fix(socket): suppress retry-storm Sentry noise + empty-token guard (OPENHUMAN-TAURI-8M) (#1568)

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Co-authored-by: Steven Enamakel <enamakel@tinyhumans.ai>

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);
+    }
 }

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<bool>,
     internal_tx: mpsc::UnboundedSender<String>,
 ) {
+    // 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
 // ---------------------------------------------------------------------------

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::<String>();
+    // 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::<String>();
+
+    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};