SQLite engine: queues — Queue/JobHandle over the writer slot, engine-side backoff curve, extent guard (task sqlite-engine-queues)

queue.rs: SqliteQueueHandle (QueueOpts-carrying, stamp-resolving
enqueue over the seam's shared resolution arithmetic), claim_one/
claim_batch through the writer slot with ADR-023 §2's extent guard
(n <= 0 -> empty Vec at the trait-impl entry), the full JobHandle impl
(one-shot ack/retry/fail as 'static boxed futures, repeatable absolute
reset heartbeat, substrate's uniform validity predicate), and the
engine-owned equal-jitter exponential backoff (std-only RandomState
hash jitter, integerized inclusive [ceil(half), cap], 1-hour cap;
no rand dep - documented).

ack_batch loses its substrate worker filter (register D-31 - ADR-019
§1's worker-less batch form); job_from_json decode + stamps_with_
override moved to their one owners (queue.rs / resolution.rs);
reader-pool helpers lifted from stream.rs into seam.rs. Store::queue
wired; 10 acceptance tests (lifecycle/stamps, extent guard,
exactly-once under concurrency, backoff range + cap + override,
validity predicate + reclaim, dead-letter defaults + get_job
visibility, cancel, ack_batch, sweep both-states + retention,
validation + closed-store). Workspace 25+3+171 green, clippy
-D warnings, fmt clean; sqlite suite 4x green.

alkstore-sqlite/src/store/queue_tests.rs

@@ -0,0 +1,999 @@
+//! The queue mechanism's acceptance tests (`sqlite-engine-queues`):
+//! the enqueue→claim→ack lifecycle works end-to-end over a temp-file
+//! store with stamps visible via `get_job` matching the resolution
+//! rules, the extent guard (ADR-023 §2) claims nothing on `n <= 0`,
+//! the exactly-once handout under concurrent claims (the POC-pinned
+//! property, engine-tested here), 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, cancel,
+//! `ack_batch`'s per-id predicate without a worker filter, and
+//! `sweep_expired`'s both-states move + retention count (ADR-010;
+//! ADR-019 §1–§3; ADR-020 §1–§3; ADR-023 §2).
+
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+use alkstore::{EnqueueOpts, JobState, QueueOpts, Store};
+use serde_json::{Value, json};
+
+use crate::queue::{EXHAUSTED_ERROR, EXPIRED_ERROR, FAILED_ERROR, backoff_delay_s};
+use crate::store::open_store;
+
+fn temp_dir(tag: &str) -> PathBuf {
+    let d = std::env::temp_dir().join(format!(
+        "alkstore-queue-{tag}-{}-{:?}",
+        std::process::id(),
+        std::thread::current().id()
+    ));
+    let _ = std::fs::remove_dir_all(&d);
+    std::fs::create_dir_all(&d).unwrap();
+    d
+}
+
+fn temp_path(tag: &str) -> PathBuf {
+    temp_dir(tag).join("store.db")
+}
+
+fn opts() -> QueueOpts {
+    QueueOpts {
+        visibility_timeout_s: 300,
+        max_attempts: 3,
+        backoff_base_s: 5,
+        dead_letter_retention_s: None,
+    }
+}
+
+/// 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, and `get_job` sees the states along
+/// the way (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 dir = temp_dir("lifecycle");
+    let path = temp_path("lifecycle");
+    let store = open_store(path.to_str().unwrap(), Default::default()).unwrap();
+
+    let qopts = QueueOpts {
+        visibility_timeout_s: 120,
+        max_attempts: 4,
+        backoff_base_s: 9,
+        dead_letter_retention_s: Some(7200),
+    };
+    let queue = store.queue("emails", qopts.clone()).await.unwrap();
+    assert_eq!(queue.name(), "emails");
+
+    // Delay wins over run_at (ADR-020 §1); relative expires resolves
+    // to the absolute row expiry (§2).
+    let before = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap()
+        .as_secs() as i64;
+
+    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),
+        "the handle's stamp"
+    );
+    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 (the resolved
+    // delay stamps the row).
+    assert!(queue.claim_one("w1").await.unwrap().is_none());
+
+    // A second enqueue with run_at in the past is claimable now, the
+    // one-clocked shape: relative expires resolves over the same
+    // instant as the ready time.
+    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.
+    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); the
+    // delayed row is still not claimable either.
+    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"
+    );
+
+    store.close();
+    let _ = std::fs::remove_dir_all(dir);
+}
+
+/// Acceptance: the extent guard (ADR-023 §2) — `claim_batch(n <= 0)`
+/// claims nothing (empty `Vec`, no error, never the whole queue — the
+/// `LIMIT -1` dialect artifact is dead) 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 dir = temp_dir("extent");
+    let path = temp_path("extent");
+    let store = open_store(path.to_str().unwrap(), Default::default()).unwrap();
+    let queue = store.queue("work", opts()).await.unwrap();
+
+    for i in 0..5 {
+        queue
+            .enqueue(json!({"i": i}), EnqueueOpts::default())
+            .await
+            .unwrap();
+    }
+
+    for n in [-1, 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);
+
+    store.close();
+    let _ = std::fs::remove_dir_all(dir);
+}
+
+/// Acceptance: exactly-once handout under concurrent claims — two
+/// workers racing claim on one row; exactly one wins (the POC-pinned
+/// property, engine-tested here through the writer slot), and a
+/// many-worker many-row drain hands every row out exactly once.
+#[tokio::test(flavor = "multi_thread")]
+async fn concurrent_claims_hand_each_row_out_exactly_once() {
+    let dir = temp_dir("exclusive");
+    let path = temp_path("exclusive");
+    let store: Arc<crate::store::SqliteStore> =
+        Arc::new(open_store(path.to_str().unwrap(), Default::default()).unwrap());
+
+    // Two workers, one row: exactly one claim wins.
+    {
+        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::<Vec<_>>();
+        assert_eq!(winners.len(), 1, "exactly one worker got the row");
+        assert_eq!(winners[0].job().id, id);
+    }
+
+    // Many workers drain a whole queue with no double handout.
+    const JOBS: i64 = 40;
+    const WORKERS: usize = 6;
+    {
+        let queue = store.queue("shared", opts()).await.unwrap();
+        for i in 0..JOBS {
+            queue
+                .enqueue(json!({"i": i}), EnqueueOpts::default())
+                .await
+                .unwrap();
+        }
+        let claimed = Arc::new(AtomicUsize::new(0));
+        let mut handles = Vec::new();
+        for w in 0..WORKERS {
+            let store = store.clone();
+            let claimed = claimed.clone();
+            handles.push(tokio::spawn(async move {
+                let queue = store.queue("shared", 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,
+            "every job handed out exactly once"
+        );
+        assert!(
+            shares.iter().any(|s| *s > 0) && shares.iter().any(|s| *s < JOBS as usize),
+            "work was shared across workers, not drained by one"
+        );
+
+        // No id in processing twice; every job is processing now.
+        let queue = store.queue("shared", opts()).await.unwrap();
+        let mut distinct = std::collections::HashSet::new();
+        for id in 2..(JOBS + 2) {
+            if let Some(job) = queue.get_job(id).await.unwrap() {
+                assert_eq!(
+                    job.state,
+                    JobState::Processing,
+                    "job {id} must be processing"
+                );
+                distinct.insert(job.id);
+            }
+        }
+        assert_eq!(distinct.len(), JOBS as usize, "no id claimed twice");
+    }
+
+    store.close();
+    let _ = std::fs::remove_dir_all(dir);
+}
+
+/// Acceptance: the backoff curve — `retry(err, None)` computes the
+/// equal-jitter exponential `delay ∈ [base·2^(a−1)/2, base·2^(a−1)]`
+/// from the job's stamped `backoff_base_s`, capped at 1 hour
+/// (ADR-010 §3): statistical pin over many draws, tolerance-bounded;
+/// the end-to-end row's `run_at` 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 → ceil(5/2)=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 dir = temp_dir("backoff");
+    let path = temp_path("backoff");
+    let store = open_store(path.to_str().unwrap(), Default::default()).unwrap();
+    let queue = 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 = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap()
+        .as_secs() as i64;
+    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 the curve: wait out the curve draw, claim
+    // again, retry with the explicit delay.
+    let deadline = tokio::time::Instant::now() + std::time::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 draw never elapsed"
+        );
+        tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+    };
+    assert!(override_landed);
+    let job = queue.get_job(id).await.unwrap().unwrap();
+    let now = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap()
+        .as_secs() as i64;
+    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
+    );
+
+    store.close();
+    let _ = std::fs::remove_dir_all(dir);
+}
+
+/// Wait until the claim deadline of the claimed row `id` on `queue`
+/// 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() + std::time::Duration::from_secs(6);
+    loop {
+        let lapsed = match queue.get_job(id).await {
+            Ok(Some(job)) => {
+                job.claim_expires_at.is_some()
+                    && job.claim_expires_at.unwrap()
+                        < std::time::SystemTime::now()
+                            .duration_since(std::time::UNIX_EPOCH)
+                            .unwrap()
+                            .as_secs() as i64
+            }
+            _ => false,
+        };
+        if lapsed {
+            return;
+        }
+        assert!(
+            tokio::time::Instant::now() < deadline,
+            "deadline: never lapsed"
+        );
+        tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+    }
+}
+
+/// Acceptance: the uniform validity predicate — once the claim
+/// deadline has lapsed, **all** handle ops refuse with `Ok(false)`
+/// (heartbeat, ack, retry, fail — never errors), the row is untouched,
+/// and a reclaim consumes an attempt (ADR-010 §2). Healthy renewal is
+/// pinned first: `heartbeat` is the repeatable `&self` case (ADR-019
+/// §3), an absolute reset; the one-shot ops consume their handles
+/// whatever the row's answer (ADR-019 §3's annotation).
+#[tokio::test(flavor = "multi_thread")]
+async fn lapsed_deadline_refuses_all_handle_ops_and_reclaim_consumes_an_attempt() {
+    let dir = temp_dir("validity");
+    let path = temp_path("validity");
+    let store = open_store(path.to_str().unwrap(), Default::default()).unwrap();
+
+    // Healthy phase (visibility 300): renewal lands and is repeatable.
+    let queue = 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 = 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<Box<dyn alkstore::JobHandle>> = 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;
+    }
+
+    // The lapsed handles refuse with Ok(false) — op by op.
+    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"));
+
+    // The original worker's side effects may complete, but its op
+    // cannot land now (reclaim won atomically): post-reclaim, the
+    // original claim's ops refuse.
+    assert!(
+        !claims[0].heartbeat(5000).await.unwrap(),
+        "the original claim's ops refuse post-reclaim"
+    );
+
+    store.close();
+    let _ = std::fs::remove_dir_all(dir);
+}
+
+/// Acceptance: dead-letter — `retry` at exhausted budget moves the row
+/// to dead with the pinned default string (`"max attempts exceeded"`),
+/// `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's annotation).
+#[tokio::test(flavor = "multi_thread")]
+async fn exhaustion_and_fail_dead_letter_with_the_pinned_strings() {
+    let dir = temp_dir("dead-letter");
+    let path = temp_path("dead-letter");
+    let store = open_store(path.to_str().unwrap(), Default::default()).unwrap();
+    let queue = 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: reschedules"
+    );
+    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, None);
+
+    // 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: reschedules"
+    );
+    let h = queue.claim_one("w1").await.unwrap().unwrap();
+    assert_eq!(h.job().attempts, 2);
+    assert!(
+        h.retry(None, None).await.unwrap(),
+        "the at-budget retry 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());
+
+    store.close();
+    let _ = std::fs::remove_dir_all(dir);
+
+    // fail(None) pins "failed"; fail(Some(e)) carries the caller's.
+    let dir2 = temp_dir("dead-letter-2");
+    let path2 = temp_path("dead-letter-2");
+    let store = open_store(path2.to_str().unwrap(), Default::default()).unwrap();
+    let queue = 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"
+    );
+
+    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");
+
+    store.close();
+    let _ = std::fs::remove_dir_all(dir2);
+}
+
+/// Acceptance: `cancel` is unconditional (either state, any holder),
+/// `ack_batch` applies the per-id validity predicate without a worker
+/// filter (partial success ordinary), and missing ids are silent
+/// (ADR-010 §1; ADR-019 §1).
+#[tokio::test(flavor = "multi_thread")]
+async fn cancel_and_ack_batch_behaviors() {
+    let dir = temp_dir("cancel-ackbatch");
+    let path = temp_path("cancel-ackbatch");
+    let store = open_store(path.to_str().unwrap(), Default::default()).unwrap();
+    let queue = store.queue("ops", opts()).await.unwrap();
+
+    // cancel: pending and processing, holder-independent. The first
+    // claim takes the head of the FIFO order (equal run_at, id ASC →
+    // the first-enqueued row) — observe, don't assume.
+    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(),
+        "cancel is 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 on the
+    // batch form); 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);
+
+    store.close();
+    let _ = std::fs::remove_dir_all(dir);
+}
+
+/// 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 dir = temp_dir("sweep");
+    let path = temp_path("sweep");
+    let store = open_store(path.to_str().unwrap(), Default::default()).unwrap();
+    let queue = store.queue("sweeping", opts()).await.unwrap();
+    let now = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap()
+        .as_secs() as i64;
+
+    // Expired pending + expired processing + unexpired: only the two
+    // expired rows move; the count is 2. `expires: Some(0)` resolves
+    // to expires_at = enqueue-instant — second-precision, so the sweep
+    // predicate (`<= unixepoch()`) sees it as lapsed from the next
+    // second on (the wait below makes that definite). The wait also
+    // holds off claims so the expired-pending row cannot be claimed
+    // first: only the not-yet-expired rows are claimable at that
+    // point.
+    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(1),
+                ..Default::default()
+            },
+        )
+        .await
+        .unwrap();
+    let unexpired = queue
+        .enqueue(
+            json!({"k": 3}),
+            EnqueueOpts {
+                expires: Some(3600),
+                ..Default::default()
+            },
+        )
+        .await
+        .unwrap();
+    // Claim the head (the expired-pending row, FIFO) then back-date
+    // both rows' expiry directly — the honest cross-connection probe —
+    // so the sweep's both-states move is exercised deterministically.
+    let h = queue.claim_one("w1").await.unwrap().unwrap();
+    assert_eq!(h.job().id, expired_pending);
+    {
+        let conn = rusqlite::Connection::open(&path).unwrap();
+        for id in [expired_pending, expired_processing] {
+            conn.execute(
+                "UPDATE __alkstore_live SET expires_at = unixepoch() - 1 WHERE id = ?1",
+                [id],
+            )
+            .unwrap();
+        }
+    }
+
+    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);
+
+    store.close();
+    let _ = std::fs::remove_dir_all(dir);
+}
+
+/// 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() {
+    let dir = temp_dir("retention");
+    let path = temp_path("retention");
+    let store = open_store(path.to_str().unwrap(), Default::default()).unwrap();
+    let keeping = store
+        .queue(
+            "keeping",
+            QueueOpts {
+                dead_letter_retention_s: None,
+                ..opts()
+            },
+        )
+        .await
+        .unwrap();
+    let expiring = store
+        .queue(
+            "expiring",
+            QueueOpts {
+                dead_letter_retention_s: Some(1),
+                ..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).
+    {
+        let conn = rusqlite::Connection::open(&path).unwrap();
+        conn.execute(
+            "UPDATE __alkstore_dead SET died_at = unixepoch() - 10 WHERE id = ?1",
+            [expire_id],
+        )
+        .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"
+    );
+
+    store.close();
+    let _ = std::fs::remove_dir_all(dir);
+}
+
+/// Acceptance: entry-point validation — the constructor rejects empty
+/// and reserved names (`InvalidName`/`ReservedName`), claims reject
+/// empty `worker_id`, and ops against a closed store fail closed with
+/// the engine-wide `Database` shape (ADR-008 §4; ADR-019 §2).
+#[tokio::test(flavor = "multi_thread")]
+async fn validation_and_closed_store_posture() {
+    let dir = temp_dir("validation");
+    let path = temp_path("validation");
+    let store = open_store(path.to_str().unwrap(), Default::default()).unwrap();
+
+    match store.queue("", opts()).await {
+        Err(alkstore::Error::InvalidName { .. }) => {}
+        _other => panic!("empty queue name must be InvalidName"),
+    }
+    match store.queue(alkstore::RESERVED_PREFIX, opts()).await {
+        Err(alkstore::Error::ReservedName { .. }) => {}
+        _other => panic!("reserved queue name must be ReservedName"),
+    }
+
+    let queue = store.queue("valid", opts()).await.unwrap();
+    match queue.claim_batch("", 1).await {
+        Err(alkstore::Error::InvalidName { .. }) => {}
+        _other => panic!("empty worker_id must be InvalidName"),
+    }
+
+    store.close();
+
+    // Closed store: all queue ops fail closed with Database.
+    let err = 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.get_job(1).await.expect_err("closed refuses get");
+    assert!(matches!(err, alkstore::Error::Database(_)), "got {err:?}");
+    let err = queue
+        .sweep_expired()
+        .await
+        .expect_err("closed refuses sweep");
+    assert!(matches!(err, alkstore::Error::Database(_)), "got {err:?}");
+
+    let _ = std::fs::remove_dir_all(dir);
+}