From 3dd83791ff0a268a54300273992d2fa3c86cf8c9 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Fri, 9 Oct 2026 08:44:59 +0000 Subject: [PATCH] =?UTF-8?q?Postgres=20engine:=20queues=20=E2=80=94=20Queue?= =?UTF-8?q?=20(validated=20constructor=20over=20the=20handle's=20QueueOpts?= =?UTF-8?q?=20stamps),=20the=20FOR=20UPDATE=20SKIP=20LOCKED=20claim=20(one?= =?UTF-8?q?=20statement,=20the=20pre-claim=20exhausted-reclaimable=20sweep?= =?UTF-8?q?=20in=20the=20atomic=20claim=20frame),=20JobHandle=20(one-shot?= =?UTF-8?q?=20ack/retry/fail=20in=20tx=20frames=20under=20the=20uniform=20?= =?UTF-8?q?validity=20predicate,=20absolute-reset=20heartbeat,=20engine-si?= =?UTF-8?q?de=20equal-jitter=20backoff),=20dead-letter=20moves=20transacti?= =?UTF-8?q?onal=20(the=20#133=20class=20excluded),=20worker-less=20ack=5Fb?= =?UTF-8?q?atch,=20unconditional=20cancel,=20dead-visible=20get=5Fjob,=20b?= =?UTF-8?q?oth-states+retention=20sweep,=20best-effort=20queue-channel=20w?= =?UTF-8?q?ake,=20wake-driven=20claim=20loop=20pinned=20(task=20pg-engine-?= =?UTF-8?q?queues)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- alkstore-postgres/src/lib.rs | 1 + alkstore-postgres/src/queue.rs | 999 ++++++++++++++ alkstore-postgres/src/store.rs | 23 +- alkstore-postgres/src/store/open_tests.rs | 20 +- alkstore-postgres/src/store/queue_tests.rs | 1439 ++++++++++++++++++++ alkstore-postgres/src/tx.rs | 89 +- tasks/pg-engine-queues.md | 122 +- 7 files changed, 2640 insertions(+), 53 deletions(-) create mode 100644 alkstore-postgres/src/queue.rs create mode 100644 alkstore-postgres/src/store/queue_tests.rs diff --git a/alkstore-postgres/src/lib.rs b/alkstore-postgres/src/lib.rs index 2c357df..e750e94 100644 --- a/alkstore-postgres/src/lib.rs +++ b/alkstore-postgres/src/lib.rs @@ -35,6 +35,7 @@ mod forwarder; mod notify; mod opts; +mod queue; mod resolution; mod schema; mod seam; diff --git a/alkstore-postgres/src/queue.rs b/alkstore-postgres/src/queue.rs new file mode 100644 index 0000000..90828b5 --- /dev/null +++ b/alkstore-postgres/src/queue.rs @@ -0,0 +1,999 @@ +//! The queue mechanism's Postgres arm (`pg-engine-queues`) — the +//! auto-commit counterpart to the tx seam's enqueue; the re-derived +//! queue machinery (ADR-004/005's greenfield posture, the pg-boss +//! schema family as design reference) with ADR-010's full depth: +//! +//! - `queue(name, opts)` — the validated constructor (construction is +//! free: validation + a closed-store check — queues are names, not +//! registered objects, ADR-010 §3a); the handle carries its +//! [`QueueOpts`] stamps future enqueues resolve over. +//! - `enqueue(payload, opts)` — the one handle-carrying shape: stamps +//! resolve over **this handle's opts** (`EnqueueOpts::max_attempts` +//! the one per-job override — ADR-020 §3a), through the shared +//! resolution arithmetic (one owner per formula — the engine's own +//! `resolution.rs`; one clock read per enqueue, delay-over-`run_at`, +//! relative-`expires`), `encode_payload` at the seam, the INSERT (the +//! tx seam's shared row-write) on a pool connection — then **wake**: +//! `pg_notify` on the queue's wake channel (the +//! mechanism-name-is-the-channel realization; the stream task's twin +//! posture — best-effort, logged and swallowed: the durable row is +//! the truth). +//! - `claim_one` / `claim_batch` — exactly-once handout under +//! concurrency (ADR-010 §1): the `FOR UPDATE SKIP LOCKED` claim on +//! a pool connection, one statement — an `UPDATE … RETURNING` over a +//! `picked` CTE. Ordering `priority DESC, run_at ASC, id` +//! (enqueue order = id); `attempts += 1` per claim (a reclaim +//! consumes an attempt); the claim sets the row's deadline from the +//! job's own stamped `visibility_timeout_s`; the pre-claim +//! dead-letter sweep for exhausted reclaimables runs first (the +//! substrate's D-12 shape — a worker dying right after its last +//! allowed claim dead-letters at the next claim, not strands); +//! `worker_id` stamps the claimant column (consumer-local, +//! non-empty — ADR-019 §2). The **extent guard** (ADR-023 §2): +//! `n <= 0` → the empty `Vec` at trait-impl entry. Claims return +//! boxed [`JobHandle`]s (ops + row value), decoded through the one +//! 18-column decode owner ([`job_from_row`] — the tx reads' shared +//! shape). +//! - The [`JobHandle`] impl: `job()` (the claimed row value), +//! `ack`/`retry`/`fail` (one-shot, `self: Box` — the consuming +//! shape; `BoxedFuture<'static>`), `heartbeat(extend)` (repeatable, +//! absolute reset). The uniform validity predicate (ADR-010 §2 — +//! `processing` + unexpired claim deadline) is conjunct in every +//! op's SQL (`worker_id` keeps its conjunct on the single-row +//! forms — the D-31 delta removed the filter from the *batch* ack +//! only); refusals are `Ok(false)`, never errors. +//! - **The backoff curve — engine-side, one owner (ADR-010 §3)**: +//! `retry(err, None)` computes the equal-jitter exponential from the +//! job's stamped `backoff_base_s`: `delay ∈ [base·2^(a−1)/2, +//! base·2^(a−1)]`, capped at 1 hour; the attempt index `a` is the +//! row's `attempts` at the retry (the claim-row value — post-claim +//! count, reclaims included). `Some(d)` passes through. The jitter's +//! randomness source is std-only (the SQLite task's reused posture — +//! a fresh [`RandomState`] hash draw, no `rand` dep; flagged in the +//! task Notes). +//! - Dead-letter moves (retry-at-budget, fail, the pre-claim sweep, +//! the expiry sweep) ride one dead-move INSERT; dead rows are +//! `get_job`-visible (ADR-010 §4). The engine-triggered default +//! strings are the contract's pinned ones (`"max attempts +//! exceeded"`, `"failed"`, `"expired"`); caller strings cross +//! verbatim. +//! - `ack_batch(ids)` — the batch ack applies the uniform validity +//! predicate per id (no worker filter — the wave-3 D-31 delta's +//! shape is the contract's); count returned, partial success +//! ordinary (ADR-010 §1). +//! - `cancel(job_id)` — unconditional delete (pending or processing, +//! not an interrupt); `Ok(false)` when row-less. +//! - `get_job(job_id)` — dead-visible pure read on a pool connection +//! (the resolved values — not raw opts — are what the row shows, +//! ADR-020). +//! - `sweep_expired()` — both-states no-stranded-rows move (any +//! past-expiry row → dead with `"expired"`) + retention enforcement +//! (ADR-010 §5), single-statement atomic; returns the moved count. +//! No leader lock required. + +use std::collections::hash_map::RandomState; +use std::hash::BuildHasher; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::{SystemTime, UNIX_EPOCH}; + +use deadpool_postgres::Pool; +use serde_json::Value; + +use alkstore::{ + BoxedFuture, EnqueueOpts, Error, Job, JobHandle, JobState, Queue, QueueOpts, Result, + validate_local_name, +}; + +use deadpool_postgres::Object; + +use crate::resolution::{Stamps, now_unix}; +use crate::schema::{QualifiedTable, tables}; +use crate::seam::{database_error, pg_error, pool_error}; +use crate::tx::{encode_payload_bytes, enqueue_row}; + +/// The dead-letter default strings (row content, ADR-010 §4 / ADR-019 +/// §3's annotation), identical to the SQLite engine's — the +/// engine-default strings land identically on both engines (the +/// backlog row's wording). +pub(crate) const EXHAUSTED_ERROR: &str = "max attempts exceeded"; +pub(crate) const FAILED_ERROR: &str = "failed"; +pub(crate) const EXPIRED_ERROR: &str = "expired"; + +/// The backoff cap (ADR-010 §3 — a pinned contract constant, no knob). +const BACKOFF_CAP_S: i64 = 3600; + +/// The queue-scoped handle (see the module docs). +pub struct PgQueueHandle { + name: String, + opts: QueueOpts, + pool: Pool, + schema: String, + closed: Arc, +} + +impl std::fmt::Debug for PgQueueHandle { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("PgQueueHandle") + .field("name", &self.name) + .field("opts", &self.opts) + .finish_non_exhaustive() + } +} + +impl PgQueueHandle { + pub(crate) fn new( + name: String, + opts: QueueOpts, + pool: Pool, + schema: String, + closed: Arc, + ) -> Self { + Self { + name, + opts, + pool, + schema, + closed, + } + } + + /// The handle's opts, as the rows stamp them (the `max_attempts` + /// override resolves per enqueue over these). + fn base_stamps(&self) -> Stamps { + Stamps { + max_attempts: self.opts.max_attempts, + visibility_timeout_s: self.opts.visibility_timeout_s, + backoff_base_s: self.opts.backoff_base_s, + dead_letter_retention_s: self.opts.dead_letter_retention_s, + } + } +} + +/// The equal-jitter exponential backoff (ADR-010 §3 — engine-side, one +/// owner; the SQLite engine's twin computes identically and wave 5 +/// pins equivalence): `delay ∈ [base·2^(a−1)/2, base·2^(a−1)]` from +/// the job's stamped `backoff_base_s`, capped at **1 hour**. +/// +/// Integerized second-precision: the draw is uniform over the integers +/// in `[ceil(half), cap]` (inclusive both ends) — never below the +/// pinned lower bound, never above the cap. The attempt index `a` is +/// the row's `attempts` at the retry (post-claim count, reclaims +/// included); the exponent clamps at 30 so the cap arithmetic +/// saturates rather than overflows. +/// +/// The jitter draw is std-only (no `rand` dependency without a +/// decision — the ADR-005 dependency posture; the SQLite engine's +/// reused approach, flagged in the task Notes): a fresh [`RandomState`] +/// hasher (per-process-random seeds) over the row id, the attempt +/// count, and the subsecond clock spreads the draw uniformly over the +/// range. Hash quality is only asked to spread the draw; the pinned +/// property is the **range**, what the acceptance test pins +/// statistically. +pub(crate) fn backoff_delay_s(base_s: i64, attempts: i64, id: i64) -> i64 { + let exponent: u32 = (attempts.max(1) as u32).saturating_sub(1).min(30); + let doubling: i64 = base_s.saturating_mul(1i64 << exponent).max(1); + let cap = doubling.min(BACKOFF_CAP_S); + let half = ((cap + 1) / 2).max(1); + let span = (cap - half + 1).max(1); + let now_nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.subsec_nanos()) + .unwrap_or(0); + let draw = RandomState::new().hash_one((id, attempts, now_nanos)); + half + (draw % span as u64) as i64 +} + +/// Qualified table reference for this engine-owned schema. +fn table(schema: &str, name: &str) -> String { + QualifiedTable { schema, name }.to_string() +} + +/// The live-row select's column list (the uniform 18-column shape the +/// dead-table select mirrors with `last_error`/`died_at` in place of +/// the NULL casts) — one shape, shared with the tx reads. +fn live_columns() -> &'static str { + "id, queue, state, payload, priority, run_at, attempts, + max_attempts, worker_id, claimed_at, claim_expires_at, + created_at, expires_at, visibility_timeout_s, + backoff_base_s, dead_letter_retention_s, + NULL::text, NULL::bigint" +} + +/// The dead-row select's column list (the 18-column live shape with +/// `last_error`/`died_at` riding the NULL casts' slots). +fn dead_columns() -> &'static str { + "id, queue, 'dead' AS state, payload, priority, run_at, attempts, + max_attempts, worker_id, claimed_at, claim_expires_at, + created_at, expires_at, visibility_timeout_s, + backoff_base_s, dead_letter_retention_s, + last_error, died_at" +} + +/// The 14 columns a dead-move carries (the live row minus state, the +/// deadline, and the claimant-clearing arms — the move deletes the row +/// whole and re-inserts it with its history). +const DEAD_MOVE_COLUMNS: &str = "id, queue, payload, priority, run_at, attempts, + max_attempts, worker_id, claimed_at, + created_at, expires_at, visibility_timeout_s, + backoff_base_s, dead_letter_retention_s"; + +/// Decode the dead-move row (the [`DEAD_MOVE_COLUMNS`] shape) into the +/// dead table with `last_error`/`died_at` stamped. One owner: every +/// dead-letter move (retry-at-budget, `fail`, the pre-claim sweep, the +/// expiry sweep) rides it (ADR-010 §4). +async fn insert_dead_row_tx( + client: &tokio_postgres::Client, + schema: &str, + row: &tokio_postgres::Row, + last_error: &str, + died_at: i64, +) -> Result<()> { + let params: [&(dyn tokio_postgres::types::ToSql + Sync); 16] = [ + &row_get_i64(row, 0), + &row_get_string(row, 1), + &row_get_bytes(row, 2), + &row_get_i64(row, 3), + &row_get_i64(row, 4), + &row_get_i64(row, 5), + &row_get_i64(row, 6), + &row_get_opt_string(row, 7), + &row_get_opt_i64(row, 8), + &row_get_i64(row, 9), + &row_get_opt_i64(row, 10), + &row_get_i64(row, 11), + &row_get_i64(row, 12), + &row_get_opt_i64(row, 13), + &last_error.to_string(), + &died_at, + ]; + client + .execute( + &format!( + "INSERT INTO {dead} + (id, queue, payload, priority, run_at, attempts, + max_attempts, worker_id, claimed_at, created_at, + expires_at, visibility_timeout_s, backoff_base_s, + dead_letter_retention_s, last_error, died_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, + $11, $12, $13, $14, $15, $16)", + dead = table(schema, tables::DEAD) + ), + ¶ms[..], + ) + .await + .map_err(pg_error)?; + Ok(()) +} + +// The dead-move row's typed column reads (the RETURNING row's decode — +// the columns arrive typed from the live table; these helpers pull +// them out as owned values so they outlive the row borrow across the +// INSERT's params borrow). + +fn row_get_i64(row: &tokio_postgres::Row, idx: usize) -> i64 { + row.get(idx) +} + +fn row_get_opt_i64(row: &tokio_postgres::Row, idx: usize) -> Option { + row.get(idx) +} + +fn row_get_string(row: &tokio_postgres::Row, idx: usize) -> String { + row.get(idx) +} + +fn row_get_opt_string(row: &tokio_postgres::Row, idx: usize) -> Option { + row.get(idx) +} + +fn row_get_bytes(row: &tokio_postgres::Row, idx: usize) -> Vec { + row.get(idx) +} + +/// The dead-move / claim transaction frame on one pooled object: an +/// explicit `BEGIN` … body … `COMMIT`/`ROLLBACK` via `batch_execute` +/// on the held client (the seam's pooled-object posture +/// ([`crate::tx::PgTxHandle`]) scoped to one op — the sqlite +/// substrate's savepoint frame's pg twin; every statement the body +/// runs through the client participates in the open frame). The arms: +/// +/// - a failed `BEGIN` discards the object (unknowable state — +/// [`Object::take`]; the pool accounting stays exact); +/// - the body's error arm rolls back explicitly and re-pools (a failed +/// `ROLLBACK` discards the object too); +/// - a failed `COMMIT` discards the object (the wave-3 review's +/// finding (c), pre-empted by the seam). +async fn in_tx(object: Object, body: F) -> Result +where + T: Send, + F: FnOnce(&tokio_postgres::Client) -> BoxedFuture<'_, Result>, +{ + let client: &tokio_postgres::Client = &object; + if let Err(e) = client.batch_execute("BEGIN").await { + drop(Object::take(object)); + return Err(pg_error(e)); + } + match body(client).await { + Ok(value) => match client.batch_execute("COMMIT").await { + Ok(()) => { + drop(object); + Ok(value) + } + Err(e) => { + drop(Object::take(object)); + Err(pg_error(e)) + } + }, + Err(e) => { + if client.batch_execute("ROLLBACK").await.is_err() { + drop(Object::take(object)); + } else { + drop(object); + } + Err(e) + } + } +} + +/// The pre-claim dead-letter sweep for exhausted reclaimables (the +/// substrate's D-12 shape — re-owned as owned pg SQL): rows at-or-past +/// budget that the claim predicate would otherwise reach are moved to +/// dead first, so the claim statement never hands out a row whose +/// next ack path is already exhausted. Same transaction frame as the +/// move paths (the sqlite substrate's savepoint-guarded #133 defect +/// class is excluded here the same way: a failing dead INSERT rolls +/// the whole move back — no stranded-in-neither-table state). +async fn sweep_exhausted_reclaimables_tx( + client: &tokio_postgres::Client, + schema: &str, + queue: &str, + now: i64, +) -> Result<()> { + let sql = format!( + "DELETE FROM {job} + WHERE queue = $1 + AND attempts >= max_attempts + AND (expires_at IS NULL OR expires_at > $2) + AND ((state = 'pending' AND run_at <= $2) + OR (state = 'processing' AND claim_expires_at < $2)) + RETURNING {DEAD_MOVE_COLUMNS}", + job = table(schema, tables::JOB) + ); + let rows = client + .query(&sql, &[&queue, &now]) + .await + .map_err(pg_error)?; + for row in &rows { + insert_dead_row_tx(client, schema, row, EXHAUSTED_ERROR, now).await?; + } + Ok(()) +} + +/// The claim statement: one `UPDATE … RETURNING` over the claimable +/// predicate under `FOR UPDATE SKIP LOCKED` (the exactly-once handout, +/// ADR-010 §1). The claim stamps `processing`, the claimant, the +/// deadline-from-the-job's-own-stamp, `claimed_at`, and +/// `attempts += 1`. +/// +/// Parameter positions: $1 worker_id, $2 now, $3 queue, $4 extent n. +fn claim_stmt_sql(schema: &str) -> String { + format!( + "UPDATE {job} SET + state = 'processing', + worker_id = $1, + claim_expires_at = $2::bigint + visibility_timeout_s, + claimed_at = $2, + attempts = attempts + 1 + WHERE id IN ( + SELECT id FROM {job} + WHERE queue = $3 + AND state IN ('pending', 'processing') + AND attempts < max_attempts + AND (expires_at IS NULL OR expires_at > $2) + AND ((state = 'pending' AND run_at <= $2) + OR (state = 'processing' AND claim_expires_at < $2)) + ORDER BY priority DESC, run_at ASC, id ASC + FOR UPDATE SKIP LOCKED + LIMIT $4 + ) + RETURNING {live}", + job = table(schema, tables::JOB), + live = live_columns(), + ) +} + +/// Decode an 18-column job row into the contract value +/// ([`Job::from_row`] — the `#[doc(hidden)]` constructor's field list +/// of record, ADR-021 §2). Malformed rows map to `Error::Codec` (the +/// decode-side posture guard). One decode owner: the tx reads and this +/// module's claim/get rows share the shape. +pub(crate) fn job_from_row(row: &tokio_postgres::Row) -> Result { + let state = match row.get::<_, Option<&str>>(2) { + Some("pending") => JobState::Pending, + Some("processing") => JobState::Processing, + Some("dead") => JobState::Dead, + other => { + return Err(Error::Codec(format!( + "job row state must be pending|processing|dead, got {other:?}" + ))); + } + }; + let payload: Option> = row + .try_get(3) + .map_err(|e| Error::Codec(format!("job row payload must be bytea: {e}")))?; + Ok(Job::from_row( + row.get(0), + row.get(1), + state, + payload.unwrap_or_default(), + row.get(4), + row.get(5), + row.get(6), + row.get(7), + row.get(8), + row.get(9), + row.get(10), + row.get(11), + row.get(12), + row.get(13), + row.get(14), + row.get(15), + row.get(16), + row.get(17), + )) +} + +/// The closed-store error shape (the engine-wide posture — one +/// message). +fn closed_error() -> Error { + database_error("the store is closed") +} + +impl Queue for PgQueueHandle { + fn name(&self) -> &str { + &self.name + } + + fn enqueue<'a>(&'a self, payload: Value, opts: EnqueueOpts) -> BoxedFuture<'a, Result> { + if self.closed.load(Ordering::Acquire) { + return Box::pin(async { Err(closed_error()) }); + } + let name = self.name.clone(); + let base = self.base_stamps(); + let pool = self.pool.clone(); + let schema = self.schema.clone(); + Box::pin(async move { + // The seam: encode_payload's typed `Codec` error `?`-ed + // before any round trip (ADR-023 §1). + let bytes = encode_payload_bytes(&payload)?; + let conn = pool.get().await.map_err(pool_error)?; + // The shared resolution arithmetic — one clock read per + // enqueue, delay-over-`run_at`, relative-`expires` — and + // the INSERT (the tx seam's shared row-write; the + // `max_attempts` override rides `stamps_with_override` + // inside `enqueue_row`). + let id = enqueue_row(&conn, &schema, &name, bytes, &opts, base).await?; + // Then wake: pg_notify on the queue's wake channel — the + // mechanism-name-is-the-channel realization (the stream + // task's twin posture). Best-effort on this auto-commit + // path: the durable row is the truth; a failed wake is + // logged and swallowed and the consumer's re-poll safety + // net covers the gap — no wake may ever fail (or stall) a + // committed enqueue. + if let Err(wake_err) = conn.query_one("SELECT pg_notify($1, '')", &[&name]).await { + eprintln!( + "alkstore-postgres: queue enqueue wake failed for {name:?} \ + (the durable row landed; the re-poll safety net covers the gap): {wake_err}" + ); + } + Ok(id) + }) + } + + fn claim_one<'a>( + &'a self, + worker_id: &str, + ) -> BoxedFuture<'a, Result>>> { + let worker_id = worker_id.to_string(); + Box::pin(async move { Ok(self.claim_batch(&worker_id, 1).await?.into_iter().next()) }) + } + + fn claim_batch<'a>( + &'a self, + worker_id: &str, + n: i64, + ) -> BoxedFuture<'a, Result>>> { + if let Err(e) = validate_local_name(worker_id) { + return Box::pin(async move { Err(e) }); + } + if self.closed.load(Ordering::Acquire) { + return Box::pin(async { Err(closed_error()) }); + } + // The extent guard (ADR-023 §2): `n <= 0` claims nothing — the + // empty `Vec`, a value, never an error — at trait-impl entry, + // before any round trip. + if n <= 0 { + return Box::pin(async { Ok(Vec::new()) }); + } + let name = self.name.clone(); + let pool = self.pool.clone(); + let schema = self.schema.clone(); + let worker = worker_id.to_string(); + let closed = self.closed.clone(); + Box::pin(async move { + let conn = pool.get().await.map_err(pool_error)?; + // The claim's statement frame: the pre-claim sweep (the + // D-12 shape — at-budget rows the claim predicate would + // reach dead-letter first, no row stranded between the + // two statements' failure arms) + the claim (one + // `UPDATE … SKIP LOCKED` — exactly-once handout under + // concurrency; `now` is the one clock read the frame's + // stamping shares), both committing atomically. + in_tx(conn, |client| { + Box::pin(async move { + let now = now_unix(); + sweep_exhausted_reclaimables_tx(client, &schema, &name, now).await?; + let rows = client + .query(&claim_stmt_sql(&schema), &[&worker, &now, &name, &n]) + .await + .map_err(pg_error)?; + let mut out = Vec::with_capacity(rows.len()); + for row in &rows { + let job = job_from_row(row)?; + out.push(Box::new(PgJobHandle { + job, + pool: pool.clone(), + schema: schema.clone(), + closed: closed.clone(), + }) as Box); + } + Ok(out) + }) + }) + .await + }) + } + + fn ack_batch<'a>(&'a self, ids: &[i64]) -> BoxedFuture<'a, Result> { + if self.closed.load(Ordering::Acquire) { + return Box::pin(async { Err(closed_error()) }); + } + let name = self.name.clone(); + let pool = self.pool.clone(); + let schema = self.schema.clone(); + let ids: Vec = ids.to_vec(); + Box::pin(async move { + // An empty batch is vacuously zero, no round trip. + if ids.is_empty() { + return Ok(0); + } + let now = now_unix(); + let conn = pool.get().await.map_err(pool_error)?; + // The batch ack carries no worker filter (the wave-3 D-31 + // delta's shape is the contract's, ADR-19 §1): the uniform + // validity predicate — processing + unexpired claim + // deadline — per id; lapsed, non-claimed, acked-elsewhere, + // or cancelled ids silently not counted; count returned, + // partial success ordinary (ADR-010 §1). Queue-scoped (the + // handle's name — a queue handle acks its own rows). + let n = conn + .execute( + &format!( + "DELETE FROM {job} + WHERE id = ANY($1) + AND queue = $2 + AND state = 'processing' + AND claim_expires_at >= $3", + job = table(&schema, tables::JOB) + ), + &[&ids, &name, &now], + ) + .await + .map_err(pg_error)?; + Ok(n as i64) + }) + } + + fn cancel<'a>(&'a self, job_id: i64) -> BoxedFuture<'a, Result> { + if self.closed.load(Ordering::Acquire) { + return Box::pin(async { Err(closed_error()) }); + } + let name = self.name.clone(); + let pool = self.pool.clone(); + let schema = self.schema.clone(); + Box::pin(async move { + let conn = pool.get().await.map_err(pool_error)?; + // Unconditional (ADR-010 §1): either state, any holder — + // not an interrupt; the holder's next op refuses (the row + // just is gone — the same shape as a deadline lapse). + // Queue-scoped (the handle's name). + let n = conn + .execute( + &format!( + "DELETE FROM {job} + WHERE id = $1 AND queue = $2 + AND state IN ('pending', 'processing')", + job = table(&schema, tables::JOB) + ), + &[&job_id, &name], + ) + .await + .map_err(pg_error)?; + Ok(n > 0) + }) + } + + fn get_job<'a>(&'a self, job_id: i64) -> BoxedFuture<'a, Result>> { + if self.closed.load(Ordering::Acquire) { + return Box::pin(async { Err(closed_error()) }); + } + let name = self.name.clone(); + let pool = self.pool.clone(); + let schema = self.schema.clone(); + Box::pin(async move { + let conn = pool.get().await.map_err(pool_error)?; + // Queue-scoped + dead-visible (the auto-commit shape, + // ADR-010 §1): dead table first (the diagnosis columns + // ride), then live — the tx seam's `get_job_tx` shape, + // two typed selects, one decode owner ([`job_from_row`]). + let row = conn + .query_opt( + &format!( + "SELECT {dead} FROM {deadt} WHERE id = $1 AND queue = $2", + dead = dead_columns(), + deadt = table(&schema, tables::DEAD) + ), + &[&job_id, &name], + ) + .await + .map_err(pg_error)?; + if let Some(row) = row { + return Ok(Some(job_from_row(&row)?)); + } + let row = conn + .query_opt( + &format!( + "SELECT {live} FROM {jobt} WHERE id = $1 AND queue = $2", + live = live_columns(), + jobt = table(&schema, tables::JOB) + ), + &[&job_id, &name], + ) + .await + .map_err(pg_error)?; + match row { + Some(row) => Ok(Some(job_from_row(&row)?)), + None => Ok(None), + } + }) + } + + fn sweep_expired<'a>(&'a self) -> BoxedFuture<'a, Result> { + if self.closed.load(Ordering::Acquire) { + return Box::pin(async { Err(closed_error()) }); + } + let name = self.name.clone(); + let pool = self.pool.clone(); + let schema = self.schema.clone(); + Box::pin(async move { + let now = now_unix(); + let conn = pool.get().await.map_err(pool_error)?; + // The no-stranded-rows property (ADR-010 §5): every + // past-`expires_at` row (any state, pending and processing + // alike) moves to dead with `"expired"`, and the dead rows + // past their retention TTL delete (the only sweeper dead + // rows ever have — nothing runs without a caller). One + // transaction frame (the move is atomic — a failing dead + // INSERT rolls the whole move back, no stranded-in- + // neither-table state); no leader lock. + in_tx(conn, |client| { + Box::pin(async move { + let move_sql = format!( + "DELETE FROM {job} + WHERE queue = $1 + AND expires_at IS NOT NULL + AND expires_at <= $2 + RETURNING {DEAD_MOVE_COLUMNS}", + job = table(&schema, tables::JOB) + ); + let rows = client + .query(&move_sql, &[&name, &now]) + .await + .map_err(pg_error)?; + for row in &rows { + insert_dead_row_tx(client, &schema, row, EXPIRED_ERROR, now).await?; + } + let retained = client + .execute( + &format!( + "DELETE FROM {dead} + WHERE queue = $1 + AND dead_letter_retention_s IS NOT NULL + AND died_at <= $2::bigint - dead_letter_retention_s", + dead = table(&schema, tables::DEAD) + ), + &[&name, &now], + ) + .await + .map_err(pg_error)?; + Ok(rows.len() as i64 + retained as i64) + }) + }) + .await + }) + } +} + +/// The claim handle (see the module docs). The one-shot ops consume +/// `self: Box` (the whole handle moves into the `'static` +/// future); `heartbeat` is the deliberate repeatable `&self` +/// counter-case (ADR-019 §3 — do not "fix" the asymmetry). +pub struct PgJobHandle { + job: Job, + pool: Pool, + schema: String, + closed: Arc, +} + +impl std::fmt::Debug for PgJobHandle { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("PgJobHandle") + .field("job", &self.job) + .finish_non_exhaustive() + } +} + +impl PgJobHandle { + /// The stored job's live row re-read under the uniform validity + /// predicate (ADR-010 §2): `processing` + the caller's unexpired + /// claim deadline + the caller's claimant. `None` = refused (a + /// lapse, a reclaim, a cancel, or an ack elsewhere consumed the + /// window) — the ops' `Ok(false)` arm; the row is untouched. + /// + /// The move-carrying 14-column shape (the dead-move's decode). + /// Run inside the op's transaction frame (`FOR UPDATE`-locked by + /// the subsequent DELETE/UPDATE of the same row in that frame — + /// the frame is the op's atomicity). + async fn fetch_move_row( + client: &tokio_postgres::Client, + schema: &str, + job_id: i64, + worker_id: &str, + now: i64, + ) -> Result> { + let row = client + .query_opt( + &format!( + "SELECT {move_cols} FROM {job} + WHERE id = $1 AND worker_id = $2 + AND state = 'processing' + AND claim_expires_at >= $3 + FOR UPDATE", + move_cols = DEAD_MOVE_COLUMNS, + job = table(schema, tables::JOB) + ), + &[&job_id, &worker_id, &now], + ) + .await + .map_err(pg_error)?; + Ok(row) + } +} + +impl JobHandle for PgJobHandle { + fn job(&self) -> &Job { + &self.job + } + + fn ack(self: Box) -> BoxedFuture<'static, Result> { + let PgJobHandle { + job, + pool, + schema, + closed, + } = *self; + let id = job.id; + let worker_id = job.worker_id.unwrap_or_default(); + Box::pin(async move { + let _ = closed; + let now = now_unix(); + let conn = pool.get().await.map_err(pool_error)?; + // Delete-on-ack (ADR-010 §1) under the validity predicate + // (ADR-010 §2): 0 rows = refused (lapsed, reclaimed, + // cancelled, or acked elsewhere) — a value, never an + // error. + let n = conn + .execute( + &format!( + "DELETE FROM {job} + WHERE id = $1 AND worker_id = $2 + AND state = 'processing' + AND claim_expires_at >= $3", + job = table(&schema, tables::JOB) + ), + &[&id, &worker_id, &now], + ) + .await + .map_err(pg_error)?; + Ok(n > 0) + }) + } + + fn retry( + self: Box, + err: Option, + delay: Option, + ) -> BoxedFuture<'static, Result> { + let PgJobHandle { + job, + pool, + schema, + closed, + } = *self; + let id = job.id; + let worker_id = job.worker_id.unwrap_or_default(); + // `None` = the queue's curve (ADR-010 §3) from the job's + // stamped base; the attempt index is the claim-row's + // post-claim count (reclaims included). + let delay_s = match delay { + Some(d) => d, + None => backoff_delay_s(job.backoff_base_s, job.attempts, id), + }; + // The exhausted path's string: `None` err = the pinned + // `"max attempts exceeded"` (the engine-observed path's owned + // string wins over the caller-None case, per ADR-019 §3's + // annotation); `Some(err)` rides the row verbatim. + let exhaust_error = err.unwrap_or_else(|| EXHAUSTED_ERROR.to_string()); + Box::pin(async move { + let _ = closed; + let conn = pool.get().await.map_err(pool_error)?; + // One transaction frame for the whole op: the validity + // predicate (the row re-read at 0 rows = refused — the row + // keeps its pre-call state; at-least-once redelivery + // governs) + the reschedule or the dead-letter move, all + // committing atomically. + in_tx(conn, |client| { + Box::pin(async move { + let now = now_unix(); + let Some(row) = + PgJobHandle::fetch_move_row(client, &schema, id, &worker_id, now).await? + else { + return Ok(false); + }; + let attempts: i64 = row.get(5); + let max_attempts: i64 = row.get(6); + if attempts >= max_attempts { + // The exhausted-budget dead-letter move + // (ADR-010 §4): the dead INSERT + the live + // DELETE under the same predicate, atomic in + // the frame (the sqlite substrate's #133 + // defect class excluded — a failing INSERT + // rolls the whole op back, no + // stranded-in-neither-table state). + insert_dead_row_tx(client, &schema, &row, &exhaust_error, now).await?; + client + .execute( + &format!( + "DELETE FROM {job} + WHERE id = $1 AND worker_id = $2 + AND state = 'processing' + AND claim_expires_at >= $3", + job = table(&schema, tables::JOB) + ), + &[&id, &worker_id, &now], + ) + .await + .map_err(pg_error)?; + } else { + // Reschedule pending: the curve draw (or the + // caller's override) is the new ready time from + // now; the claim state clears (deadline, + // claimant, claimed_at). + client + .execute( + &format!( + "UPDATE {job} SET + state = 'pending', + run_at = $2::bigint + $3::bigint, + worker_id = NULL, + claim_expires_at = NULL, + claimed_at = NULL + WHERE id = $1 AND worker_id = $4 + AND state = 'processing' + AND claim_expires_at >= $2", + job = table(&schema, tables::JOB) + ), + &[&id, &now, &delay_s, &worker_id], + ) + .await + .map_err(pg_error)?; + } + Ok(true) + }) + }) + .await + }) + } + + fn fail(self: Box, err: Option) -> BoxedFuture<'static, Result> { + let PgJobHandle { + job, + pool, + schema, + closed, + } = *self; + let id = job.id; + let worker_id = job.worker_id.unwrap_or_default(); + let error = err.unwrap_or_else(|| FAILED_ERROR.to_string()); + Box::pin(async move { + let _ = closed; + let conn = pool.get().await.map_err(pool_error)?; + // One transaction frame for the whole op: the validity + // predicate first (0 rows = refused), then the immediate + // dead-letter move (ADR-010 §4) — `fail` is the "stop + // retrying this" operator, budget-unaware; the move is + // atomic in the frame. + in_tx(conn, |client| { + Box::pin(async move { + let now = now_unix(); + let Some(row) = + PgJobHandle::fetch_move_row(client, &schema, id, &worker_id, now).await? + else { + return Ok(false); + }; + insert_dead_row_tx(client, &schema, &row, &error, now).await?; + client + .execute( + &format!( + "DELETE FROM {job} + WHERE id = $1 AND worker_id = $2 + AND state = 'processing' + AND claim_expires_at >= $3", + job = table(&schema, tables::JOB) + ), + &[&id, &worker_id, &now], + ) + .await + .map_err(pg_error)?; + Ok(true) + }) + }) + .await + }) + } + + fn heartbeat<'a>(&'a self, extend: i64) -> BoxedFuture<'a, Result> { + if self.closed.load(Ordering::Acquire) { + return Box::pin(async { Err(closed_error()) }); + } + let pool = self.pool.clone(); + let schema = self.schema.clone(); + let id = self.job.id; + let worker_id = self.job.worker_id.clone().unwrap_or_default(); + Box::pin(async move { + // Renewal — an absolute reset of the claim deadline from + // now (the new full deadline, not additive; ADR-010 §2). + // The validity predicate rides (a late heartbeat refuses — + // it can never steal the job back from a reclaimer); + // claimed_at never moves. + let now = now_unix(); + let conn = pool.get().await.map_err(pool_error)?; + let n = conn + .execute( + &format!( + "UPDATE {job} + SET claim_expires_at = $3::bigint + $4::bigint + WHERE id = $1 AND worker_id = $2 + AND state = 'processing' + AND claim_expires_at >= $3", + job = table(&schema, tables::JOB) + ), + &[&id, &worker_id, &now, &extend], + ) + .await + .map_err(pg_error)?; + Ok(n > 0) + }) + } +} diff --git a/alkstore-postgres/src/store.rs b/alkstore-postgres/src/store.rs index 76244fa..f7fc30b 100644 --- a/alkstore-postgres/src/store.rs +++ b/alkstore-postgres/src/store.rs @@ -26,7 +26,8 @@ //! the trait stubs below are the wave-3 posture — they return //! `Err(Database("… wiring lands with the … task"))` until the //! mechanism tasks replace them. `notify`/`listen` are wired (the -//! notify-listen task's) — the wake contract's pg arm rides the +//! notify-listen task's), `stream` (the streams task's), `queue` +//! (the queues task's) — the wake contract's pg arm rides the //! forwarder. use alkstore::Store; @@ -281,13 +282,24 @@ impl Store for PgStore { fn queue<'a>( &'a self, - _name: &str, - _opts: alkstore::QueueOpts, + name: &str, + opts: alkstore::QueueOpts, ) -> alkstore::BoxedFuture<'a, alkstore::Result>> { + if let Err(e) = alkstore::validate_shared_name(name) { + return Box::pin(async move { Err(e) }); + } if let Err(e) = self.closed_check() { return Box::pin(async move { Err(e) }); } - Box::pin(async { Err(stub("queue wiring lands with the queues task")) }) + let name = name.to_string(); + let handle = crate::queue::PgQueueHandle::new( + name, + opts, + self.pool.clone(), + self.schema.clone(), + self.closed.clone(), + ); + Box::pin(async move { Ok(Box::new(handle) as Box) }) } fn outbox<'a>( @@ -355,6 +367,9 @@ impl Store for PgStore { #[cfg(test)] mod open_tests; +#[cfg(test)] +mod queue_tests; + #[cfg(test)] mod notify_tests; diff --git a/alkstore-postgres/src/store/open_tests.rs b/alkstore-postgres/src/store/open_tests.rs index fdc6581..3a3c2f8 100644 --- a/alkstore-postgres/src/store/open_tests.rs +++ b/alkstore-postgres/src/store/open_tests.rs @@ -663,8 +663,18 @@ async fn store_trait_methods_are_wiring_stubs() { // named handle — full behavior the stream tests' scope. let handle = store.stream("s").await.unwrap(); assert_eq!(handle.name(), "s"); - let err = stub_err!(store.queue("q", alkstore::QueueOpts::default()).await); - assert!(matches!(err, alkstore::Error::Database(_))); + // The queue constructor is wired (the queues task's): returns a + // named handle whose enqueue lands — full behavior the queue + // tests' scope. + let queue = store + .queue("stub_q_probe", alkstore::QueueOpts::default()) + .await + .unwrap(); + assert_eq!(queue.name(), "stub_q_probe"); + queue + .enqueue(serde_json::json!({}), alkstore::EnqueueOpts::default()) + .await + .unwrap(); let err = stub_err!(store.outbox("o").await); assert!(matches!(err, alkstore::Error::Database(_))); let err = stub_err!(store.try_lock("l", "owner", 60).await); @@ -691,8 +701,8 @@ async fn store_trait_methods_are_wiring_stubs() { // The remaining stub messages name their landing task (the // wave-3 posture's "… wiring lands with the … task" shape) — in // the source chain (`Database`'s Display is opaque; ADR-008 §5's - // detail-carriage). `queue` sampled as the representative. - let err = stub_err!(store.queue("ch", alkstore::QueueOpts::default()).await); + // detail-carriage). `outbox` sampled as the representative. + let err = stub_err!(store.outbox("ch").await); match err { alkstore::Error::Database(source) => { let msg = source.to_string(); @@ -701,7 +711,7 @@ async fn store_trait_methods_are_wiring_stubs() { "stub errors name their landing task, got: {msg}" ); } - other => panic!("queue stub must be Database, got {other:?}"), + other => panic!("outbox stub must be Database, got {other:?}"), } drop(store); diff --git a/alkstore-postgres/src/store/queue_tests.rs b/alkstore-postgres/src/store/queue_tests.rs new file mode 100644 index 0000000..d8c7ff2 --- /dev/null +++ b/alkstore-postgres/src/store/queue_tests.rs @@ -0,0 +1,1439 @@ +//! The queue mechanism's acceptance tests (`pg-engine-queues`): the +//! enqueue→claim→ack lifecycle with stamps visible via `get_job` +//! matching the resolution rules, the extent guard (ADR-023 §2), the +//! exactly-once handout under concurrent claims (two workers, one row; +//! 6×40 drain, no double handouts), the backoff curve's range pin +//! (statistical, tolerance-bounded) with the 1-hour cap and the +//! `Some(d)` override, the uniform validity predicate refusing lapsed +//! ops with `Ok(false)` and a reclaim consuming an attempt, the +//! dead-letter defaults and dead-row `get_job` visibility, the +//! pre-claim exhausted-reclaimable sweep, claim ordering, cancel + +//! `ack_batch`'s per-id predicate without a worker filter, `sweep_ +//! expired`'s both-states move + retention (incl. forever-none), the +//! wake-driven claim loop pinned end-to-end (enqueue → notify → wake → +//! claim, cross-connection; the LISTEN-driven posture with the re-poll +//! safety net — the contract claim ops stay pull-shaped), queue-name +//! scoping, the validation posture (empty/reserved queue names, empty +//! `worker_id`), and the closed-store fail-closed shape on all queue +//! ops (ADR-010; ADR-019 §1–§3; ADR-020; ADR-023 §2). +//! +//! Harness convention (the schema/open tasks'): connection settings +//! ride the environment (`ALKSTORE_PG_HOST/PORT/USER/PASSWORD/DB`), +//! never hardcoded; tests without a reachable server skip cleanly so +//! the workspace gates stay green server-less. Isolation is a fresh +//! unique schema per test. + +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +use std::time::Duration; + +use alkstore::{EnqueueOpts, JobHandle, JobState, QueueOpts, Store}; +use serde_json::{Value, json}; + +use crate::queue::{EXHAUSTED_ERROR, EXPIRED_ERROR, FAILED_ERROR, backoff_delay_s}; +use crate::store::{PgStore, open_store}; + +const ENV_HOST: &str = "ALKSTORE_PG_HOST"; +const ENV_PORT: &str = "ALKSTORE_PG_PORT"; +const ENV_USER: &str = "ALKSTORE_PG_USER"; +const ENV_PASSWORD: &str = "ALKSTORE_PG_PASSWORD"; +const ENV_DB: &str = "ALKSTORE_PG_DB"; + +fn harness_dsn() -> Option { + let host = std::env::var(ENV_HOST).ok()?; + let port: u16 = std::env::var(ENV_PORT).ok()?.parse().ok()?; + let user = std::env::var(ENV_USER).ok()?; + let password = std::env::var(ENV_PASSWORD).ok()?; + let db = std::env::var(ENV_DB).unwrap_or_else(|_| "postgres".to_string()); + Some(format!( + "host={host} port={port} user={user} password={password} dbname={db}" + )) +} + +fn instance_namer(tag: &str) -> impl Fn() -> String + use<'_> { + let counter = AtomicU64::new(0); + move || { + format!( + "{tag}_{}_{}_{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos(), + counter.fetch_add(1, Ordering::SeqCst), + ) + } +} + +/// A raw admin client for cross-connection probes (NOTIFY senders, +/// schema teardown, row backdating) — separate from the store's pool. +async fn harness_client() -> Option { + let (client, connection) = tokio_postgres::connect(&harness_dsn()?, tokio_postgres::NoTls) + .await + .ok()?; + tokio::spawn(async move { + let _ = connection.await; + }); + Some(client) +} + +async fn drop_schema(client: &tokio_postgres::Client, schema: &str) { + if schema == crate::schema::DEFAULT_SCHEMA { + return; + } + let sql = format!( + "DROP SCHEMA IF EXISTS {} CASCADE", + crate::schema::quote_identifier(schema) + ); + let _ = client.batch_execute(&sql).await; +} + +fn test_opts(schema: &str) -> crate::opts::PgOpts { + crate::opts::PgOpts { + schema: schema.to_string(), + ..crate::opts::PgOpts::default() + } +} + +/// Open a store on the harness server with a fresh unique schema — +/// `None` = no harness server (the skip posture). The returned store +/// closes on `Fixture` drop (the schema teardown included). +struct Fixture { + store: PgStore, + admin: tokio_postgres::Client, + schema: String, +} + +impl Fixture { + async fn open(namer: &impl Fn() -> String) -> Option { + let dsn = harness_dsn()?; + let schema = namer(); + let store = open_store(&dsn, test_opts(&schema)).await.unwrap(); + let admin = harness_client().await?; + Some(Fixture { + store, + admin, + schema, + }) + } +} + +impl Drop for Fixture { + fn drop(&mut self) { + self.store.close(); + } +} + +fn opts() -> QueueOpts { + QueueOpts { + visibility_timeout_s: 300, + max_attempts: 3, + backoff_base_s: 5, + dead_letter_retention_s: None, + } +} + +fn now_unix() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs() as i64 +} + +/// Backdate a live row's `expires_at` past the wall clock (the honest +/// cross-connection probe for the sweep's both-states move — a fresh +/// enqueue would carry a future expiry, so the admin UPDATE backdates +/// it: schema-qualified, typed-binded). +async fn backdate_expires(admin: &tokio_postgres::Client, schema: &str, id: i64) { + admin + .execute( + &format!( + "UPDATE {} SET expires_at = $2 WHERE id = $1", + crate::schema::QualifiedTable { + schema, + name: crate::schema::tables::JOB, + } + ), + &[&id, &(now_unix() - 1)], + ) + .await + .unwrap(); +} + +/// Acceptance: the full lifecycle — enqueue stamps the resolved values +/// (resolved `run_at` after delay-over-`run_at`, absolute `expires_at` +/// after relative-`expires`, the handle's `QueueOpts` stamps with the +/// per-job `max_attempts` override), claim returns a boxed handle +/// carrying the row, ack deletes, `get_job` sees the states (ADR-010 +/// §1/§3a; ADR-020 §1–§2; ADR-019 §1). +#[tokio::test(flavor = "multi_thread")] +async fn enqueue_claim_ack_lifecycle_with_resolved_stamps() { + let Some(fx) = Fixture::open(&instance_namer("ql")).await else { + eprintln!("skip: no harness server"); + return; + }; + + let qopts = QueueOpts { + visibility_timeout_s: 120, + max_attempts: 4, + backoff_base_s: 9, + dead_letter_retention_s: Some(7200), + }; + let queue = fx.store.queue("emails", qopts.clone()).await.unwrap(); + assert_eq!(queue.name(), "emails"); + + let before = now_unix(); + let id = queue + .enqueue( + json!({"to": "a@example.com"}), + EnqueueOpts { + delay: Some(60), + run_at: Some(before + 10_000), + priority: 3, + max_attempts: Some(7), + expires: Some(1200), + }, + ) + .await + .unwrap(); + + let job = queue.get_job(id).await.unwrap().expect("row exists"); + assert_eq!(job.id, id); + assert_eq!(job.queue, "emails"); + assert_eq!(job.state, JobState::Pending); + assert_eq!(job.priority, 3); + assert!( + (before + 60..=before + 62).contains(&job.run_at), + "ready time is now + delay, delay wins over run_at: {}", + job.run_at + ); + assert_eq!( + job.expires_at, + Some(before + 1200), + "relative expires resolves to the absolute row expiry" + ); + assert_eq!(job.max_attempts, 7, "the per-job override stamps the row"); + assert_eq!(job.visibility_timeout_s, 120, "the handle's stamp"); + assert_eq!(job.backoff_base_s, 9, "the handle's stamp"); + assert_eq!(job.dead_letter_retention_s, Some(7200)); + assert_eq!(job.attempts, 0); + assert_eq!(job.worker_id, None); + assert_eq!(job.claimed_at, None); + assert_eq!(job.claim_expires_at, None); + assert_eq!(job.last_error, None); + assert_eq!(job.died_at, None); + let decoded: Value = job.payload_as().unwrap(); + assert_eq!(decoded, json!({"to": "a@example.com"})); + + // Not yet claimable — ready time is a minute out. + assert!(queue.claim_one("w1").await.unwrap().is_none()); + + // A second enqueue with run_at in the past is claimable now. + let id2 = queue + .enqueue( + json!({"n": 2}), + EnqueueOpts { + run_at: Some(before - 10), + ..Default::default() + }, + ) + .await + .unwrap(); + let job2 = queue.get_job(id2).await.unwrap().unwrap(); + assert!( + (before - 10..=before - 9).contains(&job2.run_at), + "run_at is used literally when delay is unset" + ); + + // Claim: the handle carries the row, the claim counts an attempt, + // the deadline comes from the job's own visibility stamp. + let handle = queue.claim_one("w1").await.unwrap().expect("claimable"); + let job = handle.job(); + assert_eq!(job.id, id2); + assert_eq!(job.state, JobState::Processing); + assert_eq!(job.attempts, 1); + assert_eq!(job.worker_id.as_deref(), Some("w1")); + assert!(job.claimed_at.is_some()); + assert_eq!( + job.claim_expires_at, + Some(job.claimed_at.unwrap() + 120), + "the deadline comes from the job's own visibility stamp" + ); + + // get_job reflects the claim. + let seen = queue.get_job(id2).await.unwrap().unwrap(); + assert_eq!(seen.state, JobState::Processing); + assert_eq!(seen.worker_id.as_deref(), Some("w1")); + + // A second worker cannot claim it (exactly-once handout). + assert!(queue.claim_one("w2").await.unwrap().is_none()); + + // Ack deletes the row (ADR-010 §1). + assert!(handle.ack().await.unwrap()); + assert!(queue.get_job(id2).await.unwrap().is_none()); + assert!(queue.claim_one("w2").await.unwrap().is_none()); + assert!( + queue.get_job(id).await.unwrap().is_some(), + "the delayed row survives untouched" + ); + + drop_schema(&fx.admin, &fx.schema).await; +} + +/// Acceptance: the extent guard (ADR-023 §2) — `claim_batch(n <= 0)` +/// claims nothing (the empty `Vec`, no error; pg's `LIMIT` with a +/// negative would be a server error, which the guard makes +/// unreachable) and `claim_one` on an empty queue is `None` (no-work +/// is a value). +#[tokio::test(flavor = "multi_thread")] +async fn extent_guard_claims_nothing_on_non_positive_n() { + let Some(fx) = Fixture::open(&instance_namer("qe")).await else { + eprintln!("skip: no harness server"); + return; + }; + let queue = fx.store.queue("work", opts()).await.unwrap(); + + for i in 0..5 { + queue + .enqueue(json!({"i": i}), EnqueueOpts::default()) + .await + .unwrap(); + } + + for n in [-1i64, 0] { + let got = queue.claim_batch("w1", n).await.unwrap(); + assert!( + got.is_empty(), + "claim_batch({n}) must claim nothing (empty Vec, no error)" + ); + } + // Nothing was destroyed by the guarded call. + let seen = queue.get_job(1).await.unwrap().unwrap(); + assert_eq!(seen.state, JobState::Pending); + + // A positive claim on the same state still works and hands out + // exactly the requested extent. + let got = queue.claim_batch("w1", 2).await.unwrap(); + assert_eq!(got.len(), 2); + + drop_schema(&fx.admin, &fx.schema).await; +} + +/// Acceptance: exactly-once handout under concurrent claims — two +/// workers racing claim on one row (exactly one wins; the POC's 4×4 +/// shape at 2×1) and a 6×40 concurrent drain with zero double-claims +/// (the POC-pinned property, engine-tested here; the drain tasks share +/// the store through `Arc`-free sibling stores — every +/// store instance is a stateless client of the same schema'd tables, +/// so concurrent stores over one schema are the honest multi-worker +/// shape). +#[tokio::test(flavor = "multi_thread")] +async fn concurrent_claims_hand_each_row_out_exactly_once() { + let Some(dsn) = harness_dsn() else { + eprintln!("skip: no harness server"); + return; + }; + let schema = instance_namer("qx")(); + let admin = harness_client().await.unwrap(); + + // Two workers, one row: exactly one claim wins. + { + let store = open_store(&dsn, test_opts(&schema)).await.unwrap(); + let queue = store.queue("solo", opts()).await.unwrap(); + let id = queue + .enqueue(json!({"only": true}), EnqueueOpts::default()) + .await + .unwrap(); + let q1 = store.queue("solo", opts()).await.unwrap(); + let q2 = store.queue("solo", opts()).await.unwrap(); + let (a, b) = tokio::join!(q1.claim_one("w1"), q2.claim_one("w2")); + let winners = [a.unwrap(), b.unwrap()] + .into_iter() + .flatten() + .collect::>(); + assert_eq!(winners.len(), 1, "exactly one worker got the row"); + assert_eq!(winners[0].job().id, id); + store.close(); + } + + // Many concurrent stores drain a whole queue with no double + // handout (cross-instance claims — each drain task opens its own + // pool connection; the SKIP LOCKED handout is the exactly-once + // property under connection diversity too). + const JOBS: i64 = 40; + const WORKERS: usize = 6; + { + let store = open_store(&dsn, test_opts(&schema)).await.unwrap(); + let queue = store.queue("shared_arc", opts()).await.unwrap(); + for i in 0..JOBS { + queue + .enqueue(json!({"i": i}), EnqueueOpts::default()) + .await + .unwrap(); + } + store.close(); + + let factory = Arc::new(TestFactory { + dsn: dsn.clone(), + schema: schema.clone(), + }); + let claimed = Arc::new(AtomicUsize::new(0)); + let mut handles = Vec::new(); + for w in 0..WORKERS { + let factory = factory.clone(); + let claimed = claimed.clone(); + handles.push(tokio::spawn(async move { + let store = factory.open().await; + let queue = store.queue("shared_arc", opts()).await.unwrap(); + let mut got = 0usize; + loop { + let batch = queue + .claim_batch(&format!("w{w}"), 4) + .await + .expect("claim must not error under contention"); + if batch.is_empty() { + break; + } + got += batch.len(); + // Hold the rows (no ack) — the drain semantics are + // what is under test. + } + claimed.fetch_add(got, Ordering::SeqCst); + got + })); + } + let mut shares = Vec::new(); + for h in handles { + shares.push(h.await.unwrap()); + } + assert_eq!(claimed.load(Ordering::SeqCst), JOBS as usize); + assert!( + shares.iter().any(|s| *s > 0) && shares.iter().any(|s| *s < JOBS as usize), + "work was shared across workers, not drained by one" + ); + + // Every job is processing now; no id claimed twice (probed via + // a raw admin count — the bigserial ids are not the loop's + // 1..=n range, other tables' rows interleave). + let raw = open_store(&dsn, test_opts(&schema)).await.unwrap(); + let job_table = crate::schema::QualifiedTable { + schema: &schema, + name: crate::schema::tables::JOB, + } + .to_string(); + let conn = raw.pool().get().await.unwrap(); + let processing: i64 = conn + .query_one( + &format!( + "SELECT count(*) FROM {job_table} WHERE state = 'processing' AND queue = 'shared_arc'" + ), + &[], + ) + .await + .unwrap() + .get(0); + let distinct: i64 = conn + .query_one( + &format!( + "SELECT count(DISTINCT id) FROM {job_table} WHERE state = 'processing' AND queue = 'shared_arc'" + ), + &[], + ) + .await + .unwrap() + .get(0); + assert_eq!(processing, JOBS, "every job handed out exactly once"); + assert_eq!(distinct, JOBS, "no id claimed twice"); + store.close(); + raw.close(); + } + + drop_schema(&admin, &schema).await; +} + +/// A per-test store factory for the drain tasks (a fresh store +/// instance over the same schema per call — the honest cross-instance +/// worker shape). +struct TestFactory { + dsn: String, + schema: String, +} + +impl TestFactory { + async fn open(&self) -> PgStore { + open_store(&self.dsn, test_opts(&self.schema)) + .await + .unwrap() + } +} + +/// Acceptance: the uniform validity predicate — once the claim +/// deadline lapsed, all handle ops refuse with `Ok(false)` (heartbeat, +/// ack, retry, fail — never errors), the row untouched, and a reclaim +/// consumes an attempt (ADR-010 §2). Healthy renewal is pinned first: +/// absolute reset, repeatable, `claimed_at` untouched. +#[tokio::test(flavor = "multi_thread")] +async fn lapsed_deadline_refuses_all_handle_ops_and_reclaim_consumes_an_attempt() { + let Some(fx) = Fixture::open(&instance_namer("qv")).await else { + eprintln!("skip: no harness server"); + return; + }; + + // Healthy phase (visibility 300): renewal lands and is repeatable. + let queue = fx.store.queue("healthy", opts()).await.unwrap(); + let id = queue + .enqueue(json!({"work": true}), EnqueueOpts::default()) + .await + .unwrap(); + let handle = queue.claim_one("w1").await.unwrap().unwrap(); + let first_deadline = handle.job().claim_expires_at.unwrap(); + assert!(handle.heartbeat(5000).await.unwrap(), "renewal lands"); + assert!(handle.heartbeat(5000).await.unwrap(), "repeatable"); + let renewed = queue.get_job(id).await.unwrap().unwrap(); + assert!( + renewed.claim_expires_at.unwrap() >= first_deadline + 4500, + "renewal is an absolute reset from now: {} vs {first_deadline}", + renewed.claim_expires_at.unwrap() + ); + assert!( + renewed.claimed_at == handle.job().claimed_at, + "heartbeat never touches claimed_at" + ); + assert!(handle.ack().await.unwrap(), "healthy ack lands"); + + // Lapsed phase (visibility 1): every op refuses with Ok(false). + let queue = fx + .store + .queue( + "lapsed", + QueueOpts { + visibility_timeout_s: 1, + max_attempts: 5, + backoff_base_s: 5, + dead_letter_retention_s: None, + }, + ) + .await + .unwrap(); + + let mut ids = Vec::new(); + for _ in 0..4 { + ids.push( + queue + .enqueue(json!({"work": true}), EnqueueOpts::default()) + .await + .unwrap(), + ); + } + let mut claims: Vec> = Vec::new(); + for id in &ids { + let h = queue + .claim_batch("w1", 1) + .await + .unwrap() + .into_iter() + .next() + .expect("claimable"); + assert_eq!(h.job().id, *id); + claims.push(h); + } + + for id in &ids { + until_deadline_lapsed(queue.as_ref(), *id).await; + } + + assert!( + !claims[0].heartbeat(5000).await.unwrap(), + "late heartbeat refuses" + ); + assert!(!claims[0].heartbeat(5000).await.unwrap(), "still refuses"); + assert!(!claims.remove(1).ack().await.unwrap(), "lapsed ack refuses"); + assert!( + !claims + .remove(1) + .retry(Some("boom".to_string()), Some(10)) + .await + .unwrap(), + "lapsed retry refuses" + ); + assert!( + !claims + .remove(1) + .fail(Some("boom".to_string())) + .await + .unwrap(), + "lapsed fail refuses" + ); + + // The rows are untouched by the refusals. + for id in &ids { + let job = queue.get_job(*id).await.unwrap().unwrap(); + assert_eq!(job.state, JobState::Processing); + assert_eq!(job.attempts, 1); + } + + // A reclaim wins and consumes an attempt. + let reclaimed = queue.claim_one("w2").await.unwrap().expect("reclaimable"); + assert_eq!(reclaimed.job().attempts, 2, "the reclaim counts an attempt"); + assert_eq!(reclaimed.job().worker_id.as_deref(), Some("w2")); + + // Post-reclaim, the original claim's ops refuse (the reclaim won + // atomically). + assert!( + !claims[0].heartbeat(5000).await.unwrap(), + "the original claim's ops refuse post-reclaim" + ); + + drop_schema(&fx.admin, &fx.schema).await; +} + +/// Wait until the claimed row's deadline has lapsed — the bounded-wait +/// posture (no tight-race assertions). +async fn until_deadline_lapsed(queue: &dyn alkstore::Queue, id: i64) { + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); + loop { + let lapsed = match queue.get_job(id).await { + Ok(Some(job)) => { + job.claim_expires_at.is_some() && job.claim_expires_at.unwrap() < now_unix() + } + _ => false, + }; + if lapsed { + return; + } + assert!(tokio::time::Instant::now() < deadline, "never lapsed"); + tokio::time::sleep(Duration::from_millis(20)).await; + } +} + +/// Acceptance: the backoff curve — `retry(err, None)` computes the +/// equal-jitter exponential `delay ∈ [base·2^(a−1)/2, base·2^(a−1)]`, +/// capped at 1 hour (ADR-010 §3): statistical pin over many draws, +/// tolerance-bounded; the row's `run_at` after an end-to-end retry +/// lands in the range; `Some(d)` overrides. +#[tokio::test(flavor = "multi_thread")] +async fn backoff_curve_lands_in_the_pinned_range() { + // Statistical pin of the pure formula (a = attempts index): base + // 5, a=1 → [3, 5]; draws must stay in the inclusive integer range + // and spread across most of it. + for (base, attempts, lo, hi) in [(5, 1, 3, 5), (5, 2, 5, 10), (5, 3, 10, 20), (1, 1, 1, 1)] { + let mut lo_seen = i64::MAX; + let mut hi_seen = i64::MIN; + for id in 0..400 { + let d = backoff_delay_s(base, attempts, id); + assert!( + (lo..=hi).contains(&d), + "delay {d} outside [{lo}, {hi}] for base={base} a={attempts}" + ); + lo_seen = lo_seen.min(d); + hi_seen = hi_seen.max(d); + } + if lo == hi { + assert_eq!(lo_seen, hi_seen, "degenerate range pins to the value"); + } else { + assert!( + lo_seen <= lo + (hi - lo) / 2 && hi_seen >= hi - (hi - lo) / 2, + "draws must spread over the range ([{lo_seen}, {hi_seen}] vs [{lo}, {hi}])" + ); + } + } + + // Cap at 1 hour: a huge doubling exponent still lands in [1800, 3600]. + for id in 0..200 { + let d = backoff_delay_s(5, 40, id); + assert!((1800..=3600).contains(&d), "capped delay {d} out of range"); + } + + // End-to-end: the row's run_at after retry(err, None) is now plus + // a curve draw from the job's stamped base. + let Some(fx) = Fixture::open(&instance_namer("qb")).await else { + eprintln!("skip: no harness server"); + return; + }; + let queue = fx + .store + .queue( + "retrying", + QueueOpts { + visibility_timeout_s: 300, + max_attempts: 5, + backoff_base_s: 8, + dead_letter_retention_s: None, + }, + ) + .await + .unwrap(); + let id = queue + .enqueue(json!({"task": 1}), EnqueueOpts::default()) + .await + .unwrap(); + let handle = queue.claim_one("w1").await.unwrap().unwrap(); + assert_eq!(handle.job().attempts, 1, "a = the post-claim count"); + let before = now_unix(); + assert!(handle.retry(None, None).await.unwrap()); + let job = queue.get_job(id).await.unwrap().unwrap(); + assert_eq!(job.state, JobState::Pending); + let span_lo = before - 1 + 8 / 2; + let span_hi = before + 1 + 8; + assert!( + (span_lo..=span_hi).contains(&job.run_at), + "run_at {} must be now + a draw from [4, 8] (base 8, a=1)", + job.run_at + ); + + // Some(d) overrides: wait out the curve draw, claim again, retry + // with the explicit delay. + let deadline = tokio::time::Instant::now() + Duration::from_secs(30); + let override_landed = loop { + match queue.claim_one("w1").await { + Ok(Some(h)) => { + let ok = h + .retry(Some("still broken".to_string()), Some(123)) + .await + .unwrap(); + assert!(ok, "the override retry lands"); + break true; + } + Ok(None) => {} + Err(e) => panic!("claim errored: {e}"), + } + assert!( + tokio::time::Instant::now() < deadline, + "curve never elapsed" + ); + tokio::time::sleep(Duration::from_millis(200)).await; + }; + assert!(override_landed); + let job = queue.get_job(id).await.unwrap().unwrap(); + let now = now_unix(); + assert!( + (now + 123 - 2..=now + 123 + 2).contains(&job.run_at) && job.run_at >= before + 120, + "the explicit delay passes through: {} (now {now})", + job.run_at + ); + + drop_schema(&fx.admin, &fx.schema).await; +} + +/// Acceptance: dead-letter — `retry` at exhausted budget moves the row +/// to dead with the pinned default string (`"max attempts exceeded"` — +/// or the caller's `Some(err)` verbatim), `fail(None)` records +/// `"failed"`, `get_job` sees dead rows with `last_error`/`died_at` +/// carrying the full stamp list (ADR-010 §1/§4; ADR-019 §3). +#[tokio::test(flavor = "multi_thread")] +async fn exhaustion_and_fail_dead_letter_with_the_pinned_strings() { + let Some(fx) = Fixture::open(&instance_namer("qd")).await else { + eprintln!("skip: no harness server"); + return; + }; + let queue = fx + .store + .queue( + "dying", + QueueOpts { + visibility_timeout_s: 300, + max_attempts: 2, + backoff_base_s: 5, + dead_letter_retention_s: None, + }, + ) + .await + .unwrap(); + + // Exhaustion via retry: claim (attempts=1), retry under budget → + // pending, reclaim (attempts=2 == max) → the retry dead-letters. + let id1 = queue + .enqueue(json!({"n": 1}), EnqueueOpts::default()) + .await + .unwrap(); + let h = queue.claim_one("w1").await.unwrap().unwrap(); + assert!(h.retry(None, Some(0)).await.unwrap(), "under budget"); + let job = queue.get_job(id1).await.unwrap().unwrap(); + assert_eq!(job.state, JobState::Pending); + assert_eq!(job.attempts, 1); + let h = queue.claim_one("w1").await.unwrap().unwrap(); + assert_eq!(h.job().attempts, 2, "a reclaim counts too"); + assert!( + h.retry(Some("caller string".to_string()), None) + .await + .unwrap(), + "the exhausted-budget retry lands as the dead-letter move" + ); + let dead = queue + .get_job(id1) + .await + .unwrap() + .expect("dead rows visible"); + assert_eq!(dead.state, JobState::Dead); + assert_eq!( + dead.last_error.as_deref(), + Some("caller string"), + "retry(Some(err)) carries the caller's string on the dead row" + ); + assert!(dead.died_at.is_some()); + assert_eq!(dead.max_attempts, 2, "the dead row carries its stamps"); + assert_eq!(dead.visibility_timeout_s, 300); + assert_eq!(dead.backoff_base_s, 5); + assert_eq!(dead.attempts, 2); + assert!( + dead.claimed_at.is_some(), + "the dead row keeps claim history" + ); + assert_eq!(dead.worker_id.as_deref(), Some("w1")); + + // retry(None) exhaustion pins the default string. + let id2 = queue + .enqueue(json!({"n": 2}), EnqueueOpts::default()) + .await + .unwrap(); + let h = queue.claim_one("w1").await.unwrap().unwrap(); + assert_eq!(h.job().attempts, 1, "a fresh claim is one attempt"); + assert!(h.retry(None, Some(0)).await.unwrap(), "attempts 1 < 2"); + let h = queue.claim_one("w1").await.unwrap().unwrap(); + assert_eq!(h.job().attempts, 2); + assert!(h.retry(None, None).await.unwrap(), "at-budget dead-letters"); + let dead = queue.get_job(id2).await.unwrap().unwrap(); + assert_eq!(dead.state, JobState::Dead); + assert_eq!( + dead.last_error.as_deref(), + Some(EXHAUSTED_ERROR), + "retry(None) exhaustion records the pinned default string" + ); + assert!(dead.died_at.is_some()); + + drop_schema(&fx.admin, &fx.schema).await; + + // fail(None) pins "failed"; fail(Some(e)) carries the caller's. + let Some(fx) = Fixture::open(&instance_namer("qf")).await else { + eprintln!("skip: no harness server"); + return; + }; + let queue = fx.store.queue("dying", opts()).await.unwrap(); + let id = queue + .enqueue(json!({"n": 3}), EnqueueOpts::default()) + .await + .unwrap(); + let h = queue.claim_one("w1").await.unwrap().unwrap(); + assert!(h.fail(None).await.unwrap(), "fail lands"); + let dead = queue.get_job(id).await.unwrap().unwrap(); + assert_eq!(dead.state, JobState::Dead); + assert_eq!( + dead.last_error.as_deref(), + Some(FAILED_ERROR), + "fail(None) records the pinned default string" + ); + assert!(dead.died_at.is_some()); + + let id = queue + .enqueue(json!({"n": 4}), EnqueueOpts::default()) + .await + .unwrap(); + let h = queue.claim_one("w1").await.unwrap().unwrap(); + assert!(h.fail(Some("explicit reason".to_string())).await.unwrap()); + let dead = queue.get_job(id).await.unwrap().unwrap(); + assert_eq!(dead.last_error.as_deref(), Some("explicit reason")); + + // The pinning constants are the contract's defaults. + assert_eq!(EXHAUSTED_ERROR, "max attempts exceeded"); + assert_eq!(FAILED_ERROR, "failed"); + assert_eq!(EXPIRED_ERROR, "expired"); + + drop_schema(&fx.admin, &fx.schema).await; +} + +/// Acceptance: the pre-claim dead-letter sweep — a reclaimable row +/// at-or-past its budget is dead-lettered by the next claim (the +/// substrate's D-12 shape: a worker dying right after its last allowed +/// claim dead-letters, not strands — ADR-010 §2/§3). +#[tokio::test(flavor = "multi_thread")] +async fn pre_claim_sweep_dead_letters_exhausted_reclaimables() { + let Some(fx) = Fixture::open(&instance_namer("q12")).await else { + eprintln!("skip: no harness server"); + return; + }; + let queue = fx + .store + .queue( + "d12", + QueueOpts { + visibility_timeout_s: 1, + max_attempts: 1, + backoff_base_s: 5, + dead_letter_retention_s: None, + }, + ) + .await + .unwrap(); + let spent = queue + .enqueue(json!({"k": 1}), EnqueueOpts::default()) + .await + .unwrap(); + let fresh = queue + .enqueue(json!({"k": 2}), EnqueueOpts::default()) + .await + .unwrap(); + + // Claim the budget-spent row (attempts 1 == max 1) and let its + // deadline lapse: it is now a reclaimable-past-budget row. + let h = queue.claim_one("w1").await.unwrap().unwrap(); + assert_eq!(h.job().id, spent); + until_deadline_lapsed(queue.as_ref(), spent).await; + + // The next claim dead-letters the spent row (the pre-claim sweep) + // and hands out the fresh one. + let h = queue + .claim_one("w2") + .await + .unwrap() + .expect("fresh claimable"); + assert_eq!(h.job().id, fresh); + let dead = queue.get_job(spent).await.unwrap().expect("dead visible"); + assert_eq!(dead.state, JobState::Dead); + assert_eq!( + dead.last_error.as_deref(), + Some(EXHAUSTED_ERROR), + "the exhausted reclaimable dead-letters (not strands)" + ); + assert!(dead.died_at.is_some()); + + drop_schema(&fx.admin, &fx.schema).await; +} + +/// Acceptance: claim ordering — `priority DESC`, ready-time ASC, +/// enqueue order (ADR-010 §1). +#[tokio::test(flavor = "multi_thread")] +async fn claim_ordering_is_priority_desc_run_at_then_enqueue_order() { + let Some(fx) = Fixture::open(&instance_namer("qo")).await else { + eprintln!("skip: no harness server"); + return; + }; + let queue = fx.store.queue("ordered", opts()).await.unwrap(); + let now = now_unix(); + let a = queue + .enqueue( + json!("a"), + EnqueueOpts { + run_at: Some(now), + ..Default::default() + }, + ) + .await + .unwrap(); + let b = queue + .enqueue( + json!("b"), + EnqueueOpts { + run_at: Some(now), + priority: 5, + ..Default::default() + }, + ) + .await + .unwrap(); + let c = queue + .enqueue( + json!("c"), + EnqueueOpts { + run_at: Some(now), + priority: 5, + ..Default::default() + }, + ) + .await + .unwrap(); + let d = queue + .enqueue( + json!("d"), + EnqueueOpts { + run_at: Some(now - 10), + priority: 5, + ..Default::default() + }, + ) + .await + .unwrap(); + + for _ in 0..3 { + queue.claim_one("w").await.unwrap().unwrap(); + } + // After 3 claims: d, b, c claimed (priority DESC; among equal + // priority, run_at ASC → d's earlier ready time, then enqueue + // order b then c); a (priority 0) is last, unclaimed. + let ja = queue.get_job(a).await.unwrap().unwrap(); + assert_eq!(ja.state, JobState::Pending, "the lower-priority job last"); + assert_eq!(ja.worker_id, None); + for id in [b, c, d] { + let j = queue.get_job(id).await.unwrap().unwrap(); + assert_eq!(j.state, JobState::Processing, "claimed in order"); + assert_eq!(j.worker_id.as_deref(), Some("w")); + } + + drop_schema(&fx.admin, &fx.schema).await; +} + +/// Acceptance: `cancel` is unconditional (either state, any holder), +/// `ack_batch` applies the per-id validity predicate without a worker +/// filter (partial success ordinary), missing ids silent (ADR-010 §1; +/// ADR-019 §1). +#[tokio::test(flavor = "multi_thread")] +async fn cancel_and_ack_batch_behaviors() { + let Some(fx) = Fixture::open(&instance_namer("qc")).await else { + eprintln!("skip: no harness server"); + return; + }; + let queue = fx.store.queue("ops", opts()).await.unwrap(); + + let pending = queue + .enqueue(json!({"p": 1}), EnqueueOpts::default()) + .await + .unwrap(); + let processing = queue + .enqueue(json!({"p": 2}), EnqueueOpts::default()) + .await + .unwrap(); + let h = queue.claim_one("w1").await.unwrap().unwrap(); + let claimed_first = h.job().id; + assert!(claimed_first == pending || claimed_first == processing); + let still_live = if claimed_first == pending { + processing + } else { + pending + }; + assert!(queue.cancel(claimed_first).await.unwrap()); + assert!(queue.cancel(still_live).await.unwrap(), "unconditional"); + assert!(!queue.cancel(still_live).await.unwrap(), "gone = false"); + assert!(queue.get_job(still_live).await.unwrap().is_none()); + assert!(!h.ack().await.unwrap(), "the holder's next op refuses"); + + // ack_batch: per-id predicate, no worker filter, count returned. + let a = queue + .enqueue(json!({"a": 1}), EnqueueOpts::default()) + .await + .unwrap(); + let b = queue + .enqueue(json!({"b": 1}), EnqueueOpts::default()) + .await + .unwrap(); + let c = queue + .enqueue(json!({"c": 1}), EnqueueOpts::default()) + .await + .unwrap(); + let ha = queue.claim_one("w1").await.unwrap().unwrap(); + let hb = queue.claim_one("w2").await.unwrap().unwrap(); + assert_eq!(ha.job().id, a, "FIFO: the first-enqueued row claims first"); + assert_eq!(hb.job().id, b); + assert!(queue.cancel(c).await.unwrap()); + + // a is w1's claim, b is w2's — both count (no worker filter); the + // cancelled c and the missing id do not. + let count = queue.ack_batch(&[a, b, c, 999_999]).await.unwrap(); + assert_eq!(count, 2, "only the valid ids count; partial is ordinary"); + assert!(queue.get_job(a).await.unwrap().is_none()); + assert!(queue.get_job(b).await.unwrap().is_none()); + + // An empty batch is vacuously zero. + assert_eq!(queue.ack_batch(&[]).await.unwrap(), 0); + + drop_schema(&fx.admin, &fx.schema).await; +} + +/// Acceptance: `sweep_expired` moves every past-`expires_at` row (both +/// states — the no-stranded-rows property, ADR-010 §5) to dead with +/// `"expired"`, and returns the moved count; unexpired rows are +/// untouched; no leader lock required (the bare call just runs). +#[tokio::test(flavor = "multi_thread")] +async fn sweep_expired_moves_both_states_and_applies_retention() { + let Some(fx) = Fixture::open(&instance_namer("qs")).await else { + eprintln!("skip: no harness server"); + return; + }; + let queue = fx.store.queue("sweeping", opts()).await.unwrap(); + let now = now_unix(); + + // Expired pending + expired processing + unexpired: only the two + // expired rows move; the count is 2. The claim takes the head (the + // expired-pending row, FIFO) then the honest cross-connection probe + // back-dates both rows' expiry so the both-states move is + // exercised deterministically. + let expired_pending = queue + .enqueue( + json!({"k": 1}), + EnqueueOpts { + expires: Some(3600), + ..Default::default() + }, + ) + .await + .unwrap(); + let expired_processing = queue + .enqueue( + json!({"k": 2}), + EnqueueOpts { + expires: Some(3600), + ..Default::default() + }, + ) + .await + .unwrap(); + let unexpired = queue + .enqueue( + json!({"k": 3}), + EnqueueOpts { + expires: Some(3600), + ..Default::default() + }, + ) + .await + .unwrap(); + let h = queue.claim_one("w1").await.unwrap().unwrap(); + assert_eq!(h.job().id, expired_pending); + backdate_expires(&fx.admin, &fx.schema, expired_pending).await; + backdate_expires(&fx.admin, &fx.schema, expired_processing).await; + + let moved = queue.sweep_expired().await.unwrap(); + assert_eq!(moved, 2, "both expired rows moved (pending + processing)"); + + for id in [expired_pending, expired_processing] { + let dead = queue.get_job(id).await.unwrap().expect("moved to dead"); + assert_eq!(dead.state, JobState::Dead); + assert_eq!(dead.last_error.as_deref(), Some(EXPIRED_ERROR)); + assert!(dead.died_at.is_some() && dead.died_at.unwrap() >= now); + } + let live = queue.get_job(unexpired).await.unwrap().unwrap(); + assert_eq!(live.state, JobState::Pending, "unexpired rows untouched"); + + // A bare sweep on an empty-expiry queue is 0 and not an error. + assert_eq!(queue.sweep_expired().await.unwrap(), 0); + + drop_schema(&fx.admin, &fx.schema).await; +} + +/// Acceptance: retention — dead rows past their +/// `dead_letter_retention_s` stamp are deleted by the sweep (the only +/// sweeper dead rows ever have), rows with `None` retention live +/// forever; the deleted count reports through (ADR-010 §4/§5). +#[tokio::test(flavor = "multi_thread")] +async fn sweep_enforces_dead_letter_retention_incl_forever_none() { + let Some(fx) = Fixture::open(&instance_namer("qr")).await else { + eprintln!("skip: no harness server"); + return; + }; + let keeping = fx + .store + .queue( + "keeping", + QueueOpts { + dead_letter_retention_s: None, + ..opts() + }, + ) + .await + .unwrap(); + let expiring = fx + .store + .queue( + "expiring", + QueueOpts { + dead_letter_retention_s: Some(60), + ..opts() + }, + ) + .await + .unwrap(); + + let keep_id = keeping + .enqueue(json!({"k": 1}), EnqueueOpts::default()) + .await + .unwrap(); + let expire_id = expiring + .enqueue(json!({"k": 2}), EnqueueOpts::default()) + .await + .unwrap(); + let h = keeping.claim_one("w1").await.unwrap().unwrap(); + assert!(h.fail(None).await.unwrap()); + let h = expiring.claim_one("w1").await.unwrap().unwrap(); + assert!(h.fail(None).await.unwrap()); + + // Move the expired-retention row's death back past its TTL (the + // honest probe: retention reads died_at vs now). + fx.admin + .execute( + &format!( + "UPDATE {} SET died_at = $2 WHERE id = $1", + crate::schema::QualifiedTable { + schema: &fx.schema, + name: crate::schema::tables::DEAD, + } + ), + &[&expire_id, &(now_unix() - 3600)], + ) + .await + .unwrap(); + + // The sweep on the expiring queue deletes the TTL'd row; the + // keeping queue's row (retention None) survives. + assert_eq!(expiring.sweep_expired().await.unwrap(), 1); + assert!(expiring.get_job(expire_id).await.unwrap().is_none()); + assert_eq!(keeping.sweep_expired().await.unwrap(), 0); + assert!( + keeping.get_job(keep_id).await.unwrap().is_some(), + "retention None = dead rows live forever" + ); + + drop_schema(&fx.admin, &fx.schema).await; +} + +/// Acceptance: the wake-driven claim loop pinned end-to-end — enqueue +/// → pg_notify on the queue's wake channel → the LISTEN forwarder fan +/// → cross-connection wake → the claim (the LISTEN-driven default +/// consumption posture, ADR-004; the contract claim ops are +/// pull-shaped — a consumer's loop listens and claims on wake; the +/// re-poll safety net is the loop's fallback, pinned here only as the +/// pull-shaped surface's honest shape). +#[tokio::test(flavor = "multi_thread")] +async fn wake_driven_claim_loop_end_to_end() { + let Some(fx) = Fixture::open(&instance_namer("qw")).await else { + eprintln!("skip: no harness server"); + return; + }; + let queue = fx.store.queue("wakeup", opts()).await.unwrap(); + let mut receiver = fx.store.listen("wakeup").await.unwrap(); + + // Enqueue on the handle (a pooled connection, not the listener's): + // the row lands, the wake notifies on the queue's wake channel. + let id = queue + .enqueue(json!({"work": true}), EnqueueOpts::default()) + .await + .unwrap(); + + // The wake arrives cross-connection (the enqueue's NOTIFY fired at + // its implicit transaction's commit and the forwarder forwarded + // it). + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); + loop { + match receiver.recv_timeout(Duration::from_millis(300)).await { + Ok(Some(wake)) => { + assert_eq!(wake.channel, "wakeup"); + break; + } + Ok(None) => {} + Err(e) => panic!("receiver errored: {e}"), + } + assert!( + tokio::time::Instant::now() < deadline, + "the enqueue's wake never arrived" + ); + } + + // The claim lands on the consumer's own pooled connection. + let handle = queue.claim_one("w1").await.unwrap().expect("claimable"); + assert_eq!(handle.job().id, id); + assert!(handle.ack().await.unwrap()); + assert!(queue.get_job(id).await.unwrap().is_none()); + + // A poll-only consumer (the safety net's shape — no LISTEN wake at + // all, just a claim loop) still lands the same work: a second + // enqueue is claimed by an unreactive loop alone. + let second = queue + .enqueue(json!({"work": 2}), EnqueueOpts::default()) + .await + .unwrap(); + let deadline = tokio::time::Instant::now() + Duration::from_secs(2); + let claimed = loop { + if let Some(h) = queue.claim_one("w2").await.unwrap() { + break h; + } + assert!( + tokio::time::Instant::now() < deadline, + "the poll-only loop never landed the row" + ); + tokio::time::sleep(Duration::from_millis(50)).await; + }; + assert_eq!(claimed.job().id, second); + + drop_schema(&fx.admin, &fx.schema).await; +} + +/// Acceptance: entry-point validation — the constructor rejects empty +/// and reserved names (`InvalidName`/`ReservedName`), claims reject +/// empty `worker_id` (ADR-008 §4; ADR-019 §2) — nothing round-trips. +#[tokio::test(flavor = "multi_thread")] +async fn validation_rejects_bad_names() { + let Some(fx) = Fixture::open(&instance_namer("qval")).await else { + eprintln!("skip: no harness server"); + return; + }; + + match fx.store.queue("", opts()).await { + Err(alkstore::Error::InvalidName { .. }) => {} + _other => panic!("empty queue name must be InvalidName"), + } + match fx.store.queue(alkstore::RESERVED_PREFIX, opts()).await { + Err(alkstore::Error::ReservedName { .. }) => {} + _other => panic!("reserved queue name must be ReservedName"), + } + + let queue = fx.store.queue("valid", opts()).await.unwrap(); + match queue.claim_batch("", 1).await { + Err(alkstore::Error::InvalidName { .. }) => {} + _other => panic!("empty worker_id must be InvalidName"), + } + // The non-positive extent guards to the empty value even under a + // valid worker (ADR-023 §2). + let got = queue.claim_batch("w1", 0).await.unwrap(); + assert!(got.is_empty()); + + drop_schema(&fx.admin, &fx.schema).await; +} + +/// Acceptance: closed-store fail-closed on all queue ops (the wave-3 +/// shape) — the constructor, enqueue, claims, ack_batch, cancel, +/// get_job, sweep, heartbeat. +#[tokio::test(flavor = "multi_thread")] +async fn closed_store_fails_closed_on_all_queue_ops() { + let Some(fx) = Fixture::open(&instance_namer("qcl")).await else { + eprintln!("skip: no harness server"); + return; + }; + let queue = fx.store.queue("valid", opts()).await.unwrap(); + let id = queue + .enqueue(json!({}), EnqueueOpts::default()) + .await + .unwrap(); + let handle = queue.claim_one("w1").await.unwrap().unwrap(); + + fx.store.close(); + + let err = fx + .store + .queue("valid", opts()) + .await + .err() + .expect("closed store refuses construction"); + assert!(matches!(err, alkstore::Error::Database(_)), "got {err:?}"); + let err = queue + .enqueue(json!({}), EnqueueOpts::default()) + .await + .expect_err("closed store refuses enqueue"); + assert!(matches!(err, alkstore::Error::Database(_)), "got {err:?}"); + let err = queue + .claim_batch("w1", 1) + .await + .err() + .expect("closed store refuses claim"); + assert!(matches!(err, alkstore::Error::Database(_)), "got {err:?}"); + let err = queue + .ack_batch(&[id]) + .await + .expect_err("closed store refuses ack_batch"); + assert!(matches!(err, alkstore::Error::Database(_)), "got {err:?}"); + let err = queue + .cancel(id) + .await + .expect_err("closed store refuses cancel"); + assert!(matches!(err, alkstore::Error::Database(_)), "got {err:?}"); + let err = queue + .get_job(id) + .await + .expect_err("closed store refuses get_job"); + assert!(matches!(err, alkstore::Error::Database(_)), "got {err:?}"); + let err = queue + .sweep_expired() + .await + .expect_err("closed store refuses sweep"); + assert!(matches!(err, alkstore::Error::Database(_)), "got {err:?}"); + let err = handle + .heartbeat(100) + .await + .expect_err("closed store refuses heartbeat"); + assert!(matches!(err, alkstore::Error::Database(_)), "got {err:?}"); + + drop_schema(&fx.admin, &fx.schema).await; +} + +/// Acceptance: queue scoping — two handles on two queues are isolated: +/// a claim on one queue never hands out another queue's row (the +/// queue-scoped claim predicate), `get_job` is queue-scoped, and +/// cancel/ack_batch/sweep are queue-scoped (the seam's scoping rule at +/// the auto-commit shape). +#[tokio::test(flavor = "multi_thread")] +async fn queue_handles_are_name_scoped() { + let Some(fx) = Fixture::open(&instance_namer("qsc")).await else { + eprintln!("skip: no harness server"); + return; + }; + let q1 = fx.store.queue("one", opts()).await.unwrap(); + let q2 = fx.store.queue("two", opts()).await.unwrap(); + let id1 = q1 + .enqueue(json!({"q": 1}), EnqueueOpts::default()) + .await + .unwrap(); + let id2 = q2 + .enqueue(json!({"q": 2}), EnqueueOpts::default()) + .await + .unwrap(); + + // q1's claim only ever sees q1's rows; the second claim is + // no-work. + let h = q1.claim_one("w1").await.unwrap().unwrap(); + assert_eq!(h.job().id, id1); + assert!(q1.claim_one("w1").await.unwrap().is_none()); + // q2 still hands out its own. + let h2 = q2.claim_one("w2").await.unwrap().unwrap(); + assert_eq!(h2.job().id, id2); + + // get_job is queue-scoped: q1 cannot see q2's row. + assert!(q1.get_job(id2).await.unwrap().is_none()); + assert!(q2.get_job(id1).await.unwrap().is_none()); + assert!(q1.get_job(id1).await.unwrap().is_some()); + + // Cancel/ack_batch are queue-scoped too. + assert!(!q1.cancel(id2).await.unwrap(), "cancel is q-scoped"); + assert!(q2.cancel(id2).await.unwrap()); + assert_eq!(q1.ack_batch(&[id2]).await.unwrap(), 0); + + drop_schema(&fx.admin, &fx.schema).await; +} + +/// The dead-letter string constants are the contract's pinned values +/// (the engine-default strings land identically on both engines — the +/// backlog row's wording; server-less pin, the behavioral rows above +/// exercise them through the harness). +#[test] +fn dead_letter_strings_match_the_pinned_defaults() { + assert_eq!(EXHAUSTED_ERROR, "max attempts exceeded"); + assert_eq!(FAILED_ERROR, "failed"); + assert_eq!(EXPIRED_ERROR, "expired"); +} + +/// The backoff formula's unit pins (server-less): the curve's +/// attempt-index clamping and the pure formula's range behavior — the +/// statistical rows run harness-side. +#[test] +fn backoff_formula_unit_pins() { + // attempts 1 == the first retry draws from [ceil(base/2), base]. + for base in [1, 2, 5, 100] { + for id in 0..200 { + let d = backoff_delay_s(base, 1, id); + assert!(((base + 1) / 2..=base).contains(&d), "base {base} d {d}"); + } + } + // A zero attempts index clamps to the first-retry attempt (the + // formula's `max(1)` floor — the draw's subsecond input varies, so + // pin the *range* the clamp produces, not call equality). + for id in 0..200 { + let d = backoff_delay_s(5, 0, id); + assert!( + (3..=5).contains(&d), + "clamped-attempts draw {d} out of range" + ); + } + // The cap saturates the doubling arithmetic (no overflow at any + // attempt count). + for attempts in [30, 31, 60, 10_000] { + for id in 0..100 { + let d = backoff_delay_s(1, attempts, id); + assert!((1..=3600).contains(&d)); + } + } +} diff --git a/alkstore-postgres/src/tx.rs b/alkstore-postgres/src/tx.rs index 72c7cd7..b97fd9d 100644 --- a/alkstore-postgres/src/tx.rs +++ b/alkstore-postgres/src/tx.rs @@ -192,20 +192,10 @@ impl PgTxHandle { .to_string() } - fn enqueue_row_sql(&self) -> String { - format!( - "INSERT INTO {} (queue, payload, run_at, priority, max_attempts, - expires_at, created_at, visibility_timeout_s, - backoff_base_s, dead_letter_retention_s) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) - RETURNING id", - self.table(tables::JOB) - ) - } - /// The plain-queue enqueue's shared write (the `enqueue_tx` / /// `outbox_enqueue_tx` INSERT), called with this handle's - /// qualified job table. + /// qualified job table — the module-level [`enqueue_row`]'s + /// handle-side bridge. async fn enqueue_row( &self, client: &tokio_postgres::Client, @@ -214,31 +204,62 @@ impl PgTxHandle { opts: &EnqueueOpts, base: crate::resolution::Stamps, ) -> Result { - let now = crate::resolution::now_unix(); - let ResolvedEnqueue { run_at, expires_at } = resolve_enqueue_opts(now, opts); - let stamps = stamps_with_override(base, opts); - let row = client - .query_one( - &self.enqueue_row_sql(), - &[ - &queue, - &bytes, - &run_at, - &opts.priority, - &stamps.max_attempts, - &expires_at, - &now, - &stamps.visibility_timeout_s, - &stamps.backoff_base_s, - &stamps.dead_letter_retention_s, - ], - ) - .await - .map_err(pg_error)?; - Ok(row.get(0)) + enqueue_row(client, &self.schema, queue, bytes, opts, base).await } } +/// The enqueue INSERT — the one owner of the plain-queue row write. +/// The tx paths (`enqueue_tx` over the plain-queue defaults, +/// `outbox_enqueue_tx` over the outbox's 60/5/5 set) and the auto-commit +/// `Queue::enqueue` over the handle's `QueueOpts` ride it; one clock +/// read per enqueue (the `now` the resolution and the write share). +pub(crate) async fn enqueue_row( + client: &tokio_postgres::Client, + schema: &str, + queue: &str, + bytes: Vec, + opts: &EnqueueOpts, + base: crate::resolution::Stamps, +) -> Result { + let now = crate::resolution::now_unix(); + let ResolvedEnqueue { run_at, expires_at } = resolve_enqueue_opts(now, opts); + let stamps = stamps_with_override(base, opts); + let row = client + .query_one( + &enqueue_row_sql(schema), + &[ + &queue, + &bytes, + &run_at, + &opts.priority, + &stamps.max_attempts, + &expires_at, + &now, + &stamps.visibility_timeout_s, + &stamps.backoff_base_s, + &stamps.dead_letter_retention_s, + ], + ) + .await + .map_err(pg_error)?; + Ok(row.get(0)) +} + +/// The enqueue INSERT's SQL (schema-qualified job table). +fn enqueue_row_sql(schema: &str) -> String { + format!( + "INSERT INTO {} (queue, payload, run_at, priority, max_attempts, + expires_at, created_at, visibility_timeout_s, + backoff_base_s, dead_letter_retention_s) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) + RETURNING id", + QualifiedTable { + schema, + name: tables::JOB + } + ) +} + /// Encode a trait-crossing payload `Value` into the exact stored bytes /// (the serde_json serialization — ADR-020 §4). The notify carriage /// reuses it: the JSON serialization is valid UTF-8 text by diff --git a/tasks/pg-engine-queues.md b/tasks/pg-engine-queues.md index db5c739..45e25ff 100644 --- a/tasks/pg-engine-queues.md +++ b/tasks/pg-engine-queues.md @@ -1,7 +1,7 @@ --- id: pg-engine-queues name: Postgres engine — queues (`Queue`, `FOR UPDATE SKIP LOCKED` claims, job handles, backoff curve) -status: pending +status: completed depends_on: [pg-engine-seam-tx, pg-engine-notify-listen] scope: moderate risk: medium @@ -95,28 +95,28 @@ tables and the forwarder's wake substrate: ## Acceptance Criteria -- [ ] `enqueue` resolves opts over the handle's stamps (the shared +- [x] `enqueue` resolves opts over the handle's stamps (the shared resolution arithmetic; `max_attempts` the one override); payload bytes exact; wake notify fires on the queue's channel -- [ ] `claim_one`/`claim_batch`: exactly-once under concurrency +- [x] `claim_one`/`claim_batch`: exactly-once under concurrency (two-workers-one-row; 6×40 concurrent drain, zero double-claims); ordering row pinned; reclaim consumes an attempt; deadline from the row's stamp; extent guard (`n <= 0` → empty `Vec`); pre-claim exhausted-reclaimable sweep -- [ ] `JobHandle`: validity predicate uniform (lapse refuses all ops, +- [x] `JobHandle`: validity predicate uniform (lapse refuses all ops, `Ok(false)`); heartbeat absolute-reset + late-refused; ack deletes; retry curve matches ADR-010 §3's range/cap (statistical + cap tests, the SQLite task's shape); dead-letter defaults (`"max attempts exceeded"`, `"failed"`, caller strings verbatim); dead rows `get_job`-visible -- [ ] `ack_batch` per-id predicate (mixed lapsed/valid ids); `cancel` +- [x] `ack_batch` per-id predicate (mixed lapsed/valid ids); `cancel` unconditional both states; `sweep_expired` both-states + retention (incl. forever-none) -- [ ] Wake-driven claim loop pinned end-to-end (enqueue → wake → +- [x] Wake-driven claim loop pinned end-to-end (enqueue → wake → claim, cross-connection) -- [ ] Validation (empty/reserved queue names, empty `worker_id`) + +- [x] Validation (empty/reserved queue names, empty `worker_id`) + closed-store fail-closed on all queue ops -- [ ] `cargo test -p alkstore-postgres` (harness server), clippy +- [x] `cargo test -p alkstore-postgres` (harness server), clippy `-D warnings`, fmt clean; gates green server-less ## References @@ -134,8 +134,110 @@ tables and the forwarder's wake substrate: ## Notes -> To be filled by implementation agent +- **The claim is one statement — the task's "one statement or a + CTE-scoped transaction — implementer's choice" resolved to the one + statement**: an `UPDATE … RETURNING` over an `id IN (SELECT … FOR + UPDATE SKIP LOCKED ORDER BY … LIMIT n)` subquery (the picked-rows + CTE inline). The exactly-once property and the ordering row are + what the tests pin; the statement shape rides. +- **The dead-letter moves are transactional — the #133 defect class + explicitly excluded on the pg arm**: the naive DELETE-then-INSERT + pair across bare auto-commit statements would strand a row in + *neither* table when the dead INSERT failed (the substrate's + "stranded in NEITHER table is the property" test row). All four + move paths (retry-at-budget, `fail`, the pre-claim exhausted- + reclaimable sweep, `sweep_expired`'s both-states move + retention) + run inside an explicit `BEGIN`…`COMMIT` frame through a pool + `in_tx` helper — the seam's pooled-object posture scoped to one op + (a failed `BEGIN`/`COMMIT`/`ROLLBACK` discards the object via + `Object::take`, the seam's unknowable-state arm verbatim; the + body's error arm rolls back and re-pools) — and the single-job ops + (`retry`/`fail`) ride one frame each: the validity-predicate row + re-read (`SELECT … FOR UPDATE`) + the dead INSERT + the live + DELETE commit atomically, and a reclaim racing the op resolves + inside the frame (the loser's predicate re-read sees 0 rows). +- **The claim frame includes the pre-claim sweep atomically**: the + at-budget reclaimables' dead-letter move and the claim statement + commit together (the sweep alone would strand on its second + statement's failure; together they're one atomic unit, and the + claim predicate skipping already-dead rows is automatic). +- **The jitter's randomness source (the flagged choice)**: the + SQLite engine's std-only `RandomState`-hash draw reused verbatim — + no `rand` dep (ADR-005's posture), the same formula body (the + exponent clamps at 30 so the doubling saturates; the draw is + uniform over the integerized `[ceil(half), cap]` inclusive), its + own `backoff_delay_s` (wave 5 pins the twins' equivalence). +- **The enqueue wake is best-effort, logged-and-swallowed** (the + stream task's twin posture): the durable row is the truth, the + re-poll safety net / the consumer's re-read covers a failed wake — + never may a wake fee fail (or stall) a committed enqueue. The wake + channel is the queue's own name (the mechanism-name-is-the-channel + realization); the wake payload is empty (`''` — the wake carries + no content, ADR-008 §3). +- **The ack_batch predicate rides `claim_expires_at >= now` with the + queue's name scope** (the seam's scoping rule at the auto-commit + shape): a queue handle acks its own rows only — matching the + queue-scoped `get_job`/`cancel`/sweep predicates (the tx seam's + scoping rule carried to every queue op). +- **Params riding SQL arithmetic need explicit `::bigint` casts + where they meet column expressions** (`$2::bigint + + visibility_timeout_s`, the run_at/heartbeat arithmetic, the + retention TTL subtraction) — the seam task's noted + extended-protocol param-typing posture (the server infers param + types from context; `$2 + col` where `$2` sits beside no literal + is the E42725 "operator is not unique" arm, observed and fixed in + this task's first harness run). +- **`enqueue_row` lifted from `PgTxHandle`'s inherent method into + the module-level function the auto-commit path shares** (one owner + of the INSERT — the tx paths bridge it through a one-line inherent + wrapper; the schema string is the parameter — the tx handle keeps + its field, the queue handle passes its own). +- **PgStore's `queue` wiring replaced the stub; the open task's + `store_trait_methods_are_wiring_stubs` test updated accordingly** + (the constructor pinned as wired alongside notify/listen/stream; + `outbox` is the remaining stub-message representative). +- **Test details**: the drain test opens sibling stores over the + same schema per task (the honest cross-instance worker shape — + queue rows are schema state, not store-instance state; the + bigserial id range is probed via a raw admin count, not a + `1..=n` loop); the expiry backdating rides admin UPDATEs + (schema-qualified); the wake-driven loop test pins + enqueue→notify→wake→claim cross-connection plus the poll-only + fallback's shape (a claim loop with no wake still lands work). ## Summary -> To be filled on completion \ No newline at end of file +Landed alkstore-postgres's queue mechanism: `queue.rs` (the +`PgQueueHandle` carrying its `QueueOpts` stamps; `enqueue` resolving +over the handle's stamps through the shared resolution arithmetic +with `encode_payload` at the seam and the best-effort +queue-channel wake; `claim_one`/`claim_batch` as the one-statement +`FOR UPDATE SKIP LOCKED` claim in a `BEGIN`-framed transaction with +the atomically-included pre-claim exhausted-reclaimable sweep, the +job's-own-stamp deadline, the attempt-counting, the ADR-023 §2 +extent guard, and the ADR-019 §2 worker-id validation; the full +`JobHandle` impl — one-shot `ack`/`retry`/`fail` over `in_tx` +frames with the uniform validity predicate + `FOR UPDATE` row +re-read + atomic dead-letter moves, repeatable absolute-reset +`heartbeat`, the engine-side equal-jitter backoff curve (std-only +`RandomState` jitter, capped 3600 s, flagged), the pinned +dead-letter default strings, dead-visible queue-scoped `get_job`, +unconditional `cancel`, worker-less per-id `ack_batch`, and the +both-states + retention `sweep_expired`), the `Store::queue` wiring +replacing its stub, and `enqueue_row` lifted to the shared +module-level INSERT owner. neither-table state). 15 new acceptance tests + 2 server-less unit +pins (89 lib tests green ×3 runs at 8/4/1 threads against the harness +server, server-less skips clean; workspace `cargo test` green), +covering the lifecycle + stamp resolutions, the extent guard, the +two-workers-one-row + 6×40 cross-instance drain exactly-once, the +ordering row, the statistical + capped + end-to-end backoff curve, +the healthy-renewal + lapsed-refusals + reclaim-consumes-an-attempt +predicate row, the D-12 pre-claim sweep, the dead-letter defaults + +dead-row `get_job` visibility, cancel + ack_batch mixed-worker +counts, the both-states sweep + retention + forever-none, the +wake-driven claim loop end-to-end (cross-connection wake → claim, +plus the poll-only fallback), queue-name scoping, validation +(empty/reserved queue names, empty `worker_id`), and the +closed-store fail-closed on all queue ops; `cargo clippy +--all-targets -- -D warnings` and `cargo fmt --check` clean +workspace-wide. \ No newline at end of file