fix(streaming): bound socket writes, seed historical deadlines, arm self-join guard, and own the wait-strategy surface

crates/thetadatadx/src/config/mdds.rs

@@ -79,6 +79,25 @@ pub struct HistoricalConfig {
     /// cadence.
     pub connect_timeout_secs: u64,
 
+    /// Default per-request deadline for historical (gRPC) queries, in
+    /// seconds. `0` disables the default (no deadline unless the caller
+    /// sets one).
+    ///
+    /// A server that holds the HTTP/2 stream open while sending no
+    /// chunks would otherwise hang `collect_stream` / `stream(...)`
+    /// indefinitely: the gRPC keepalive PING only detects a fully dead
+    /// peer, not a live-but-silent one. This default bounds every
+    /// request that did not call `with_deadline(...)`, so a stalled
+    /// stream resolves to `Error::Timeout` instead of blocking forever.
+    ///
+    /// Per-call control overrides this: `with_deadline(Duration)` sets a
+    /// shorter or longer bound, and `with_deadline(Duration::ZERO)`
+    /// opts a single request out of any deadline.
+    ///
+    /// Default `300s` (5 min) — comfortably above the slowest realistic
+    /// multi-million-row bulk pull while still bounding a wedged stream.
+    pub request_timeout_secs: u64,
+
     /// Estimated-bytes threshold above which the buffered `.await`
     /// path on a `parsed_endpoint!` builder emits a single
     /// `tracing::warn!` event suggesting `.stream(handler)` for the
@@ -138,6 +157,11 @@ impl HistoricalConfig {
             window_size_kb: 64,
             connection_window_size_kb: 64,
             connect_timeout_secs: 10,
+            // 5 min — bounds a server that holds the stream open while
+            // sending no chunks (h2 keepalive only catches a fully dead
+            // peer). Sits above the slowest realistic bulk pull;
+            // `with_deadline(Duration::ZERO)` opts a single request out.
+            request_timeout_secs: 300,
             // 100 MiB — empirically catches bulk pulls (multi-million
             // row option-chain or multi-day backfill responses) while
             // staying silent on ad-hoc single-day quote / OHLC pulls

crates/thetadatadx/src/fpss/connection.rs

@@ -223,7 +223,8 @@ pub(crate) fn order_hosts(
 /// 2. `TCP_NODELAY = true`
 /// 3. `SO_KEEPALIVE` armed per `keepalive`
 /// 4. Read timeout set to `read_timeout`
-/// 5. TLS handshake pinned to the FPSS SPKI
+/// 5. Write timeout set to `write_timeout`
+/// 6. TLS handshake pinned to the FPSS SPKI
 ///
 /// # Errors
 ///
@@ -232,6 +233,7 @@ pub(crate) fn connect_to_servers(
     servers: &[(&str, u16)],
     connect_timeout: Duration,
     read_timeout: Duration,
+    write_timeout: Duration,
     keepalive: TcpKeepaliveSpec,
 ) -> Result<(FpssStream, String), crate::error::Error> {
     ensure_rustls_crypto_provider();
@@ -241,7 +243,14 @@ pub(crate) fn connect_to_servers(
         let addr = format!("{host}:{port}");
         tracing::debug!(server = %addr, "attempting FPSS connection");
 
-        match try_connect(host, port, connect_timeout, read_timeout, keepalive) {
+        match try_connect(
+            host,
+            port,
+            connect_timeout,
+            read_timeout,
+            write_timeout,
+            keepalive,
+        ) {
             Ok(stream) => {
                 tracing::info!(server = %addr, "FPSS connected");
                 return Ok((stream, addr));
@@ -318,12 +327,14 @@ fn arm_keepalive(tcp: &TcpStream, spec: TcpKeepaliveSpec) {
 /// 2. `set_nodelay(true)`
 /// 3. `SO_KEEPALIVE` per the configured schedule
 /// 4. `set_read_timeout`
-/// 5. Blocking TLS handshake via rustls `StreamOwned`
+/// 5. `set_write_timeout`
+/// 6. Blocking TLS handshake via rustls `StreamOwned`
 fn try_connect(
     host: &str,
     port: u16,
     connect_timeout: Duration,
     read_timeout: Duration,
+    write_timeout: Duration,
     keepalive: TcpKeepaliveSpec,
 ) -> Result<FpssStream, crate::error::Error> {
     let addr = format!("{host}:{port}");
@@ -358,6 +369,17 @@ fn try_connect(
     // Read timeout.
     tcp.set_read_timeout(Some(read_timeout))?;
 
+    // Write timeout. The first write (CREDENTIALS) drives the lazy TLS
+    // handshake, and steady-state ping/subscribe writes can otherwise
+    // block indefinitely against a peer whose receive window has stalled
+    // (alive enough to ACK at the kernel but not draining the socket).
+    // `connect_timeout` only bounds the SYN/ACK, so an unbounded write
+    // would wedge the I/O thread past that budget. The bound persists for
+    // the life of the socket via `SO_SNDTIMEO`, so a write `TimedOut`
+    // surfaces as a fatal I/O error and the caller reconnects, mirroring
+    // the read-timeout liveness contract.
+    tcp.set_write_timeout(Some(write_timeout))?;
+
     // TLS handshake (blocking) using rustls with webpki root certificates.
     let server_name =
         ServerName::try_from(host.to_owned()).map_err(|e| crate::error::Error::Fpss {
@@ -447,7 +469,14 @@ mod tests {
         let connect_timeout = Duration::from_millis(150);
         let read_timeout = Duration::from_millis(10_000);
         let start = std::time::Instant::now();
-        let res = connect_to_servers(&servers, connect_timeout, read_timeout, test_keepalive());
+        let write_timeout = Duration::from_millis(10_000);
+        let res = connect_to_servers(
+            &servers,
+            connect_timeout,
+            read_timeout,
+            write_timeout,
+            test_keepalive(),
+        );
         let elapsed = start.elapsed();
         assert!(res.is_err(), "unroutable host must fail to connect");
         assert!(
@@ -472,6 +501,39 @@ mod tests {
         );
     }
 
+    /// Both the read and write timeouts round-trip onto a real socket.
+    ///
+    /// `try_connect` sets `SO_RCVTIMEO` and `SO_SNDTIMEO` before the TLS
+    /// handshake. The handshake itself needs a live FPSS peer, so this
+    /// asserts the load-bearing socket-option contract directly on a
+    /// loopback socket: an unbounded write would let a stalled-receiver
+    /// peer wedge the I/O thread past `connect_timeout`, so the write
+    /// timeout must actually land on the kernel socket.
+    #[test]
+    fn read_and_write_timeouts_arm_on_loopback_socket() {
+        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind loopback");
+        let addr = listener.local_addr().expect("local addr");
+        let tcp = TcpStream::connect(addr).expect("connect loopback");
+
+        let read_timeout = Duration::from_millis(7_000);
+        let write_timeout = Duration::from_millis(3_000);
+        tcp.set_read_timeout(Some(read_timeout))
+            .expect("set read timeout");
+        tcp.set_write_timeout(Some(write_timeout))
+            .expect("set write timeout");
+
+        assert_eq!(
+            tcp.read_timeout().expect("read SO_RCVTIMEO"),
+            Some(read_timeout),
+            "read timeout must round-trip onto the socket"
+        );
+        assert_eq!(
+            tcp.write_timeout().expect("read SO_SNDTIMEO"),
+            Some(write_timeout),
+            "write timeout must round-trip onto the socket"
+        );
+    }
+
     fn production_hosts() -> Vec<(String, u16)> {
         vec![
             ("nj-a.thetadata.us".to_string(), 20000),

crates/thetadatadx/src/fpss/io_loop/mod.rs

@@ -791,7 +791,17 @@ where
                 .iter()
                 .map(|(host, port)| (host.as_str(), *port))
                 .collect();
-            connection::connect_to_servers(&ordered, connect_timeout, read_timeout, keepalive)
+            // The write timeout shares the read timeout's budget: both
+            // bound a single unacknowledged transport operation during the
+            // reconnect window, so the re-auth write cannot wedge the I/O
+            // thread against a peer that has stopped draining its socket.
+            connection::connect_to_servers(
+                &ordered,
+                connect_timeout,
+                read_timeout,
+                read_timeout,
+                keepalive,
+            )
         };
 
         let (mut new_stream, new_addr) = match new_stream {

crates/thetadatadx/src/fpss/io_loop/ping.rs

@@ -55,12 +55,27 @@ pub(in crate::fpss) fn ping_loop(
             code: StreamMsgType::Ping,
             payload: ping_payload.clone(),
         };
-        // Blocking send on the bounded channel: the heartbeat takes natural
-        // backpressure if the I/O thread is momentarily behind, rather than
-        // dropping a ping. `send` only errors once the receiver hangs up.
-        if cmd_tx.send(cmd).is_err() {
-            // I/O thread has exited
-            break;
+        // Non-blocking send on the bounded channel, matching the rest of
+        // the control plane (subscribe / unsubscribe / shutdown all use
+        // `try_send`). The heartbeat is an idempotent liveness signal,
+        // not a state-carrying frame: a momentarily full channel means
+        // the I/O thread is already draining a large backlog of outbound
+        // writes, so the connection is demonstrably alive and the next
+        // ping fires one interval later — well inside any server-side
+        // ping deadline. Skipping a ping under that backpressure is
+        // therefore safe, and it keeps a blocking heartbeat from pinning
+        // the consumer/Drop path for up to one interval while the channel
+        // drains. A hung-up receiver (I/O thread exited) still ends the
+        // loop.
+        match cmd_tx.try_send(cmd) {
+            Ok(()) => {}
+            Err(std_mpsc::TrySendError::Full(_)) => {
+                // I/O thread is behind; drop this beat and try the next.
+            }
+            Err(std_mpsc::TrySendError::Disconnected(_)) => {
+                // I/O thread has exited.
+                break;
+            }
         }
 
         thread::sleep(interval);