pg: forwarder TOCTOU closed + hang-shaped arms bounded (review 004 items 4+6)

- (a) the register/unregister decide-and-queue critical sections: the
  refcount change and the derived LISTEN/UNLISTEN enqueue run under one
  registry guard (ChannelSet::enqueue_register/enqueue_unregister), so
  the command queue's order is the registry transitions' linearization —
  the old thread-preempted decide/send split (no await inside, but
  mid-unit preemption across workers) could leave a live refcount with
  the server-side LISTEN cleared until the next reconnect re-issue.
  The redundant per-registration LISTEN stays as the ack's vehicle, so
  a listen() that returned still awaits a server-processed LISTEN.
- (b) a failed generation-verify no longer parks on the poll task
  forever: the bounded settle (LISTEN_SETTLE_TIMEOUT=2s, shutdown-
  selected) reaps a genuinely dead session promptly or aborts the
  alive-but-unusable connection's reaper at the bound — reconnect
  either way.
- (c) close()/Drop can no longer leak the listener session: every
  listener-client query (verify batch, replay commands, serve
  commands) rides bounded_wait (LISTENER_QUERY_TIMEOUT=5s) with the
  shutdown watch selected alongside — the old unbounded run_command
  await sat outside the shutdown select; an overrun is the honest
  death posture (abort + reconnect, the snapshot re-issue recovers).
- pins: the interleaved register/unregister server-model equality pin
  (with a decision-anchored deterministic finale), the bounded-wait +
  failed-verify-settle paused-time pins, and the black-hole-poxy
  harness pin (close reclaims the session past a stuck serve arm);
  the refcount unit pin reworked onto the decide-and-queue API.
- replay-proofs (both ways, verified live): the old split shape fails
  the interleave pin on the review's exact mismatch; the old
  unbounded serve await fails the black-hole pin (session leaked past
  close).
- the preserved postures re-verified green: reconnect survival, the
  fanout release guard, receivers-terminal-iff-loop-gone, the
  stale-UNLISTEN generation machinery, the register-failure
  unregister discipline, the poison-recovery idiom, and both
  POC-pinned deadlock-pitfall exclusions.
- gates: pg suite with the harness server green (131 lib + 25
  contract + 9 schema; lib battery run twice full), workspace
  build/test/fuzz gates server-less green, clippy -D warnings, fmt,
  and the rustdoc gate all green; task -> completed

alkstore-postgres/src/forwarder.rs

@@ -55,6 +55,30 @@
 //! reconnect or a fresh registration). Queued commands carry the
 //! generation they were queued under; the loop bumps the shared
 //! counter at each reconnect.
+//!
+//! # The listener-path bounds (review 004 item 6)
+//!
+//! Every client query on the listener connection is **bounded**
+//! ([`LISTENER_QUERY_TIMEOUT`]): the generation-verify batch (probe or
+//! snapshot LISTEN), the post-reconnect replay commands, and the serve
+//! loop's LISTEN/UNLISTEN commands all ride it — and the shutdown
+//! watch is selected with each of them, so a close surfaces even while
+//! a query sits unresponded (the review 004 item 6 arm 2 shape: the
+//! unbounded `run_command` await used to sit outside the shutdown
+//! select and could leak the listener session past `close()`/`Drop` on
+//! a black-holed connection). A bound expiry is an **honest
+//! connection-death posture**: the generation falls through to
+//! [`reconnect_with_retry`], the poll task (the connection's reaper)
+//! is reaped or aborted, and the registry snapshot re-issue recovers —
+//! no in-loop state rides the aborted query.
+//!
+//! The failed-verify settle (review 004 item 6 arm 1) is bounded too:
+//! a failed generation-verify batch no longer parks forever on the
+//! poll task's completion (the session may be alive-but-unusable — a
+//! query error response leaves a working transport): the loop's
+//! settle path gives the poll task (the connection's reaper) a
+//! bounded window ([`LISTEN_SETTLE_TIMEOUT`] via [`bounded_wait`])
+//! and falls through to the reconnect either way.
 
 use std::future::Future;
 use std::sync::Arc;
@@ -102,6 +126,28 @@ pub(crate) const RECONNECT_MAX_MS: u64 = 2000;
 /// not silent).
 const FANOUT_CAPACITY: usize = 1024;
 
+/// The listener-path client-query bound (review 004 item 6): the
+/// generation-verify batch, the replayed commands, and the serve
+/// loop's LISTEN/UNLISTEN all ride it, together with the shutdown
+/// watch's selection — a black-holed or wedged connection turns into
+/// the honest death posture (fall through to the reconnect) instead
+/// of parking the loop forever (which used to hold the shutdown
+/// select hostage and leak the listener session past `close()`/`Drop`).
+/// Generous against a merely-slow server (an overrun re-issues from
+/// the snapshot on reconnect, so a spuriously-bounded healthy server
+/// recovers; nothing is lost either way).
+pub(crate) const LISTENER_QUERY_TIMEOUT: Duration = Duration::from_secs(5);
+
+/// The failed-verify settle bound (review 004 item 6 arm 1): a
+/// generation whose verify batch failed gives the poll task — the
+/// connection's reaper — this bounded window to observe the death,
+/// then the falls to the reconnect either way (an aborted settle
+/// closes the alive-but-unusable connection alongside). Far above the
+/// real death-detection latency (a socket error/EOF surfaces in
+/// `poll_message` at once); a longer settle would only mean the
+/// session was usable and the abort is the recovery's cost.
+pub(crate) const LISTEN_SETTLE_TIMEOUT: Duration = Duration::from_secs(2);
+
 /// Quote a channel name as a SQL identifier (the POC's `quote_ident`
 /// shape — Postgres doubles embedded `"`). Channel names are arbitrary
 /// legal names (colons etc.); quoting is what makes them work.
@@ -120,7 +166,9 @@ pub(crate) struct RawNotification {
 /// The dynamic channel set: the registry the forwarder re-issues
 /// LISTENs from after every reconnect. `listen` (registration) and
 /// receiver-drop (unregistration, the refcounted last-subscriber
-/// rule) ride [`Forwarder`]'s register / unregister.
+/// rule) ride [`Forwarder`]'s register / unregister — whose
+/// decide-and-queue critical sections (review 004 item 4) run under
+/// this set's guard.
 #[derive(Debug, Default)]
 pub(crate) struct ChannelSet {
     /// BTreeMap for the deterministic snapshot; the value is the
@@ -129,40 +177,6 @@ pub(crate) struct ChannelSet {
 }
 
 impl ChannelSet {
-    /// Add a subscriber to `channel` (inserting the entry on the
-    /// first); returns `true` if this is a brand-new channel entry
-    /// (a LISTEN must be issued). Test-side: the refcount unit pin.
-    pub(crate) fn add_subscriber(&self, channel: &str) -> bool {
-        let mut map = self
-            .channels
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner);
-        let is_new = !map.contains_key(channel);
-        *map.entry(channel.to_string()).or_insert(0) += 1;
-        is_new
-    }
-
-    /// Remove a subscriber from `channel`; returns `true` if the last
-    /// subscriber left (an UNLISTEN must be issued and the registry
-    /// entry dropped). A channel not present is a no-op (`false`).
-    /// Test-side: the refcount unit pin.
-    pub(crate) fn remove_subscriber(&self, channel: &str) -> bool {
-        let mut map = self
-            .channels
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner);
-        let Some(count) = map.get_mut(channel) else {
-            return false;
-        };
-        if *count > 1 {
-            *count -= 1;
-            false
-        } else {
-            map.remove(channel);
-            true
-        }
-    }
-
     /// The registry's live channel list — the reconnect re-issue
     /// source, and the tests' registry read.
     pub(crate) fn snapshot(&self) -> Vec<String> {
@@ -173,6 +187,103 @@ impl ChannelSet {
             .cloned()
             .collect()
     }
+
+    /// The register path's decide-and-queue (review 004 item 4's
+    /// TOCTOU fix): the refcount change and the derived LISTEN enqueue
+    /// are **one critical section** under the registry guard — an
+    /// interleaved register/unregister across tasks can never leave a
+    /// live refcount with the server-side LISTEN cleared, because the
+    /// command queue's order is the registry transitions'
+    /// linearization (a last-subscriber drop whose UNLISTEN reaches
+    /// the server before a later re-register's LISTEN is exactly the
+    /// guarded order; the reverse queue order — UNLISTEN replayed
+    /// after a live registration's LISTEN — is structurally excluded).
+    /// The `connected` read and the generation tag load ride the same
+    /// section, so the queued command carries the refcount state (the
+    /// decision) it was derived from.
+    ///
+    /// The registry write happens first (the ordering is the
+    /// registration-recovery discipline: during a gap the entry stays
+    /// and the reconnect re-issue covers it — the transient
+    /// `Database`-errored register's honest posture). Each registration
+    /// enqueues its own LISTEN even for a shared (non-new) refcount —
+    /// the redundant LISTEN is the ack's vehicle: a `listen()` that
+    /// returned sees every later commit (`listen` starts from now), so
+    /// the registration must await a *server-processed* LISTEN, not
+    /// infer it from the shared refcount (whose LISTEN may still sit
+    /// queued). Returns the ack receiver — await it *outside* the
+    /// registry guard (never hold a std lock across an await).
+    pub(crate) fn enqueue_register(
+        &self,
+        channel: &str,
+        commands: &mpsc::UnboundedSender<ListenCommand>,
+        generation: &AtomicU64,
+        connected: &AtomicBool,
+    ) -> Result<tokio::sync::oneshot::Receiver<alkstore::Result<()>>, alkstore::Error> {
+        let mut map = self
+            .channels
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        *map.entry(channel.to_string()).or_insert(0) += 1;
+        if !connected.load(Ordering::Acquire) {
+            return Err(database_error(
+                "the listener connection is not currently up (mid-reconnect); \
+                 the channel is registered and re-issued on reconnect",
+            ));
+        }
+        let (ack_tx, ack_rx) = tokio::sync::oneshot::channel();
+        let sent = commands
+            .send(ListenCommand::Listen {
+                generation: generation.load(Ordering::Acquire),
+                channel: channel.to_string(),
+                ack: ack_tx,
+            })
+            .is_ok();
+        if !sent {
+            return Err(database_error("the forwarder loop is gone (shutting down)"));
+        }
+        Ok(ack_rx)
+    }
+
+    /// The unregister path's decide-and-queue (review 004 item 4's
+    /// TOCTOU fix, symmetric with [`ChannelSet::enqueue_register`]):
+    /// the refcount decrement and the derived UNLISTEN enqueue are one
+    /// critical section under the registry guard — the UNLISTEN queues
+    /// *with* the last-subscriber decision it was derived from, and
+    /// (same generation or not) no separate registration's LISTEN can
+    /// be forced ahead of it into the queue. The UNLISTEN itself stays
+    /// best-effort (a lost one re-issues nothing; UNLISTEN is not
+    /// recovery-relevant).
+    pub(crate) fn enqueue_unregister(
+        &self,
+        channel: &str,
+        commands: &mpsc::UnboundedSender<ListenCommand>,
+        generation: &AtomicU64,
+    ) {
+        {
+            let mut map = self
+                .channels
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner);
+            let is_last = match map.get_mut(channel) {
+                Some(count) if *count > 1 => {
+                    *count -= 1;
+                    false
+                }
+                Some(_) => {
+                    map.remove(channel);
+                    true
+                }
+                None => false,
+            };
+            if is_last {
+                let _ = commands.send(ListenCommand::Unlisten {
+                    generation: generation.load(Ordering::Acquire),
+                    channel: channel.to_string(),
+                });
+            }
+        }
+    }
 }
 
 /// The forwarder handle the store holds: the subscription point for
@@ -313,6 +424,43 @@ pub(crate) fn reconcile_queued_commands(
     (replay, dropped)
 }
 
+/// The bounded-wait outcome — the future's completion, its bound's
+/// expiry, or the shutdown flip, whichever came first ([`bounded_wait`]).
+#[derive(Debug)]
+pub(crate) enum BoundedOutcome<T> {
+    Done(T),
+    Overran,
+    Shutdown,
+}
+
+/// The bounded-wait primitive (review 004 item 6's treatment family)
+/// over the listener path's awaits: the future's output, a bound
+/// expiry, or the shutdown flip. The failed-verify settle and every
+/// client query on the listener connection (the verify batch, the
+/// replayed commands, the serve loop's commands) ride it — a
+/// black-holed connection can no longer hold the loop's shutdown
+/// select hostage (the close-time listener-session leak) and a failed
+/// verify can no longer park forever on the poll task's completion.
+pub(crate) async fn bounded_wait<T, F>(
+    bound: Duration,
+    shutdown: &mut watch::Receiver<bool>,
+    fut: F,
+) -> BoundedOutcome<T>
+where
+    F: Future<Output = T>,
+{
+    tokio::select! {
+        biased;
+
+        _ = shutdown.changed() => BoundedOutcome::Shutdown,
+
+        done = tokio::time::timeout(bound, fut) => match done {
+            Ok(value) => BoundedOutcome::Done(value),
+            Err(_elapsed) => BoundedOutcome::Overran,
+        },
+    }
+}
+
 /// The mid-LISTEN transient error — a `register` whose LISTEN did not
 /// land (died mid-command, or dropped stale at a reconnect) acks/
 /// returns this; the retry recovers it (the registry entry stays
@@ -424,12 +572,22 @@ impl Forwarder {
     /// that returned sees every later commit; `listen` starts from
     /// now).
     ///
-    /// The registry write happens first: if the connection dies
+    /// The registry write and the derived LISTEN enqueue are **one
-    /// between the write and the LISTEN delivery, the reconnect path
+    /// critical section** under the registry guard
-    /// re-issues from the snapshot — a registration is never lost
+    /// ([`ChannelSet::enqueue_register`] — review 004 item 4's
-    /// (the ordering is the registration-recovery discipline).
+    /// TOCTOU fix): interleaved register/unregister across tasks
+    /// linearize through the guard, so the command queue's order
+    /// mirrors the registry's transitions and a live refcount can
+    /// never sit against a cleared server-side LISTEN. The write is
+    /// first: if the connection dies between it and the LISTEN
+    /// delivery, the reconnect path re-issues from the snapshot — a
+    /// registration is never lost (the ordering is the
+    /// registration-recovery discipline).
     ///
-    /// Refcounted: N subscribers on one channel share one LISTEN.
+    /// Refcounted: N subscribers on one channel share one LISTEN (the
+    /// redundant per-registration LISTEN is the ack's vehicle — the
+    /// registration awaits a *server-processed* LISTEN, not an
+    /// inference from the shared refcount).
     ///
     /// Fails `Database` when no listener connection is currently live
     /// (mid-reconnect) or the LISTEN itself failed — transient; a
@@ -442,25 +600,12 @@ impl Forwarder {
     /// last-subscriber UNLISTEN rule and keep the channel LISTENing
     /// after every receiver is gone.
     pub(crate) async fn register(&self, channel: &str) -> Result<(), alkstore::Error> {
-        self.channels.add_subscriber(channel);
+        let ack_rx = self.channels.enqueue_register(
-        if !self.connected.load(Ordering::Acquire) {
+            channel,
-            return Err(database_error(
+            &self.commands_tx,
-                "the listener connection is not currently up (mid-reconnect); \
+            &self.generation,
-                 the channel is registered and re-issued on reconnect",
+            &self.connected,
-            ));
+        )?;
-        }
-        let (ack_tx, ack_rx) = tokio::sync::oneshot::channel();
-        let sent = self
-            .commands_tx
-            .send(ListenCommand::Listen {
-                generation: self.generation.load(Ordering::Acquire),
-                channel: channel.to_string(),
-                ack: ack_tx,
-            })
-            .is_ok();
-        if !sent {
-            return Err(database_error("the forwarder loop is gone (shutting down)"));
-        }
         ack_rx
             .await
             .map_err(|_| database_error("the forwarder loop is gone (shutting down)"))?
@@ -469,22 +614,21 @@ impl Forwarder {
     /// Unregister one subscriber's reference on `channel` AND issue
     /// `UNLISTEN` when the **last** subscriber for the channel drops
     /// (the refcount; zero subscribers ⇒ no LISTEN needed). Registry
-    /// write first (ordering symmetric with [`Forwarder::register`]'s
+    /// write and the derived UNLISTEN enqueue are **one critical
-    /// recovery logic); the UNLISTEN is best-effort — a lost one
+    /// section** under the registry guard
-    /// re-issues nothing (UNLISTEN is not recovery-relevant: the
+    /// ([`ChannelSet::enqueue_unregister`] — review 004 item 4's
-    /// snapshot won't carry the removed channel). The queued UNLISTEN
+    /// TOCTOU fix, symmetric with [`Forwarder::register`]). The
-    /// rides its send-time generation: if the connection died before
+    /// UNLISTEN is best-effort — a lost one re-issues nothing
-    /// processing it, the reconnect drops it rather than replaying —
+    /// (UNLISTEN is not recovery-relevant: the snapshot won't carry
-    /// a stale UNLISTEN after the re-issued LISTENs would otherwise
+    /// the removed channel). The queued UNLISTEN rides its send-time
-    /// silently cancel a re-registered channel (review 002 Finding 4;
+    /// generation: if the connection died before processing it, the
-    /// the fresh connection's LISTEN set is the snapshot's).
+    /// reconnect drops it rather than replaying — a stale UNLISTEN
+    /// after the re-issued LISTENs would otherwise silently cancel a
+    /// re-registered channel (review 002 Finding 4; the fresh
+    /// connection's LISTEN set is the snapshot's).
     pub(crate) fn unregister(&self, channel: &str) {
-        if self.channels.remove_subscriber(channel) {
+        self.channels
-            let _ = self.commands_tx.send(ListenCommand::Unlisten {
+            .enqueue_unregister(channel, &self.commands_tx, &self.generation);
-                generation: self.generation.load(Ordering::Acquire),
-                channel: channel.to_string(),
-            });
-        }
     }
 
     /// Shutdown: flip the switch — the loop exits, dropping the
@@ -603,14 +747,28 @@ impl Drop for FanoutRelease {
 ///
 /// Per connection generation: spawn the poll task (it owns the
 /// `Connection` and fans out `poll_message`'s notifications) → issue
-/// LISTENs from the live channel-set snapshot → serve commands and
+/// LISTENs from the live channel-set snapshot *bounded and
-/// wait for the connection's death → reconnect via
+/// shutdown-selected* ([`bounded_wait`]/[`LISTENER_QUERY_TIMEOUT`]) →
-/// [`reconnect_with_retry`] (exponential backoff 50 ms → 2 s cap;
+/// on a failed verify, the bounded settle
-/// failed connects never leave the arm). The loop-carried connection
+/// ([`LISTEN_SETTLE_TIMEOUT`]) reaps-or-aborts the poll task → serve
-/// is *always present* at the generation's top by construction: the
+/// commands (each also bounded, the shutdown watch selected with the
-/// first iteration holds the handed-in connection, and only a
+/// await) → reconnect via [`reconnect_with_retry`] (exponential
-/// successful connect re-binds it — there is no empty-slot arm left to
+/// backoff 50 ms → 2 s cap; failed connects never leave the arm). The
-/// fall back into.
+/// loop-carried connection is *always present* at the generation's
+/// top by construction: the first iteration holds the handed-in
+/// connection, and only a successful connect re-binds it — there is
+/// no empty-slot arm left to fall back into.
+///
+/// The bounded arms (review 004 item 6) are honest-death postures,
+/// never stalls: a failed verify — even with the session
+/// alive-but-unusable — falls to the reconnect after a settled bound
+/// instead of parking on the poll task forever, and an unresponded
+/// serve/replay command aborts the generation and reconnects instead
+/// of holding the shutdown select hostage (a close can no longer leak
+/// the listener session past `close()`/`Drop` on a black-holed
+/// connection). An overrun query's costs are absorbed by the
+/// recovery: the registry snapshot re-issues on the fresh generation,
+/// no replay, no lost registration.
 ///
 /// The reconnect bumps the shared generation counter (before the
 /// snapshot re-issue, so commands queued past the bump — a healthy
@@ -701,21 +859,58 @@ async fn forwarder_loop(
         // reconnect it carries everything `listen` ever registered).
         // With no channels yet, a `SELECT 1` liveness probe stands in
         // — `connected` must reflect a verified-alive session, not an
-        // assumed one.
+        // assumed one. The verify is bounded and shutdown-selected
+        // (review 004 item 6): a black-holed or wedged connection
+        // cannot park the generation top forever, and a close
+        // surfaces here too.
         let listen_sql = channels
             .snapshot()
             .iter()
             .map(|ch| format!("LISTEN {};\n", quote_ident(ch)))
             .collect::<String>();
-        let listen_ok = if listen_sql.is_empty() {
+        let verify = async {
-            client.simple_query("SELECT 1").await.is_ok()
+            if listen_sql.is_empty() {
-        } else {
+                client.simple_query("SELECT 1").await.is_ok()
-            client.batch_execute(&listen_sql).await.is_ok()
+            } else {
+                client.batch_execute(&listen_sql).await.is_ok()
+            }
+        };
+        let listen_ok = match bounded_wait(LISTENER_QUERY_TIMEOUT, &mut shutdown, verify).await {
+            BoundedOutcome::Done(ok) => ok,
+            BoundedOutcome::Overran => false,
+            BoundedOutcome::Shutdown => {
+                poll_task.abort();
+                connected.store(false, Ordering::Release);
+                return;
+            }
         };
 
-        if listen_ok {
+        if !listen_ok {
+            connected.store(false, Ordering::Release);
+            // The bounded settle (review 004 item 6 arm 1): a failed
+            // verify used to park forever on the poll task's
+            // completion — but the session may be alive-but-unusable
+            // (a failed query response leaves a working transport, and
+            // only the poll task would end the generation never). The
+            // reaper gets this bounded window; either way the
+            // generation falls through to the reconnect below (an
+            // unsettled window aborts the task, closing the
+            // alive-but-unusable connection; its reaper's job is
+            // absorbed by the socket close).
+            match bounded_wait(LISTEN_SETTLE_TIMEOUT, &mut shutdown, &mut poll_task).await {
+                BoundedOutcome::Done(_) => {}
+                BoundedOutcome::Overran => poll_task.abort(),
+                BoundedOutcome::Shutdown => {
+                    poll_task.abort();
+                    return;
+                }
+            }
+        } else {
             backoff_ms = RECONNECT_BASE_MS;
             connected.store(true, Ordering::Release);
+            let was_first = first_listen;
+            first_listen = false;
+
             // The stale command drain (review 002 Finding 4): the
             // re-issue above is the server-side LISTEN set's authority
             // — the predecessor generation's queued commands are
@@ -732,91 +927,145 @@ async fn forwarder_loop(
             for cmd in dropped {
                 cmd.abandon_stale();
             }
+            let mut replay_dead = false;
             for cmd in replay {
-                run_command(&client, cmd).await;
+                // The replay rides the listener-path bound, with the
-            }
+                // shutdown watch selected alongside — a stuck replay
-            if !first_listen {
+                // command cannot park the generation top forever, and
-                // The synthetic reconnect-wake (ADR-008 §4's reserved
+                // a close surfaces mid-replay too.
-                // channel; ADR-006's recovery semantics) broadcast to
+                match bounded_wait(
-                // every subscriber's receiver — the no-replay hole's
+                    LISTENER_QUERY_TIMEOUT,
-                // honest communication. The wake-content mapping is
+                    &mut shutdown,
-                // the notify-listen task's; the skeleton broadcasts
+                    run_command(&client, cmd),
-                // the reserved channel's name so the fanout carries it.
+                )
-                let _ = fanout.send(RawNotification {
+                .await
-                    channel: alkstore::RESERVED_LISTENER_RECONNECTED.to_string(),
+                {
-                });
+                    BoundedOutcome::Done(()) => {}
-            }
+                    BoundedOutcome::Shutdown => {
-            first_listen = false;
-        } else {
-            // The connection died before/during the LISTEN — fall
-            // through to the reconnect path with the poll task's
-            // completion below.
-            connected.store(false, Ordering::Release);
-        }
-
-        // Drive the connection until it dies (or shutdown). Client
-        // queries (LISTEN/UNLISTEN commands) run while the poll task
-        // polls the same session — the co-residency that excludes
-        // pitfall 1's starvation permanently.
-        if listen_ok {
-            loop {
-                tokio::select! {
-                    biased;
-
-                    // Shutdown: exit the whole forwarder (dropping the
-                    // client closes the listener session).
-                    _ = shutdown.changed() => {
                         poll_task.abort();
                         return;
                     }
-
+                    BoundedOutcome::Overran => {
-                    cmd = commands.recv() => {
+                        // An unresponded replay command: the
-                        match cmd {
+                        // connection is unusable — the honest death
-                            Some(command) if command.generation() < generation => {
+                        // posture, straight to the reconnect (the
-                                // A stale command queued under a dead
+                        // reaper is aborted now; the next generation's
-                                // generation (a send racing the
+                        // re-issue is the recovery).
-                                // reconnect's drain): dropped here for
+                        connected.store(false, Ordering::Release);
-                                // the same reason the generation-top
+                        poll_task.abort();
-                                // drain drops its batch — it must not
+                        replay_dead = true;
-                                // touch this connection (review 002
-                                // Finding 4).
-                                command.abandon_stale();
-                            }
-                            Some(command) => {
-                                // Acked after the server processed the
-                                // LISTEN: a `listen()` return stamps
-                                // "registered from now" end-to-end. A
-                                // failed LISTEN (connection died
-                                // mid-command) acks the failure — the
-                                // caller sees it; the registry entry
-                                // rides the reconnect re-issue.
-                                run_command(&client, command).await;
-                            }
-                            // The command sender is gone (the store —
-                            // and its forwarder handle — went away
-                            // without a shutdown flip): serve no more,
-                            // exit (the release guard fires; the
-                            // receivers' terminal arm).
-                            None => {
-                                poll_task.abort();
-                                return;
-                            }
-                        }
-                    }
-
-                    // The poll task ended = the connection died (a
-                    // fatal poll error or server close). The session
-                    // is already gone server-side; reconnect below.
-                    _ = &mut poll_task => {
                         break;
                     }
                 }
             }
-        } else {
+            if !replay_dead {
-            // A failed probe/LISTEN means the connection died: wait
+                if !was_first {
-            // the poll task out (it owns the connection and is its
+                    // The synthetic reconnect-wake (ADR-008 §4's reserved
-            // reaper).
+                    // channel; ADR-006's recovery semantics) broadcast to
-            let _ = poll_task.await;
+                    // every subscriber's receiver — the no-replay hole's
+                    // honest communication. The wake-content mapping is
+                    // the notify-listen task's; the skeleton broadcasts
+                    // the reserved channel's name so the fanout carries it.
+                    // (A generation that dies mid-replay never broadcasts —
+                    // the next generation's top carries the wake; the
+                    // `first_listen` flip above already happened at this
+                    // generation's verified-alive entry.)
+                    let _ = fanout.send(RawNotification {
+                        channel: alkstore::RESERVED_LISTENER_RECONNECTED.to_string(),
+                    });
+                }
+
+                // Drive the connection until it dies (or shutdown). Client
+                // queries (LISTEN/UNLISTEN commands) run while the poll task
+                // polls the same session — the co-residency that excludes
+                // pitfall 1's starvation permanently.
+                loop {
+                    tokio::select! {
+                        biased;
+
+                        // Shutdown: exit the whole forwarder (dropping the
+                        // client closes the listener session).
+                        _ = shutdown.changed() => {
+                            poll_task.abort();
+                            return;
+                        }
+
+                        cmd = commands.recv() => {
+                            match cmd {
+                                Some(command) if command.generation() < generation => {
+                                    // A stale command queued under a dead
+                                    // generation (a send racing the
+                                    // reconnect's drain): dropped here for
+                                    // the same reason the generation-top
+                                    // drain drops its batch — it must not
+                                    // touch this connection (review 002
+                                    // Finding 4).
+                                    command.abandon_stale();
+                                }
+                                Some(command) => {
+                                    // Acked after the server processed the
+                                    // LISTEN: a `listen()` return stamps
+                                    // "registered from now" end-to-end. A
+                                    // failed LISTEN (connection died
+                                    // mid-command) acks the failure — the
+                                    // caller sees it; the registry entry
+                                    // rides the reconnect re-issue.
+                                    //
+                                    // The command rides the listener-path
+                                    // bound with the shutdown watch selected
+                                    // alongside (review 004 item 6 arm 2) —
+                                    // the old unbounded `run_command` await
+                                    // sat in the recv arm *outside* the
+                                    // shutdown select: a black-holed
+                                    // connection held the loop hostage past
+                                    // any close and leaked the listener
+                                    // session. An overrun command turns the
+                                    // connection honest-dead; the shutdown
+                                    // arm resolves nothing (a stalled
+                                    // LISTEN's caller sees the ack-drop —
+                                    // the "loop is gone (shutting down)"
+                                    // posture — and the caller's
+                                    // re-issuable state is the registry's).
+                                    match bounded_wait(
+                                        LISTENER_QUERY_TIMEOUT,
+                                        &mut shutdown,
+                                        run_command(&client, command),
+                                    )
+                                    .await
+                                    {
+                                        BoundedOutcome::Done(()) => {}
+                                        BoundedOutcome::Shutdown => {
+                                            poll_task.abort();
+                                            return;
+                                        }
+                                        BoundedOutcome::Overran => {
+                                            connected.store(false, Ordering::Release);
+                                            poll_task.abort();
+                                            break;
+                                        }
+                                    }
+                                }
+                                // The command sender is gone (the store —
+                                // and its forwarder handle — went away
+                                // without a shutdown flip): serve no more,
+                                // exit (the release guard fires; the
+                                // receivers' terminal arm).
+                                None => {
+                                    poll_task.abort();
+                                    return;
+                                }
+                            }
+                        }
+
+                        // The poll task ended = the connection died (a
+                        // fatal poll error or server close). The session
+                        // is already gone server-side; reconnect below.
+                        _ = &mut poll_task => {
+                            break;
+                        }
+                    }
+                }
+            }
         }
 
         // Reconnect: the Client is dead with the session (dropping it
@@ -843,8 +1092,11 @@ async fn forwarder_loop(
             reconnect_with_retry(attempt, &mut backoff_ms, &mut shutdown).await
         else {
             // Shutdown mid-backoff/mid-retry: the poll task is already
-            // done (the generation's end preceded this arm); plain
+            // reaped or aborted (every run_command path above — the
-            // return — the guard and the client drop carry the rest.
+            // serve loop's poll-death break, the bounded-await death
+            // postures, the settle — ends the reaper before reaching
+            // here); plain return — the guard and the client drop
+            // carry the rest.
             return;
         };
         // The generation bump happens here — *before* the fresh

alkstore-postgres/src/store/notify_tests.rs

@@ -726,22 +726,114 @@ async fn pitfall_2_long_lived_listener_survives_client_churn() {
     drop_schema(&admin, &schema).await;
 }
 
+/// The "server" side of the ack protocol for the decide-and-queue
+/// unit pins: the next queued LISTEN, resolved after the model
+/// processed it.
+async fn drain_one_listen(
+    rx: &mut tokio::sync::mpsc::UnboundedReceiver<crate::forwarder::ListenCommand>,
+) -> String {
+    use crate::forwarder::ListenCommand;
+
+    match rx.recv().await.expect("the command stream is open") {
+        ListenCommand::Listen { channel, ack, .. } => {
+            let _ = ack.send(Ok(()));
+            channel
+        }
+        ListenCommand::Unlisten { channel, .. } => {
+            panic!(
+                "an UNLISTEN must be derived exactly by the last-subscriber decision, got one for {channel:?}"
+            )
+        }
+    }
+}
+
 /// Unit pin (server-less): the forwarder's channel-set refcount —
-/// add/remove subscribers and the last-subscriber rule, and the
+/// through the decide-and-queue registry API (review 004 item 4): the
-/// snapshot the reconnect re-issues from.
+/// last-subscriber rule (an UNLISTEN is derived exactly when the last
-#[test]
+/// reference drops, and only then), drop-past-zero as a no-op, and the
-fn channel_set_refcounts_last_subscriber_rule() {
+/// snapshot the reconnect re-issues from. Each registration's LISTEN
-    let set = crate::forwarder::ChannelSet::default();
+/// is queued (the redundant per-registration LISTEN is the ack's
-    assert!(set.add_subscriber("a"), "first subscriber inserts");
+/// vehicle) — asserted too.
-    assert!(
+#[tokio::test]
-        !set.add_subscriber("a"),
+async fn channel_set_refcounts_last_subscriber_rule() {
-        "second subscriber doesn't reinsert"
+    use crate::forwarder::{ChannelSet, GenerationSlot, ListenCommand};
-    );
+    use std::sync::Arc;
-    assert!(set.add_subscriber("b"));
+    use std::sync::atomic::AtomicBool;
+
+    let set = ChannelSet::default();
+    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ListenCommand>();
+    let generation: GenerationSlot = Arc::new(AtomicU64::new(0));
+    let connected = Arc::new(AtomicBool::new(true));
+
+    // First subscriber on "a": the entry inserts and its LISTEN is
+    // derived (queued; the "server" resolves the ack, then the
+    // registration's await sees it).
+    let ack = set
+        .enqueue_register("a", &tx, &generation, &connected)
+        .expect("the decide-and-queue register enqueued");
+    let (result, first_channel) = tokio::join!(ack, drain_one_listen(&mut rx));
+    result.expect("the ack resolves").expect("the LISTEN acked");
+    assert_eq!(first_channel, "a");
+    assert!(!rx.try_recv().is_ok(), "no extra commands queued");
+
+    // Second subscriber on "a": the same entry (no reinsertion), the
+    // redundant per-registration LISTEN rides for its own ack.
+    let ack = set
+        .enqueue_register("a", &tx, &generation, &connected)
+        .expect("the decide-and-queue register enqueued");
+    let (result, second_channel) = tokio::join!(ack, drain_one_listen(&mut rx));
+    result.expect("the ack resolves").expect("the LISTEN acked");
+    assert_eq!(second_channel, "a");
+
+    // "b": a second entry inserts; the snapshot is deterministic.
+    let ack = set
+        .enqueue_register("b", &tx, &generation, &connected)
+        .expect("the decide-and-queue register enqueued");
+    let (result, third_channel) = tokio::join!(ack, drain_one_listen(&mut rx));
+    result.expect("the ack resolves").expect("the LISTEN acked");
+    assert_eq!(third_channel, "b");
     assert_eq!(set.snapshot(), vec!["a".to_string(), "b".to_string()]);
 
-    assert!(!set.remove_subscriber("a"), "first drop keeps the entry");
+    // The first drop of "a" keeps the entry (an UNLISTEN must NOT be
-    assert!(set.remove_subscriber("a"), "last drop removes it");
+    // derived — the refcount still holds).
-    assert!(!set.remove_subscriber("a"), "drop past zero is a no-op");
+    set.enqueue_unregister("a", &tx, &generation);
+    assert_eq!(set.snapshot(), vec!["a".to_string(), "b".to_string()]);
+    assert!(!rx.try_recv().is_ok(), "no UNLISTEN before the last drop");
+
+    // The last drop of "a" removes it and derives the UNLISTEN.
+    set.enqueue_unregister("a", &tx, &generation);
     assert_eq!(set.snapshot(), vec!["b".to_string()]);
+    match rx.recv().await.expect("the command stream is open") {
+        ListenCommand::Unlisten { channel, .. } => assert_eq!(channel, "a"),
+        ListenCommand::Listen { channel, .. } => {
+            panic!("a LISTEN after the last drop breaks the last-subscriber rule: {channel:?}")
+        }
+    }
+
+    // Drop past zero is a no-op.
+    set.enqueue_unregister("a", &tx, &generation);
+    assert_eq!(set.snapshot(), vec!["b".to_string()]);
+    assert!(!rx.try_recv().is_ok(), "drop past zero queues nothing");
+
+    // The mid-reconnect arm keeps the registry-write-first order: the
+    // entry stays written (the reconnect re-issue's recovery) and NOT
+    // a thing is enqueued against the down connection.
+    connected.store(false, std::sync::atomic::Ordering::SeqCst);
+    let err = match set.enqueue_register("c", &tx, &generation, &connected) {
+        Err(e) => e,
+        Ok(_) => panic!("a register during a gap must fail the honest transient"),
+    };
+    assert!(
+        matches!(err, alkstore::Error::Database(_)),
+        "the mid-reconnect register fails typed, got: {err:?}"
+    );
+    assert_eq!(
+        set.snapshot(),
+        vec!["b".to_string(), "c".to_string()],
+        "the failed register's entry stays (registry-write-first)"
+    );
+    assert!(
+        !rx.try_recv().is_ok(),
+        "the failed register enqueued nothing"
+    );
 }

alkstore-postgres/src/store/open_tests.rs

@@ -10,7 +10,8 @@
 //! schema per test (the POC's shared-server parallel-interference
 //! caveat answered by schema-per-test isolation).
 
-use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::time::Duration;
 
 use crate::forwarder::RawNotification;
@@ -41,6 +42,17 @@ fn harness_dsn() -> Option<String> {
     ))
 }
 
+/// The harness DSN's parts — the black-hole proxy test's building
+/// blocks (`None` = no harness server).
+fn harness_parts() -> Option<(String, u16, String, String, String)> {
+    let host = std::env::var(ENV_HOST).ok()?;
+    let port: u16 = std::env::var(ENV_PORT).ok()?.parse().ok()?;
+    let user = std::env::var(ENV_USER).ok()?;
+    let password = std::env::var(ENV_PASSWORD).ok()?;
+    let db = std::env::var(ENV_DB).unwrap_or_else(|_| "postgres".to_string());
+    Some((host, port, user, password, db))
+}
+
 fn unreachable_dsn() -> String {
     "host=127.0.0.1 port=1 user=x password=x dbname=x connect_timeout=2".to_string()
 }
@@ -1355,3 +1367,519 @@ async fn forwarder_survives_failed_reconnects_and_recovers_fully() {
     store.close();
     drop_schema(&admin, &schema).await;
 }
+
+/// Unit pin (server-less, review 004 item 4): interleaved
+/// register/unregister across tasks keep the server-side LISTEN state
+/// matching the registry. The server-model task applies the command
+/// stream to a LISTEN set **in queue order** (exactly what the live
+/// connection's statement stream does), and the refcount's transitions
+/// must read back consistent — the review's failure (a live refcount
+/// whose server-side LISTEN ended cleared, surviving until the next
+/// reconnect re-issue, both commands sharing a generation so no stale
+/// check saves it) is exactly a registry/model divergence. The two
+/// interleave on one channel with yields between every decide-and-queue
+/// unit; the final states are asserted equal.
+#[tokio::test(flavor = "multi_thread")]
+async fn interleaved_register_unregister_keep_server_listen_set_matching_registry() {
+    use crate::forwarder::{ChannelSet, GenerationSlot, ListenCommand};
+    use std::collections::BTreeSet;
+    use std::sync::Arc;
+    use std::sync::atomic::AtomicBool;
+
+    let channels = Arc::new(ChannelSet::default());
+    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ListenCommand>();
+    let generation: GenerationSlot = Arc::new(AtomicU64::new(0));
+    let connected = Arc::new(AtomicBool::new(true));
+
+    // "The server": applies the commands in queue order, resolving
+    // each LISTEN's ack after the model processed it (the acked
+    // registration's synchronous end-to-end contract).
+    let server = tokio::spawn(async move {
+        let mut model = BTreeSet::new();
+        while let Some(cmd) = rx.recv().await {
+            match cmd {
+                ListenCommand::Listen { channel, ack, .. } => {
+                    model.insert(channel.clone());
+                    let _ = ack.send(Ok(()));
+                }
+                ListenCommand::Unlisten { channel, .. } => {
+                    model.remove(&channel);
+                }
+            }
+        }
+        model
+    });
+
+    for channel in ["interleave-a", "interleave-b"] {
+        // Seed one reference (the mismatch's live-refcount operand).
+        let ack = channels
+            .enqueue_register(channel, &tx, &generation, &connected)
+            .expect("the decide-and-queue register enqueued");
+        let _ = tokio::time::timeout(Duration::from_secs(2), ack)
+            .await
+            .expect("the seeded LISTEN acked")
+            .expect("the LISTEN ok");
+
+        // The two interleavers: one keeps dropping references, the
+        // other keeps re-adding them — every step yields between the
+        // units, so the multi-thread runtime interleaves the tasks
+        // arbitrarily (the review's "across tasks" shape; the old
+        // split shape's window is a thread-preempted mid-unit).
+        const ITERATIONS: usize = 4;
+        let adds = channels.clone();
+        let adds_tx = tx.clone();
+        let adds_gen = generation.clone();
+        let adds_connected = connected.clone();
+        let channel_owned = channel.to_string();
+        let adder = tokio::spawn(async move {
+            for _ in 0..ITERATIONS {
+                let ack = adds
+                    .enqueue_register(&channel_owned, &adds_tx, &adds_gen, &adds_connected)
+                    .expect("the decide-and-queue register enqueued");
+                let _ = ack.await;
+                tokio::task::yield_now().await;
+            }
+        });
+        let drops = channels.clone();
+        let drops_tx = tx.clone();
+        let drops_gen = generation.clone();
+        let channel_owned = channel.to_string();
+        let dropper = tokio::spawn(async move {
+            for _ in 0..ITERATIONS {
+                drops.enqueue_unregister(&channel_owned, &drops_tx, &drops_gen);
+                tokio::task::yield_now().await;
+            }
+        });
+
+        tokio::time::timeout(Duration::from_secs(60), adder)
+            .await
+            .expect("the adder finished")
+            .expect("the adder did not panic");
+        tokio::time::timeout(Duration::from_secs(60), dropper)
+            .await
+            .expect("the dropper finished")
+            .expect("the dropper did not panic");
+
+        // The deterministic finale on a fresh channel (count seeded to
+        // one, its queued LISTEN acked so the model is proven drained
+        // in FIFO order): a last-subscriber drop is spawned; the main
+        // re-register waits for the drop's *decision* to become
+        // visible (the entry vanishing from the registry snapshot) and
+        // races the drop's derived UNLISTEN inside that window. Under
+        // the fixed decide-and-queue shape the entry can only vanish
+        // *with* the UNLISTEN already queued — the re-register lands
+        // after it ([U, L]) and every legal schedule reads back
+        // consistent; under the old decide/send split the entry
+        // vanishes at the decision, the re-registration's LISTEN lands
+        // in the split's window, and the UNLISTEN replays after it —
+        // the review's exact mismatch, caught by the final equality (a
+        // live refcount at a cleared server-side LISTEN).
+        let finale = format!("{channel}-finale");
+        let ack = channels
+            .enqueue_register(&finale, &tx, &generation, &connected)
+            .expect("the finale's seed register enqueued");
+        let _ = tokio::time::timeout(Duration::from_secs(2), ack)
+            .await
+            .expect("the finale's seed LISTEN acked (also the queue drain)")
+            .expect("the LISTEN ok");
+        let drops = channels.clone();
+        let drops_tx = tx.clone();
+        let drops_gen = generation.clone();
+        let finale_owned = finale.clone();
+        let finale_dropper = tokio::spawn(async move {
+            drops.enqueue_unregister(&finale_owned, &drops_tx, &drops_gen);
+        });
+        // The decide anchor: wait for the drop's decision to become
+        // visible (the entry vanishing), then race the UNLISTEN inside
+        // its post-decision window.
+        let undecided_deadline = tokio::time::Instant::now() + Duration::from_secs(2);
+        while channels.snapshot().contains(&finale) {
+            assert!(
+                tokio::time::Instant::now() < undecided_deadline,
+                "the drop's decision never became visible"
+            );
+            tokio::time::sleep(Duration::from_millis(1)).await;
+        }
+        let ack = channels
+            .enqueue_register(&finale, &tx, &generation, &connected)
+            .expect("the finale's racing register enqueued");
+        let _ = tokio::time::timeout(Duration::from_secs(2), ack)
+            .await
+            .expect("the finale's racing LISTEN acked")
+            .expect("the LISTEN ok");
+        tokio::time::timeout(Duration::from_secs(60), finale_dropper)
+            .await
+            .expect("the finale dropper finished")
+            .expect("the finale dropper did not panic");
+    }
+    drop(tx);
+    let model = tokio::time::timeout(Duration::from_secs(10), server)
+        .await
+        .expect("the server model drained")
+        .expect("the server model did not panic");
+    let model: Vec<String> = model.into_iter().collect();
+
+    assert_eq!(
+        channels.snapshot(),
+        model,
+        "the server-side LISTEN state must match the registry refcounts \
+         — a live refcount with the server-side LISTEN cleared (or its \
+         inverse) is the review 004 item 4 mismatch"
+    );
+}
+
+/// Unit pin (server-less, paused time) — the bounded-wait primitive's
+/// arms (review 004 item 6): a completing future yields `Done` inside
+/// its bound; a never-completing future yields `Overran` exactly at
+/// the hard deadline (the failed-verify settle's and the listener-
+/// command bounds' mechanism — the old unbounded `poll_task.await`
+/// arm pinned to end); the shutdown flip wins over a pending future
+/// ahead of the bound (a close surfaces through any stuck
+/// listener-path await — the old unbounded `run_command` await sat
+/// outside the shutdown select).
+#[tokio::test(start_paused = true)]
+async fn bounded_wait_done_overran_and_shutdown_arms() {
+    use crate::forwarder::{BoundedOutcome, bounded_wait};
+    use tokio::sync::watch;
+
+    let (shutdown_tx, mut shutdown_rx) = watch::channel(false);
+    // The pinned arms never flip this watch (the flip arms ride their
+    // own watch below); the sender is just kept alive.
+    let _keep = shutdown_tx;
+
+    // Done inside the bound.
+    let outcome = bounded_wait(Duration::from_secs(3), &mut shutdown_rx, async { 7u32 }).await;
+    assert!(
+        matches!(outcome, BoundedOutcome::Done(7)),
+        "got {outcome:?}"
+    );
+
+    // Overran exactly at the hard deadline (the paused runtime
+    // advances the deadline sleep; the future never completes).
+    let started = tokio::time::Instant::now();
+    let outcome = bounded_wait(
+        Duration::from_secs(3),
+        &mut shutdown_rx,
+        std::future::pending::<u32>(),
+    )
+    .await;
+    assert!(
+        matches!(outcome, BoundedOutcome::Overran),
+        "got {outcome:?}"
+    );
+    assert_eq!(
+        started.elapsed(),
+        Duration::from_secs(3),
+        "the bound is a hard deadline, not best-effort"
+    );
+
+    // Shutdown pre-wins: the flip resolves the select at its moment,
+    // ahead of both the pending future and a much wider bound.
+    let (flip_tx, mut flip_rx) = watch::channel(false);
+    let flipper = tokio::spawn(async move {
+        tokio::time::sleep(Duration::from_secs(1)).await;
+        let _ = flip_tx.send(true);
+    });
+    let started = tokio::time::Instant::now();
+    let outcome = bounded_wait(
+        Duration::from_secs(99),
+        &mut flip_rx,
+        std::future::pending::<()>(),
+    )
+    .await;
+    assert!(
+        matches!(outcome, BoundedOutcome::Shutdown),
+        "got {outcome:?}"
+    );
+    assert_eq!(
+        started.elapsed(),
+        Duration::from_secs(1),
+        "the flip surfaces through the bounded await"
+    );
+    flipper.await.expect("the flipper finished");
+}
+
+/// Unit pin (server-less, paused time) — the failed-verify settle's
+/// poll-task semantics (review 004 item 6 arm 1, exactly the loop's
+/// use): a reaper that never ends (the session alive-but-unusable —
+/// the old shape's park-forever operand) is bounded and then aborted
+/// by the fall-through (the connection closes with it); a reaper that
+/// ends promptly reaps cleanly (the genuinely-dead session's shape,
+/// the existing LISTEN-window-death harness test's path); and a close
+/// during the settle surfaces through the bounded wait in place of
+/// the park.
+#[tokio::test(start_paused = true)]
+async fn failed_verify_settle_bounds_aborts_and_surfaces_shutdown() {
+    use crate::forwarder::{BoundedOutcome, LISTEN_SETTLE_TIMEOUT, bounded_wait};
+    use tokio::sync::watch;
+
+    let (shutdown_tx, mut shutdown_rx) = watch::channel(false);
+    // The first two arms never flip this watch (the flip arm rides
+    // its own watch below).
+    let _keep = shutdown_tx;
+
+    // Alive-but-unusable: the reaper never ends — bounded, then abort.
+    let mut wedged = tokio::spawn(std::future::pending::<()>());
+    let started = tokio::time::Instant::now();
+    match bounded_wait(LISTEN_SETTLE_TIMEOUT, &mut shutdown_rx, &mut wedged).await {
+        BoundedOutcome::Overran => wedged.abort(),
+        other => panic!("an unsettled reaper must overrun its bound, got: {other:?}"),
+    }
+    assert_eq!(
+        started.elapsed(),
+        LISTEN_SETTLE_TIMEOUT,
+        "the settle is bounded, not an unbounded park"
+    );
+    let err = wedged
+        .await
+        .expect_err("the aborted reaper completed cancelled");
+    assert!(
+        err.is_cancelled(),
+        "the fall-through's abort closed the connection with the reaper: {err:?}"
+    );
+
+    // Genuinely dead: the reaper ends promptly (the existing
+    // harness-tested death path reaps under the bound).
+    let mut dead = tokio::spawn(async {});
+    match bounded_wait(LISTEN_SETTLE_TIMEOUT, &mut shutdown_rx, &mut dead).await {
+        BoundedOutcome::Done(_) => {}
+        other => panic!("a promptly-dying reaper must reap, got: {other:?}"),
+    }
+
+    // A close during the settle surfaces promptly (prompt reclaim —
+    // the settle must not hold shutdown hostage either).
+    let (flip_tx, mut flip_rx) = watch::channel(false);
+    let mut wedged2 = tokio::spawn(std::future::pending::<()>());
+    let flipper_tx = flip_tx.clone();
+    let started = tokio::time::Instant::now();
+    let flipper = tokio::spawn(async move {
+        tokio::time::sleep(Duration::from_millis(300)).await;
+        let _ = flipper_tx.send(true);
+    });
+    match bounded_wait(LISTEN_SETTLE_TIMEOUT, &mut flip_rx, &mut wedged2).await {
+        BoundedOutcome::Shutdown => {
+            wedged2.abort();
+            let _ = wedged2.await;
+        }
+        other => panic!("a close must surface through the settle, got: {other:?}"),
+    }
+    assert_eq!(
+        started.elapsed(),
+        Duration::from_millis(300),
+        "the settle's shutdown exit is prompt"
+    );
+    flipper.await.expect("the flipper finished");
+}
+
+/// The black-hole proxy binding: the local listen address to point the
+/// store's DSN at, the engage flag, and the eaten-bytes counter
+/// (engaged = server→client bytes read-and-silently-dropped for every
+/// proxied connection — the reads on a *live* connection hang forever
+/// while both TCP ends stay up, the one failure shape backend kills
+/// cannot reach: the session is alive, its transport attached, but
+/// unresponsive; the counter makes "the server's response arrived and
+/// was eaten" observable, so the test can pin that the serve loop is
+/// demonstrably parked inside its command await before closing).
+struct BlackHoleProxy {
+    addr: std::net::SocketAddr,
+    engaged: Arc<AtomicBool>,
+    eaten: Arc<AtomicU64>,
+}
+
+impl BlackHoleProxy {
+    /// Drop the eaten responses from here on (per connection: applies
+    /// to every proxied connection's backend→client direction).
+    fn engage(&self) {
+        self.engaged.store(true, Ordering::SeqCst);
+    }
+}
+
+/// Spawn the black-hole proxy in front of `upstream_host:port` (the
+/// harness server). Pass-through until engaged; the client→backend
+/// direction always relays (so the backend stays alive and healthy —
+/// only the responses vanish), and the client's close propagates to
+/// the upstream as the backend's write-half shutdown — which is how
+/// the reclaimed session is measured server-side.
+fn spawn_black_hole_proxy(
+    upstream_host: &str,
+    upstream_port: u16,
+) -> std::io::Result<BlackHoleProxy> {
+    let engaged = Arc::new(AtomicBool::new(false));
+    let eaten = Arc::new(AtomicU64::new(0));
+    let listener = std::net::TcpListener::bind(("127.0.0.1", 0))?;
+    listener.set_nonblocking(true)?;
+    let addr = listener.local_addr()?;
+    let listener = tokio::net::TcpListener::from_std(listener)?;
+    let engaged_loop = engaged.clone();
+    let eaten_loop = eaten.clone();
+    let upstream = (upstream_host.to_string(), upstream_port);
+    tokio::spawn(async move {
+        loop {
+            let Ok((client_sock, _)) = listener.accept().await else {
+                return;
+            };
+            let Ok(backend_sock) = tokio::net::TcpStream::connect(upstream.clone()).await else {
+                continue;
+            };
+            let (mut client_read, mut client_write) = tokio::io::split(client_sock);
+            let (mut backend_read, mut backend_write) = tokio::io::split(backend_sock);
+            tokio::spawn(async move {
+                let _ = tokio::io::copy(&mut client_read, &mut backend_write).await;
+                let _ = tokio::io::AsyncWriteExt::shutdown(&mut backend_write).await;
+            });
+            let engaged = engaged_loop.clone();
+            let eaten = eaten_loop.clone();
+            tokio::spawn(async move {
+                let mut buf = [0u8; 8192];
+                loop {
+                    match tokio::io::AsyncReadExt::read(&mut backend_read, &mut buf).await {
+                        Ok(0) | Err(_) => {
+                            let _ = tokio::io::AsyncWriteExt::shutdown(&mut client_write).await;
+                            return;
+                        }
+                        Ok(n) => {
+                            if engaged.load(Ordering::SeqCst) {
+                                eaten.fetch_add(n as u64, Ordering::SeqCst);
+                                continue;
+                            }
+                            if tokio::io::AsyncWriteExt::write_all(&mut client_write, &buf[..n])
+                                .await
+                                .is_err()
+                            {
+                                let _ = tokio::io::AsyncWriteExt::shutdown(&mut client_write).await;
+                                return;
+                            }
+                        }
+                    }
+                }
+            });
+        }
+    });
+    Ok(BlackHoleProxy {
+        addr,
+        engaged,
+        eaten,
+    })
+}
+
+/// Acceptance (review 004 item 6 arm 2, end-to-end through the
+/// black-hole proxy seam): a queued command hangs the serve loop
+/// inside its LISTEN/UNLISTEN await on a black-holed (alive but
+/// unresponsive) connection — the review's old shape: that unbounded
+/// await sat outside the shutdown select, so `close()` could not reach
+/// it and the listener session leaked past shutdown. With the
+/// listener-path bounds the close surfaces through the stuck await
+/// (the bounded wait's shutdown arm), and the loop's exit closes the
+/// connection — reclaimed: the store's exact `application_name`
+/// vanishes from `pg_stat_activity` inside a bounded window (the
+/// proxy propagating the client's close to the backend).
+#[tokio::test(flavor = "multi_thread")]
+async fn close_reclaims_the_listener_session_through_a_black_holed_connection() {
+    let (host, port, user, password, db) = match harness_parts() {
+        Some(parts) => parts,
+        None => {
+            eprintln!("skip: no harness server");
+            return;
+        }
+    };
+    let proxy = match spawn_black_hole_proxy(&host, port) {
+        Ok(proxy) => proxy,
+        Err(e) => panic!("the black-hole proxy must bind: {e}"),
+    };
+    let proxied_dsn = format!(
+        "host={} port={} user={user} password={password} dbname={db}",
+        proxy.addr.ip(),
+        proxy.addr.port()
+    );
+    let schema = instance_namer("blackhole")();
+    let store = open_store(&proxied_dsn, test_opts(&schema)).await.unwrap();
+    let forwarder = store.forwarder().clone();
+    let admin = harness_client().await.unwrap();
+    let app_name = store.listener_application_name().to_string();
+
+    let channel = format!("bh_probe_{}", instance_namer("ch")());
+    forwarder.register(&channel).await.unwrap();
+    assert_eq!(
+        listener_sessions(&admin, &app_name).await,
+        1,
+        "the listener session is up pre-engagement (riding the proxy)"
+    );
+
+    // Engage: the server's responses are eaten from here on; the
+    // listener connection's reads hang (both TCP ends up, the session
+    // alive-and-attached server-side).
+    proxy.engage();
+
+    // A queued command then hangs the serve loop inside its UNLISTEN
+    // await: the command reaches the backend (always relayed), its
+    // response is eaten — once the eaten counter moves, the serve loop
+    // is demonstrably parked inside the unresponsive await.
+    forwarder.unregister(&channel);
+    let eaten_before = std::time::Instant::now();
+    loop {
+        if proxy.eaten.load(Ordering::SeqCst) > 0 {
+            break;
+        }
+        assert!(
+            eaten_before.elapsed() < Duration::from_secs(5),
+            "the UNLISTEN's response must have been eaten while the client hangs \
+             (the serve loop parked inside the black-holed await)"
+        );
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    }
+    assert_eq!(
+        listener_sessions(&admin, &app_name).await,
+        1,
+        "the session is still alive server-side (alive-but-unresponsive)"
+    );
+
+    // Close: the flip reaches the stuck serve loop through the bounded
+    // wait (never the old hostage-taking); the loop's exit closes the
+    // connection, the proxy propagates the close, and the backend
+    // session ends inside a bounded window.
+    store.close();
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(15);
+    let reclaimed = loop {
+        if tokio::time::Instant::now() >= deadline {
+            break false;
+        }
+        if listener_sessions(&admin, &app_name).await == 0 {
+            break true;
+        }
+        tokio::time::sleep(Duration::from_millis(250)).await;
+    };
+    assert!(
+        reclaimed,
+        "close must reclaim the listener session even with the serve loop \
+         stuck awaiting a command on the black-holed connection"
+    );
+
+    // It stays reclaimed: the loop exited (no reconnect re-issued a
+    // listener through the still-engaged proxy's pass-through).
+    tokio::time::sleep(Duration::from_millis(600)).await;
+    assert_eq!(
+        listener_sessions(&admin, &app_name).await,
+        0,
+        "no listener reappears after close (the loop exited)"
+    );
+
+    // Post-close ops fail closed (the engine-wide posture, through the
+    // same store — closed refuses before any round trip).
+    let err = store
+        .notify("post_blackhole_close", serde_json::json!({}))
+        .await
+        .unwrap_err();
+    assert!(
+        matches!(err, alkstore::Error::Database(_)),
+        "post-close notify fails closed, got: {err:?}"
+    );
+
+    // The subscribers' terminal arm: subscribe past the loop's exit
+    // fails closed (the release invariant at this exit shape).
+    assert!(
+        forwarder.subscribe().is_none(),
+        "subscribe after the reclaimed loop must fail closed"
+    );
+
+    drop_schema(&admin, &schema).await;
+}

tasks/pg-fix-forwarder-hardening.md

@@ -1,7 +1,7 @@
 ---
 id: pg-fix-forwarder-hardening
 name: Forwarder TOCTOU restructure + hang-shaped arms (review 004 items 4+6)
-status: pending
+status: completed
 depends_on: []
 scope: broad
 risk: high
@@ -56,20 +56,20 @@ connector injection, the generation tests) for the new pins.
 
 ## Acceptance Criteria
 
-- [ ] The TOCTOU is closed: register/unregister interleavings across
+- [x] The TOCTOU is closed: register/unregister interleavings across
       tasks cannot leave a live refcount with the server-side LISTEN
       cleared — pinned by a test that interleaves the two and asserts
       the server-side LISTEN state matches the refcount
-- [ ] A failed generation-verify/LISTEN batch no longer parks forever:
+- [x] A failed generation-verify/LISTEN batch no longer parks forever:
       bounded wait with an honest error or reconnect — pinned by test
-- [ ] `close()`/`Drop` reclaims the listener session even when the
+- [x] `close()`/`Drop` reclaims the listener session even when the
       serve loop is stuck awaiting on a dead connection — pinned by
       test (black-holed connection seam)
-- [ ] The preserved postures listed above re-verified green (existing
+- [x] The preserved postures listed above re-verified green (existing
       forwarder pins: reconnect survival, no-replay, shutdown close,
       stale-UNLISTEN generation tests, the register-failure
       unregister pin)
-- [ ] `cargo test -p alkstore-postgres` (harness server), clippy
+- [x] `cargo test -p alkstore-postgres` (harness server), clippy
       `-D warnings`, fmt clean; workspace gates green server-less
 
 ## References
@@ -86,8 +86,171 @@ connector injection, the generation tests) for the new pins.
 
 ## Notes
 
-> To be filled by implementation agent
+> Decisions of record the implementation made that the description
+> didn't pin:
+
+- **The real interleaving window was thread preemption, not a yield.**
+  Locating the TOCTOU on pickup: the old `register` (add → connected
+  check → send) and `unregister` (remove → send) each had no `await`
+  between their decide and their send, so the split is not an
+  interleave point under tokio's cooperative scheduling *within one
+  thread* — but the multi-thread runtime's worker threads preempt
+  mid-unit, so thread-A's decide can land, thread-B's whole unit runs
+  (decide + send), and thread-A resumes and sends its stale-contrary
+  command. That is the review item 4's shape, and it quiesces
+  persistently: registry count 1 with the queue `[L, U]` (UNLISTEN
+  replayed after the live registration's LISTEN). The fix makes
+  decide+send one critical section under the registry guard, so the
+  queue order is the registry transitions' linearization.
+- **Every registration keeps its own (redundant) LISTEN + ack.** A
+  shape that skipped the redundant LISTEN for shared (non-new)
+  refcounts was considered and rejected: it would let a later
+  `listen()` return `Ok` while the *first* registration's LISTEN still
+  sat queued — breaking the "a `listen()` that returned sees every
+  later commit / starts from now" contract for that caller. The
+  redundant per-registration LISTEN is the ack's vehicle; semantics
+  preserved exactly (documented in `enqueue_register`).
+- **The decide-and-queue lives on `ChannelSet`** (`enqueue_register` /
+  `enqueue_unregister`), not on `Forwarder`: the registry guard hosts
+  the critical section; the `connected` read and the generation-tag
+  load ride the same section; the ack receiver is returned so the
+  caller awaits *outside* the guard (never a std lock across an
+  await). The old standalone `add_subscriber`/`remove_subscriber`
+  were folded into the under-guard decisions and deleted — the
+  refcount unit pin (`channel_set_refcounts_last_subscriber_rule`) was
+  reworked to drive the decide-and-queue API (now it also pins that
+  each registration derives its own LISTEN, and the mid-reconnect
+  Err-with-entry-stays posture).
+- **One bounded-wait primitive for the whole treatment family**
+  (`bounded_wait`: output / bound expiry / shutdown flip, `biased`
+  toward shutdown): the verify batch, the replay commands, the serve
+  commands, and the failed-verify settle all ride it. Constants:
+  `LISTENER_QUERY_TIMEOUT` (5 s) and `LISTEN_SETTLE_TIMEOUT` (2 s) —
+  generous against a merely-slow server (an overrun re-issues from
+  the snapshot on reconnect; nothing is lost either way; the settle
+  sits far above real death-detection latency, which surfaces in
+  `poll_message` at once).
+- **Death postures, two shapes.** (1) A verify overrun/failure runs
+  the bounded settle: the poll task (the reaper) gets
+  `LISTEN_SETTLE_TIMEOUT`; an unsettled reaper is *aborted* — the
+  connection closes with it (an aborted settle's socket close absorbs
+  the reaper's job). (2) A serve/replay command overrun is treated as
+  honest connection death *directly* (abort + fall to the reconnect),
+  not via a second settle wait — the unresponsive connection is
+  already proven; the reaper is aborted immediately and the
+  reconnect arm proceeds. Every bounded await selects the shutdown
+  watch alongside (`biased`), so `close()` surfaces through any stuck
+  listener-path await — the review item 6 arm 2's hostage shape is
+  closed at *all* of its await sites (the review cited only the serve
+  arm's `run_command`; the verify batch and the generation-top replay
+  had the same shape and got the same treatment family).
+- **`first_listen` flip moved to the verify's success** (before the
+  drain/replay): a generation that dies mid-replay keeps the wake for
+  the *next* generation's top (a replay-death generation never
+  broadcasts its own wake). The existing exactly-one-wake pin
+  (LISTEN-window death) stays green — re-derived against the new
+  order.
+- **(c)'s live harness shape is a TCP black-hole proxy seam** (the
+  task's "black-holed connection seam"): a local pass-through TCP
+  proxy in front of the harness server (the store's DSN points at
+  it); engagement eats server→client bytes (reads hang forever, both
+  TCP ends up — the one failure shape backend kills cannot reach:
+  alive, attached, unresponsive). An eaten-bytes counter pins that
+  the serve loop is demonstrably parked inside the UNLISTEN await
+  before `close()`, and the reclaim is measured server-side by the
+  store's exact `application_name` vanishing from `pg_stat_activity`
+  (the proxy propagates the client's close to the backend). No
+  fabricatable alternatives exist for the connection types (the
+  Client/Connection pair cannot be faked), and OS-level freezes
+  (SIGSTOP) cannot reach dockerized harness backends — the network
+  layer is the testable seam. `LISTENER_QUERY_TIMEOUT`,
+  `LISTEN_SETTLE_TIMEOUT`, `bounded_wait`, and `BoundedOutcome` are
+  `pub(crate)`.
+- **(b)'s harness-level live arrangement is not separately added** —
+  the stale-unlisten precedent's bar applies ("the unit pin is the
+  acceptance bar"): the failed-verify settle is pinned server-less in
+  paused time (a wedged reaper bounded + aborted + cancelled, a
+  promptly-dying reaper reaped, a close surfacing through the settle
+  promptly), and the genuinely-dead-verify path through the real loop
+  stays covered by the existing LISTEN-window-death harness test
+  (`forwarder_recovers_from_connection_death_during_the_listen_window`),
+  which reaps under the bound and recovers unchanged.
+- The stale-generation machinery (tags, counter bump, the
+  generation-top drain, the recv-arm staleness guard), the retry
+  core, the fanout release guard, the register-failure unregister
+  discipline, and the poison-recovery lock idiom are all preserved —
+  re-verified green (existing pins' full battery in Gates).
+- **Replay-proofs (both ways, verified live):**
+  (1) with the old decide/send split temporarily reintroduced in
+  `enqueue_unregister`/`enqueue_register` (a widened post-decision
+  window), the interleave pin fails on exactly the review's assert
+  (registry/model divergence — a live refcount at a cleared
+  server-side LISTEN); with the fix restored it passes instantly and
+  deterministically (the finale anchors on the observable decision —
+  the entry vanishing — so a commit-time run never flirts with the
+  timing).
+  (2) with the serve arm's bounded await temporarily reverted to the
+  old unbounded `run_command` await, the black-hole harness test
+  fails on "close must reclaim the listener session…" (the session
+  leaked past close, as reviewed); with the fix restored it reclaims
+  (2.4 s wall, well inside the 5 s command bound + the 15 s
+  observation window).
 
 ## Summary
 
-> To be filled on completion
+> What landed, verified how:
+
+- **`alkstore-postgres/src/forwarder.rs`**:
+  (a) the TOCTOU — `ChannelSet` gained the decide-and-queue critical
+  sections (`enqueue_register`: refcount + connected-check + acked
+  LISTEN enqueue under one registry guard; `enqueue_unregister`:
+  last-subscriber decision + UNLISTEN enqueue under one guard);
+  `Forwarder::register`/`unregister` ride them (the ack await is
+  outside the guard); the standalone refcount helpers folded away;
+  the module/method docs state the invariant (the queue's order is
+  the registry transitions' linearization; a live refcount can never
+  sit against a cleared server-side LISTEN).
+  (b) the failed-verify settle — the old unbounded
+  `poll_task.await` is replaced by the bounded settle
+  (`LISTEN_SETTLE_TIMEOUT`=2 s, shutdown-selected): reaped fast when
+  the session is genuinely dead, aborted at the bound (the connection
+  closes with the reaper) when it is alive-but-unusable, then the
+  reconnect either way.
+  (c) the close-time leak — a new `bounded_wait` primitive bounds
+  every listener-client query (5 s) with the shutdown watch selected
+  alongside: the verify batch, the generation-top replay commands,
+  and the serve loop's LISTEN/UNLISTEN commands; an overrun is the
+  honest death posture (abort + reconnect; the snapshot re-issue
+  recovers, the stalled LISTEN's caller sees the transient error
+  family), so `close()`/`Drop` can no longer leak the listener
+  session through a stuck await.
+- **Pins added** (`open_tests.rs`, 4 new; `notify_tests.rs`, 1
+  reworked):
+  `interleaved_register_unregister_keep_server_listen_set_matching_registry`
+  (server-less: a server-model task re-plays the command queue in
+  FIFO order onto a LISTEN model, two interleaving tasks hammer
+  register/unregister on two channels, a per-channel deterministic
+  finale races a last-subscriber drop against one re-register with
+  the decision anchored on the registry snapshot — the final
+  registry/model equality is the pin);
+  `bounded_wait_done_overran_and_shutdown_arms` and
+  `failed_verify_settle_bounds_aborts_and_surfaces_shutdown`
+  (paused-time, the settle/query bounds' arms incl. prompt close);
+  `close_reclaims_the_listener_session_through_a_black_holed_connection`
+  (harness, the TCP black-hole proxy seam: eaten-response proof the
+  serve arm is parked, close reclaims inside a bounded window, the
+  session stays reclaimed, post-close ops fail closed, subscribe past
+  the loop exit fails closed);
+  `channel_set_refcounts_last_subscriber_rule` reworked onto the
+  decide-and-queue API (last-subscriber rule, redundant-LISTEN ack
+  vehicle, past-zero no-op, mid-reconnect Err + entry-stays + nothing
+  enqueued).
+- **`Gates`**: `cargo test -p alkstore_postgres` with the harness
+  server green (131 lib + 25 contract-suite + 9 schema; lib battery
+  run twice full), every pre-existing forwarder/reconnect/no-replay/
+  shutdown/stale-UNLISTEN/LISTEN-window-death pin green unchanged;
+  workspace `cargo build`, server-less `cargo test --workspace`,
+  `cargo clippy --all-targets -- -D warnings` (workspace),
+  `cargo fmt --check`
+  — all green; the doc gate (`RUSTDOCFLAGS="-D warnings" cargo doc
+  --no-deps`) green.