mem engine: pin notify/listen burst floor and close-at-drop arms (mem-engine-notify-listen)

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();