From f96fb006de98e92acf4751eb8b8f1d773fbf518f Mon Sep 17 00:00:00 2001 From: brettchien Date: Sat, 15 Aug 2026 15:55:24 +0800 Subject: [PATCH] feat(remote): exponential reconnect backoff + classified disconnect reason MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The reconnect loop retried on a flat 5s cadence forever and surfaced every failure as a raw error blob. With the gateway's handful of session slots, a flapping link hammered it on a fixed beat, and the operator couldn't tell a network blip from an auth reject or the server being at capacity. - acp-tunnel: new pure `reconnect` module (unit-tested, no GTK needed): - `backoff_delay(attempt, salt)` — exponential 1→2→4→8→16s capped at 30s with per-connection jitter. - `DisconnectReason::classify(&str)` + `label()` — network / auth rejected / server at capacity / protocol / other. - remote.rs run_reconnecting: escalate backoff on consecutive fast failures, reset once a connection has held ≥15s, and put the classified reason in the status line (e.g. "error: server at capacity") + log. Tests: 3 new reconnect unit tests (classification + backoff shape/cap/jitter); the Session state machine is already covered in acp-tunnel. The pure policy now lives in the workspace-tested crate so the Tauri driver stays thin. Co-Authored-By: Claude Opus 4.8 --- crates/acp-tunnel/src/lib.rs | 2 + crates/acp-tunnel/src/reconnect.rs | 144 +++++++++++++++++++++++++++++ src-tauri/src/remote.rs | 30 +++++- 3 files changed, 171 insertions(+), 5 deletions(-) create mode 100644 crates/acp-tunnel/src/reconnect.rs diff --git a/crates/acp-tunnel/src/lib.rs b/crates/acp-tunnel/src/lib.rs index 103c829..e2da35f 100644 --- a/crates/acp-tunnel/src/lib.rs +++ b/crates/acp-tunnel/src/lib.rs @@ -25,6 +25,8 @@ use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; pub mod config; +pub mod reconnect; +pub use reconnect::{backoff_delay, DisconnectReason}; /// The ACP sub-protocol token the server echoes back on the `/acp` upgrade. pub const ACP_SUBPROTOCOL: &str = "acp.v1"; diff --git a/crates/acp-tunnel/src/reconnect.rs b/crates/acp-tunnel/src/reconnect.rs new file mode 100644 index 0000000..c5f683a --- /dev/null +++ b/crates/acp-tunnel/src/reconnect.rs @@ -0,0 +1,144 @@ +//! Reconnect policy for the remote `/acp` client: how long to wait between +//! attempts, and how to classify *why* an attempt ended so the operator sees a +//! meaningful reason instead of a raw error string. Pure + unit-tested here so the +//! Tauri driver (`src-tauri/src/remote.rs`) stays thin. + +use std::time::Duration; + +/// Why a connection attempt ended — drives the operator-facing status line. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum DisconnectReason { + /// DNS / TCP / TLS / timeout — the socket never came up, or dropped mid-stream. + Network, + /// The gateway rejected our credentials (bad / expired `/acp` bearer). + Auth, + /// The gateway is at capacity (`max_sessions` reached); a retry may land once a + /// slot frees. + SlotsFull, + /// Handshake / framing / JSON-RPC level failure once connected. + Protocol, + /// Anything not matched above. + Other, +} + +impl DisconnectReason { + /// Best-effort classification from an error string — case-insensitive substring + /// match, most-specific buckets first. Errors here are cosmetic (they pick the + /// status label), never load-bearing, so an unmatched string just falls to + /// [`DisconnectReason::Other`]. + pub fn classify(err: &str) -> Self { + let e = err.to_ascii_lowercase(); + if e.contains("max_sessions") + || e.contains("too many sessions") + || e.contains("capacity") + || e.contains("no free slot") + || e.contains("503") + || e.contains("service unavailable") + { + Self::SlotsFull + } else if e.contains("401") + || e.contains("403") + || e.contains("unauthorized") + || e.contains("forbidden") + || e.contains("invalid token") + { + Self::Auth + } else if e.contains("dns") + || e.contains("lookup address") + || e.contains("connect") + || e.contains("connection reset") + || e.contains("timed out") + || e.contains("timeout") + || e.contains("broken pipe") + || e.contains("io error") + || e.contains("os error") + { + Self::Network + } else if e.contains("handshake") + || e.contains("subprotocol") + || e.contains("protocol") + || e.contains("-32") + || e.contains("unexpected") + { + Self::Protocol + } else { + Self::Other + } + } + + /// Short operator-facing phrase for the status line / log. + pub fn label(&self) -> &'static str { + match self { + Self::Network => "network", + Self::Auth => "auth rejected", + Self::SlotsFull => "server at capacity", + Self::Protocol => "protocol error", + Self::Other => "error", + } + } +} + +/// Backoff before the next reconnect attempt: exponential (1s, 2s, 4s, 8s, 16s) +/// capped at 30s, plus a small deterministic jitter derived from `salt` (e.g. the +/// first byte of the connection id) so a flapping link doesn't retry on an exact +/// cadence. `attempt` is 0-based (0 = first retry after a drop). +pub fn backoff_delay(attempt: u32, salt: u8) -> Duration { + const CAP_SECS: u64 = 30; + let secs = (1u64 << attempt.min(5)).min(CAP_SECS); // 1,2,4,8,16,32→cap 30 + let jitter_ms = (salt as u64) * 4 % 1000; // 0..=996 ms, always < 1s + Duration::from_secs(secs) + Duration::from_millis(jitter_ms) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn classify_maps_known_errors() { + use DisconnectReason::*; + // the exact DNS storm string seen in the Activity log + assert_eq!( + DisconnectReason::classify( + "dial wss://…/acp: IO error: failed to lookup address information: nodename nor servname provided" + ), + Network + ); + assert_eq!( + DisconnectReason::classify("ws read: Connection reset by peer"), + Network + ); + assert_eq!(DisconnectReason::classify("HTTP 401 Unauthorized"), Auth); + assert_eq!( + DisconnectReason::classify("gateway refused: max_sessions reached"), + SlotsFull + ); + assert_eq!( + DisconnectReason::classify("503 Service Unavailable"), + SlotsFull + ); + assert_eq!( + DisconnectReason::classify("handshake failed: bad subprotocol"), + Protocol + ); + assert_eq!(DisconnectReason::classify("something inexplicable"), Other); + } + + #[test] + fn backoff_is_exponential_and_capped() { + assert_eq!(backoff_delay(0, 0).as_secs(), 1); + assert_eq!(backoff_delay(1, 0).as_secs(), 2); + assert_eq!(backoff_delay(2, 0).as_secs(), 4); + assert_eq!(backoff_delay(3, 0).as_secs(), 8); + assert_eq!(backoff_delay(4, 0).as_secs(), 16); + assert_eq!(backoff_delay(5, 0).as_secs(), 30); // 32 → cap + assert_eq!(backoff_delay(20, 0).as_secs(), 30); // stays capped + } + + #[test] + fn backoff_jitter_is_bounded_under_one_second() { + for salt in [0u8, 1, 42, 127, 249, 255] { + let extra_ms = backoff_delay(0, salt).as_millis() as u64 - 1000; + assert!(extra_ms < 1000, "jitter {extra_ms}ms should be < 1s"); + } + } +} diff --git a/src-tauri/src/remote.rs b/src-tauri/src/remote.rs index 165a2c2..9db6868 100644 --- a/src-tauri/src/remote.rs +++ b/src-tauri/src/remote.rs @@ -17,7 +17,7 @@ use std::time::{Duration, Instant}; use acp_tunnel as acp; use acp_tunnel::config::RemoteConfig; -use acp_tunnel::{Inbound, Session}; +use acp_tunnel::{DisconnectReason, Inbound, Session}; use futures_util::{Sink, SinkExt, StreamExt}; use serde_json::{json, Value}; use tauri::{AppHandle, Emitter, Manager, Runtime}; @@ -218,6 +218,10 @@ async fn run_reconnecting( stop: Arc, ) { let mut session = Session::new(vec![acp::oab_server(&uuid::Uuid::new_v4().to_string())]); + // Consecutive-failure counter driving the reconnect backoff. Reset to 0 once an + // attempt has held a live connection for a while (see below), so a long-running + // session that blips reconnects promptly instead of at the capped delay. + let mut attempt: u32 = 0; loop { if stop.load(Ordering::SeqCst) { break; @@ -233,6 +237,7 @@ async fn run_reconnecting( ); } + let started = Instant::now(); let result = run_once(&app, &cfg, &client, &mut session, &conn_id).await; // The socket is gone — retract the outbound-chat channel so a prompt // between attempts fails fast rather than dropping into a dead sink. @@ -242,19 +247,34 @@ async fn run_reconnecting( if stop.load(Ordering::SeqCst) { break; } + // A drop after a decently long-lived connection is a fresh incident, not an + // escalating failure — reset the backoff so we retry quickly. A fast failure + // (bad dial / handshake) keeps escalating. + if started.elapsed() >= Duration::from_secs(15) { + attempt = 0; + } if let Err(e) = result { - emit_status(&app, &format!("error: {e}")); + // Classify so the status line says *why* (network / auth rejected / + // server at capacity / protocol) instead of a raw error blob. + let reason = DisconnectReason::classify(&e); + emit_status(&app, &format!("error: {}", reason.label())); let _ = app.emit( "app-log", - json!({ "level": "error", "msg": format!("remote: {e}") }), + json!({ "level": "error", "msg": format!("remote: {} — {e}", reason.label()) }), ); } + // Exponential backoff (capped 30s) + per-connection jitter, so a flapping + // link doesn't hammer the gateway — which only has a handful of session + // slots — on a fixed cadence. + let salt = conn_id.as_bytes().first().copied().unwrap_or(0); + let delay = acp::backoff_delay(attempt, salt); + attempt = attempt.saturating_add(1); emit_status(&app, "connecting"); let _ = app.emit( "app-log", - json!({ "level": "info", "msg": "remote: reconnecting in 5s…" }), + json!({ "level": "info", "msg": format!("remote: reconnecting in {}s…", delay.as_secs()) }), ); - tokio::time::sleep(std::time::Duration::from_secs(5)).await; + tokio::time::sleep(delay).await; } }