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

Functions

detach_job(→ pathlib.Path)	Fence and seal one job, leaving no schedulable source marker.
validate_bundle(→ dict[str, Any])	Validate a sealed bundle and return its manifest.
import_bundle(→ dict[str, object])	Idempotently import a sealed bundle and publish its prior state.
acknowledge_transfer(→ pathlib.Path)	Validate an acknowledgement and retire the sealed source bundle.
recover_transfers(→ list[dict[str, object]])	Finish source sealing and inventory every retained bundle.
offer_transfers(→ list[dict[str, object]])	Seal every finished job of workspace into a bundle for one destination.
retire_transfers(→ list[dict[str, object]])	Retire the sealed source bundle of every named job.
discard_staged_bundle(→ None)	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')

httk.workflow.transfers.detach_job(workspace: httk.workflow.workspace.Workspace, job_id: str, *, destination_workspace_id: str, destination_placement: str | pathlib.PurePosixPath | None = None, transfer_id: str | None = None) → pathlib.Path

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

httk.workflow.transfers.validate_bundle(bundle: str | os.PathLike[str]) → dict[str, Any]

Validate a sealed bundle and return its manifest.

httk.workflow.transfers.import_bundle(workspace: httk.workflow.workspace.Workspace, bundle: str | os.PathLike[str]) → dict[str, object]

Idempotently import a sealed bundle and publish its prior state.

httk.workflow.transfers.acknowledge_transfer(workspace: httk.workflow.workspace.Workspace, acknowledgement: collections.abc.Mapping[str, object]) → pathlib.Path

Validate an acknowledgement and retire the sealed source bundle.

An acknowledgement that carries an identity signature must carry a valid one: retiring a source is irreversible enough that a damaged or forged attribution is refused rather than recorded. An acknowledgement with no signature is accepted exactly as before, so a destination without an identity key keeps working.

httk.workflow.transfers.recover_transfers(workspace: httk.workflow.workspace.Workspace) → list[dict[str, object]]

Finish source sealing and inventory every retained bundle.

httk.workflow.transfers.offer_transfers(workspace: httk.workflow.workspace.Workspace, *, destination_workspace_id: str, states: collections.abc.Iterable[str] = DEFAULT_OFFER_STATES, placement: str | pathlib.PurePosixPath | None = None) → list[dict[str, object]]

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.

httk.workflow.transfers.retire_transfers(workspace: httk.workflow.workspace.Workspace, job_ids: collections.abc.Sequence[str], *, destination_workspace_id: str | None = None) → list[dict[str, object]]

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 moves the bundle rather than deleting it, and a bundle already retired is reported as such, so calling this twice is the same as calling it once.

httk.workflow.transfers.discard_staged_bundle(workspace: httk.workflow.workspace.Workspace, staging: pathlib.Path) → None

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.