Dispatch¶

zeroth.platform.dispatch ¶

Durable run dispatch infrastructure: lease claiming and ARQ wakeups.

LeaseManager owns the SQL lease protocol over the runs table; the ARQ helpers provide best-effort wakeup signals so workers can skip a poll interval. The run worker that drives claimed runs through the orchestrator is runtime code and lives in :mod:zeroth.runtime.orchestration.run_worker.

LeaseManager `dataclass` ¶

LeaseManager(
    database: AsyncDatabase,
    lease_duration_seconds: int = 60,
)

Manages worker leases on runs stored in an async database.

A lease is an exclusive claim on a run. Workers use leases to prevent two concurrent workers from both executing the same run. Leases expire after lease_duration_seconds so a crashed worker's work can be reclaimed.

claim_pending `async` ¶

claim_pending(
    deployment_ref: str,
    worker_id: str,
    *,
    tenant_id: str | None = None,
    workspace_id: str | None | object = _UNSCOPED_WORKSPACE,
    max_concurrency: int | None = None,
) -> str | None

Atomically claim one PENDING run for this worker.

Dispatches to _claim_pending_pg (Postgres) or _claim_pending_sqlite (SQLite) based on the database backend.

Returns the run_id that was claimed, or None if no work is available. The claimed run's status is left as PENDING -- the worker transitions it to RUNNING once execution actually starts.

claim_pending_result `async` ¶

claim_pending_result(
    deployment_ref: str,
    worker_id: str,
    *,
    tenant_id: str | None = None,
    workspace_id: str | None | object = _UNSCOPED_WORKSPACE,
    max_concurrency: int | None = None,
) -> LeaseClaimResult

Claim once and return saturation observed by this exact call.

claim_orphaned `async` ¶

claim_orphaned(
    deployment_ref: str,
    worker_id: str,
    *,
    tenant_id: str | None = None,
    workspace_id: str | None | object = _UNSCOPED_WORKSPACE,
    max_concurrency: int | None = None,
    claim_limit: int | None = None,
) -> list[str]

Claim all RUNNING runs with expired leases for this deployment.

Called at worker startup to recover work abandoned by crashed workers. Sets recovery_checkpoint_id to the latest checkpoint for each claimed run so the worker knows where to resume.

claim_orphaned_result `async` ¶

claim_orphaned_result(
    deployment_ref: str,
    worker_id: str,
    *,
    tenant_id: str | None = None,
    workspace_id: str | None | object = _UNSCOPED_WORKSPACE,
    max_concurrency: int | None = None,
    claim_limit: int | None = None,
) -> _OrphanClaimResult

Claim expired RUNNING runs and distinguish saturation from exhaustion.

renew_lease `async` ¶

renew_lease(
    run_id: str,
    worker_id: str,
    *,
    generation: int | None = None,
    tenant_id: str | None = None,
    workspace_id: str | None | object = _UNSCOPED_WORKSPACE,
) -> bool

Extend the lease expiry for an active run.

Returns True if the lease was renewed (i.e. we still own it), False if another worker has taken over or the run no longer exists.

generation qualifies the renewal on top of ownership. Worker ids are fresh per process, so owner-qualification alone already catches takeover by a different worker; the generation additionally catches the case where the lease was released and re-acquired, and is what the caller must then present to :meth:commit_fenced.

current_generation `async` ¶

current_generation(
    run_id: str,
    *,
    tenant_id: str | None = None,
    workspace_id: str | None | object = _UNSCOPED_WORKSPACE,
) -> int | None

The run's current lease generation, or None if the run is unknown.

current_holder `async` ¶

current_holder(
    run_id: str,
    *,
    tenant_id: str | None = None,
    workspace_id: str | None | object = _UNSCOPED_WORKSPACE,
) -> str | None

The worker id currently holding the run's lease, or None.

A write fence is only meaningful for the worker that actually holds the lease; installing one without ownership would reject every save.

commit_fenced `async` ¶

commit_fenced(
    run_id: str,
    worker_id: str,
    *,
    generation: int,
    tenant_id: str | None = None,
    workspace_id: str | None | object = _UNSCOPED_WORKSPACE,
    metrics_collector: object | None = None,
    **columns: object,
) -> bool

Apply a run-state write only if the caller still holds the lease.

The fence is part of the UPDATE predicate rather than a preceding check, because a check-then-write leaves a window in which ownership can move between the two statements -- precisely the race this exists to close.

Returns True when the write landed, False when the lease expired or a newer generation (or a different owner) superseded the caller.

release_lease `async` ¶

release_lease(
    run_id: str,
    worker_id: str,
    *,
    generation: int,
    tenant_id: str | None = None,
    workspace_id: str | None | object = _UNSCOPED_WORKSPACE,
) -> bool

Clear only the exact owned lease generation after execution.

hand_back_to_pending `async` ¶

hand_back_to_pending(
    run_id: str,
    worker_id: str,
    *,
    generation: int,
    tenant_id: str | None = None,
    workspace_id: str | None | object = _UNSCOPED_WORKSPACE,
) -> bool

Atomically return one exact RUNNING lease generation to PENDING.

expire_recovery_lease `async` ¶

expire_recovery_lease(
    run_id: str,
    worker_id: str,
    *,
    generation: int,
    tenant_id: str | None = None,
    workspace_id: str | None | object = _UNSCOPED_WORKSPACE,
) -> bool

Make owned recovery work immediately reclaimable without losing its checkpoint.

clear_lease `async` ¶

clear_lease(
    run_id: str,
    *,
    tenant_id: str | None = None,
    workspace_id: str | None | object = _UNSCOPED_WORKSPACE,
) -> None

Clear the lease columns regardless of the current lease owner.

get_recovery_checkpoint_id `async` ¶

get_recovery_checkpoint_id(
    run_id: str,
    *,
    tenant_id: str | None = None,
    workspace_id: str | None | object = _UNSCOPED_WORKSPACE,
) -> str | None

Return the recovery_checkpoint_id stored on the run, if any.

arq_settings_from_zeroth ¶

arq_settings_from_zeroth(redis_settings: Any) -> Any

Convert ZerothSettings.redis to ARQ RedisSettings.

Parameters:

Name	Type	Description	Default
`redis_settings`	`Any`	A RedisSettings instance from zeroth.platform.config.settings.	required

Returns:

Type	Description
`Any`	An arq.connections.RedisSettings instance.

create_arq_pool `async` ¶

create_arq_pool(redis_settings: Any) -> Any

Create an ARQ connection pool from Zeroth Redis settings.

Returns None if ARQ or Redis is unavailable.

enqueue_wakeup `async` ¶

enqueue_wakeup(arq_pool: Any, run_id: str) -> None

Best-effort wakeup enqueue. Never raises.

Enqueues a minimal ARQ job carrying only the run_id as a signal for a worker to check the lease store. The job itself does nothing -- the act of receiving it IS the wakeup signal.

run_arq_consumer `async` ¶

run_arq_consumer(
    redis_settings: Any,
    on_wakeup: Callable[[str], Awaitable[None]],
) -> None

Run ARQ consumer as a background task.

Calls on_wakeup(run_id) for each wakeup signal received. Runs until cancelled. Designed to be wrapped in asyncio.create_task.

Parameters:

Name	Type	Description	Default
`redis_settings`	`Any`	Zeroth RedisSettings (will be converted to ARQ format).	required
`on_wakeup`	`Callable[[str], Awaitable[None]]`	Async callback invoked when a wakeup signal arrives.	required

zeroth.runtime.orchestration.run_worker ¶

Durable run worker that replaces asyncio.create_task dispatch.

A RunWorker polls SQLite for PENDING runs, claims them via lease, drives them through the RuntimeOrchestrator, and releases the lease on completion. On startup it reclaims any orphaned RUNNING runs whose leases have expired.

The worker runs as a single asyncio background task started in the app lifespan. Graceful shutdown cancels the poll loop without interrupting runs that are currently executing (the semaphore ensures bounded concurrency).

RunWorker `dataclass` ¶

RunWorker(
    deployment_ref: str,
    run_repository: RunRepository,
    orchestrator: RuntimeOrchestrator,
    graph: Graph,
    lease_manager: LeaseManager,
    tenant_id: str | None = None,
    workspace_id: str | None = None,
    max_concurrency: int = 8,
    poll_interval: float = 0.5,
    orphan_sweep_interval: float = 5.0,
    worker_id: str = _new_worker_id(),
    dead_letter_manager: DeadLetterManager | None = None,
    metrics_collector: MetricsCollector | None = None,
    shutdown_timeout: float = 30.0,
)

Long-lived worker that drives PENDING runs to completion.

Attributes:

Name	Type	Description
`deployment_ref`	`str`	The deployment this worker serves.
`run_repository`	`RunRepository`	Used to load/transition runs.
`orchestrator`	`RuntimeOrchestrator`	Drives execution for each run.
`graph`	`Graph`	The deployment graph passed to the orchestrator.
`lease_manager`	`LeaseManager`	Manages SQLite-backed leases.
`max_concurrency`	`int`	Maximum simultaneous runs (default 8).
`poll_interval`	`float`	Seconds between poll ticks when idle (default 0.5).
`worker_id`	`str`	Unique ID for this worker instance.
`dead_letter_manager`	`DeadLetterManager \| None`	Optional; marks repeatedly-failing runs as dead-letter.
`metrics_collector`	`MetricsCollector \| None`	Optional; records execution metrics.

start `async` ¶

start() -> None

Recover orphaned runs from crashed workers, then begin the poll loop.

interrupt_active_run `async` ¶

interrupt_active_run(run_id: str) -> None

Stop this worker's active drive without changing persisted status.

poll_loop `async` ¶

poll_loop() -> None

Continuously claim and dispatch PENDING runs until cancelled.

handle_wakeup `async` ¶

handle_wakeup(run_id: str) -> None

ARQ wakeup callback -- attempt to claim from DB immediately.

The run_id is informational only; the worker always claims from the lease store (not from the ARQ job payload) per D-06.

graceful_shutdown `async` ¶

graceful_shutdown() -> None

Wait for in-flight tasks then release remaining leases to PENDING.

Called on SIGTERM. Steps: 1. Set stopping flag so poll_loop exits cleanly 2. Wait for active tasks to complete (up to shutdown_timeout) 3. For any tasks still running, cancel them and release their leases back to PENDING so another worker can claim them

Dispatch¶

zeroth.platform.dispatch ¶

LeaseManager dataclass ¶

claim_pending async ¶

claim_pending_result async ¶

claim_orphaned async ¶

claim_orphaned_result async ¶

renew_lease async ¶

current_generation async ¶

current_holder async ¶

commit_fenced async ¶

release_lease async ¶

hand_back_to_pending async ¶

expire_recovery_lease async ¶

clear_lease async ¶

get_recovery_checkpoint_id async ¶