Owned SQL worker queues¶

The validation, pipeline and streaming SQL job sources provide explicit owned submission, claim recovery and fenced mutations. These are queue primitives; existing worker/launcher loops do not yet carry these fences. They do not enable agent execution, provider submission, evaluation or publication. See agent step definitions for the execution boundary.

Sources and identities¶

Use the existing configured source and its public job envelope:

from pycharter.worker.job_sources.validation import ValidationJobSource
from pycharter.worker.job_sources.pipeline import PipelineJobSource
from pycharter.worker.job_sources.streaming import StreamingJobSource
from pycharter.worker.models import ValidationJob, PipelineJob, StreamingJob
from pycharter.worker.config import WorkerConfig
from pycharter.worker.claims import QueueClaimFence, OwnedStreamingState
from pycharter.worker.job_sources.owned import (
    OwnedValidationJobSource,
    OwnedPipelineJobSource,
    OwnedStreamingJobSource,
)

Sources retain their existing (db_url, config=None) constructors and asynchronous close() method. Trusted outer setup selects the database and existing WorkerConfig; requests cannot select another database. The new capabilities are described by runtime-checkable protocols alongside existing interfaces. For example, trusted setup can construct a validation source with ValidationJobSource(configured_db_url, WorkerConfig(db_url=configured_db_url)). Apply the queue migration first and call await source.close() when finished; constructing the source does not qualify or migrate the database.

Source	Submission envelope	`submit_owned` returns	Owned claim returns
`ValidationJobSource`	`ValidationJob`	Original job/event ID	`ClaimedJob`
`PipelineJobSource`	`PipelineJob`	Original run ID	`ClaimedPipelineJob`
`StreamingJobSource`	`StreamingJob`	Original launch-job ID	`ClaimedStreamingJob`

Each successful owned claim includes claim_fence. Its frozen, slotted QueueClaimFence contains queue kind, job ID, worker/launcher ID, generation and the opaque claim token. For pipeline work the fence contains the job ID, not the returned run ID. Use the returned fence unchanged; identifiers and nonces are correlation, not authorization grants. Tokens are hidden from fence repr and ordinary job-list responses.

Retaining one original submission¶

submit_owned(job, *, submission_id) requires a nonempty caller-known identity before dispatch, even when job_id, run_id or the legacy idempotency key is None. The trusted caller retains that identity and the original request for recovery. Do not allocate a replacement identity to resolve an uncertain write. The example only stores a synthetic queue request; no contract is executed:

request = ValidationJob(
    contract_name="synthetic_orders",
    contract_version="1.0.0",
    data_source="synthetic-orders.jsonl",
    options={"fixture": True},
)
job_id = await source.submit_owned(request, submission_id=known_submission_id)

Before its first asynchronous bridge, the method validates the exact expected job type and copies nested finite JSON into an immutable original snapshot. Later mutation of the caller's dictionaries cannot change that dispatch. The snapshot checks request shape; it does not execute a Pipeline or validate data against a referenced DataContract.

A versioned SHA-256 fingerprint covers the queue, submission ID and every authored job field. Canonical encoding uses UTF-8, sorted object keys, compact separators, ordered arrays and explicit nulls, without Unicode normalization. None and an empty mapping remain distinct. Explicit fires_at requires an aware datetime, normalized to UTC with six fractional digits; equal instants with different offsets match. None remains different from an explicit timestamp.

Generated job/run IDs and the actual default enqueue time are assigned once. An original authored None stays None in the fingerprint: retrying with a returned generated ID filled into the request conflicts. Attempts, lease/status, backoff, result/error and PID state do not rewrite the original fingerprint.

On a genuine PostgreSQL UUID job-ID column, submission and claim projection use canonical database UUID text; pipeline submission still returns its run ID. An explicit authored UUID spelling remains unchanged in the fingerprint. Replay with the original request, retaining its authored spelling or None. Changing that spelling or replacing an authored None with a returned generated ID conflicts.

Exact replay returns the original queue-specific ID, including after completion, failure or cancellation. Changed content under the same identity conflicts; existing legacy rows are never promoted into owned submissions. Replay is limited to the selected database and queue table. The fingerprint does not retain a replayable original request or replace an acquisition ledger.

Claims, recovery and final writes¶

The common owned methods are:

Operation	Contract
`claim_next_owned(owner_id, *, claim_token)`	Fresh caller nonce; returns the next eligible claimed envelope or None
`recover_owned_claim(owner_id, *, claim_token)`	Original nonce lookup; does not claim again or renew a lease
`heartbeat_owned(fence)`	Renew the exact current live claim
`complete_owned(fence, result=None)`	Complete the exact current live claim; repeated completion returns False
`fail_owned(fence, error, *, retry, dlq_record_id=None)`	Validation/pipeline failure or configured retry/backoff; returns a Boolean

The first parameter is named worker_id on validation/pipeline sources and launcher_id on streaming sources. Streaming fail_owned has the required retry keyword and no dlq_record_id parameter. A DLQ ID on a finite queue is retained correlation; this method does not implement a DLQ write transaction.

A fresh nonce is known and retained before each new claim dispatch. Repeating claim_next_owned with a stored nonce raises a claim conflict; use recovery for an uncertain original claim. The next eligible job can be any submission in that queue, not necessarily the one most recently submitted by this caller.

The requested owner and nonce must survive storage as the exact strings sent, with the expected integer generation. If coercion or a trigger changes them, the whole claim rolls back with QUEUE_CLAIM_CONFLICT/claim_identity_changed; no changed attempt, generation or token is committed. Inputs are never padded or trimmed. PostgreSQL CHAR owners must already be supplied in their canonical padded representation to round-trip faithfully; normal VARCHAR and SQLite CHAR owners retain their exact values.

claimed = await source.claim_next_owned(worker_id, claim_token=known_fresh_nonce)
if claimed is not None:
    fence = claimed.claim_fence
    assert fence is not None
    if await source.heartbeat_owned(fence):
        committed = await source.complete_owned(fence, {"queue_fixture": True})
        # False means this final write did not apply; never fall back to complete(id).

Owned mutations and streaming state reads first compare the locked row's canonical string job ID, exact owner/token, strict integer generation and queue with the supplied fence. SQL case folding or fixed-character comparison cannot substitute another spelling. The final SQL predicate checks the full fence, claimed status and a non-null lease strictly after current database time. SQLite uses a held write transaction; PostgreSQL locks the current row. Time is sampled after locking. A lock wait past expiry cannot authorize a stale write. Generation advances on every successful claim independently of retry attempts; the same worker name or an attempt reset cannot restore an earlier fence.

True means the exact current mutation applied. False means known missing, stale, expired, terminal or operation-ineligible state. Database failure raises; it never becomes False. A representable wrong-case UUID fence refuses by the exact locked-row comparison. A syntactically invalid native UUID ID can instead be rejected before row ownership is established: it raises the safe QUEUE_PERSISTENCE_FAILED/unavailable boundary with no input/driver details or row changes. That error is not evidence of an absent claim. Carry the returned fence unchanged. A recovered None means no matching live claim was observed, not proof that an outstanding original request cannot still commit. Neither None nor a lost lease authorizes another external execution attempt.

Unknown acknowledgements and errors¶

Owned operations raise package errors with bounded authored reasons:

Error	Meaning
`ConfigValidationError`, code `CONFIG`	Invalid owned request, result or fence
`PyCharterError`, code `QUEUE_SUBMISSION_CONFLICT`	Submission mismatch, collision or inconsistent protection
`PyCharterError`, code `QUEUE_CLAIM_CONFLICT`	Nonce collision, invalid/exhausted generation or a stored claim identity change
`PyCharterError`, code `QUEUE_PERSISTENCE_FAILED`	Required persistence failed, including an uncertain commit acknowledgement

The owned error surface does not echo supplied IDs/tokens, payloads, connection strings or driver details. Handle codes, not exception-message text. Submission recovery retries the same original request and submission ID; claim recovery uses recover_owned_claim with the original owner and nonce:

from pycharter.shared.errors import ErrorCode, PyCharterError

try:
    claimed = await source.claim_next_owned(worker_id, claim_token=known_fresh_nonce)
except PyCharterError as error:
    if error.code != ErrorCode.QUEUE_PERSISTENCE_FAILED:
        raise
    claimed = await source.recover_owned_claim(
        worker_id, claim_token=known_fresh_nonce
    )

Recovery itself can fail. These primitives supply no automatic retry loop, provider cancellation, remote uncertainty resolution or release decision.

Streaming cancellation and replacement¶

Streaming additionally exposes set_pid_owned(fence, pid), get_owned_state(fence) and acknowledge_cancel_owned(fence). A state read returns frozen OwnedStreamingState(cancel_requested, pid) or known nonownership as None. It is a snapshot, not authority for a later action or process signal.

Completion, failure and PID changes refuse a cancel-requested claim. Its live fence can still heartbeat for cleanup. Cancellation acknowledgement requires that live fence plus the current cancellation request; it records local cancelled state, clears lease/PID and releases the legacy active key. An expired cancel-requested stream claim becomes cancelled instead of executable again. None of these operations proves an external process stopped.

A fresh owned force/rerun requires a new submission ID. Active-key predecessor cancellation/key release and replacement insertion are atomic. Exact replay is checked first and rechecked under the active-key predecessor lock, including when another identical request commits after the initial lookup. Retrying an old force request returns its original job ID and cannot cancel a successor. Changed originals conflict before cancellation or active-key release. An active owned predecessor requires force=True; a terminal owned predecessor can release its active key for a fresh rerun. A legacy predecessor collision refuses; legacy force cannot cancel protected owned work. This does not add a general per-config admission policy.

SQL schema, legacy behavior and current limits¶

The scope is the existing SQLite/PostgreSQL queues; there is no MongoDB, Redis or in-memory worker adapter added here. No topology, failover, disk durability or provider/account profile is qualified by these methods. Lease duration and retry/backoff come from existing WorkerConfig; this page adds no numeric execution policy or defaults. No queue payload-byte quota or expiry policy is introduced by this slice.

Owned submission IDs, claim tokens and owner IDs are nonempty strings of at most 255 characters, compared exactly without case-folding, Unicode normalization or whitespace trimming. Job IDs are at most 36 characters and pipeline run IDs at most 32; finite-queue DLQ IDs are at most 32. Owned max_attempts is a strict positive signed-32-bit integer; generation is a positive signed-64-bit claim value and refuses overflow. Booleans do not satisfy integer fields. JSON inputs require built-in finite JSON values; variables require string values, and streaming PIDs are positive signed-32-bit integers. These are technical validation bounds, not live budgets.

Deploy the additive queue migration through the normal SQL migration workflow. It adds generation/token and retained submission protection. Migration admission requires canonical VARCHAR types for the new string metadata, exact full unique claim-token/submission-ID indexes and compatible nullability/defaults. These two new identity columns require BINARY comparison on SQLite and deterministic collations on PostgreSQL. Fixed CHAR metadata, partial or differently compared indexes, and defaults that manufacture identity or invalid Boolean/generation values refuse. Safe quoted numeric zero remains compatible; SQLite quoted 'false' refuses, while PostgreSQL typed quoted false is valid. The SQLite native collation probe is rolled back and retains no probe objects or rows. These checks happen during migration, not on every owned call; privileged schema mutation after admission is outside their guarantee. Genuine old rows start with generation zero, no token and no owned submission identity. Existing owner/job-ID column types are retained; exact locked-row binding guards their comparisons. Request fingerprints are never fabricated from live rows.

In this version, legacy claims exclude fence_required submissions. Legacy heartbeat/complete/fail/PID writes cannot change a protected submission or a currently owned incarnation. An owned claim of an ordinary legacy row protects that incarnation but does not create owned-submission replay metadata. Administrative cancellation may invalidate ownership while retaining its submission identity. No old ID-only worker fallback receives the owned guarantee.

Owned submission ID/fingerprint/flag survive terminal, retry, cancellation and force transitions. Existing validation/pipeline retention skips protected rows; validation public deletion returns False for them. Controlled downgrade locks before checking and refuses retained protected submissions or active non-null-token claims, including owned legacy work. A separate reviewed retirement procedure is still needed to remove protected identity. No indefinite replay or tombstone guarantee survives privileged raw SQL purge/restore or incompatible old binaries.