pg suite-infra hardening (task pg-suite-infra-hardening): the wave-4 review's F-1 defense-in-depth candidates — must_recv_event re-shaped from the 20 ms try_recv polling loop to a parked recv().await under the 15 s timeout wrapper (stream_tests.rs, the sole pg copy — SQLite twin untouched; the poll loop's wake-subscription interaction surface the F-1 flake suspected is out of the hot path, and the deadline assert still bounds the property). The parked form's terminal arms are explicit: Some(Err(e)) keeps the errored-instead-of-idling panic, None gets a receiver-closed panic the poll form never saw. The deadline miss self-diagnoses (the small-honest realization of candidate (b)'s discriminator): read_since(offset, 1000) over rows beyond the receiver's position names the residual class — rows durable undelivered (wake/re-drain class) vs no rows (publish-visibility class); with the parked recv a wake-arrived-but-re-drain-missed arm is structurally implausible (only a read error could miss, and it surfaces Err). Candidate (b)'s literal bridge-side counter skipped, reason recorded in the task (surfacing the bridge wake count across the dyn EventReceiver boundary needs downcast/keyed-global machinery beyond the small-honest bar); the small honest piece did land: wait_wake's Lagged(n) arm now logs (eprintln, house posture matching notify.rs's bridge_loop — and forwarder.rs's 'the receiver bridge's Lagged arm logs / recovers' doc claim now true at both bridges), stream name threaded into wait_wake's signature so the log attributes the lag. Verified: pg stream module 21/21 vs harness server (full module run), tx_publishes_compose_with_the_handle 5 solo re-runs green (determinism re-check under the parked shape), two consecutive full pg harness runs green (121 lib + 25 suite + 9 schema, ~72 s each), cargo test -p alkstore-postgres green server-less (skips clean), workspace cargo test green (12 binaries), clippy -D warnings, fmt clean

alkstore-postgres/src/store/stream_tests.rs

@@ -86,22 +86,39 @@ fn test_opts(schema: &str) -> PgOpts {
     }
 }
 
-/// Wait for the next event with a bounded deadline — polling loosely
-/// so no tight-race assertions ride (the suite's stability posture).
+/// Wait for the next event with a bounded deadline — parked on the
+/// receiver's `recv()` under the timeout wrapper (no polling loop:
+/// the deadline still bounds the property, and the wave-4 F-1 flake's
+/// poll-loop wake-subscription surface is out of the hot path). A
+/// deadline miss self-diagnoses: the durable rows beyond the
+/// receiver's position discriminate a wake/re-drain class (rows exist
+/// undelivered) from a publish that never landed (none do).
 async fn must_recv_event(receiver: &mut dyn EventReceiver, tag: &str) -> alkstore::StreamEvent {
-    let deadline = tokio::time::Instant::now() + Duration::from_secs(15);
-    loop {
-        match receiver.try_recv() {
-            Ok(Some(event)) => return event,
-            Ok(None) => {}
-            Err(e) => panic!("{tag}: receiver errored instead of idling: {e}"),
-        }
-        assert!(
-            tokio::time::Instant::now() < deadline,
-            "{tag}: event never arrived"
-        );
-        tokio::time::sleep(Duration::from_millis(20)).await;
+    match tokio::time::timeout(Duration::from_secs(15), receiver.recv()).await {
+        Ok(Some(Ok(event))) => return event,
+        Ok(Some(Err(e))) => panic!("{tag}: receiver errored instead of idling: {e}"),
+        Ok(None) => panic!("{tag}: receiver closed before the event arrived"),
+        Err(_) => {}
     }
+    let offset = receiver.offset();
+    let durable_beyond = match receiver.read_since(offset, 1000).await {
+        Ok(rows) => rows.len(),
+        Err(e) => panic!(
+            "{tag}: event never arrived after 15 s (receiver at offset {offset}; \
+             the diagnosis read also failed: {e})"
+        ),
+    };
+    if durable_beyond > 0 {
+        panic!(
+            "{tag}: event never arrived after 15 s (receiver at offset {offset}; \
+             {durable_beyond} durable rows exist beyond it undelivered — \
+             wake/re-drain class)"
+        );
+    }
+    panic!(
+        "{tag}: event never arrived after 15 s (receiver at offset {offset}; \
+         no durable rows beyond it — publish-visibility class)"
+    );
 }
 
 /// Kill this store's listener backend (the POC's

alkstore-postgres/src/stream.rs

@@ -635,11 +635,18 @@ enum WakeTick {
     Closed,
 }
 
-async fn wait_wake(rx: &mut tokio::sync::broadcast::Receiver<RawNotification>) -> WakeTick {
+async fn wait_wake(
+    stream: &str,
+    rx: &mut tokio::sync::broadcast::Receiver<RawNotification>,
+) -> WakeTick {
     match rx.recv().await {
         Ok(_raw) => WakeTick::Ok,
-        Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
+        Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
             // Still surfaced — treat like a wake: re-drain.
+            eprintln!(
+                "alkstore-postgres: stream bridge on {stream:?} lagged the wake feed, \
+                 {n} notifications skipped (recovery is the re-drain)"
+            );
             WakeTick::Ok
         }
         Err(tokio::sync::broadcast::error::RecvError::Closed) => WakeTick::Closed,
@@ -727,7 +734,7 @@ async fn run_subscribe_loop(
         // 3. Park on the wake feed (this stream's channel — and the
         //    reserved reconnect-wake: the same re-drain heals a
         //    gap's missed wakes by reading the durable rows).
-        match wait_wake(&mut wake_rx).await {
+        match wait_wake(&stream, &mut wake_rx).await {
             WakeTick::Closed => return,
             WakeTick::Ok => {}
         }