Struct PgStore 

pub struct PgStore { /* private fields */ }

The Postgres adapter. Construct it with PgStore::new around a pool YOU own — failure classification: headgate never closes a pool it did not open, and every entry point accepts an existing one. PgStore::connect is a convenience for tools and tests.

CONNECTION BUDGET (failure classification, live-proven in both languages/backends): store calls borrow transiently; each in-flight TRANSACTIONAL handler (once, step_once) HOLDS one; connect*/with_listen adds ONE dedicated LISTEN connection OUTSIDE the pool. If T callbacks may retain transactions concurrently across every worker sharing this pool, size the command pool at T + 2. No internal path holds one pooled connection while acquiring another. See docs/connection-budget.md for the exact physical formula and the user-created nested-acquisition caveat.

push wakeups push wakeup needs a dedicated LISTEN connection, which needs connection config — so the Notifying capability exists only on stores built via connect* (or PgStore::with_listen). A pool-only store polls, honestly.

Implementations

impl PgStore

pub async fn set_archive_policy( &self, queue: &str, retention: Duration, ) -> Result<(), StoreError>

pub async fn clear_archive_policy(&self, queue: &str) -> Result<(), StoreError>

pub async fn prune_archive_month(&self, month: &str) -> Result<u64, StoreError>

Truncate one closed monthly archive partition after every row’s retained audit lifetime has elapsed. A closed month cannot receive a normal store-time eviction, which makes the safety check stable until TRUNCATE takes its table lock.

pub async fn index_health(&self) -> Result<Vec<IndexHealth>, StoreError>

pub async fn reindex_concurrently(&self, name: &str) -> Result<(), StoreError>

pub fn new(pool: Pool) -> Self

failure classification caller-supplied pool. Never closed by this crate.

pub fn with_options(pool: Pool, opts: PgStoreOptions) -> Self

pub fn in_schema(pool: Pool, schema: &str) -> Result<Self, StoreError>

Wrap a caller-owned pool while explicitly qualifying every headgate object with schema. The pool may be shared by stores for other schemas: no session-level search_path state is changed or trusted.

pub fn with_options_in_schema( pool: Pool, opts: PgStoreOptions, schema: &str, ) -> Result<Self, StoreError>

pub fn schema(&self) -> Option<&str>

The explicit schema configured at store construction. None is the legacy/default connection namespace and preserves byte-identical SQL for existing callers.

pub fn with_listen(self, config: Config) -> Self

Enable push wakeups push wakeup on a pool-constructed store by supplying connection config for the dedicated LISTEN connection.

pub fn connect(conninfo: &str, max_size: usize) -> Result<Self, StoreError>

Convenience constructor from a libpq conninfo string / URL.

pub fn connect_in_schema( conninfo: &str, max_size: usize, schema: &str, ) -> Result<Self, StoreError>

pub fn connect_with_options( conninfo: &str, max_size: usize, opts: PgStoreOptions, ) -> Result<Self, StoreError>

pub fn connect_in_schema_with_options( conninfo: &str, max_size: usize, opts: PgStoreOptions, schema: &str, ) -> Result<Self, StoreError>

pub async fn begin(&self) -> Result<PgTx, StoreError>

Begin a store transaction for the Transactional port. The handle owns its connection; dropping it without commit detaches the connection from the pool so the open transaction aborts server-side instead of leaking into the next lease of that connection.

pub async fn enqueue_on<C: GenericClient>( &self, c: &C, batch: &[Envelope], ) -> Result<(), StoreError>

Enqueue on an explicit transaction — the shared path behind enqueue, enqueue_tx, and direct tokio_postgres::Transaction interop (caller-owned transaction contract).

Trait Implementations

impl CheckpointInspect for PgStore

fn get_job_checkpoint<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<Checkpoint>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Explicit checkpoint access. None means the job does not exist; an existing job with no resumable progress returns an empty Checkpoint.

impl Inspect for PgStore

fn as_result_inspect(&self) -> Option<&dyn ResultInspect>

Optional explicit result-byte reader. Keeping this separate from job/list reads prevents an accidental payload leak and preserves capability honesty.

fn as_output_inspect(&self) -> Option<&dyn OutputInspect>

Optional explicit reader for mid-run output bytes. Ordinary job/list reads keep omitting them for the same PII posture as final results and payloads.

fn as_progress_inspect(&self) -> Option<&dyn ProgressInspect>

Optional explicit reader for operator-facing progress. It stays outside ordinary job/list reads because even a short application message may contain sensitive data.

fn as_checkpoint_inspect(&self) -> Option<&dyn CheckpointInspect>

Optional explicit reader for resumable-step checkpoints. Cursor bytes may carry application data, so ordinary job/list responses never include them.

fn get_job<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, include_payload: bool, ) -> Pin<Box<dyn Future<Output = Result<Option<JobSummary>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn list_jobs<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, filter: &'life1 JobFilter, cursor: Option<&'life2 str>, limit: u32, ) -> Pin<Box<dyn Future<Output = Result<JobPage, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

fn counts<'life0, 'life1, 'async_trait>( &'life0 self, queue: Option<&'life1 str>, ) -> Pin<Box<dyn Future<Output = Result<StateCounts, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn queue_stats<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<QueueStats>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn set_queue_paused<'life0, 'life1, 'async_trait>( &'life0 self, queue: &'life1 str, paused: bool, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn set_queue_weight<'life0, 'life1, 'async_trait>( &'life0 self, queue: &'life1 str, weight: u32, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Invariant 16: queue-selection weight is fleet policy, not a worker-local polling hint. weight == 0 is invalid; omitted/unconfigured queues read as weight 1.

fn set_enqueue_limit<'life0, 'life1, 'async_trait>( &'life0 self, queue: &'life1 str, max_unfinished_jobs: Option<u64>, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Configure the fleet-wide enqueue bound. The store may accept a limit below the current depth; that immediately stops growth until drain catches up.

fn rate_classes<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<RateClassState>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn upsert_rate_class<'life0, 'life1, 'async_trait>( &'life0 self, cfg: &'life1 RateClassConfig, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Invariant 16: any policy the gate reads, the API can write — a fleet limit you cannot change without a redeploy is not an operational feature.

fn concurrency_limits<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<ConcurrencyLimitConfig>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn upsert_concurrency_limit<'life0, 'life1, 'async_trait>( &'life0 self, cfg: &'life1 ConcurrencyLimitConfig, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn partitions<'life0, 'life1, 'async_trait>( &'life0 self, queue: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Vec<PartitionState>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn quarantine_list<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<QuarantineEntry>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn quarantine_release<'life0, 'life1, 'async_trait>( &'life0 self, fingerprint: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<u64, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

crash quarantine deliberate operator action: quarantined jobs of this fingerprint become available (operator_release) and new enqueues are accepted again. Returns how many jobs were released. A released job re-quarantines on its next crash.

fn operator_retry<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

archived → available (operator_retry). Any other state is an error — the transition table defines exactly which rows exist.

fn operator_cancel<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

scheduled|available|running → cancelled (operator_cancel). Cancelling a running job clears its lease, so the holder’s next renew/ack/checkpoint is rejected and its handler stops within a heartbeat.

fn delete_job<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Delete a non-running job. Deleting mid-flight is refused (asynq’s rule).

fn explain_admission<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<AdmissionExplain>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn history<'life0, 'life1, 'async_trait>( &'life0 self, queue: &'life1 str, since_ms: i64, bucket_ms: i64, ) -> Pin<Box<dyn Future<Output = Result<Vec<HistoryBucket>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

backlog metrics time series from the incrementally-maintained counters — never a scan.

fn quarantine_sweep<'life0, 'async_trait>( &'life0 self, limit: i64, ) -> Pin<Box<dyn Future<Output = Result<u64, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

crash quarantine the quarantine sweeper (singleton duties’s duty): waiting jobs whose fingerprint is quarantined move to the terminal quarantined state, VISIBLY — without this they sit gate-excluded forever, which is an invisible skip. Returns how many moved. Bounded per call; run under a duty lease.

fn reschedule_job<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, at_ms: i64, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Move a waiting job’s run time. Defined only for scheduled and retryable — no state changes, so no transition-table row is needed.

fn edit_payload<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, id: &'life1 str, payload: &'life2 [u8], schema_version: u32, fingerprint: &'life3 str, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Edit-then-retry (control API contract). Non-running jobs only. The fingerprint is derived caller-side (content fingerprinting) and passed in, because it must change with the payload.

fn upsert_schedule<'life0, 'life1, 'async_trait>( &'life0 self, s: &'life1 Schedule, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Idempotent upsert (BullMQ’s upsertJobScheduler). next_run_ms is kept from the existing row when the spec is unchanged, so re-deploying a config does not reset the phase of a running schedule.

fn delete_schedule<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn list_schedules<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<Schedule>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn due_schedules<'life0, 'async_trait>( &'life0 self, limit: i64, ) -> Pin<Box<dyn Future<Output = Result<(Vec<Schedule>, i64), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Due entries plus STORE time — tick math must not use a worker clock.

fn advance_schedule<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, from_next_run_ms: i64, to_next_run_ms: i64, ) -> Pin<Box<dyn Future<Output = Result<bool, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Compare-and-set advance: succeeds only if next_run_ms still equals from. Losing the race means another node already advanced — never an error.

fn record_schedule_event<'life0, 'life1, 'async_trait>( &'life0 self, event: &'life1 ScheduleEvent, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Append one scheduler enqueue attempt and trim older history atomically.

fn list_schedule_events<'life0, 'life1, 'async_trait>( &'life0 self, schedule_id: &'life1 str, before_event_id: Option<u64>, limit: u32, ) -> Pin<Box<dyn Future<Output = Result<Vec<ScheduleEvent>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Newest first. limit must be in 1..=SCHEDULE_EVENT_LIMIT.

fn heartbeat_worker<'life0, 'life1, 'async_trait>( &'life0 self, w: &'life1 WorkerMeta, ) -> Pin<Box<dyn Future<Output = Result<Option<String>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Upsert the worker row and return any pending operator COMMAND for it — the control channel rides the heartbeat that is already happening (Faktory’s BEAT): “quiet” stops admitting, “resume” resumes, “restart” drains without a timeout, “terminate” performs a bounded shutdown, and “resign” releases singleton duties.

fn signal_worker<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, worker_id: &'life1 str, command: Option<&'life2 str>, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Set (or clear, with None) a worker’s pending command. It is delivered on the next heartbeat; runtimes clear it after applying it, then publish acknowledged state.

fn distinct_kinds<'life0, 'async_trait>( &'life0 self, limit: i64, ) -> Pin<Box<dyn Future<Output = Result<Vec<String>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

typed dispatch distinct kinds currently present among waiting jobs (bounded sample), so a runner can warn at startup about kinds no registered handler answers.

fn list_workers<'life0, 'async_trait>( &'life0 self, stale_after_ms: i64, ) -> Pin<Box<dyn Future<Output = Result<Vec<WorkerMeta>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Workers whose heartbeat is within stale_after_ms of store-now.

fn create_operation<'life0, 'life1, 'async_trait>( &'life0 self, req: &'life1 BulkRequest, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn get_operation<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<OperationStatus>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn run_pending_operations<'life0, 'async_trait>( &'life0 self, batch: i64, ) -> Pin<Box<dyn Future<Output = Result<u64, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Execute one bounded batch of each pending operation (run under a duty lease). Returns rows affected this sweep; an operation whose batch comes back short is marked completed.

fn promote_job<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

pending -> available. No timer or dependency watcher may perform this change.

fn delete_queue<'life0, 'life1, 'async_trait>( &'life0 self, queue: &'life1 str, force: bool, ) -> Pin<Box<dyn Future<Output = Result<Option<String>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Refuse a non-empty queue unless force; forced deletion is represented by a bounded async operation and therefore returns its operation id.

fn sample_queue_memory<'life0, 'async_trait>( &'life0 self, limit: u32, ) -> Pin<Box<dyn Future<Output = Result<u32, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Refresh bounded queue memory samples. Implementations must cap work to limit; ordinary queue reads only return the last stored sample.

impl Notifying for PgStore

fn wait_wakeup<'life0, 'life1, 'async_trait>( &'life0 self, queues: &'life1 [String], timeout: Duration, ) -> Pin<Box<dyn Future<Output = Result<Option<String>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Wait up to timeout for a hint that work may be available. An empty queues slice matches ANY queue (the UI’s one-subscription case, bounded live-control contract). Returns the waking queue’s name, or None on timeout. Wakeups may be spurious; callers admit either way when their poll timer expires — a wakeup only shortcuts it.

impl OutputInspect for PgStore

fn get_job_output<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<JobOutput>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Explicit output access. A previous attempt’s latest output may remain visible until the current holder replaces it; JobOutput::fence identifies its author.

impl OutputStore for PgStore

fn write_job_output<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, lease: &'life1 LeaseRef, output: &'life2 JobResult, ) -> Pin<Box<dyn Future<Output = Result<JobOutput, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

impl ProgressInspect for PgStore

fn get_job_progress<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<JobProgress>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

A previous attempt’s last report may remain until the current holder replaces it; JobProgress::fence makes that provenance explicit.

impl ProgressStore for PgStore

fn write_job_progress<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, lease: &'life1 LeaseRef, update: &'life2 ProgressUpdate, ) -> Pin<Box<dyn Future<Output = Result<JobProgress, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

impl ResultInspect for PgStore

fn get_job_result<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<JobResult>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Explicit result access. Implementations return None for a missing job or a job with no completed result; payload/list reads never include these bytes implicitly.

impl ResultStore for PgStore

fn ack_success_with_result<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, lease: &'life1 LeaseRef, logs: &'life2 [String], actual_weight: Option<u32>, result: &'life3 JobResult, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

impl Store for PgStore

fn admit<'life0, 'async_trait>( &'life0 self, req: AdmitRequest, ) -> Pin<Box<dyn Future<Output = Result<Vec<AdmissionUnit>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

The whole admission decision: policy + claim + lease, atomically, store-side.

fn ack_attempt_with_actual_weight<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, lease: &'life1 LeaseRef, outcome: Outcome, err: Option<&'life2 str>, delay_ms: Option<i64>, logs: &'life3 [String], actual_weight: Option<u32>, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Store::ack_attempt plus surveyed policy behavior cost reconciliation. Admission charges the envelope’s estimated weight; Some(actual) corrects that charge under the same fence and in the same atomic write as the state transition. Some(0) is a real full refund; None means the estimate was exact. The extra method is kept coarse on purpose: a separate reconcile call could commit after a rejected ack.

fn renew<'life0, 'life1, 'async_trait>( &'life0 self, leases: &'life1 [LeaseRef], lease: Duration, ) -> Pin<Box<dyn Future<Output = Result<Vec<String>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Extend leases; return the job ids of the leases that were LOST. A worker that lost a lease must be able to stop — silently succeeding here is how asynq stranded jobs in ACTIVE since 2022.

fn enqueue<'life0, 'life1, 'async_trait>( &'life0 self, batch: &'life1 [Envelope], ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn checkpoint<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, lease: &'life1 LeaseRef, cp: &'life2 Checkpoint, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

step replay persist step progress, fence-verified: the write succeeds only while the caller still holds the lease, so it doubles as the step boundary’s lease check. LeaseRejected here means STOP — do not run the next step’s side effects. Durable BEFORE the step runs, never after the worker returns (River’s mistake).

fn reclaim_expired<'life0, 'async_trait>( &'life0 self, limit: i64, ) -> Pin<Box<dyn Future<Output = Result<Vec<Reclaimed>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

lease fencing/crash quarantine the lease reclaimer’s sweep. An expired lease is Outcome::LeaseLost, NEVER Retry: it increments crash_attempt and leaves attempt alone. At the crash limit the job parks in quarantined and its fingerprint is registered. Safe under contention; run it via a duty lease to avoid redundant sweeps.

fn promote_due<'life0, 'async_trait>( &'life0 self, limit: i64, ) -> Pin<Box<dyn Future<Output = Result<u64, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

The schedule_due/backoff_due sweep: due scheduled and retryable jobs become available. Returns how many were promoted.

fn evict_retained<'life0, 'async_trait>( &'life0 self, limit: i64, ) -> Pin<Box<dyn Future<Output = Result<u64, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

retention and eviction contract the retention sweep: TERMINAL jobs whose finalized_at_ms + retention_ms has lapsed are deleted (the transition table’s completed -> deleted by retention; retention_ms = 0 already deleted at ack time). quarantined is exempt — it parks VISIBLY until an operator acts, never silently expires. Bounded per call; run under the retention duty lease.

fn claim_duty<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, name: &'life1 str, holder: &'life2 str, lease: Duration, ) -> Pin<Box<dyn Future<Output = Result<bool, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

singleton duties claim (or renew) a singleton duty. Same compare-and-set as claiming a job, on store time — a skewed node cannot steal a duty early. false = someone else holds it; skip this tick, never block on it.

fn release_duty<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, name: &'life1 str, holder: &'life2 str, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

singleton duties step down by expiring the duty immediately, so takeover is fast. A no-match (not the holder) is fine — release is best-effort on shutdown.

fn caps(&self) -> Caps

fn as_transactional(&self) -> Option<&dyn Transactional>

runtime capability boundary Runtime capability upcast. None means genuinely unsupported — never a silent no-op, and never a config knob that does nothing.

fn as_result_store(&self) -> Option<&dyn ResultStore>

Optional capability surface for a success transition that records result bytes under the same fence. A backend implementing ResultStore returns it here.

fn as_output_store(&self) -> Option<&dyn OutputStore>

Optional capability for fence-verified mid-run output writes. Unlike a final result this does not transition the job; it succeeds only for the current running lease and returns store-stamped attempt/time metadata.

fn as_progress_store(&self) -> Option<&dyn ProgressStore>

Optional capability for operator-facing progress writes. Like mid-run output, this is a fenced write that does not transition the job.

fn as_inspect(&self) -> Option<&dyn Inspect>

control plane the inspection/control surface. Same rule as as_transactional.

fn as_notifying(&self) -> Option<&dyn Notifying>

push wakeups push wakeup. MySQL never has this (poll only); PgBouncer in transaction pooling breaks it, which is why poll-only remains a first-class mode.

fn ack<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, lease: &'life1 LeaseRef, outcome: Outcome, err: Option<&'life2 str>, delay_ms: Option<i64>, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, Self: 'async_trait,

Apply the transition table for outcome, write the error history, honour the fence. delay_ms: required for Snooze (must be > 0); for Retry it overrides the store’s default backoff (the retry-policy port computes it caller-side); ignored otherwise. LeaseLost is never acked — it is the reclaimer’s transition. Convenience over Store::ack_attempt with no logs.

fn ack_attempt<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, lease: &'life1 LeaseRef, outcome: Outcome, err: Option<&'life2 str>, delay_ms: Option<i64>, logs: &'life3 [String], ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, Self: 'async_trait,

Store::ack plus attempt-log contract per-attempt execution logs (River’s riverlog): captured handler log lines land INSIDE the attempt’s error-history entry, so the console shows why an attempt failed, not just that it did. Recorded for success / retry / skip / undecodable (a non-empty logs on success writes a success entry — the only time one exists); dropped for snooze / rate_limited / revoke, which by design record no attempt entry.

impl Transactional for PgStore

fn begin_tx<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Box<dyn TxHandle>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Open a store transaction for the dyn path (transactional API). Callers holding their own driver transaction wrap it instead (caller-owned transaction contract) — this is for code that only knows dyn Transactional, like [Job.Once]-style helpers.

fn commit_tx<'life0, 'async_trait>( &'life0 self, tx: Box<dyn TxHandle>, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn rollback_tx<'life0, 'async_trait>( &'life0 self, tx: Box<dyn TxHandle>, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn claim_effect<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, tx: &'life1 mut dyn TxHandle, key: &'life2 str, ) -> Pin<Box<dyn Future<Output = Result<bool, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

transactional effects claim an effect key inside the caller’s transaction. false means the key was already claimed by a COMMITTED transaction — the effect ran; skip the work. The claim commits (or vanishes) with everything else in the transaction, which is the entire mechanism behind at-most-once effects.

fn checkpoint_tx<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, tx: &'life1 mut dyn TxHandle, lease: &'life2 LeaseRef, cp: &'life3 Checkpoint, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

step replay × transactional effects write the checkpoint inside the caller’s transaction, fence-verified. This is what makes a step’s effects and its completion marker ONE commit: a step-scoped once claims {job}/{step}, does its writes, and records the step complete — atomically. A superseded holder fails here and everything rolls back.

fn enqueue_tx<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, tx: &'life1 mut dyn TxHandle, batch: &'life2 [Envelope], ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

fn complete_tx_with_actual_weight<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, tx: &'life1 mut dyn TxHandle, lease: &'life2 LeaseRef, actual_weight: Option<u32>, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Transactional completion with the same surveyed policy behavior post-hoc correction as ack. This exists separately because once completes inside the caller’s transaction; reconciling outside it could charge an effect whose fenced completion rolled back.

fn complete_tx<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, tx: &'life1 mut dyn TxHandle, lease: &'life2 LeaseRef, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, Self: Sync + 'async_trait,