From 4aa9023ca4f09d556ff0b71aacf90c313f9f16be Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Sat, 10 Oct 2026 12:01:26 +0000 Subject: [PATCH] mem engine: pin notify/listen burst floor and close-at-drop arms (mem-engine-notify-listen) --- alkstore-mem/src/notify.rs | 58 ++++++++++++++++++++++++++ tasks/mem-engine-notify-listen.md | 69 +++++++++++++++++++++++++++++-- 2 files changed, 124 insertions(+), 3 deletions(-) diff --git a/alkstore-mem/src/notify.rs b/alkstore-mem/src/notify.rs index 0839707..b87b86f 100644 --- a/alkstore-mem/src/notify.rs +++ b/alkstore-mem/src/notify.rs @@ -215,6 +215,64 @@ mod tests { ); } + #[tokio::test(flavor = "current_thread")] + async fn burst_delivers_at_least_one_wake_and_never_coalesces() { + let store = MemStore::new(); + let mut rx = listener(&store, "orders").await; + for n in 0..5 { + store.notify("orders", serde_json::json!(n)).await.unwrap(); + } + let wake = tokio::time::timeout(std::time::Duration::from_millis(200), rx.recv()) + .await + .expect("the burst delivers at least one wake — the floor pin") + .expect("the wake arrives with content"); + assert_eq!(wake.channel, "orders"); + let mut delivered = 1; + while rx.try_recv().unwrap().is_some() { + delivered += 1; + } + assert_eq!( + delivered, 5, + "mem's column is per-notify: every commit's wake surfaces, never coalesced" + ); + } + + #[tokio::test(flavor = "current_thread")] + async fn engine_drop_closes_every_open_receiver_on_the_terminal_arms() { + let store = MemStore::new(); + let mut orders = listener(&store, "orders").await; + let mut payments = listener(&store, "payments").await; + store.notify("orders", serde_json::json!(1)).await.unwrap(); + assert!( + orders.recv().await.is_some(), + "pre-drop delivery works on the same receiver that closes below" + ); + drop(store); + assert!( + orders.recv().await.is_none(), + "terminal None — engine drop is the one close path; no reconnect-wake ever follows" + ); + assert!( + payments.recv().await.is_none(), + "the close is whole-engine: every open receiver, not per-channel" + ); + // The closed state stays stable across the other two shapes, and + // distinguishes itself from idle (Err(Closed), never Ok(None)). + assert!(matches!( + orders + .recv_timeout(std::time::Duration::from_millis(30)) + .await, + Err(alkstore::Error::Closed) + )); + assert!(matches!(payments.try_recv(), Err(alkstore::Error::Closed))); + assert!(matches!( + payments + .recv_timeout(std::time::Duration::from_millis(30)) + .await, + Err(alkstore::Error::Closed) + )); + } + #[tokio::test(flavor = "current_thread")] async fn receiver_drop_detaches_its_feed() { let store = MemStore::new(); diff --git a/tasks/mem-engine-notify-listen.md b/tasks/mem-engine-notify-listen.md index b5ee833..c43898d 100644 --- a/tasks/mem-engine-notify-listen.md +++ b/tasks/mem-engine-notify-listen.md @@ -1,7 +1,7 @@ --- id: mem-engine-notify-listen name: Mem engine — notify/listen surface (`WakeReceiver`, close-at-engine-drop) -status: pending +status: completed depends_on: [mem-engine-seam-tx] scope: narrow risk: medium @@ -66,8 +66,71 @@ terminal `None`, commit-boundary delivery timing through ## 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 `notify`/`listen`/`notify_tx` and the + `MemWakeReceiver` bridge in `alkstore-mem/src/notify.rs`, the + `WakeBus` substrate in `wakes.rs`, and the tx overlay's staged + notify) is all-or-nothing per trait impl, so this task found its + mechanism *implemented*. Its work was the remaining engine-side + unit pins its description enumerates, against that foundation. +2. **Coverage reconciliation at pickup** — already pinned by the seam + task: channel-scoped delivery with the field carried + foreign- + channel silence (`structurally` silent on mem — the vacuous-column + analog of the suite row's scoping observable), never-coalesced + (per-notify) counts, attach-after-commit no-replay, entry + validation on both paths (empty/blank/reserved-prefix/the reserved + reconnect string rejected on `notify` and `listen`), + `PayloadTooLarge` never at 2 MiB (ADR-016 §5 non-occurrence), the + three recv forms' pinned arms (idle `Ok(None)`, timeout `Ok(None)`, + post-drop `try_recv → Err(Closed)` + `recv → None`), + receiver-drop-detaches, and the `notify_tx` commit-atomicity legs + (silence pre-merge → one wake post-merge, per-notify, rollback + no-ghosts, `with_tx` panics) in `tx.rs`. The FIFO-by-commit-order + property is pinned at the substrate with the commit watermark + (`wakes.rs`: `per_subscriber_fifo_by_commit_order`). +3. **What this task added** — the two unpinned shapes from the + description's list: (a) the **burst floor** — a 5-notify burst + delivers ≥1 wake through `recv` (the suite row's tolerance shape) + and drains exactly 5 through `try_recv` (mem's per-notify column, + the exact-count coalescing pin); (b) the dedicated **close-at-drop + pin** — engine drop closes *every* open receiver across channels + (terminal `None` on both, whole-engine not per-channel), the + closed state stable across all three recv shapes with + `recv_timeout`/`try_recv` carrying `Err(Closed)` (distinguishable + from idle) and no reconnect-wake ever following (no reconnection + concept; the reserved string stays rejected as a consumer name — + pinned in the seam task's validation test). +4. **No production code changed** — the seam task's realization was + verified correct as-found; this task is tests + surface review + only (`notify.rs` tests module, +2 tests). ## Summary -> To be filled on completion \ No newline at end of file +> Filled on completion, 2026-10-10. + +Completed the mem engine's notify/listen task against the seam task's +foundation: reconciled the description's pin list against +`alkstore-mem/src/notify.rs`'s existing suite (13 pre-existing +mechanism tests across `notify.rs` + the substrate pins in `wakes.rs` ++ the tx-atomicity legs in `tx.rs`), found everything but the burst +floor and the multi-receiver close-at-drop shape already pinned, and +added those two: `burst_delivers_at_least_one_wake_and_never_coalesces` +(floor ≥1 through `recv`, exact per-notify count through `try_recv`) +and `engine_drop_closes_every_open_receiver_on_the_terminal_arms` +(terminal `None` on every open receiver across channels, closed state +stable across `recv`/`try_recv`/`recv_timeout`, `Err(Closed)` +distinguishable from idle, no reconnect-wake). Acceptance criteria: +trait-complete surface + entry validation + core payload encoding ✓ +(seam task, validation test), commit-ordered channel-scoped no-replay +never-coalesced delivery ✓ (subsurface watermark test + store-level +pins), engine-drop close ✓ (now dedicated-pinned), `notify_tx` +post-merge-only delivery ✓ (seam task's `tx.rs` tests, criterion's +"pinned with the seam task's commit-atomicity tests" satisfied +as-written). Verified: `cargo test -p alkstore-mem` 91/91 (89 + 2 +new), workspace `cargo test` green (core 25, suite-harness 3, +mem 91, 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. \ No newline at end of file