diff --git a/tests/common/mod.rs b/tests/common/mod.rs index c971f6d8..c629a3d2 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,601 @@ 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); +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)] +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, + ServerArrival = 17, + ServerRequestFirstData = 18, + ServerRequestEof = 19, + ServerRequestError = 20, + ServerRequestDrop = 21, + ServerResponseHead = 22, + ServerResponseFirstData = 23, + ServerResponseEof = 24, + ServerResponseError = 25, + ServerResponseDrop = 26, + TargetPath = 27, + ServerPort = 28, +} + +impl HttpLifecyclePhase { + fn label(value: u8) -> &'static str { + match value { + 1 => "submit", + 2 => "target-port", + 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", + 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", + 28 => "server-port", + _ => "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); + self.publish(sequence, request, phase, detail); + } + + fn publish(&self, sequence: usize, request: usize, phase: HttpLifecyclePhase, detail: u16) { + 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); + let slot = &self.slots[sequence % HTTP_LIFECYCLE_CAP]; + let mut current = slot.load(Ordering::Acquire); + loop { + let current_sequence = current >> 40; + 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) + { + Ok(_) => return, + Err(observed) => current = observed, + } + } + } + + 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.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 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; + 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, + journal: Arc, +} + +impl HttpLifecycleTrace { + fn new() -> Self { + Self { + invocation: NEXT_HTTP_TRACE_INVOCATION.fetch_add(1, Ordering::Relaxed), + journal: Arc::new(HttpLifecycleJournal::new()), + } + } + + fn next_request(&self) -> usize { + NEXT_HTTP_TRACE_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); + 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, + ); + } + + fn snapshot(&self) -> String { + self.journal.snapshot(self.invocation) + } +} + +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 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( + 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(request_id: usize, status: http::StatusCode) { + test_server_http_trace().record( + request_id, + HttpLifecyclePhase::ServerResponseHead, + status.as_u16(), + ); +} + +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) +} + +/// 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, + ("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); + } +} + +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.inner.as_ref().is_end_stream() { + self.event("drop-before-terminal"); + } + } +} + +#[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 +} + +#[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, + 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(); + let request = http::Request::from_parts( + parts, + TracedHttpBody::new(body, trace.clone(), request_id, "request").boxed_unsync(), + ); + 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")] + fn send_request( + &mut self, + 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 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, + TracedHttpBody::new(body, trace.clone(), request_id, "request").boxed_unsync(), + ); + 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)) + } +} + +#[derive(Clone)] +struct P3HttpTraceHooks(HttpLifecycleTrace); + +impl wasmtime_wasi_http::p3::WasiHttpHooks for P3HttpTraceHooks { + fn send_request( + &mut self, + mut 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(); + 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( + 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; @@ -605,6 +1205,145 @@ 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_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()); + 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 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() { @@ -1836,10 +2575,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 +2676,10 @@ 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(); + let server_http_lifecycle = test_server_http_trace().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}\ntest-server HTTP lifecycle:\n{server_http_lifecycle}\nhost trace:\n{host_trace}" )) }); @@ -2527,6 +3271,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 +3342,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 +3546,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() diff --git a/tests/common/test_server.rs b/tests/common/test_server.rs index ade981e7..1df69bc1 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,35 @@ 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()); + 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, + request_id, + "server-request", + )), + ); + let response = next.run(request).await; + 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, + request_id, + "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 +258,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 +295,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(); });