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, *, destination_workspace_id)

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)

Idempotently import a sealed bundle and publish its prior state.

acknowledge_transfer(workspace, acknowledgement)

Validate an acknowledgement and retire the sealed source bundle.

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'[source]
httk.workflow.transfers.TRANSFER_MANIFEST = 'manifest.json'[source]
httk.workflow.transfers.TRANSFER_RUNNERS = 'runners'[source]
httk.workflow.transfers.TRANSFER_FORMAT = 'httk-workflow-detached-transfer'[source]
httk.workflow.transfers.TRANSFER_FORMAT_VERSION = 2[source]
httk.workflow.transfers.TRANSFER_OFFER_FORMAT = 'httk-workflow-transfer-offer'[source]
httk.workflow.transfers.TRANSFER_RETIREMENT_FORMAT = 'httk-workflow-transfer-retirement'[source]
httk.workflow.transfers.DEFAULT_OFFER_STATES = ('succeeded', 'failed')[source]
class httk.workflow.transfers.TransferCandidate[source]

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

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.

  • 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)[source]

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)[source]

Idempotently import a sealed bundle and publish its prior state.

Runners are installed and verified before the imported job becomes schedulable; the returned acknowledgement identifies the imported payload and transfer.

Parameters:

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

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

Returns:

The destination acknowledgement.

Raises:
Return type:

dict[str, object]

httk.workflow.transfers.acknowledge_transfer(workspace, acknowledgement)[source]

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.

Parameters:

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

  • acknowledgement (collections.abc.Mapping[str, object]) – Supply the destination acknowledgement.

Returns:

The retired source bundle path.

Raises:

httk.workflow.errors.FormatError – If the acknowledgement format, signature, or identity is invalid.

Return type:

pathlib.Path

httk.workflow.transfers.recover_transfers(workspace)[source]

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)[source]

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.

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)[source]

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.

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)[source]

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.

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)[source]

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.