httk.workflow.transfers

Crash-recoverable detached job transfer.

Attributes

TRANSFER_DIRECTORY

TRANSFER_MANIFEST

TRANSFER_RUNNERS

TRANSFER_FORMAT

TRANSFER_FORMAT_VERSION

TRANSFER_OFFER_FORMAT

TRANSFER_RETIREMENT_FORMAT

DEFAULT_OFFER_STATES

Classes

TransferCandidate

One job a transfer offer or resume can actually inspect.

Functions

detach_job(workspace, job_id, *[, marker, ...])

Fence and seal one job, leaving no schedulable source marker.

validate_bundle(bundle)

Validate a sealed bundle and return its manifest.

import_bundle(workspace, bundle)

Import one sealed bundle, or validate and acknowledge its replay.

acknowledge_transfer(workspace, acknowledgement)

Validate one destination receipt and retire its exact source transfer.

recover_transfers(workspace)

Finish source sealing and inventory every retained bundle.

select_transfer_jobs(workspace, *, ...[, states, ...])

Select sealed and live jobs without changing transfer state.

offer_transfers(workspace, *, destination_workspace_id)

Seal every finished job of workspace into a bundle for one destination.

retire_transfers(workspace, job_ids, *[, ...])

Retire the sealed source bundle of every named job.

discard_staged_bundle(workspace, staging)

Drop a staged incoming bundle whose payload the workspace now owns.

Module Contents

httk.workflow.transfers.TRANSFER_DIRECTORY = '.httk-transfer'
httk.workflow.transfers.TRANSFER_MANIFEST = 'manifest.json'
httk.workflow.transfers.TRANSFER_RUNNERS = 'runners'
httk.workflow.transfers.TRANSFER_FORMAT = 'httk-workflow-detached-transfer'
httk.workflow.transfers.TRANSFER_FORMAT_VERSION = 2
httk.workflow.transfers.TRANSFER_OFFER_FORMAT = 'httk-workflow-transfer-offer'
httk.workflow.transfers.TRANSFER_RETIREMENT_FORMAT = 'httk-workflow-transfer-retirement'
httk.workflow.transfers.DEFAULT_OFFER_STATES = ('succeeded', 'failed')
class httk.workflow.transfers.TransferCandidate

One job a transfer offer or resume can actually inspect.

Parameters:

  • job_id – Identify the job.

  • job_key – Preserve the stable job key.

  • prior_kind – State the job had before transfer.

  • source_placement – Placement in the source workspace.

  • bundle – Sealed payload, when this is a resumed transfer.

  • marker – Live source marker, when this is a new offer.

  • job – Readable immutable job definition, when available.

  • manifest – Validated sealed-bundle manifest, when available.

  • problem – Readability problem, if the candidate cannot be validated.

job_id: str
job_key: str
prior_kind: str
source_placement: pathlib.PurePosixPath
bundle: pathlib.Path | None
marker: httk.workflow.models.Marker | None
job: httk.workflow.models.JobDefinition | None
manifest: collections.abc.Mapping[str, Any] | None = None
problem: str | None = None
httk.workflow.transfers.detach_job(workspace, job_id, *, marker=None, waiting_parent_map=None, destination_workspace_id, destination_remote=None, destination_placement=None, transfer_id=None)

Fence and seal one job, leaving no schedulable source marker.

Parameters:

  • workspace (httk.workflow.workspace.Workspace) – Provide the source workspace.

  • job_id (str) – Identify the job to detach.

  • marker (httk.workflow.models.Marker | None) – An already resolved source marker; avoids scanning all states.

  • waiting_parent_map (collections.abc.Mapping[str, set[str]] | None) – Precomputed child-to-parent map for this transfer batch.

  • destination_workspace_id (str) – Identify the destination workspace.

  • destination_remote (str | None) – Preserve the destination’s remote identifier.

  • destination_placement (str | pathlib.PurePosixPath | None) – Override the destination placement.

  • transfer_id (str | None) – Reuse a transfer id when resuming sealing.

Returns:

The sealed source bundle path.

Raises:

ValueError – If the job is missing, active, joined, or already transferring incompatibly.

Return type:

pathlib.Path

httk.workflow.transfers.validate_bundle(bundle)

Validate a sealed bundle and return its manifest.

Parameters:

bundle (str | os.PathLike[str]) – Locate the sealed transfer bundle.

Returns:

The validated transfer manifest.

Raises:

httk.workflow.errors.FormatError – If the bundle format, digest, marker, or runner is invalid.

Return type:

dict[str, Any]

httk.workflow.transfers.import_bundle(workspace, bundle)

Import one sealed bundle, or validate and acknowledge its replay.

httk.workflow.transfers.acknowledge_transfer(workspace, acknowledgement)

Validate one destination receipt and retire its exact source transfer.

httk.workflow.transfers.recover_transfers(workspace)

Finish source sealing and inventory every retained bundle.

Parameters:

workspace (httk.workflow.workspace.Workspace) – Provide the workspace whose transfers to recover.

Returns:

The recovered and retained transfer records.

Return type:

list[dict[str, object]]

httk.workflow.transfers.select_transfer_jobs(workspace, *, destination_workspace_id, states=DEFAULT_OFFER_STATES, placement=None, job_ids=None, destination_remote=None, include_transferring=False, known_markers=None, waiting_parent_map=None)

Select sealed and live jobs without changing transfer state.

Parameters:

  • workspace (httk.workflow.workspace.Workspace) – Provide the source workspace.

  • destination_workspace_id (str) – Select one destination’s sealed ledgers.

  • states (collections.abc.Iterable[str]) – Select quiescent states eligible for offering.

  • placement (str | pathlib.PurePosixPath | None) – Restrict source placements by normalized prefix.

  • job_ids (collections.abc.Iterable[str] | None) – Restrict selection to explicit job ids when supplied.

  • destination_remote (str | None) – Restrict sealed ledgers to one remote name.

  • include_transferring (bool) – Include live interrupted-transfer markers for advisory checks.

  • known_markers (collections.abc.Sequence[httk.workflow.models.Marker] | None) – Use these already-resolved live markers instead of scanning the workspace.

  • waiting_parent_map (collections.abc.Mapping[str, set[str]] | None) – Reuse one waiting-parent map across a transfer batch.

Returns:

Candidates in the same placement/key order as offer_transfers().

Raises:

ValueError – If the destination id or requested states are invalid.

Return type:

list[TransferCandidate]

httk.workflow.transfers.offer_transfers(workspace, *, destination_workspace_id, states=DEFAULT_OFFER_STATES, placement=None, job_ids=None)

Seal every finished job of workspace into a bundle for one destination.

This is the far side of a results fetch: the remote that ran the work offers what stopped there, and the workspace that asked pulls each bundle and imports it. Offering is idempotent because a sealed bundle is reported from its ledger rather than sealed again, so the jobs a first call detached — which no longer have a schedulable marker — are exactly the jobs a second call re-offers, and an interrupted fetch resumes by simply asking again.

A job that cannot leave right now is skipped rather than fatal: one still referenced by an unresolved join keeps the campaign it belongs to consistent, and reporting the rest lets the fetch make progress.

Parameters:

  • workspace (httk.workflow.workspace.Workspace) – Provide the source workspace.

  • destination_workspace_id (str) – Identify the destination workspace.

  • states (collections.abc.Iterable[str]) – Select quiescent states eligible for offering.

  • placement (str | pathlib.PurePosixPath | None) – Restrict offers to this placement prefix.

  • job_ids (collections.abc.Iterable[str] | None) – Restrict the offer to these explicit job ids.

Returns:

Offered transfer records in placement order.

Raises:

ValueError – If the destination id or requested states are invalid.

Return type:

list[dict[str, object]]

httk.workflow.transfers.retire_transfers(workspace, job_ids, *, destination_workspace_id=None)

Retire the sealed source bundle of every named job.

A fetch retires at the source only once the destination holds an acknowledgement, so the identity of the job is all this side needs; naming the destination as well refuses to retire a bundle that was sealed for somebody else. Retirement renames the bundle, publishes its retired ledger, then reclaims the bundle and unprotected source journal segments unless retention says to keep them. Repeated retirement also retries cleanup.

Parameters:

  • workspace (httk.workflow.workspace.Workspace) – Provide the source workspace.

  • job_ids (collections.abc.Sequence[str]) – Identify the jobs whose bundles to retire.

  • destination_workspace_id (str | None) – Restrict retirement to one destination.

Returns:

Retirement records for the named jobs.

Raises:

ValueError – If a job id has no matching detached transfer.

Return type:

list[dict[str, object]]

httk.workflow.transfers.discard_staged_bundle(workspace, staging)

Drop a staged incoming bundle whose payload the workspace now owns.

The staging tree is renamed out of the incoming directory before it is removed, so an interrupted removal can never leave a partial bundle where a resumed fetch would find one and mistake it for the real thing.

Parameters:

  • workspace (httk.workflow.workspace.Workspace) – Provide the workspace owning the staging directory.

  • staging (pathlib.Path) – Locate the staged bundle to discard.