fix(pair-relay): keep pairing socket open past desktop timeout

crates/buzz-pair-relay/src/lib.rs

@@ -9,18 +9,22 @@
 //! This binary binds **loopback only** and MUST run behind a reverse proxy
 //! (nginx, caddy, etc.) that:
 //! - Routes only `/pair` to this sidecar
-//! - Enforces HTTP read timeouts (mitigates slowloris at the TCP layer)
+//! - Keeps the pre-upgrade HTTP header/request timeout short and applies
+//!   connection/rate limits before forwarding to the sidecar
+//! - Keeps post-upgrade WebSocket read/send/idle timeouts **greater than 140
+//!   seconds** so the proxy cannot preempt [`CONN_TIMEOUT`]
 //! - Terminates TLS
 //!
-//! The relay does not enforce path restrictions or pre-upgrade connection
-//! limits — those are the reverse proxy's responsibility.
+//! The relay independently enforces [`HTTP_HEADER_READ_TIMEOUT`]. Path
+//! restrictions and pre-upgrade connection/rate limits remain the reverse
+//! proxy's responsibility.
 //!
 //! # Security Model
 //!
 //! - **Signature verification** — Schnorr signatures are verified against the
 //!   NIP-01 event ID hash. Events with invalid signatures are rejected.
 //! - **No persistence** — events exist only in-flight between matched pub/sub.
-//! - **Bounded resources** — 128 max WS connections, 4 KiB max frame, 120s TTL.
+//! - **Bounded resources** — 128 max WS connections, 4 KiB max frame, 140s TTL.
 //! - **Session cap** — at most 6 accepted EVENTs per connection.
 //! - **Freshness** — `created_at` must be within ±120 s of relay wall-clock.
 //! - **Deduplication** — duplicate event IDs are rejected; dedup entries expire after 300 s.
@@ -41,7 +45,7 @@ use hyper::server::conn::http1;
 use hyper::service::service_fn;
 use hyper::upgrade::Upgraded;
 use hyper::{Method, Request, Response, StatusCode, Version};
-use hyper_util::rt::TokioIo;
+use hyper_util::rt::{TokioIo, TokioTimer};
 use parking_lot::Mutex;
 use secp256k1::schnorr::Signature as SchnorrSig;
 use secp256k1::XOnlyPublicKey;
@@ -55,8 +59,20 @@ use tokio_tungstenite::tungstenite::protocol::{Message, Role, WebSocketConfig};
 use tokio_tungstenite::WebSocketStream;
 use tokio_util::sync::CancellationToken;
 
-/// Hard per-connection lifetime. `pub(crate)` for test access.
-pub(crate) const CONN_TIMEOUT: Duration = Duration::from_secs(120);
+/// Hard per-connection lifetime.
+///
+/// Must outlive desktop `PAIRING_HARD_TIMEOUT` in
+/// `desktop/src-tauri/src/commands/pairing.rs`. If this sidecar closes first,
+/// the desktop surfaces the transport string "relay connection closed" instead
+/// of its own "Session timed out" expired state.
+pub const CONN_TIMEOUT: Duration = Duration::from_secs(140);
+
+/// Maximum time allowed to receive a complete pre-upgrade HTTP header.
+///
+/// This is intentionally much shorter than [`CONN_TIMEOUT`]: ordinary HTTP
+/// parsing is the slowloris boundary, while the longer timeout applies only
+/// after a valid WebSocket upgrade has reserved a bounded relay slot.
+pub const HTTP_HEADER_READ_TIMEOUT: Duration = Duration::from_secs(10);
 
 const MAX_CONNS: u32 = 128;
 const CHANNEL_CAP: usize = 4;
@@ -128,6 +144,12 @@ impl Relay {
         }
     }
 
+    /// Number of connections that completed a valid WebSocket upgrade and
+    /// currently hold one of the bounded relay slots.
+    pub fn active_connection_count(&self) -> u32 {
+        self.conn_count.load(Ordering::Relaxed)
+    }
+
     /// Atomically check-and-reserve an event ID. Evicts expired entries first.
     /// Returns `Ok(true)` if duplicate (already seen), `Ok(false)` if new
     /// (reserved — caller MUST call `unreserve_id` if delivery fails),
@@ -1013,11 +1035,10 @@ pub async fn run_server(listener: TcpListener, relay: Arc<Relay>) {
         tokio::spawn(async move {
             let io = TokioIo::new(tcp);
             let svc = service_fn(move |req| http_service(Arc::clone(&relay), req));
-            if let Err(e) = http1::Builder::new()
-                .serve_connection(io, svc)
-                .with_upgrades()
-                .await
-            {
+            let mut http = http1::Builder::new();
+            http.timer(TokioTimer::new())
+                .header_read_timeout(HTTP_HEADER_READ_TIMEOUT);
+            if let Err(e) = http.serve_connection(io, svc).with_upgrades().await {
                 eprintln!("http error: {e}");
             }
         });

crates/buzz-pair-relay/tests/integration.rs

@@ -6,12 +6,13 @@
 use std::sync::Arc;
 use std::time::Duration;
 
-use futures_util::{SinkExt, StreamExt};
+use futures_util::{FutureExt, SinkExt, StreamExt};
 use serde_json::{json, Value};
-use tokio::net::TcpListener;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::{TcpListener, TcpStream};
 use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream};
 
-use buzz_pair_relay::{run_server, Relay};
+use buzz_pair_relay::{run_server, Relay, CONN_TIMEOUT};
 
 use secp256k1::{Keypair, Secp256k1, SecretKey};
 use sha2::{Digest, Sha256};
@@ -32,11 +33,28 @@ type WS = WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>>;
 
 /// Start a relay on a random port, return the WebSocket URL.
 async fn start_relay() -> String {
+    start_relay_with_state().await.0
+}
+
+/// Start a relay and retain its state for resource-accounting assertions.
+async fn start_relay_with_state() -> (String, Arc<Relay>) {
     let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
     let addr = listener.local_addr().unwrap();
     let relay = Arc::new(Relay::new());
-    tokio::spawn(run_server(listener, relay));
-    format!("ws://127.0.0.1:{}", addr.port())
+    tokio::spawn(run_server(listener, Arc::clone(&relay)));
+    (format!("ws://127.0.0.1:{}", addr.port()), relay)
+}
+
+/// Wait for connection cleanup to reach an observable state before asserting
+/// client-side EOF in virtual-time tests.
+async fn wait_for_active_connections(relay: &Relay, expected: u32) {
+    for _ in 0..50 {
+        if relay.active_connection_count() == expected {
+            return;
+        }
+        tokio::task::yield_now().await;
+    }
+    assert_eq!(relay.active_connection_count(), expected);
 }
 
 /// Connect a WebSocket client to the relay.
@@ -86,6 +104,21 @@ async fn assert_closed(ws: &mut WS) {
     }
 }
 
+/// Assert closure without a virtual-time timeout, which can elapse before the
+/// detached WebSocket writer observes cancellation.
+async fn assert_closed_after_yields(ws: &mut WS) {
+    for _ in 0..100 {
+        if let Some(result) = ws.next().now_or_never() {
+            match result {
+                None | Some(Ok(Message::Close(_))) | Some(Err(_)) => return,
+                Some(Ok(other)) => panic!("expected close, got {other:?}"),
+            }
+        }
+        tokio::task::yield_now().await;
+    }
+    panic!("connection did not close after server cleanup");
+}
+
 /// Generate a random keypair; returns `(SecretKey, pubkey_hex)`.
 fn gen_keypair() -> (SecretKey, String) {
     let secp = Secp256k1::new();
@@ -438,18 +471,17 @@ async fn test_second_sub_same_id() {
     assert_eq!(ev_msg[1], "s1");
 }
 
-/// 9. Connection closes after 120 s (virtual time).
+/// 9. Connection closes after CONN_TIMEOUT (virtual time).
 #[tokio::test(start_paused = true)]
-async fn test_120s_timeout() {
-    let url = start_relay().await;
+async fn test_conn_timeout() {
+    let (url, relay) = start_relay_with_state().await;
     let mut ws = connect(&url).await;
 
     // Advance virtual time past the connection timeout.
-    tokio::time::advance(Duration::from_secs(121)).await;
-    // Yield to let the relay task run its deadline branch.
-    tokio::task::yield_now().await;
+    tokio::time::advance(CONN_TIMEOUT + Duration::from_secs(1)).await;
+    wait_for_active_connections(&relay, 0).await;
 
-    assert_closed(&mut ws).await;
+    assert_closed_after_yields(&mut ws).await;
 }
 
 /// 10. Backpressure unit test: bounded mpsc channel rejects when full.
@@ -605,6 +637,37 @@ async fn test_global_conn_cap() {
     let _new = connect(&url).await;
 }
 
+/// An incomplete pre-upgrade request is bounded independently of the longer
+/// WebSocket lifetime and never consumes one of the 128 upgraded slots.
+#[tokio::test]
+async fn test_incomplete_http_header_times_out_before_ws_pool() {
+    const HEADER_CLOSE_DEADLINE: Duration = Duration::from_secs(12);
+
+    let (url, relay) = start_relay_with_state().await;
+    let addr = url.strip_prefix("ws://").unwrap();
+    let mut slow = TcpStream::connect(addr).await.unwrap();
+    slow.write_all(b"GET /pair HTTP/1.1\r\nHost: localhost\r\n")
+        .await
+        .unwrap();
+
+    // The incomplete request has not upgraded and therefore must not consume
+    // any of the bounded WebSocket pool.
+    tokio::task::yield_now().await;
+    assert_eq!(relay.active_connection_count(), 0);
+
+    let mut byte = [0_u8; 1];
+    let read = tokio::time::timeout(HEADER_CLOSE_DEADLINE, slow.read(&mut byte)).await;
+    assert!(
+        matches!(read, Ok(Ok(0)) | Ok(Err(_))),
+        "incomplete HTTP header remained open after {HEADER_CLOSE_DEADLINE:?}: {read:?}"
+    );
+
+    // A valid upgrade still acquires the first pool slot after the slow
+    // request is discarded.
+    let _ws = connect(&url).await;
+    assert_eq!(relay.active_connection_count(), 1);
+}
+
 /// 18. Session event cap: 6 EVENTs are accepted; 7th is rejected with
 ///     "session event limit reached".  (The relay's hard cap is 6 per
 ///     connection, which is tighter than the per-window rate limit of 10.)
@@ -1176,18 +1239,18 @@ async fn test_reader_backpressure_closes() {
     }
 }
 
-/// 42. Connection closes promptly after 120 s (virtual time).
+/// 42. Connection closes promptly after CONN_TIMEOUT (virtual time).
 ///     Explicit duplicate of test 9 with a slightly different assertion style.
 #[tokio::test(start_paused = true)]
 async fn test_cancellation_immediate() {
-    let url = start_relay().await;
+    let (url, relay) = start_relay_with_state().await;
     let mut ws = connect(&url).await;
 
-    tokio::time::advance(Duration::from_secs(121)).await;
-    tokio::task::yield_now().await;
+    tokio::time::advance(CONN_TIMEOUT + Duration::from_secs(1)).await;
+    wait_for_active_connections(&relay, 0).await;
 
     // The connection must be closed — not just slow.
-    assert_closed(&mut ws).await;
+    assert_closed_after_yields(&mut ws).await;
 }
 
 /// 43. Client-initiated graceful close receives a Close reply.

deploy/charts/buzz/README.md

@@ -167,6 +167,24 @@ clients connect directly to the dedicated endpoint. The chart does not create
 an Ingress or HTTPRoute for the pairing Service; route the public hostname to
 `<release>-buzz-pairing:5000` with your platform's ingress configuration.
 
+Configure two distinct timeout classes on the public ingress/proxy:
+
+- Keep the **pre-upgrade HTTP header/request timeout short** and apply
+  connection/rate limits before traffic reaches the sidecar. The sidecar also
+  closes incomplete HTTP headers after 10 seconds, but the public proxy is the
+  first resource-exhaustion boundary.
+- Set **post-upgrade WebSocket read/send/idle timeouts greater than 140
+  seconds** (for example, 150 or 180 seconds). This lets Desktop own the
+  user-visible 130-second expiry and keeps the sidecar's 140-second timeout as
+  the orphaned-client safety net.
+
+For ingress-nginx, set `nginx.ingress.kubernetes.io/proxy-read-timeout` and
+`nginx.ingress.kubernetes.io/proxy-send-timeout` to `150` or higher on the
+pairing Ingress while keeping the controller's `client-header-timeout` short.
+For an AWS ALB, set `idle_timeout.timeout_seconds` above 140; for a Google
+Cloud BackendService, set `timeoutSec` above 140. Do not raise the ordinary
+HTTP header/request timeout to match the WebSocket lifetime.
+
 ## HA (production)
 
 `replicaCount > 1` hard-requires Redis: