diff --git a/src/adapters/from_wss.rs b/src/adapters/from_wss.rs index 8a3d68f..ac25b45 100644 --- a/src/adapters/from_wss.rs +++ b/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 -//! sweeps the pending map once a second while the session lives: calls -//! registered *after* the initial `fail_all` (still possible on the -//! forgotten-session path, where imports may race the drop) are failed -//! on the next sweep tick once EOF has been observed. No hang: pendings -//! resolve retryable regardless of registration-vs-EOF ordering -//! (CON-02). Subsequent handler calls fail on write; reconnect policy is -//! the assembly layer's job. +//! session was forgotten (fire-and-forget import). Once EOF has been +//! observed, registrations that land in the pending map *after* the +//! initial `fail_all` (still possible on the forgotten-session path, +//! where imports may race the drop) are failed fast: the monitor +//! watches the pending map at a short interval during a bounded +//! post-EOF window and drains any entry that lands — no wait for the +//! next one-second sweep tick (CON-18). No hang: pendings resolve +//! 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 -/// registered after the last `fail_all` (only effective once EOF has -/// been observed — see the module doc). -const PENDING_SWEEP_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1); +/// How often the drop monitor drains the pending map once EOF has been +/// observed, failing `fail_all`-after registrations fast (CON-18). +const PENDING_SWEEP_INTERVAL: std::time::Duration = std::time::Duration::from_millis(50); + +/// 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>, + #[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"); + } }