From af000523ac25c2a650898cefe0c7d4972010925f Mon Sep 17 00:00:00 2001 From: vshengbro Date: Tue, 13 Oct 2026 08:41:26 +0800 Subject: [PATCH] fix(hyperlane): stop dropping pipelined requests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit HTTP pipelining was completely broken: a client that sent several requests in a single write received a response for only the first one, and every later request was silently discarded. `fill_http_from_stream` created a fresh `PooledReader` per request. A single socket read can deliver bytes belonging to the next pipelined request, and those bytes were held in the reader's `start..end` window. Dropping the reader returned the whole buffer to the thread-local pool via `mem::take`, discarding the read-ahead bytes. The next request took a recycled buffer with `start`/`end` reset to zero, so the read-ahead data was gone for good. `Stream` now carries the unconsumed bytes across requests: `leftover` plus its `start`/`end` offsets. They are seeded into the reader before parsing and stored back after the request is parsed. All `self` access happens before the reader borrows the socket or after it is dropped, so the mutable borrows of `Stream` and `TcpStream` never overlap. Measured on an 18-core Mac, release build, closed-loop keep-alive client (128 connections, 10s runs, three repetitions each): - pipelined depth 64: 0 QPS (32/32 connections failed) -> 863,653 QPS - pipelined depth 16: 0 QPS -> 836,190 QPS - pipelined depth 4: 0 QPS -> 523,048 QPS - serial (depth 1): 189,065 -> 189,812 QPS, unchanged in run variance The serial number is flat by design: this restores correctness rather than speeding up the single-request path. Adds a regression test that writes 8 pipelined requests on one connection and asserts all 8 responses come back. It fails against the unfixed parser with "expected 8 pipelined responses, got 1". 🤖 Generated with [Hermes](https://hermes-agent.nousresearch.com) --- core/tests/mod.rs | 8 +++- core/tests/server/fn.rs | 84 +++++++++++++++++++++++++++++++++++++++ type/src/stream/impl.rs | 37 +++++++++++++++++ type/src/stream/struct.rs | 22 ++++++++++ 4 files changed, 150 insertions(+), 1 deletion(-) diff --git a/core/tests/mod.rs b/core/tests/mod.rs index 59ed11d..648e8fe 100644 --- a/core/tests/mod.rs +++ b/core/tests/mod.rs @@ -13,4 +13,10 @@ use std::{ time::{Duration, Instant}, }; -use tokio::{io::AsyncWriteExt, net::TcpStream, spawn, task::JoinHandle, time::sleep}; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::TcpStream, + spawn, + task::JoinHandle, + time::sleep, +}; diff --git a/core/tests/server/fn.rs b/core/tests/server/fn.rs index 2e7274f..aab9a8d 100644 --- a/core/tests/server/fn.rs +++ b/core/tests/server/fn.rs @@ -454,3 +454,87 @@ async fn client_concurrent_servers_on_distinct_ports_e2e() { control_a.wait().await; control_b.wait().await; } + +#[tokio::test] +async fn server_pipelined_requests_all_get_responses() { + let control_hook: ServerControlHook; + let port: u16; + { + let (hook, bound_port) = start_server_with(|server: &mut Server| { + server.route::("/"); + }) + .await; + control_hook = hook; + port = bound_port; + } + sleep(Duration::from_millis(120)).await; + + const REQUESTS: usize = 8; + let mut stream: TcpStream = TcpStream::connect(("127.0.0.1", port)).await.unwrap(); + let request: Vec = + b"GET / HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: keep-alive\r\n\r\n".to_vec(); + let mut batch: Vec = Vec::with_capacity(request.len() * REQUESTS); + for _ in 0..REQUESTS { + batch.extend_from_slice(&request); + } + stream.write_all(&batch).await.unwrap(); + stream.flush().await.unwrap(); + + let mut received: Vec = Vec::new(); + let deadline: Instant = Instant::now() + Duration::from_secs(5); + while Instant::now() < deadline { + let mut chunk: [u8; 8192] = [0; 8192]; + match stream.read(&mut chunk).await { + Ok(0) => break, + Ok(n) => received.extend_from_slice(&chunk[..n]), + Err(_) => break, + } + if count_pipelined_responses(&received) >= REQUESTS { + break; + } + } + + let responses: usize = count_pipelined_responses(&received); + assert_eq!( + responses, REQUESTS, + "expected {REQUESTS} pipelined responses, got {responses}" + ); + control_hook.shutdown().await; +} + +fn count_pipelined_responses(buffer: &[u8]) -> usize { + let mut count: usize = 0; + let mut position: usize = 0; + while let Some(head_end) = find_subslice(buffer, position, b"\r\n\r\n") { + let head: String = String::from_utf8_lossy(&buffer[position..head_end]).to_string(); + let mut content_length: Option = None; + for line in head.split("\r\n") { + if let Some((key, value)) = line.split_once(':') + && key.trim().eq_ignore_ascii_case("content-length") + { + content_length = value.trim().parse::().ok(); + } + } + let body_length: usize = match content_length { + Some(length) => length, + None => break, + }; + let body_start: usize = head_end + 4; + if buffer.len() < body_start + body_length { + break; + } + count += 1; + position = body_start + body_length; + } + count +} + +fn find_subslice(haystack: &[u8], from: usize, needle: &[u8]) -> Option { + if from > haystack.len() { + return None; + } + haystack[from..] + .windows(needle.len()) + .position(|window: &[u8]| window == needle) + .map(|index: usize| index + from) +} diff --git a/type/src/stream/impl.rs b/type/src/stream/impl.rs index dd55dc8..c7d25d2 100644 --- a/type/src/stream/impl.rs +++ b/type/src/stream/impl.rs @@ -269,8 +269,31 @@ impl Stream { let config: RequestConfig = *self.get_request_config(); let buffer_size: usize = config.get_buffer_size(); let max_path_size: usize = config.get_max_path_size(); + // Take any bytes the previous request read ahead, BEFORE borrowing the + // socket: those bytes already left the socket and cannot be re-read, so + // dropping them would silently discard a pipelined request. + let seeded: Vec = { + let start: usize = self.get_leftover_start(); + let end: usize = self.get_leftover_end(); + let mut pending: Vec = Vec::new(); + if start < end { + let leftover: Vec = mem::take(self.get_mut_leftover()); + let take: usize = (end - start).min(leftover.len()); + pending.extend_from_slice(&leftover[start..start + take]); + } + self.set_leftover(Vec::new()); + self.set_leftover_start(0); + self.set_leftover_end(0); + pending + }; let buffer: Vec = take_read_buffer(buffer_size); let mut reader: PooledReader<'_> = PooledReader::new(self.get_mut_stream(), buffer); + if !seeded.is_empty() { + let target: &mut Vec = reader.get_mut_buffer(); + let copy_len: usize = seeded.len().min(target.len()); + target[..copy_len].copy_from_slice(&seeded[..copy_len]); + reader.set_end(copy_len); + } let mut line: String = String::with_capacity(REQUEST_LINE_BUFFER_CAPACITY); AsyncBufReadExt::read_line(&mut reader, &mut line).await?; let (method, path, version): (RequestMethod, &str, RequestVersion) = @@ -286,6 +309,20 @@ impl Stream { request.set_method(method); request.set_version(version); Request::fill_http_body(&mut reader, request.get_mut_body(), content_size).await?; + // Preserve bytes read ahead for the next pipelined request: they have + // already left the socket and cannot be read again. + let reader_start: usize = *reader.get_start(); + let reader_end: usize = *reader.get_end(); + let pending: Vec = if reader_start < reader_end { + reader.get_buffer()[reader_start..reader_end].to_vec() + } else { + Vec::new() + }; + drop(reader); + let pending_len: usize = pending.len(); + self.set_leftover(pending); + self.set_leftover_start(0); + self.set_leftover_end(pending_len); Ok(()) } diff --git a/type/src/stream/struct.rs b/type/src/stream/struct.rs index 2ecfe66..30da1f1 100644 --- a/type/src/stream/struct.rs +++ b/type/src/stream/struct.rs @@ -11,6 +11,28 @@ pub struct Stream { #[get(type(copy))] #[get_mut(pub(super))] pub(super) closed: bool, + /// Bytes read ahead of the current request, already consumed from the + /// socket but not yet parsed into a request. + /// + /// A single socket read can deliver bytes belonging to the next pipelined + /// request. Those bytes are carried here between requests so + /// HTTP pipelining is not silently dropped. + #[get_mut(pub(super))] + #[set(pub(super))] + #[new(skip)] + pub(super) leftover: Vec, + /// Offset of the first unconsumed byte in `leftover`. + #[get(type(copy))] + #[get_mut(pub(super))] + #[set(pub(super))] + #[new(skip)] + pub(super) leftover_start: usize, + /// Offset one past the last valid byte in `leftover`. + #[get(type(copy))] + #[get_mut(pub(super))] + #[set(pub(super))] + #[new(skip)] + pub(super) leftover_end: usize, } /// A buffered reader over a `TcpStream` with a reusable read buffer.