Struct QueueStorage 

pub struct QueueStorage { /* private fields */ }

Segmented queue storage backend.

Design goals:

• append-only queue segments in a rotated ring
• append-only completion segments keyed back to the queue segment
• a separate, faster rotating lease ring so delete churn is bounded by the lease cycle rather than by queue retention
• hot mutable state restricted to queue cursors, narrow leases, and optional per-attempt runtime state only when needed

Implementations

impl QueueStorage

pub fn new(config: QueueStorageConfig) -> Result<Self, AwaError>

pub fn from_existing_schema(schema: impl Into<String>) -> Result<Self, AwaError>

pub fn schema(&self) -> &str

pub fn slot_count(&self) -> usize

pub fn queue_slot_count(&self) -> usize

pub fn lease_slot_count(&self) -> usize

pub fn claim_slot_count(&self) -> usize

pub fn queue_stripe_count(&self) -> usize

pub fn lease_claim_receipts(&self) -> bool

pub fn ready_child_relname(&self, slot: usize) -> String

pub fn done_child_relname(&self, slot: usize) -> String

pub fn leases_relname(&self) -> &'static str

pub fn lease_claims_relname(&self) -> &'static str

pub fn lease_claim_closures_relname(&self) -> &'static str

pub fn leases_child_relname(&self, slot: usize) -> String

pub fn attempt_state_relname(&self) -> &'static str

pub async fn active_schema(pool: &PgPool) -> Result<Option<String>, AwaError>

pub async fn active_schema_in_tx( tx: &mut Transaction<'_, Postgres>, ) -> Result<Option<String>, AwaError>

Transaction-aware variant of Self::active_schema — read the active queue-storage schema name inside the caller’s transaction rather than acquiring a separate pool connection.

pub async fn prepare_schema(&self, pool: &PgPool) -> Result<(), AwaError>

Idempotently install the queue-storage substrate (schema, tables, indexes, functions) for this configuration.

Runs entirely inside one transaction on a single pooled connection, guarded by pg_advisory_xact_lock so concurrent worker startups serialize cleanly rather than fighting over pool slots. The lock auto-releases on COMMIT/ROLLBACK.

Note on index builds: the CREATE INDEX IF NOT EXISTS calls below are not CONCURRENTLY (Postgres bans CONCURRENTLY inside a transaction). On a fresh install that’s a no-op cost. On a redeploy against a database that already has rows in the queue tables, each fresh-index build takes ShareUpdateExclusiveLock on its table for the duration of the build, and concurrent writers to that table queue up. For typical operator scenarios this is bounded by the IF NOT EXISTS guard — only the new indexes added by a runtime upgrade get built; the rest are no-ops.

pub async fn activate_backend(&self, pool: &PgPool) -> Result<(), AwaError>

pub async fn install(&self, pool: &PgPool) -> Result<(), AwaError>

pub async fn reset(&self, pool: &PgPool) -> Result<(), AwaError>

pub async fn enqueue_batch( &self, pool: &PgPool, queue: &str, priority: i16, count: i64, ) -> Result<i64, AwaError>

pub async fn enqueue_params_batch( &self, pool: &PgPool, jobs: &[InsertParams], ) -> Result<usize, AwaError>

pub async fn enqueue_params_copy( &self, pool: &PgPool, jobs: &[InsertParams], ) -> Result<usize, AwaError>

Enqueue prepared jobs into queue storage using PostgreSQL COPY.

This follows the same preparation, queue striping, lane sequencing, uniqueness, and notification semantics as Self::enqueue_params_batch, but streams materialized rows directly into ready_entries and deferred_jobs instead of building multi-row INSERT statements.

pub async fn claim_batch( &self, pool: &PgPool, queue: &str, max_batch: i64, ) -> Result<Vec<ClaimedEntry>, AwaError>

pub async fn claim_runtime_batch( &self, pool: &PgPool, queue: &str, max_batch: i64, deadline_duration: Duration, ) -> Result<Vec<ClaimedRuntimeJob>, AwaError>

pub async fn claim_runtime_batch_with_aging( &self, pool: &PgPool, queue: &str, max_batch: i64, deadline_duration: Duration, aging_interval: Duration, ) -> Result<Vec<ClaimedRuntimeJob>, AwaError>

pub async fn acquire_queue_claimer( &self, pool: &PgPool, queue: &str, instance_id: Uuid, max_claimers: i16, lease_ttl: Duration, idle_threshold: Duration, ) -> Result<Option<QueueClaimerLease>, AwaError>

pub async fn mark_queue_claimer_active( &self, pool: &PgPool, queue: &str, instance_id: Uuid, lease: QueueClaimerLease, lease_ttl: Duration, ) -> Result<bool, AwaError>

pub async fn claim_runtime_batch_with_aging_for_instance( &self, pool: &PgPool, queue: &str, max_batch: i64, deadline_duration: Duration, aging_interval: Duration, instance_id: Uuid, max_claimers: i16, lease_ttl: Duration, idle_threshold: Duration, ) -> Result<Vec<ClaimedRuntimeJob>, AwaError>

pub async fn claim_job_batch( &self, pool: &PgPool, queue: &str, max_batch: i64, deadline_duration: Duration, ) -> Result<Vec<JobRow>, AwaError>

pub async fn complete_batch( &self, pool: &PgPool, claimed: &[ClaimedEntry], ) -> Result<usize, AwaError>

pub async fn complete_claimed_batch( &self, pool: &PgPool, claimed: &[ClaimedEntry], ) -> Result<Vec<(i64, i64)>, AwaError>

pub async fn complete_runtime_batch( &self, pool: &PgPool, claimed: &[ClaimedRuntimeJob], ) -> Result<Vec<(i64, i64)>, AwaError>

pub async fn complete_runtime_batch_slow_in_tx( &self, tx: &mut Transaction<'_, Postgres>, claimed: &[ClaimedRuntimeJob], ) -> Result<Vec<(i64, i64)>, AwaError>

Same as Self::complete_runtime_batch_slow but runs on the caller’s transaction so additional writes — for example ADR-029 follow-up job inserts — can join the same commit. The caller is responsible for commit() / rollback(). Handles both receipt-claimed and materialised leases.

pub async fn complete_job_batch_by_id( &self, pool: &PgPool, completions: &[(i64, i64)], ) -> Result<Vec<(i64, i64)>, AwaError>

pub async fn complete_job_batch_by_id_in_tx( &self, tx: &mut Transaction<'_, Postgres>, completions: &[(i64, i64)], ) -> Result<Vec<(i64, i64)>, AwaError>

Same as Self::complete_job_batch_by_id but runs on the caller’s transaction so additional writes — for example ADR-029 follow-up job inserts — can join the same commit. The caller is responsible for commit() / rollback().

pub async fn queue_counts( &self, pool: &PgPool, queue: &str, ) -> Result<QueueCounts, AwaError>

pub async fn queue_counts_fast( &self, pool: &PgPool, queue: &str, ) -> Result<QueueCounts, AwaError>

Index-only queue depth probe — for observability / depth-target throttling. Returns the same shape as Self::queue_counts but skips the table scans that Self::queue_counts_exact needs for exact terminal accounting:

• available is the same as the dispatcher’s claim signal: sum(GREATEST(enqueue_seq - claim_seq, 0)) over the shard head tables. No scan of ready_entries. This is an upper bound: admin DELETEs, committed gaps, and uncommitted enqueue reservations can leave the enqueue sequence ahead of the actual ready row count. Acceptable for depth-target throttling and dashboards; not suitable for exact billing-style counts.
• running matches Self::queue_counts’s strict definition: leases.state = 'running' only. Receipt-plane claims that have not yet materialised a lease row are omitted (the exact path catches them via the lease_claims anti-join, but that anti-join is what this fast variant exists to avoid). waiting_external is not included — it’s reported as part of admin’s parked-callback view, not running.
• terminal is read from the persisted queue_terminal_rollups.pruned_completed_count denormaliser. Rows currently in done_entries that have not yet rolled up are not included. Strictly a lower bound; converges to the exact count when rotation prunes the live done_entries segment. (The name terminal is honest — this number counts completed, failed, and cancelled rows in done_entries, the same semantics as Self::queue_counts_exact; renamed from completed in #290.)

All three counters are O(num shards) lookups against small head tables and leases index probes. Use this for high-cadence pollers (admin dashboards, depth-target producers, soak observability); use Self::queue_counts for admin tooling that needs the exact terminal count.

pub async fn retry_job( &self, pool: &PgPool, job_id: i64, ) -> Result<Option<JobRow>, AwaError>

pub async fn retry_jobs_by_ids( &self, pool: &PgPool, ids: &[i64], ) -> Result<Vec<JobRow>, AwaError>

pub async fn cancel_job( &self, pool: &PgPool, job_id: i64, ) -> Result<Option<JobRow>, AwaError>

pub async fn cancel_jobs_by_ids( &self, pool: &PgPool, ids: &[i64], ) -> Result<Vec<JobRow>, AwaError>

pub async fn set_priority( &self, pool: &PgPool, job_id: i64, priority: i16, ) -> Result<bool, AwaError>

pub async fn set_priority_tx<'a>( &self, tx: &mut Transaction<'a, Postgres>, job_id: i64, priority: i16, ) -> Result<bool, AwaError>

pub async fn move_queue( &self, pool: &PgPool, job_id: i64, queue: &str, priority: Option<i16>, ) -> Result<bool, AwaError>

pub async fn move_queue_tx<'a>( &self, tx: &mut Transaction<'a, Postgres>, job_id: i64, queue: &str, priority: Option<i16>, ) -> Result<bool, AwaError>

pub async fn age_waiting_priorities( &self, pool: &PgPool, aging_interval: Duration, limit: i64, ) -> Result<Vec<i64>, AwaError>

pub async fn load_job( &self, pool: &PgPool, job_id: i64, ) -> Result<Option<JobRow>, AwaError>

pub async fn register_callback( &self, pool: &PgPool, job_id: i64, run_lease: i64, timeout: Duration, ) -> Result<Uuid, AwaError>

pub async fn register_callback_with_config( &self, pool: &PgPool, job_id: i64, run_lease: i64, timeout: Duration, config: &CallbackConfig, ) -> Result<Uuid, AwaError>

pub async fn cancel_callback( &self, pool: &PgPool, job_id: i64, run_lease: i64, ) -> Result<bool, AwaError>

pub async fn load_active_lease_in_tx( &self, tx: &mut Transaction<'_, Postgres>, job_id: i64, run_lease: i64, ) -> Result<Option<JobRow>, AwaError>

Load the currently-active lease row for job_id (running or waiting_external) inside a caller-owned transaction. Used by ADR-029 helpers that need the post-park snapshot — including the callback_id and callback_timeout_at written by register_callback() — without leaving the transaction that just performed the parking UPDATE.

pub async fn enter_callback_wait( &self, pool: &PgPool, job_id: i64, run_lease: i64, callback_id: Uuid, ) -> Result<bool, AwaError>

pub async fn enter_callback_wait_in_tx( &self, tx: &mut Transaction<'_, Postgres>, job_id: i64, run_lease: i64, callback_id: Uuid, ) -> Result<bool, AwaError>

Transaction-aware variant of Self::enter_callback_wait (ADR-029). Returns whether the row transitioned to waiting_external.

pub async fn check_callback_state( &self, pool: &PgPool, job_id: i64, callback_id: Uuid, ) -> Result<CallbackPollResult, AwaError>

pub async fn callback_job( &self, pool: &PgPool, callback_id: Uuid, run_lease: Option<i64>, ) -> Result<Option<JobRow>, AwaError>

pub async fn callback_job_in_tx( &self, tx: &mut Transaction<'_, Postgres>, callback_id: Uuid, run_lease: Option<i64>, for_update: bool, ) -> Result<Option<JobRow>, AwaError>

Transaction-aware variant of Self::callback_job (ADR-029). When for_update is true the join’s lease row is locked with FOR UPDATE OF lease, mirroring the canonical resolve_callback lookup that takes a row lock on awa.jobs_hot before evaluating the callback policy and committing the resulting transition.

pub async fn complete_external( &self, pool: &PgPool, callback_id: Uuid, payload: Option<Value>, run_lease: Option<i64>, resume: bool, ) -> Result<JobRow, AwaError>

pub async fn complete_external_in_tx( &self, tx: &mut Transaction<'_, Postgres>, callback_id: Uuid, payload: Option<Value>, run_lease: Option<i64>, resume: bool, ) -> Result<JobRow, AwaError>

Transaction-aware variant of Self::complete_external (ADR-029). The non-resume path returns the post-completion JobRow directly from the done_row insert. The resume path returns the parked-row snapshot reloaded inside the same transaction via Self::load_active_lease_in_tx — i.e. it does not leave the caller’s transaction to refresh state.

pub async fn fail_external( &self, pool: &PgPool, callback_id: Uuid, error: &str, run_lease: Option<i64>, ) -> Result<JobRow, AwaError>

pub async fn fail_external_with_error_entry( &self, pool: &PgPool, callback_id: Uuid, error_entry: Value, run_lease: Option<i64>, ) -> Result<JobRow, AwaError>

pub async fn fail_external_with_error_entry_in_tx( &self, tx: &mut Transaction<'_, Postgres>, callback_id: Uuid, error_entry: Value, run_lease: Option<i64>, ) -> Result<JobRow, AwaError>

Transaction-aware variant of Self::fail_external_with_error_entry (ADR-029).

pub async fn retry_external( &self, pool: &PgPool, callback_id: Uuid, run_lease: Option<i64>, ) -> Result<JobRow, AwaError>

pub async fn retry_external_in_tx( &self, tx: &mut Transaction<'_, Postgres>, callback_id: Uuid, run_lease: Option<i64>, ) -> Result<JobRow, AwaError>

Transaction-aware variant of Self::retry_external (ADR-029).

pub async fn heartbeat_callback( &self, pool: &PgPool, callback_id: Uuid, timeout: Duration, ) -> Result<JobRow, AwaError>

pub async fn flush_progress( &self, pool: &PgPool, job_id: i64, run_lease: i64, progress: Value, ) -> Result<(), AwaError>

pub async fn heartbeat_batch( &self, pool: &PgPool, jobs: &[(i64, i64)], ) -> Result<usize, AwaError>

pub async fn heartbeat_progress_batch( &self, pool: &PgPool, jobs: &[(i64, i64, Value)], ) -> Result<usize, AwaError>

pub async fn retry_after( &self, pool: &PgPool, job_id: i64, run_lease: i64, retry_after: Duration, progress: Option<Value>, ) -> Result<Option<JobRow>, AwaError>

pub async fn retry_after_in_tx( &self, tx: &mut Transaction<'_, Postgres>, job_id: i64, run_lease: i64, retry_after: Duration, progress: Option<Value>, ) -> Result<Option<JobRow>, AwaError>

Transaction-aware variant of Self::retry_after. Caller owns the transaction lifecycle so the move can commit atomically alongside follow-up INSERTs (ADR-029).

pub async fn snooze( &self, pool: &PgPool, job_id: i64, run_lease: i64, snooze_for: Duration, progress: Option<Value>, ) -> Result<Option<JobRow>, AwaError>

pub async fn cancel_running( &self, pool: &PgPool, job_id: i64, run_lease: i64, reason: &str, progress: Option<Value>, ) -> Result<Option<JobRow>, AwaError>

pub async fn cancel_running_in_tx( &self, tx: &mut Transaction<'_, Postgres>, job_id: i64, run_lease: i64, reason: &str, progress: Option<Value>, ) -> Result<Option<JobRow>, AwaError>

Transaction-aware variant of Self::cancel_running (ADR-029).

pub async fn fail_terminal( &self, pool: &PgPool, job_id: i64, run_lease: i64, error: &str, progress: Option<Value>, ) -> Result<Option<JobRow>, AwaError>

pub async fn fail_terminal_in_tx( &self, tx: &mut Transaction<'_, Postgres>, job_id: i64, run_lease: i64, error: &str, progress: Option<Value>, ) -> Result<Option<JobRow>, AwaError>

Transaction-aware variant of Self::fail_terminal (ADR-029).

pub async fn fail_to_dlq( &self, pool: &PgPool, job_id: i64, run_lease: i64, dlq_reason: &str, error: &str, progress: Option<Value>, ) -> Result<Option<JobRow>, AwaError>

pub async fn fail_to_dlq_in_tx( &self, tx: &mut Transaction<'_, Postgres>, job_id: i64, run_lease: i64, dlq_reason: &str, error: &str, progress: Option<Value>, ) -> Result<Option<JobRow>, AwaError>

Transaction-aware variant of Self::fail_to_dlq (ADR-029).

pub async fn move_failed_to_dlq( &self, pool: &PgPool, job_id: i64, dlq_reason: &str, ) -> Result<Option<JobRow>, AwaError>

pub async fn bulk_move_failed_to_dlq( &self, pool: &PgPool, kind: Option<&str>, queue: Option<&str>, dlq_reason: &str, ) -> Result<u64, AwaError>

pub async fn retry_from_dlq( &self, pool: &PgPool, job_id: i64, opts: &RetryFromDlqOpts, ) -> Result<Option<JobRow>, AwaError>

pub async fn bulk_retry_from_dlq( &self, pool: &PgPool, filter: &ListDlqFilter, ) -> Result<u64, AwaError>

pub async fn discard_failed_by_kind( &self, pool: &PgPool, kind: &str, ) -> Result<u64, AwaError>

pub async fn fail_retryable( &self, pool: &PgPool, job_id: i64, run_lease: i64, error: &str, progress: Option<Value>, ) -> Result<Option<JobRow>, AwaError>

pub async fn fail_retryable_in_tx( &self, tx: &mut Transaction<'_, Postgres>, job_id: i64, run_lease: i64, error: &str, progress: Option<Value>, ) -> Result<Option<JobRow>, AwaError>

Transaction-aware variant of Self::fail_retryable (ADR-029).

pub async fn rescue_stale_heartbeats( &self, pool: &PgPool, staleness: Duration, ) -> Result<Vec<JobRow>, AwaError>

pub async fn rescue_expired_deadlines( &self, pool: &PgPool, ) -> Result<Vec<JobRow>, AwaError>

pub async fn rescue_expired_callbacks( &self, pool: &PgPool, ) -> Result<Vec<JobRow>, AwaError>

pub async fn promote_due( &self, pool: &PgPool, state: JobState, batch_size: i64, ) -> Result<usize, AwaError>

pub async fn rotate(&self, pool: &PgPool) -> Result<RotateOutcome, AwaError>

pub async fn rotate_leases( &self, pool: &PgPool, ) -> Result<RotateOutcome, AwaError>

pub async fn rebuild_terminal_counters( &self, pool: &PgPool, ) -> Result<i64, AwaError>

Rebuild terminal counters from scratch by truncating folded live counts and pending deltas, then re-aggregating done_entries. Run this when:

• upgrading from a pre-#290 fleet, where in-flight binaries wrote done_entries without maintaining the counter,
• after any incident that may have left the counter inconsistent with done_entries, or
• as a routine drift-recovery step before relying on counter-fed reads for billing-grade accuracy.

The rebuild is wrapped in an advisory lock so concurrent writers don’t interleave new inserts mid-rebuild. The lock key is scoped to the queue-storage schema, so other schemas / other operations are unaffected. Writers that hit the lock will block briefly rather than fail.

Operator note: this is best run on a quiesced fleet (workers paused or fully drained). Concurrent inserts during the rebuild will block on the lock; long-held locks can stall the fleet. The rebuild itself is O(rows in done_entries).

pub async fn terminal_counter_trusted( &self, pool: &PgPool, ) -> Result<bool, AwaError>

Check whether the live terminal counter has been marked trusted for exact reads. Returns true for fresh installs (the trust marker is auto-set by prepare_schema when done_entries is empty) and after a successful Self::rebuild_terminal_counters; returns false on an existing install that has not yet been rebuilt under the new counter-aware code path.

queue_counts_exact consults this to decide between the counter-fed fast path and the scan-based fallback. Single-row PK fetch; negligible cost per call.

pub async fn rollup_terminal_count_deltas( &self, pool: &PgPool, max_slots: usize, ) -> Result<TerminalDeltaRollupOutcome, AwaError>

Fold append-only terminal-count deltas into queue_terminal_live_counts for sealed queue slots.

Completions and terminal deletes append signed rows into queue_terminal_count_deltas instead of updating the live counter on the user-facing hot path. Exact reads sum folded live counts plus pending deltas, so this rollup can run asynchronously. It intentionally skips the current queue slot and any slot with active leases or open receipt claims; that keeps rollup away from the segment receiving hot completions and avoids racing future terminal deltas for the same ready generation. Rollup also stands down while another backend pins the MVCC horizon, leaving the append-only deltas visible to exact reads until the horizon clears.

pub async fn prune_oldest( &self, pool: &PgPool, ) -> Result<PruneOutcome, AwaError>

pub async fn prune_oldest_leases( &self, pool: &PgPool, ) -> Result<PruneOutcome, AwaError>

pub async fn vacuum_leases(&self, pool: &PgPool) -> Result<(), AwaError>

pub async fn rotate_claims( &self, pool: &PgPool, ) -> Result<RotateOutcome, AwaError>

ADR-023 claim-ring rotation. Parallel of rotate_leases.

Advances claim_ring_state.current_slot via compare-and-swap. Before flipping the cursor the target partition must be drained: both the lease_claims_<next> and lease_claim_closures_<next> child tables must be empty. This is what the rotate → prune → rotate ring invariant requires — we only hand out a slot to new claims when a prior prune has truncated it.

pub async fn prune_oldest_claims( &self, pool: &PgPool, ) -> Result<PruneOutcome, AwaError>

ADR-023 claim-ring prune. Parallel of prune_oldest_leases.

Reclaims the oldest initialized (sealed) claim-ring slot by TRUNCATE-ing both its lease_claims_<slot> and lease_claim_closures_<slot> children. Takes the full ADR-023 lock sequence:

1. FOR UPDATE on claim_ring_state (serialises with rotate).
2. FOR UPDATE on the target claim_ring_slots row.
3. SET LOCAL lock_timeout = '50ms' then LOCK TABLE ACCESS EXCLUSIVE on both children (serialises with in-flight claim/complete/rescue writers; gives up gracefully under contention).
4. Verifies the slot is not the current one, and that every claim in the partition has a matching closure row.
5. TRUNCATE both children in a single statement.

The “every claim has a closure” precondition is what ADR-023 calls PartitionTruncateSafety. If an open claim remains in the partition, prune returns SkippedActive and the claim has to drain by normal completion or be rescued by rescue_stale_receipt_claims_tx before prune will try again.

Trait Implementations

impl Debug for QueueStorage

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.