Postgres engine: LISTEN forwarder full behavior + notify/listen — wake contract pg arm

- notify: auto-commit pg_notify path, 8000-byte typed client-side check
  through the tx seam's NOTIFY_PAYLOAD_LIMIT (one limit owner), closed-store
  fail-closed before payload work
- listen: acked synchronous channel registration (listen starts-from-now —
  a notify racing the LISTEN cannot be lost), refcounted ChannelSet
  (UNLISTEN at last-subscriber drop), PgWakeReceiver bridging Wake { channel }
  only, channel-scoped fanout + reserved reconnect-wake to every subscriber,
  Lagged(n) surfaced-not-silent
- receiver close semantics (ADR-021 §5 pg arm): stays open across forwarder
  reconnects, terminal None only at engine shutdown — Forwarder::shutdown
  takes the fanout sender so receiver-held Arc lifetimes can't pin the
  broadcast open
- tests: wake-arrives, 7999/8000/8002 boundary both entry points, no-replay
  during connection gaps, backend-kill reconnect through the receiver stack,
  drop-unregisters (behavioral probe — pg_listening_channels is per-session),
  validation both paths, the two POC deadlock pitfalls re-pinned, 11 new
  tests green against the harness server, gates green server-less

(task pg-engine-notify-listen)

alkstore-postgres/src/forwarder.rs

@@ -77,55 +77,67 @@ pub(crate) fn quote_ident(name: &str) -> String {
     format!("\"{}\"", name.replace('"', "\"\""))
 }
 
-/// The raw notification the poll loop fans out — the skeleton's
-/// broadcast content; the notify-listen task maps it to the opaque
-/// [`alkstore::Wake`].
-///
-/// `channel` is read by the notify-listen task's receiver bridging
-/// (and this wave's fanout-proof tests); ungated in tests, gated in
-/// the lib build until that task wires it (the wave-3 intermediate
-/// posture — clippy `-D warnings` rejects dead code).
+/// The raw notification the poll loop fans out — the notify task's
+/// receiver bridging maps it to the opaque [`alkstore::Wake`]
+/// (`Wake { channel }` only — payloads never surface, ADR-008 §3).
 #[derive(Debug, Clone)]
 pub(crate) struct RawNotification {
-    #[cfg_attr(not(test), allow(dead_code))]
     pub channel: String,
 }
 
 /// The dynamic channel set: the registry the forwarder re-issues
-/// LISTENs from after every reconnect. The notify-listen task's
-/// `listen` (registration) and receiver-drop (unregistration, the
-/// refcounted last-subscriber rule) ride [`Forwarder`]'s register /
-/// unregister.
+/// LISTENs from after every reconnect. `listen` (registration) and
+/// receiver-drop (unregistration, the refcounted last-subscriber
+/// rule) ride [`Forwarder`]'s register / unregister.
 #[derive(Debug, Default)]
 pub(crate) struct ChannelSet {
-    channels: std::sync::Mutex<std::collections::BTreeSet<String>>,
+    /// BTreeMap for the deterministic snapshot; the value is the
+    /// subscriber refcount (the last-subscriber UNLISTEN rule).
+    channels: std::sync::Mutex<std::collections::BTreeMap<String, u64>>,
 }
 
 impl ChannelSet {
-    // `insert`/`remove` are the notify-listen task's registry entry
-    // points (this wave's tests exercise them); gated in the lib
-    // build until that task wires them.
-    #[cfg_attr(not(test), allow(dead_code))]
-    fn insert(&self, channel: &str) -> bool {
-        self.channels
+    /// 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()
-            .expect("channel registry mutex poisoned")
-            .insert(channel.to_string())
+            .expect("channel registry mutex poisoned");
+        let is_new = !map.contains_key(channel);
+        *map.entry(channel.to_string()).or_insert(0) += 1;
+        is_new
     }
 
-    #[cfg_attr(not(test), allow(dead_code))]
-    fn remove(&self, channel: &str) -> bool {
-        self.channels
+    /// 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()
-            .expect("channel registry mutex poisoned")
-            .remove(channel)
+            .expect("channel registry mutex poisoned");
+        let Some(count) = map.get_mut(channel) else {
+            return false;
+        };
+        if *count > 1 {
+            *count -= 1;
+            false
+        } else {
+            map.remove(channel);
+            true
+        }
     }
 
-    fn snapshot(&self) -> Vec<String> {
+    /// The registry's live channel list — the reconnect re-issue
+    /// source, and the tests' registry read.
+    pub(crate) fn snapshot(&self) -> Vec<String> {
         self.channels
             .lock()
             .expect("channel registry mutex poisoned")
-            .iter()
+            .keys()
             .cloned()
             .collect()
     }
@@ -141,38 +153,41 @@ impl ChannelSet {
 /// "listener connection dropped" arm).
 #[derive(Debug)]
 pub(crate) struct Forwarder {
-    /// Read by the notify-listen task's receiver bridging (this
-    /// wave's tests exercise it); gated in the lib build until then.
-    #[cfg_attr(not(test), allow(dead_code))]
-    fanout: broadcast::Sender<RawNotification>,
-    /// Read by the forwarder's reconnect re-issue within the spawned
-    /// loop's own context (`Arc` move); the field itself is read by
-    /// the register/unregister surfaces below.
-    #[cfg_attr(not(test), allow(dead_code))]
+    /// The wake fanout — every receiver bridge holds a subscription;
+    /// its release (this handle's take at shutdown, the loop's drop at
+    /// its exit) is the receivers' terminal `Closed`/`None` arm.
+    /// `Option` because shutdown empties it while receiver-held
+    /// `Arc<Forwarder>`s keep the struct alive.
+    fanout: std::sync::Mutex<Option<broadcast::Sender<RawNotification>>>,
+    /// The channel registry the reconnect re-issues from; read by the
+    /// register/unregister surfaces below.
     channels: Arc<ChannelSet>,
     /// LISTEN/UNLISTEN commands to the live connection (the loop
     /// issues them; queued commands replay harmlessly across
     /// reconnects alongside the registry re-issue). Read by the
     /// register/unregister surfaces below.
-    #[cfg_attr(not(test), allow(dead_code))]
     commands_tx: mpsc::UnboundedSender<ListenCommand>,
     /// Whether a listener connection is currently live — the
     /// register-path's honest transient state (registration during a
     /// gap fails with `Database`; the registry entry's recovery rides
     /// the reconnect re-issue). Read by `register` below.
-    #[cfg_attr(not(test), allow(dead_code))]
     connected: Arc<AtomicBool>,
     shutdown: watch::Sender<bool>,
 }
 
 /// Commands to the live listener connection (built by
-/// `register`/`unregister` below — gated in the lib build until the
-/// notify-listen task wires them).
+/// `register`/`unregister` below). `Listen` carries an ack the
+/// forwarder resolves after the server processed the LISTEN — the
+/// registration is synchronous end-to-end, so a `listen()` that
+/// returned sees every later commit (`listen` **starts from now** —
+/// a notify racing the LISTEN cannot be lost; delivery only starts
+/// at server-side registration).
 #[derive(Debug)]
 enum ListenCommand {
-    #[cfg_attr(not(test), allow(dead_code))]
-    Listen(String),
-    #[cfg_attr(not(test), allow(dead_code))]
+    Listen {
+        channel: String,
+        ack: tokio::sync::oneshot::Sender<alkstore::Result<()>>,
+    },
     Unlisten(String),
 }
 
@@ -209,7 +224,7 @@ impl Forwarder {
         }));
 
         Forwarder {
-            fanout,
+            fanout: std::sync::Mutex::new(Some(fanout)),
             channels,
             commands_tx,
             connected,
@@ -217,57 +232,85 @@ impl Forwarder {
         }
     }
 
-    /// Subscribe to the raw fanout (the notify-listen task bridges
-    /// broadcast receivers into `WakeReceiver`s). Gated in the lib
-    /// build until that task wires it (tests exercise it now).
-    #[cfg_attr(not(test), allow(dead_code))]
-    pub(crate) fn subscribe(&self) -> broadcast::Receiver<RawNotification> {
-        self.fanout.subscribe()
+    /// Subscribe to the raw fanout (the notify task bridges broadcast
+    /// receivers into `WakeReceiver`s). `None` after shutdown (the
+    /// senders are being released) — a subscribe racing `close()` or
+    /// arriving past it fails closed rather than panicking.
+    pub(crate) fn subscribe(&self) -> Option<broadcast::Receiver<RawNotification>> {
+        self.fanout
+            .lock()
+            .expect("fanout mutex poisoned")
+            .as_ref()
+            .map(|s| s.subscribe())
     }
 
-    /// Register a channel AND issue `LISTEN` on the live connection.
+    /// Register a channel AND issue `LISTEN` on the live connection,
+    /// waiting for the server-side registration (the ack — a `listen`
+    /// that returned sees every later commit; `listen` starts from
+    /// now).
+    ///
     /// The registry write happens first: if the connection dies
-    /// between the write and the listendelivery, the reconnect path
+    /// between the write and the LISTEN delivery, the reconnect path
     /// re-issues from the snapshot — a registration is never lost
     /// (the ordering is the registration-recovery discipline).
     ///
-    /// Fails `Database` only when no listener connection is currently
-    /// live (mid-reconnect) — transient; a retry (or the notify-listen
-    /// task's refined posture) recovers it.
-    #[cfg_attr(not(test), allow(dead_code))]
-    pub(crate) fn register(&self, channel: &str) -> Result<(), alkstore::Error> {
-        self.channels.insert(channel);
+    /// Refcounted: N subscribers on one channel share one LISTEN.
+    ///
+    /// Fails `Database` when no listener connection is currently live
+    /// (mid-reconnect) or the LISTEN itself failed — transient; a
+    /// retry recovers it (the registry entry stays either way and the
+    /// re-issue covers the channel on reconnect).
+    pub(crate) async fn register(&self, channel: &str) -> Result<(), alkstore::Error> {
+        self.channels.add_subscriber(channel);
         if !self.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 _ = self
+        let (ack_tx, ack_rx) = tokio::sync::oneshot::channel();
+        let sent = self
             .commands_tx
-            .send(ListenCommand::Listen(channel.to_string()));
-        Ok(())
+            .send(ListenCommand::Listen {
+                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)"))?
     }
 
-    /// Unregister a channel AND issue `UNLISTEN` (the notify-listen
-    /// task's last-subscriber rule). Registry write first (ordering
-    /// symmetric with [`Forwarder::register`]'s recovery logic); the
-    /// UNLISTEN is best-effort — a lost one re-issues nothing
-    /// (UNLISTEN is not recovery-relevant: the snapshot won't carry
-    /// the removed channel).
-    #[cfg_attr(not(test), allow(dead_code))]
+    /// 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
+    /// recovery logic); the UNLISTEN is best-effort — a lost one
+    /// re-issues nothing (UNLISTEN is not recovery-relevant: the
+    /// snapshot won't carry the removed channel).
     pub(crate) fn unregister(&self, channel: &str) {
-        self.channels.remove(channel);
+        if self.channels.remove_subscriber(channel) {
             let _ = self
                 .commands_tx
                 .send(ListenCommand::Unlisten(channel.to_string()));
         }
+    }
 
     /// Shutdown: flip the switch — the loop exits, dropping the
     /// listener `Client` and its session (the listener connection is
-    /// gone). Idempotent.
+    /// gone) — and release this handle's fanout sender. The broadcast
+    /// closes once every sender is dropped (the loop's clone dies with
+    /// its exit; this handle's dies here — receivers unaffected by
+    /// `Arc<Forwarder>` lifetimes see the terminal arm). Idempotent.
     pub(crate) fn shutdown(&self) {
         let _ = self.shutdown.send(true);
+        // Take the sender even if a receiver holds this Arc: the
+        // receivers' terminal `None` must fire at engine shutdown
+        // whatever lifetime stands on the handle.
+        *self.fanout.lock().expect("fanout mutex poisoned") = None;
     }
 }
 
@@ -281,6 +324,15 @@ impl Forwarder {
 /// backoff (50 ms → 2 s cap). The loop consumes the handed-in
 /// connection on its first generation and reconnects into the slot
 /// afterwards.
+///
+/// The receiver-facing close shape (ADR-021 §5's pg arm): the fanout
+/// sender clones live here (loop) and on the handle (taken at
+/// shutdown); a connection death and its reconnect are invisible to
+/// subscribers (the fanout never drops on a generation boundary —
+/// receivers stay open across reconnects, they receive the reserved
+/// reconnect-wake instead); the senders drop only at shutdown — the
+/// loop's clone at its exit, the handle's at the shutdown take — the
+/// receivers' terminal arm.
 struct LoopCtx {
     client: tokio_postgres::Client,
     connection: Option<ListenerConnection>,
@@ -376,10 +428,30 @@ async fn forwarder_loop(
 
                     cmd = commands.recv() => {
                         match cmd {
-                            Some(ListenCommand::Listen(ch)) => {
-                                let _ = client
-                                    .batch_execute(&format!("LISTEN {};", quote_ident(&ch)))
-                                    .await;
+                            Some(ListenCommand::Listen { channel, ack }) => {
+                                // 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.
+                                let result = if client
+                                    .batch_execute(&format!(
+                                        "LISTEN {};",
+                                        quote_ident(&channel)
+                                    ))
+                                    .await
+                                    .is_ok()
+                                {
+                                    Ok(())
+                                } else {
+                                    Err(database_error(
+                                        "the listener connection died mid-LISTEN; \
+                                         the channel is registered and re-issued on reconnect",
+                                    ))
+                                };
+                                let _ = ack.send(result);
                             }
                             Some(ListenCommand::Unlisten(ch)) => {
                                 let _ = client

alkstore-postgres/src/lib.rs

@@ -33,6 +33,7 @@
 //! scheduler/outbox); each task replaces the prior stubs.
 
 mod forwarder;
+mod notify;
 mod opts;
 mod resolution;
 mod schema;

alkstore-postgres/src/notify.rs

@@ -0,0 +1,219 @@
+//! The notify/listen mechanism's Postgres arm — the wake half
+//! (`pg-engine-notify-listen`; `notify_tx` landed with the seam):
+//!
+//! - `notify` — the auto-commit path: pool checkout, `SELECT
+//!   pg_notify($1, $2)` in its own implicit transaction, return to
+//!   pool. Commit-atomic by `NOTIFY`'s native transactional delivery
+//!   (ADR-007): the fire happens at commit, so a rollback is
+//!   impossible on this path (nothing holds it back). The 8000-byte
+//!   boundary is the client-side typed check (ADR-016 §5) — `len >=
+//!   limit` (the server's NUL-inclusive wire budget), measured on the
+//!   serde_json serialization, before any round trip. One owner of
+//!   the limit constant: [`crate::tx::NOTIFY_PAYLOAD_LIMIT`].
+//! - `listen` — the forwarder-fanout bridge: `validate`, register the
+//!   channel with the forwarder (the registry entry rides the
+//!   reconnect re-issue), subscribe its broadcast, and return the
+//!   boxed [`PgWakeReceiver`] bridging raw notifications into
+//!   [`Wake`] values scoped per channel (`Wake { channel }` only —
+//!   ADR-008 §3). The broadcast bridge handles the three subscriber
+//!   arms: `Ok` (a wake), `Lagged(n)` (logged-and-skipped — lag is
+//!   surfaced, not silent; the missed wakes are the contract's
+//!   best-effort-hint shape, recovery is re-read), `Closed` (engine
+//!   shutdown — the terminal arm).
+//!
+//! Close semantics (ADR-021 §5's pg arm — the documented engine
+//! asymmetry): the receiver **stays open across forwarder
+//! reconnects** (reconnect-wakes arrive on it — the broadcast
+//! subscription outlives connection generations) and closes only at
+//! engine shutdown (`close()`/[`Drop` for PgStore] → the forwarder's
+//! fanout sender drops → every receiver terminal `None`). `try_recv`
+//! / `recv_timeout` distinguish `Err(Closed)` from idle's `Ok(None)`;
+//! `recv_timeout`'s timeout expiry without a wake is `Ok(None)` —
+//! never a timeout signal.
+//!
+//! Unsubscribe is the receiver's own [`Drop`]: it drops its broadcast
+//! subscription and decrements the channel's subscriber refcount, and
+//! unregisters (= `UNLISTEN`) when the **last** subscriber for that
+//! channel drops — zero subscribers ⇒ no LISTEN needed.
+
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::Duration;
+
+use deadpool_postgres::Pool;
+use serde_json::Value;
+
+use alkstore::{BoxedFuture, Error, Result, Wake, WakeReceiver};
+
+use crate::forwarder::Forwarder;
+use crate::seam::{database_error, pg_error, pool_error};
+use crate::tx::{NOTIFY_PAYLOAD_LIMIT, encode_payload_bytes};
+
+/// The auto-commit notify path (see the module docs). Pool checkout,
+/// one `SELECT pg_notify`, return to pool. Name validation fires at
+/// the entry point, before any round trip and before the payload is
+/// touched (ADR-008 §4).
+pub(crate) fn notify(
+    pool: Pool,
+    closed: Arc<AtomicBool>,
+    channel: String,
+    payload: Value,
+) -> BoxedFuture<'static, Result<()>> {
+    if let Err(e) = alkstore::validate_shared_name(&channel) {
+        return Box::pin(async move { Err(e) });
+    }
+    Box::pin(async move {
+        // Closed-store fail-closed dominates the payload work (a
+        // post-close notify is a `Database` whatever the payload —
+        // the engine-wide closed-store posture).
+        if closed.load(Ordering::Acquire) {
+            return Err(database_error("the store is closed"));
+        }
+        let bytes = encode_payload_bytes(&payload)?;
+        // One owner of the limit: the same constant `notify_tx` checks
+        // (the seam task's boundary — this task wires the auto-commit
+        // twin through it). Same `len >= limit` predicate (the
+        // NUL-inclusive wire budget), same typed
+        // `PayloadTooLarge { limit: 8000 }` before any round trip.
+        if bytes.len() >= NOTIFY_PAYLOAD_LIMIT {
+            return Err(Error::PayloadTooLarge {
+                limit: NOTIFY_PAYLOAD_LIMIT,
+            });
+        }
+        let client = pool.get().await.map_err(pool_error)?;
+        let text = String::from_utf8(bytes)
+            .map_err(|e| Error::Codec(format!("notify payload must be utf-8 text: {e}")))?;
+        client
+            .query_one("SELECT pg_notify($1, $2)", &[&channel, &text])
+            .await
+            .map_err(pg_error)?;
+        Ok(())
+    })
+}
+
+/// The listen path (see the module docs). Name validation fires at
+/// the entry point; a listen against a closed store fails closed with
+/// [`Error::Database`] (the engine-wide closed-store posture —
+/// `begin_tx`/`notify` yield the same shape).
+pub(crate) fn listen(
+    forwarder: Arc<Forwarder>,
+    closed: Arc<AtomicBool>,
+    channel: String,
+) -> BoxedFuture<'static, Result<Box<dyn WakeReceiver>>> {
+    if let Err(e) = alkstore::validate_shared_name(&channel) {
+        return Box::pin(async move { Err(e) });
+    }
+    Box::pin(async move {
+        if closed.load(Ordering::Acquire) {
+            return Err(Error::database(std::io::Error::other(
+                "the store is closed",
+            )));
+        }
+        forwarder.register(&channel).await?;
+        let broadcast_rx = forwarder.subscribe();
+        let Some(broadcast_rx) = broadcast_rx else {
+            // The subscribe raced (or arrived past) a shutdown: fail
+            // closed — no receiver can ever deliver again. Unregister
+            // first: the acked LISTEN must not leak past the failed
+            // listen (a next-closest `listen` re-issues it).
+            forwarder.unregister(&channel);
+            return Err(database_error("the store is closed"));
+        };
+        let (wake_tx, wake_rx) = tokio::sync::mpsc::channel(bridge_capacity());
+        let bridge = tokio::spawn(bridge_loop(broadcast_rx, wake_tx, channel.clone()));
+        Ok(Box::new(PgWakeReceiver {
+            wake_rx,
+            _bridge: bridge,
+            forwarder,
+            channel,
+        }) as Box<dyn WakeReceiver>)
+    })
+}
+
+/// Per-subscriber capacity for the mpsc bridge (the wake handoff
+/// broadcast → receiver). The wake rate is the notify rate (bursty at
+/// worst); 1024 matches the fanout's bounded shape, lag surfaced at
+/// the broadcast layer (not here).
+fn bridge_capacity() -> usize {
+    1024
+}
+
+/// The broadcast bridge: loops the raw fanout, maps notifications on
+/// this receiver's channel (plus the reserved reconnect-wake — every
+/// subscriber receives it) into [`Wake`]s, and hands them to the
+/// consumer side. Lag on the raw subscription is surfaced, not silent:
+/// logged-and-skipped (a lagged subscriber's missed wakes are the
+/// contract's best-effort shape), and the loop keeps streaming. The
+/// loop exits when either side drops (consumer drop, or the forwarder
+/// shutdown drops the fanout sender) — the bridge task ends, the
+/// mpsc sender drops, the receiver's `recv()` yields its terminal.
+async fn bridge_loop(
+    mut broadcast_rx: tokio::sync::broadcast::Receiver<crate::forwarder::RawNotification>,
+    wake_tx: tokio::sync::mpsc::Sender<Wake>,
+    channel: String,
+) {
+    loop {
+        match broadcast_rx.recv().await {
+            Ok(raw) => {
+                if raw.channel != channel && raw.channel != alkstore::RESERVED_LISTENER_RECONNECTED
+                {
+                    continue;
+                }
+                if wake_tx.send(Wake::new(raw.channel)).await.is_err() {
+                    return;
+                }
+            }
+            Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
+                eprintln!(
+                    "alkstore-postgres: listener subscriber on channel {channel:?} lagged, \
+                     skipped {n} notifications (recovery is the consumer's re-read)"
+                );
+            }
+            Err(tokio::sync::broadcast::error::RecvError::Closed) => return,
+        }
+    }
+}
+
+/// The consumer-visible wake receiver: the tokio side of the
+/// broadcast bridge (see the module docs for the close semantics —
+/// stays open across reconnects, terminal `None` only at engine
+/// shutdown).
+pub(crate) struct PgWakeReceiver {
+    wake_rx: tokio::sync::mpsc::Receiver<Wake>,
+    /// The bridge task — kept so a receiver drop terminates its
+    /// wake pump (the forwarder may live on with other subscribers).
+    _bridge: tokio::task::JoinHandle<()>,
+    forwarder: Arc<Forwarder>,
+    channel: String,
+}
+impl Drop for PgWakeReceiver {
+    fn drop(&mut self) {
+        self.forwarder.unregister(&self.channel);
+    }
+}
+
+impl WakeReceiver for PgWakeReceiver {
+    fn recv<'a>(&'a mut self) -> BoxedFuture<'a, Option<Wake>> {
+        Box::pin(async move { self.wake_rx.recv().await })
+    }
+
+    fn try_recv(&mut self) -> Result<Option<Wake>> {
+        match self.wake_rx.try_recv() {
+            Ok(wake) => Ok(Some(wake)),
+            Err(tokio::sync::mpsc::error::TryRecvError::Empty) => Ok(None),
+            // Terminal only at engine shutdown (the forwarder's
+            // broadcast dropped) — never at a reconnect gap.
+            Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => Err(Error::Closed),
+        }
+    }
+
+    fn recv_timeout<'a>(&'a mut self, timeout: Duration) -> BoxedFuture<'a, Result<Option<Wake>>> {
+        Box::pin(async move {
+            match tokio::time::timeout(timeout, self.wake_rx.recv()).await {
+                Ok(Some(wake)) => Ok(Some(wake)),
+                Ok(None) => Err(Error::Closed),
+                Err(_expired) => Ok(None),
+            }
+        })
+    }
+}

alkstore-postgres/src/store.rs

@@ -25,7 +25,9 @@
 //! No `spawn_blocking` seam on this engine (the crate docs' posture);
 //! the trait stubs below are the wave-3 posture — they return
 //! `Err(Database("… wiring lands with the … task"))` until the
-//! mechanism tasks replace them.
+//! mechanism tasks replace them. `notify`/`listen` are wired (the
+//! notify-listen task's) — the wake contract's pg arm rides the
+//! forwarder.
 
 use alkstore::Store;
 use std::sync::Arc;
@@ -236,23 +238,25 @@ impl Store for PgStore {
 
     fn notify<'a>(
         &'a self,
-        _channel: &str,
-        _payload: serde_json::Value,
+        channel: &str,
+        payload: serde_json::Value,
     ) -> alkstore::BoxedFuture<'a, alkstore::Result<()>> {
-        if let Err(e) = self.closed_check() {
-            return Box::pin(async move { Err(e) });
-        }
-        Box::pin(async { Err(stub("notify wiring lands with the notify-listen task")) })
+        crate::notify::notify(
+            self.pool.clone(),
+            self.closed.clone(),
+            channel.to_string(),
+            payload,
+        )
     }
 
     fn listen<'a>(
         &'a self,
-        _channel: &str,
+        channel: &str,
     ) -> alkstore::BoxedFuture<'a, alkstore::Result<Box<dyn alkstore::WakeReceiver>>> {
-        if let Err(e) = self.closed_check() {
-            return Box::pin(async move { Err(e) });
-        }
-        Box::pin(async { Err(stub("listen wiring lands with the notify-listen task")) })
+        let forwarder = self.forwarder.clone();
+        let closed = self.closed.clone();
+        let channel = channel.to_string();
+        Box::pin(async move { crate::notify::listen(forwarder, closed, channel).await })
     }
 
     fn stream<'a>(
@@ -341,5 +345,8 @@ impl Store for PgStore {
 #[cfg(test)]
 mod open_tests;
 
+#[cfg(test)]
+mod notify_tests;
+
 #[cfg(test)]
 mod tx_tests;

alkstore-postgres/src/store/notify_tests.rs

@@ -0,0 +1,747 @@
+//! The notify/listen mechanism's acceptance tests
+//! (`pg-engine-notify-listen`): commit-atomic auto-commit notifies
+//! (the typed 8000-byte check on both entry points, one limit owner),
+//! the forwarder-fanout wakes through the bridged receiver
+//! (`Wake { channel }` only, channel-scoped, N subscribers,
+//! multi-channel), the receiver close arms (stays open across
+//! reconnects — the reconnect-wake arrives; terminal `None` only at
+//! engine shutdown; `try_recv`/`recv_timeout`'s pinned shapes), the
+//! no-replay honesty pin, drop-unregistration (UNLISTEN at
+//! last-subscriber drop), entry-point validation, closed-store
+//! fail-closed on both paths, and the two POC deadlock pitfalls
+//! re-pinned through the full receiver stack.
+//!
+//! Harness convention (the schema task's): connection settings ride
+//! the environment (`ALKSTORE_PG_HOST/PORT/USER/PASSWORD/DB`), never
+//! hardcoded; tests without a reachable server skip cleanly so the
+//! workspace gates stay green server-less. Isolation is a fresh unique
+//! schema per test.
+
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
+
+use alkstore::{Error, Store, WakeReceiver};
+use serde_json::json;
+
+use crate::opts::PgOpts;
+use crate::schema::DEFAULT_SCHEMA;
+use crate::store::open_store;
+use crate::tx::NOTIFY_PAYLOAD_LIMIT;
+
+const ENV_HOST: &str = "ALKSTORE_PG_HOST";
+const ENV_PORT: &str = "ALKSTORE_PG_PORT";
+const ENV_USER: &str = "ALKSTORE_PG_USER";
+const ENV_PASSWORD: &str = "ALKSTORE_PG_PASSWORD";
+const ENV_DB: &str = "ALKSTORE_PG_DB";
+
+fn harness_dsn() -> Option<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(format!(
+        "host={host} port={port} user={user} password={password} dbname={db}"
+    ))
+}
+
+fn instance_namer(tag: &str) -> impl Fn() -> String + use<'_> {
+    let counter = AtomicU64::new(0);
+    move || {
+        format!(
+            "{tag}_{}_{}_{}",
+            std::process::id(),
+            std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .unwrap_or_default()
+                .as_nanos(),
+            counter.fetch_add(1, Ordering::SeqCst),
+        )
+    }
+}
+
+/// A raw admin client for cross-connection probes (NOTIFY senders,
+/// backend kills, `pg_listening_channels` accounting) — separate from
+/// the store's pool and from `notify`'s checkout.
+async fn harness_client() -> Option<tokio_postgres::Client> {
+    let (client, connection) = tokio_postgres::connect(&harness_dsn()?, tokio_postgres::NoTls)
+        .await
+        .ok()?;
+    tokio::spawn(async move {
+        let _ = connection.await;
+    });
+    Some(client)
+}
+
+async fn drop_schema(client: &tokio_postgres::Client, schema: &str) {
+    if schema == DEFAULT_SCHEMA {
+        // Never drop the shared default schema in tests — tests that
+        // exercise the default name only assert its shape.
+        return;
+    }
+    let sql = format!(
+        "DROP SCHEMA IF EXISTS {} CASCADE",
+        crate::schema::quote_identifier(schema)
+    );
+    let _ = client.batch_execute(&sql).await;
+}
+
+fn test_opts(schema: &str) -> PgOpts {
+    PgOpts {
+        schema: schema.to_string(),
+        ..PgOpts::default()
+    }
+}
+
+/// Wait for a wake on the receiver with a bounded deadline — polling
+/// loosely so no tight-race assertions ride (the suite's stability
+/// posture). Channel-agnostic: the reconnect-wake and the channel's
+/// own wakes both match (each caller pins the wakes it waits for).
+async fn must_recv(receiver: &mut dyn WakeReceiver, tag: &str) -> alkstore::Wake {
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
+    loop {
+        match receiver.try_recv() {
+            Ok(Some(wake)) => return wake,
+            Ok(None) => {}
+            Err(e) => panic!("{tag}: receiver errored instead of idling: {e}"),
+        }
+        assert!(
+            tokio::time::Instant::now() < deadline,
+            "{tag}: wake never arrived"
+        );
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    }
+}
+
+/// Drain-and-filter: the next wake on `channel` (skipping any
+/// reconnect-wakes interleaved), bounded.
+async fn recv_channel_wake(
+    receiver: &mut dyn WakeReceiver,
+    channel: &str,
+    tag: &str,
+) -> alkstore::Wake {
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
+    loop {
+        let wake = must_recv(receiver, tag).await;
+        if wake.channel == channel {
+            return wake;
+        }
+        assert_eq!(
+            wake.channel,
+            alkstore::RESERVED_LISTENER_RECONNECTED,
+            "{tag}: unexpected wake channel {wake:?}"
+        );
+        assert!(
+            tokio::time::Instant::now() < deadline,
+            "{tag}: the channel's wake never arrived"
+        );
+    }
+}
+
+async fn notify_admin(admin: &tokio_postgres::Client, channel: &str, payload: &str) {
+    admin
+        .query_one("SELECT pg_notify($1, $2)", &[&channel, &payload])
+        .await
+        .unwrap();
+}
+
+/// Kill this store's listener backend (the POC's
+/// `pg_terminate_backend` shape, targeted at the store's unique
+/// application_name).
+async fn kill_listener_backend(admin: &tokio_postgres::Client, app_name: &str) -> bool {
+    admin
+        .query_one(
+            "SELECT pg_terminate_backend(pid) FROM pg_stat_activity
+             WHERE application_name = $1 AND pid <> pg_backend_pid()",
+            &[&app_name],
+        )
+        .await
+        .unwrap()
+        .get(0)
+}
+
+/// Acceptance: `notify` delivers commit-atomic auto-commit notifies —
+/// heard through `listen` on a second store connection (the
+/// cross-connection wake-arrives pin; `notify`'s `pg_notify` fires at
+/// its implicit transaction's commit), payload-less wakes only
+/// (`Wake { channel }`).
+#[tokio::test(flavor = "multi_thread")]
+async fn notify_delivers_commit_atomic_wake_to_listener() {
+    let Some(dsn) = harness_dsn() else {
+        eprintln!("skip: no harness server");
+        return;
+    };
+    let schema = instance_namer("notify")();
+    let store = open_store(&dsn, test_opts(&schema)).await.unwrap();
+
+    let channel = instance_namer("ch")();
+    let mut receiver = store.listen(&channel).await.unwrap();
+
+    store.notify(&channel, json!({"order": 7})).await.unwrap();
+
+    let wake = recv_channel_wake(receiver.as_mut(), &channel, "wake").await;
+    assert_eq!(wake.channel, channel, "the wake carries the channel only");
+    assert!(
+        receiver
+            .recv_timeout(Duration::from_millis(50))
+            .await
+            .unwrap()
+            .is_none(),
+        "per-notify delivery: exactly one wake per committed NOTIFY"
+    );
+
+    // A second listener on the same channel hears it too (N
+    // subscribers per channel — pinned again below at 4 via admin
+    // NOTIFY; this one pins the store-notify path delivers to every
+    // subscriber).
+    let mut second = store.listen(&channel).await.unwrap();
+    store.notify(&channel, json!("again")).await.unwrap();
+    recv_channel_wake(receiver.as_mut(), &channel, "second-first").await;
+    recv_channel_wake(second.as_mut(), &channel, "second-second").await;
+
+    store.close();
+    drop(store);
+    let admin = harness_client().await.unwrap();
+    drop_schema(&admin, &schema).await;
+}
+
+/// Acceptance: the 8000-byte boundary on the auto-commit path — typed
+/// `PayloadTooLarge { limit: 8000 }` client-side before any round
+/// trip, through the one limit constant `notify_tx` checks.
+#[tokio::test(flavor = "multi_thread")]
+async fn notify_payload_boundary_is_client_side_typed() {
+    let Some(dsn) = harness_dsn() else {
+        eprintln!("skip: no harness server");
+        return;
+    };
+    let schema = instance_namer("notify-boundary")();
+    let store = open_store(&dsn, test_opts(&schema)).await.unwrap();
+
+    let channel = instance_namer("ch")();
+    let mut receiver = store.listen(&channel).await.unwrap();
+
+    // The largest crossing payload: a 7997-char string serializes to
+    // 7999 bytes (≤ the NUL-inclusive budget) — delivers.
+    let under = json!("b".repeat(7997));
+    store.notify(&channel, under).await.unwrap();
+    let wake = recv_channel_wake(receiver.as_mut(), &channel, "under").await;
+    assert_eq!(wake.channel, channel);
+
+    // 8000-byte serialization → the typed arm, client-side.
+    let at = json!("a".repeat(7998)); // exactly 8000 bytes
+    let err = store.notify(&channel, at).await.unwrap_err();
+    match err {
+        Error::PayloadTooLarge { limit } => assert_eq!(limit, NOTIFY_PAYLOAD_LIMIT),
+        other => panic!("expect PayloadTooLarge, got {other:?}"),
+    }
+    let over = json!("a".repeat(8000));
+    let err = store.notify(&channel, over).await.unwrap_err();
+    assert!(
+        matches!(err, Error::PayloadTooLarge { limit: 8000 }),
+        "the typed arm at any over-size, got {err:?}"
+    );
+
+    // Client-side before any round trip: nothing delivered while the
+    // rejects fired (no server round trip happened — the listener
+    // hears only the under-boundary notify).
+    assert!(
+        receiver
+            .recv_timeout(Duration::from_millis(150))
+            .await
+            .unwrap()
+            .is_none(),
+        "rejected notifies deliver nothing"
+    );
+
+    // The same check on the tx twin rides the same constant: the
+    // typed arm there is pinned by the tx tests' boundary test; here
+    // we pin `notify_tx` and `notify` reject at the identical size
+    // (one owner of the limit — a drift would fail this).
+    let mut tx = store.begin_tx().await.unwrap();
+    let err = tx
+        .notify_tx(&channel, json!("a".repeat(7998)))
+        .await
+        .unwrap_err();
+    assert!(
+        matches!(err, Error::PayloadTooLarge { limit: 8000 }),
+        "notify_tx and notify reject at the identical boundary"
+    );
+    tx.commit().await.unwrap();
+
+    store.close();
+    drop(store);
+    let admin = harness_client().await.unwrap();
+    drop_schema(&admin, &schema).await;
+}
+
+/// Acceptance: channel-scoped fanout, multi-channel on one listener
+/// connection, N subscribers per channel, wake content
+/// `Wake { channel }` only.
+#[tokio::test(flavor = "multi_thread")]
+async fn fanout_is_channel_scoped_multichannel_and_multi_subscriber() {
+    let Some(dsn) = harness_dsn() else {
+        eprintln!("skip: no harness server");
+        return;
+    };
+    let schema = instance_namer("fanout")();
+    let store = open_store(&dsn, test_opts(&schema)).await.unwrap();
+    let admin = harness_client().await.unwrap();
+
+    let ch_a = instance_namer("ch_a")();
+    let ch_b = instance_namer("ch_b")();
+
+    // One listener connection, two channels, two subscribers on a.
+    let mut a1 = store.listen(&ch_a).await.unwrap();
+    let mut a2 = store.listen(&ch_a).await.unwrap();
+    let mut b1 = store.listen(&ch_b).await.unwrap();
+
+    notify_admin(&admin, &ch_a, "for-a").await;
+    recv_channel_wake(a1.as_mut(), &ch_a, "a1").await;
+    recv_channel_wake(a2.as_mut(), &ch_a, "a2").await;
+
+    // B's subscriber never heard a's notify: its next wake is b's —
+    // the strict filter (a wake for b arrives, a's doesn't).
+    notify_admin(&admin, &ch_b, "for-b").await;
+    let wake = recv_channel_wake(b1.as_mut(), &ch_b, "b1").await;
+    assert_eq!(wake.channel, ch_b);
+
+    // All 4 subscribers on one channel hear one notify (multi-
+    // subscriber fanout count — the SQLite twin's shape, pg-side
+    // per-notify: four wakes, one per subscriber, one per notify).
+    let mut subs: Vec<Box<dyn WakeReceiver>> = Vec::new();
+    for _ in 0..4 {
+        subs.push(store.listen(&ch_a).await.unwrap());
+    }
+    notify_admin(&admin, &ch_a, "fan-four").await;
+    for (i, sub) in subs.iter_mut().enumerate() {
+        let wake = recv_channel_wake(sub.as_mut(), &ch_a, &format!("fan{i}")).await;
+        assert_eq!(wake.channel, ch_a);
+    }
+
+    store.close();
+    drop(store);
+    drop_schema(&admin, &schema).await;
+}
+
+/// Acceptance: channel names are quoted identifiers end-to-end —
+/// arbitrary legal names (colons, dots, embedded quotes, spaces)
+/// LISTEN/NOTIFY/UNLISTEN correctly.
+#[tokio::test(flavor = "multi_thread")]
+async fn arbitrary_channel_names_deliver_via_quoting() {
+    let Some(dsn) = harness_dsn() else {
+        eprintln!("skip: no harness server");
+        return;
+    };
+    let schema = instance_namer("quoting")();
+    let store = open_store(&dsn, test_opts(&schema)).await.unwrap();
+    let admin = harness_client().await.unwrap();
+
+    for name in ["orders:created", "ns.sub.events", "has space", "has\"quote"] {
+        let channel = instance_namer("ch")();
+        let channel = format!("{channel}::{name}");
+        let mut receiver = store.listen(&channel).await.unwrap();
+        store.notify(&channel, json!("q")).await.unwrap();
+        let wake = recv_channel_wake(receiver.as_mut(), &channel, "quoted").await;
+        assert_eq!(wake.channel, channel, "the quoted name round-trips");
+        drop(receiver);
+    }
+
+    store.close();
+    drop(store);
+    drop_schema(&admin, &schema).await;
+}
+
+/// Acceptance: receiver close semantics (ADR-021 §5's pg arm) —
+/// the receiver stays open across forwarder reconnects (a backend
+/// kill is invisible: the reconnect-wake arrives, then
+/// post-reconnect deliveries work), and closes only at engine
+/// shutdown (terminal `None`; `Closed` on the bounded arms).
+#[tokio::test(flavor = "multi_thread")]
+async fn receiver_stays_open_across_reconnects_closes_at_shutdown() {
+    let Some(dsn) = harness_dsn() else {
+        eprintln!("skip: no harness server");
+        return;
+    };
+    let schema = instance_namer("close-sem")();
+    let store = open_store(&dsn, test_opts(&schema)).await.unwrap();
+    let admin = harness_client().await.unwrap();
+
+    let channel = instance_namer("ch")();
+    let mut receiver = store.listen(&channel).await.unwrap();
+
+    // Pre-kill delivery (the subscription is verified).
+    store.notify(&channel, json!("pre")).await.unwrap();
+    recv_channel_wake(receiver.as_mut(), &channel, "pre").await;
+
+    // Kill the listener's backend: the receiver does NOT close —
+    // the reconnect-wake arrives (the forwarder reconnected),
+    // followed by working post-reconnect deliveries.
+    assert!(
+        kill_listener_backend(&admin, store.listener_application_name()).await,
+        "the listener backend was found and killed"
+    );
+    let wake = must_recv(receiver.as_mut(), "reconnect-wake").await;
+    assert_eq!(
+        wake.channel,
+        alkstore::RESERVED_LISTENER_RECONNECTED,
+        "the synthetic reconnect-wake arrives across the gap"
+    );
+    store.notify(&channel, json!("post")).await.unwrap();
+    let wake = recv_channel_wake(receiver.as_mut(), &channel, "post").await;
+    assert_eq!(wake.channel, channel, "post-reconnect deliveries work");
+
+    // Engine shutdown → terminal None (the broadcast sender drops).
+    store.close();
+    let wake = receiver.recv().await;
+    assert!(wake.is_none(), "shutdown closes the receiver terminal");
+    assert!(
+        matches!(
+            receiver.recv_timeout(Duration::from_millis(50)).await,
+            Err(Error::Closed)
+        ),
+        "recv_timeout at shutdown carries Closed, not Ok(None)"
+    );
+    assert!(matches!(receiver.try_recv(), Err(Error::Closed)));
+
+    drop(store);
+    drop_schema(&admin, &schema).await;
+}
+
+/// Acceptance: the no-replay hole's honest communication — a commit
+/// during a connection gap is never re-delivered on the receiver
+/// (`!saw_replay` posture): the listener hears only the
+/// reconnect-wake after the gap, then post-reconnect activity.
+/// Recovery is the reconnect-wake + consumer re-read; the wake stream
+/// is best-effort, never a replay.
+#[tokio::test(flavor = "multi_thread")]
+async fn commits_during_a_gap_are_never_replayed() {
+    let Some(dsn) = harness_dsn() else {
+        eprintln!("skip: no harness server");
+        return;
+    };
+    let schema = instance_namer("no-replay")();
+    let store = open_store(&dsn, test_opts(&schema)).await.unwrap();
+    let admin = harness_client().await.unwrap();
+
+    let channel = instance_namer("ch")();
+    let mut receiver = store.listen(&channel).await.unwrap();
+
+    // Pre-gap delivery verified.
+    store.notify(&channel, json!("pre-gap")).await.unwrap();
+    recv_channel_wake(receiver.as_mut(), &channel, "pre-gap").await;
+
+    // Kill the listener backend; the notify fired while the session
+    // is down (during the gap — the server buffers nothing for the
+    // dead session).
+    assert!(kill_listener_backend(&admin, store.listener_application_name()).await);
+    let reconnect_wake = must_recv(receiver.as_mut(), "reconnect-wake").await;
+    assert_eq!(
+        reconnect_wake.channel,
+        alkstore::RESERVED_LISTENER_RECONNECTED
+    );
+
+    // The gap's delivery: the only wake seen after the kill is the
+    // reconnect-wake — the gap-commit was not replayed (NOTIFY is
+    // per-session: a dead session's notification is simply gone).
+    assert!(
+        receiver
+            .recv_timeout(Duration::from_millis(150))
+            .await
+            .unwrap()
+            .is_none(),
+        "no replay: the gap commit never re-delivered"
+    );
+
+    store.notify(&channel, json!("post-gap")).await.unwrap();
+    let wake = recv_channel_wake(receiver.as_mut(), &channel, "post-gap").await;
+    assert_eq!(wake.channel, channel, "post-reconnect delivery works");
+
+    store.close();
+    drop(store);
+    drop_schema(&admin, &schema).await;
+}
+
+/// Acceptance: receiver drop unregisters the channel — UNLISTEN at
+/// the last-subscriber drop, refcounted per subscriber. Probed
+/// behaviorally (pg_listening_channels is a per-session function —
+/// no cross-session view exists): while subscribers remain, an
+/// admin-NOTIFY delivers to a remaining subscriber; after the last
+/// drop, deliveries to this store's listener session cease (the raw
+/// forwarder fanout — which sees every channel this session LISTENs —
+/// carries no wake for the dropped channel across repeated NOTIFYs).
+#[tokio::test(flavor = "multi_thread")]
+async fn receiver_drop_unregisters_at_last_subscriber() {
+    let Some(dsn) = harness_dsn() else {
+        eprintln!("skip: no harness server");
+        return;
+    };
+    let schema = instance_namer("unlisten")();
+    let store = open_store(&dsn, test_opts(&schema)).await.unwrap();
+    let admin = harness_client().await.unwrap();
+
+    let channel = instance_namer("ch")();
+    let channel2 = instance_namer("other")();
+    let mut r1 = store.listen(&channel).await.unwrap();
+    let mut r2 = store.listen(&channel).await.unwrap();
+    let mut keep = store.listen(&channel2).await.unwrap();
+
+    // Registered state: admin NOTIFY delivers to both subscribers.
+    notify_admin(&admin, &channel, "registered").await;
+    recv_channel_wake(r1.as_mut(), &channel, "r1-registered").await;
+    recv_channel_wake(r2.as_mut(), &channel, "r2-registered").await;
+
+    // First subscriber drop: refcount 2 → 1 — the LISTEN stays:
+    // delivery to the remaining subscriber keeps working.
+    drop(r1);
+    tokio::time::sleep(Duration::from_millis(100)).await;
+    notify_admin(&admin, &channel, "still-listening").await;
+    recv_channel_wake(r2.as_mut(), &channel, "r2-still-listening").await;
+
+    // Last subscriber drop: UNLISTEN — the raw fanout (every channel
+    // this listener session is registered for delivers through it)
+    // carries no wake for the dropped channel across repeated NOTIFYs
+    // (retried until the async UNLISTEN command lands; a wake arriving
+    // at all means the LISTEN was still live).
+    let mut raw_rx = store.forwarder().subscribe().unwrap();
+    drop(r2);
+    let mut unlistened = false;
+    for _ in 0..25 {
+        notify_admin(&admin, &channel, "probe").await;
+        let heard = raw_fanout_hears(&mut raw_rx, &channel, Duration::from_millis(120)).await;
+        if !heard {
+            unlistened = true;
+            break;
+        }
+    }
+    assert!(
+        unlistened,
+        "the last-subscriber drop must UNLISTEN the channel"
+    );
+
+    // The untouched channel stayed registered throughout.
+    notify_admin(&admin, &channel2, "untouched").await;
+    recv_channel_wake(keep.as_mut(), &channel2, "keep").await;
+
+    store.close();
+    drop(store);
+    drop_schema(&admin, &schema).await;
+}
+
+/// Whether the raw fanout hears `channel` within `timeout` (the
+/// listener session's LISTEN registry is live for that channel).
+async fn raw_fanout_hears(
+    rx: &mut tokio::sync::broadcast::Receiver<crate::forwarder::RawNotification>,
+    channel: &str,
+    timeout: Duration,
+) -> bool {
+    let deadline = tokio::time::Instant::now() + timeout;
+    loop {
+        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
+        if remaining.is_zero() {
+            return false;
+        }
+        match tokio::time::timeout(remaining, rx.recv()).await {
+            Ok(Ok(n)) if n.channel == channel => return true,
+            Ok(Ok(_)) => continue,
+            Ok(Err(tokio::sync::broadcast::error::RecvError::Lagged(_))) => continue,
+            Ok(Err(tokio::sync::broadcast::error::RecvError::Closed)) => return false,
+            Err(_) => return false,
+        }
+    }
+}
+
+/// Acceptance: entry-point validation on both paths — empty/
+/// whitespace-only → `InvalidName`, reserved-prefix → `ReservedName`,
+/// identically on `notify` and `listen`, before any engine round
+/// trip (a rejected listen registers nothing server-side).
+#[tokio::test(flavor = "multi_thread")]
+async fn entry_points_validate_names() {
+    let Some(dsn) = harness_dsn() else {
+        eprintln!("skip: no harness server");
+        return;
+    };
+    let schema = instance_namer("validation")();
+    let store = open_store(&dsn, test_opts(&schema)).await.unwrap();
+    let admin = harness_client().await.unwrap();
+
+    let check = |err: Error, expect: &str| match err {
+        Error::InvalidName { .. } => assert_eq!(expect, "empty"),
+        Error::ReservedName { .. } => assert_eq!(expect, "reserved"),
+        other => panic!("{expect} case got the wrong error: {other}"),
+    };
+
+    check(store.notify("", json!({})).await.unwrap_err(), "empty");
+    check(store.notify("  ", json!({})).await.unwrap_err(), "empty");
+    check(
+        store
+            .notify("__alkstore_listener_reconnected__", json!({}))
+            .await
+            .unwrap_err(),
+        "reserved",
+    );
+    check(
+        store
+            .notify("__alkstore_outbox:x", json!({}))
+            .await
+            .unwrap_err(),
+        "reserved",
+    );
+    let err = match store.listen("").await {
+        Err(e) => e,
+        Ok(_) => panic!("empty channel must be rejected"),
+    };
+    check(err, "empty");
+    let err = match store.listen("  ").await {
+        Err(e) => e,
+        Ok(_) => panic!("whitespace channel must be rejected"),
+    };
+    check(err, "empty");
+    let err = match store.listen("__alkstore_x").await {
+        Err(e) => e,
+        Ok(_) => panic!("reserved channel must be rejected"),
+    };
+    check(err, "reserved");
+
+    // The rejected listens registered nothing (the validation fired
+    // before any engine round trip): the raw fanout — which sees
+    // every channel this listener session LISTENs — carries no wake
+    // for probed channels across repeated admin NOTIFYs.
+    let mut raw_rx = store.forwarder().subscribe().unwrap();
+    for probed in ["clean_probe_a", "clean_probe_b"] {
+        let mut heard = false;
+        for _ in 0..5 {
+            notify_admin(&admin, probed, "probe").await;
+            if raw_fanout_hears(&mut raw_rx, probed, Duration::from_millis(80)).await {
+                heard = true;
+                break;
+            }
+        }
+        assert!(
+            !heard,
+            "a rejected listen registered nothing server-side ({probed} heard)"
+        );
+    }
+
+    store.close();
+    drop(store);
+    drop_schema(&admin, &schema).await;
+}
+
+/// Acceptance: closed-store fail-closed on both paths — `notify` and
+/// `listen` past `close()` yield `Database` (the engine-wide
+/// closed-store posture).
+#[tokio::test(flavor = "multi_thread")]
+async fn closed_store_fails_closed_on_both_paths() {
+    let Some(dsn) = harness_dsn() else {
+        eprintln!("skip: no harness server");
+        return;
+    };
+    let schema = instance_namer("closed")();
+    let store = open_store(&dsn, test_opts(&schema)).await.unwrap();
+    let admin = harness_client().await.unwrap();
+
+    store.close();
+
+    let err = store.notify("ch", json!({})).await.unwrap_err();
+    assert!(
+        matches!(err, Error::Database(_)),
+        "notify fails closed, got {err:?}"
+    );
+    let err = match store.listen("ch").await {
+        Err(e) => e,
+        Ok(_) => panic!("listen must fail closed on a closed store"),
+    };
+    assert!(matches!(err, Error::Database(_)));
+
+    drop(store);
+    drop_schema(&admin, &schema).await;
+}
+
+/// Acceptance, pitfall 1 re-pinned through the full receiver stack:
+/// the poll loop runs before the first client query — a `listen`
+/// issued immediately after `open` (which itself rides the listener
+/// connection's LISTEN) works: a query issued immediately after the
+/// listener spawn completes.
+#[tokio::test(flavor = "multi_thread")]
+async fn pitfall_1_listen_immediately_after_spawn_delivers() {
+    let Some(dsn) = harness_dsn() else {
+        eprintln!("skip: no harness server");
+        return;
+    };
+    let schema = instance_namer("pit1")();
+    let store = open_store(&dsn, test_opts(&schema)).await.unwrap();
+    let admin = harness_client().await.unwrap();
+
+    // listen immediately after open (its LISTEN is the first client
+    // query racing the poll loop), then notify immediately after.
+    let channel = instance_namer("ch")();
+    let mut receiver = store.listen(&channel).await.unwrap();
+    store.notify(&channel, json!("instant")).await.unwrap();
+    let wake = recv_channel_wake(receiver.as_mut(), &channel, "instant").await;
+    assert_eq!(
+        wake.channel, channel,
+        "no query-vs-poll starvation deadlock"
+    );
+
+    store.close();
+    drop(store);
+    drop_schema(&admin, &schema).await;
+}
+
+/// Acceptance, pitfall 2 re-pinned through the full receiver stack:
+/// the listener's `Client` stays alive for the listener's lifetime —
+/// a long-lived listener survives unrelated client churn (pool
+/// checkouts and admin connections cycling around it).
+#[tokio::test(flavor = "multi_thread")]
+async fn pitfall_2_long_lived_listener_survives_client_churn() {
+    let Some(dsn) = harness_dsn() else {
+        eprintln!("skip: no harness server");
+        return;
+    };
+    let schema = instance_namer("pit2")();
+    let store = open_store(&dsn, test_opts(&schema)).await.unwrap();
+    let admin = harness_client().await.unwrap();
+
+    let channel = instance_namer("ch")();
+    let mut receiver = store.listen(&channel).await.unwrap();
+
+    // Unrelated client churn: pool checkouts held and dropped, twice,
+    // while the listener session must stay alive.
+    for round in 0..2 {
+        let mut hold = Vec::new();
+        for _ in 0..3 {
+            hold.push(store.pool().get().await.unwrap());
+        }
+        store
+            .notify(&channel, json!(format!("churn-{round}")))
+            .await
+            .unwrap();
+        let wake = recv_channel_wake(receiver.as_mut(), &channel, "churn").await;
+        assert_eq!(wake.channel, channel, "the listener session survived churn");
+        drop(hold);
+    }
+
+    store.close();
+    drop(store);
+    drop_schema(&admin, &schema).await;
+}
+
+/// Unit pin (server-less): the forwarder's channel-set refcount —
+/// add/remove subscribers and the last-subscriber rule, and the
+/// snapshot the reconnect re-issues from.
+#[test]
+fn channel_set_refcounts_last_subscriber_rule() {
+    let set = crate::forwarder::ChannelSet::default();
+    assert!(set.add_subscriber("a"), "first subscriber inserts");
+    assert!(
+        !set.add_subscriber("a"),
+        "second subscriber doesn't reinsert"
+    );
+    assert!(set.add_subscriber("b"));
+    assert_eq!(set.snapshot(), vec!["a".to_string(), "b".to_string()]);
+
+    assert!(!set.remove_subscriber("a"), "first drop keeps the entry");
+    assert!(set.remove_subscriber("a"), "last drop removes it");
+    assert!(!set.remove_subscriber("a"), "drop past zero is a no-op");
+    assert_eq!(set.snapshot(), vec!["b".to_string()]);
+}

alkstore-postgres/src/store/open_tests.rs

@@ -177,8 +177,8 @@ async fn open_boots_pool_bootstrap_and_forwarder() {
     // delivers through the fanout (the minimal fanout proof — the
     // notify-listen task owns the receiver bridging).
     let channel = format!("forwarder_probe_{}", instance_namer("ch")());
-    store.forwarder().register(&channel).unwrap();
-    let mut rx = store.forwarder().subscribe();
+    store.forwarder().register(&channel).await.unwrap();
+    let mut rx = store.forwarder().subscribe().unwrap();
 
     let admin2 = harness_client().await.unwrap();
     let notify_sql = format!(
@@ -220,10 +220,10 @@ async fn forwarder_poll_loop_survives_command_churn() {
     let store = open_store(&dsn, test_opts(&schema)).await.unwrap();
 
     let channel = format!("churn_probe_{}", instance_namer("ch")());
-    let mut rx = store.forwarder().subscribe();
+    let mut rx = store.forwarder().subscribe().unwrap();
     // Pitfall 1: LISTEN issued immediately after open — if the poll
     // loop were not already running this would deadlock.
-    store.forwarder().register(&channel).unwrap();
+    store.forwarder().register(&channel).await.unwrap();
 
     let admin = harness_client().await.unwrap();
     for i in 0..5 {
@@ -263,8 +263,8 @@ async fn forwarder_reconnects_after_backend_kill_and_delivers() {
     let store = open_store(&dsn, test_opts(&schema)).await.unwrap();
 
     let channel = format!("reconn_probe_{}", instance_namer("ch")());
-    let mut rx = store.forwarder().subscribe();
-    store.forwarder().register(&channel).unwrap();
+    let mut rx = store.forwarder().subscribe().unwrap();
+    store.forwarder().register(&channel).await.unwrap();
 
     let admin = harness_client().await.unwrap();
     // Deliver once pre-kill (the subscription path is verified).
@@ -532,9 +532,12 @@ async fn close_and_drop_teardown_cleanly() {
     let after = listener_sessions(&admin, &app_name).await;
     assert_eq!(after, 0, "close terminated this store's listener session");
 
-    // Post-close ops fail closed (Database, not panic).
+    // Post-close ops fail closed (Database, not panic). A plain
+    // channel name: notify's closed-store check fires (validation
+    // passes for it; a reserved-prefixed channel would fail
+    // validation first — its own pin).
     let err = store
-        .notify("__alkstore_post_close", serde_json::json!({}))
+        .notify("post_close", serde_json::json!({}))
         .await
         .unwrap_err();
     assert!(
@@ -617,10 +620,11 @@ async fn open_fails_database_on_unparseable_and_unreachable() {
     }
 }
 
-/// Acceptance: the trait surface is the wave-3 stub posture — every
-/// `Store` method returns `Err(Database(… wiring lands with the …
-/// task))`; `with_tx` surfaces the `begin_tx` stub naturally; no
-/// panics.
+/// Acceptance: the trait surface — `begin_tx`, `notify`, and
+/// `listen` are wired (the seam + notify-listen tasks); the remaining
+/// methods are the wave-3 stub posture — every unwired `Store` method
+/// returns `Err(Database(… wiring lands with the … task))`; `with_tx`
+/// surfaces the wired `begin_tx` naturally; no panics.
 #[tokio::test(flavor = "multi_thread")]
 async fn store_trait_methods_are_wiring_stubs() {
     let Some(dsn) = harness_dsn() else {
@@ -639,16 +643,22 @@ async fn store_trait_methods_are_wiring_stubs() {
         .with_tx(Box::new(|_tx| Box::pin(async { Ok(()) })))
         .await;
     assert!(err.is_ok(), "with_tx surfaces the wired begin_tx naturally");
-    let err = store
-        .notify("ch", serde_json::json!("x"))
+    // `notify`/`listen` are wired (the notify-listen task's): a
+    // valid-channel listen returns a working receiver, `notify`
+    // returns `Ok` — full behavior the notify tests' scope.
+    let mut receiver = store.listen("stub_ch_probe").await.unwrap();
+    assert!(
+        receiver
+            .recv_timeout(Duration::from_millis(50))
             .await
-        .unwrap_err();
-    assert!(matches!(err, alkstore::Error::Database(_)));
-    let err = match store.listen("ch").await {
-        Err(e) => e,
-        Ok(_) => panic!("listen must be a stub"),
-    };
-    assert!(matches!(err, alkstore::Error::Database(_)));
+            .unwrap()
+            .is_none(),
+        "an idle wired receiver idles (Ok(None))"
+    );
+    store
+        .notify("stub_ch_probe", serde_json::json!("x"))
+        .await
+        .unwrap();
     let err = match store.stream("s").await {
         Err(e) => e,
         Ok(_) => panic!("stream must be a stub"),
@@ -679,13 +689,11 @@ async fn store_trait_methods_are_wiring_stubs() {
         .unwrap_err();
     assert!(matches!(err, alkstore::Error::Database(_)));
 
-    // The stub messages name their landing task (the wave-3 posture's
-    // "… wiring lands with the … task" shape) — in the source chain
-    // (`Database`'s Display is opaque; ADR-008 §5's detail-carriage).
-    let err = store
-        .notify("ch", serde_json::json!("x"))
-        .await
-        .unwrap_err();
+    // The remaining stub messages name their landing task (the
+    // wave-3 posture's "… wiring lands with the … task" shape) — in
+    // the source chain (`Database`'s Display is opaque; ADR-008 §5's
+    // detail-carriage). `stream` sampled as the representative.
+    let err = stub_err!(store.stream("ch").await);
     match err {
         alkstore::Error::Database(source) => {
             let msg = source.to_string();
@@ -694,7 +702,7 @@ async fn store_trait_methods_are_wiring_stubs() {
                 "stub errors name their landing task, got: {msg}"
             );
         }
-        other => panic!("notify stub must be Database, got {other:?}"),
+        other => panic!("stream stub must be Database, got {other:?}"),
     }
 
     drop(store);

alkstore-postgres/src/store/tx_tests.rs

@@ -505,8 +505,8 @@ async fn rollback_ghosts_nothing_across_all_writes() {
     // notified ONLY by rolled-back writes in this test: the only
     // delivery that could arrive is a ghost.
     let rollback_channel = instance_namer("rollback_ch")();
-    store.forwarder().register(&rollback_channel).unwrap();
-    let mut rollback_rx = store.forwarder().subscribe();
+    store.forwarder().register(&rollback_channel).await.unwrap();
+    let mut rollback_rx = store.forwarder().subscribe().unwrap();
 
     let mut tx = store.begin_tx().await.unwrap();
     let ghost_id = tx
@@ -568,8 +568,8 @@ async fn rollback_ghosts_nothing_across_all_writes() {
     // The committed-side positive control on its OWN channel: a
     // committed notify_tx delivers at commit.
     let commit_channel = instance_namer("commit_ch")();
-    store.forwarder().register(&commit_channel).unwrap();
-    let mut commit_rx = store.forwarder().subscribe();
+    store.forwarder().register(&commit_channel).await.unwrap();
+    let mut commit_rx = store.forwarder().subscribe().unwrap();
     let mut tx = store.begin_tx().await.unwrap();
     tx.notify_tx(&commit_channel, serde_json::json!({}))
         .await

alkstore-postgres/src/tx.rs

@@ -83,6 +83,8 @@ use crate::seam::{pg_error, pool_error};
 /// The notification payload boundary (the contract's pinned limit —
 /// client-side checked, typed before any round trip; the predicate
 /// rejects at `len >= limit`, the server's NUL-inclusive wire budget).
+/// One owner of the limit: both `notify_tx` (below) and the
+/// auto-commit `notify` (the notify task's) check through it.
 pub(crate) const NOTIFY_PAYLOAD_LIMIT: usize = 8000;
 
 /// The caller-held pg transaction handle: the pooled-object handle.
@@ -240,7 +242,7 @@ impl PgTxHandle {
 /// (the serde_json serialization — ADR-020 §4). The notify carriage
 /// reuses it: the JSON serialization is valid UTF-8 text by
 /// construction (ADR-016 §5).
-fn encode_payload_bytes(payload: &Value) -> Result<Vec<u8>> {
+pub(crate) fn encode_payload_bytes(payload: &Value) -> Result<Vec<u8>> {
     alkstore::encode_payload(payload)
 }
 

tasks/pg-engine-notify-listen.md

@@ -1,7 +1,7 @@
 ---
 id: pg-engine-notify-listen
 name: Postgres engine — LISTEN forwarder (full behavior) + notify/listen surface
-status: pending
+status: completed
 depends_on: [pg-engine-seam-tx]
 scope: moderate
 risk: high
@@ -132,8 +132,96 @@ hand-rolled posture's owned failure modes, ADR-004):
 
 ## Notes
 
-> To be filled by implementation agent
+> Decisions of record the implementation made that the description
+> didn't pin:
+
+- **Registration is synchronous end-to-end** (beyond the skeleton's
+  fire-and-forget commands): `Forwarder::register` now carries an
+  ack — the forwarder loop responds after the server processed the
+  LISTEN, and a failed LISTEN (connection died mid-command) acks the
+  failure to the caller. Rationale: the trait pins `listen` as
+  **starting from now** (no replay) — a listen that returned must
+  see every later commit; the skeleton's async command queue left a
+  race where a notify issued right after `listen()` could beat the
+  server-side registration and be silently lost (observed as a real
+  test failure, not hypothetical). A mid-LISTEN failure still
+  recovers via the registry re-issue (the registry entry is written
+  before the command, unchanged).
+- **`Forwarder` shutdown takes the fanout sender** (the struct's
+  sender became `Mutex<Option<Sender>>`; `subscribe` returns
+  `Option` → the notify path fails closed on a taken sender).
+  Rationale: receivers hold `Arc<Forwarder>` (for drop-unregister),
+  so the struct's sender never drops on receiver lifetimes alone —
+  without the take, a `close()`d store's receivers hung forever on
+  `recv()` (observed). The shutdown take releases the broadcast;
+  the loop's clone dies with the loop's shutdown exit. The terminal
+  `None` arm is thus purely shutdown-driven, independent of who
+  holds which Arc — ADR-021 §5's pg arm as pinned.
+- **`ChannelSet` refcounts per subscriber** (`BTreeSet` →
+  `BTreeMap<String, u64>`; `insert`/`remove` →
+  `add_subscriber`/`remove_subscriber` with last-subscriber
+  detection) — the description's refcount rule, made concrete on
+  the registry itself; snapshot/re-issue logic unchanged.
+- **Server-side LISTEN-registry probes are behavioral, not
+  catalog**: `pg_listening_channels()` is a per-session function —
+  no cross-session view of another session's LISTEN set exists, so
+  the drop-unregisters and validation tests probe through the raw
+  forwarder fanout (which sees every channel the session LISTENs)
+  instead of a catalog query.
+- **`notify`'s closed-store check precedes payload work** (the
+  fail-closed dominates the 8000-byte check's ordering — a
+  post-close notify is `Database` whatever the payload).
+- **Forwarder-loop fan eprintln for lag**: `Lagged(n)` is surfaced
+  via stderr (the POC's bridge posture — log and keep going); no
+  structured logging dependency adopted for this task.
+- The skeleton test file renamed/restructured accordingly:
+  `open_tests.rs`'s stub-surface test now pins `notify`/`listen` as
+  wired (idle `Ok(None)`, `notify` success) alongside the remaining
+  stubs; its three forwarder tests gained `.await`/`.unwrap()` for
+  the synchronous register. The tx tests' two raw-fanout probes
+  updated for the `Option`-returning subscribe.
 
 ## Summary
 
-> To be filled on completion
+> What landed, verified how:
+
+- **`alkstore-postgres/src/notify.rs`** (new): the wake half —
+  `notify` (auto-commit: pool checkout → `SELECT pg_notify($1,$2)` →
+  return to pool; shared-namespace validation; the 8000-byte check
+  typed client-side through `tx::NOTIFY_PAYLOAD_LIMIT` — one limit
+  owner with `notify_tx` — before any round trip) and `listen`
+  (validation → closed-store fail-closed → synchronous acked
+  registration → broadcast subscription → `PgWakeReceiver` bridging
+  `Wake { channel }` wakes, channel-scoped + the reserved
+  reconnect-wake, `Lagged(n)` logged-and-streamed).
+- **`PgWakeReceiver`**: `recv()` → per-notify wakes, terminal `None`
+  only at engine shutdown; `try_recv`/`recv_timeout` with `Closed`
+  distinguishable from idle (`recv_timeout` expiry = `Ok(None)`);
+  Drop unregisters (UNLISTEN at last-subscriber drop). Stays open
+  across forwarder reconnects — pinned by backend-kill tests.
+- **`forwarder.rs`** (completed from the skeleton): refcounted
+  channel registry, acked `register`, shutdown sender-take; reconnect
+  (50 ms → 2 s exponential, re-LISTEN from the snapshot, synthetic
+  reconnect-wake on `__alkstore_listener_reconnected__` to every
+  subscriber) was already the skeleton's — now exercised through the
+  full receiver stack.
+- **`store.rs`**: `Store::notify`/`Store::listen` wired (stubs land
+  only for streams/queue/outbox/lock/scheduler now).
+- **Tests** (`store/notify_tests.rs`, 11 new): commit-atomic wake-
+  arrives (incl. N-subscriber fanout), the 8000 boundary on both
+  entry points at 7999/8000/8002 (one-constant pin via `notify_tx`
+  at the identical boundary), channel-scoped multi-channel + 4-
+  subscriber fanout, arbitrary quoted channel names (colons, dots,
+  spaces, embedded quotes), receiver-stays-open-across-reconnect +
+  shutdown-terminal (`None`, `Closed` on the bounded arms), the
+  no-replay pin (a gap commit never re-delivered; only the
+  reconnect-wake, then post-gap activity), drop-unregisters-
+  behaviorally (deliveries cease after the last drop; the untouched
+  channel stays), entry-point validation on both paths (nothing
+  registers), closed-store fail-closed on both, the two deadlock
+  pitfalls re-pinned through the receiver stack, and the channel-set
+  refcount unit pin (server-less). Harness-server runs: 51 lib + 9
+  schema tests green, stable across `--test-threads` 1/4/8 repeats;
+  server-less: all skip cleanly, workspace `cargo build` / `cargo
+  test` / `cargo clippy --all-targets -- -D warnings` / `cargo fmt
+  --check` green.