diff --git a/crates/edgezero-adapter-fastly/src/delivery_compression.rs b/crates/edgezero-adapter-fastly/src/delivery_compression.rs new file mode 100644 index 000000000..d148d2f3c --- /dev/null +++ b/crates/edgezero-adapter-fastly/src/delivery_compression.rs @@ -0,0 +1,286 @@ +//! Final delivery-compression headers shared by standard and custom senders. + +use edgezero_core::delivery_compression::DeliveryCompressionIntent; +use edgezero_core::http::{HeaderValue, Response, header}; + +/// Restore the delivery contract of an explicitly marked identity response. +/// +/// Standard conversion calls this automatically. Custom senders must call it +/// after their last header effects and before consuming response parts or +/// committing headers. It never polls or modifies the body. Unmarked responses +/// are unchanged, including already encoded output and manually supplied hints. +/// +/// The intent asserts that the installed body produces identity bytes. A body +/// transformer must remove it before replacing or re-encoding that representation. +/// This function only corrects stale headers; it cannot decode an encoded body. +#[inline] +pub fn finalize_response(response: &mut Response) { + if response + .extensions() + .get::() + .is_none() + { + return; + } + response.headers_mut().remove(header::CONTENT_ENCODING); + response.headers_mut().remove(header::CONTENT_LENGTH); + response + .headers_mut() + .insert("x-compress-hint", HeaderValue::from_static("on")); + let varies_on_encoding = response + .headers() + .get_all(header::VARY) + .iter() + .any(|value| { + value.to_str().is_ok_and(|text| { + text.split(',').any(|token| { + token.trim() == "*" || token.trim().eq_ignore_ascii_case("accept-encoding") + }) + }) + }); + if !varies_on_encoding { + response + .headers_mut() + .append(header::VARY, HeaderValue::from_static("Accept-Encoding")); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use bytes::Bytes; + use edgezero_core::body::Body; + use edgezero_core::delivery_compression::{ + DeliveryCompressionIntent, DeliveryCompressionSupport, + }; + use edgezero_core::http::{HeaderValue, StatusCode, header}; + use futures::executor::block_on; + use futures_util::stream; + use std::cell::Cell; + use std::rc::Rc; + use std::task::Poll; + + fn response(streaming: bool) -> Response { + let body = if streaming { + Body::stream(stream::iter(vec![Bytes::from_static(b"identity output")])) + } else { + Body::text("identity output") + }; + let mut response = Response::new(body); + *response.status_mut() = StatusCode::CREATED; + for (name, value) in [ + ("content-length", "15"), + ("cache-control", "private, no-store"), + ("surrogate-control", "no-store"), + ("etag", "\"original\""), + ("set-cookie", "a=1"), + ("set-cookie", "b=2"), + ("vary", "Origin"), + ("x-other", "preserve"), + ] { + response + .headers_mut() + .append(name, HeaderValue::from_static(value)); + } + response.extensions_mut().insert(42_u32); + response + } + + #[test] + fn unmarked_identity_encoded_and_manual_hints_remain_unchanged() { + for streaming in [false, true] { + for maybe_encoding in [ + None, + Some("gzip"), + Some("br"), + Some("deflate"), + Some("zstd"), + ] { + let mut response = response(streaming); + if let Some(encoding) = maybe_encoding { + response + .headers_mut() + .insert(header::CONTENT_ENCODING, HeaderValue::from_static(encoding)); + } + response + .headers_mut() + .insert("x-compress-hint", HeaderValue::from_static("off")); + let headers = response.headers().clone(); + finalize_response(&mut response); + assert_eq!(response.headers(), &headers); + assert_eq!(response.status(), StatusCode::CREATED); + assert_eq!(response.extensions().get::(), Some(&42)); + assert_eq!( + block_on(response.into_body().into_bytes_bounded(1024)).unwrap(), + "identity output" + ); + } + } + } + + #[test] + fn marked_once_and_stream_restore_late_headers_without_changing_output_or_privacy() { + for streaming in [false, true] { + let mut response = response(streaming); + DeliveryCompressionSupport::GzipBrotli + .mark_identity_response(&mut response) + .unwrap(); + response + .headers_mut() + .insert(header::CONTENT_ENCODING, HeaderValue::from_static("gzip")); + response + .headers_mut() + .insert(header::CONTENT_LENGTH, HeaderValue::from_static("9999")); + response + .headers_mut() + .insert("x-compress-hint", HeaderValue::from_static("off")); + finalize_response(&mut response); + let headers = response.headers().clone(); + finalize_response(&mut response); + assert_eq!(response.headers(), &headers); + assert!(!response.headers().contains_key(header::CONTENT_ENCODING)); + assert!(!response.headers().contains_key(header::CONTENT_LENGTH)); + assert_eq!(response.headers()["x-compress-hint"], "on"); + assert_eq!( + response.headers().get_all("x-compress-hint").iter().count(), + 1 + ); + assert_eq!(response.status(), StatusCode::CREATED); + assert_eq!( + response.headers()[header::CACHE_CONTROL], + "private, no-store" + ); + assert_eq!(response.headers()["surrogate-control"], "no-store"); + assert_eq!(response.headers()[header::ETAG], "\"original\""); + assert_eq!(response.headers()["x-other"], "preserve"); + assert_eq!( + response + .headers() + .get_all(header::SET_COOKIE) + .iter() + .collect::>(), + vec!["a=1", "b=2"] + ); + assert_eq!(response.extensions().get::(), Some(&42)); + assert!( + response + .extensions() + .get::() + .is_some() + ); + assert_eq!( + block_on(response.into_body().into_bytes_bounded(1024)).unwrap(), + "identity output" + ); + } + } + + #[test] + fn vary_preserves_fields_and_avoids_duplicate_or_wildcard_variation() { + for (values, expected) in [ + (vec![], vec!["Accept-Encoding"]), + (vec!["Origin"], vec!["Origin", "Accept-Encoding"]), + ( + vec!["Origin", "Accept-Language"], + vec!["Origin", "Accept-Language", "Accept-Encoding"], + ), + ( + vec!["Origin, AcCePt-EnCoDiNg"], + vec!["Origin, AcCePt-EnCoDiNg"], + ), + ( + vec!["Origin", " accept-encoding \t"], + vec!["Origin", " accept-encoding \t"], + ), + (vec!["*"], vec!["*"]), + ( + vec!["Origin", "Accept-Language, *"], + vec!["Origin", "Accept-Language, *"], + ), + ] { + let mut response = Response::new(Body::empty()); + for value in &values { + response + .headers_mut() + .append(header::VARY, HeaderValue::from_static(value)); + } + DeliveryCompressionSupport::GzipBrotli + .mark_identity_response(&mut response) + .unwrap(); + finalize_response(&mut response); + let actual = response + .headers() + .get_all(header::VARY) + .iter() + .map(|value| value.to_str().unwrap()) + .collect::>(); + assert_eq!(actual, expected); + } + } + + #[test] + fn opaque_vary_value_is_preserved_alongside_required_variation() { + let mut response = Response::new(Body::empty()); + let opaque = HeaderValue::from_bytes(&[0xff]).unwrap(); + response.headers_mut().append(header::VARY, opaque.clone()); + DeliveryCompressionSupport::GzipBrotli + .mark_identity_response(&mut response) + .unwrap(); + finalize_response(&mut response); + assert_eq!( + response + .headers() + .get_all(header::VARY) + .iter() + .collect::>(), + vec![&opaque, &HeaderValue::from_static("Accept-Encoding")] + ); + } + + #[test] + fn finalization_does_not_poll_or_replace_a_lazy_stream() { + let polls = Rc::new(Cell::new(0_usize)); + let observed = Rc::clone(&polls); + let mut chunk = Some(Bytes::from_static(b"prefix")); + let body = Body::stream(stream::poll_fn(move |_| { + observed.set(observed.get() + 1); + Poll::Ready(chunk.take()) + })); + let mut response = Response::new(body); + DeliveryCompressionSupport::GzipBrotli + .mark_identity_response(&mut response) + .unwrap(); + finalize_response(&mut response); + finalize_response(&mut response); + assert_eq!(polls.get(), 0); + assert!(response.body().is_stream()); + assert_eq!( + block_on(response.into_body().into_bytes_bounded(1024)).unwrap(), + "prefix" + ); + assert_eq!(polls.get(), 2); + } + + #[test] + fn intent_removed_before_reencoding_preserves_encoded_representation() { + let mut response = Response::new(Body::text("identity output")); + DeliveryCompressionSupport::GzipBrotli + .mark_identity_response(&mut response) + .unwrap(); + response + .extensions_mut() + .remove::(); + let encoded = vec![0x1f, 0x8b, 0x08, 0x00]; + *response.body_mut() = Body::from(encoded.clone()); + response + .headers_mut() + .insert(header::CONTENT_ENCODING, HeaderValue::from_static("gzip")); + response + .headers_mut() + .insert(header::CONTENT_LENGTH, HeaderValue::from_static("4")); + let headers = response.headers().clone(); + finalize_response(&mut response); + assert_eq!(response.headers(), &headers); + assert_eq!(response.body().as_bytes(), Some(encoded.as_slice())); + } +} diff --git a/crates/edgezero-adapter-fastly/src/lib.rs b/crates/edgezero-adapter-fastly/src/lib.rs index ba3751931..202053ff1 100644 --- a/crates/edgezero-adapter-fastly/src/lib.rs +++ b/crates/edgezero-adapter-fastly/src/lib.rs @@ -18,6 +18,7 @@ pub mod cli; #[cfg(feature = "fastly")] pub mod config_store; pub mod context; +pub mod delivery_compression; #[cfg(feature = "fastly")] pub mod key_value_store; pub mod lifecycle; diff --git a/crates/edgezero-adapter-fastly/src/request.rs b/crates/edgezero-adapter-fastly/src/request.rs index e1f5962a3..19167db8c 100644 --- a/crates/edgezero-adapter-fastly/src/request.rs +++ b/crates/edgezero-adapter-fastly/src/request.rs @@ -7,6 +7,7 @@ use std::sync::{Arc, Mutex, OnceLock, PoisonError}; use edgezero_core::app::{App, StoreMetadata, StoresMetadata}; use edgezero_core::body::Body; use edgezero_core::config_store::ConfigStoreHandle; +use edgezero_core::delivery_compression::DeliveryCompressionSupport; use edgezero_core::error::EdgeError; use edgezero_core::http::header::HOST; use edgezero_core::http::{Extensions, HeaderValue, Request, Uri, request_builder}; @@ -676,6 +677,9 @@ fn convert_request(mut req: FastlyRequest) -> Result { client_ip: req.get_client_ip_addr(), }; FastlyRequestContext::insert(&mut request, context); + request + .extensions_mut() + .insert(DeliveryCompressionSupport::GzipBrotli); request .extensions_mut() .insert(ProxyHandle::with_client(FastlyProxyClient)); diff --git a/crates/edgezero-adapter-fastly/src/response.rs b/crates/edgezero-adapter-fastly/src/response.rs index f9eb4d020..b1d87bc7b 100644 --- a/crates/edgezero-adapter-fastly/src/response.rs +++ b/crates/edgezero-adapter-fastly/src/response.rs @@ -1,3 +1,4 @@ +use crate::delivery_compression::finalize_response; use edgezero_core::body::Body; use edgezero_core::error::EdgeError; use edgezero_core::http::{Response, Uri}; @@ -9,7 +10,8 @@ use std::io::Write as _; /// # Errors /// Returns [`EdgeError::Internal`] if the response body cannot be streamed to the Fastly send-channel. #[inline] -pub fn from_core_response(response: Response) -> Result { +pub fn from_core_response(mut response: Response) -> Result { + finalize_response(&mut response); let (parts, body) = response.into_parts(); let mut fastly_response = FastlyResponse::from_status(parts.status.as_u16()); diff --git a/crates/edgezero-adapter-fastly/tests/contract.rs b/crates/edgezero-adapter-fastly/tests/contract.rs index 09c5dbef1..4308b0a64 100644 --- a/crates/edgezero-adapter-fastly/tests/contract.rs +++ b/crates/edgezero-adapter-fastly/tests/contract.rs @@ -28,6 +28,9 @@ mod tests { use edgezero_core::body::Body; use edgezero_core::config_store::{ConfigStore, ConfigStoreError, ConfigStoreHandle}; use edgezero_core::context::RequestContext; + use edgezero_core::delivery_compression::{ + DeliveryCompressionIntent, DeliveryCompressionSupport, + }; use edgezero_core::error::EdgeError; use edgezero_core::http::{Method, Response, StatusCode, response_builder}; use edgezero_core::router::RouterService; @@ -127,6 +130,10 @@ mod tests { let core_request = into_core_request(req).expect("core request"); + assert_eq!( + DeliveryCompressionSupport::from_request(&core_request), + DeliveryCompressionSupport::GzipBrotli + ); assert_eq!(core_request.method(), &Method::POST); assert_eq!(core_request.uri().path(), "/mirror"); assert_eq!(core_request.uri().query(), Some("foo=bar")); @@ -167,6 +174,10 @@ mod tests { }, ) .expect("registry-aware conversion"); + assert_eq!( + DeliveryCompressionSupport::from_request(&core), + DeliveryCompressionSupport::GzipBrotli + ); assert!(core.extensions().get::().is_some()); assert_eq!( core.extensions().get::().unwrap(), @@ -195,6 +206,183 @@ mod tests { assert_eq!(fastly_response.take_body_bytes(), b"hello world"); } + #[test] + fn delivery_compression_standard_converter_finalizes_once_and_stream() { + for streaming in [false, true] { + let body = if streaming { + Body::stream(stream::iter(vec![ + Bytes::from_static(b"identity "), + Bytes::from_static(b"output"), + ])) + } else { + Body::text("identity output") + }; + let mut response = response_builder() + .status(StatusCode::CREATED) + .header("set-cookie", "a=1") + .header("set-cookie", "b=2") + .header("cache-control", "private, no-store") + .header("etag", "\"keep\"") + .header("vary", "Origin") + .body(body) + .unwrap(); + DeliveryCompressionSupport::GzipBrotli + .mark_identity_response(&mut response) + .unwrap(); + // Late header effects do not change the identity-producing body. + response + .headers_mut() + .insert("content-encoding", "gzip".parse().unwrap()); + response + .headers_mut() + .insert("content-length", "9999".parse().unwrap()); + response + .headers_mut() + .insert("x-compress-hint", "off".parse().unwrap()); + let mut native = from_core_response(response).unwrap(); + assert_eq!(native.get_status(), FastlyStatus::CREATED); + assert_eq!(native.get_header_str("x-compress-hint"), Some("on")); + assert!(native.get_header("content-encoding").is_none()); + assert!(native.get_header("content-length").is_none()); + assert_eq!( + native.get_header_str("cache-control"), + Some("private, no-store") + ); + assert_eq!(native.get_header_str("etag"), Some("\"keep\"")); + assert_eq!( + native + .get_header_all("set-cookie") + .map(|value| value.to_str().unwrap()) + .collect::>(), + vec!["a=1", "b=2"] + ); + assert_eq!( + native + .get_header_all("vary") + .map(|value| value.to_str().unwrap()) + .collect::>(), + vec!["Origin", "Accept-Encoding"] + ); + assert_eq!(native.take_body_bytes(), b"identity output"); + } + } + + #[test] + fn delivery_compression_unmarked_gzip_survives_standard_conversion() { + use flate2::{Compression, write::GzEncoder}; + use std::io::Write as _; + let mut compressor = GzEncoder::new(Vec::new(), Compression::default()); + compressor.write_all(b"encoded output").unwrap(); + let encoded = compressor.finish().unwrap(); + for streaming in [false, true] { + let body = if streaming { + Body::stream(stream::iter(vec![Bytes::from(encoded.clone())])) + } else { + Body::from(encoded.clone()) + }; + let response = response_builder() + .header("content-encoding", "gzip") + .header("content-length", encoded.len()) + .header("vary", "Origin") + .body(body) + .unwrap(); + let mut native = from_core_response(response).unwrap(); + assert_eq!(native.get_header_str("content-encoding"), Some("gzip")); + assert_eq!( + native.get_header_str("content-length"), + Some(encoded.len().to_string().as_str()) + ); + assert_eq!(native.get_header_str("vary"), Some("Origin")); + assert!(native.get_header("x-compress-hint").is_none()); + assert_eq!(native.take_body_bytes(), encoded); + } + } + + #[test] + fn delivery_compression_stream_errors_keep_existing_propagation() { + use std::io::Error as IoError; + let body = Body::from_stream(stream::iter(vec![Err::(IoError::other( + "delivery source failed", + ))])); + let mut response = Response::new(body); + DeliveryCompressionSupport::GzipBrotli + .mark_identity_response(&mut response) + .unwrap(); + let error = from_core_response(response).expect_err("source failure must propagate"); + assert!(error.to_string().contains("delivery source failed")); + } + + #[test] + fn delivery_compression_dispatch_exposes_support_before_output_selection() { + use edgezero_adapter_fastly::request::dispatch_with_registries; + use edgezero_core::app::StoresMetadata; + + #[edgezero_core::action] + async fn deliver(ctx: RequestContext) -> Result { + let support = DeliveryCompressionSupport::from_request(ctx.request()); + assert_eq!(support, DeliveryCompressionSupport::GzipBrotli); + let mut response = Response::new(Body::text("identity output")); + support.mark_identity_response(&mut response)?; + Ok(response) + } + let app = App::new(RouterService::builder().get("/delivery", deliver).build()); + let mut service_response = FastlyService::new(&app) + .dispatch(fastly_request(FastlyMethod::GET, "/delivery", None)) + .unwrap(); + assert_eq!( + service_response.get_header_str("x-compress-hint"), + Some("on") + ); + assert_eq!(service_response.take_body_bytes(), b"identity output"); + let mut registry_response = dispatch_with_registries( + &app, + fastly_request(FastlyMethod::GET, "/delivery", None), + StoresMetadata::default(), + |_, _| {}, + ) + .unwrap(); + assert_eq!( + registry_response.get_header_str("x-compress-hint"), + Some("on") + ); + assert_eq!(registry_response.take_body_bytes(), b"identity output"); + } + + #[test] + fn delivery_compression_custom_finalizer_does_not_poll_before_parts_split() { + use edgezero_adapter_fastly::delivery_compression::finalize_response; + use std::cell::Cell; + use std::rc::Rc; + use std::task::Poll; + let request = + into_core_request(fastly_request(FastlyMethod::GET, "/delivery", None)).unwrap(); + let support = DeliveryCompressionSupport::from_request(&request); + assert_eq!(support, DeliveryCompressionSupport::GzipBrotli); + let polls = Rc::new(Cell::new(0_usize)); + let observed = Rc::clone(&polls); + let body = Body::stream(stream::poll_fn(move |_| { + observed.set(observed.get() + 1); + Poll::>::Pending + })); + let mut response = Response::new(body); + support.mark_identity_response(&mut response).unwrap(); + response + .headers_mut() + .insert("content-length", "9999".parse().unwrap()); + finalize_response(&mut response); + assert_eq!(polls.get(), 0); + let (parts, output_body) = response.into_parts(); + assert!( + parts + .extensions + .get::() + .is_some() + ); + assert_eq!(parts.headers["x-compress-hint"], "on"); + assert!(!parts.headers.contains_key("content-length")); + assert!(output_body.is_stream()); + } + #[test] fn dispatch_runs_router_and_returns_response() { let app = build_test_app(); diff --git a/crates/edgezero-core/src/delivery_compression.rs b/crates/edgezero-core/src/delivery_compression.rs new file mode 100644 index 000000000..c1324c244 --- /dev/null +++ b/crates/edgezero-core/src/delivery_compression.rs @@ -0,0 +1,318 @@ +//! Adapter delivery compression capabilities and explicit identity-output intent. + +use crate::error::EdgeError; +use crate::http::{Request, Response, header::CONTENT_ENCODING}; + +/// Compression an adapter can apply when delivering identity output to a reader. +/// +/// This does not select a codec or negotiate the reader's `Accept-Encoding`. +/// Applications must check support before choosing to skip their own encoder. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub enum DeliveryCompressionSupport { + /// The adapter supports reader-negotiated gzip and Brotli delivery. + GzipBrotli, + /// The application retains responsibility for output encoding. + #[default] + Unavailable, +} + +/// An application assertion that the response already produces identity bytes. +/// +/// Attach through [`DeliveryCompressionSupport::mark_identity_response`] only +/// after installing completed identity bytes or the final identity-producing +/// stream. This marker does not prove the body's encoding. A component that +/// later replaces or re-encodes the body representation must remove this +/// extension first. Header-only mutations retain the intent for the adapter's +/// finalization step. +#[derive(Clone, Copy, Debug)] +pub struct DeliveryCompressionIntent { + _private: (), +} + +impl DeliveryCompressionSupport { + /// Read adapter support from request extensions, defaulting to unavailable. + #[must_use] + #[inline] + pub fn from_request(request: &Request) -> Self { + request + .extensions() + .get::() + .copied() + .unwrap_or_default() + } + + /// Select adapter compression for an already identity-producing response. + /// + /// Does not decode, inspect, poll, or modify the body or headers. The caller + /// owns eligibility, negotiation, and installing the identity output first. + /// Query or remove the intent through ordinary response extensions. + /// + /// # Errors + /// Returns [`EdgeError::Internal`] without modifying the response when + /// support is unavailable or `Content-Encoding` is neither absent nor a + /// single case-insensitive `identity` field with optional whitespace. + /// + /// # Examples + /// ``` + /// use edgezero_core::body::Body; + /// use edgezero_core::delivery_compression::{ + /// DeliveryCompressionIntent, DeliveryCompressionSupport, + /// }; + /// use edgezero_core::http::{Request, Response}; + /// # let mut request = Request::new(Body::empty()); + /// # request.extensions_mut().insert(DeliveryCompressionSupport::GzipBrotli); + /// + /// // Check support before choosing to skip the output encoder. + /// let support = DeliveryCompressionSupport::from_request(&request); + /// if support == DeliveryCompressionSupport::GzipBrotli { + /// let mut response = Response::new(Body::text("identity output")); + /// support.mark_identity_response(&mut response)?; + /// assert!(response.extensions().get::().is_some()); + /// response.extensions_mut().remove::(); + /// } + /// # Ok::<(), edgezero_core::error::EdgeError>(()) + /// ``` + #[inline] + pub fn mark_identity_response(self, response: &mut Response) -> Result<(), EdgeError> { + if self == Self::Unavailable { + return Err(EdgeError::internal(anyhow::anyhow!( + "adapter delivery compression is unavailable" + ))); + } + let encoding_fields = response.headers().get_all(CONTENT_ENCODING); + let mut encodings = encoding_fields.iter(); + if let Some(encoding) = encodings.next() + && (!encoding + .to_str() + .is_ok_and(|value| value.trim().eq_ignore_ascii_case("identity")) + || encodings.next().is_some()) + { + return Err(EdgeError::internal(anyhow::anyhow!( + "adapter delivery compression requires identity output" + ))); + } + response + .extensions_mut() + .insert(DeliveryCompressionIntent { _private: () }); + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::body::Body; + use crate::http::{HeaderValue, header::CONTENT_ENCODING}; + use bytes::Bytes; + use futures_util::stream; + use std::cell::Cell; + use std::rc::Rc; + use std::task::Poll; + + #[test] + fn support_defaults_to_unavailable_and_reads_only_typed_extensions() { + let mut request = Request::new(Body::empty()); + request + .headers_mut() + .insert("x-compress-hint", HeaderValue::from_static("on")); + request + .headers_mut() + .insert("surrogate-control", HeaderValue::from_static("max-age=60")); + assert_eq!( + DeliveryCompressionSupport::default(), + DeliveryCompressionSupport::Unavailable + ); + assert_eq!( + DeliveryCompressionSupport::from_request(&request), + DeliveryCompressionSupport::Unavailable + ); + request + .extensions_mut() + .insert(DeliveryCompressionSupport::GzipBrotli); + assert_eq!( + DeliveryCompressionSupport::from_request(&request), + DeliveryCompressionSupport::GzipBrotli + ); + request + .extensions_mut() + .insert(DeliveryCompressionSupport::Unavailable); + assert_eq!( + DeliveryCompressionSupport::from_request(&request), + DeliveryCompressionSupport::Unavailable + ); + } + + #[test] + fn marking_identity_output_preserves_headers_body_and_other_extensions() { + for maybe_encoding in [None, Some("identity"), Some(" \tIdEnTiTy \t")] { + let mut response = Response::new(Body::text("identity output")); + response + .headers_mut() + .insert("content-length", HeaderValue::from_static("15")); + if let Some(encoding) = maybe_encoding { + response + .headers_mut() + .insert(CONTENT_ENCODING, HeaderValue::from_str(encoding).unwrap()); + } + response.extensions_mut().insert(42_u32); + let headers = response.headers().clone(); + DeliveryCompressionSupport::GzipBrotli + .mark_identity_response(&mut response) + .unwrap(); + assert!( + response + .extensions() + .get::() + .is_some() + ); + assert_eq!(response.extensions().get::(), Some(&42)); + assert_eq!(response.headers(), &headers); + assert_eq!( + response.body().as_bytes(), + Some(b"identity output".as_slice()) + ); + } + } + + #[test] + fn unavailable_marking_does_not_mutate_response() { + let mut response = Response::new(Body::text("unchanged")); + response + .headers_mut() + .insert("x-test", HeaderValue::from_static("original")); + response.extensions_mut().insert(42_u32); + let headers = response.headers().clone(); + assert!( + DeliveryCompressionSupport::Unavailable + .mark_identity_response(&mut response) + .is_err() + ); + assert_eq!(response.headers(), &headers); + assert_eq!(response.body().as_bytes(), Some(b"unchanged".as_slice())); + assert_eq!(response.extensions().get::(), Some(&42)); + assert!( + response + .extensions() + .get::() + .is_none() + ); + } + + #[test] + fn coded_stacked_repeated_and_malformed_encodings_are_rejected_unchanged() { + let opaque = HeaderValue::from_bytes(&[0xff]).unwrap(); + for encodings in [ + vec![HeaderValue::from_static("gzip")], + vec![HeaderValue::from_static("br")], + vec![HeaderValue::from_static("deflate")], + vec![HeaderValue::from_static("zstd")], + vec![HeaderValue::from_static("identity, gzip")], + vec![HeaderValue::from_static("identity, identity")], + vec![HeaderValue::from_static("")], + vec![HeaderValue::from_static(" ")], + vec![HeaderValue::from_static("identity;q=1")], + vec![ + HeaderValue::from_static("identity"), + HeaderValue::from_static("gzip"), + ], + vec![ + HeaderValue::from_static("identity"), + HeaderValue::from_static("identity"), + ], + vec![opaque], + ] { + let mut response = Response::new(Body::from(vec![0x1f, 0x8b, 0x08])); + for encoding in encodings { + response.headers_mut().append(CONTENT_ENCODING, encoding); + } + response.extensions_mut().insert(42_u32); + let headers = response.headers().clone(); + assert!( + DeliveryCompressionSupport::GzipBrotli + .mark_identity_response(&mut response) + .is_err() + ); + assert_eq!(response.headers(), &headers); + assert_eq!( + response.body().as_bytes(), + Some([0x1f, 0x8b, 0x08].as_slice()) + ); + assert_eq!(response.extensions().get::(), Some(&42)); + assert!( + response + .extensions() + .get::() + .is_none() + ); + } + } + + #[test] + fn failed_remarking_preserves_existing_intent() { + let mut response = Response::new(Body::text("identity output")); + DeliveryCompressionSupport::GzipBrotli + .mark_identity_response(&mut response) + .unwrap(); + response + .headers_mut() + .insert(CONTENT_ENCODING, HeaderValue::from_static("gzip")); + assert!( + DeliveryCompressionSupport::GzipBrotli + .mark_identity_response(&mut response) + .is_err() + ); + assert!( + response + .extensions() + .get::() + .is_some() + ); + assert_eq!(response.headers()[CONTENT_ENCODING], "gzip"); + } + + #[test] + fn marking_does_not_poll_streams_on_success_or_failure() { + let polls = Rc::new(Cell::new(0_usize)); + let observed = Rc::clone(&polls); + let body = Body::stream(stream::poll_fn(move |_| { + observed.set(observed.get() + 1); + Poll::>::Pending + })); + let mut response = Response::new(body); + DeliveryCompressionSupport::GzipBrotli + .mark_identity_response(&mut response) + .unwrap(); + response + .headers_mut() + .insert(CONTENT_ENCODING, HeaderValue::from_static("gzip")); + assert!( + DeliveryCompressionSupport::GzipBrotli + .mark_identity_response(&mut response) + .is_err() + ); + assert_eq!(polls.get(), 0); + assert!(response.body().is_stream()); + } + + #[test] + fn intent_survives_parts_and_can_be_removed_before_reencoding() { + let mut response = Response::new(Body::text("identity output")); + DeliveryCompressionSupport::GzipBrotli + .mark_identity_response(&mut response) + .unwrap(); + let (parts, body) = response.into_parts(); + let mut rebuilt = Response::from_parts(parts, body); + assert!( + rebuilt + .extensions_mut() + .remove::() + .is_some() + ); + assert!( + rebuilt + .extensions() + .get::() + .is_none() + ); + } +} diff --git a/crates/edgezero-core/src/lib.rs b/crates/edgezero-core/src/lib.rs index c0f1e89e8..c0a641fa5 100644 --- a/crates/edgezero-core/src/lib.rs +++ b/crates/edgezero-core/src/lib.rs @@ -22,6 +22,7 @@ pub mod canonical_form; pub mod compression; pub mod config_store; pub mod context; +pub mod delivery_compression; pub mod env_config; pub mod error; pub mod extractor; diff --git a/docs/guide/adapters/fastly.md b/docs/guide/adapters/fastly.md index 582575bb0..80155ba3b 100644 --- a/docs/guide/adapters/fastly.md +++ b/docs/guide/adapters/fastly.md @@ -175,6 +175,60 @@ edgezero_core::app!("edgezero.toml", owns_logging = true); or on a hand-written `Hooks` impl (`fn owns_logging() -> bool { true }`). Every adapter's `run_app` honors it, so the app is responsible for logger setup. +## Delivery compression + +The Fastly request converter advertises +`edgezero_core::delivery_compression::DeliveryCompressionSupport::GzipBrotli` +through a typed request extension. Read it with +`DeliveryCompressionSupport::from_request(&request)` before choosing an output +encoder. Missing support defaults to `Unavailable`; the other adapters do not +advertise this facility. + +An application that has selected compatible adapter delivery can install identity +bytes or its final identity-producing stream, then mark that response: + +```rust +use edgezero_core::delivery_compression::DeliveryCompressionSupport; + +let support = DeliveryCompressionSupport::from_request(&request); +// Consult support and the reader's preferences before choosing your encoder. +// Only after installing the selected identity output: +support.mark_identity_response(&mut response)?; +``` + +Marking does not decode or inspect the body. It rejects unavailable support and +non-identity, stacked, repeated, or malformed `Content-Encoding` fields without +changing the response. A later component that re-encodes the body must first +remove `DeliveryCompressionIntent` from the response extensions. + +Standard Fastly response conversion honors the intent automatically. It sets +`X-Compress-Hint: on`, removes stale encoding/length headers, and preserves existing +Vary fields while covering `Accept-Encoding`. Fastly negotiates gzip or Brotli +with the original reader request. Applications retain their eligibility, +negotiation/fallback, privacy, and cache policy. Deflate is not supported by this +delivery facility. Unmarked responses keep their existing behavior. + +Custom senders must use the same finalizer after their last response-header +mutations and before splitting the response or committing headers: + +```rust +edgezero_adapter_fastly::delivery_compression::finalize_response(&mut response); +let (parts, body) = response.into_parts(); +// Continue with the application's existing buffered or streaming sender. +``` + +This finalizer does not poll or collect the body and can be called repeatedly. +The standard `from_core_response` converter still collects streamed core bodies +before returning a native response; this feature does not change that transport +behavior. Custom progressive senders should call the finalizer directly rather +than switch to that collecting converter. + +Viceroy tests establish guest-body and header behavior, not actual CDN compression +or performance. Verify delivery encoding, reader negotiation, and transfer/vCPU +measurements on real Fastly before making those claims. With shielding or service +chaining, the matching client may itself be Fastly; final-hop policy remains with +the application. + ## Building Build for Fastly's Wasm target: diff --git a/docs/superpowers/plans/2026-11-02-362-delivery-compression.md b/docs/superpowers/plans/2026-11-02-362-delivery-compression.md new file mode 100644 index 000000000..8776e0037 --- /dev/null +++ b/docs/superpowers/plans/2026-11-02-362-delivery-compression.md @@ -0,0 +1,123 @@ +# Adapter delivery compression implementation plan + +## Outcome and boundaries + +Implement the [design](../specs/2026-11-02-362-delivery-compression-design.md) +for [EdgeZero #362](https://github.com/IABTechLab/edgezero/issues/362), then +integrate its exact PR commit into +[Trusted Server #884](https://github.com/IABTechLab/trusted-server/pull/884). +Both pull requests remain draft for joint review. Do not deploy either application. + +Preserve all unmarked response behavior, standard converter buffering, and custom +sender laziness. Only the application decides when its output is identity and +eligible for platform delivery compression. + +## 1. Establish the contract + +- [x] Create isolated `feature/362-delivery-compression` worktree from current main. +- [x] Read existing superpowers specs, adapter response conversion, and the actual + Trusted Server consumer. +- [x] Write this design and plan before source changes. +- [x] Review the design independently against source. Both agent reviews found + no blockers; the design records the selected minimal API and consumer invariant. + +Compare keeping application-local glue, standard-converter-only handling, and the +shared capability/intent/finalizer approach. Select the smallest approach that +also covers custom streaming senders. Do not broaden into a transport rewrite. + +## 2. Implement portable capability and intent + +- [x] Add a focused core module and export it from `edgezero-core/src/lib.rs`. +- [x] Default capability to unavailable and expose supported-codec information + before callers choose their output encoder. +- [x] Add explicit typed identity-body delivery selection and query operations. + Unsupported selection must leave the response unchanged. +- [x] Add focused tests and rustdoc examples. Keep the implementation dependency- + free beyond the existing core HTTP types. + +Run core tests for the new module, then neighboring response/body tests. Record +commands and tested source fingerprints outside the repository. + +## 3. Implement Fastly finalization + +- [x] Expose the supported capability and a reusable finalizer in the Fastly adapter. +- [x] Call the same finalizer from standard core-response conversion. +- [x] Preserve body ownership and no-poll semantics for custom senders. +- [x] Test header/body consistency, Vary preservation, idempotence, late overrides, + unmarked encoded responses, and unchanged privacy/cache metadata. +- [x] Exercise standard conversion in the actual Fastly/Viceroy target. Do not + substitute a host-only helper test for the SDK conversion path. +- [x] Document both normal and custom streaming integration and their ordering. + +Add focused regressions before the behavior where practical, observe a meaningful +failure, then implement and rerun the affected neighboring tests. + +## 4. Validate and publish EdgeZero first + +- [x] Run workspace tests, formatting, strict all-feature Clippy, the + `fastly cloudflare spin` feature check, and Spin `wasm32-wasip2` check. +- [x] Run targeted Fastly WASM tests with the repository-pinned Viceroy. +- [x] Run docs formatting, lint, and build. Classify any unrelated baseline failure + once rather than repeatedly rerunning an unchanged failure. +- [x] Independently inspect the complete source diff and actual test output. +- [ ] Commit and push only this feature and its docs. Open a draft EdgeZero PR + linked to #362 and Trusted Server #858/#884. Record its exact head SHA. + +One owner runs Cargo per target directory. Other repositories may use independent +build directories. Do not use mutable local path dependencies as the PR contract. + +## 5. Rebuild and integrate Trusted Server + +Start only after the EdgeZero draft PR exists. + +- [ ] Preserve the old Trusted Server branch as a reference. +- [ ] Rebuild the feature's old-main-to-head delta onto latest Trusted Server main, + without replaying inherited obsolete parser/merge history. +- [ ] Pin all six workspace EdgeZero declarations to the same exact published PR + SHA and verify all eight lockfile packages resolve to that source. +- [ ] Replace application-local generic marker, capability type, and Fastly header + helper with the EdgeZero API. Keep application options only where they still + bundle meaningful publisher policy. +- [ ] Keep input/output codec separation and SSAT eligibility, consent, auction, + EOF, cache, trace/deadline/ingress, and terminal privacy behavior. Carry an explicit + local pending output plan through inline, buffered, template, lazy, and ordered- + EOF paths. Template encoders must use that plan instead of mutable headers. + Clear representation metadata and mark only after installing the final identity + bytes or identity-producing stream; the old pre-transform marker sites cannot + call the checked EdgeZero API directly. +- [ ] Call the EdgeZero finalizer after late Fastly effects in the custom sender; + never route it through the collecting standard converter. +- [ ] Adapt and run the compression scenarios plus relevant new-main regressions. +- [ ] Update the Trusted Server design and guide to describe the new ownership + and temporary exact-commit dependency, with a removal condition of a merged + EdgeZero release. + +Rebase conflict resolution must preserve main's current trace and absolute- +deadline behavior. Resolve compression changes semantically, not by replacing +whole files from the old branch. Use a lease tied to the verified old remote head +when publishing the approved history rewrite. + +## 6. Validate and publish Trusted Server + +- [ ] Run focused publisher, adapter, and terminal-effect checks first. +- [ ] Run current repository CI gates and actual required EOF runtime controls, + preserving proof tied to final relevant inputs. +- [ ] Independently inspect the combined consumer diff and dependency consistency. +- [ ] Push the rebuilt branch, update #884, and keep both PRs draft. +- [ ] Update both issue descriptions and PR descriptions with bidirectional links, + exact dependency SHA, results, and remaining staging work. + +## Completion record + +Core, pure Fastly, and actual Viceroy contract tests observed meaningful failures +before implementation, then passed on the final source. The final focused runs +passed core 510, native Fastly 68, Viceroy contracts 12, SDK converter neighbors 4, +and the new rustdoc example. Scoped native/WASM strict Clippy passed. An independent +source review found no defects or smaller contract-preserving shape. Full workspace +tests, all-feature strict Clippy, feature compilation, Spin WASM checks, source/pin +guards, core documentation, and docs checks subsequently passed. Publication and +cross-repository integration results are recorded in the paired PR descriptions. + +Record finished steps here as work proceeds. PR descriptions hold final check +commands and outcomes. Runtime delivery negotiation and performance remain +unverified until real Fastly staging, even when all local checks pass. diff --git a/docs/superpowers/specs/2026-11-02-362-delivery-compression-design.md b/docs/superpowers/specs/2026-11-02-362-delivery-compression-design.md new file mode 100644 index 000000000..e4ae75f2b --- /dev/null +++ b/docs/superpowers/specs/2026-11-02-362-delivery-compression-design.md @@ -0,0 +1,167 @@ +# Adapter delivery compression + +## Goal and scope + +[Issue #362](https://github.com/IABTechLab/edgezero/issues/362) tracks reusable +adapter delivery compression. The immediate consumer is +[Trusted Server #858](https://github.com/IABTechLab/trusted-server/issues/858) +and its draft [PR #884](https://github.com/IABTechLab/trusted-server/pull/884). +Applications should be able to select identity output before encoding and ask a +capable adapter to compress delivery. Existing unmarked responses must not change. + +Baseline: EdgeZero main `87456c708ffe9edf86b3a0f8b06dfd15d60b07b5`. +Implementation steps are in the [matching plan](../plans/2026-11-02-362-delivery-compression.md). + +```mermaid +flowchart TD + A[Adapter advertises request capability] --> B[Application chooses output codec] + B --> C[Application installs identity bytes or final identity stream] + C --> D[Application attaches typed delivery intent] + D --> E[Late application header effects] + E --> F[Fastly finalizer restores delivery metadata without polling body] + F --> G[Existing buffered or custom streaming sender] + G --> H[Fastly negotiates compression with original reader request] +``` + +## Ownership + +EdgeZero owns the portable delivery capability and response intent, plus Fastly's +final compression headers. The application owns eligibility, response rewriting, +decoding, output-codec selection, privacy and cache policy. No SSAT, auction, +consent, template-cache, or publisher configuration enters EdgeZero. + +Fastly supports dynamic gzip/Brotli compression through `X-Compress-Hint: on`. +Axum, Cloudflare, and Spin retain their current delivery behavior. Do not claim a +compatible facility exists on them without a separate implementation and proof. + +## Design + +### Capability before encoding + +Add `edgezero_core::delivery_compression::DeliveryCompressionSupport` with +`Unavailable` as its default and `GzipBrotli` as the only supported capability. +`from_request(&Request)` reads the typed request extension, defaulting to +unavailable. This is a normal trusted adapter extension, not an authenticity token. +It describes delivery codecs, not a cache-header family or upstream-response +streaming capability. Consult it before choosing to skip the application's encoder. + +Fastly's common `convert_request` inserts `GzipBrotli`, covering standard runners +and custom callers of the raw request converters. Other adapters do not insert +support, so their default remains unavailable. Do not add a new registry, macro, +configuration flag, or per-adapter application literal for this capability. + +Keep reader negotiation outside this extraction. Fastly uses the original native +reader request at delivery. The application retains existing negotiation and +fallback logic; Trusted Server keeps in-guest output when negotiation fails or +selects deflate. Add no parallel `Accept-Encoding` parser or synthetic reader +headers in EdgeZero. + +### Explicit identity-body intent + +Define an opaque `DeliveryCompressionIntent` response extension in the same core +module. It is public for query/removal but privately constructed. +`DeliveryCompressionSupport::mark_identity_response(&mut Response)` returns +`Result<(), EdgeError>` and attaches the intent only for supported identity output. +Reject unavailable support or an existing non-identity encoding without changing +headers, body, or extensions. Accept absent encoding or a single case-insensitive +`identity` field with optional surrounding whitespace; reject coded, stacked, +repeated, or malformed fields. This prevents the concrete encoded-origin relabeling +risk, without inspecting body bytes. + +Do not use a configurable or network-supplied header as the authoritative marker. +The application must install completed identity bytes or its final identity- +producing stream before marking, not mark a raw origin body awaiting a decoder. +The intent survives ordinary response-header mutations. A component that later +re-encodes or replaces body representation must first remove the intent. + +The API must document that it neither decodes nor inspects body bytes and cannot +prove a stream's encoding. It must not silently opt an unavailable adapter into +offload. Unmarked responses, including already encoded bodies, remain unchanged. +Do not relabel compressed bytes as identity merely to enable the hint. + +### Fastly finalization + +Expose `edgezero_adapter_fastly::delivery_compression::finalize_response(&mut Response)` +in a feature-independent module using only core HTTP types. Standard +`from_core_response` calls it before consuming parts. Custom senders call the same +operation after their last header effects, before committing headers or consuming +response parts. No new universal sender abstraction is needed. + +For a selected identity-body response, finalization: + +- removes stale `Content-Encoding` and `Content-Length`; +- sets `X-Compress-Hint: on`; +- preserves all existing `Vary` values and adds `Accept-Encoding` only when neither + that token nor `*` already covers it; +- leaves body bytes, status, cookies, cache directives, validators, and privacy + policy untouched. + +Finalization is idempotent and does not poll, collect, replace, or wrap the body. +It reasserts the selected identity-output contract after late effects, not a new +codec decision. Applications must not replace identity bytes with encoded bytes +while retaining that intent. + +The current standard Fastly converter collects streamed core bodies before +returning a native response. Preserve that behavior rather than redesigning the +converter here. Trusted Server's custom sender must call the new finalizer +without going through that collecting path, preserving prefix-before-auction and +lazy origin consumption. + +### Alternatives + +- Keep all helpers application-local. Smallest immediate change, but duplicates + platform behavior and prevents other EdgeZero consumers from sharing it. +- Add only the hint to the standard Fastly converter. Misses custom streaming + senders and gives applications no way to choose their output codec beforehand. +- Rewrite response transport and all adapters. Unnecessary for this feature and + increases lifecycle and compatibility risk. + +The proposed capability, typed intent, and shared finalizer are the smallest split +that supports the actual consumer without moving application policy. + +## Compatibility and failure behavior + +This is an additive API with no manifest, route, or configuration changes. The +absence of capability/intent preserves existing behavior. Do not add body reads, +new buffering, codecs, retries, tasks, limits, or dependencies for this mechanism. +Existing conversion and body-read errors keep their current propagation. + +Other adapters must not interpret the intent as proof they can compress. Their +capability remains unavailable. Consumers are responsible for checking the +capability before changing output and for marking only response families whose +encoding they own. Existing HEAD/range/error policy remains application-owned. + +## Verification + +Core tests cover unavailable defaults, capability reporting, intent selection/query, +rejection of unavailable and encoded/malformed marking, and zero mutation on +failure. Fastly tests cover marked identity responses and unmarked identity/encoded +responses, repeated/case-insensitive/wildcard Vary, idempotence, stale framing, +late header effects, intent removal before re-encoding, and untouched privacy. +A stream that records polls must remain unpolled after the reusable finalizer. +Exercise the real standard converter under Viceroy as well as the custom-sender +header operation. Keep current multi-value header/body conversion regressions. + +Run focused tests before workspace format, tests, strict all-feature Clippy, +feature compilation, Spin WASM checks, and docs checks. The consumer integration +must pin one exact EdgeZero PR commit across all EdgeZero dependencies and prove +its actual buffered, lazy, template, and ordered-EOF paths. + +Local/Viceroy checks prove guest bytes and metadata only. Actual Fastly gzip/br +negotiation, identity refusal, progressive wire delivery, transfer sizes, and vCPU +benefits need a later staging run. Do not claim those from this implementation. + +## Review log + +- Design investigation, independent agent using `openai-codex/gpt-6.1-sol`: + confirmed the common Fastly conversion and custom sender boundaries. Required + marking only after installing identity output, not on an origin body awaiting + transformation. The consumer must carry a pending application output plan until + its buffered/lazy/template/EOF finalizers return the actual identity response. + The written contract was then checked against the same source with no blockers. +- Second solution/spec review, fresh `openai-codex/gpt-6.1-sol` session: + no blocking design defects. Confirmed the smallest API needs only the support + value, opaque intent, request lookup, checked marking, and shared finalizer. + Ordinary response extensions already supply query/removal. Required an explicit + pending-output plan for the consumer migration, including template encoding, + rather than mechanically replacing its early markers. diff --git a/tests/fixtures/reusable-app/crates/fixture-harness/src/net.rs b/tests/fixtures/reusable-app/crates/fixture-harness/src/net.rs index 8c5653245..d1b39ede9 100644 --- a/tests/fixtures/reusable-app/crates/fixture-harness/src/net.rs +++ b/tests/fixtures/reusable-app/crates/fixture-harness/src/net.rs @@ -156,6 +156,19 @@ fn serve( stream.write_all( b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\nConnection: close\r\n\r\npartial", )?; + if path.split('?').next() == Some("/abort-after-first-byte") { + stream.flush()?; + // Inject EOF only after the downstream client observes body bytes. + // Failed probes remain bounded by the client timeout; Backend::drop + // releases this waiter during cleanup. Keep plain /abort immediate + // for faults that occur before any downstream response exists. + drop( + signal + .1 + .wait_while(signal.0.lock().unwrap(), |ready| !*ready) + .map_err(|_| "release poisoned")?, + ); + } } else { stream.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok")?; } @@ -346,6 +359,85 @@ mod tests { assert_eq!(partial.body, b"partial"); } + #[test] + fn controlled_abort_waits_for_body_observation_before_closing() { + let backend = Backend::start().unwrap(); + let mut stream = TcpStream::connect(("127.0.0.1", backend.port)).unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + stream + .write_all(b"GET /abort-after-first-byte HTTP/1.1\r\nHost: localhost\r\n\r\n") + .unwrap(); + let mut reader = BufReader::new(stream); + loop { + let mut line = String::new(); + assert!(reader.read_line(&mut line).unwrap() > 0); + if line == "\r\n" { + break; + } + } + let mut prefix = [0; 7]; + reader.read_exact(&mut prefix).unwrap(); + assert_eq!(&prefix, b"partial"); + reader + .get_ref() + .set_read_timeout(Some(Duration::from_millis(100))) + .unwrap(); + let error = reader + .read(&mut [0]) + .expect_err("origin must stay open until the client acknowledges body bytes"); + assert!(matches!( + error.kind(), + std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut + )); + reader + .get_ref() + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + release(&backend.release); + assert_eq!(reader.read(&mut [0]).unwrap(), 0); + } + + #[test] + fn controlled_abort_preserves_the_observed_partial_response() { + let backend = Backend::start().unwrap(); + let failure = request( + backend.port, + "/abort-after-first-byte", + "x", + None, + Some(&backend.release), + ) + .unwrap_err(); + let partial = failure + .downcast_ref::() + .expect("response body should precede the controlled abort"); + assert_eq!(partial.status, 200); + assert_eq!(partial.body, b"partial"); + assert!(*backend.release.0.lock().unwrap()); + } + + #[test] + fn connection_closed_before_headers_is_not_an_interrupted_response() { + let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap(); + let port = listener.local_addr().unwrap().port(); + let server = thread::spawn(move || { + let (stream, _) = listener.accept().unwrap(); + let mut reader = BufReader::new(stream); + loop { + let mut line = String::new(); + assert!(reader.read_line(&mut line).unwrap() > 0); + if line == "\r\n" { + break; + } + } + }); + let failure = request(port, "/abort", "x", None, None).unwrap_err(); + server.join().unwrap(); + assert!(failure.downcast_ref::().is_none()); + } + #[test] fn connection_failure_is_not_an_interrupted_response() { let unused = port().unwrap(); diff --git a/tests/fixtures/reusable-app/crates/fixture-harness/src/runners.rs b/tests/fixtures/reusable-app/crates/fixture-harness/src/runners.rs index 53d1792e2..10499af7d 100644 --- a/tests/fixtures/reusable-app/crates/fixture-harness/src/runners.rs +++ b/tests/fixtures/reusable-app/crates/fixture-harness/src/runners.rs @@ -480,13 +480,14 @@ fn custom_stream_checks( large, json!({"variant":variant,"repetition":repetition,"assertion":"large_progressive_stream"}), ))?; + *backend.release.0.lock().unwrap() = false; let failure_token = crate::token(); let failure = net::request( port, "/stream-error", &failure_token, - Some(&backend.url(&format!("/abort?token={failure_token}"))), - None, + Some(&backend.url(&format!("/abort-after-first-byte?token={failure_token}"))), + Some(&backend.release), ); let failure = match failure { Ok(reply) => { @@ -496,7 +497,11 @@ fn custom_stream_checks( }; let partial = failure .downcast_ref::() - .ok_or("stream failure occurred before response body: not post-commit evidence")?; + .ok_or_else(|| { + format!( + "stream failure did not capture a partial response: not post-commit evidence: {failure}" + ) + })?; require( partial.status == 200 && partial.body == b"partial", "missing partial committed response",