From c9d8872d9cac9a09c004c1e59cf1775f3f7998fc Mon Sep 17 00:00:00 2001 From: "detail-app[bot]" <180357370+detail-app[bot]@users.noreply.github.com> Date: Fri, 16 Oct 2026 23:43:51 +0000 Subject: [PATCH] fix(http): route HTTP-mode sockets through close_socket on force close --- crates/bemo/src/abi/socket.rs | 6 + crates/bemo/tests/http.rs | 217 ++++++++++++++++++++++++++++++++++ 2 files changed, 223 insertions(+) diff --git a/crates/bemo/src/abi/socket.rs b/crates/bemo/src/abi/socket.rs index 2cd0d68..0684bc2 100644 --- a/crates/bemo/src/abi/socket.rs +++ b/crates/bemo/src/abi/socket.rs @@ -656,6 +656,12 @@ pub fn elide_transport_socket_close(driver: u64, socket: u64) -> i32 { let Some(state) = drivers.0.get_mut(&driver) else { return INVALID; }; + if super::http::is_http(state, socket) { + let mut events = std::collections::VecDeque::new(); + super::http::close_socket(state, socket, &mut events); + state.http.overflow.extend(events); + return 0; + } let Some(connection) = state.sockets.remove(&socket) else { return INVALID; }; diff --git a/crates/bemo/tests/http.rs b/crates/bemo/tests/http.rs index 49d9a54..8f6832c 100644 --- a/crates/bemo/tests/http.rs +++ b/crates/bemo/tests/http.rs @@ -3834,3 +3834,220 @@ fn native_h2_free_before_responding_cancels_the_stream() { ] ); } + +/// A parked HTTP socket (receive lane idle, no pending operation) must still tear down fully +/// when the consumer force-closes it: `EVENT_CLOSED` is delivered, `state.http.sockets` is +/// emptied, and the owner budget drains. This is the core bug — on epoll and io_uring with +/// non-default capacity, the raw close used to leave the HTTP entry behind with no completion +/// to drive teardown. +#[test] +#[cfg_attr(miri, ignore = "io_uring and real sockets are unavailable under miri")] +fn parked_http_socket_close_emits_closed_and_cleans_up() { + let mut h = harness(); + // Send a POST body larger than BODY_WINDOW_BYTES and never ack the segments, parking the + // receive lane with no pending operation in state.operations. + let body: Vec = (0..BODY_WINDOW_BYTES * 4).map(|i| (i % 251) as u8).collect(); + let head = format!( + "POST /big HTTP/1.1\r\nHost: a\r\nContent-Length: {}\r\n\r\n", + body.len() + ); + let mut peer = h.peer.try_clone().unwrap(); + let payload = body.clone(); + let writer = std::thread::spawn(move || { + peer.write_all(head.as_bytes()).unwrap(); + let _ = peer.write_all(&payload); + }); + let _exchange = h.take(EVENT_REQUEST, 1)[0].value; + let held = h.drain_segments(); + let delivered: usize = held.iter().map(|s| s.result as usize).sum(); + assert!(delivered < body.len(), "window did not stall: {delivered}"); + // The lane is now parked: confirm no events arrive. + for _ in 0..4 { + assert_eq!(h.poll(50_000_000), 0, "receives must stay parked"); + } + // Force-close the socket while the lane is parked. + assert_eq!(elide_transport_socket_close(h.driver, h.socket), 0); + // EVENT_CLOSED must be delivered on the next poll. + let closed = h.take(EVENT_CLOSED, 1); + assert_eq!(closed.len(), 1); + assert_eq!(closed[0].socket, h.socket); + // No more events for this socket. + for _ in 0..4 { + assert_eq!(h.poll(50_000_000), 0, "no spurious events after close"); + } + writer.join().unwrap(); + h.release_segments(&held); + assert_eq!(elide_transport_buffer_release(h.batch), 0); + assert_eq!(elide_transport_driver_release(h.driver), 0); + assert_eq!(elide_transport_owner_used(h.owner), 0); + assert_eq!(elide_transport_owner_release(h.owner), 0); +} + +/// Force-closing an HTTP socket with queued streaming response parts must fail each unsent part +/// with an `EVENT_PART_SENT` error event, then emit `EVENT_CLOSED`. Before the fix, the parts +/// were silently dropped on non-io_uring backends. +#[test] +#[cfg_attr(miri, ignore = "io_uring and real sockets are unavailable under miri")] +fn parked_http_socket_close_fails_queued_streaming_parts() { + let mut h = harness(); + h.peer.write_all(b"GET /stream HTTP/1.1\r\nHost: a\r\n\r\n").unwrap(); + let exchange = h.take(EVENT_REQUEST, 1)[0].value; + // Start a streaming response and queue two parts without a final. + respond_stream(h.driver, exchange, 200, &[], u64::MAX); + send_chunk(h.driver, exchange, b"part1", false); + send_chunk(h.driver, exchange, b"part2", false); + // Force-close while parts are queued. Some parts may have already been sent successfully + // (the head), but any unsent streaming parts must be failed with EVENT_PART_SENT errors. + assert_eq!(elide_transport_socket_close(h.driver, h.socket), 0); + // Collect all events: at least one EVENT_PART_SENT must carry a negative (error) result, + // proving queued parts were failed rather than silently dropped. + let mut saw_failed_part = false; + for _ in 0..64 { + h.poll(50_000_000); + for ev in &h.pending { + if ev.kind == EVENT_PART_SENT && ev.value == exchange && ev.result < 0 { + saw_failed_part = true; + } + } + if h.pending.iter().any(|e| e.kind == EVENT_CLOSED) { + break; + } + } + assert!( + saw_failed_part, + "at least one queued streaming part must be failed with EVENT_PART_SENT(error): {:?}", + h.pending + ); + // EVENT_CLOSED must be delivered. + let closed = h.take(EVENT_CLOSED, 1); + assert_eq!(closed.len(), 1); + assert_eq!(closed[0].socket, h.socket); + assert_eq!(elide_transport_buffer_release(h.batch), 0); + assert_eq!(elide_transport_driver_release(h.driver), 0); + assert_eq!(elide_transport_owner_used(h.owner), 0); + assert_eq!(elide_transport_owner_release(h.owner), 0); +} + +/// A non-HTTP socket close must follow the raw path and return 0; no HTTP state is touched. +#[test] +#[cfg_attr(miri, ignore = "io_uring and real sockets are unavailable under miri")] +fn non_http_socket_close_is_unaffected() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let owner = elide_transport_owner_new(1024 * 1024); + let driver = elide_transport_driver_new(common::workload(), common::backend() as u32, 16); + let endpoint = endpoint(owner, listener.local_addr().unwrap()); + let socket = elide_transport_socket_connect(common::workload(), driver, endpoint); + assert_ne!(socket, 0); + let batch = elide_transport_buffer_new(owner, 8 * size_of::() as u64); + let accepted = std::thread::spawn(move || listener.accept().unwrap().0); + assert_eq!(events(driver, batch, 1, 1)[0].2, 0); + let _peer = accepted.join().unwrap(); + // Close without activating HTTP mode — must return 0. + assert_eq!(elide_transport_socket_close(driver, socket), 0); + // No HTTP state was ever created. + assert_eq!(elide_transport_buffer_release(endpoint), 0); + assert_eq!(elide_transport_buffer_release(batch), 0); + assert_eq!(elide_transport_driver_release(driver), 0); + assert_eq!(elide_transport_owner_used(owner), 0); + assert_eq!(elide_transport_owner_release(owner), 0); +} + +/// Double close of an HTTP socket: the first returns 0, the second returns INVALID. +#[test] +#[cfg_attr(miri, ignore = "io_uring and real sockets are unavailable under miri")] +fn double_close_of_http_socket_returns_invalid() { + let mut h = harness(); + h.peer.write_all(b"GET / HTTP/1.1\r\nHost: a\r\n\r\n").unwrap(); + let _ = h.take(EVENT_REQUEST, 1); + assert_eq!(elide_transport_socket_close(h.driver, h.socket), 0); + h.take(EVENT_CLOSED, 1); + // Second close must return INVALID — the socket is already gone. + assert_eq!(elide_transport_socket_close(h.driver, h.socket), INVALID); + assert_eq!(elide_transport_buffer_release(h.batch), 0); + assert_eq!(elide_transport_driver_release(h.driver), 0); + assert_eq!(elide_transport_owner_used(h.owner), 0); + assert_eq!(elide_transport_owner_release(h.owner), 0); +} + +/// Force-closing an HTTP socket with an in-flight persistent receive (io_uring default capacity) +/// must emit exactly one EVENT_CLOSED — not zero, not two. This guards against double-close +/// regressions on the path where the cancellation completion also routes through close_socket. +#[test] +#[cfg_attr(miri, ignore = "io_uring and real sockets are unavailable under miri")] +fn close_http_socket_with_inflight_receive_emits_one_closed() { + let mut h = harness(); + h.peer.write_all(b"GET / HTTP/1.1\r\nHost: a\r\n\r\n").unwrap(); + let _ = h.take(EVENT_REQUEST, 1); + // The receive lane is in flight (persistent on io_uring default, one-shot elsewhere). + assert_eq!(elide_transport_socket_close(h.driver, h.socket), 0); + let closed = h.take(EVENT_CLOSED, 1); + assert_eq!(closed.len(), 1, "exactly one EVENT_CLOSED expected"); + assert_eq!(closed[0].socket, h.socket); + // No second EVENT_CLOSED. + for _ in 0..4 { + let n = h.poll(50_000_000); + assert_eq!(n, 0, "no spurious events after close"); + } + assert!( + !h.pending.iter().any(|e| e.kind == EVENT_CLOSED), + "no duplicate EVENT_CLOSED" + ); + assert_eq!(elide_transport_buffer_release(h.batch), 0); + assert_eq!(elide_transport_driver_release(h.driver), 0); + assert_eq!(elide_transport_owner_used(h.owner), 0); + assert_eq!(elide_transport_owner_release(h.owner), 0); +} + +/// After force-closing a parked HTTP socket, acking held segments must not resurrect the +/// socket or produce new HTTP request/body events — the zombie cannot self-heal because the +/// socket is already torn down. +#[test] +#[cfg_attr(miri, ignore = "io_uring and real sockets are unavailable under miri")] +fn ack_after_parked_close_produces_no_events() { + let mut h = harness(); + let body: Vec = (0..BODY_WINDOW_BYTES * 4).map(|i| (i % 251) as u8).collect(); + let head = format!( + "POST /big HTTP/1.1\r\nHost: a\r\nContent-Length: {}\r\n\r\n", + body.len() + ); + let mut peer = h.peer.try_clone().unwrap(); + let writer = std::thread::spawn(move || { + peer.write_all(head.as_bytes()).unwrap(); + let _ = peer.write_all(&body); + }); + let _exchange = h.take(EVENT_REQUEST, 1)[0].value; + let held = h.drain_segments(); + // Park confirmed. + for _ in 0..4 { + assert_eq!(h.poll(50_000_000), 0); + } + // Force-close. + assert_eq!(elide_transport_socket_close(h.driver, h.socket), 0); + h.take(EVENT_CLOSED, 1); + // Ack all held segments — this must not produce any new HTTP request/body events for the + // closed socket. The socket is gone from state.http.sockets, so arm_receive cannot re-arm. + h.ack_segments(&held); + for _ in 0..8 { + h.poll(50_000_000); + } + // No new HTTP request or body events for the closed socket. + assert!( + !h.pending + .iter() + .any(|e| e.kind == EVENT_REQUEST || e.kind == EVENT_BODY), + "acking after close must not produce HTTP request/body events: {:?}", + h.pending + ); + // No second EVENT_CLOSED. + assert!( + !h.pending.iter().any(|e| e.kind == EVENT_CLOSED), + "no duplicate EVENT_CLOSED from acking: {:?}", + h.pending + ); + writer.join().unwrap(); + h.release_segments(&held); + assert_eq!(elide_transport_buffer_release(h.batch), 0); + assert_eq!(elide_transport_driver_release(h.driver), 0); + assert_eq!(elide_transport_owner_used(h.owner), 0); + assert_eq!(elide_transport_owner_release(h.owner), 0); +}