diff --git a/src/adapters/from_wss.rs b/src/adapters/from_wss.rs index 449e521..f647ed4 100644 --- a/src/adapters/from_wss.rs +++ b/src/adapters/from_wss.rs @@ -23,10 +23,18 @@ //! Reconnect policy (OQ-03 disposition): v1 = none. A connection drop //! fails in-flight calls retryable: alkcall's client-side read pump only //! routes envelopes and does not observe EOF, so the adapter owns drop -//! semantics — the [`WssSession`] monitor awaits the WS read pump and -//! fails all pending calls with retryable `CONNECTION_CLOSED`. Subsequent -//! handler calls fail on write; reconnect policy is the assembly layer's -//! job. +//! semantics — the [`WssSession`] monitor awaits the WS read pump's EOF +//! 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. //! //! [ADR-070]: https://docs.rs/alkhttp (docs/architecture/decisions) @@ -43,6 +51,11 @@ 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); + pub struct FromWss { endpoint: String, auth_token: Option, @@ -151,18 +164,38 @@ impl WssSession { })?; let call_connection = Arc::new(call_connection); - let (close_tx, close_rx) = tokio::sync::oneshot::channel(); + let (close_tx, mut close_rx) = tokio::sync::oneshot::channel(); let pending = Arc::clone(call_connection.pending()); + let mut eof_rx = pumps.read_eof(); tokio::spawn(async move { - tokio::select! { - _ = pumps.read_eof() => {}, - _ = close_rx => {}, + let mut sweep = tokio::time::interval(PENDING_SWEEP_INTERVAL); + sweep.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { + tokio::select! { + _ = crate::websocket::wait_for_eof(&mut eof_rx) => break, + _ = &mut close_rx => break, + _ = sweep.tick() => { + if *eof_rx.borrow() { + pending.lock().fail_all( + CallError::new( + "CONNECTION_CLOSED", + "from_wss connection dropped", + true, + ), + ); + } + } + } + } + let failure = CallError::new("CONNECTION_CLOSED", "from_wss connection dropped", true); + let mut guard = pending.lock(); + guard.fail_all(failure.clone()); + if *eof_rx.borrow() { + let swept = guard.fail_all(failure); + if !swept.is_empty() { + tracing::debug!("from_wss: sweep failed {} late pendings", swept.len()); + } } - pending.lock().fail_all(CallError::new( - "CONNECTION_CLOSED", - "from_wss connection dropped", - true, - )); }); Ok(Self { diff --git a/src/websocket/byte_adapter.rs b/src/websocket/byte_adapter.rs index 437a497..bf7e217 100644 --- a/src/websocket/byte_adapter.rs +++ b/src/websocket/byte_adapter.rs @@ -36,9 +36,6 @@ use std::{ task::{Context, Poll}, }; -#[cfg(any(test, feature = "wss"))] -use std::sync::Arc; - use axum::extract::ws::{CloseFrame, Message as AxumMessage, WebSocket}; use futures::channel::mpsc as futures_mpsc; use futures::{SinkExt, StreamExt}; @@ -84,7 +81,7 @@ pub struct WsPumps { read_task: tokio::task::JoinHandle<()>, write_task: tokio::task::JoinHandle<()>, #[cfg(feature = "wss")] - read_eof: Arc, + read_eof: tokio::sync::watch::Sender, } impl WsPumps { @@ -93,12 +90,35 @@ impl WsPumps { self.write_task.abort(); } - /// Fires when the WS read side reaches EOF (socket close from either - /// side) — used by `from_wss`'s connection-drop monitor to await - /// socket EOF (ADR-070). + /// A lossless receiver for the WS read-EOF signal (socket close from + /// either side): the watch channel retains the latest value, so an + /// EOF signaled at any point — including before the receiver is + /// taken or the observer starts awaiting — is still observed, and + /// may be observed repeatedly. Used by `from_wss`'s + /// connection-drop monitor (ADR-070); await it with + /// [`wait_for_eof`]. #[cfg(feature = "wss")] - pub(crate) async fn read_eof(&self) { - self.read_eof.notified().await; + pub(crate) fn read_eof(&self) -> tokio::sync::watch::Receiver { + self.read_eof.subscribe() + } +} + +/// Resolves once the given WS read-EOF watch receiver observes `true`. +/// +/// Lossless by construction: a `watch::Receiver` retains the latest +/// value, so an EOF signaled before this call (or before the receiver +/// existed) is still observed — the property a one-shot `Notify` lacked +/// (WS-02). Returns early (treated as EOF) if the sender half is +/// dropped, e.g. the read pump was aborted. +#[cfg(feature = "wss")] +pub(crate) async fn wait_for_eof(rx: &mut tokio::sync::watch::Receiver) { + loop { + if *rx.borrow_and_update() { + return; + } + if rx.changed().await.is_err() { + return; + } } } @@ -111,11 +131,11 @@ pub fn split_ws_to_bytes(socket: WebSocket) -> (WsByteStream, WsPumps) { let (write_tx, mut write_rx) = futures_mpsc::channel::(WRITE_SLOTS); #[cfg(feature = "wss")] - let read_eof = Arc::new(tokio::sync::Notify::new()); + let read_eof = tokio::sync::watch::channel(false).0; let write_tx_for_read = write_tx.clone(); #[cfg(feature = "wss")] - let read_eof_for_task = Arc::clone(&read_eof); + let read_eof_for_task = read_eof.clone(); let read_task = tokio::spawn(async move { while let Some(msg) = ws_stream.next().await { match msg { @@ -136,7 +156,9 @@ pub fn split_ws_to_bytes(socket: WebSocket) -> (WsByteStream, WsPumps) { } } #[cfg(feature = "wss")] - read_eof_for_task.notify_waiters(); + { + let _ = read_eof_for_task.send(true); + } }); let write_task = tokio::spawn(async move { @@ -293,10 +315,10 @@ where let (read_tx, read_rx) = mpsc::channel::>(READ_SLOTS); let (write_tx, mut write_rx) = futures_mpsc::channel::(WRITE_SLOTS); - let read_eof = Arc::new(tokio::sync::Notify::new()); + let read_eof = tokio::sync::watch::channel(false).0; let write_tx_for_read = write_tx.clone(); - let read_eof_for_task = Arc::clone(&read_eof); + let read_eof_for_task = read_eof.clone(); let read_task = tokio::spawn(async move { while let Some(msg) = ws_stream.next().await { match msg { @@ -316,7 +338,7 @@ where Ok(_) => {} } } - read_eof_for_task.notify_waiters(); + let _ = read_eof_for_task.send(true); }); let write_task = tokio::spawn(async move { diff --git a/src/websocket/mod.rs b/src/websocket/mod.rs index 1219736..ee2fac7 100644 --- a/src/websocket/mod.rs +++ b/src/websocket/mod.rs @@ -18,6 +18,8 @@ pub use byte_adapter::{ #[cfg(any(test, feature = "wss"))] pub use byte_adapter::split_tungstenite_to_bytes; +#[cfg(feature = "wss")] +pub(crate) use byte_adapter::wait_for_eof; pub use upgrade::{run_channels_session, ws_bearer_auth, ws_upgrade_handler}; #[cfg(any(test, feature = "test-support"))]