From 8337378c6aaf233bfdb2e48e924b89c92f7a02c8 Mon Sep 17 00:00:00 2001 From: "detail-app[bot]" <180357370+detail-app[bot]@users.noreply.github.com> Date: Wed, 14 Oct 2026 12:33:03 +0000 Subject: [PATCH 1/2] fix(h2): clamp driver poll to armed stall deadline for indefinite timeouts --- crates/bemo/src/abi/http.rs | 77 ++++++++++++ crates/bemo/src/abi/http_h2.rs | 19 +++ crates/bemo/src/abi/socket.rs | 41 ++++++- crates/bemo/tests/http.rs | 211 +++++++++++++++++++++++++++++++++ 4 files changed, 344 insertions(+), 4 deletions(-) diff --git a/crates/bemo/src/abi/http.rs b/crates/bemo/src/abi/http.rs index f5e5384..5f44e72 100644 --- a/crates/bemo/src/abi/http.rs +++ b/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 { + 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) { + let stalled: Vec = 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) { @@ -2810,4 +2846,45 @@ 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] + 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); + } } diff --git a/crates/bemo/src/abi/http_h2.rs b/crates/bemo/src/abi/http_h2.rs index 62a6a23..510dbd7 100644 --- a/crates/bemo/src/abi/http_h2.rs +++ b/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 { + 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(), diff --git a/crates/bemo/src/abi/socket.rs b/crates/bemo/src/abi/socket.rs index 2cd0d68..b010e81 100644 --- a/crates/bemo/src/abi/socket.rs +++ b/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 }); diff --git a/crates/bemo/tests/http.rs b/crates/bemo/tests/http.rs index 49d9a54..d2ffaa4 100644 --- a/crates/bemo/tests/http.rs +++ b/crates/bemo/tests/http.rs @@ -3107,6 +3107,217 @@ fn native_h2_free_on_flow_control_stalled_stream_force_resets_and_closes() { assert_eq!(elide_transport_owner_release(h.owner), 0); } +/// A freed stream stalled by withheld flow-control credit must be force-reset within the bounded +/// `FLOW_CONTROL_STALL` window even when the embedder polls with the documented indefinite timeout +/// (`u64::MAX`). The driver clamps its internal wait to the earliest armed deadline so the lazy +/// `Instant::now()` stall check in `drive` is evaluated without relying on host tick frequency or +/// an external wake. Without the fix, the single `poll(u64::MAX)` blocks indefinitely (until I/O +/// or an explicit `driverWake`) and the 5 s deadline is never evaluated — a regression watchdog +/// `elide_transport_driver_wake` caps the test at 8 s so it fails fast instead of hanging. +#[test] +#[cfg_attr(miri, ignore = "io_uring and real sockets are unavailable under miri")] +fn native_h2_free_flow_control_stalled_stream_resets_under_indefinite_poll() { + use rustls::pki_types::ServerName; + use std::sync::Arc; + let (mut h, config) = tls_fixture_with_protocol(&[&rustls::version::TLS13], b"h2"); + let peer = h.peer.try_clone().unwrap(); + // The stall detector force-resets after `FLOW_CONTROL_STALL` (5 s). The cloned peer inherits the + // harness's 5 s read timeout, which races the detector and loses on slower platforms; allow the + // reset and TLS close to land before the client's read times out. + peer.set_read_timeout(Some(Duration::from_secs(30))).unwrap(); + let (closed, observed) = std::sync::mpsc::channel(); + let body = vec![b'x'; 200 * 1024]; + let body_len = body.len(); + let client = std::thread::spawn(move || { + let client = rustls::ClientConnection::new(Arc::new(config), ServerName::try_from("localhost").unwrap()).unwrap(); + let mut stream = rustls::StreamOwned::new(client, peer); + stream.write_all(b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n").unwrap(); + stream.write_all(&[0, 0, 0, 4, 0, 0, 0, 0, 0]).unwrap(); + // Static HPACK entries GET, https, /; literal :authority localhost. + let mut head = vec![0x82, 0x87, 0x84, 0x01, 9]; + head.extend_from_slice(b"localhost"); + stream.write_all(&[0, 0, head.len() as u8, 1, 5, 0, 0, 0, 1]).unwrap(); + stream.write_all(&head).unwrap(); + // Half-close the client side so the server observes peer_closed; then withhold all + // flow-control credit by never sending a WINDOW_UPDATE and just drain the response. + stream.conn.send_close_notify(); + stream.flush().unwrap(); + closed.send(()).unwrap(); + let mut reply = Vec::new(); + stream.read_to_end(&mut reply).unwrap(); + // The initial flow-control window flushes; the stall detector force-resets the remainder. + let mut data = 0usize; + let mut reset = false; + let mut offset = 0; + while offset < reply.len() { + assert!(reply.len() - offset >= 9, "trailing frame header"); + let frame = &reply[offset..]; + let length = ((frame[0] as usize) << 16) | ((frame[1] as usize) << 8) | frame[2] as usize; + assert!(frame.len() >= 9 + length); + match frame[3] { + 0 if frame[5..9] == [0, 0, 0, 1] => data += length, + 3 if frame[5..9] == [0, 0, 0, 1] => reset = true, + _ => {} + } + offset += 9 + length; + } + assert!(data > 0, "the initial flow-control window must flush"); + assert!(data < body_len, "the body must not fully flush under a withheld window"); + assert!(reset, "the stalled stream must be force-reset by the stall detector"); + }); + let request = h.take(EVENT_REQUEST, 1)[0].value; + observed.recv_timeout(Duration::from_secs(5)).unwrap(); + // Authenticate the half-close so the server observes peer_closed before the response stalls. + h.poll(100_000_000); + h.poll(100_000_000); + assert!(h.pending.iter().all(|event| event.kind != EVENT_CLOSED)); + respond(h.driver, request, 200, &[], &body); + assert_eq!(elide_transport_http_free(h.driver, request), 0); + // Immediately after freeing, poll with a short timeout to confirm no premature close. + h.poll(50_000_000); + assert!( + h.pending.iter().all(|event| event.kind != EVENT_CLOSED), + "a freshly-stalled stream must not close before the stall timeout" + ); + // The regression watchdog: if the fix is broken, the indefinite poll blocks forever. Wake the + // driver after 8 s so the test fails with an assertion rather than a hang. A passing fix wakes + // the driver internally at the 5 s deadline, long before the watchdog fires. + let watchdog = { + let driver = h.driver; + std::thread::spawn(move || { + std::thread::sleep(Duration::from_secs(8)); + elide_transport_driver_wake(driver); + }) + }; + // Poll indefinitely (u64::MAX). The fix clamps the internal wait to FLOW_CONTROL_STALL so + // the stall detector fires without any external wake. + let freed_at = Instant::now(); + h.poll(u64::MAX); + let reset_elapsed = freed_at.elapsed(); + // The stall deadline is 5 s; allow scheduling overhead. The watchdog at 8 s must not fire. + assert!( + reset_elapsed < Duration::from_secs(7), + "indefinite poll must wake at the ~5s stall deadline, not block indefinitely (elapsed {:?})", + reset_elapsed + ); + // The stall detector resets the stream in the poll above, but the RST_STREAM/TLS close must + // drain through the wire before the peer-closed connection emits EVENT_CLOSED. A bounded loop + // completes that cleanup; the deadline proof is that the indefinite poll returned at ~5 s. + let deadline = Instant::now() + Duration::from_secs(10); + loop { + if h.pending.iter().any(|event| event.kind == EVENT_CLOSED) { + break; + } + assert!( + Instant::now() < deadline, + "EVENT_CLOSED never fired; stalled stream was not retired by the stall detector" + ); + h.poll(200_000_000); + } + watchdog.join().unwrap(); + h.take(EVENT_CLOSED, 1); + client.join().unwrap(); + assert_eq!(elide_transport_driver_release(h.driver), 0); + assert_eq!(elide_transport_buffer_release(h.batch), 0); + assert_eq!(elide_transport_owner_used(h.owner), 0); + assert_eq!(elide_transport_owner_release(h.owner), 0); +} + +/// A cooperative h2 peer (one that sends WINDOW_UPDATE and lets the response reach the wire) +/// retires normally under an indefinite poll — the stall detector is NOT armed and the deadline +/// clamp does not cause a busy-loop. The connection returns to idle blocking after retirement. +#[test] +#[cfg_attr(miri, ignore = "io_uring and real sockets are unavailable under miri")] +fn native_h2_cooperative_peer_retires_normally_under_indefinite_poll() { + use rustls::pki_types::ServerName; + use std::sync::Arc; + let (mut h, config) = tls_fixture_with_protocol(&[&rustls::version::TLS13], b"h2"); + let peer = h.peer.try_clone().unwrap(); + peer.set_read_timeout(Some(Duration::from_secs(30))).unwrap(); + let (closed, observed) = std::sync::mpsc::channel(); + let client = std::thread::spawn(move || { + let client = rustls::ClientConnection::new(Arc::new(config), ServerName::try_from("localhost").unwrap()).unwrap(); + let mut stream = rustls::StreamOwned::new(client, peer); + stream.write_all(b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n").unwrap(); + stream.write_all(&[0, 0, 0, 4, 0, 0, 0, 0, 0]).unwrap(); + // Static HPACK entries GET, https, /; literal :authority localhost. + let mut head = vec![0x82, 0x87, 0x84, 0x01, 9]; + head.extend_from_slice(b"localhost"); + stream.write_all(&[0, 0, head.len() as u8, 1, 5, 0, 0, 0, 1]).unwrap(); + stream.write_all(&head).unwrap(); + // Half-close the client side so the server observes peer_closed; the small body + // fits the initial flow-control window so the response reaches the wire cooperatively. + stream.conn.send_close_notify(); + stream.flush().unwrap(); + closed.send(()).unwrap(); + let mut reply = Vec::new(); + stream.read_to_end(&mut reply).unwrap(); + // Verify we got a DATA frame with the body and a RST_STREAM is NOT present (cooperative). + let mut data = 0usize; + let mut reset = false; + let mut offset = 0; + while offset < reply.len() { + assert!(reply.len() - offset >= 9, "trailing frame header"); + let frame = &reply[offset..]; + let length = ((frame[0] as usize) << 16) | ((frame[1] as usize) << 8) | frame[2] as usize; + assert!(frame.len() >= 9 + length); + match frame[3] { + 0 if frame[5..9] == [0, 0, 0, 1] => data += length, + 3 if frame[5..9] == [0, 0, 0, 1] => reset = true, + _ => {} + } + offset += 9 + length; + } + assert!(data > 0, "the cooperative body must flush"); + assert!(!reset, "a cooperative peer must not see a RST_STREAM"); + }); + let request = h.take(EVENT_REQUEST, 1)[0].value; + observed.recv_timeout(Duration::from_secs(5)).unwrap(); + h.poll(100_000_000); + // Respond with a small body that fits the initial flow-control window. + respond(h.driver, request, 200, &[], b"cooperative"); + assert_eq!(elide_transport_http_free(h.driver, request), 0); + // Poll with bounded timeouts to let the response flush and the stream retire normally. + for _ in 0..10 { + h.poll(100_000_000); + } + // The stream has retired normally — no stall detector armed. An indefinite poll with no + // armed deadline and no pending I/O must block indefinitely (G6: no busy-loop). A watchdog + // wakes the driver after 3 s to prove the poll was truly blocking, not spinning. + let watchdog = { + let driver = h.driver; + std::thread::spawn(move || { + std::thread::sleep(Duration::from_secs(3)); + elide_transport_driver_wake(driver); + }) + }; + let poll_start = Instant::now(); + let _count = h.poll(u64::MAX); + let poll_elapsed = poll_start.elapsed(); + // The poll must have blocked for ~3 s (until the watchdog woke it), proving no busy-loop. + assert!( + poll_elapsed >= Duration::from_secs(2), + "indefinite poll without armed deadline must block, not busy-loop (elapsed {:?})", + poll_elapsed + ); + watchdog.join().unwrap(); + // Drain any remaining events and close. + let deadline = Instant::now() + Duration::from_secs(10); + loop { + if h.pending.iter().any(|event| event.kind == EVENT_CLOSED) { + break; + } + assert!(Instant::now() < deadline, "EVENT_CLOSED never fired"); + h.poll(200_000_000); + } + h.take(EVENT_CLOSED, 1); + client.join().unwrap(); + assert_eq!(elide_transport_driver_release(h.driver), 0); + assert_eq!(elide_transport_buffer_release(h.batch), 0); + assert_eq!(elide_transport_owner_used(h.owner), 0); + assert_eq!(elide_transport_owner_release(h.owner), 0); +} + #[test] #[cfg_attr(miri, ignore = "io_uring and real sockets are unavailable under miri")] fn parsed_header_metadata_is_advertised_and_preserves_first_host() { From e3b85f349dc19394af69ff3d85b6212e97ea6bff Mon Sep 17 00:00:00 2001 From: "detail-app[bot]" <180357370+detail-app[bot]@users.noreply.github.com> Date: Wed, 14 Oct 2026 13:41:04 +0000 Subject: [PATCH 2/2] fix(h2): ignore next_deadline driver test under miri The new next_deadline test opens a real driver via elide_transport_driver_new, which builds an io_uring proactor unsupported under miri. Add the same cfg_attr(miri, ignore) gate that every other driver-opening test in this module already carries, per crates/bemo/AGENTS.md. --- crates/bemo/src/abi/http.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/crates/bemo/src/abi/http.rs b/crates/bemo/src/abi/http.rs index 5f44e72..bf5d2e2 100644 --- a/crates/bemo/src/abi/http.rs +++ b/crates/bemo/src/abi/http.rs @@ -2851,6 +2851,7 @@ mod tests { /// 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.