mem engine queues — the wave-6 mechanism task lands: coverage reconciliation against the seam task's foundation, with the six unpinned engine-side pins (mem-engine-queues)

The queue mechanism (ADR-010 state machine over the in-memory job
table, mem's third-owner equal-jitter curve + opts resolution)
already landed implemented with the seam task's whole-engine scope
(mem-engine-seam-tx Note 1); this task's work, per the disposition the
notify/listen twin recorded, was reconciling the description's pin
list against the existing suite and adding the missing engine-side
unit pins — no production code changed, the realization verified
correct as-found.

Already pinned by the seam task (reconciled, recorded in Notes):
lifecycle + stamps + resolved get_job values, claim ordering, extent
clamps at entry, worker-id validation, the uniform validity predicate,
late-heartbeat refusal + absolute-reset renewal, reclaim-eats-attempt
+ loser-ack-refuses, the pinned dead-letter default strings, dead-row
get_job visibility, both-states sweep + retention + scoping, the tx
stamp shapes, and the resolution legs + base-5 curve range/cap/spread.

Added (6): the in-process claim-race atomicity pin (8 concurrent
workers over 4 rows — every row handed exactly once through the
guarded section, the description's claim-vs-claim bullet); the
claim-vs-expiry boundary determinism pin (at now == deadline: the d <
now reclaim predicate and the d >= now holder-op predicate are
coherent — no reclaim, heartbeat/ack still valid; one second past:
reclaimable with its attempt consumed; lapsed ids silently uncounted
by ack_batch); the engine-side mirror of the suite's
backoff_curve_equivalence row — base 2 s at [1,2]/[2,4]/[4,8] driven
end-to-end through claim -> retry(None) -> get_job's run_at with the
Some(5)-verbatim and Some(0)-ready-now override legs, exact under the
frozen clock — both through the handle and pinned at backoff_delay_s
itself (the cap leg stays in resolution.rs's base-5 pins per
suite-opts-backoff-rows' recorded disposition); the D-12 pre-claim
sweep pin (a lapsed row with its budget spent claims nothing and
dead-letters with the exhaust string at the next claim, never
strands); and unconditional cancel across both states (the holder's
post-cancel ack refusing — the not-an-interrupt honesty). One
test-authoring note recorded: heartbeat's absolute reset moves the
deadline, so boundary legs stage on fresh rows to keep each
observation exact.

Task file: status completed, Notes (the seam-carried disposition +
reconciliation inventory + the six adds + the heartbeat-observation
note) + Summary filled. Verified: cargo test -p alkstore-mem 97/97
(91 + 6 new; new tests green solo x3), workspace cargo test green
(core 25, suite-harness 3, mem 97, pg 124+25+9, SQLite 191+25),
clippy --all-targets -D warnings clean, cargo fmt --check clean,
cargo check --target wasm32-unknown-unknown -p alkstore-mem clean

alkstore-mem/src/queue.rs

@@ -1247,6 +1247,230 @@ mod tests {
         assert_eq!(kept.state, JobState::Dead, "retention None = forever");
     }
 
+    #[tokio::test(flavor = "current_thread")]
+    async fn concurrent_claims_hand_out_work_exactly_once() {
+        let store = MemStore::new();
+        store.clock.freeze(1_000);
+        let handle = handle(&store, "q");
+        let mut ids = Vec::new();
+        for i in 0..4 {
+            ids.push(
+                handle
+                    .enqueue(serde_json::json!(i), ready_now())
+                    .await
+                    .unwrap(),
+            );
+        }
+        let mut tasks = Vec::new();
+        for w in 0..8 {
+            let state = store.state.clone();
+            let clock = store.clock.clone();
+            tasks.push(tokio::spawn(async move {
+                MemQueueHandle::new("q".to_string(), QueueOpts::default(), state, clock)
+                    .claim_batch(&format!("w{w}"), 1)
+                    .await
+                    .expect("claim")
+                    .into_iter()
+                    .map(|h| h.job().id)
+                    .collect::<Vec<i64>>()
+            }));
+        }
+        let mut handed = Vec::new();
+        for task in tasks {
+            handed.extend(task.await.expect("task joins"));
+        }
+        handed.sort();
+        ids.sort();
+        assert_eq!(
+            handed, ids,
+            "8 racing workers over 4 rows: every row handed once, never twice — \
+             the guarded section makes the in-process claim race row-atomic"
+        );
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn claim_vs_expiry_is_deterministic_at_the_row_boundary() {
+        let store = MemStore::new();
+        store.clock.freeze(1_000);
+        let handle = handle(&store, "q");
+        let first = handle
+            .enqueue(serde_json::json!("x"), ready_now())
+            .await
+            .unwrap();
+        let w1 = handle.claim_batch("w1", 1).await.unwrap().pop().unwrap();
+        let deadline = w1.job().claim_expires_at.unwrap();
+        assert_eq!(deadline, 1_300);
+        store.clock.advance(300);
+        let contested = handle.claim_batch("w2", 1).await.unwrap();
+        assert!(
+            contested.is_empty(),
+            "at now == deadline the claim is not lapsed (d < now is false) — no reclaim"
+        );
+        assert!(
+            w1.heartbeat(60).await.unwrap(),
+            "the same boundary second keeps the holder's ops valid (d >= now)"
+        );
+        assert!(w1.ack().await.unwrap(), "the boundary ack lands");
+        assert!(handle.get_job(first).await.unwrap().is_none());
+
+        let second = handle
+            .enqueue(serde_json::json!("y"), ready_now())
+            .await
+            .unwrap();
+        let w3 = handle.claim_batch("w3", 1).await.unwrap().pop().unwrap();
+        assert_eq!(w3.job().id, second);
+        let deadline2 = w3.job().claim_expires_at.unwrap();
+        assert_eq!(deadline2, 1_600);
+        store.clock.advance(301);
+        let reclaimed = handle.claim_batch("w4", 1).await.unwrap();
+        assert_eq!(
+            reclaimed.len(),
+            1,
+            "one second past the deadline the row is reclaimable"
+        );
+        assert_eq!(
+            reclaimed[0].job().attempts,
+            2,
+            "the reclaim consumes its attempt"
+        );
+        let lapsed_id = reclaimed[0].job().id;
+        store.clock.advance(301);
+        assert_eq!(
+            handle.ack_batch(&[lapsed_id]).await.unwrap(),
+            0,
+            "a lapsed id is silently not counted by the batch form"
+        );
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn the_equal_jitter_curve_mirrors_the_suite_rows_numeric_bounds() {
+        let store = MemStore::new();
+        store.clock.freeze(10_000);
+        let curve = MemQueueHandle::new(
+            "curve".to_string(),
+            QueueOpts {
+                max_attempts: 7,
+                backoff_base_s: 2,
+                ..QueueOpts::default()
+            },
+            store.state.clone(),
+            store.clock.clone(),
+        );
+        let id = curve
+            .enqueue(serde_json::json!({"leg": "curve"}), EnqueueOpts::default())
+            .await
+            .unwrap();
+        // Base 2 s at the first three attempts — the suite row's
+        // identical ranges ([1,2], [2,4], [4,8]); the frozen clock makes
+        // the observation exact (no straddle tolerance needed).
+        for (attempt, lower, upper) in [(1i64, 1i64, 2i64), (2, 2, 4), (3, 4, 8)] {
+            store.clock.advance(upper + 1);
+            let held = curve.claim_batch("w", 1).await.unwrap().pop().unwrap();
+            assert_eq!(
+                held.job().attempts,
+                attempt,
+                "the attempt index is the claim count"
+            );
+            let now = store.clock.unix_now();
+            assert!(held.retry(Some("boom".to_string()), None).await.unwrap());
+            let job = curve.get_job(id).await.unwrap().unwrap();
+            assert_eq!(job.state, JobState::Pending);
+            let observed = job.run_at - now;
+            assert!(
+                (lower..=upper).contains(&observed),
+                "attempt {attempt}: delay {observed} outside the suite row's [{lower}, {upper}]"
+            );
+        }
+        store.clock.advance(9);
+        let held = curve.claim_batch("w", 1).await.unwrap().pop().unwrap();
+        let now = store.clock.unix_now();
+        assert!(held.retry(Some("boom".to_string()), Some(5)).await.unwrap());
+        let job = curve.get_job(id).await.unwrap().unwrap();
+        assert_eq!(job.run_at, now + 5, "Some(5) overrides the curve verbatim");
+        store.clock.advance(6);
+        let held = curve.claim_batch("w", 1).await.unwrap().pop().unwrap();
+        let now = store.clock.unix_now();
+        assert!(held.retry(Some("boom".to_string()), Some(0)).await.unwrap());
+        let job = curve.get_job(id).await.unwrap().unwrap();
+        assert_eq!(
+            job.run_at, now,
+            "Some(0) resolves ready-now — the boundary-total rule"
+        );
+        assert_eq!(
+            curve.claim_batch("w", 1).await.unwrap().len(),
+            1,
+            "a ready-now row claims immediately"
+        );
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn the_pre_claim_sweep_dead_letters_exhausted_reclaimables() {
+        let store = MemStore::new();
+        store.clock.freeze(1_000);
+        let handle = handle(&store, "q");
+        let id = handle
+            .enqueue(
+                serde_json::json!("x"),
+                EnqueueOpts {
+                    max_attempts: Some(1),
+                    ..ready_now()
+                },
+            )
+            .await
+            .unwrap();
+        let claimed = handle.claim_batch("w1", 1).await.unwrap().pop().unwrap();
+        assert_eq!(claimed.job().attempts, 1);
+        store.clock.advance(301);
+        let reclaimed = handle.claim_batch("w2", 1).await.unwrap();
+        assert!(
+            reclaimed.is_empty(),
+            "a lapsed row with its budget spent claims nothing — it dead-letters instead"
+        );
+        let dead = handle.get_job(id).await.unwrap().unwrap();
+        assert_eq!(dead.state, JobState::Dead);
+        assert_eq!(
+            dead.last_error.as_deref(),
+            Some("max attempts exceeded"),
+            "exhaustion via reclaim pins the engine default string"
+        );
+        assert_eq!(dead.died_at, Some(1_301));
+        assert_eq!(dead.attempts, 1, "the row never got a second claim");
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn cancel_is_unconditional_across_states() {
+        let store = MemStore::new();
+        store.clock.freeze(1_000);
+        let handle = handle(&store, "q");
+        let pending = handle
+            .enqueue(serde_json::json!("p"), ready_now())
+            .await
+            .unwrap();
+        assert!(
+            handle.cancel(pending).await.unwrap(),
+            "a pending row cancels"
+        );
+        assert!(handle.get_job(pending).await.unwrap().is_none());
+        assert!(
+            !handle.cancel(999).await.unwrap(),
+            "a row-less id is Ok(false)"
+        );
+        let processing = handle
+            .enqueue(serde_json::json!("w"), ready_now())
+            .await
+            .unwrap();
+        let holder = handle.claim_batch("w1", 1).await.unwrap().pop().unwrap();
+        assert!(
+            handle.cancel(processing).await.unwrap(),
+            "a processing row cancels"
+        );
+        assert!(
+            !holder.ack().await.unwrap(),
+            "cancel is not an interrupt — the holder's next op refuses, the expiry shape"
+        );
+        assert!(handle.get_job(processing).await.unwrap().is_none());
+    }
+
     #[tokio::test(flavor = "current_thread")]
     async fn sweep_is_scoped_to_its_queue() {
         let store = MemStore::new();

alkstore-mem/src/resolution.rs

@@ -218,6 +218,22 @@ mod tests {
         assert_eq!(resolve_enqueue_opts(now, &none).expires_at, None);
     }
 
+    #[test]
+    fn backoff_base_two_lands_in_the_suite_rows_exact_bounds() {
+        // The `backoff_curve_equivalence` suite row's identical ranges
+        // (base 2 s: [1,2], [2,4], [4,8]) — the engine-side numeric pin
+        // of the equivalence the row asserts across engines.
+        for (attempts, lower, upper) in [(1i64, 1i64, 2i64), (2, 2, 4), (3, 4, 8)] {
+            for id in [1i64, 2, 42, 999_999] {
+                let delay = backoff_delay_s(2, attempts, id);
+                assert!(
+                    (lower..=upper).contains(&delay),
+                    "base 2 attempt {attempts} id {id}: delay {delay} outside [{lower}, {upper}]"
+                );
+            }
+        }
+    }
+
     #[test]
     fn backoff_lands_in_the_pinned_equal_jitter_range() {
         for attempts in [1i64, 2, 3, 5, 30] {