From f82f2d524d09c7df6152c009bc40d20cc0b71b25 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Wed, 13 May 2026 09:29:41 +0530 Subject: [PATCH] fix(socket): suppress retry-storm Sentry noise + empty-token guard (OPENHUMAN-TAURI-8M) (#1568) Co-authored-by: Claude Opus 4.7 (1M context) Co-authored-by: Steven Enamakel --- src/openhuman/socket/manager.rs | 32 ++++++++++ src/openhuman/socket/ws_loop.rs | 88 ++++++++++++++++++++++++++- src/openhuman/socket/ws_loop_tests.rs | 81 ++++++++++++++++++++++++ 3 files changed, 199 insertions(+), 2 deletions(-) diff --git a/src/openhuman/socket/manager.rs b/src/openhuman/socket/manager.rs index a530f7861..1ad219963 100644 --- a/src/openhuman/socket/manager.rs +++ b/src/openhuman/socket/manager.rs @@ -128,7 +128,18 @@ impl SocketManager { /// /// Spawns a background `ws_loop` that manages the connection with automatic /// reconnection and exponential backoff. + /// + /// Returns `Err` immediately if `token` is empty — every reconnect attempt + /// would either 401 at the SIO CONNECT step or fail upstream at the gateway, + /// producing exactly the kind of retry-storm noise this module is designed to + /// suppress. Callers receive an actionable error and the RPC response reflects + /// the actual outcome rather than optimistically reporting `{"status":"Connecting"}`. pub async fn connect(&self, url: &str, token: &str) -> Result<(), String> { + if token.trim().is_empty() { + log::error!("[socket] connect: refusing to start — empty session token"); + return Err("empty session token — authenticate first".to_string()); + } + // Ensure the rustls crypto provider is installed (needed for wss:// TLS). // This is a no-op if already installed. let _ = rustls::crypto::ring::default_provider().install_default(); @@ -338,4 +349,25 @@ mod tests { let err = mgr.emit("x", json!({})).await.unwrap_err(); assert_eq!(err, "Not connected"); } + + /// Empty-token guard at the `SocketManager::connect` boundary: + /// the RPC caller must receive an `Err` immediately — not + /// `{"status":"Connecting"}` — so the UI can surface an actionable error. + #[tokio::test] + async fn connect_rejects_empty_token_and_returns_err() { + let mgr = SocketManager::new(); + + // Bare empty string. + let err = mgr.connect("http://localhost:1", "").await.unwrap_err(); + assert!( + err.contains("empty session token"), + "expected 'empty session token' in error, got: {err}" + ); + assert_eq!(mgr.get_state().status, ConnectionStatus::Disconnected); + + // Whitespace-only string (trim check). + let err = mgr.connect("http://localhost:1", " ").await.unwrap_err(); + assert!(err.contains("empty session token"), "{err}"); + assert_eq!(mgr.get_state().status, ConnectionStatus::Disconnected); + } } diff --git a/src/openhuman/socket/ws_loop.rs b/src/openhuman/socket/ws_loop.rs index 9c6a6eef3..72c214413 100644 --- a/src/openhuman/socket/ws_loop.rs +++ b/src/openhuman/socket/ws_loop.rs @@ -29,6 +29,24 @@ const MAX_REDIRECT_HOPS: u8 = 3; // Background loop // --------------------------------------------------------------------------- +/// Number of consecutive `ConnectionOutcome::Failed` attempts at which the +/// loop fires exactly one `error`-level log (and therefore one Sentry event). +/// Below the threshold, repeated transient failures (gateway 5xx, TLS +/// handshake resets, DNS blips) stay at `warn` and don't reach the Sentry +/// tracing layer. Above the threshold, subsequent retries return to `warn` — +/// the one-shot `error` at the threshold is sufficient to page on a sustained +/// outage without generating unbounded events. +/// +/// The value is 5 intentionally: the Sentry event fires on the **5th +/// consecutive failed attempt**, which corresponds to ~15 seconds of +/// accumulated backoff sleep (1 s + 2 s + 4 s + 8 s before the 5th try). +/// Transient blips that recover within 4 attempts produce zero Sentry noise; +/// sustained outages produce exactly one event per affected client. +/// +/// See OPENHUMAN-TAURI-8M — a single gateway 503 incident generated 549 +/// Sentry events because every retry was logged at `error`. +const FAIL_ESCALATE_THRESHOLD: u32 = 5; + /// Background loop that manages the WebSocket connection and reconnection. pub(super) async fn ws_loop( url: String, @@ -38,8 +56,24 @@ pub(super) async fn ws_loop( mut shutdown_rx: watch::Receiver, internal_tx: mpsc::UnboundedSender, ) { + // Defense in depth: `SocketManager::connect` is the primary guard and + // will reject an empty token before spawning this task. This check + // covers any future caller that invokes `ws_loop` directly (e.g. tests + // or internal plumbing bypassing the manager). Without it, every attempt + // would either 401 at the SIO CONNECT step or fail upstream at the + // gateway, producing the retry-storm Sentry noise this module is designed + // to suppress. + if token.trim().is_empty() { + log::error!("[socket] ws_loop: refusing to start — empty session token"); + *shared.status.write() = ConnectionStatus::Disconnected; + *shared.socket_id.write() = None; + emit_state_change(&shared); + return; + } + let mut backoff = Duration::from_millis(1000); let max_backoff = Duration::from_secs(30); + let mut consecutive_failures: u32 = 0; // `ws_url` is the *resolved* socket URL we're currently connecting to. // If the backend responds with an HTTP 3xx during the upgrade (typical when @@ -73,11 +107,23 @@ pub(super) async fn ws_loop( break; } ConnectionOutcome::Lost(reason) => { + // `Lost` is only returned after a successful SIO CONNECT ACK + // (see `run_connection`), so reaching this arm proves the + // backend is reachable and the token is valid. Reset both + // the backoff and the failure streak. + if consecutive_failures > 0 { + log::debug!( + "[socket] Connection re-established; resetting failure streak ({} cleared)", + consecutive_failures + ); + } + consecutive_failures = 0; log::warn!("[socket] Connection lost: {}", reason); - backoff = Duration::from_millis(1000); // reset on established-then-lost + backoff = Duration::from_millis(1000); } ConnectionOutcome::Failed(reason) => { - log::error!("[socket] Connection failed: {}", reason); + consecutive_failures = consecutive_failures.saturating_add(1); + log_connection_failure(consecutive_failures, &reason); // keep growing backoff } } @@ -106,6 +152,44 @@ pub(super) async fn ws_loop( emit_state_change(&shared); } +// --------------------------------------------------------------------------- +// Failure logging +// --------------------------------------------------------------------------- + +/// Log a connection failure at the appropriate level based on how many +/// consecutive failures have occurred. +/// +/// - Below `FAIL_ESCALATE_THRESHOLD`: `warn` — transient blips (DNS, gateway +/// 5xx, TLS resets) stay out of Sentry. +/// - Exactly at the threshold: `error` — fires the one-shot Sentry event that +/// signals a sustained outage. +/// - Above the threshold: `warn` — already paged once; avoid unbounded events +/// during a long outage. +/// +/// Extracted as a pure function so it can be unit-tested without running an +/// async event loop or touching the WS stack. +fn log_connection_failure(consecutive: u32, reason: &str) { + if consecutive == FAIL_ESCALATE_THRESHOLD { + // One-shot escalation: fire exactly one Sentry-visible `error` event so + // sustained outages surface without generating unbounded events. + log::error!( + "[socket] Connection failed (sustained outage after {} attempts): {}", + consecutive, + reason + ); + } else { + // Below threshold (transient blips) or above threshold (already fired + // the one-shot error): stay at `warn` so subsequent retries don't pile + // up additional Sentry events. + log::warn!( + "[socket] Connection failed (attempt {}/{}): {}", + consecutive, + FAIL_ESCALATE_THRESHOLD, + reason + ); + } +} + // --------------------------------------------------------------------------- // Single connection attempt // --------------------------------------------------------------------------- diff --git a/src/openhuman/socket/ws_loop_tests.rs b/src/openhuman/socket/ws_loop_tests.rs index 19b008401..765a53449 100644 --- a/src/openhuman/socket/ws_loop_tests.rs +++ b/src/openhuman/socket/ws_loop_tests.rs @@ -243,6 +243,41 @@ fn handle_sio_packet_unknown_type_is_noop() { assert_eq!(*shared.status.read(), ConnectionStatus::Connected); } +// ── log_connection_failure ───────────────────────────────────── + +/// Verify that `log_connection_failure` does not panic for any call count +/// (below, at, or above the threshold). The one-shot escalation fires at +/// exactly `FAIL_ESCALATE_THRESHOLD`; above it reverts to `warn`. We can't +/// assert on log output in unit tests, but the no-panic invariant combined +/// with `fail_escalate_threshold_is_five` keeps the doc comment honest. +#[test] +fn log_connection_failure_does_not_panic_below_threshold() { + // Calls 1 through threshold-1 stay at warn — must complete without panic. + for i in 1..FAIL_ESCALATE_THRESHOLD { + log_connection_failure(i, "simulated transient failure"); + } +} + +#[test] +fn log_connection_failure_does_not_panic_at_and_above_threshold() { + // Calls at and above threshold escalate to error — must also not panic. + for i in FAIL_ESCALATE_THRESHOLD..=FAIL_ESCALATE_THRESHOLD + 3 { + log_connection_failure(i, "simulated sustained failure"); + } +} + +#[test] +fn fail_escalate_threshold_is_five() { + // Threshold of 5 is load-bearing (doc says "~15s of accumulated backoff"). + // If the value changes the doc comment and the backoff math must be updated + // together — this test surfaces the discrepancy immediately. + assert_eq!( + FAIL_ESCALATE_THRESHOLD, 5, + "FAIL_ESCALATE_THRESHOLD changed — update the doc comment to reflect the \ + new backoff accumulation before the first Sentry event" + ); +} + // ── End-to-end handshake tests against a local WS server ─────── // // These tests drive the real `ws_loop` / `run_connection` code path @@ -495,6 +530,52 @@ fn connect_behavior_variants_are_distinct() { } } +/// Empty-token guard: if a caller invokes the bare `connect(url, token)` RPC +/// with an empty string the loop must bail immediately rather than spin a +/// doomed reconnect cycle that fires Sentry events on every retry. The +/// status must end up `Disconnected` and the function must return without +/// completing the reconnect loop. +#[tokio::test] +async fn ws_loop_refuses_to_start_with_empty_token() { + let shared = make_shared(); + *shared.status.write() = ConnectionStatus::Connecting; + *shared.socket_id.write() = Some("stale".into()); + + // `_emit_tx` is kept alive so the emit channel is not closed before the + // task starts — a closed sender would give ws_loop an `emit_rx` that + // immediately returns `None`, potentially exiting via the Shutdown arm + // before the empty-token guard is ever reached and masking a regression. + let (_emit_tx, emit_rx) = mpsc::unbounded_channel::(); + // Shutdown channel is never signalled — if the guard fails, the test + // will time out waiting for the spawned task to complete. + let (_shutdown_tx, shutdown_rx) = watch::channel(false); + let (internal_tx, _internal_rx) = mpsc::unbounded_channel::(); + + let loop_shared = Arc::clone(&shared); + let handle = tokio::spawn(async move { + ws_loop( + // URL is deliberately invalid — if the guard misfires, the + // task would error on connect rather than return immediately. + "http://invalid.example.invalid:1".into(), + " ".into(), // whitespace-only counts as empty + loop_shared, + emit_rx, + shutdown_rx, + internal_tx, + ) + .await; + }); + + let res = tokio::time::timeout(tokio::time::Duration::from_secs(2), handle).await; + assert!( + matches!(res, Ok(Ok(()))), + "ws_loop must return cleanly on empty token (no timeout, no panic)" + ); + + assert_eq!(*shared.status.read(), ConnectionStatus::Disconnected); + assert!(shared.socket_id.read().is_none()); +} + // ── End-to-end redirect-follow (the real fix for the 301 noise) ── use tokio::io::{AsyncReadExt, AsyncWriteExt};