Merge branch 'wt/review-002-con18-wss-sweep-exit'

docs/architecture/decisions/070-from-wss-consumer-adapter.md

@@ -116,8 +116,14 @@ layer is done with the import. Consequences, stated explicitly:
   surface as live for the process lifetime. A reconnecting assembly
   layer should tear down its whole registry and re-import, accepting
   the accumulated server-side sessions until the remote times them out.
-- A close/teardown handle (and with it, safe reconnect) is future work;
+- **Close/teardown handle**: future work; there is no
-  v1 deliberately does not build a reconnect layer (OQ-03).
+  `WssSession::close` in v1 (OQ-03). The connection-drop monitor is
+  however self-limiting (review-002 CON-18): once WS read EOF is
+  observed, pending calls fail with retryable `CONNECTION_CLOSED`,
+  registrations racing the drop are drained by a fast-fail sweep
+  (50 ms interval) during a bounded post-EOF grace window (8
+  consecutive empty drains), and the monitor task then ends — a dead
+  import leaves no permanent task behind.
 
 ## Consequences
 
@@ -129,6 +135,10 @@ layer is done with the import. Consequences, stated explicitly:
   construction (server upgrade path + consumer path share it).
 - Same-protocol import means zero translation: the remote node's ops
   appear in the local registry with their real specs and error schemas.
+- A dead import (peer gone) self-cleans: after EOF the drop monitor
+  fails everything pending, drains drop-racing registrations for a
+  bounded grace window, and ends — no per-dead-session task leak
+  scales with import count (review-002 CON-18).
 
 **Negative:**
 
@@ -138,6 +148,11 @@ layer is done with the import. Consequences, stated explicitly:
   keepalive/timeout tuning — deployment concern, but worth documenting.
 - Reconnection semantics are initially minimal (OQ-03); a consumer
   wanting hot re-registration must wait for or build the policy.
+- Registrations landing in the pending map after the bounded post-EOF
+  grace window elapses wait for the 30 s sweeper deadline instead of
+  failing fast — acceptable because registration-vs-EOF races resolve
+  well within the window in practice, and the mux write-failure path
+  covers the rest.
 
 ## References
 

src/adapters/from_wss.rs

@@ -36,14 +36,20 @@
 //! signal and fails all pending calls with retryable `CONNECTION_CLOSED`.
 //! The EOF signal is lossless (a retained watch value, WS-02): EOF is
 //! observed even if it fires before the monitor starts, or after the
-//! session was forgotten (fire-and-forget import). The monitor also
+//! session was forgotten (fire-and-forget import). Once EOF has been
-//! sweeps the pending map once a second while the session lives: calls
+//! observed, registrations that land in the pending map *after* the
-//! registered *after* the initial `fail_all` (still possible on the
+//! initial `fail_all` (still possible on the forgotten-session path,
-//! forgotten-session path, where imports may race the drop) are failed
+//! where imports may race the drop) are failed fast: the monitor
-//! on the next sweep tick once EOF has been observed. No hang: pendings
+//! watches the pending map at a short interval during a bounded
-//! resolve retryable regardless of registration-vs-EOF ordering
+//! post-EOF window and drains any entry that lands — no wait for the
-//! (CON-02). Subsequent handler calls fail on write; reconnect policy is
+//! next one-second sweep tick (CON-18). No hang: pendings resolve
-//! the assembly layer's job.
+//! retryable regardless of registration-vs-EOF ordering (CON-02), and
+//! when the map stays drained past the grace window the monitor task
+//! ends: no per-dead-session permanent task remains (the WS-02
+//! losslessness invariant holds — fast-fail is an additional
+//! resolution path, never a replacement for the retained watch
+//! signal). Subsequent handler calls fail on write; reconnect policy
+//! is the assembly layer's job.
 //!
 //! Session teardown (explicit limitation, review-001 CON-09): `import()`
 //! detaches the session fire-and-forget — there is no close/shutdown
@@ -69,10 +75,15 @@ use tokio_tungstenite::tungstenite::client::IntoClientRequest;
 
 use crate::websocket::split_tungstenite_to_bytes;
 
-/// How often the drop monitor sweeps the pending map for calls
+/// How often the drop monitor drains the pending map once EOF has been
-/// registered after the last `fail_all` (only effective once EOF has
+/// observed, failing `fail_all`-after registrations fast (CON-18).
-/// been observed — see the module doc).
+const PENDING_SWEEP_INTERVAL: std::time::Duration = std::time::Duration::from_millis(50);
-const PENDING_SWEEP_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1);
+
+/// The bounded post-EOF grace window, as a number of consecutive idle
+/// drains: when this many drains in a row find the pending map empty,
+/// the monitor ends — no per-dead-session task remains for the process
+/// lifetime (CON-18).
+const PENDING_SWEEP_MAX_POST_EOF: u32 = 8;
 
 fn connection_closed_error() -> CallError {
     CallError::new("CONNECTION_CLOSED", "from_wss connection dropped", true)
@@ -151,10 +162,31 @@ pub struct WssSession {
     _monitor: WssDropMonitor,
 }
 
+impl WssSession {
+    /// The drop monitor's task handle, for observability: tests assert
+    /// the monitor ends after EOF + the bounded grace window (CON-18).
+    /// The handle lives on the session; dropping the session detaches
+    /// the handle (tokio semantics) — the close signal already ends
+    /// the task on explicit drop, and the bounded sweep ends it on EOF.
+    #[cfg(test)]
+    fn monitor_handle(&mut self) -> &mut tokio::task::JoinHandle<()> {
+        &mut self._monitor.monitor_task
+    }
+}
+
 /// The drop monitor: fires `fail_all` on WS read EOF or on explicit
 /// session close (the session's `Drop` impl sends the close signal).
+/// When the session drops, the close signal resolves `fail_all` and
+/// the task ends; on read-EOF the bounded post-EOF sweep ends the task
+/// (CON-18). The task handle is retained on the session in test builds
+/// so the monitor's lifecycle stays observable; in non-test builds it
+/// is detached (tokio semantics) — the close signal already ends the
+/// task on explicit drop, and the bounded sweep ends it on EOF.
+#[cfg_attr(not(test), allow(dead_code))]
 struct WssDropMonitor {
     close_tx: Option<tokio::sync::oneshot::Sender<()>>,
+    #[cfg(test)]
+    monitor_task: tokio::task::JoinHandle<()>,
 }
 
 impl Drop for WssDropMonitor {
@@ -243,29 +275,36 @@ impl WssSession {
 
         let (close_tx, mut close_rx) = tokio::sync::oneshot::channel();
         let pending = Arc::clone(call_connection.pending());
-        tokio::spawn(async move {
+        #[cfg_attr(not(test), allow(unused_variables))]
+        let monitor_task = tokio::spawn(async move {
             let mut eof_rx = pumps.read_eof();
             let mut sweep = tokio::time::interval(PENDING_SWEEP_INTERVAL);
             sweep.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
-            let mut eof_observed = *eof_rx.borrow_and_update();
+            let eof_observed = *eof_rx.borrow_and_update();
             if eof_observed {
                 pending.lock().fail_all(connection_closed_error());
             }
+            let mut idle_sweeps_post_eof: u32 = 0;
             loop {
                 tokio::select! {
-                    changed = eof_rx.changed() => {
-                        eof_observed = changed.is_err() || *eof_rx.borrow_and_update();
-                        if eof_observed {
-                            pending.lock().fail_all(connection_closed_error());
-                        }
-                    }
                     _ = &mut close_rx => {
                         pending.lock().fail_all(connection_closed_error());
                         return;
                     }
                     _ = sweep.tick() => {
                         if *eof_rx.borrow() {
-                            pending.lock().fail_all(connection_closed_error());
+                            if pending
+                                .lock()
+                                .fail_all(connection_closed_error())
+                                .is_empty()
+                            {
+                                idle_sweeps_post_eof += 1;
+                            } else {
+                                idle_sweeps_post_eof = 0;
+                            }
+                            if idle_sweeps_post_eof >= PENDING_SWEEP_MAX_POST_EOF {
+                                return;
+                            }
                         }
                     }
                 }
@@ -277,6 +316,8 @@ impl WssSession {
             call_connection,
             _monitor: WssDropMonitor {
                 close_tx: Some(close_tx),
+                #[cfg(test)]
+                monitor_task,
             },
         })
     }
@@ -962,7 +1003,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn call_registered_after_eof_resolves_via_pending_sweep() {
+    async fn call_registered_after_eof_fails_fast_retryable() {
         let (endpoint, pumps_slot) = drop_on_signal_producer(slow_producer_registry());
         let session = WssSession::connect(&endpoint, Some("tok-1"), true)
             .await
@@ -983,20 +1024,41 @@ mod tests {
 
         std::mem::forget(session);
         abort_pumps_and_wait(&pumps_slot).await;
-        // Headroom past the monitor's 1 s sweep tick: the call
-        // registered below must be caught by the next sweep, not hang.
-        tokio::time::sleep(std::time::Duration::from_millis(1100)).await;
 
         let ctx = noop_context("req-late");
         let call_task = tokio::spawn(async move { handler(serde_json::json!({}), ctx).await });
 
-        let response = tokio::time::timeout(std::time::Duration::from_secs(5), call_task)
+        let response = tokio::time::timeout(std::time::Duration::from_millis(500), call_task)
             .await
-            .expect("post-EOF registered call resolves via sweep (no hang)")
+            .expect("post-EOF registered call fails fast (no 1s sweep wait, no hang)")
             .expect("join");
         match response.result {
-            Err(e) => assert!(e.retryable, "drop error must be retryable, got {e:?}"),
+            Err(e) => {
+                assert_eq!(e.code, "CONNECTION_CLOSED", "fast-fail code, got {e:?}");
+                assert!(e.retryable, "drop error must be retryable, got {e:?}");
+            }
             Ok(_) => panic!("expected Err after connection drop"),
         }
     }
+
+    #[tokio::test]
+    async fn drop_monitor_ends_after_eof_plus_grace_window() {
+        let (endpoint, pumps_slot) = drop_on_signal_producer(slow_producer_registry());
+        let mut session = WssSession::connect(&endpoint, Some("tok-1"), true)
+            .await
+            .expect("connect");
+
+        let config = FromCallConfig::new();
+        let bundles = import_from_call(&session.call_connection, config)
+            .await
+            .expect("import");
+        assert!(!bundles.is_empty());
+
+        abort_pumps_and_wait(&pumps_slot).await;
+
+        tokio::time::timeout(std::time::Duration::from_secs(5), session.monitor_handle())
+            .await
+            .expect("monitor ends after EOF + bounded grace window (no permanent task)")
+            .expect("monitor task was not aborted, no panic payload");
+    }
 }