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 denormaliser (pruned_completed_count + pruned_failed_count). Rows currently in done_entries or receipt_completion_batches that have not yet rolled up are not included. Strictly a lower bound; converges to the exact count when rotation prunes the live queue segment. (The name terminal is honest — this number counts completed, failed, and cancelled terminal facts with 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 pruned_failed_count_for_queue( &self, pool: &PgPool, queue: &str, ) -> Result<u64, AwaError>

Cumulative count of failed terminal rows pruned past the retention floor for a queue, summed over the queue’s queue_terminal_rollups priorities (and stripes, for striped queues). Monotonic — rollups never decrease. These rows no longer exist in done_entries and cannot be retried.

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>, u64), AwaError>

Retry jobs by id inside a single transaction. Duplicate ids are collapsed so each job is attempted at most once. Returns the retried rows together with the number of unique ids attempted: ids whose terminal row raced to another state or was pruned between the caller’s scan and this call are skipped, so attempted - retried.len() is the count of unique ids that were requested but not retried.

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 terminal history. Run this when:

• upgrading from a pre-#290 fleet, where in-flight binaries wrote terminal history without maintaining the counter,
• after any incident that may have left the counter inconsistent with terminal history, 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 {schema}.done_entries). Compact receipt completions remain counted directly from retained receipt_completion_batches until queue prune folds them into permanent rollups.

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 done_entries terminal-count deltas into queue_terminal_live_counts for sealed queue slots.

done_entries terminal inserts and deletes append signed rows into queue_terminal_count_deltas instead of updating the live counter on the user-facing hot path. Compact receipt completions are counted from retained batch rows directly. Exact reads sum folded live counts plus pending deltas and retained compact batches, 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, failed_retention: Duration, ) -> Result<PruneOutcome, AwaError>

Prune the oldest sealed queue ring slot.

failed_retention is the floor for failed terminal rows: rows with finalized_at inside the floor are re-homed into the live done_entries segment (still retryable) instead of being truncated with the slot. Duration::ZERO disables the floor and prunes every terminal row. Granularity is whole seconds, matching DLQ retention. The floor applies to failed only — cancelled is an explicit operator action and is always pruned.

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: the lease_claims_<next>, lease_claim_batches_<next>, lease_claim_closures_<next>, and lease_claim_closure_batches_<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 its lease_claims_<slot>, lease_claim_batches_<slot>, lease_claim_closures_<slot>, and lease_claim_closure_batches_<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. Proves every claim in the sealed partition has closure evidence.
4. LOCK TABLE ACCESS EXCLUSIVE on all claim-ring children with a short transaction-local lock_timeout (serialises with in-flight claim/complete/rescue writers; gives up under sustained contention).
5. Rechecks the slot is not the current one and that every claim still has closure evidence.
6. TRUNCATE all claim-ring children in a single statement.

The “every claim has closure evidence” 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.