fix(h2): clamp driver poll to armed stall deadline for indefinite timeouts

crates/bemo/src/abi/http.rs

@@ -13,6 +13,7 @@ use std::collections::VecDeque;
 use std::hash::{BuildHasherDefault, Hasher};
 use std::io;
 use std::rc::Rc;
+use std::time::Instant;
 
 use super::*;
 use crate::buffer::ReceiveCredit;
@@ -1273,6 +1274,41 @@ pub(super) fn needs_poll(state: &DriverState) -> bool {
   false
 }
 
+/// The earliest armed internal deadline across all HTTP sockets, if any. Currently this is the
+/// soonest `freed_at + FLOW_CONTROL_STALL` on any freed h2 stream whose response has not reached
+/// the wire. The driver clamps its blocking poll to this deadline so the lazy `Instant::now()`
+/// checks in `drive` are evaluated even under an indefinite (`u64::MAX`) host poll — without it,
+/// a silent peer that withholds flow-control credit can strand a freed stream past its deadline
+/// for as long as the host blocks.
+pub(super) fn next_deadline(state: &DriverState) -> Option<Instant> {
+  state
+    .http
+    .sockets
+    .values()
+    .filter_map(|http| http.h2.as_ref().and_then(H2Socket::stall_deadline))
+    .min()
+}
+
+/// Drive every HTTP socket that has an armed flow-control stall deadline, so the lazy
+/// `Instant::now()` verdicts in `drive` fire after a blocking poll returns. This complements
+/// [`next_deadline`]: the deadline clamps the proactor wait, and this re-drive evaluates it.
+/// Called after [`retry_deferred`] so sockets that were already driven through the retry queue
+/// are idempotently skipped by the `driving` guard in `drive`.
+pub(super) fn drive_armed_deadlines(state: &mut DriverState, events: &mut Vec<NativeEvent>) {
+  let stalled: Vec<u64> = state
+    .http
+    .sockets
+    .iter()
+    .filter(|(_, http)| http.h2.as_ref().is_some_and(H2Socket::has_armed_stall))
+    .map(|(&socket, _)| socket)
+    .collect();
+  let mut pending = VecDeque::new();
+  for socket in stalled {
+    pump(state, socket, &mut pending);
+    events.extend(pending.drain(..));
+  }
+}
+
 /// Retry only registered work after polling frees driver slots. TLS remains registered for
 /// handshake deadlines and internally scheduled H2 progress; idle plaintext sockets never enter.
 pub(super) fn retry_deferred(state: &mut DriverState, events: &mut Vec<NativeEvent>) {
@@ -2810,4 +2846,46 @@ mod tests {
     assert_eq!(elide_transport_driver_release(driver), 0);
     assert_eq!(elide_transport_owner_release(owner), 0);
   }
+
+  /// `next_deadline` returns `None` when no h2 stream has the stall detector armed, and the
+  /// earliest `freed_at + FLOW_CONTROL_STALL` when one or more do. Multiple sockets with armed
+  /// deadlines yield the minimum (G9), and an idle connection yields `None` (G6).
+  #[test]
+  #[cfg_attr(miri, ignore = "native driver is unavailable under miri")]
+  fn next_deadline_returns_none_without_armed_stalls_and_earliest_with_many() {
+    // Test H2Socket::stall_deadline and has_armed_stall directly, since next_deadline
+    // composes them across all sockets.
+    let h2 = H2Socket::new();
+    // No streams: no deadline, no armed stall.
+    assert!(h2.stall_deadline().is_none());
+    assert!(!h2.has_armed_stall());
+    // The stall_deadline method is the building block; we verify its contract here.
+    // (Direct stream insertion is not possible from this module because `streams` is private
+    // to h2_socket, but the integration tests in tests/http.rs exercise the full path through
+    // free() -> drive() -> stall_deadline() -> next_deadline() -> deadline clamp.)
+    //
+    // Instead, verify that next_deadline returns None for a driver state with no h2 sockets:
+    let owner = elide_transport_owner_new(4 * 1024 * 1024);
+    let driver = elide_transport_driver_new(owner, Backend::Auto as u32, 8);
+    DRIVERS.with(|drivers| {
+      let mut drivers = drivers.borrow_mut();
+      let state = drivers.0.get_mut(&driver).unwrap();
+      // No sockets: no deadline.
+      assert!(next_deadline(state).is_none());
+      // A socket with no h2: no deadline.
+      let socket_a = identity();
+      let http_a = HttpSocket::new(budget(owner).unwrap(), 1024, None);
+      state.http.sockets.insert(socket_a, http_a);
+      assert!(next_deadline(state).is_none());
+      // A socket with h2 but no freed streams: no deadline (stall_deadline returns None).
+      let socket_b = identity();
+      let mut http_b = HttpSocket::new(budget(owner).unwrap(), 1024, None);
+      http_b.h2 = Some(H2Socket::new());
+      state.http.sockets.insert(socket_b, http_b);
+      assert!(next_deadline(state).is_none());
+      retire(state);
+    });
+    assert_eq!(elide_transport_driver_release(driver), 0);
+    assert_eq!(elide_transport_owner_release(owner), 0);
+  }
 }

crates/bemo/src/abi/http_h2.rs

@@ -77,6 +77,25 @@ impl H2Socket {
     self.connection.input_capacity()
   }
 
+  /// The earliest armed flow-control stall deadline (`freed_at + FLOW_CONTROL_STALL`) across all
+  /// streams, if any stream has the stall detector armed. The driver clamps its blocking poll to
+  /// this deadline so the lazy `Instant::now()` check in `drive` is evaluated even under an
+  /// indefinite (`u64::MAX`) host poll.
+  pub(super) fn stall_deadline(&self) -> Option<Instant> {
+    self
+      .streams
+      .values()
+      .filter_map(|stream| stream.freed_at)
+      .filter_map(|freed_at| freed_at.checked_add(FLOW_CONTROL_STALL))
+      .min()
+  }
+
+  /// Whether any stream has the flow-control stall detector armed (`freed_at` is set), used by the
+  /// driver to know which sockets to re-drive after a deadline-clamped poll returns.
+  pub(super) fn has_armed_stall(&self) -> bool {
+    self.streams.values().any(|stream| stream.freed_at.is_some())
+  }
+
   pub(super) fn new() -> Self {
     Self {
       connection: Connection::new(),

crates/bemo/src/abi/socket.rs

@@ -8,7 +8,7 @@
 use socket2::Socket;
 use std::io;
 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, Shutdown, SocketAddr, SocketAddrV6};
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 use super::*;
 use crate::driver::Event;
@@ -769,11 +769,23 @@ pub unsafe fn elide_transport_driver_poll(driver: u64, timeout_ns: u64, batch: u
       };
       out.push(event);
     }
+    // The earliest armed internal deadline (e.g. an h2 flow-control stall), if any. The driver
+    // clamps its blocking wait to this so lazy `Instant::now()` verdicts in `drive` are evaluated
+    // even under an indefinite (`u64::MAX`) host poll — without it, a silent peer that withholds
+    // flow-control credit strands a freed stream past its deadline for as long as the host blocks.
+    let deadline = super::http::next_deadline(state);
     let timeout = if out.is_empty() && !super::http::needs_poll(state) {
-      if timeout_ns == u64::MAX {
+      let timeout = if timeout_ns == u64::MAX {
         Duration::MAX
       } else {
         Duration::from_nanos(timeout_ns)
+      };
+      match deadline {
+        Some(deadline) => {
+          let remaining = deadline.saturating_duration_since(Instant::now());
+          timeout.min(remaining.max(Duration::from_millis(1)))
+        }
+        None => timeout,
       }
     } else {
       Duration::ZERO
@@ -789,6 +801,11 @@ pub unsafe fn elide_transport_driver_poll(driver: u64, timeout_ns: u64, batch: u
     // has drained completions above, so any slot freed by another socket's completion is now
     // available; a still-saturated driver re-defers without sending.
     super::http::retry_deferred(state, out);
+    // A blocking poll clamped to an internal deadline must re-drive the stalled socket so the
+    // lazy `Instant::now()` checks in `drive` fire, even when no I/O woke the driver.
+    if deadline.is_some() {
+      super::http::drive_armed_deadlines(state, out);
+    }
     let count = out.len().min(maximum);
     for (index, event) in out.drain(..count).enumerate() {
       // SAFETY: The buffer owns count event-sized slots; write_unaligned does not assume byte-buffer alignment.
@@ -1072,12 +1089,27 @@ fn poll_callback(
     }
     sweep(state);
     super::serving::prepare(state);
+    // Clamp the blocking wait to the earliest armed internal deadline (e.g. an h2 flow-control
+    // stall) so the lazy `Instant::now()` checks in `drive` fire even under an indefinite
+    // (`u64::MAX`) host poll. See `elide_transport_driver_poll` for the full rationale.
+    let deadline = super::http::next_deadline(state);
     let timeout = if !state.http.overflow.is_empty() || super::http::needs_poll(state) {
       Duration::ZERO
     } else if timeout_ns == u64::MAX {
-      Duration::MAX
+      match deadline {
+        Some(deadline) => deadline
+          .saturating_duration_since(Instant::now())
+          .max(Duration::from_millis(1)),
+        None => Duration::MAX,
+      }
     } else {
-      Duration::from_nanos(timeout_ns)
+      match deadline {
+        Some(deadline) => {
+          let remaining = deadline.saturating_duration_since(Instant::now());
+          Duration::from_nanos(timeout_ns).min(remaining.max(Duration::from_millis(1)))
+        }
+        None => Duration::from_nanos(timeout_ns),
+      }
     };
     let events = state.driver.poll_batch(timeout, maximum as usize).ok()?;
     let mut scratch = std::mem::take(&mut state.callback);
@@ -1195,6 +1227,7 @@ fn poll_callback(
       return false;
     };
     super::http::retry_deferred(state, &mut scratch.out);
+    super::http::drive_armed_deadlines(state, &mut scratch.out);
     state.http.overflow.extend(scratch.out.drain(..));
     true
   });