From e82b4ef995d6396ae7afefd6f6efe5320260936e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Da=CC=81vid=20Istva=CC=81n=20Bi=CC=81ro=CC=81?= Date: Sat, 29 Aug 2026 14:17:27 +0200 Subject: [PATCH 1/9] test(http): capture transport lifecycle on failures --- tests/common/mod.rs | 410 +++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 406 insertions(+), 4 deletions(-) diff --git a/tests/common/mod.rs b/tests/common/mod.rs index c971f6d8..984f5347 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -7,13 +7,18 @@ use camino::{Utf8Path, Utf8PathBuf}; use camino_tempfile::{NamedUtf8TempFile, Utf8TempDir}; use futures::FutureExt; use heck::ToSnakeCase; +use http_body::{Body as HttpBody, Frame, SizeHint}; +use http_body_util::BodyExt; use std::collections::{BTreeMap, BTreeSet, HashMap}; use std::fs; +use std::future::Future; use std::hash::{Hash, Hasher}; use std::io::{Read, Write}; +use std::pin::Pin; use std::process::Command; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex, OnceLock}; +use std::task::{Context, Poll}; use std::thread; use std::time::{Duration, Instant, SystemTime}; use tokio::time::timeout; @@ -31,7 +36,7 @@ use wasmtime_wasi::cli::OutputFile; use wasmtime_wasi::p2::bindings; use wasmtime_wasi::{DirPerms, FilePerms, WasiCtx, WasiCtxView, WasiView}; use wasmtime_wasi_http::WasiHttpCtx; -use wasmtime_wasi_http::p2::{WasiHttpCtxView, WasiHttpView, default_hooks}; +use wasmtime_wasi_http::p2::{WasiHttpCtxView, WasiHttpView}; pub mod ws_mock_p2 { wasmtime::component::bindgen!({ @@ -244,6 +249,397 @@ const TEST_WASMTIME_CACHE_ENV: &str = "WASM_RQUICKJS_TEST_WASMTIME_CACHE"; /// The buffer is shared by all tests in the process and capped, keeping the most recent output. static HOST_TRACE: Mutex> = Mutex::new(Vec::new()); const HOST_TRACE_CAP: usize = 256 * 1024; +static NEXT_HTTP_TRACE_INVOCATION: AtomicUsize = AtomicUsize::new(1); +const HTTP_LIFECYCLE_CAP: usize = 256; + +#[repr(u8)] +#[derive(Clone, Copy)] +enum HttpLifecyclePhase { + Submit = 1, + Target = 2, + RequestFirstData = 3, + RequestFirstTrailers = 4, + RequestEof = 5, + RequestError = 6, + RequestDrop = 7, + ResponseHead = 8, + ResponseFirstData = 9, + ResponseFirstTrailers = 10, + ResponseEof = 11, + ResponseError = 12, + ResponseDrop = 13, + SendError = 14, + ResponseIoOk = 15, + ResponseIoError = 16, +} + +impl HttpLifecyclePhase { + fn label(value: u8) -> &'static str { + match value { + 1 => "submit", + 2 => "target", + 3 => "request-first-data", + 4 => "request-first-trailers", + 5 => "request-eof", + 6 => "request-error", + 7 => "request-drop-before-terminal", + 8 => "response-head", + 9 => "response-first-data", + 10 => "response-first-trailers", + 11 => "response-eof", + 12 => "response-error", + 13 => "response-drop-before-terminal", + 14 => "send-error", + 15 => "response-io-ok", + 16 => "response-io-error", + _ => "unknown", + } + } +} + +struct HttpLifecycleJournal { + next_event: AtomicUsize, + slots: [std::sync::atomic::AtomicU64; HTTP_LIFECYCLE_CAP], +} + +impl HttpLifecycleJournal { + fn new() -> Self { + Self { + next_event: AtomicUsize::new(1), + slots: std::array::from_fn(|_| std::sync::atomic::AtomicU64::new(0)), + } + } + + fn record(&self, request: usize, phase: HttpLifecyclePhase, detail: u16) { + let sequence = self.next_event.fetch_add(1, Ordering::Relaxed); + let packed = ((sequence as u64 & 0x00ff_ffff) << 40) + | ((request as u64 & 0xffff) << 24) + | ((phase as u64) << 16) + | u64::from(detail); + self.slots[sequence % HTTP_LIFECYCLE_CAP].store(packed, Ordering::Release); + } + + fn snapshot(&self, invocation: usize) -> String { + let mut events = self + .slots + .iter() + .map(|slot| slot.load(Ordering::Acquire)) + .filter(|event| *event != 0) + .collect::>(); + events.sort_unstable_by_key(|event| event >> 40); + let mut result = format!("invocation={invocation}\n"); + for event in events { + let sequence = event >> 40; + let request = (event >> 24) & 0xffff; + let phase = ((event >> 16) & 0xff) as u8; + let detail = event & 0xffff; + use std::fmt::Write as _; + let _ = writeln!( + result, + "seq={sequence} request={request} phase={} detail={detail}", + HttpLifecyclePhase::label(phase) + ); + } + result + } +} + +/// Per-component correlation state for the test harness' outgoing HTTP lifecycle trace. +/// +/// The trace is intentionally implemented in the host rather than the embedded skeleton: it is +/// test-only, covers both P2 and P3, and can observe the Wasmtime transport boundary without +/// changing a generated component. Events use a fixed atomic journal, so the hot path has no +/// locks, allocation, clocks, or output; the journal is formatted only after an invocation fails. +#[derive(Clone)] +struct HttpLifecycleTrace { + invocation: usize, + next_request: Arc, + journal: Arc, +} + +impl HttpLifecycleTrace { + fn new() -> Self { + Self { + invocation: NEXT_HTTP_TRACE_INVOCATION.fetch_add(1, Ordering::Relaxed), + next_request: Arc::new(AtomicUsize::new(1)), + journal: Arc::new(HttpLifecycleJournal::new()), + } + } + + fn next_request(&self) -> usize { + self.next_request.fetch_add(1, Ordering::Relaxed) + } + + fn record(&self, request: usize, phase: HttpLifecyclePhase, detail: u16) { + self.journal.record(request, phase, detail); + } + + fn record_submit(&self, request: usize, method: &http::Method, uri: &http::Uri) { + let method = if method == http::Method::GET { + 1 + } else if method == http::Method::POST { + 2 + } else if method == http::Method::PUT { + 3 + } else if method == http::Method::DELETE { + 4 + } else if method == http::Method::HEAD { + 5 + } else { + 0 + }; + self.record(request, HttpLifecyclePhase::Submit, method); + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + uri.hash(&mut hasher); + self.record( + request, + HttpLifecyclePhase::Target, + (hasher.finish() & 0xffff) as u16, + ); + } + + fn snapshot(&self) -> String { + self.journal.snapshot(self.invocation) + } +} + +/// A transparent body observer. It never polls ahead or adds an await: every host poll is +/// delegated exactly once, with only the first frame and the terminal outcome recorded. +struct TracedHttpBody { + inner: Pin>, + trace: HttpLifecycleTrace, + request: usize, + side: &'static str, + saw_frame: bool, + terminal: bool, +} + +impl TracedHttpBody { + fn new(inner: B, trace: HttpLifecycleTrace, request: usize, side: &'static str) -> Self { + Self { + inner: Box::pin(inner), + trace, + request, + side, + saw_frame: false, + terminal: false, + } + } + + fn event(&self, outcome: &'static str) { + let phase = match (self.side, outcome) { + ("request", "first-data") => HttpLifecyclePhase::RequestFirstData, + ("request", "first-trailers") => HttpLifecyclePhase::RequestFirstTrailers, + ("request", "eof") => HttpLifecyclePhase::RequestEof, + ("request", "error") => HttpLifecyclePhase::RequestError, + ("request", "drop-before-terminal") => HttpLifecyclePhase::RequestDrop, + ("response", "first-data") => HttpLifecyclePhase::ResponseFirstData, + ("response", "first-trailers") => HttpLifecyclePhase::ResponseFirstTrailers, + ("response", "eof") => HttpLifecyclePhase::ResponseEof, + ("response", "error") => HttpLifecyclePhase::ResponseError, + ("response", "drop-before-terminal") => HttpLifecyclePhase::ResponseDrop, + _ => return, + }; + self.trace.record(self.request, phase, 0); + } +} + +impl HttpBody for TracedHttpBody { + type Data = B::Data; + type Error = B::Error; + + fn poll_frame( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll, Self::Error>>> { + let this = self.get_mut(); + match this.inner.as_mut().poll_frame(cx) { + Poll::Ready(Some(Ok(frame))) => { + if !this.saw_frame { + this.saw_frame = true; + this.event(if frame.is_data() { + "first-data" + } else { + "first-trailers" + }); + } + Poll::Ready(Some(Ok(frame))) + } + Poll::Ready(Some(Err(error))) => { + this.terminal = true; + this.event("error"); + Poll::Ready(Some(Err(error))) + } + Poll::Ready(None) => { + this.terminal = true; + this.event("eof"); + Poll::Ready(None) + } + Poll::Pending => Poll::Pending, + } + } + + fn is_end_stream(&self) -> bool { + self.inner.as_ref().is_end_stream() + } + + fn size_hint(&self) -> SizeHint { + self.inner.as_ref().size_hint() + } +} + +impl Drop for TracedHttpBody { + fn drop(&mut self) { + if !self.terminal { + self.event("drop-before-terminal"); + } + } +} + +#[derive(Clone)] +struct P2HttpTraceHooks(HttpLifecycleTrace); + +impl wasmtime_wasi_http::p2::WasiHttpHooks for P2HttpTraceHooks { + #[cfg(not(feature = "use-golem-wasmtime"))] + fn send_request( + &mut self, + request: http::Request, + config: wasmtime_wasi_http::p2::types::OutgoingRequestConfig, + ) -> wasmtime_wasi_http::p2::HttpResult + { + let trace = self.0.clone(); + let request_id = trace.next_request(); + trace.record_submit(request_id, request.method(), request.uri()); + + let (parts, body) = request.into_parts(); + let request = http::Request::from_parts( + parts, + TracedHttpBody::new(body, trace.clone(), request_id, "request").boxed_unsync(), + ); + Ok(wasmtime_wasi_http::p2::default_send_request( + request, config, + )) + } + + #[cfg(feature = "use-golem-wasmtime")] + fn send_request( + &mut self, + request: http::Request, + config: wasmtime_wasi_http::p2::types::OutgoingRequestConfig, + body_completion: Option, + ) -> wasmtime_wasi_http::p2::HttpResult + { + let trace = self.0.clone(); + let request_id = trace.next_request(); + trace.record_submit(request_id, request.method(), request.uri()); + let (parts, body) = request.into_parts(); + let request = http::Request::from_parts( + parts, + TracedHttpBody::new(body, trace, request_id, "request").boxed_unsync(), + ); + Ok(wasmtime_wasi_http::p2::default_send_request_with_pool( + request, + config, + body_completion, + None, + )) + } +} + +#[derive(Clone)] +struct P3HttpTraceHooks(HttpLifecycleTrace); + +impl wasmtime_wasi_http::p3::WasiHttpHooks for P3HttpTraceHooks { + fn send_request( + &mut self, + request: http::Request< + http_body_util::combinators::UnsyncBoxBody< + bytes::Bytes, + wasmtime_wasi_http::p3::bindings::http::types::ErrorCode, + >, + >, + options: Option, + response_processing: Box< + dyn Future< + Output = Result<(), wasmtime_wasi_http::p3::bindings::http::types::ErrorCode>, + > + Send, + >, + ) -> Box< + dyn Future< + Output = Result< + ( + http::Response< + http_body_util::combinators::UnsyncBoxBody< + bytes::Bytes, + wasmtime_wasi_http::p3::bindings::http::types::ErrorCode, + >, + >, + Box< + dyn Future< + Output = Result< + (), + wasmtime_wasi_http::p3::bindings::http::types::ErrorCode, + >, + > + Send, + >, + ), + wasmtime_wasi::TrappableError< + wasmtime_wasi_http::p3::bindings::http::types::ErrorCode, + >, + >, + > + Send, + > { + // Match Wasmtime's default hook: response-processing is currently not wired into the + // default client. Keep that ownership behavior unchanged while tracing the returned I/O + // future, which is the transport's actual connection lifetime signal. + drop(response_processing); + + let trace = self.0.clone(); + let request_id = trace.next_request(); + trace.record_submit(request_id, request.method(), request.uri()); + let (parts, body) = request.into_parts(); + let request = http::Request::from_parts( + parts, + TracedHttpBody::new(body, trace.clone(), request_id, "request").boxed_unsync(), + ); + + Box::new(async move { + let result = wasmtime_wasi_http::p3::default_send_request(request, options).await; + let (response, io) = match result { + Ok(value) => value, + Err(error) => { + trace.record(request_id, HttpLifecyclePhase::SendError, 0); + return Err(error.into()); + } + }; + trace.record( + request_id, + HttpLifecyclePhase::ResponseHead, + response.status().as_u16(), + ); + let (parts, body) = response.into_parts(); + let response = http::Response::from_parts( + parts, + TracedHttpBody::new(body, trace.clone(), request_id, "response").boxed_unsync(), + ); + let io_trace = trace.clone(); + let io = Box::new(async move { + let result = io.await; + io_trace.record( + request_id, + if result.is_ok() { + HttpLifecyclePhase::ResponseIoOk + } else { + HttpLifecyclePhase::ResponseIoError + }, + 0, + ); + result + }) as Box> + Send>; + Ok((response, io)) + }) + } +} #[derive(Clone, Copy)] struct HostTraceWriter; @@ -1836,10 +2232,13 @@ impl TestInstance { #[cfg(not(feature = "use-golem-wasmtime"))] let ctx = ctx_builder.build(); let http_ctx = WasiHttpCtx::new(); + let http_trace = HttpLifecycleTrace::new(); let host = Host { table: Arc::new(Mutex::new(ResourceTable::new())), wasi: Arc::new(Mutex::new(ctx)), wasi_http: Arc::new(Mutex::new(http_ctx)), + p2_http_hooks: P2HttpTraceHooks(http_trace.clone()), + p3_http_hooks: P3HttpTraceHooks(http_trace), started_at: Instant::now(), timeout: Duration::from_secs(120), log_messages: Arc::new(Mutex::new(Vec::new())), @@ -1934,8 +2333,9 @@ impl TestInstance { // on CI). let results = results.map_err(|err| { let host_trace = host_trace(); + let http_lifecycle = self.store.data().p2_http_hooks.0.snapshot(); err.context(format!( - "guest stdout:\n{stdout}\nguest stderr:\n{stderr}\nhost trace:\n{host_trace}" + "guest stdout:\n{stdout}\nguest stderr:\n{stderr}\nHTTP lifecycle:\n{http_lifecycle}\nhost trace:\n{host_trace}" )) }); @@ -2527,6 +2927,8 @@ pub struct Host { pub table: Arc>, pub wasi: Arc>, pub wasi_http: Arc>, + p2_http_hooks: P2HttpTraceHooks, + p3_http_hooks: P3HttpTraceHooks, pub started_at: Instant, pub timeout: Duration, pub log_messages: Arc>>, @@ -2596,7 +2998,7 @@ impl WasiHttpView for Host { .expect("ResourceTable is shared and cannot be borrowed mutably") .get_mut() .expect("ResourceTable mutex must never fail"), - hooks: default_hooks(), + hooks: &mut self.p2_http_hooks, } } } @@ -2800,7 +3202,7 @@ fn add_websocket_client_mock(linker: &mut Linker, target: TestTarget) -> a impl wasmtime_wasi_http::p3::WasiHttpView for Host { fn http(&mut self) -> wasmtime_wasi_http::p3::WasiHttpCtxView<'_> { wasmtime_wasi_http::p3::WasiHttpCtxView { - hooks: wasmtime_wasi_http::p3::default_hooks(), + hooks: &mut self.p3_http_hooks, table: Arc::get_mut(&mut self.table) .expect("ResourceTable is shared and cannot be borrowed mutably") .get_mut() From 2e53877209c6f51abbec1ab7f564e375ba8d01c2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Da=CC=81vid=20Istva=CC=81n=20Bi=CC=81ro=CC=81?= Date: Sat, 29 Aug 2026 14:30:16 +0200 Subject: [PATCH 2/9] test(http): correlate response and server ownership --- tests/common/mod.rs | 270 +++++++++++++++++++++++++++++++++--- tests/common/test_server.rs | 28 ++++ 2 files changed, 279 insertions(+), 19 deletions(-) diff --git a/tests/common/mod.rs b/tests/common/mod.rs index 984f5347..17a20b8b 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -250,6 +250,7 @@ const TEST_WASMTIME_CACHE_ENV: &str = "WASM_RQUICKJS_TEST_WASMTIME_CACHE"; static HOST_TRACE: Mutex> = Mutex::new(Vec::new()); const HOST_TRACE_CAP: usize = 256 * 1024; static NEXT_HTTP_TRACE_INVOCATION: AtomicUsize = AtomicUsize::new(1); +static TEST_SERVER_HTTP_TRACE: OnceLock = OnceLock::new(); const HTTP_LIFECYCLE_CAP: usize = 256; #[repr(u8)] @@ -271,13 +272,24 @@ enum HttpLifecyclePhase { SendError = 14, ResponseIoOk = 15, ResponseIoError = 16, + ServerArrival = 17, + ServerRequestFirstData = 18, + ServerRequestEof = 19, + ServerRequestError = 20, + ServerRequestDrop = 21, + ServerResponseHead = 22, + ServerResponseFirstData = 23, + ServerResponseEof = 24, + ServerResponseError = 25, + ServerResponseDrop = 26, + TargetPath = 27, } impl HttpLifecyclePhase { fn label(value: u8) -> &'static str { match value { 1 => "submit", - 2 => "target", + 2 => "target-port", 3 => "request-first-data", 4 => "request-first-trailers", 5 => "request-eof", @@ -292,6 +304,17 @@ impl HttpLifecyclePhase { 14 => "send-error", 15 => "response-io-ok", 16 => "response-io-error", + 17 => "server-arrival", + 18 => "server-request-first-data", + 19 => "server-request-eof", + 20 => "server-request-error", + 21 => "server-request-drop-before-terminal", + 22 => "server-response-head", + 23 => "server-response-first-data", + 24 => "server-response-eof", + 25 => "server-response-error", + 26 => "server-response-drop-before-terminal", + 27 => "target-path-hash", _ => "unknown", } } @@ -312,11 +335,27 @@ impl HttpLifecycleJournal { fn record(&self, request: usize, phase: HttpLifecyclePhase, detail: u16) { let sequence = self.next_event.fetch_add(1, Ordering::Relaxed); + self.publish(sequence, request, phase, detail); + } + + fn publish(&self, sequence: usize, request: usize, phase: HttpLifecyclePhase, detail: u16) { let packed = ((sequence as u64 & 0x00ff_ffff) << 40) | ((request as u64 & 0xffff) << 24) | ((phase as u64) << 16) | u64::from(detail); - self.slots[sequence % HTTP_LIFECYCLE_CAP].store(packed, Ordering::Release); + let slot = &self.slots[sequence % HTTP_LIFECYCLE_CAP]; + let mut current = slot.load(Ordering::Acquire); + loop { + let current_sequence = current >> 40; + if current_sequence >= (sequence as u64 & 0x00ff_ffff) { + return; + } + match slot.compare_exchange_weak(current, packed, Ordering::Release, Ordering::Acquire) + { + Ok(_) => return, + Err(observed) => current = observed, + } + } } fn snapshot(&self, invocation: usize) -> String { @@ -389,11 +428,16 @@ impl HttpLifecycleTrace { 0 }; self.record(request, HttpLifecyclePhase::Submit, method); - let mut hasher = std::collections::hash_map::DefaultHasher::new(); - uri.hash(&mut hasher); self.record( request, HttpLifecyclePhase::Target, + uri.port_u16().unwrap_or_default(), + ); + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + uri.path_and_query().hash(&mut hasher); + self.record( + request, + HttpLifecyclePhase::TargetPath, (hasher.finish() & 0xffff) as u16, ); } @@ -403,9 +447,44 @@ impl HttpLifecycleTrace { } } +fn test_server_http_trace() -> &'static HttpLifecycleTrace { + TEST_SERVER_HTTP_TRACE.get_or_init(HttpLifecycleTrace::new) +} + +fn record_test_server_arrival(port: u16, uri: &http::Uri) { + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + uri.path_and_query().hash(&mut hasher); + test_server_http_trace().record( + usize::from(port), + HttpLifecyclePhase::ServerArrival, + (hasher.finish() & 0xffff) as u16, + ); +} + +fn record_test_server_response_head(port: u16, status: http::StatusCode) { + test_server_http_trace().record( + usize::from(port), + HttpLifecyclePhase::ServerResponseHead, + status.as_u16(), + ); +} + +fn traced_test_server_body( + body: B, + port: u16, + side: &'static str, +) -> TracedHttpBody { + TracedHttpBody::new( + body, + test_server_http_trace().clone(), + usize::from(port), + side, + ) +} + /// A transparent body observer. It never polls ahead or adds an await: every host poll is /// delegated exactly once, with only the first frame and the terminal outcome recorded. -struct TracedHttpBody { +struct TracedHttpBody { inner: Pin>, trace: HttpLifecycleTrace, request: usize, @@ -414,7 +493,7 @@ struct TracedHttpBody { terminal: bool, } -impl TracedHttpBody { +impl TracedHttpBody { fn new(inner: B, trace: HttpLifecycleTrace, request: usize, side: &'static str) -> Self { Self { inner: Box::pin(inner), @@ -438,6 +517,14 @@ impl TracedHttpBody { ("response", "eof") => HttpLifecyclePhase::ResponseEof, ("response", "error") => HttpLifecyclePhase::ResponseError, ("response", "drop-before-terminal") => HttpLifecyclePhase::ResponseDrop, + ("server-request", "first-data") => HttpLifecyclePhase::ServerRequestFirstData, + ("server-request", "eof") => HttpLifecyclePhase::ServerRequestEof, + ("server-request", "error") => HttpLifecyclePhase::ServerRequestError, + ("server-request", "drop-before-terminal") => HttpLifecyclePhase::ServerRequestDrop, + ("server-response", "first-data") => HttpLifecyclePhase::ServerResponseFirstData, + ("server-response", "eof") => HttpLifecyclePhase::ServerResponseEof, + ("server-response", "error") => HttpLifecyclePhase::ServerResponseError, + ("server-response", "drop-before-terminal") => HttpLifecyclePhase::ServerResponseDrop, _ => return, }; self.trace.record(self.request, phase, 0); @@ -488,9 +575,9 @@ impl HttpBody for TracedHttpBody { } } -impl Drop for TracedHttpBody { +impl Drop for TracedHttpBody { fn drop(&mut self) { - if !self.terminal { + if !self.terminal && !self.inner.as_ref().is_end_stream() { self.event("drop-before-terminal"); } } @@ -499,6 +586,43 @@ impl Drop for TracedHttpBody { #[derive(Clone)] struct P2HttpTraceHooks(HttpLifecycleTrace); +fn trace_p2_result( + trace: &HttpLifecycleTrace, + request_id: usize, + result: Result< + wasmtime_wasi_http::p2::types::IncomingResponse, + wasmtime_wasi_http::p2::bindings::http::types::ErrorCode, + >, +) -> Result< + wasmtime_wasi_http::p2::types::IncomingResponse, + wasmtime_wasi_http::p2::bindings::http::types::ErrorCode, +> { + match result { + Ok(mut incoming) => { + trace.record( + request_id, + HttpLifecyclePhase::ResponseHead, + incoming.resp.status().as_u16(), + ); + let (parts, body) = incoming.resp.into_parts(); + incoming.resp = http::Response::from_parts( + parts, + TracedHttpBody::new(body, trace.clone(), request_id, "response").boxed_unsync(), + ); + Ok(incoming) + } + Err(error) => { + trace.record(request_id, HttpLifecyclePhase::SendError, 0); + Err(error) + } + } +} + +#[cfg(feature = "use-golem-wasmtime")] +fn p2_method_expects_body(method: &http::Method) -> bool { + method == http::Method::POST || method == http::Method::PUT || method == http::Method::PATCH +} + impl wasmtime_wasi_http::p2::WasiHttpHooks for P2HttpTraceHooks { #[cfg(not(feature = "use-golem-wasmtime"))] fn send_request( @@ -516,9 +640,12 @@ impl wasmtime_wasi_http::p2::WasiHttpHooks for P2HttpTraceHooks { parts, TracedHttpBody::new(body, trace.clone(), request_id, "request").boxed_unsync(), ); - Ok(wasmtime_wasi_http::p2::default_send_request( - request, config, - )) + let handle = wasmtime_wasi::runtime::spawn(async move { + let result = + wasmtime_wasi_http::p2::default_send_request_handler(request, config).await; + Ok(trace_p2_result(&trace, request_id, result)) + }); + Ok(wasmtime_wasi_http::p2::types::HostFutureIncomingResponse::pending(handle)) } #[cfg(feature = "use-golem-wasmtime")] @@ -532,17 +659,44 @@ impl wasmtime_wasi_http::p2::WasiHttpHooks for P2HttpTraceHooks { let trace = self.0.clone(); let request_id = trace.next_request(); trace.record_submit(request_id, request.method(), request.uri()); + let collect_before_send = + !p2_method_expects_body(request.method()) && body_completion.is_some(); let (parts, body) = request.into_parts(); let request = http::Request::from_parts( parts, - TracedHttpBody::new(body, trace, request_id, "request").boxed_unsync(), + TracedHttpBody::new(body, trace.clone(), request_id, "request").boxed_unsync(), ); - Ok(wasmtime_wasi_http::p2::default_send_request_with_pool( - request, - config, - body_completion, - None, - )) + let handle = wasmtime_wasi::runtime::spawn(async move { + let request = if collect_before_send { + let body_completion = body_completion.expect("checked above"); + let (parts, body) = request.into_parts(); + let completion = async { + match body_completion.await { + Ok(Ok(())) => Ok(()), + Ok(Err(error)) => Err(error), + Err(_) => Err( + wasmtime_wasi_http::p2::bindings::http::types::ErrorCode::HttpProtocolError, + ), + } + }; + let collect = async { + BodyExt::collect(body).await.map(|collected| { + collected + .map_err(|_: std::convert::Infallible| unreachable!()) + .boxed_unsync() + }) + }; + let (completion, collected) = futures::future::join(completion, collect).await; + completion?; + http::Request::from_parts(parts, collected?) + } else { + request + }; + let result = + wasmtime_wasi_http::p2::default_send_request_handler(request, config).await; + Ok(trace_p2_result(&trace, request_id, result)) + }); + Ok(wasmtime_wasi_http::p2::types::HostFutureIncomingResponse::pending(handle)) } } @@ -1001,6 +1155,83 @@ mod tests { use super::*; use test_r::test; + #[test] + fn http_lifecycle_ring_rejects_delayed_old_generation() { + let journal = HttpLifecycleJournal::new(); + journal.publish(1, 1, HttpLifecyclePhase::Submit, 0); + journal.publish( + 1 + HTTP_LIFECYCLE_CAP, + 2, + HttpLifecyclePhase::ResponseHead, + 200, + ); + journal.publish(1, 3, HttpLifecyclePhase::SendError, 0); + + let retained = journal.slots[1].load(Ordering::Acquire); + assert_eq!(retained >> 40, (1 + HTTP_LIFECYCLE_CAP) as u64); + assert_eq!((retained >> 24) & 0xffff, 2); + } + + #[test] + fn http_lifecycle_ring_remains_valid_during_concurrent_snapshots() { + let journal = Arc::new(HttpLifecycleJournal::new()); + let writers = (0..8) + .map(|request| { + let journal = journal.clone(); + thread::spawn(move || { + for _ in 0..1_000 { + journal.record(request, HttpLifecyclePhase::Submit, 0); + } + }) + }) + .collect::>(); + + for _ in 0..100 { + let snapshot = journal.snapshot(1); + assert!(snapshot.starts_with("invocation=1\n")); + } + for writer in writers { + writer.join().unwrap(); + } + + let newest = journal.next_event.load(Ordering::Acquire) - 1; + let retained = journal + .slots + .iter() + .map(|slot| slot.load(Ordering::Acquire) >> 40) + .collect::>(); + assert!(retained.contains(&(newest as u64))); + assert!( + retained + .iter() + .all(|sequence| *sequence > (newest - HTTP_LIFECYCLE_CAP) as u64) + ); + } + + #[test] + fn http_lifecycle_empty_body_drop_is_terminal() { + let trace = HttpLifecycleTrace::new(); + let body = http_body_util::Empty::::new(); + drop(TracedHttpBody::new(body, trace.clone(), 1, "request")); + + let snapshot = trace.snapshot(); + assert!(!snapshot.contains("drop-before-terminal")); + } + + #[test] + fn http_lifecycle_p2_send_error_is_terminal_boundary() { + let trace = HttpLifecycleTrace::new(); + let result = trace_p2_result( + &trace, + 7, + Err(wasmtime_wasi_http::p2::bindings::http::types::ErrorCode::HttpProtocolError), + ); + + assert!(result.is_err()); + let snapshot = trace.snapshot(); + assert!(snapshot.contains("request=7 phase=send-error")); + } + #[test] fn artifact_cache_stamp_must_not_be_older_than_output() -> anyhow::Result<()> { if test_drop_cache_enabled() { @@ -2334,8 +2565,9 @@ impl TestInstance { let results = results.map_err(|err| { let host_trace = host_trace(); let http_lifecycle = self.store.data().p2_http_hooks.0.snapshot(); + let server_http_lifecycle = test_server_http_trace().snapshot(); err.context(format!( - "guest stdout:\n{stdout}\nguest stderr:\n{stderr}\nHTTP lifecycle:\n{http_lifecycle}\nhost trace:\n{host_trace}" + "guest stdout:\n{stdout}\nguest stderr:\n{stderr}\nHTTP lifecycle:\n{http_lifecycle}\ntest-server HTTP lifecycle:\n{server_http_lifecycle}\nhost trace:\n{host_trace}" )) }); diff --git a/tests/common/test_server.rs b/tests/common/test_server.rs index ade981e7..398456c1 100644 --- a/tests/common/test_server.rs +++ b/tests/common/test_server.rs @@ -1,6 +1,8 @@ use axum::body::Body; +use axum::extract::Request; use axum::extract::{Multipart, Path}; use axum::http::HeaderMap; +use axum::middleware::Next; use axum::response::{AppendHeaders, IntoResponse}; use axum::routing::{get, post}; use axum::{Json, Router}; @@ -28,6 +30,30 @@ impl Drop for TestServerHandle { } } +fn trace_http_lifecycle(router: Router, port: u16) -> Router { + router.layer(axum::middleware::from_fn( + move |request: Request, next: Next| async move { + super::record_test_server_arrival(port, request.uri()); + let (parts, body) = request.into_parts(); + let request = Request::from_parts( + parts, + Body::new(super::traced_test_server_body(body, port, "server-request")), + ); + let response = next.run(request).await; + super::record_test_server_response_head(port, response.status()); + let (parts, body) = response.into_parts(); + axum::response::Response::from_parts( + parts, + Body::new(super::traced_test_server_body( + body, + port, + "server-response", + )), + ) + }, + )) +} + pub async fn start_test_server() -> (u16, TestServerHandle) { let listener = tokio::net::TcpListener::bind("0.0.0.0:0").await.unwrap(); let host_http_port = listener.local_addr().unwrap().port(); @@ -227,6 +253,7 @@ pub async fn start_test_server() -> (u16, TestServerHandle) { .into_response() }), ); + let router = trace_http_lifecycle(router, host_http_port); axum::serve(listener, router).await.unwrap(); }); @@ -263,6 +290,7 @@ pub async fn start_abort_test_server() -> (u16, TestServerHandle, mpsc::Unbounde axum::routing::any(async || (StatusCode::FOUND, [("Location", "/slow-response")])), ) .route("/abort-ready", ready); + let router = trace_http_lifecycle(router, port); axum::serve(listener, router).await.unwrap(); }); From b8830bf05671b05abadc38a3ddcb2039132bca67 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Da=CC=81vid=20Istva=CC=81n=20Bi=CC=81ro=CC=81?= Date: Sat, 29 Aug 2026 14:39:48 +0200 Subject: [PATCH 3/9] test(http): preserve correlation under concurrency --- tests/common/mod.rs | 160 ++++++++++++++++++++++++++++++------ tests/common/test_server.rs | 13 ++- 2 files changed, 145 insertions(+), 28 deletions(-) diff --git a/tests/common/mod.rs b/tests/common/mod.rs index 17a20b8b..c629a3d2 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -250,8 +250,11 @@ const TEST_WASMTIME_CACHE_ENV: &str = "WASM_RQUICKJS_TEST_WASMTIME_CACHE"; static HOST_TRACE: Mutex> = Mutex::new(Vec::new()); const HOST_TRACE_CAP: usize = 256 * 1024; static NEXT_HTTP_TRACE_INVOCATION: AtomicUsize = AtomicUsize::new(1); +static NEXT_HTTP_TRACE_REQUEST: AtomicUsize = AtomicUsize::new(1); static TEST_SERVER_HTTP_TRACE: OnceLock = OnceLock::new(); const HTTP_LIFECYCLE_CAP: usize = 256; +const HTTP_LIFECYCLE_SEQUENCE_MASK: u64 = 0x00ff_ffff; +const HTTP_LIFECYCLE_SEQUENCE_HALF: u64 = 0x0080_0000; #[repr(u8)] #[derive(Clone, Copy)] @@ -283,6 +286,7 @@ enum HttpLifecyclePhase { ServerResponseError = 25, ServerResponseDrop = 26, TargetPath = 27, + ServerPort = 28, } impl HttpLifecyclePhase { @@ -315,6 +319,7 @@ impl HttpLifecyclePhase { 25 => "server-response-error", 26 => "server-response-drop-before-terminal", 27 => "target-path-hash", + 28 => "server-port", _ => "unknown", } } @@ -339,7 +344,8 @@ impl HttpLifecycleJournal { } fn publish(&self, sequence: usize, request: usize, phase: HttpLifecyclePhase, detail: u16) { - let packed = ((sequence as u64 & 0x00ff_ffff) << 40) + let encoded_sequence = sequence as u64 & HTTP_LIFECYCLE_SEQUENCE_MASK; + let packed = (encoded_sequence << 40) | ((request as u64 & 0xffff) << 24) | ((phase as u64) << 16) | u64::from(detail); @@ -347,8 +353,12 @@ impl HttpLifecycleJournal { let mut current = slot.load(Ordering::Acquire); loop { let current_sequence = current >> 40; - if current_sequence >= (sequence as u64 & 0x00ff_ffff) { - return; + if current != 0 { + let advance = + encoded_sequence.wrapping_sub(current_sequence) & HTTP_LIFECYCLE_SEQUENCE_MASK; + if advance == 0 || advance >= HTTP_LIFECYCLE_SEQUENCE_HALF { + return; + } } match slot.compare_exchange_weak(current, packed, Ordering::Release, Ordering::Acquire) { @@ -359,16 +369,31 @@ impl HttpLifecycleJournal { } fn snapshot(&self, invocation: usize) -> String { + let newest_full_sequence = self.next_event.load(Ordering::Acquire).saturating_sub(1); + let newest_encoded_sequence = newest_full_sequence as u64 & HTTP_LIFECYCLE_SEQUENCE_MASK; let mut events = self .slots .iter() .map(|slot| slot.load(Ordering::Acquire)) .filter(|event| *event != 0) .collect::>(); - events.sort_unstable_by_key(|event| event >> 40); + events.retain(|event| { + let sequence = event >> 40; + let age = newest_encoded_sequence.wrapping_sub(sequence) & HTTP_LIFECYCLE_SEQUENCE_MASK; + age < HTTP_LIFECYCLE_SEQUENCE_HALF + }); + events.sort_unstable_by_key(|event| { + let sequence = event >> 40; + std::cmp::Reverse( + newest_encoded_sequence.wrapping_sub(sequence) & HTTP_LIFECYCLE_SEQUENCE_MASK, + ) + }); let mut result = format!("invocation={invocation}\n"); for event in events { - let sequence = event >> 40; + let encoded_sequence = event >> 40; + let age = newest_encoded_sequence.wrapping_sub(encoded_sequence) + & HTTP_LIFECYCLE_SEQUENCE_MASK; + let sequence = newest_full_sequence.saturating_sub(age as usize); let request = (event >> 24) & 0xffff; let phase = ((event >> 16) & 0xff) as u8; let detail = event & 0xffff; @@ -392,7 +417,6 @@ impl HttpLifecycleJournal { #[derive(Clone)] struct HttpLifecycleTrace { invocation: usize, - next_request: Arc, journal: Arc, } @@ -400,13 +424,12 @@ impl HttpLifecycleTrace { fn new() -> Self { Self { invocation: NEXT_HTTP_TRACE_INVOCATION.fetch_add(1, Ordering::Relaxed), - next_request: Arc::new(AtomicUsize::new(1)), journal: Arc::new(HttpLifecycleJournal::new()), } } fn next_request(&self) -> usize { - self.next_request.fetch_add(1, Ordering::Relaxed) + NEXT_HTTP_TRACE_REQUEST.fetch_add(1, Ordering::Relaxed) } fn record(&self, request: usize, phase: HttpLifecyclePhase, detail: u16) { @@ -451,19 +474,35 @@ fn test_server_http_trace() -> &'static HttpLifecycleTrace { TEST_SERVER_HTTP_TRACE.get_or_init(HttpLifecycleTrace::new) } -fn record_test_server_arrival(port: u16, uri: &http::Uri) { +fn attach_http_correlation(request: &mut http::Request, request_id: usize) { + request.headers_mut().insert( + http::HeaderName::from_static("x-wrq-http-trace-id"), + http::HeaderValue::try_from(request_id.to_string()).expect("numeric header is valid"), + ); +} + +fn test_server_http_correlation(headers: &http::HeaderMap) -> usize { + headers + .get("x-wrq-http-trace-id") + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.parse().ok()) + .unwrap_or_default() +} + +fn record_test_server_arrival(request_id: usize, port: u16, uri: &http::Uri) { let mut hasher = std::collections::hash_map::DefaultHasher::new(); uri.path_and_query().hash(&mut hasher); test_server_http_trace().record( - usize::from(port), + request_id, HttpLifecyclePhase::ServerArrival, (hasher.finish() & 0xffff) as u16, ); + test_server_http_trace().record(request_id, HttpLifecyclePhase::ServerPort, port); } -fn record_test_server_response_head(port: u16, status: http::StatusCode) { +fn record_test_server_response_head(request_id: usize, status: http::StatusCode) { test_server_http_trace().record( - usize::from(port), + request_id, HttpLifecyclePhase::ServerResponseHead, status.as_u16(), ); @@ -471,15 +510,10 @@ fn record_test_server_response_head(port: u16, status: http::StatusCode) { fn traced_test_server_body( body: B, - port: u16, + request_id: usize, side: &'static str, ) -> TracedHttpBody { - TracedHttpBody::new( - body, - test_server_http_trace().clone(), - usize::from(port), - side, - ) + TracedHttpBody::new(body, test_server_http_trace().clone(), request_id, side) } /// A transparent body observer. It never polls ahead or adds an await: every host poll is @@ -623,16 +657,30 @@ fn p2_method_expects_body(method: &http::Method) -> bool { method == http::Method::POST || method == http::Method::PUT || method == http::Method::PATCH } +#[cfg(feature = "use-golem-wasmtime")] +fn p2_body_completion_for_dispatch( + method: &http::Method, + body_completion: Option, +) -> Option { + if p2_method_expects_body(method) { + drop(body_completion); + None + } else { + body_completion + } +} + impl wasmtime_wasi_http::p2::WasiHttpHooks for P2HttpTraceHooks { #[cfg(not(feature = "use-golem-wasmtime"))] fn send_request( &mut self, - request: http::Request, + mut request: http::Request, config: wasmtime_wasi_http::p2::types::OutgoingRequestConfig, ) -> wasmtime_wasi_http::p2::HttpResult { let trace = self.0.clone(); let request_id = trace.next_request(); + attach_http_correlation(&mut request, request_id); trace.record_submit(request_id, request.method(), request.uri()); let (parts, body) = request.into_parts(); @@ -651,16 +699,17 @@ impl wasmtime_wasi_http::p2::WasiHttpHooks for P2HttpTraceHooks { #[cfg(feature = "use-golem-wasmtime")] fn send_request( &mut self, - request: http::Request, + mut request: http::Request, config: wasmtime_wasi_http::p2::types::OutgoingRequestConfig, body_completion: Option, ) -> wasmtime_wasi_http::p2::HttpResult { let trace = self.0.clone(); let request_id = trace.next_request(); + attach_http_correlation(&mut request, request_id); trace.record_submit(request_id, request.method(), request.uri()); - let collect_before_send = - !p2_method_expects_body(request.method()) && body_completion.is_some(); + let body_completion = p2_body_completion_for_dispatch(request.method(), body_completion); + let collect_before_send = body_completion.is_some(); let (parts, body) = request.into_parts(); let request = http::Request::from_parts( parts, @@ -706,7 +755,7 @@ struct P3HttpTraceHooks(HttpLifecycleTrace); impl wasmtime_wasi_http::p3::WasiHttpHooks for P3HttpTraceHooks { fn send_request( &mut self, - request: http::Request< + mut request: http::Request< http_body_util::combinators::UnsyncBoxBody< bytes::Bytes, wasmtime_wasi_http::p3::bindings::http::types::ErrorCode, @@ -750,6 +799,7 @@ impl wasmtime_wasi_http::p3::WasiHttpHooks for P3HttpTraceHooks { let trace = self.0.clone(); let request_id = trace.next_request(); + attach_http_correlation(&mut request, request_id); trace.record_submit(request_id, request.method(), request.uri()); let (parts, body) = request.into_parts(); let request = http::Request::from_parts( @@ -1172,6 +1222,29 @@ mod tests { assert_eq!((retained >> 24) & 0xffff, 2); } + #[test] + fn http_lifecycle_ring_accepts_encoded_sequence_rollover() { + let journal = HttpLifecycleJournal::new(); + let before_rollover = HTTP_LIFECYCLE_SEQUENCE_MASK as usize; + journal.publish( + before_rollover - (HTTP_LIFECYCLE_CAP - 1), + 1, + HttpLifecyclePhase::Submit, + 0, + ); + journal.publish( + before_rollover + 1, + 2, + HttpLifecyclePhase::ResponseHead, + 200, + ); + + let retained = + journal.slots[(before_rollover + 1) % HTTP_LIFECYCLE_CAP].load(Ordering::Acquire); + assert_eq!(retained >> 40, 0); + assert_eq!((retained >> 24) & 0xffff, 2); + } + #[test] fn http_lifecycle_ring_remains_valid_during_concurrent_snapshots() { let journal = Arc::new(HttpLifecycleJournal::new()); @@ -1232,6 +1305,45 @@ mod tests { assert!(snapshot.contains("request=7 phase=send-error")); } + #[test] + fn http_lifecycle_correlation_distinguishes_same_path_requests() { + let mut first = http::Request::get("http://127.0.0.1:1234/same") + .body(()) + .unwrap(); + let mut second = http::Request::get("http://127.0.0.1:1234/same") + .body(()) + .unwrap(); + attach_http_correlation(&mut first, 41); + attach_http_correlation(&mut second, 42); + + assert_eq!(test_server_http_correlation(first.headers()), 41); + assert_eq!(test_server_http_correlation(second.headers()), 42); + + let uri = Arc::new(first.uri().clone()); + let arrivals = [41, 42].map(|request_id| { + let uri = uri.clone(); + thread::spawn(move || record_test_server_arrival(request_id, 1234, &uri)) + }); + for arrival in arrivals { + arrival.join().unwrap(); + } + let snapshot = test_server_http_trace().snapshot(); + assert!(snapshot.contains("request=41 phase=server-arrival")); + assert!(snapshot.contains("request=42 phase=server-arrival")); + } + + #[cfg(feature = "use-golem-wasmtime")] + #[test] + fn p2_body_method_drops_unused_completion_before_dispatch() { + let (sender, receiver) = tokio::sync::oneshot::channel::< + Result<(), wasmtime_wasi_http::p2::bindings::http::types::ErrorCode>, + >(); + let retained = p2_body_completion_for_dispatch(&http::Method::POST, Some(receiver)); + + assert!(retained.is_none()); + assert!(sender.is_closed()); + } + #[test] fn artifact_cache_stamp_must_not_be_older_than_output() -> anyhow::Result<()> { if test_drop_cache_enabled() { diff --git a/tests/common/test_server.rs b/tests/common/test_server.rs index 398456c1..1df69bc1 100644 --- a/tests/common/test_server.rs +++ b/tests/common/test_server.rs @@ -33,20 +33,25 @@ impl Drop for TestServerHandle { fn trace_http_lifecycle(router: Router, port: u16) -> Router { router.layer(axum::middleware::from_fn( move |request: Request, next: Next| async move { - super::record_test_server_arrival(port, request.uri()); + let request_id = super::test_server_http_correlation(request.headers()); + super::record_test_server_arrival(request_id, port, request.uri()); let (parts, body) = request.into_parts(); let request = Request::from_parts( parts, - Body::new(super::traced_test_server_body(body, port, "server-request")), + Body::new(super::traced_test_server_body( + body, + request_id, + "server-request", + )), ); let response = next.run(request).await; - super::record_test_server_response_head(port, response.status()); + super::record_test_server_response_head(request_id, response.status()); let (parts, body) = response.into_parts(); axum::response::Response::from_parts( parts, Body::new(super::traced_test_server_body( body, - port, + request_id, "server-response", )), ) From be0b1995ea4d39ce00568ca0808b28d2ef9f2161 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Da=CC=81vid=20Istva=CC=81n=20Bi=CC=81ro=CC=81?= Date: Sat, 29 Aug 2026 16:57:49 +0200 Subject: [PATCH 4/9] test(http): trace server socket ownership --- tests/common/mod.rs | 587 +++++++++++++++++++++++++++++++++++- tests/common/test_server.rs | 28 +- 2 files changed, 600 insertions(+), 15 deletions(-) diff --git a/tests/common/mod.rs b/tests/common/mod.rs index c629a3d2..3e2cf032 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -251,6 +251,7 @@ static HOST_TRACE: Mutex> = Mutex::new(Vec::new()); const HOST_TRACE_CAP: usize = 256 * 1024; static NEXT_HTTP_TRACE_INVOCATION: AtomicUsize = AtomicUsize::new(1); static NEXT_HTTP_TRACE_REQUEST: AtomicUsize = AtomicUsize::new(1); +static NEXT_HTTP_TRACE_CONNECTION: AtomicUsize = AtomicUsize::new(1); static TEST_SERVER_HTTP_TRACE: OnceLock = OnceLock::new(); const HTTP_LIFECYCLE_CAP: usize = 256; const HTTP_LIFECYCLE_SEQUENCE_MASK: u64 = 0x00ff_ffff; @@ -287,6 +288,20 @@ enum HttpLifecyclePhase { ServerResponseDrop = 26, TargetPath = 27, ServerPort = 28, + ServerConnectionAccept = 29, + ServerRequestConnection = 30, + ServerConnectionFirstRead = 31, + ServerConnectionReadEof = 32, + ServerConnectionReadError = 33, + ServerConnectionFirstWrite = 34, + ServerConnectionWriteError = 35, + ServerConnectionFlushError = 36, + ServerConnectionShutdown = 37, + ServerConnectionShutdownError = 38, + ServerConnectionDrop = 39, + ServerResponseWriteAfterFrame = 40, + ServerResponseWriteError = 41, + ServerResponsePendingAtDrop = 42, } impl HttpLifecyclePhase { @@ -320,6 +335,20 @@ impl HttpLifecyclePhase { 26 => "server-response-drop-before-terminal", 27 => "target-path-hash", 28 => "server-port", + 29 => "server-connection-accept", + 30 => "server-request-connection", + 31 => "server-connection-first-read", + 32 => "server-connection-read-eof", + 33 => "server-connection-read-error", + 34 => "server-connection-first-write", + 35 => "server-connection-write-error", + 36 => "server-connection-flush-error", + 37 => "server-connection-shutdown", + 38 => "server-connection-shutdown-error", + 39 => "server-connection-drop", + 40 => "server-response-write-after-frame", + 41 => "server-response-write-error", + 42 => "server-response-pending-at-drop", _ => "unknown", } } @@ -508,12 +537,310 @@ fn record_test_server_response_head(request_id: usize, status: http::StatusCode) ); } +fn record_test_server_connection( + request_id: usize, + connection: Option<&TracedTestServerConnection>, +) { + if let Some(connection) = connection { + connection.state.trace.record( + request_id, + HttpLifecyclePhase::ServerRequestConnection, + connection.state.connection, + ); + } +} + +pub(crate) fn traced_test_server_listener( + listener: tokio::net::TcpListener, +) -> TracedTestServerListener { + TracedTestServerListener { + listener, + trace: test_server_http_trace().clone(), + } +} + +pub(crate) struct TracedTestServerListener { + listener: tokio::net::TcpListener, + trace: HttpLifecycleTrace, +} + +impl axum::serve::Listener for TracedTestServerListener { + type Io = TracedTestServerIo; + type Addr = std::net::SocketAddr; + + async fn accept(&mut self) -> (Self::Io, Self::Addr) { + let (stream, peer) = + ::accept(&mut self.listener).await; + let connection = (NEXT_HTTP_TRACE_CONNECTION.fetch_add(1, Ordering::Relaxed) + % usize::from(u16::MAX)) + + 1; + let state = Arc::new(HttpConnectionState::new( + self.trace.clone(), + connection as u16, + )); + self.trace.record( + connection, + HttpLifecyclePhase::ServerConnectionAccept, + peer.port(), + ); + (TracedTestServerIo::new(stream, state), peer) + } + + fn local_addr(&self) -> std::io::Result { + self.listener.local_addr() + } +} + +#[derive(Clone)] +pub(crate) struct TracedTestServerConnection { + state: Arc, +} + +impl + axum::extract::connect_info::Connected< + axum::serve::IncomingStream<'_, TracedTestServerListener>, + > for TracedTestServerConnection +{ + fn connect_info(target: axum::serve::IncomingStream<'_, TracedTestServerListener>) -> Self { + Self { + state: target.io().state.clone(), + } + } +} + +struct HttpConnectionState { + trace: HttpLifecycleTrace, + connection: u16, + pending_response: AtomicUsize, +} + +impl HttpConnectionState { + fn new(trace: HttpLifecycleTrace, connection: u16) -> Self { + Self { + trace, + connection, + pending_response: AtomicUsize::new(0), + } + } + + fn arm_response(&self, request_id: usize) { + self.pending_response.store(request_id, Ordering::Release); + } + + fn record_response_write(&self, phase: HttpLifecyclePhase) { + let request_id = self.pending_response.swap(0, Ordering::AcqRel); + if request_id != 0 { + self.trace.record(request_id, phase, self.connection); + } + } +} + +pub(crate) struct TracedTestServerIo { + inner: T, + state: Arc, + saw_read: bool, + saw_write: bool, + read_terminal: bool, + write_terminal: bool, +} + +impl TracedTestServerIo { + fn new(inner: T, state: Arc) -> Self { + Self { + inner, + state, + saw_read: false, + saw_write: false, + read_terminal: false, + write_terminal: false, + } + } + + fn record_connection(&self, phase: HttpLifecyclePhase, detail: u16) { + self.state + .trace + .record(usize::from(self.state.connection), phase, detail); + } + + fn record_write_success(&mut self, bytes: usize) { + if !self.saw_write { + self.saw_write = true; + self.record_connection( + HttpLifecyclePhase::ServerConnectionFirstWrite, + bytes.min(usize::from(u16::MAX)) as u16, + ); + } + if bytes != 0 { + self.state + .record_response_write(HttpLifecyclePhase::ServerResponseWriteAfterFrame); + } + } + + fn record_write_error(&mut self, error: &std::io::Error) { + self.write_terminal = true; + self.state + .record_response_write(HttpLifecyclePhase::ServerResponseWriteError); + self.record_connection( + HttpLifecyclePhase::ServerConnectionWriteError, + error.raw_os_error().unwrap_or_default() as u16, + ); + } +} + +impl tokio::io::AsyncRead for TracedTestServerIo { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buffer: &mut tokio::io::ReadBuf<'_>, + ) -> Poll> { + let filled_before = buffer.filled().len(); + match Pin::new(&mut self.inner).poll_read(cx, buffer) { + Poll::Ready(Ok(())) => { + let bytes = buffer.filled().len().saturating_sub(filled_before); + if bytes != 0 && !self.saw_read { + self.saw_read = true; + self.record_connection( + HttpLifecyclePhase::ServerConnectionFirstRead, + bytes.min(usize::from(u16::MAX)) as u16, + ); + } else if bytes == 0 && !self.read_terminal { + self.read_terminal = true; + self.record_connection(HttpLifecyclePhase::ServerConnectionReadEof, 0); + } + Poll::Ready(Ok(())) + } + Poll::Ready(Err(error)) => { + self.read_terminal = true; + self.record_connection( + HttpLifecyclePhase::ServerConnectionReadError, + error.raw_os_error().unwrap_or_default() as u16, + ); + Poll::Ready(Err(error)) + } + Poll::Pending => Poll::Pending, + } + } +} + +impl tokio::io::AsyncWrite for TracedTestServerIo { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buffer: &[u8], + ) -> Poll> { + match Pin::new(&mut self.inner).poll_write(cx, buffer) { + Poll::Ready(Ok(bytes)) => { + self.record_write_success(bytes); + Poll::Ready(Ok(bytes)) + } + Poll::Ready(Err(error)) => { + self.record_write_error(&error); + Poll::Ready(Err(error)) + } + Poll::Pending => Poll::Pending, + } + } + + fn poll_write_vectored( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buffers: &[std::io::IoSlice<'_>], + ) -> Poll> { + match Pin::new(&mut self.inner).poll_write_vectored(cx, buffers) { + Poll::Ready(Ok(bytes)) => { + self.record_write_success(bytes); + Poll::Ready(Ok(bytes)) + } + Poll::Ready(Err(error)) => { + self.record_write_error(&error); + Poll::Ready(Err(error)) + } + Poll::Pending => Poll::Pending, + } + } + + fn is_write_vectored(&self) -> bool { + self.inner.is_write_vectored() + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + match Pin::new(&mut self.inner).poll_flush(cx) { + Poll::Ready(Err(error)) => { + self.write_terminal = true; + self.state + .record_response_write(HttpLifecyclePhase::ServerResponseWriteError); + self.record_connection( + HttpLifecyclePhase::ServerConnectionFlushError, + error.raw_os_error().unwrap_or_default() as u16, + ); + Poll::Ready(Err(error)) + } + result => result, + } + } + + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + match Pin::new(&mut self.inner).poll_shutdown(cx) { + Poll::Ready(Ok(())) => { + if !self.write_terminal { + self.write_terminal = true; + self.record_connection(HttpLifecyclePhase::ServerConnectionShutdown, 0); + } + Poll::Ready(Ok(())) + } + Poll::Ready(Err(error)) => { + self.write_terminal = true; + self.state + .record_response_write(HttpLifecyclePhase::ServerResponseWriteError); + self.record_connection( + HttpLifecyclePhase::ServerConnectionShutdownError, + error.raw_os_error().unwrap_or_default() as u16, + ); + Poll::Ready(Err(error)) + } + Poll::Pending => Poll::Pending, + } + } +} + +impl Drop for TracedTestServerIo { + fn drop(&mut self) { + self.state + .record_response_write(HttpLifecyclePhase::ServerResponsePendingAtDrop); + let detail = u16::from(self.saw_read) + | (u16::from(self.saw_write) << 1) + | (u16::from(self.read_terminal) << 2) + | (u16::from(self.write_terminal) << 3); + self.record_connection(HttpLifecyclePhase::ServerConnectionDrop, detail); + } +} + fn traced_test_server_body( body: B, request_id: usize, side: &'static str, ) -> TracedHttpBody { - TracedHttpBody::new(body, test_server_http_trace().clone(), request_id, side) + TracedHttpBody::new( + body, + test_server_http_trace().clone(), + request_id, + side, + None, + ) +} + +fn traced_test_server_response_body( + body: B, + request_id: usize, + connection: Option, +) -> TracedHttpBody { + TracedHttpBody::new( + body, + test_server_http_trace().clone(), + request_id, + "server-response", + connection.map(|connection| connection.state), + ) } /// A transparent body observer. It never polls ahead or adds an await: every host poll is @@ -523,17 +850,25 @@ struct TracedHttpBody { trace: HttpLifecycleTrace, request: usize, side: &'static str, + server_connection: Option>, saw_frame: bool, terminal: bool, } impl TracedHttpBody { - fn new(inner: B, trace: HttpLifecycleTrace, request: usize, side: &'static str) -> Self { + fn new( + inner: B, + trace: HttpLifecycleTrace, + request: usize, + side: &'static str, + server_connection: Option>, + ) -> Self { Self { inner: Box::pin(inner), trace, request, side, + server_connection, saw_frame: false, terminal: false, } @@ -578,11 +913,17 @@ impl HttpBody for TracedHttpBody { Poll::Ready(Some(Ok(frame))) => { if !this.saw_frame { this.saw_frame = true; - this.event(if frame.is_data() { + let outcome = if frame.is_data() { "first-data" } else { "first-trailers" - }); + }; + this.event(outcome); + if outcome == "first-data" + && let Some(connection) = &this.server_connection + { + connection.arm_response(this.request); + } } Poll::Ready(Some(Ok(frame))) } @@ -641,7 +982,8 @@ fn trace_p2_result( let (parts, body) = incoming.resp.into_parts(); incoming.resp = http::Response::from_parts( parts, - TracedHttpBody::new(body, trace.clone(), request_id, "response").boxed_unsync(), + TracedHttpBody::new(body, trace.clone(), request_id, "response", None) + .boxed_unsync(), ); Ok(incoming) } @@ -686,7 +1028,7 @@ impl wasmtime_wasi_http::p2::WasiHttpHooks for P2HttpTraceHooks { let (parts, body) = request.into_parts(); let request = http::Request::from_parts( parts, - TracedHttpBody::new(body, trace.clone(), request_id, "request").boxed_unsync(), + TracedHttpBody::new(body, trace.clone(), request_id, "request", None).boxed_unsync(), ); let handle = wasmtime_wasi::runtime::spawn(async move { let result = @@ -713,7 +1055,7 @@ impl wasmtime_wasi_http::p2::WasiHttpHooks for P2HttpTraceHooks { let (parts, body) = request.into_parts(); let request = http::Request::from_parts( parts, - TracedHttpBody::new(body, trace.clone(), request_id, "request").boxed_unsync(), + TracedHttpBody::new(body, trace.clone(), request_id, "request", None).boxed_unsync(), ); let handle = wasmtime_wasi::runtime::spawn(async move { let request = if collect_before_send { @@ -804,7 +1146,7 @@ impl wasmtime_wasi_http::p3::WasiHttpHooks for P3HttpTraceHooks { let (parts, body) = request.into_parts(); let request = http::Request::from_parts( parts, - TracedHttpBody::new(body, trace.clone(), request_id, "request").boxed_unsync(), + TracedHttpBody::new(body, trace.clone(), request_id, "request", None).boxed_unsync(), ); Box::new(async move { @@ -824,7 +1166,8 @@ impl wasmtime_wasi_http::p3::WasiHttpHooks for P3HttpTraceHooks { let (parts, body) = response.into_parts(); let response = http::Response::from_parts( parts, - TracedHttpBody::new(body, trace.clone(), request_id, "response").boxed_unsync(), + TracedHttpBody::new(body, trace.clone(), request_id, "response", None) + .boxed_unsync(), ); let io_trace = trace.clone(); let io = Box::new(async move { @@ -1204,6 +1547,92 @@ impl Drop for TestCacheLock { mod tests { use super::*; use test_r::test; + use tokio::io::{AsyncRead as _, AsyncWrite as _}; + + #[derive(Default)] + struct ScriptedIoCalls { + reads: AtomicUsize, + writes: AtomicUsize, + vectored_writes: AtomicUsize, + flushes: AtomicUsize, + shutdowns: AtomicUsize, + } + + struct ScriptedIo { + calls: Arc, + fail_write: bool, + } + + impl tokio::io::AsyncRead for ScriptedIo { + fn poll_read( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + buffer: &mut tokio::io::ReadBuf<'_>, + ) -> Poll> { + self.calls.reads.fetch_add(1, Ordering::Relaxed); + buffer.put_slice(b"r"); + Poll::Ready(Ok(())) + } + } + + impl tokio::io::AsyncWrite for ScriptedIo { + fn poll_write( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + buffer: &[u8], + ) -> Poll> { + self.calls.writes.fetch_add(1, Ordering::Relaxed); + if self.fail_write { + Poll::Ready(Err(std::io::Error::other("scripted write failure"))) + } else { + Poll::Ready(Ok(buffer.len())) + } + } + + fn poll_write_vectored( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + buffers: &[std::io::IoSlice<'_>], + ) -> Poll> { + self.calls.vectored_writes.fetch_add(1, Ordering::Relaxed); + if self.fail_write { + Poll::Ready(Err(std::io::Error::other("scripted write failure"))) + } else { + Poll::Ready(Ok(buffers.iter().map(|buffer| buffer.len()).sum())) + } + } + + fn is_write_vectored(&self) -> bool { + true + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + self.calls.flushes.fetch_add(1, Ordering::Relaxed); + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + self.calls.shutdowns.fetch_add(1, Ordering::Relaxed); + Poll::Ready(Ok(())) + } + } + + fn scripted_connection( + trace: HttpLifecycleTrace, + connection: u16, + fail_write: bool, + ) -> (TracedTestServerIo, Arc) { + let calls = Arc::new(ScriptedIoCalls::default()); + let state = Arc::new(HttpConnectionState::new(trace, connection)); + let io = TracedTestServerIo::new( + ScriptedIo { + calls: calls.clone(), + fail_write, + }, + state, + ); + (io, calls) + } #[test] fn http_lifecycle_ring_rejects_delayed_old_generation() { @@ -1285,7 +1714,7 @@ mod tests { fn http_lifecycle_empty_body_drop_is_terminal() { let trace = HttpLifecycleTrace::new(); let body = http_body_util::Empty::::new(); - drop(TracedHttpBody::new(body, trace.clone(), 1, "request")); + drop(TracedHttpBody::new(body, trace.clone(), 1, "request", None)); let snapshot = trace.snapshot(); assert!(!snapshot.contains("drop-before-terminal")); @@ -1332,6 +1761,144 @@ mod tests { assert!(snapshot.contains("request=42 phase=server-arrival")); } + #[test] + fn http_lifecycle_server_io_delegates_each_poll_once() { + let trace = HttpLifecycleTrace::new(); + let (mut io, calls) = scripted_connection(trace.clone(), 7, false); + let waker = futures::task::noop_waker(); + let mut cx = Context::from_waker(&waker); + + let mut storage = [0; 4]; + let mut read = tokio::io::ReadBuf::new(&mut storage); + assert!(Pin::new(&mut io).poll_read(&mut cx, &mut read).is_ready()); + assert_eq!(read.filled(), b"r"); + assert_eq!(calls.reads.load(Ordering::Relaxed), 1); + + assert!(matches!( + Pin::new(&mut io).poll_write(&mut cx, b"abc"), + Poll::Ready(Ok(3)) + )); + assert_eq!(calls.writes.load(Ordering::Relaxed), 1); + + let buffers = [std::io::IoSlice::new(b"d"), std::io::IoSlice::new(b"ef")]; + assert!(matches!( + Pin::new(&mut io).poll_write_vectored(&mut cx, &buffers), + Poll::Ready(Ok(3)) + )); + assert_eq!(calls.vectored_writes.load(Ordering::Relaxed), 1); + assert!(io.is_write_vectored()); + + assert!(matches!( + Pin::new(&mut io).poll_flush(&mut cx), + Poll::Ready(Ok(())) + )); + assert_eq!(calls.flushes.load(Ordering::Relaxed), 1); + assert!(matches!( + Pin::new(&mut io).poll_shutdown(&mut cx), + Poll::Ready(Ok(())) + )); + assert_eq!(calls.shutdowns.load(Ordering::Relaxed), 1); + + let snapshot = trace.snapshot(); + assert!(snapshot.contains("request=7 phase=server-connection-first-read detail=1")); + assert!(snapshot.contains("request=7 phase=server-connection-first-write detail=3")); + assert!(snapshot.contains("request=7 phase=server-connection-shutdown")); + } + + #[test] + fn http_lifecycle_server_response_frame_arms_exactly_one_correlated_write() { + let trace = HttpLifecycleTrace::new(); + let state = Arc::new(HttpConnectionState::new(trace.clone(), 9)); + let body = http_body_util::Full::new(bytes::Bytes::from_static(b"body")); + let mut body = Box::pin(TracedHttpBody::new( + body, + trace.clone(), + 41, + "server-response", + Some(state.clone()), + )); + let waker = futures::task::noop_waker(); + let mut cx = Context::from_waker(&waker); + assert!(matches!( + body.as_mut().poll_frame(&mut cx), + Poll::Ready(Some(Ok(_))) + )); + assert_eq!(state.pending_response.load(Ordering::Acquire), 41); + + let calls = Arc::new(ScriptedIoCalls::default()); + let mut io = TracedTestServerIo::new( + ScriptedIo { + calls, + fail_write: false, + }, + state.clone(), + ); + let buffers = [ + std::io::IoSlice::new(b"head"), + std::io::IoSlice::new(b"body"), + ]; + assert!(matches!( + Pin::new(&mut io).poll_write_vectored(&mut cx, &buffers), + Poll::Ready(Ok(8)) + )); + assert_eq!(state.pending_response.load(Ordering::Acquire), 0); + assert!(matches!( + Pin::new(&mut io).poll_write(&mut cx, b"later"), + Poll::Ready(Ok(5)) + )); + + let snapshot = trace.snapshot(); + assert_eq!( + snapshot + .matches("request=41 phase=server-response-write-after-frame detail=9") + .count(), + 1 + ); + } + + #[test] + fn http_lifecycle_server_response_write_error_preserves_correlation() { + let trace = HttpLifecycleTrace::new(); + let (mut io, _) = scripted_connection(trace.clone(), 11, true); + io.state.arm_response(42); + let waker = futures::task::noop_waker(); + let mut cx = Context::from_waker(&waker); + + assert!(matches!( + Pin::new(&mut io).poll_write(&mut cx, b"response"), + Poll::Ready(Err(_)) + )); + let snapshot = trace.snapshot(); + assert!(snapshot.contains("request=42 phase=server-response-write-error detail=11")); + assert!(snapshot.contains("request=11 phase=server-connection-write-error")); + } + + #[test] + fn http_lifecycle_server_connection_drop_reports_armed_response() { + let trace = HttpLifecycleTrace::new(); + let (io, _) = scripted_connection(trace.clone(), 13, false); + io.state.arm_response(43); + drop(io); + + let snapshot = trace.snapshot(); + assert!(snapshot.contains("request=43 phase=server-response-pending-at-drop detail=13")); + assert!(snapshot.contains("request=13 phase=server-connection-drop")); + } + + #[test] + fn http_lifecycle_server_connection_maps_multiple_requests_without_conflation() { + let trace = HttpLifecycleTrace::new(); + let connection = TracedTestServerConnection { + state: Arc::new(HttpConnectionState::new(trace.clone(), 17)), + }; + record_test_server_connection(51, Some(&connection)); + record_test_server_connection(52, Some(&connection)); + + let snapshot = trace.snapshot(); + assert!(snapshot.contains("request=51 phase=server-request-connection detail=17")); + assert!(snapshot.contains("request=52 phase=server-request-connection detail=17")); + } + #[cfg(feature = "use-golem-wasmtime")] #[test] fn p2_body_method_drops_unused_completion_before_dispatch() { diff --git a/tests/common/test_server.rs b/tests/common/test_server.rs index 1df69bc1..fb60664e 100644 --- a/tests/common/test_server.rs +++ b/tests/common/test_server.rs @@ -1,6 +1,6 @@ use axum::body::Body; use axum::extract::Request; -use axum::extract::{Multipart, Path}; +use axum::extract::{ConnectInfo, Multipart, Path}; use axum::http::HeaderMap; use axum::middleware::Next; use axum::response::{AppendHeaders, IntoResponse}; @@ -34,7 +34,13 @@ fn trace_http_lifecycle(router: Router, port: u16) -> Router { router.layer(axum::middleware::from_fn( move |request: Request, next: Next| async move { let request_id = super::test_server_http_correlation(request.headers()); + let connection = request + .extensions() + .get::>() + .map(|connection| connection.0.clone()); + let trace_response_write = request.version() == http::Version::HTTP_11; super::record_test_server_arrival(request_id, port, request.uri()); + super::record_test_server_connection(request_id, connection.as_ref()); let (parts, body) = request.into_parts(); let request = Request::from_parts( parts, @@ -49,10 +55,10 @@ fn trace_http_lifecycle(router: Router, port: u16) -> Router { let (parts, body) = response.into_parts(); axum::response::Response::from_parts( parts, - Body::new(super::traced_test_server_body( + Body::new(super::traced_test_server_response_body( body, request_id, - "server-response", + trace_response_write.then_some(connection).flatten(), )), ) }, @@ -260,7 +266,13 @@ pub async fn start_test_server() -> (u16, TestServerHandle) { ); let router = trace_http_lifecycle(router, host_http_port); - axum::serve(listener, router).await.unwrap(); + let listener = super::traced_test_server_listener(listener); + axum::serve( + listener, + router.into_make_service_with_connect_info::(), + ) + .await + .unwrap(); }); (host_http_port, TestServerHandle::new(handle)) @@ -296,7 +308,13 @@ pub async fn start_abort_test_server() -> (u16, TestServerHandle, mpsc::Unbounde ) .route("/abort-ready", ready); let router = trace_http_lifecycle(router, port); - axum::serve(listener, router).await.unwrap(); + let listener = super::traced_test_server_listener(listener); + axum::serve( + listener, + router.into_make_service_with_connect_info::(), + ) + .await + .unwrap(); }); (port, TestServerHandle::new(handle), arrived_rx) From b06b4da23a70d9c2f648b3cd8e40e54c324374cc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Da=CC=81vid=20Istva=CC=81n=20Bi=CC=81ro=CC=81?= Date: Sat, 29 Aug 2026 17:05:11 +0200 Subject: [PATCH 5/9] test(http): preserve ambiguous response correlation --- tests/common/mod.rs | 104 ++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 101 insertions(+), 3 deletions(-) diff --git a/tests/common/mod.rs b/tests/common/mod.rs index 3e2cf032..68eabedc 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -302,6 +302,7 @@ enum HttpLifecyclePhase { ServerResponseWriteAfterFrame = 40, ServerResponseWriteError = 41, ServerResponsePendingAtDrop = 42, + ServerResponseCorrelationOverlap = 43, } impl HttpLifecyclePhase { @@ -349,6 +350,7 @@ impl HttpLifecyclePhase { 40 => "server-response-write-after-frame", 41 => "server-response-write-error", 42 => "server-response-pending-at-drop", + 43 => "server-response-correlation-overlap", _ => "unknown", } } @@ -624,7 +626,18 @@ impl HttpConnectionState { } fn arm_response(&self, request_id: usize) { - self.pending_response.store(request_id, Ordering::Release); + if let Err(pending_request) = self.pending_response.compare_exchange( + 0, + request_id, + Ordering::AcqRel, + Ordering::Acquire, + ) { + self.trace.record( + request_id, + HttpLifecyclePhase::ServerResponseCorrelationOverlap, + pending_request.min(usize::from(u16::MAX)) as u16, + ); + } } fn record_response_write(&self, phase: HttpLifecyclePhase) { @@ -1547,7 +1560,7 @@ impl Drop for TestCacheLock { mod tests { use super::*; use test_r::test; - use tokio::io::{AsyncRead as _, AsyncWrite as _}; + use tokio::io::{AsyncRead as _, AsyncReadExt as _, AsyncWrite as _, AsyncWriteExt as _}; #[derive(Default)] struct ScriptedIoCalls { @@ -1886,7 +1899,7 @@ mod tests { } #[test] - fn http_lifecycle_server_connection_maps_multiple_requests_without_conflation() { + fn http_lifecycle_server_connection_maps_multiple_requests_to_same_connection() { let trace = HttpLifecycleTrace::new(); let connection = TracedTestServerConnection { state: Arc::new(HttpConnectionState::new(trace.clone(), 17)), @@ -1899,6 +1912,91 @@ mod tests { assert!(snapshot.contains("request=52 phase=server-request-connection detail=17")); } + #[test] + fn http_lifecycle_server_response_overlap_is_explicit_and_non_overwriting() { + let trace = HttpLifecycleTrace::new(); + let state = HttpConnectionState::new(trace.clone(), 19); + state.arm_response(61); + state.arm_response(62); + + assert_eq!(state.pending_response.load(Ordering::Acquire), 61); + let snapshot = trace.snapshot(); + assert!( + snapshot.contains("request=62 phase=server-response-correlation-overlap detail=61") + ); + } + + #[test] + async fn http_lifecycle_axum_pipeline_preserves_response_correlation() { + use axum::body::Body; + use axum::extract::{ConnectInfo, Request}; + use axum::routing::any; + use bytes::Bytes; + use http_body_util::Full; + + let trace = HttpLifecycleTrace::new(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let listener = TracedTestServerListener { + listener, + trace: trace.clone(), + }; + let router = axum::Router::new().route( + "/", + any( + |ConnectInfo(connection): ConnectInfo, + request: Request| async move { + let request_id = test_server_http_correlation(request.headers()); + record_test_server_connection(request_id, Some(&connection)); + axum::response::Response::new(Body::new(TracedHttpBody::new( + Full::new(Bytes::from_static(b"ok")), + connection.state.trace.clone(), + request_id, + "server-response", + Some(connection.state), + ))) + }, + ), + ); + let server = tokio::spawn(async move { + axum::serve( + listener, + router.into_make_service_with_connect_info::(), + ) + .await + .unwrap(); + }); + + let mut client = tokio::net::TcpStream::connect(address).await.unwrap(); + client + .write_all( + b"GET / HTTP/1.1\r\nHost: localhost\r\nx-wrq-http-trace-id: 71\r\n\r\n\ + GET / HTTP/1.1\r\nHost: localhost\r\nx-wrq-http-trace-id: 72\r\nConnection: close\r\n\r\n", + ) + .await + .unwrap(); + let mut response = Vec::new(); + client.read_to_end(&mut response).await.unwrap(); + server.abort(); + + assert_eq!( + String::from_utf8_lossy(&response) + .matches("HTTP/1.1 200 OK") + .count(), + 2 + ); + let snapshot = trace.snapshot(); + for request_id in [71, 72] { + assert!(snapshot.contains(&format!( + "request={request_id} phase=server-request-connection" + ))); + assert!(snapshot.contains(&format!( + "request={request_id} phase=server-response-write-after-frame" + ))); + } + assert!(!snapshot.contains("server-response-correlation-overlap")); + } + #[cfg(feature = "use-golem-wasmtime")] #[test] fn p2_body_method_drops_unused_completion_before_dispatch() { From 3f51d2c400a950d239dae23a544978fc60b5b1ec Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Da=CC=81vid=20Istva=CC=81n=20Bi=CC=81ro=CC=81?= Date: Sat, 29 Aug 2026 20:47:39 +0200 Subject: [PATCH 6/9] test(http): account response body and socket bytes --- tests/common/mod.rs | 553 +++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 519 insertions(+), 34 deletions(-) diff --git a/tests/common/mod.rs b/tests/common/mod.rs index 68eabedc..063e1835 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -3,6 +3,7 @@ pub mod test_server; use crate::common::WasmSource::Precompiled; use anyhow::anyhow; +use bytes::Buf; use camino::{Utf8Path, Utf8PathBuf}; use camino_tempfile::{NamedUtf8TempFile, Utf8TempDir}; use futures::FutureExt; @@ -303,6 +304,18 @@ enum HttpLifecyclePhase { ServerResponseWriteError = 41, ServerResponsePendingAtDrop = 42, ServerResponseCorrelationOverlap = 43, + ServerResponseBodyExpectedBytes = 44, + ServerResponseBodyExpectedUnknown = 45, + ServerResponseBodyPolledBytes = 46, + ServerResponseWritePartial = 47, + ServerResponseWriteBoundaryBytes = 48, + ServerResponseFlushBytes = 49, + ServerResponseTerminalBytes = 50, + ServerResponseByteCountOverflow = 51, + ServerConnectionFlushPending = 52, + ServerConnectionFlushOk = 53, + ServerConnectionShutdownPending = 54, + ServerResponseCorrelationBoundary = 55, } impl HttpLifecyclePhase { @@ -351,11 +364,44 @@ impl HttpLifecyclePhase { 41 => "server-response-write-error", 42 => "server-response-pending-at-drop", 43 => "server-response-correlation-overlap", + 44 => "server-response-body-expected-bytes", + 45 => "server-response-body-expected-unknown", + 46 => "server-response-body-polled-bytes", + 47 => "server-response-write-partial", + 48 => "server-response-write-boundary-bytes", + 49 => "server-response-flush-bytes", + 50 => "server-response-terminal-bytes", + 51 => "server-response-byte-count-overflow", + 52 => "server-connection-flush-pending", + 53 => "server-connection-flush-ok", + 54 => "server-connection-shutdown-pending", + 55 => "server-response-correlation-boundary", _ => "unknown", } } } +const HTTP_BYTE_COUNT_EXPECTED: u16 = 1; +const HTTP_BYTE_COUNT_POLLED: u16 = 2; +const HTTP_BYTE_COUNT_WRITE: u16 = 3; + +fn record_http_byte_count( + trace: &HttpLifecycleTrace, + request: usize, + phase: HttpLifecyclePhase, + count: u64, + overflow_kind: u16, +) { + if count > u64::from(u16::MAX) { + trace.record( + request, + HttpLifecyclePhase::ServerResponseByteCountOverflow, + overflow_kind, + ); + } + trace.record(request, phase, count.min(u64::from(u16::MAX)) as u16); +} + struct HttpLifecycleJournal { next_event: AtomicUsize, slots: [std::sync::atomic::AtomicU64; HTTP_LIFECYCLE_CAP], @@ -640,14 +686,19 @@ impl HttpConnectionState { } } - fn record_response_write(&self, phase: HttpLifecyclePhase) { - let request_id = self.pending_response.swap(0, Ordering::AcqRel); - if request_id != 0 { - self.trace.record(request_id, phase, self.connection); + fn take_pending_response(&self) -> Option { + match self.pending_response.swap(0, Ordering::AcqRel) { + 0 => None, + request_id => Some(request_id), } } } +struct ActiveResponseWrite { + request_id: usize, + accepted_bytes: u64, +} + pub(crate) struct TracedTestServerIo { inner: T, state: Arc, @@ -655,6 +706,9 @@ pub(crate) struct TracedTestServerIo { saw_write: bool, read_terminal: bool, write_terminal: bool, + active_response: Option, + flush_pending: bool, + shutdown_pending: bool, } impl TracedTestServerIo { @@ -666,6 +720,9 @@ impl TracedTestServerIo { saw_write: false, read_terminal: false, write_terminal: false, + active_response: None, + flush_pending: false, + shutdown_pending: false, } } @@ -675,24 +732,106 @@ impl TracedTestServerIo { .record(usize::from(self.state.connection), phase, detail); } - fn record_write_success(&mut self, bytes: usize) { + fn record_response_bytes(&self, request_id: usize, phase: HttpLifecyclePhase, bytes: u64) { + record_http_byte_count( + &self.state.trace, + request_id, + phase, + bytes, + HTTP_BYTE_COUNT_WRITE, + ); + } + + fn activate_pending_response(&mut self, first_accepted_bytes: Option) -> Option { + let pending = self.state.take_pending_response(); + if let Some(request_id) = pending { + match self.active_response.take() { + Some(active) if active.request_id != request_id => { + self.record_response_bytes( + active.request_id, + HttpLifecyclePhase::ServerResponseWriteBoundaryBytes, + active.accepted_bytes, + ); + self.state.trace.record( + active.request_id, + HttpLifecyclePhase::ServerResponseCorrelationBoundary, + request_id.min(usize::from(u16::MAX)) as u16, + ); + } + Some(active) => { + self.active_response = Some(active); + return Some(request_id); + } + None => {} + } + if let Some(first_accepted_bytes) = first_accepted_bytes { + self.record_response_bytes( + request_id, + HttpLifecyclePhase::ServerResponseWriteAfterFrame, + first_accepted_bytes as u64, + ); + } + self.active_response = Some(ActiveResponseWrite { + request_id, + accepted_bytes: 0, + }); + } + self.active_response + .as_ref() + .map(|active| active.request_id) + } + + fn snapshot_active_response(&mut self, phase: HttpLifecyclePhase) { + if let Some(active) = self.active_response.take() { + self.record_response_bytes(active.request_id, phase, active.accepted_bytes); + } + } + + fn record_pending_response(&self, phase: HttpLifecyclePhase) { + if let Some(request_id) = self.state.take_pending_response() { + self.state + .trace + .record(request_id, phase, self.state.connection); + } + } + + fn record_write_success(&mut self, offered: usize, accepted: usize) { if !self.saw_write { self.saw_write = true; self.record_connection( HttpLifecyclePhase::ServerConnectionFirstWrite, - bytes.min(usize::from(u16::MAX)) as u16, + accepted.min(usize::from(u16::MAX)) as u16, ); } - if bytes != 0 { - self.state - .record_response_write(HttpLifecyclePhase::ServerResponseWriteAfterFrame); + if let Some(request_id) = self.activate_pending_response(Some(accepted)) { + if let Some(active) = &mut self.active_response { + active.accepted_bytes = active.accepted_bytes.saturating_add(accepted as u64); + } + if accepted < offered { + self.record_response_bytes( + request_id, + HttpLifecyclePhase::ServerResponseWritePartial, + offered.saturating_sub(accepted) as u64, + ); + } } } fn record_write_error(&mut self, error: &std::io::Error) { self.write_terminal = true; - self.state - .record_response_write(HttpLifecyclePhase::ServerResponseWriteError); + self.activate_pending_response(None); + if let Some(request_id) = self + .active_response + .as_ref() + .map(|active| active.request_id) + { + self.state.trace.record( + request_id, + HttpLifecyclePhase::ServerResponseWriteError, + self.state.connection, + ); + } + self.snapshot_active_response(HttpLifecyclePhase::ServerResponseTerminalBytes); self.record_connection( HttpLifecyclePhase::ServerConnectionWriteError, error.raw_os_error().unwrap_or_default() as u16, @@ -743,7 +882,7 @@ impl tokio::io::AsyncWrite for TracedTestServe ) -> Poll> { match Pin::new(&mut self.inner).poll_write(cx, buffer) { Poll::Ready(Ok(bytes)) => { - self.record_write_success(bytes); + self.record_write_success(buffer.len(), bytes); Poll::Ready(Ok(bytes)) } Poll::Ready(Err(error)) => { @@ -761,7 +900,10 @@ impl tokio::io::AsyncWrite for TracedTestServe ) -> Poll> { match Pin::new(&mut self.inner).poll_write_vectored(cx, buffers) { Poll::Ready(Ok(bytes)) => { - self.record_write_success(bytes); + let offered = buffers + .iter() + .fold(0usize, |total, buffer| total.saturating_add(buffer.len())); + self.record_write_success(offered, bytes); Poll::Ready(Ok(bytes)) } Poll::Ready(Err(error)) => { @@ -778,48 +920,93 @@ impl tokio::io::AsyncWrite for TracedTestServe fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { match Pin::new(&mut self.inner).poll_flush(cx) { + Poll::Ready(Ok(())) => { + self.flush_pending = false; + self.snapshot_active_response(HttpLifecyclePhase::ServerResponseFlushBytes); + self.record_connection(HttpLifecyclePhase::ServerConnectionFlushOk, 0); + Poll::Ready(Ok(())) + } Poll::Ready(Err(error)) => { + self.flush_pending = false; self.write_terminal = true; - self.state - .record_response_write(HttpLifecyclePhase::ServerResponseWriteError); + if let Some(request_id) = self + .active_response + .as_ref() + .map(|active| active.request_id) + { + self.state.trace.record( + request_id, + HttpLifecyclePhase::ServerResponseWriteError, + self.state.connection, + ); + } + self.snapshot_active_response(HttpLifecyclePhase::ServerResponseTerminalBytes); + self.record_pending_response(HttpLifecyclePhase::ServerResponseWriteError); self.record_connection( HttpLifecyclePhase::ServerConnectionFlushError, error.raw_os_error().unwrap_or_default() as u16, ); Poll::Ready(Err(error)) } - result => result, + Poll::Pending => { + if !self.flush_pending { + self.flush_pending = true; + self.record_connection(HttpLifecyclePhase::ServerConnectionFlushPending, 0); + } + Poll::Pending + } } } fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { match Pin::new(&mut self.inner).poll_shutdown(cx) { Poll::Ready(Ok(())) => { + self.shutdown_pending = false; if !self.write_terminal { self.write_terminal = true; + self.snapshot_active_response(HttpLifecyclePhase::ServerResponseTerminalBytes); + self.record_pending_response(HttpLifecyclePhase::ServerResponsePendingAtDrop); self.record_connection(HttpLifecyclePhase::ServerConnectionShutdown, 0); } Poll::Ready(Ok(())) } Poll::Ready(Err(error)) => { + self.shutdown_pending = false; self.write_terminal = true; - self.state - .record_response_write(HttpLifecyclePhase::ServerResponseWriteError); + if let Some(request_id) = self + .active_response + .as_ref() + .map(|active| active.request_id) + { + self.state.trace.record( + request_id, + HttpLifecyclePhase::ServerResponseWriteError, + self.state.connection, + ); + } + self.snapshot_active_response(HttpLifecyclePhase::ServerResponseTerminalBytes); + self.record_pending_response(HttpLifecyclePhase::ServerResponseWriteError); self.record_connection( HttpLifecyclePhase::ServerConnectionShutdownError, error.raw_os_error().unwrap_or_default() as u16, ); Poll::Ready(Err(error)) } - Poll::Pending => Poll::Pending, + Poll::Pending => { + if !self.shutdown_pending { + self.shutdown_pending = true; + self.record_connection(HttpLifecyclePhase::ServerConnectionShutdownPending, 0); + } + Poll::Pending + } } } } impl Drop for TracedTestServerIo { fn drop(&mut self) { - self.state - .record_response_write(HttpLifecyclePhase::ServerResponsePendingAtDrop); + self.snapshot_active_response(HttpLifecyclePhase::ServerResponseTerminalBytes); + self.record_pending_response(HttpLifecyclePhase::ServerResponsePendingAtDrop); let detail = u16::from(self.saw_read) | (u16::from(self.saw_write) << 1) | (u16::from(self.read_terminal) << 2) @@ -857,7 +1044,7 @@ fn traced_test_server_response_body( } /// A transparent body observer. It never polls ahead or adds an await: every host poll is -/// delegated exactly once, with only the first frame and the terminal outcome recorded. +/// delegated exactly once while frame byte totals and the terminal outcome are recorded. struct TracedHttpBody { inner: Pin>, trace: HttpLifecycleTrace, @@ -866,6 +1053,9 @@ struct TracedHttpBody { server_connection: Option>, saw_frame: bool, terminal: bool, + expected_body_bytes: Option, + polled_body_bytes: u64, + body_bytes_recorded: bool, } impl TracedHttpBody { @@ -876,7 +1066,10 @@ impl TracedHttpBody { side: &'static str, server_connection: Option>, ) -> Self { - Self { + let expected_body_bytes = (side == "server-response") + .then(|| inner.size_hint().exact()) + .flatten(); + let body = Self { inner: Box::pin(inner), trace, request, @@ -884,7 +1077,28 @@ impl TracedHttpBody { server_connection, saw_frame: false, terminal: false, + expected_body_bytes, + polled_body_bytes: 0, + body_bytes_recorded: false, + }; + if side == "server-response" { + if let Some(expected) = body.expected_body_bytes { + record_http_byte_count( + &body.trace, + request, + HttpLifecyclePhase::ServerResponseBodyExpectedBytes, + expected, + HTTP_BYTE_COUNT_EXPECTED, + ); + } else { + body.trace.record( + request, + HttpLifecyclePhase::ServerResponseBodyExpectedUnknown, + 0, + ); + } } + body } fn event(&self, outcome: &'static str) { @@ -911,6 +1125,27 @@ impl TracedHttpBody { }; self.trace.record(self.request, phase, 0); } + + fn record_body_bytes(&mut self) { + if self.side == "server-response" && !self.body_bytes_recorded { + self.body_bytes_recorded = true; + record_http_byte_count( + &self.trace, + self.request, + HttpLifecyclePhase::ServerResponseBodyPolledBytes, + self.polled_body_bytes, + HTTP_BYTE_COUNT_POLLED, + ); + } + } + + fn record_terminal(&mut self, outcome: &'static str) { + if !self.terminal { + self.terminal = true; + self.event(outcome); + } + self.record_body_bytes(); + } } impl HttpBody for TracedHttpBody { @@ -924,6 +1159,11 @@ impl HttpBody for TracedHttpBody { let this = self.get_mut(); match this.inner.as_mut().poll_frame(cx) { Poll::Ready(Some(Ok(frame))) => { + if let Some(data) = frame.data_ref() { + this.polled_body_bytes = this + .polled_body_bytes + .saturating_add(data.remaining() as u64); + } if !this.saw_frame { this.saw_frame = true; let outcome = if frame.is_data() { @@ -938,16 +1178,17 @@ impl HttpBody for TracedHttpBody { connection.arm_response(this.request); } } + if this.inner.as_ref().is_end_stream() { + this.record_terminal("eof"); + } Poll::Ready(Some(Ok(frame))) } Poll::Ready(Some(Err(error))) => { - this.terminal = true; - this.event("error"); + this.record_terminal("error"); Poll::Ready(Some(Err(error))) } Poll::Ready(None) => { - this.terminal = true; - this.event("eof"); + this.record_terminal("eof"); Poll::Ready(None) } Poll::Pending => Poll::Pending, @@ -968,6 +1209,7 @@ impl Drop for TracedHttpBody { if !self.terminal && !self.inner.as_ref().is_end_stream() { self.event("drop-before-terminal"); } + self.record_body_bytes(); } } @@ -1559,6 +1801,7 @@ impl Drop for TestCacheLock { #[cfg(test)] mod tests { use super::*; + use std::collections::VecDeque; use test_r::test; use tokio::io::{AsyncRead as _, AsyncReadExt as _, AsyncWrite as _, AsyncWriteExt as _}; @@ -1574,6 +1817,9 @@ mod tests { struct ScriptedIo { calls: Arc, fail_write: bool, + accepted_writes: VecDeque, + flush_pending_once: bool, + shutdown_pending_once: bool, } impl tokio::io::AsyncRead for ScriptedIo { @@ -1590,7 +1836,7 @@ mod tests { impl tokio::io::AsyncWrite for ScriptedIo { fn poll_write( - self: Pin<&mut Self>, + mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buffer: &[u8], ) -> Poll> { @@ -1598,12 +1844,16 @@ mod tests { if self.fail_write { Poll::Ready(Err(std::io::Error::other("scripted write failure"))) } else { - Poll::Ready(Ok(buffer.len())) + Poll::Ready(Ok(self + .accepted_writes + .pop_front() + .unwrap_or(buffer.len()) + .min(buffer.len()))) } } fn poll_write_vectored( - self: Pin<&mut Self>, + mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buffers: &[std::io::IoSlice<'_>], ) -> Poll> { @@ -1611,7 +1861,12 @@ mod tests { if self.fail_write { Poll::Ready(Err(std::io::Error::other("scripted write failure"))) } else { - Poll::Ready(Ok(buffers.iter().map(|buffer| buffer.len()).sum())) + let offered = buffers.iter().map(|buffer| buffer.len()).sum(); + Poll::Ready(Ok(self + .accepted_writes + .pop_front() + .unwrap_or(offered) + .min(offered))) } } @@ -1619,13 +1874,24 @@ mod tests { true } - fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { self.calls.flushes.fetch_add(1, Ordering::Relaxed); + if std::mem::take(&mut self.flush_pending_once) { + cx.waker().wake_by_ref(); + return Poll::Pending; + } Poll::Ready(Ok(())) } - fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + fn poll_shutdown( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll> { self.calls.shutdowns.fetch_add(1, Ordering::Relaxed); + if std::mem::take(&mut self.shutdown_pending_once) { + cx.waker().wake_by_ref(); + return Poll::Pending; + } Poll::Ready(Ok(())) } } @@ -1641,6 +1907,9 @@ mod tests { ScriptedIo { calls: calls.clone(), fail_write, + accepted_writes: VecDeque::new(), + flush_pending_once: false, + shutdown_pending_once: false, }, state, ); @@ -1843,6 +2112,9 @@ mod tests { ScriptedIo { calls, fail_write: false, + accepted_writes: VecDeque::new(), + flush_pending_once: false, + shutdown_pending_once: false, }, state.clone(), ); @@ -1863,10 +2135,162 @@ mod tests { let snapshot = trace.snapshot(); assert_eq!( snapshot - .matches("request=41 phase=server-response-write-after-frame detail=9") + .matches("request=41 phase=server-response-write-after-frame detail=8") .count(), 1 ); + assert!(snapshot.contains("request=41 phase=server-response-body-expected-bytes detail=4")); + assert!(snapshot.contains("request=41 phase=server-response-body-polled-bytes detail=4")); + } + + #[test] + fn http_lifecycle_server_response_counts_partial_and_cumulative_writes() { + let trace = HttpLifecycleTrace::new(); + let (mut io, calls) = scripted_connection(trace.clone(), 10, false); + io.inner.accepted_writes = VecDeque::from([4, 6]); + io.state.arm_response(44); + let waker = futures::task::noop_waker(); + let mut cx = Context::from_waker(&waker); + + assert!(matches!( + Pin::new(&mut io).poll_write(&mut cx, b"0123456789"), + Poll::Ready(Ok(4)) + )); + assert!(matches!( + Pin::new(&mut io).poll_write(&mut cx, b"456789"), + Poll::Ready(Ok(6)) + )); + assert!(matches!( + Pin::new(&mut io).poll_flush(&mut cx), + Poll::Ready(Ok(())) + )); + assert_eq!(calls.writes.load(Ordering::Relaxed), 2); + assert_eq!(calls.flushes.load(Ordering::Relaxed), 1); + + let snapshot = trace.snapshot(); + assert!(snapshot.contains("request=44 phase=server-response-write-after-frame detail=4")); + assert!(snapshot.contains("request=44 phase=server-response-write-partial detail=6")); + assert!(snapshot.contains("request=44 phase=server-response-flush-bytes detail=10")); + } + + #[test] + fn http_lifecycle_server_response_records_full_accepted_write_at_terminal() { + let trace = HttpLifecycleTrace::new(); + let (mut io, _) = scripted_connection(trace.clone(), 11, false); + io.state.arm_response(54); + let waker = futures::task::noop_waker(); + let mut cx = Context::from_waker(&waker); + let response = [0; 151]; + + assert!(matches!( + Pin::new(&mut io).poll_write(&mut cx, &response), + Poll::Ready(Ok(151)) + )); + drop(io); + + let snapshot = trace.snapshot(); + assert!(snapshot.contains("request=54 phase=server-response-write-after-frame detail=151")); + assert!(snapshot.contains("request=54 phase=server-response-terminal-bytes detail=151")); + assert!(!snapshot.contains("request=54 phase=server-response-write-partial")); + } + + #[test] + fn http_lifecycle_server_response_counters_are_independent_after_flush() { + let trace = HttpLifecycleTrace::new(); + let (mut io, _) = scripted_connection(trace.clone(), 12, false); + let waker = futures::task::noop_waker(); + let mut cx = Context::from_waker(&waker); + + for (request, bytes) in [(45, b"first".as_slice()), (46, b"second".as_slice())] { + io.state.arm_response(request); + assert!(Pin::new(&mut io).poll_write(&mut cx, bytes).is_ready()); + assert!(Pin::new(&mut io).poll_flush(&mut cx).is_ready()); + } + + let snapshot = trace.snapshot(); + assert!(snapshot.contains("request=45 phase=server-response-flush-bytes detail=5")); + assert!(snapshot.contains("request=46 phase=server-response-flush-bytes detail=6")); + assert!(!snapshot.contains("server-response-correlation-boundary")); + } + + #[test] + fn http_lifecycle_server_io_records_pending_flush_and_shutdown_once() { + let trace = HttpLifecycleTrace::new(); + let (mut io, calls) = scripted_connection(trace.clone(), 14, false); + io.inner.flush_pending_once = true; + io.inner.shutdown_pending_once = true; + let waker = futures::task::noop_waker(); + let mut cx = Context::from_waker(&waker); + + assert!(Pin::new(&mut io).poll_flush(&mut cx).is_pending()); + assert!(Pin::new(&mut io).poll_flush(&mut cx).is_ready()); + assert!(Pin::new(&mut io).poll_shutdown(&mut cx).is_pending()); + assert!(Pin::new(&mut io).poll_shutdown(&mut cx).is_ready()); + assert_eq!(calls.flushes.load(Ordering::Relaxed), 2); + assert_eq!(calls.shutdowns.load(Ordering::Relaxed), 2); + + let snapshot = trace.snapshot(); + assert_eq!( + snapshot.matches("server-connection-flush-pending").count(), + 1 + ); + assert_eq!(snapshot.matches("server-connection-flush-ok").count(), 1); + assert_eq!( + snapshot + .matches("server-connection-shutdown-pending") + .count(), + 1 + ); + assert_eq!(snapshot.matches("server-connection-shutdown").count(), 2); + } + + #[test] + fn http_lifecycle_server_response_records_unknown_and_overflow_body_sizes() { + let trace = HttpLifecycleTrace::new(); + let unknown = http_body_util::StreamBody::new(futures::stream::iter([Ok::< + _, + std::convert::Infallible, + >( + Frame::data(bytes::Bytes::from_static(b"unknown")), + )])); + let mut unknown = Box::pin(TracedHttpBody::new( + unknown, + trace.clone(), + 47, + "server-response", + None, + )); + let waker = futures::task::noop_waker(); + let mut cx = Context::from_waker(&waker); + assert!(unknown.as_mut().poll_frame(&mut cx).is_ready()); + drop(unknown); + + let overflow = http_body_util::Full::new(bytes::Bytes::from(vec![0; 70_000])); + let mut overflow = Box::pin(TracedHttpBody::new( + overflow, + trace.clone(), + 48, + "server-response", + None, + )); + assert!(overflow.as_mut().poll_frame(&mut cx).is_ready()); + drop(overflow); + + let snapshot = trace.snapshot(); + assert!(snapshot.contains("request=47 phase=server-response-body-expected-unknown")); + assert!(snapshot.contains("request=47 phase=server-response-body-polled-bytes detail=7")); + assert!( + snapshot.contains("request=48 phase=server-response-body-expected-bytes detail=65535") + ); + assert!( + snapshot.contains("request=48 phase=server-response-body-polled-bytes detail=65535") + ); + assert_eq!( + snapshot + .matches("request=48 phase=server-response-byte-count-overflow") + .count(), + 2 + ); } #[test] @@ -1886,6 +2310,38 @@ mod tests { assert!(snapshot.contains("request=11 phase=server-connection-write-error")); } + #[test] + fn http_lifecycle_server_response_error_and_drop_snapshot_accepted_bytes() { + let trace = HttpLifecycleTrace::new(); + let (mut io, _) = scripted_connection(trace.clone(), 15, false); + let waker = futures::task::noop_waker(); + let mut cx = Context::from_waker(&waker); + + io.state.arm_response(49); + assert!(matches!( + Pin::new(&mut io).poll_write(&mut cx, b"abc"), + Poll::Ready(Ok(3)) + )); + io.inner.fail_write = true; + assert!(matches!( + Pin::new(&mut io).poll_write(&mut cx, b"failure"), + Poll::Ready(Err(_)) + )); + + let (mut dropped, _) = scripted_connection(trace.clone(), 16, false); + dropped.state.arm_response(50); + assert!(matches!( + Pin::new(&mut dropped).poll_write(&mut cx, b"drop"), + Poll::Ready(Ok(4)) + )); + drop(dropped); + + let snapshot = trace.snapshot(); + assert!(snapshot.contains("request=49 phase=server-response-terminal-bytes detail=3")); + assert!(snapshot.contains("request=49 phase=server-response-write-error detail=15")); + assert!(snapshot.contains("request=50 phase=server-response-terminal-bytes detail=4")); + } + #[test] fn http_lifecycle_server_connection_drop_reports_armed_response() { let trace = HttpLifecycleTrace::new(); @@ -1926,6 +2382,28 @@ mod tests { ); } + #[test] + fn http_lifecycle_server_response_boundary_without_flush_is_explicit() { + let trace = HttpLifecycleTrace::new(); + let (mut io, _) = scripted_connection(trace.clone(), 20, false); + let waker = futures::task::noop_waker(); + let mut cx = Context::from_waker(&waker); + + io.state.arm_response(63); + assert!(Pin::new(&mut io).poll_write(&mut cx, b"one").is_ready()); + io.state.arm_response(64); + assert!(Pin::new(&mut io).poll_write(&mut cx, b"two").is_ready()); + + let snapshot = trace.snapshot(); + assert!( + snapshot.contains("request=63 phase=server-response-write-boundary-bytes detail=3") + ); + assert!( + snapshot.contains("request=63 phase=server-response-correlation-boundary detail=64") + ); + assert!(snapshot.contains("request=64 phase=server-response-write-after-frame detail=3")); + } + #[test] async fn http_lifecycle_axum_pipeline_preserves_response_correlation() { use axum::body::Body; @@ -1993,8 +2471,15 @@ mod tests { assert!(snapshot.contains(&format!( "request={request_id} phase=server-response-write-after-frame" ))); + assert!(snapshot.contains(&format!( + "request={request_id} phase=server-response-body-expected-bytes detail=2" + ))); + assert!(snapshot.contains(&format!( + "request={request_id} phase=server-response-body-polled-bytes detail=2" + ))); } assert!(!snapshot.contains("server-response-correlation-overlap")); + assert!(!snapshot.contains("server-response-correlation-boundary")); } #[cfg(feature = "use-golem-wasmtime")] From 34edf2e3ac8ac0b01d4212d3984fe0c44a8fd955 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Da=CC=81vid=20Istva=CC=81n=20Bi=CC=81ro=CC=81?= Date: Sat, 29 Aug 2026 20:54:26 +0200 Subject: [PATCH 7/9] test(http): cover terminal accounting errors --- tests/common/mod.rs | 178 ++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 164 insertions(+), 14 deletions(-) diff --git a/tests/common/mod.rs b/tests/common/mod.rs index 063e1835..04254339 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -1814,12 +1814,46 @@ mod tests { shutdowns: AtomicUsize, } + struct CountedBody { + frames: VecDeque, + polls: Arc, + exact_size: Option, + end_when_empty: bool, + } + + impl HttpBody for CountedBody { + type Data = bytes::Bytes; + type Error = std::convert::Infallible; + + fn poll_frame( + mut self: Pin<&mut Self>, + _cx: &mut Context<'_>, + ) -> Poll, Self::Error>>> { + self.polls.fetch_add(1, Ordering::Relaxed); + Poll::Ready(self.frames.pop_front().map(|data| Ok(Frame::data(data)))) + } + + fn is_end_stream(&self) -> bool { + self.end_when_empty && self.frames.is_empty() + } + + fn size_hint(&self) -> SizeHint { + let mut hint = SizeHint::new(); + if let Some(exact_size) = self.exact_size { + hint.set_exact(exact_size); + } + hint + } + } + struct ScriptedIo { calls: Arc, fail_write: bool, accepted_writes: VecDeque, flush_pending_once: bool, shutdown_pending_once: bool, + fail_flush: bool, + fail_shutdown: bool, } impl tokio::io::AsyncRead for ScriptedIo { @@ -1880,6 +1914,9 @@ mod tests { cx.waker().wake_by_ref(); return Poll::Pending; } + if self.fail_flush { + return Poll::Ready(Err(std::io::Error::other("scripted flush failure"))); + } Poll::Ready(Ok(())) } @@ -1892,6 +1929,9 @@ mod tests { cx.waker().wake_by_ref(); return Poll::Pending; } + if self.fail_shutdown { + return Poll::Ready(Err(std::io::Error::other("scripted shutdown failure"))); + } Poll::Ready(Ok(())) } } @@ -1910,6 +1950,8 @@ mod tests { accepted_writes: VecDeque::new(), flush_pending_once: false, shutdown_pending_once: false, + fail_flush: false, + fail_shutdown: false, }, state, ); @@ -2115,6 +2157,8 @@ mod tests { accepted_writes: VecDeque::new(), flush_pending_once: false, shutdown_pending_once: false, + fail_flush: false, + fail_shutdown: false, }, state.clone(), ); @@ -2245,31 +2289,131 @@ mod tests { } #[test] - fn http_lifecycle_server_response_records_unknown_and_overflow_body_sizes() { + fn http_lifecycle_server_io_errors_snapshot_active_and_pending_responses() { + let trace = HttpLifecycleTrace::new(); + let waker = futures::task::noop_waker(); + let mut cx = Context::from_waker(&waker); + + let (mut flush_io, _) = scripted_connection(trace.clone(), 21, false); + flush_io.state.arm_response(55); + assert!(matches!( + Pin::new(&mut flush_io).poll_write(&mut cx, b"four"), + Poll::Ready(Ok(4)) + )); + flush_io.state.arm_response(56); + flush_io.inner.fail_flush = true; + assert!(matches!( + Pin::new(&mut flush_io).poll_flush(&mut cx), + Poll::Ready(Err(_)) + )); + drop(flush_io); + + let (mut shutdown_io, _) = scripted_connection(trace.clone(), 22, false); + shutdown_io.state.arm_response(57); + assert!(matches!( + Pin::new(&mut shutdown_io).poll_write(&mut cx, b"five!"), + Poll::Ready(Ok(5)) + )); + shutdown_io.state.arm_response(58); + shutdown_io.inner.fail_shutdown = true; + assert!(matches!( + Pin::new(&mut shutdown_io).poll_shutdown(&mut cx), + Poll::Ready(Err(_)) + )); + drop(shutdown_io); + + let snapshot = trace.snapshot(); + for (active, pending, connection, bytes, connection_phase) in [ + (55, 56, 21, 4, "server-connection-flush-error"), + (57, 58, 22, 5, "server-connection-shutdown-error"), + ] { + assert_eq!( + snapshot + .matches(&format!( + "request={active} phase=server-response-write-error detail={connection}" + )) + .count(), + 1 + ); + assert_eq!( + snapshot + .matches(&format!( + "request={active} phase=server-response-terminal-bytes detail={bytes}" + )) + .count(), + 1 + ); + assert_eq!( + snapshot + .matches(&format!( + "request={pending} phase=server-response-write-error detail={connection}" + )) + .count(), + 1 + ); + assert!(snapshot.contains(&format!("request={connection} phase={connection_phase}"))); + } + } + + #[test] + fn http_lifecycle_server_response_body_counts_terminate_without_extra_poll() { let trace = HttpLifecycleTrace::new(); - let unknown = http_body_util::StreamBody::new(futures::stream::iter([Ok::< - _, - std::convert::Infallible, - >( - Frame::data(bytes::Bytes::from_static(b"unknown")), - )])); + let exact_polls = Arc::new(AtomicUsize::new(0)); + let exact = CountedBody { + frames: VecDeque::from([ + bytes::Bytes::from_static(b"abc"), + bytes::Bytes::from_static(b"defg"), + ]), + polls: exact_polls.clone(), + exact_size: Some(7), + end_when_empty: true, + }; + let mut exact = Box::pin(TracedHttpBody::new( + exact, + trace.clone(), + 47, + "server-response", + None, + )); + let unknown_polls = Arc::new(AtomicUsize::new(0)); + let unknown = CountedBody { + frames: VecDeque::from([ + bytes::Bytes::from_static(b"un"), + bytes::Bytes::from_static(b"known"), + ]), + polls: unknown_polls.clone(), + exact_size: None, + end_when_empty: true, + }; let mut unknown = Box::pin(TracedHttpBody::new( unknown, trace.clone(), - 47, + 48, "server-response", None, )); let waker = futures::task::noop_waker(); let mut cx = Context::from_waker(&waker); - assert!(unknown.as_mut().poll_frame(&mut cx).is_ready()); + for body in [&mut exact, &mut unknown] { + assert!(matches!( + body.as_mut().poll_frame(&mut cx), + Poll::Ready(Some(Ok(_))) + )); + assert!(matches!( + body.as_mut().poll_frame(&mut cx), + Poll::Ready(Some(Ok(_))) + )); + } + assert_eq!(exact_polls.load(Ordering::Relaxed), 2); + assert_eq!(unknown_polls.load(Ordering::Relaxed), 2); + drop(exact); drop(unknown); let overflow = http_body_util::Full::new(bytes::Bytes::from(vec![0; 70_000])); let mut overflow = Box::pin(TracedHttpBody::new( overflow, trace.clone(), - 48, + 49, "server-response", None, )); @@ -2277,17 +2421,23 @@ mod tests { drop(overflow); let snapshot = trace.snapshot(); - assert!(snapshot.contains("request=47 phase=server-response-body-expected-unknown")); + assert!(snapshot.contains("request=47 phase=server-response-body-expected-bytes detail=7")); assert!(snapshot.contains("request=47 phase=server-response-body-polled-bytes detail=7")); + assert!(snapshot.contains("request=47 phase=server-response-eof")); + assert!(!snapshot.contains("request=47 phase=server-response-drop-before-terminal")); + assert!(snapshot.contains("request=48 phase=server-response-body-expected-unknown")); + assert!(snapshot.contains("request=48 phase=server-response-body-polled-bytes detail=7")); + assert!(snapshot.contains("request=48 phase=server-response-eof")); + assert!(!snapshot.contains("request=48 phase=server-response-drop-before-terminal")); assert!( - snapshot.contains("request=48 phase=server-response-body-expected-bytes detail=65535") + snapshot.contains("request=49 phase=server-response-body-expected-bytes detail=65535") ); assert!( - snapshot.contains("request=48 phase=server-response-body-polled-bytes detail=65535") + snapshot.contains("request=49 phase=server-response-body-polled-bytes detail=65535") ); assert_eq!( snapshot - .matches("request=48 phase=server-response-byte-count-overflow") + .matches("request=49 phase=server-response-byte-count-overflow") .count(), 2 ); From be7547cfcca106f61868e589661fdd1ed9b73ba7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Da=CC=81vid=20Istva=CC=81n=20Bi=CC=81ro=CC=81?= Date: Fri, 4 Sep 2026 11:31:31 +0200 Subject: [PATCH 8/9] Keep HTTP lifecycle traces off the wire (GOL-420) --- tests/common/mod.rs | 95 +++++++++++++++++++------------------ tests/common/test_server.rs | 4 +- 2 files changed, 53 insertions(+), 46 deletions(-) diff --git a/tests/common/mod.rs b/tests/common/mod.rs index 04254339..73a6b913 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -551,19 +551,14 @@ fn test_server_http_trace() -> &'static HttpLifecycleTrace { TEST_SERVER_HTTP_TRACE.get_or_init(HttpLifecycleTrace::new) } -fn attach_http_correlation(request: &mut http::Request, request_id: usize) { - request.headers_mut().insert( - http::HeaderName::from_static("x-wrq-http-trace-id"), - http::HeaderValue::try_from(request_id.to_string()).expect("numeric header is valid"), - ); +fn record_http_submit(trace: &HttpLifecycleTrace, request: &http::Request) -> usize { + let request_id = trace.next_request(); + trace.record_submit(request_id, request.method(), request.uri()); + request_id } -fn test_server_http_correlation(headers: &http::HeaderMap) -> usize { - headers - .get("x-wrq-http-trace-id") - .and_then(|value| value.to_str().ok()) - .and_then(|value| value.parse().ok()) - .unwrap_or_default() +fn next_test_server_http_request() -> usize { + test_server_http_trace().next_request() } fn record_test_server_arrival(request_id: usize, port: u16, uri: &http::Uri) { @@ -1271,14 +1266,12 @@ impl wasmtime_wasi_http::p2::WasiHttpHooks for P2HttpTraceHooks { #[cfg(not(feature = "use-golem-wasmtime"))] fn send_request( &mut self, - mut request: http::Request, + request: http::Request, config: wasmtime_wasi_http::p2::types::OutgoingRequestConfig, ) -> wasmtime_wasi_http::p2::HttpResult { let trace = self.0.clone(); - let request_id = trace.next_request(); - attach_http_correlation(&mut request, request_id); - trace.record_submit(request_id, request.method(), request.uri()); + let request_id = record_http_submit(&trace, &request); let (parts, body) = request.into_parts(); let request = http::Request::from_parts( @@ -1296,15 +1289,13 @@ impl wasmtime_wasi_http::p2::WasiHttpHooks for P2HttpTraceHooks { #[cfg(feature = "use-golem-wasmtime")] fn send_request( &mut self, - mut request: http::Request, + request: http::Request, config: wasmtime_wasi_http::p2::types::OutgoingRequestConfig, body_completion: Option, ) -> wasmtime_wasi_http::p2::HttpResult { let trace = self.0.clone(); - let request_id = trace.next_request(); - attach_http_correlation(&mut request, request_id); - trace.record_submit(request_id, request.method(), request.uri()); + let request_id = record_http_submit(&trace, &request); let body_completion = p2_body_completion_for_dispatch(request.method(), body_completion); let collect_before_send = body_completion.is_some(); let (parts, body) = request.into_parts(); @@ -1352,7 +1343,7 @@ struct P3HttpTraceHooks(HttpLifecycleTrace); impl wasmtime_wasi_http::p3::WasiHttpHooks for P3HttpTraceHooks { fn send_request( &mut self, - mut request: http::Request< + request: http::Request< http_body_util::combinators::UnsyncBoxBody< bytes::Bytes, wasmtime_wasi_http::p3::bindings::http::types::ErrorCode, @@ -1395,9 +1386,7 @@ impl wasmtime_wasi_http::p3::WasiHttpHooks for P3HttpTraceHooks { drop(response_processing); let trace = self.0.clone(); - let request_id = trace.next_request(); - attach_http_correlation(&mut request, request_id); - trace.record_submit(request_id, request.method(), request.uri()); + let request_id = record_http_submit(&trace, &request); let (parts, body) = request.into_parts(); let request = http::Request::from_parts( parts, @@ -2059,30 +2048,28 @@ mod tests { } #[test] - fn http_lifecycle_correlation_distinguishes_same_path_requests() { - let mut first = http::Request::get("http://127.0.0.1:1234/same") + fn http_lifecycle_submit_does_not_mutate_guest_headers() { + let trace = HttpLifecycleTrace::new(); + let request = http::Request::get("http://127.0.0.1:1234/same") + .header("x-guest-header", "preserved") + .header("x-wrq-http-trace-id", "guest-value") .body(()) .unwrap(); - let mut second = http::Request::get("http://127.0.0.1:1234/same") + let request_without_trace_header = http::Request::get("http://127.0.0.1:1234/same") .body(()) .unwrap(); - attach_http_correlation(&mut first, 41); - attach_http_correlation(&mut second, 42); - assert_eq!(test_server_http_correlation(first.headers()), 41); - assert_eq!(test_server_http_correlation(second.headers()), 42); + let first_request_id = record_http_submit(&trace, &request); + let second_request_id = record_http_submit(&trace, &request_without_trace_header); - let uri = Arc::new(first.uri().clone()); - let arrivals = [41, 42].map(|request_id| { - let uri = uri.clone(); - thread::spawn(move || record_test_server_arrival(request_id, 1234, &uri)) - }); - for arrival in arrivals { - arrival.join().unwrap(); - } - let snapshot = test_server_http_trace().snapshot(); - assert!(snapshot.contains("request=41 phase=server-arrival")); - assert!(snapshot.contains("request=42 phase=server-arrival")); + assert_ne!(first_request_id, second_request_id); + assert_eq!(request.headers()["x-guest-header"], "preserved"); + assert_eq!(request.headers()["x-wrq-http-trace-id"], "guest-value"); + assert!( + !request_without_trace_header + .headers() + .contains_key("x-wrq-http-trace-id") + ); } #[test] @@ -2574,7 +2561,14 @@ mod tests { any( |ConnectInfo(connection): ConnectInfo, request: Request| async move { - let request_id = test_server_http_correlation(request.headers()); + assert!(matches!( + request + .headers() + .get("x-wrq-http-trace-id") + .and_then(|value| value.to_str().ok()), + Some("guest-one" | "guest-two") + )); + let request_id = next_test_server_http_request(); record_test_server_connection(request_id, Some(&connection)); axum::response::Response::new(Body::new(TracedHttpBody::new( Full::new(Bytes::from_static(b"ok")), @@ -2598,8 +2592,8 @@ mod tests { let mut client = tokio::net::TcpStream::connect(address).await.unwrap(); client .write_all( - b"GET / HTTP/1.1\r\nHost: localhost\r\nx-wrq-http-trace-id: 71\r\n\r\n\ - GET / HTTP/1.1\r\nHost: localhost\r\nx-wrq-http-trace-id: 72\r\nConnection: close\r\n\r\n", + b"GET / HTTP/1.1\r\nHost: localhost\r\nx-wrq-http-trace-id: guest-one\r\n\r\n\ + GET / HTTP/1.1\r\nHost: localhost\r\nx-wrq-http-trace-id: guest-two\r\nConnection: close\r\n\r\n", ) .await .unwrap(); @@ -2614,7 +2608,18 @@ mod tests { 2 ); let snapshot = trace.snapshot(); - for request_id in [71, 72] { + let mut request_ids = snapshot + .lines() + .filter(|line| line.contains("phase=server-request-connection")) + .filter_map(|line| { + line.split_whitespace() + .find_map(|field| field.strip_prefix("request=")) + }) + .collect::>(); + request_ids.sort_unstable(); + request_ids.dedup(); + assert_eq!(request_ids.len(), 2, "{snapshot}"); + for request_id in request_ids { assert!(snapshot.contains(&format!( "request={request_id} phase=server-request-connection" ))); diff --git a/tests/common/test_server.rs b/tests/common/test_server.rs index fb60664e..42caa3d5 100644 --- a/tests/common/test_server.rs +++ b/tests/common/test_server.rs @@ -33,7 +33,9 @@ impl Drop for TestServerHandle { fn trace_http_lifecycle(router: Router, port: u16) -> Router { router.layer(axum::middleware::from_fn( move |request: Request, next: Next| async move { - let request_id = super::test_server_http_correlation(request.headers()); + // Server-side IDs are intentionally independent from client-side IDs. Passing a + // private correlation value over the wire would mutate guest-visible requests. + let request_id = super::next_test_server_http_request(); let connection = request .extensions() .get::>() From f02408c23612d3dd1a93fb483de2f2048ca738b8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Da=CC=81vid=20Istva=CC=81n=20Bi=CC=81ro=CC=81?= Date: Tue, 8 Sep 2026 16:33:41 +0200 Subject: [PATCH 9/9] Record terminal empty HTTP bodies (GOL-420) --- tests/common/mod.rs | 38 ++++++++++++++++++++++++++++++++++---- 1 file changed, 34 insertions(+), 4 deletions(-) diff --git a/tests/common/mod.rs b/tests/common/mod.rs index 327c26b9..9ff8a9b8 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -1064,10 +1064,11 @@ impl TracedHttpBody { side: &'static str, server_connection: Option>, ) -> Self { + let is_end_stream = inner.is_end_stream(); let expected_body_bytes = (side == "server-response") .then(|| inner.size_hint().exact()) .flatten(); - let body = Self { + let mut body = Self { inner: Box::pin(inner), trace, request, @@ -1096,6 +1097,9 @@ impl TracedHttpBody { ); } } + if is_end_stream { + body.record_terminal("eof"); + } body } @@ -2045,12 +2049,38 @@ mod tests { } #[test] - fn http_lifecycle_empty_body_drop_is_terminal() { + fn http_lifecycle_empty_bodies_record_eof_without_polling() { let trace = HttpLifecycleTrace::new(); - let body = http_body_util::Empty::::new(); - drop(TracedHttpBody::new(body, trace.clone(), 1, "request", None)); + for (request, side) in [(1, "request"), (2, "response"), (3, "server-request")] { + drop(TracedHttpBody::new( + http_body_util::Empty::::new(), + trace.clone(), + request, + side, + None, + )); + } + + let server_connection = TracedTestServerConnection { + state: Arc::new(HttpConnectionState::new(trace.clone(), 17)), + }; + record_test_server_connection(4, Some(&server_connection)); + drop(TracedHttpBody::new( + http_body_util::Empty::::new(), + trace.clone(), + 4, + "server-response", + Some(server_connection.state), + )); let snapshot = trace.snapshot(); + assert!(snapshot.contains("request=1 phase=request-eof")); + assert!(snapshot.contains("request=2 phase=response-eof")); + assert!(snapshot.contains("request=3 phase=server-request-eof")); + assert!(snapshot.contains("request=4 phase=server-response-eof")); + assert!(snapshot.contains("request=4 phase=server-request-connection detail=17")); + assert!(snapshot.contains("request=4 phase=server-response-body-expected-bytes detail=0")); + assert!(snapshot.contains("request=4 phase=server-response-body-polled-bytes detail=0")); assert!(!snapshot.contains("drop-before-terminal")); }