diff --git a/alkstore-postgres/src/forwarder.rs b/alkstore-postgres/src/forwarder.rs index dbc1011..4a6f761 100644 --- a/alkstore-postgres/src/forwarder.rs +++ b/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 { @@ -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, + generation: &AtomicU64, + connected: &AtomicBool, + ) -> 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, + 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 { + 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( + bound: Duration, + shutdown: &mut watch::Receiver, + fut: F, +) -> BoundedOutcome +where + F: Future, +{ + 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 - /// 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). + /// The registry write and the derived LISTEN enqueue are **one + /// critical section** under the registry guard + /// ([`ChannelSet::enqueue_register`] — review 004 item 4's + /// 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); - 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 (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)")); - } + let ack_rx = self.channels.enqueue_register( + channel, + &self.commands_tx, + &self.generation, + &self.connected, + )?; 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 - /// 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). The queued UNLISTEN - /// rides its send-time generation: if the connection died before - /// processing it, the 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). + /// write and the derived UNLISTEN enqueue are **one critical + /// section** under the registry guard + /// ([`ChannelSet::enqueue_unregister`] — review 004 item 4's + /// TOCTOU fix, symmetric with [`Forwarder::register`]). The + /// UNLISTEN is best-effort — a lost one re-issues nothing + /// (UNLISTEN is not recovery-relevant: the snapshot won't carry + /// the removed channel). The queued UNLISTEN rides its send-time + /// generation: if the connection died before processing it, the + /// 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) { - let _ = self.commands_tx.send(ListenCommand::Unlisten { - generation: self.generation.load(Ordering::Acquire), - channel: channel.to_string(), - }); - } + self.channels + .enqueue_unregister(channel, &self.commands_tx, &self.generation); } /// 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 -/// wait for the connection's death → reconnect via -/// [`reconnect_with_retry`] (exponential backoff 50 ms → 2 s cap; -/// failed connects never leave the arm). The 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. +/// LISTENs from the live channel-set snapshot *bounded and +/// shutdown-selected* ([`bounded_wait`]/[`LISTENER_QUERY_TIMEOUT`]) → +/// on a failed verify, the bounded settle +/// ([`LISTEN_SETTLE_TIMEOUT`]) reaps-or-aborts the poll task → serve +/// commands (each also bounded, the shutdown watch selected with the +/// await) → reconnect via [`reconnect_with_retry`] (exponential +/// backoff 50 ms → 2 s cap; failed connects never leave the arm). The +/// 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::(); - let listen_ok = if listen_sql.is_empty() { - client.simple_query("SELECT 1").await.is_ok() - } else { - client.batch_execute(&listen_sql).await.is_ok() + let verify = async { + if listen_sql.is_empty() { + client.simple_query("SELECT 1").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; - } - if !first_listen { - // The synthetic reconnect-wake (ADR-008 §4's reserved - // channel; ADR-006's recovery semantics) broadcast to - // 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. - let _ = fanout.send(RawNotification { - channel: alkstore::RESERVED_LISTENER_RECONNECTED.to_string(), - }); - } - 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() => { + // The replay rides the listener-path bound, with the + // shutdown watch selected alongside — a stuck replay + // command cannot park the generation top forever, and + // a close surfaces mid-replay too. + match bounded_wait( + LISTENER_QUERY_TIMEOUT, + &mut shutdown, + run_command(&client, cmd), + ) + .await + { + BoundedOutcome::Done(()) => {} + BoundedOutcome::Shutdown => { 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. - 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 => { + BoundedOutcome::Overran => { + // An unresponded replay command: the + // connection is unusable — the honest death + // posture, straight to the reconnect (the + // reaper is aborted now; the next generation's + // re-issue is the recovery). + connected.store(false, Ordering::Release); + poll_task.abort(); + replay_dead = true; break; } } } - } else { - // A failed probe/LISTEN means the connection died: wait - // the poll task out (it owns the connection and is its - // reaper). - let _ = poll_task.await; + if !replay_dead { + if !was_first { + // The synthetic reconnect-wake (ADR-008 §4's reserved + // channel; ADR-006's recovery semantics) broadcast to + // 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 - // return — the guard and the client drop carry the rest. + // reaped or aborted (every run_command path above — the + // 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 diff --git a/alkstore-postgres/src/store/notify_tests.rs b/alkstore-postgres/src/store/notify_tests.rs index 4de56aa..95bca86 100644 --- a/alkstore-postgres/src/store/notify_tests.rs +++ b/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, +) -> 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 -/// 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")); +/// through the decide-and-queue registry API (review 004 item 4): the +/// last-subscriber rule (an UNLISTEN is derived exactly when the last +/// reference drops, and only then), drop-past-zero as a no-op, and the +/// snapshot the reconnect re-issues from. Each registration's LISTEN +/// is queued (the redundant per-registration LISTEN is the ack's +/// vehicle) — asserted too. +#[tokio::test] +async fn channel_set_refcounts_last_subscriber_rule() { + use crate::forwarder::{ChannelSet, GenerationSlot, ListenCommand}; + use std::sync::Arc; + use std::sync::atomic::AtomicBool; + + let set = ChannelSet::default(); + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::(); + 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"); - assert!(set.remove_subscriber("a"), "last drop removes it"); - assert!(!set.remove_subscriber("a"), "drop past zero is a no-op"); + // The first drop of "a" keeps the entry (an UNLISTEN must NOT be + // derived — the refcount still holds). + 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" + ); } diff --git a/alkstore-postgres/src/store/open_tests.rs b/alkstore-postgres/src/store/open_tests.rs index f89c535..7005cf0 100644 --- a/alkstore-postgres/src/store/open_tests.rs +++ b/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 { )) } +/// 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::(); + 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 = 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::(), + ) + .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, + eaten: Arc, +} + +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 { + 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; +} diff --git a/tasks/pg-fix-forwarder-hardening.md b/tasks/pg-fix-forwarder-hardening.md index d548d63..db9f890 100644 --- a/tasks/pg-fix-forwarder-hardening.md +++ b/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 \ No newline at end of file +> 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. \ No newline at end of file