diff --git a/alkstore-mem/src/outbox.rs b/alkstore-mem/src/outbox.rs index cb59e33..2b7dd00 100644 --- a/alkstore-mem/src/outbox.rs +++ b/alkstore-mem/src/outbox.rs @@ -155,6 +155,8 @@ mod tests { struct RecordingDelivery { saw_payload: Option>, stamps: Option<(i64, i64, i64, Option)>, + handover_deadline: Option, + manual_heartbeat: bool, outcome: Option>, } @@ -163,6 +165,8 @@ mod tests { Self { saw_payload: None, stamps: None, + handover_deadline: None, + manual_heartbeat: false, outcome: Some(outcome), } } @@ -182,8 +186,18 @@ mod tests { j.backoff_base_s, j.dead_letter_retention_s, )); + self.handover_deadline = j.claim_expires_at; + let beat = self.manual_heartbeat; let outcome = self.outcome.take().unwrap_or(Ok(())); - Box::pin(async move { outcome }) + Box::pin(async move { + if beat { + assert!( + job.heartbeat(600).await.unwrap(), + "the boxed JobHandle's own heartbeat stays available to the closure" + ); + } + outcome + }) } } @@ -287,6 +301,37 @@ mod tests { assert!(dead.died_at == 2_000 || dead.died_at > 0); } + #[tokio::test(flavor = "current_thread")] + async fn run_once_issues_no_engine_heartbeat_in_delivery() { + let store = MemStore::new(); + store.clock.freeze(1_000); + let mut ob = outbox(&store, "mail"); + ob.enqueue(serde_json::json!(1), EnqueueOpts::default()) + .await + .unwrap(); + let mut delivery = RecordingDelivery::with(Ok(())); + assert!(ob.run_once("w1", &mut delivery).await.unwrap()); + assert_eq!( + delivery.handover_deadline, + Some(1_060), + "handover carries the raw claim deadline — \ + run_once issues no engine-side renewal for the closure" + ); + assert!(store.state.queues.guard().live.is_empty(), "Ok ⇒ ack"); + let id = ob + .enqueue(serde_json::json!(2), EnqueueOpts::default()) + .await + .unwrap(); + let mut renewal = RecordingDelivery::with(Ok(())); + renewal.manual_heartbeat = true; + assert!(ob.run_once("w2", &mut renewal).await.unwrap()); + assert!( + store.state.queues.guard().live.is_empty(), + "renewal inside the window is the closure's obligation — \ + and the boxed handle's heartbeat carries it ({id} acked)" + ); + } + #[tokio::test(flavor = "current_thread")] async fn empty_worker_id_is_invalid() { let store = MemStore::new(); diff --git a/tasks/mem-engine-scheduler-outbox.md b/tasks/mem-engine-scheduler-outbox.md index f6d696e..d683d5b 100644 --- a/tasks/mem-engine-scheduler-outbox.md +++ b/tasks/mem-engine-scheduler-outbox.md @@ -1,7 +1,7 @@ --- id: mem-engine-scheduler-outbox name: Mem engine — scheduler (`schedule`/`unschedule`/`run_schedules`) and outbox helper -status: pending +status: completed depends_on: [mem-engine-queues, mem-engine-locks] scope: moderate risk: medium @@ -56,16 +56,16 @@ derived-name format. ## Acceptance Criteria -- [ ] Scheduler surface + tick: boundary fire, catch-up (≤64), +- [x] Scheduler surface + tick: boundary fire, catch-up (≤64), skip-forward — unit-pinned with tokio-time determinism -- [ ] Leadership contention through the `__alkstore_scheduler` lock; +- [x] Leadership contention through the `__alkstore_scheduler` lock; `LeadershipLost` on loss, `Ok(())` on clean stop — pinned -- [ ] Schedule name + queue argument validation; `@every`-only parser +- [x] Schedule name + queue argument validation; `@every`-only parser (cron → `InvalidSpec`) — pinned -- [ ] Outbox: `enqueue`/`run_once` semantics (no engine heartbeat); +- [x] Outbox: `enqueue`/`run_once` semantics (no engine heartbeat); derived backing-queue name format identical to both engines — pinned -- [ ] `cargo check --target wasm32-unknown-unknown -p alkstore-mem` +- [x] `cargo check --target wasm32-unknown-unknown -p alkstore-mem` clean; workspace gates green ## References @@ -80,8 +80,65 @@ derived-name format. ## Notes -> To be filled by implementation agent +> Filled by implementation agent, 2026-10-10. + +1. **The implementation already landed in the seam task** — the same + all-or-nothing trait-impl disposition the previous wave-6 mechanism + tasks recorded: the scheduler (`scheduler.rs`'s register table, + `run_tick`, the runner) and the outbox helper (`outbox.rs`) were + implemented with the seam's whole-engine scope; this task found its + mechanisms *implemented*. +2. **Coverage reconciliation**: already pinned by the seam suite — + the boundary-fire row (deterministic `run_tick` fires, ready-at- + fire-instant rows stamped with the registered values + plain + defaults, no-double-fire floor, fire+advance as one guarded op), + the bounded-catch-up row (600 s behind → exactly 64 fires, never + 300; skip-forward lands the next boundary strictly future with the + grid honest; never-due schedules untouched), the leadership rows + (two contenders through the `__alkstore_scheduler` lock: one wins, + the loser `Err(LeadershipLost)`; a blocked lock loses before any + fire; clean stop → `Ok(())` and the lease released for the next + runner), the register rows (upsert-by-name + stamp resolution + + read-back; both name-bearing arguments validated + cron rejected + pre-storage with nothing written; the `@every` grammar matrix), + `unschedule` (removal + absent-false + reserved-name rejection), + `fired_rows_are_ordinary_queue_work` (the fire claims through the + ordinary queue), the outbox rows (Ok ⇒ ack / Err ⇒ retry with the + queue's equal-jitter curve and the delivery error riding the + exhaust path onto the dead row; the derived 60/5/5 stamps with + `max_attempts` the one override; no-work `Ok(false)`; empty + `worker_id` rejected), and the derived backing-queue name format + + `enqueue_tx` unreachability. +3. **What this task added** — the one unpinned posture from the + description: `run_once_issues_no_engine_heartbeat_in_delivery` — + the handover carries the raw claim deadline (no engine-side + renewal before or inside delivery), and the boxed + [`JobHandle`]'s own `heartbeat` remains available to the closure + (renewal is the consumer's obligation, ADR-019 §5). The delivery + fixture grew `handover_deadline` + a manual-heartbeat arm for it. +4. **No production code changed** — verified correct as-found; tests + only (+1 in `outbox.rs`, fixture widened, 100 mem tests total). + The scheduler's boundary waits ride realtime tokio time with + stamps on the engine clock; the tick/catch-up/leadership pins + drive the crate-internal `run_tick`/`run_schedules` directly, so + determinism holds without realtime waits. ## Summary -> To be filled on completion \ No newline at end of file +> Filled on completion, 2026-10-10. + +Completed the mem engine's scheduler/outbox task against the seam +task's foundation: reconciled the description's pin list against +`alkstore-mem/src/scheduler.rs` + `outbox.rs`'s existing suite (14 +pre-existing tests), found every tick/catch-up/leadership/validation/ +outbox shape already pinned, and added the one gap — the +no-engine-issued-heartbeat posture pinned end-to-end (raw claim +deadline at handover; the closure's own heartbeat path available and +honored). Acceptance criteria: scheduler surface + tick determinism ✓ +(seam rows), leadership ✓ (seam rows), validation + grammar ✓ (seam +rows), outbox semantics + derived name format ✓ (seam rows + the new +heartbeat row), gates ✓. Verified: `cargo test -p alkstore-mem` +100/100 (99 + 1 new, new test green solo), workspace `cargo test` +green, `cargo clippy --all-targets -- -D warnings` clean, +`cargo fmt --check` clean, `cargo check --target +wasm32-unknown-unknown -p alkstore-mem` clean. \ No newline at end of file