From d8c51dc191e8020f12ad6deca6d0ffe9337c114d Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Sun, 30 Aug 2026 21:22:01 +0000 Subject: [PATCH 1/2] fix(adapters): dead-connection fast-fail + bounded from_wss sweep exit (CON-18) The review-001 CON-02 monitor swept the pending map once EOF held but never exited: every fire-and-forget import whose peer died left a spawned task + pending map + watch receiver alive for the process lifetime, and once the read pump ended the watch sender dropped so the old select's 'changed()' error branch spun. - replaced the 1s never-exiting sweep with a 50ms post-EOF drain (fast-fail) and a bounded grace window: the monitor ends after PENDING_SWEEP_MAX_POST_EOF (8) consecutive empty drains (~400ms) - post-EOF registrations now resolve in ~50ms with retryable CONNECTION_CLOSED instead of waiting up to 1s for the next sweep - removed the silent busy-spin: the loop now only wakes on the sweep tick or the session-close signal (close path still fails all + exits) - retained the monitor JoinHandle on WssDropMonitor (test builds) and added a lifecycle test asserting the monitor joins after EOF+grace - tightened call_registered_after_eof test: 500ms bound + CONNECTION_ CLOSED code assertion (was a pre-sleep + 5s timeout) Verification: - cargo test --features wss: 358 passed, 0 failed - cargo test --all-features: 454 passed, 0 failed - cargo clippy --all-targets -- -D warnings (default, wss, all): clean - cargo fmt --check: clean - scripts/verify.sh + verify.sh --all-features: VERIFY OK --- src/adapters/from_wss.rs | 118 +++++++++++++++++++++++++++++---------- 1 file changed, 90 insertions(+), 28 deletions(-) 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"); + } } From bb709bd10186f85934a78d838a9023283eb0ba08 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Sun, 30 Aug 2026 21:22:04 +0000 Subject: [PATCH 2/2] docs(adr 070): note the self-limiting from_wss drop monitor (CON-18) Documents the CON-18 disposition in the v1 session-lifetime section (teardown handle still future work, but a dead import no longer leaves a permanent monitor task) and adds the consequence pair: dead imports self-clean after EOF + bounded grace; registrations landing past the grace window fall back to the 30s sweeper deadline. --- .../070-from-wss-consumer-adapter.md | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/docs/architecture/decisions/070-from-wss-consumer-adapter.md b/docs/architecture/decisions/070-from-wss-consumer-adapter.md index b4cb1e1..c9640a1 100644 --- a/docs/architecture/decisions/070-from-wss-consumer-adapter.md +++ b/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; - v1 deliberately does not build a reconnect layer (OQ-03). +- **Close/teardown handle**: future work; there is no + `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