From 61a1148165313df32edc408a5b94206573b08faf Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Sat, 10 Oct 2026 12:15:04 +0000 Subject: [PATCH] =?UTF-8?q?mem=20engine=20queues=20=E2=80=94=20the=20wave-?= =?UTF-8?q?6=20mechanism=20task=20lands:=20coverage=20reconciliation=20aga?= =?UTF-8?q?inst=20the=20seam=20task's=20foundation,=20with=20the=20six=20u?= =?UTF-8?q?npinned=20engine-side=20pins=20(mem-engine-queues)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 224 +++++++++++++++++++++++++++++++++ alkstore-mem/src/resolution.rs | 16 +++ tasks/mem-engine-queues.md | 108 ++++++++++++++-- 3 files changed, 340 insertions(+), 8 deletions(-) diff --git a/alkstore-mem/src/queue.rs b/alkstore-mem/src/queue.rs index ec84654..2b37a40 100644 --- a/alkstore-mem/src/queue.rs +++ b/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::>() + })); + } + 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(); diff --git a/alkstore-mem/src/resolution.rs b/alkstore-mem/src/resolution.rs index 71370fc..12029b1 100644 --- a/alkstore-mem/src/resolution.rs +++ b/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] { diff --git a/tasks/mem-engine-queues.md b/tasks/mem-engine-queues.md index 3ad9063..c3b0601 100644 --- a/tasks/mem-engine-queues.md +++ b/tasks/mem-engine-queues.md @@ -1,7 +1,7 @@ --- id: mem-engine-queues name: Mem engine — queues (in-memory job table, ADR-010 state machine, engine-owned backoff curve) -status: pending +status: completed depends_on: [mem-engine-seam-tx, mem-engine-notify-listen] scope: moderate risk: medium @@ -57,15 +57,15 @@ machine realized over an in-memory job table — all driver-free. ## Acceptance Criteria -- [ ] Full `Queue`/`JobHandle` surfaces over the in-memory job table -- [ ] The ADR-010 state machine properties unit-pinned: visibility +- [x] Full `Queue`/`JobHandle` surfaces over the in-memory job table +- [x] The ADR-010 state machine properties unit-pinned: visibility reclaim consumes an attempt, dead-letter moves, no-stranded-rows sweep, handle-op validity predicate, late-heartbeat boundary -- [ ] Equal-jitter curve + opts resolution pinned to the identical +- [x] Equal-jitter curve + opts resolution pinned to the identical numeric behavior the suite's equivalence rows assert (per the engine-side pin posture used for the cap legs) -- [ ] Numeric-domain guards (extents) at entry — pinned -- [ ] `cargo check --target wasm32-unknown-unknown -p alkstore-mem` +- [x] Numeric-domain guards (extents) at entry — pinned +- [x] `cargo check --target wasm32-unknown-unknown -p alkstore-mem` clean; workspace gates green ## References @@ -79,8 +79,100 @@ machine realized over an in-memory job table — all driver-free. ## Notes -> To be filled by implementation agent +> Filled by implementation agent, 2026-10-10. + +1. **The implementation already landed in the seam task** — + `mem-engine-seam-tx` Note 1 records that the mem engine's full + `Store` surface (including the `MemQueueHandle`/`MemJobHandle` + mechanism in `alkstore-mem/src/queue.rs` and mem's third-owner + `resolution.rs` arithmetic) is all-or-nothing per trait impl, so + this task found its mechanism *implemented*. Its work was the + coverage reconciliation against the description's pin list plus the + missing engine-side unit pins — the same disposition the + notify/listen twin task recorded (`mem-engine-notify-listen` + Notes 1–3). +2. **Coverage reconciliation**: already pinned by the seam task's + suite in `queue.rs`/`resolution.rs`/`tx.rs` — the full lifecycle + + stamp resolutions (300/3/5/none + per-job override, resolved + values shown by `get_job`), claim ordering (priority DESC → + ready-time ASC → id ASC), extent clamps (`n = 0`/`n < 0` → empty + at the entry, before any effect), empty-`worker_id` rejection, the + uniform validity predicate (lapse refuses all ops, refusals leave + the row intact), late-heartbeat refusal + absolute-reset renewal + (with `claimed_at`/`attempts` untouched), reclaim-consumes-an- + attempt + the loser's ack refusing (the dual-execution honesty), + retry/fail default strings (`"max attempts exceeded"` / `"failed"`) + + caller strings via the exhaust path, dead-row `get_job` + visibility with `last_error`/`died_at`, the both-states + no-stranded-rows sweep + retention (incl. forever-`None`) + queue + scoping, the tx/no-handle stamp shapes (plain 300/3/5/none, + outbox 60/5/5 in `tx.rs`), and the resolution legs (delay-wins- + over-`run_at`, `run_at` literal, neither→now, negative-delay total, + relative-`expires`) + the base-5 curve range/cap/spread pins in + `resolution.rs`. +3. **What this task added** — the six unpinned shapes: (a) + `concurrent_claims_hand_out_work_exactly_once` (8 racing workers + over 4 rows via spawned `claim_batch` handles: every row exactly + once — the guarded-section claim-vs-claim atomicity pin the + description's "claim race is in-process" bullet asks for); (b) + `claim_vs_expiry_is_deterministic_at_the_row_boundary` (at + `now == claim_expires_at`: no reclaim (`d < now` false), the + holder's heartbeat/ack still valid (`d >= now`) — the two + predicates' boundary coherence; one second past: reclaimable with + its attempt consumed; the lapsed id silently uncounted by + `ack_batch`); (c) `the_equal_jitter_curve_mirrors_the_suite_rows_ + numeric_bounds` (base 2, `max_attempts` 7 — the suite row's own + shape: three claim→`retry(None)` cycles asserting [1,2]/[2,4]/[4,8] + read back through `get_job`'s `run_at`, exact under the frozen + clock; plus the override legs `Some(5)` verbatim and `Some(0)` + ready-now per the boundary-total rule) — the engine-side pin of + AC3's "identical numeric behavior"; the cap leg stays pinned in + `resolution.rs`'s base-5 tests per `suite-opts-backoff-rows`'s + recorded disposition; (d) `backoff_base_two_lands_in_the_suite_ + rows_exact_bounds` in `resolution.rs` — the same base-2 bounds at + the arithmetic function itself; (e) `the_pre_claim_sweep_dead_ + letters_exhausted_reclaimables` (the D-12 shape: a lapsed row with + its budget spent claims nothing and dead-letters with the exhaust + string at the *next* claim — never strands claimable); (f) + `cancel_is_unconditional_across_states` (pending + processing + deleted, row-less `Ok(false)`, the holder's post-cancel ack + refusing — the not-an-interrupt honesty). +4. **No production code changed** — the seam task's realization was + verified correct as-found; this task is tests only (+5 in + `queue.rs`, +1 in `resolution.rs`, 97 mem tests total). One + test-authoring note: `heartbeat`'s absolute reset moves the + deadline (`now + extend`), so a boundary-reclaim leg staged after a + heartbeat on the *same* row reads the shifted deadline — the + boundary test stages its legs on fresh rows to keep each + observation exact. ## Summary -> To be filled on completion \ No newline at end of file +> Filled on completion, 2026-10-10. + +Completed the mem engine's queues task against the seam task's +foundation: reconciled the description's pin list against +`alkstore-mem/src/queue.rs` + `resolution.rs` + `tx.rs`'s existing +suite (13 pre-existing queue mechanism tests + the resolution and tx +stamp pins), found the full `Queue`/`JobHandle` surfaces and nearly +all ADR-010 properties already pinned, and added the six gaps: the +in-process claim-race atomicity pin (8 concurrent workers, exact-once +handout over the guarded section), the claim-vs-expiry boundary +determinism pin (the `d < now` / `d >= now` predicates' edge-second +coherence + the lapsed `ack_batch` non-count), the engine-side mirror +of the suite's `backoff_curve_equivalence` row (base 2 → [1,2]/[2,4]/ +[4,8] end-to-end through claim→`retry(None)`→`get_job.run_at` with +the `Some(5)`/`Some(0)` override legs, and the same bounds pinned at +`backoff_delay_s` itself; the 1-hour cap stays in `resolution.rs`'s +base-5 pins per the recorded suite disposition), the D-12 pre-claim +exhausted-reclaimable dead-letter sweep, and the both-states +unconditional `cancel`. Acceptance criteria: full surfaces ✓ (seam +task, lifecycle test), ADR-010 properties ✓ (seam tests + the new +boundary/reclaim/sweep pins), curve + opts resolution numeric +equivalence ✓ (new base-2 pins), extent guards ✓ (seam clamp test), +gates ✓. Verified: `cargo test -p alkstore-mem` 97/97 (91 + 6 new, +new tests green solo ×3), workspace `cargo test` green (core 25, +suite-harness 3, mem 97, pg 124+25+9, SQLite 191+25), +`cargo clippy --all-targets -- -D warnings` clean, `cargo fmt --check` +clean, `cargo check --target wasm32-unknown-unknown -p alkstore-mem` +clean — the wave-6 wasm gate stays green. \ No newline at end of file