diff --git a/Cargo.lock b/Cargo.lock index a19be7ab..cd222788 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5277,6 +5277,7 @@ dependencies = [ name = "trusted-server-core" version = "0.1.0" dependencies = [ + "async-stream", "async-trait", "base64", "brotli", diff --git a/Cargo.toml b/Cargo.toml index 27411acb..0512371e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -27,6 +27,7 @@ debug = 1 [workspace.dependencies] anyhow = "1" +async-stream = "0.3" async-trait = "0.1" axum = "0.8" base64 = "0.22" diff --git a/crates/trusted-server-adapter-fastly/src/app.rs b/crates/trusted-server-adapter-fastly/src/app.rs index e56498b1..b6889707 100644 --- a/crates/trusted-server-adapter-fastly/src/app.rs +++ b/crates/trusted-server-adapter-fastly/src/app.rs @@ -65,10 +65,10 @@ //! run on these responses. Legacy ran EC finalization on its own auth //! challenges. Like the 401 geo-skip, this is privacy-conservative: no EC //! cookies are issued to unauthenticated callers. -//! - **Publisher responses** are buffered (bounded by -//! `publisher.max_buffered_body_bytes`) instead of streamed to the client. -//! Asset responses are streamed straight to the client (see -//! [`dispatch_asset_fallback`]), matching legacy. +//! - **Publisher responses** keep Fastly origin bodies streaming through the +//! `EdgeZero` response body when the body is processable or pass-through. +//! Adapters without streaming-body support still use the bounded buffered +//! finalizer. //! - **Router-level 405s** (unregistered verbs) skip EC finalization along //! with the middleware chain; the entry point still adds TS headers. //! @@ -116,8 +116,8 @@ use trusted_server_core::proxy::{ handle_first_party_proxy_rebuild, handle_first_party_proxy_sign, AssetProxyCachePolicy, }; use trusted_server_core::publisher::{ - buffer_publisher_response_async, handle_page_bids, handle_publisher_request, - handle_tsjs_dynamic, page_bids_preflight_denied, AuctionDispatch, + handle_page_bids, handle_publisher_request, handle_tsjs_dynamic, page_bids_preflight_denied, + publisher_response_into_streaming_response, AuctionDispatch, }; use trusted_server_core::request_signing::{ handle_deactivate_key, handle_rotate_key, handle_trusted_server_discovery, @@ -721,10 +721,9 @@ async fn dispatch_fallback( let result = if uses_dynamic_tsjs_fallback(&method, &path) { handle_tsjs_dynamic(&req, &state.registry) } else if state.registry.has_route(&method, &path) { - // Integration-proxy responses are not bounded by publisher.max_buffered_body_bytes. - // Only the handle_publisher_request branch below routes through - // buffer_publisher_response_async. Integration responses are small in practice - // and the EdgeZero flag is off by default; extend the cap here if that changes. + // Integration-proxy responses are not bounded by + // publisher.max_buffered_body_bytes. Publisher fallback below uses the + // publisher-specific streaming finalizer instead. state .registry .handle_proxy(ProxyDispatchInput { @@ -773,9 +772,8 @@ async fn dispatch_fallback( match runtime_services_for_consent_route(&state.settings, services) { Ok(publisher_services) => { // Run the server-side auction with the configured creative- - // opportunity slots and collect the dispatched bids in the - // buffered finalize (`buffer_publisher_response_async`), matching - // the legacy streaming path. `handle_publisher_request` matches the + // opportunity slots and collect dispatched bids from the lazy + // publisher body stream. `handle_publisher_request` matches the // slots against the request path. The partner registry plus the // EC identity-graph KV (`ec.kv_graph`) enrich the bid request with // server-side EIDs, same as the legacy auction. @@ -798,13 +796,13 @@ async fn dispatch_fallback( .await { Ok(pub_response) => { - buffer_publisher_response_async( + publisher_response_into_streaming_response( pub_response, &method, - &state.settings, - &state.registry, - &state.orchestrator, - &publisher_services, + Arc::clone(&state.settings), + state.registry.as_ref(), + Arc::clone(&state.orchestrator), + publisher_services.clone(), ) .await } @@ -2155,6 +2153,10 @@ mod tests { #[async_trait::async_trait(?Send)] impl PlatformHttpClient for StreamingHttpClient { + fn supports_streaming_responses(&self) -> bool { + true + } + async fn send( &self, request: PlatformHttpRequest, @@ -2265,6 +2267,36 @@ mod tests { ); } + #[test] + fn dispatch_fallback_streams_publisher_body_without_buffering() { + // Regression guard for the publisher streaming cutover (#849): a + // successful publisher origin fetch must hand `edgezero_main` a lazy + // streaming body (`Body::Stream`) so headers commit at origin first + // byte, rather than draining the processed page into a buffered + // `Body::Once`. Core tests cover the rewrite pipeline itself; this + // guards the adapter wiring that could silently re-buffer. + let settings = test_settings(); + let state = build_state_from_settings(settings).expect("should build state"); + let services = streaming_runtime_services(); + let req = empty_request(Method::GET, "/article"); + + let response = block_on(super::dispatch_fallback(&state, &services, req)); + + assert_eq!( + response.status(), + StatusCode::OK, + "publisher proxy should succeed against the streaming origin stub" + ); + assert!( + matches!(response.body(), Body::Stream(_)), + "EdgeZero publisher dispatch must attach the lazy streaming body, not buffer it" + ); + assert!( + !response.headers().contains_key(header::CONTENT_LENGTH), + "processed streaming publisher responses must not carry a stale Content-Length" + ); + } + #[test] fn dispatch_runs_request_filter_and_threads_response_effects() { // Regression guard for the EdgeZero request-filter bypass: the publisher diff --git a/crates/trusted-server-adapter-fastly/src/main.rs b/crates/trusted-server-adapter-fastly/src/main.rs index d20de533..963686cb 100644 --- a/crates/trusted-server-adapter-fastly/src/main.rs +++ b/crates/trusted-server-adapter-fastly/src/main.rs @@ -321,10 +321,9 @@ fn run_edgezero_pull_sync_after_send( /// Sends a finalized `EdgeZero` response to the client. /// -/// Asset streams commit headers first, then pipe the origin body chunk by chunk -/// so large responses do not materialize in the Wasm heap. Publisher responses -/// are buffered by the server-side auction path so bids can be injected into the -/// document, and are sent in one shot along with all other responses. +/// Streaming `EdgeZero` bodies commit headers first, then pipe chunks to Fastly's +/// client stream so large asset and publisher-origin responses do not +/// materialize in the Wasm heap. fn send_edgezero_response( mut response: HttpResponse, request_filter_effects: Option<&RequestFilterEffects>, @@ -350,11 +349,11 @@ fn send_edgezero_response( match futures::executor::block_on(stream_asset_body(body, &mut streaming_body)) { Ok(()) => { if let Err(e) = streaming_body.finish() { - log::error!("failed to finish EdgeZero asset streaming body: {e}"); + log::error!("failed to finish EdgeZero streaming body: {e}"); } } Err(e) => { - log::error!("EdgeZero asset streaming failed: {e:?}"); + log::error!("EdgeZero streaming failed: {e:?}"); drop(streaming_body); } } diff --git a/crates/trusted-server-adapter-fastly/src/platform.rs b/crates/trusted-server-adapter-fastly/src/platform.rs index 106ced78..65d0f4b0 100644 --- a/crates/trusted-server-adapter-fastly/src/platform.rs +++ b/crates/trusted-server-adapter-fastly/src/platform.rs @@ -426,6 +426,10 @@ pub struct FastlyPlatformHttpClient; #[async_trait::async_trait(?Send)] impl PlatformHttpClient for FastlyPlatformHttpClient { + fn supports_streaming_responses(&self) -> bool { + true + } + async fn send( &self, request: PlatformHttpRequest, diff --git a/crates/trusted-server-core/Cargo.toml b/crates/trusted-server-core/Cargo.toml index ba88f361..bedefc32 100644 --- a/crates/trusted-server-core/Cargo.toml +++ b/crates/trusted-server-core/Cargo.toml @@ -13,6 +13,7 @@ workspace = true [dependencies] async-trait = { workspace = true } +async-stream = { workspace = true } base64 = { workspace = true } brotli = { workspace = true } bytes = { workspace = true } diff --git a/crates/trusted-server-core/src/platform/http.rs b/crates/trusted-server-core/src/platform/http.rs index 80bb2312..9e9337ed 100644 --- a/crates/trusted-server-core/src/platform/http.rs +++ b/crates/trusted-server-core/src/platform/http.rs @@ -276,6 +276,17 @@ pub trait PlatformHttpClient: Send + Sync { true } + /// Whether [`send`](Self::send) can preserve upstream response bodies as + /// [`Body::Stream`](edgezero_core::body::Body::Stream) when requested via + /// [`PlatformHttpRequest::with_stream_response`]. + /// + /// Adapters that cannot preserve streaming response bodies must keep the + /// default `false` so callers do not request a contract the adapter will + /// reject or silently buffer. + fn supports_streaming_responses(&self) -> bool { + false + } + /// Wait for one of the in-flight requests to complete. /// /// # Errors diff --git a/crates/trusted-server-core/src/platform/test_support.rs b/crates/trusted-server-core/src/platform/test_support.rs index ee7201fb..0c86b959 100644 --- a/crates/trusted-server-core/src/platform/test_support.rs +++ b/crates/trusted-server-core/src/platform/test_support.rs @@ -224,6 +224,9 @@ pub(crate) struct StubHttpClient { // Reported by supports_concurrent_fanout(); set false to emulate // platforms whose send_async executes eagerly (e.g. Cloudflare Workers). concurrent_fanout: std::sync::atomic::AtomicBool, + // Reported by supports_streaming_responses(); set true to emulate Fastly's + // streaming response support. + streaming_responses_supported: std::sync::atomic::AtomicBool, image_optimizer_options: Mutex>>, stream_response_flags: Mutex>, request_methods: Mutex>, @@ -246,6 +249,7 @@ impl StubHttpClient { request_headers: Mutex::new(Vec::new()), select_errors: Mutex::new(VecDeque::new()), concurrent_fanout: std::sync::atomic::AtomicBool::new(true), + streaming_responses_supported: std::sync::atomic::AtomicBool::new(false), image_optimizer_options: Mutex::new(Vec::new()), stream_response_flags: Mutex::new(Vec::new()), request_methods: Mutex::new(Vec::new()), @@ -260,6 +264,12 @@ impl StubHttpClient { .store(supported, std::sync::atomic::Ordering::Relaxed); } + /// Make `supports_streaming_responses()` report the given value. + pub fn set_streaming_responses_supported(&self, supported: bool) { + self.streaming_responses_supported + .store(supported, std::sync::atomic::Ordering::Relaxed); + } + /// Queue a canned response by status code and body bytes. pub fn push_response(&self, status: u16, body: Vec) { self.push_response_with_headers(status, body, Vec::<(String, String)>::new()); @@ -363,6 +373,11 @@ impl PlatformHttpClient for StubHttpClient { .load(std::sync::atomic::Ordering::Relaxed) } + fn supports_streaming_responses(&self) -> bool { + self.streaming_responses_supported + .load(std::sync::atomic::Ordering::Relaxed) + } + async fn send( &self, request: PlatformHttpRequest, diff --git a/crates/trusted-server-core/src/proxy.rs b/crates/trusted-server-core/src/proxy.rs index 19c10a80..00444b38 100644 --- a/crates/trusted-server-core/src/proxy.rs +++ b/crates/trusted-server-core/src/proxy.rs @@ -246,7 +246,10 @@ fn platform_response_to_fastly_asset(platform_resp: PlatformResponse) -> AssetPr } } -/// Stream an asset response body directly to a writable client stream. +/// Stream a platform response body directly to a writable client stream. +/// +/// Asset routes and Fastly `EdgeZero` publisher fallback both use this bridge +/// after headers have been committed through `stream_to_client()`. /// /// # Errors /// @@ -261,7 +264,7 @@ pub async fn stream_asset_body( output .write_all(bytes.as_ref()) .change_context(TrustedServerError::Proxy { - message: "failed to write buffered asset response body".to_string(), + message: "failed to write buffered platform response body".to_string(), })?; } EdgeBody::Stream(mut stream) => { @@ -274,7 +277,7 @@ pub async fn stream_asset_body( output .write_all(chunk.as_ref()) .change_context(TrustedServerError::Proxy { - message: "failed to write streaming asset response body".to_string(), + message: "failed to write streaming platform response body".to_string(), })?; } } diff --git a/crates/trusted-server-core/src/publisher.rs b/crates/trusted-server-core/src/publisher.rs index 6b0ea3a5..4d7e38cf 100644 --- a/crates/trusted-server-core/src/publisher.rs +++ b/crates/trusted-server-core/src/publisher.rs @@ -22,9 +22,15 @@ use std::io::Write; use std::sync::{Arc, Mutex}; use std::time::Duration; +use brotli::enc::writer::CompressorWriter; +use brotli::enc::BrotliEncoderParams; +use brotli::Decompressor; use cookie::CookieJar; use edgezero_core::body::Body as EdgeBody; use error_stack::{Report, ResultExt}; +use flate2::read::{GzDecoder, ZlibDecoder}; +use flate2::write::{GzEncoder, ZlibEncoder}; +use futures::StreamExt as _; use http::{header, HeaderValue, Method, Request, Response, StatusCode, Uri}; use crate::auction::endpoints::{ @@ -53,19 +59,132 @@ use crate::platform::{GeoInfo, PlatformBackendSpec, PlatformHttpRequest, Runtime use crate::price_bucket::{price_bucket, PriceGranularity}; use crate::rsc_flight::RscFlightUrlRewriter; use crate::settings::Settings; -use crate::streaming_processor::{Compression, PipelineConfig, StreamProcessor, StreamingPipeline}; +use crate::streaming_processor::{ + BodyStreamDecoder, BodyStreamEncoder, Compression, PipelineConfig, StreamProcessor, + StreamingPipeline, STREAM_CHUNK_SIZE, +}; use crate::streaming_replacer::create_url_replacer; const SUPPORTED_ENCODING_VALUES: [&str; 3] = ["gzip", "deflate", "br"]; const DEFAULT_PUBLISHER_FIRST_BYTE_TIMEOUT: Duration = Duration::from_secs(15); -/// Read buffer size for streaming body processing and brotli internal buffers. -/// Both the `Decompressor` and `CompressorWriter` use this value so all -/// brotli I/O layers operate on consistently-sized chunks. -const STREAM_CHUNK_SIZE: usize = 8192; +fn body_as_reader( + body: EdgeBody, +) -> Result, Report> { + let bytes = body.into_bytes().ok_or_else(|| { + Report::new(TrustedServerError::Proxy { + message: "streaming body cannot be processed by sync publisher pipeline".to_string(), + }) + })?; + Ok(std::io::Cursor::new(bytes)) +} + +struct BodyChunkSource { + body: Option, + chunk_size: usize, + max_bytes: usize, + bytes_seen: usize, + once_offset: usize, +} + +impl BodyChunkSource { + fn new(body: EdgeBody, chunk_size: usize) -> Self { + Self { + body: Some(body), + chunk_size, + max_bytes: usize::MAX, + bytes_seen: 0, + once_offset: 0, + } + } + + fn with_max_bytes(mut self, max_bytes: usize) -> Self { + self.max_bytes = max_bytes; + self + } + + async fn next_chunk(&mut self) -> Result, Report> { + // The body is polled in place (never moved out across an await) so a + // cancelled `next_chunk` future leaves the source resumable instead of + // silently reporting end-of-stream on the next call. + let pulled = match &mut self.body { + None => Ok(None), + Some(EdgeBody::Once(bytes)) => { + let end = (self.once_offset + self.chunk_size).min(bytes.len()); + if self.once_offset >= end { + Ok(None) + } else { + let chunk = bytes.slice(self.once_offset..end); + self.once_offset = end; + Ok(Some(chunk)) + } + } + Some(EdgeBody::Stream(stream)) => match stream.next().await { + Some(Ok(chunk)) => Ok(Some(chunk)), + Some(Err(err)) => Err(Report::new(TrustedServerError::Proxy { + message: format!("Failed to read publisher origin body stream: {err}"), + })), + None => Ok(None), + }, + }; + + let chunk = match pulled { + Ok(Some(chunk)) => chunk, + Ok(None) => { + self.body = None; + return Ok(None); + } + Err(err) => { + self.body = None; + return Err(err); + } + }; + + self.bytes_seen = self.bytes_seen.checked_add(chunk.len()).ok_or_else(|| { + Report::new(TrustedServerError::Proxy { + message: "publisher origin body byte count overflowed".to_string(), + }) + })?; + if self.bytes_seen > self.max_bytes { + return Err(Report::new(TrustedServerError::Proxy { + message: format!( + "publisher origin body exceeded {}-byte streaming limit", + self.max_bytes + ), + })); + } + + Ok(Some(chunk)) + } +} + +fn process_and_encode_chunk( + processor: &mut P, + encoder: &mut BodyStreamEncoder, + chunk: &[u8], + is_last: bool, + process_error: &str, +) -> Result, Report> { + let processed = + processor + .process_chunk(chunk, is_last) + .change_context(TrustedServerError::Proxy { + message: process_error.to_string(), + })?; + if processed.is_empty() { + return Ok(None); + } + let encoded = encoder.encode_chunk(processed)?; + if encoded.is_empty() { + return Ok(None); + } + Ok(Some(bytes::Bytes::from(encoded))) +} -fn body_as_reader(body: EdgeBody) -> std::io::Cursor { - std::io::Cursor::new(body.into_bytes().unwrap_or_default()) +// By-value signature so `map_err(publisher_stream_error)` works directly. +#[allow(clippy::needless_pass_by_value)] +fn publisher_stream_error(err: Report) -> std::io::Error { + std::io::Error::other(format!("{err:?}")) } fn not_found_response() -> Response { @@ -233,6 +352,55 @@ struct ProcessResponseParams<'a> { ad_bids_state: &'a Arc>>, } +struct PublisherBodyProcessor { + inner: Box, +} + +impl PublisherBodyProcessor { + fn new( + params: &OwnedProcessResponseParams, + settings: &Settings, + integration_registry: &IntegrationRegistry, + ) -> Result> { + let is_html = is_html_content_type(¶ms.content_type); + let is_rsc_flight = + content_type_contains_ascii_case_insensitive(¶ms.content_type, "text/x-component"); + let inner: Box = if is_html { + Box::new(create_html_stream_processor( + ¶ms.origin_host, + ¶ms.request_host, + ¶ms.request_scheme, + settings, + integration_registry, + params.ad_slots_script.as_deref().map(str::to_string), + Arc::clone(¶ms.ad_bids_state), + )?) + } else if is_rsc_flight { + Box::new(RscFlightUrlRewriter::new( + ¶ms.origin_host, + ¶ms.origin_url, + ¶ms.request_host, + ¶ms.request_scheme, + )) + } else { + Box::new(create_url_replacer( + ¶ms.origin_host, + ¶ms.origin_url, + ¶ms.request_host, + ¶ms.request_scheme, + )) + }; + + Ok(Self { inner }) + } +} + +impl StreamProcessor for PublisherBodyProcessor { + fn process_chunk(&mut self, chunk: &[u8], is_last: bool) -> Result, std::io::Error> { + self.inner.process_chunk(chunk, is_last) + } +} + /// Process response body through the streaming pipeline. /// /// Selects the appropriate processor based on content type (HTML rewriter, @@ -276,7 +444,7 @@ fn process_response_streaming( params.ad_slots_script.map(str::to_string), params.ad_bids_state.clone(), )?; - StreamingPipeline::new(config, processor).process(body_as_reader(body), output)?; + StreamingPipeline::new(config, processor).process(body_as_reader(body)?, output)?; } else if is_rsc_flight { // RSC Flight responses are length-prefixed (T rows). A naive string replacement will // corrupt the stream by changing byte lengths without updating the prefixes. @@ -286,7 +454,7 @@ fn process_response_streaming( params.request_host, params.request_scheme, ); - StreamingPipeline::new(config, processor).process(body_as_reader(body), output)?; + StreamingPipeline::new(config, processor).process(body_as_reader(body)?, output)?; } else { let replacer = create_url_replacer( params.origin_host, @@ -294,12 +462,408 @@ fn process_response_streaming( params.request_host, params.request_scheme, ); - StreamingPipeline::new(config, replacer).process(body_as_reader(body), output)?; + StreamingPipeline::new(config, replacer).process(body_as_reader(body)?, output)?; + } + + Ok(()) +} + +async fn process_response_streaming_async( + body: EdgeBody, + output: &mut W, + params: &OwnedProcessResponseParams, + settings: &Settings, + integration_registry: &IntegrationRegistry, +) -> Result<(), Report> { + log::debug!( + "process_response_streaming_async: content_type={}, content_encoding={}", + params.content_type, + params.content_encoding + ); + + let compression = Compression::from_content_encoding(¶ms.content_encoding); + let mut processor = PublisherBodyProcessor::new(params, settings, integration_registry)?; + process_body_chunks_async( + body, + output, + &mut processor, + compression, + settings.publisher.max_buffered_body_bytes, + ) + .await +} + +/// Pull, decode, process, and encode the next chunk of a no-hold pipeline. +/// +/// Returns `Ok(None)` when the source is exhausted; the caller must then emit +/// [`passthrough_finish_segments`]. Shared by the write-sink driver +/// ([`process_body_chunks_async`]) and the lazy publisher body stream so the +/// two no-hold paths cannot drift apart. +async fn passthrough_step( + source: &mut BodyChunkSource, + decoder: &mut BodyStreamDecoder, + encoder: &mut BodyStreamEncoder, + processor: &mut P, +) -> Result>, Report> { + let Some(raw_chunk) = source.next_chunk().await? else { + return Ok(None); + }; + let decoded = decoder.decode_chunk(raw_chunk)?; + if decoded.is_empty() { + return Ok(Some(Vec::new())); + } + let mut segments = Vec::new(); + if let Some(encoded) = process_and_encode_chunk( + processor, + encoder, + &decoded, + false, + "Failed to process chunk", + )? { + segments.push(encoded); + } + Ok(Some(segments)) +} + +async fn process_body_chunks_async( + body: EdgeBody, + writer: &mut W, + processor: &mut P, + compression: Compression, + max_body_bytes: usize, +) -> Result<(), Report> { + let mut decoder = BodyStreamDecoder::new(compression, max_body_bytes); + let mut encoder = BodyStreamEncoder::new(compression); + let mut source = BodyChunkSource::new(body, STREAM_CHUNK_SIZE).with_max_bytes(max_body_bytes); + + while let Some(segments) = + passthrough_step(&mut source, &mut decoder, &mut encoder, processor).await? + { + for encoded in segments { + write_encoded_segment(writer, &encoded)?; + } + } + + for encoded in passthrough_finish_segments(processor, &mut decoder, &mut encoder)? { + write_encoded_segment(writer, &encoded)?; } + writer.flush().change_context(TrustedServerError::Proxy { + message: "Failed to flush output".to_string(), + })?; Ok(()) } +/// Write one encoded output segment produced by the chunk pipeline. +fn write_encoded_segment( + writer: &mut W, + encoded: &[u8], +) -> Result<(), Report> { + writer + .write_all(encoded) + .change_context(TrustedServerError::Proxy { + message: "Failed to write encoded chunk".to_string(), + }) +} + +/// Finalize a no-hold chunk pipeline: drain the decoder tail through the +/// processor, signal end-of-stream to the processor, and emit the encoder +/// trailer. Returns the encoded segments for the caller to emit. +fn passthrough_finish_segments( + processor: &mut P, + decoder: &mut BodyStreamDecoder, + encoder: &mut BodyStreamEncoder, +) -> Result, Report> { + let mut segments = Vec::new(); + let decoded_tail = decoder.finish()?; + if !decoded_tail.is_empty() { + if let Some(encoded) = process_and_encode_chunk( + processor, + encoder, + &decoded_tail, + false, + "Failed to process decoded tail", + )? { + segments.push(encoded); + } + } + if let Some(encoded) = process_and_encode_chunk( + processor, + encoder, + &[], + true, + "Failed to finalize processor", + )? { + segments.push(encoded); + } + let trailer = encoder.finish()?; + if !trailer.is_empty() { + segments.push(bytes::Bytes::from(trailer)); + } + Ok(segments) +} + +/// Owns a [`DispatchedAuction`] and logs if it is dropped uncollected. +/// +/// The lazy publisher body stream can be dropped at any await point — a +/// client disconnect aborts the transfer mid-body, or the response may never +/// be polled at all. Async telemetry cannot run in `Drop`, so the loss is +/// surfaced in logs; the abandoned-auction telemetry event is only emitted on +/// error paths that can still await (see [`abandon_hold_auction`]). +struct DispatchedAuctionGuard { + dispatched: Option, +} + +impl DispatchedAuctionGuard { + fn new(dispatched: DispatchedAuction) -> Self { + Self { + dispatched: Some(dispatched), + } + } + + fn take(&mut self) -> Option { + self.dispatched.take() + } +} + +impl Drop for DispatchedAuctionGuard { + fn drop(&mut self) { + if self.dispatched.is_some() { + log::warn!( + "Dispatched server-side auction dropped without collection; SSP bid responses discarded (publisher body stream aborted or never polled)" + ); + } + } +} + +/// Mutable auction-hold state threaded through the streaming hold pipeline. +struct AuctionHoldState { + hold: Option, + dispatched: DispatchedAuctionGuard, + telemetry: AuctionTelemetryCarry, +} + +impl AuctionHoldState { + fn new(dispatched: DispatchedAuctionGuard, telemetry: AuctionTelemetryCarry) -> Self { + Self { + hold: Some(BodyCloseHoldBuffer::new()), + dispatched, + telemetry, + } + } +} + +/// Abandon the in-flight auction (if still pending) with the given telemetry +/// reason. No-op once the auction has been collected or already abandoned. +async fn abandon_hold_auction( + state: &mut AuctionHoldState, + services: &RuntimeServices, + reason: &'static str, +) { + if let Some(dispatched) = state.dispatched.take() { + emit_abandoned_auction( + services, + state.telemetry.observation.take(), + dispatched, + reason, + ) + .await; + } +} + +/// Feed one decoded chunk through the close-body hold and processor. +/// +/// Returns the encoded output segments for the caller to emit — written to a +/// client stream by [`body_close_hold_loop_stream`], yielded from the lazy +/// body by [`publisher_response_into_streaming_response`]. Both async hold +/// paths share this function so their behavior cannot drift apart. +/// +/// When the raw `( + processor: &mut P, + encoder: &mut BodyStreamEncoder, + chunk: &[u8], + state: &mut AuctionHoldState, + collect_refs: &AuctionHoldCollectRefs<'_>, +) -> Result, Report> { + let mut segments = Vec::new(); + if let Some(hold_buffer) = state.hold.as_mut() { + let ready = hold_buffer.push(chunk); + match process_and_encode_chunk(processor, encoder, &ready, false, "Failed to process chunk") + { + Ok(Some(encoded)) => segments.push(encoded), + Ok(None) => {} + Err(err) => { + abandon_hold_auction(state, collect_refs.services, "stream_process_error").await; + return Err(err); + } + } + + if state + .hold + .as_ref() + .is_some_and(BodyCloseHoldBuffer::found_close) + { + let dispatched = state + .dispatched + .take() + .expect("should have dispatched auction to collect"); + collect_stream_auction( + dispatched, + state.telemetry.take(), + collect_refs.price_granularity, + collect_refs.ad_bids_state, + collect_refs.orchestrator, + collect_refs.services, + collect_refs.settings, + ) + .await; + + let held = state + .hold + .take() + .expect("should have close-body hold buffer") + .finish(); + if let Some(encoded) = process_and_encode_chunk( + processor, + encoder, + &held, + false, + "Failed to process held body close", + )? { + segments.push(encoded); + } + } + } else { + match process_and_encode_chunk(processor, encoder, chunk, false, "Failed to process chunk") + { + Ok(Some(encoded)) => segments.push(encoded), + Ok(None) => {} + Err(err) => { + abandon_hold_auction(state, collect_refs.services, "stream_process_error").await; + return Err(err); + } + } + } + Ok(segments) +} + +/// Pull and decode the next chunk of the close-body hold pipeline, feeding it +/// through [`hold_step_decoded_chunk`]. +/// +/// Returns `Ok(None)` when the source is exhausted; the caller must then emit +/// [`hold_finish_segments`]. On read or decode failure the pending auction is +/// abandoned before the error is returned. Shared by the write-sink driver +/// ([`body_close_hold_loop_stream`]) and the lazy publisher body stream so +/// the two hold paths cannot drift apart. +async fn hold_step_next_chunk( + source: &mut BodyChunkSource, + decoder: &mut BodyStreamDecoder, + encoder: &mut BodyStreamEncoder, + processor: &mut P, + state: &mut AuctionHoldState, + collect_refs: &AuctionHoldCollectRefs<'_>, +) -> Result>, Report> { + let raw_chunk = match source.next_chunk().await { + Ok(Some(chunk)) => chunk, + Ok(None) => return Ok(None), + Err(err) => { + abandon_hold_auction(state, collect_refs.services, "stream_read_error").await; + return Err(err); + } + }; + let decoded = match decoder.decode_chunk(raw_chunk) { + Ok(decoded) => decoded, + Err(err) => { + abandon_hold_auction(state, collect_refs.services, "stream_decode_error").await; + return Err(err); + } + }; + if decoded.is_empty() { + return Ok(Some(Vec::new())); + } + hold_step_decoded_chunk(processor, encoder, &decoded, state, collect_refs) + .await + .map(Some) +} + +/// Finalize the close-body hold pipeline at end of the origin stream. +/// +/// Drains the decoder tail through the hold (or straight through when the +/// hold was already released mid-stream), collects the auction if the +/// close-body tag never streamed, processes the held tail plus the +/// processor's final chunk, and emits the encoder trailer. Returns the +/// encoded segments for the caller to emit. On decoder failure the pending +/// auction is abandoned before the error is returned. +async fn hold_finish_segments( + processor: &mut P, + decoder: &mut BodyStreamDecoder, + encoder: &mut BodyStreamEncoder, + state: &mut AuctionHoldState, + collect_refs: &AuctionHoldCollectRefs<'_>, +) -> Result, Report> { + let mut segments = Vec::new(); + + let decoded_tail = match decoder.finish() { + Ok(decoded_tail) => decoded_tail, + Err(err) => { + abandon_hold_auction(state, collect_refs.services, "stream_decode_error").await; + return Err(err); + } + }; + if !decoded_tail.is_empty() { + segments.extend( + hold_step_decoded_chunk(processor, encoder, &decoded_tail, state, collect_refs).await?, + ); + } + + if let Some(hold) = state.hold.take() { + let dispatched = state + .dispatched + .take() + .expect("should have dispatched auction to collect"); + collect_stream_auction( + dispatched, + state.telemetry.take(), + collect_refs.price_granularity, + collect_refs.ad_bids_state, + collect_refs.orchestrator, + collect_refs.services, + collect_refs.settings, + ) + .await; + + let held = hold.finish(); + if let Some(encoded) = process_and_encode_chunk( + processor, + encoder, + &held, + false, + "Failed to process held body close", + )? { + segments.push(encoded); + } + } + + if let Some(encoded) = process_and_encode_chunk( + processor, + encoder, + &[], + true, + "Failed to finalize processor", + )? { + segments.push(encoded); + } + let trailer = encoder.finish()?; + if !trailer.is_empty() { + segments.push(bytes::Bytes::from(trailer)); + } + Ok(segments) +} + /// Create a unified HTML stream processor. /// /// Builds the config via [`HtmlProcessorConfig::from_settings`] and then @@ -333,22 +897,23 @@ fn create_html_stream_processor( /// Result of publisher request handling, indicating whether the response body /// should be streamed or has already been buffered. pub enum PublisherResponse { - /// Response is fully buffered and ready to send via `send_to_client()`. + /// Response returned unmodified, ready to send via `send_to_client()`. + /// + /// On streaming adapters the unmodified body may still be a live + /// [`EdgeBody::Stream`] (the origin fetch requested streaming before the + /// response was classified); it passes through to the client untouched. Buffered(Response), /// Response headers are ready for a streaming response. Covers processable /// content on any status (2xx or non-2xx — e.g., branded 404/500 HTML and /// error JSON still get URL rewriting) where the encoding is supported. /// Post-processors run inside the streaming processor, so processable HTML - /// is streamed regardless of whether any are registered. The caller must: - /// 1. Call `finalize_response()` on the response - /// 2. Call `response.stream_to_client()` to get a `StreamingBody` - /// 3. Call `stream_publisher_body()` with the body and streaming writer - /// 4. Call `StreamingBody::finish()` + /// is streamed regardless of whether any are registered. /// - /// **Interim (PR 15):** `body` has already been fully materialised into - /// WASM heap by the platform HTTP client. `stream_publisher_body` reads - /// from an in-memory buffer, not a live origin stream. The origin-side - /// peak is bounded by `MAX_PLATFORM_RESPONSE_BODY_BYTES`. + /// Adapters with platform streaming support preserve `body` as + /// [`EdgeBody::Stream`] and attach a lazy processed stream via + /// [`publisher_response_into_streaming_response`]. Buffered adapters use + /// [`buffer_publisher_response_async`] and are bounded by + /// `settings.publisher.max_buffered_body_bytes`. Stream { /// Response with all headers set (EC ID, cookies, etc.) /// but body not yet written. `Content-Length` already removed. @@ -363,12 +928,9 @@ pub enum PublisherResponse { /// `finalize_response()` and `send_to_client()` are applied at the outer /// response-dispatch level, not in this arm. /// - /// `Content-Length` is preserved — the body is unmodified. - /// - /// **Interim (PR 15):** `body` has been fully materialised into WASM heap. - /// Previously, binary assets streamed lazily from origin with no WASM - /// buffering. This path is now bounded by `MAX_PLATFORM_RESPONSE_BODY_BYTES`; - /// assets exceeding that limit return an error instead of exhausting heap. + /// `Content-Length` is preserved — the body is unmodified. Streaming + /// adapters reattach the origin body directly so non-processable 2xx bodies + /// can pass through without materializing in WASM memory. PassThrough { /// Response with all headers set but body not yet written. response: Response, @@ -465,7 +1027,7 @@ pub struct OwnedProcessResponseParams { /// statuses (204, 304) carry no body but may advertise the `GET` representation's /// length, so they skip the buffer and length rewrite. /// -/// Every adapter (Axum, Cloudflare, Spin, and the Fastly `EdgeZero` path) calls +/// Buffered adapters (Axum, Cloudflare, Spin, and non-streaming fallbacks) call /// this: it drives /// [`stream_publisher_body_async`], which awaits /// [`AuctionOrchestrator::collect_dispatched_auction`], writes the winning bids @@ -530,41 +1092,214 @@ pub async fn buffer_publisher_response_async( } } -/// Returns `true` when a buffered publisher response should carry a body and a -/// recomputed `Content-Length`. +/// Convert a [`PublisherResponse`] into a response that preserves streaming +/// bodies where possible. /// -/// `HEAD` responses and bodiless statuses (204, 304) carry no body; rewriting -/// their `Content-Length` to the (empty) buffered length would mislead clients -/// and caches, so the origin metadata is preserved instead. -fn response_carries_body(method: &Method, status: StatusCode) -> bool { - *method != Method::HEAD - && status != StatusCode::NO_CONTENT - && status != StatusCode::NOT_MODIFIED -} - -/// A [`Write`] sink that buffers into a `Vec` but fails once the configured -/// byte limit would be exceeded. +/// Buffered adapters should keep using [`buffer_publisher_response_async`]. +/// Fastly uses this helper before the entry point commits headers, allowing the +/// response body to be pulled lazily by `stream_to_client()`. /// -/// Used to bound in-WASM-heap buffering of decoded/re-written publisher bodies. -/// A highly-compressible origin response can sit under the platform raw-body cap -/// yet expand past a safe heap size after decode and post-processing; this writer -/// turns that into a recoverable error instead of an out-of-memory abort. -pub struct BoundedWriter { - inner: Vec, - limit: usize, -} - -impl BoundedWriter { - /// Creates a writer that accepts at most `limit` bytes before erroring. - #[must_use] - pub fn new(limit: usize) -> Self { - Self { - inner: Vec::new(), - limit, +/// # Errors +/// +/// Returns an error if processor construction fails before the streaming body +/// is created; a dispatched auction is abandoned with `processor_init_error` +/// telemetry first, matching the buffered finalizer. +pub async fn publisher_response_into_streaming_response( + publisher_response: PublisherResponse, + method: &Method, + settings: Arc, + integration_registry: &IntegrationRegistry, + orchestrator: Arc, + services: RuntimeServices, +) -> Result, Report> { + match publisher_response { + PublisherResponse::Buffered(response) => Ok(response), + PublisherResponse::PassThrough { mut response, body } => { + if response_carries_body(method, response.status()) { + *response.body_mut() = body; + } + Ok(response) } - } - - /// Consumes the writer and returns the buffered bytes. + PublisherResponse::Stream { + mut response, + body, + params, + } => { + if !response_carries_body(method, response.status()) { + if params.dispatched_auction.is_some() { + // A bodiless response (HEAD navigation, 204/304) has no + // `` to inject bids into, so the dispatched SSP + // requests are wasted — surface it for quota observability, + // matching the buffered finalizer. + log::warn!( + "Server-side auction dispatched but response is bodiless (method: {}, status: {}); in-flight SSP bid requests will not be collected", + method, + response.status(), + ); + } + return Ok(response); + } + + response.headers_mut().remove(header::CONTENT_LENGTH); + let mut params = *params; + let mut processor = + match PublisherBodyProcessor::new(¶ms, &settings, integration_registry) { + Ok(processor) => processor, + Err(err) => { + // Parity with the buffered finalizer: a processor + // construction failure abandons the dispatched auction + // with telemetry instead of dropping the in-flight SSP + // responses silently. + if let Some(dispatched) = params.dispatched_auction.take() { + emit_abandoned_auction( + &services, + params.auction_observation.take(), + dispatched, + "processor_init_error", + ) + .await; + } + return Err(err); + } + }; + // The guard is created before the lazy stream so an auction whose + // response body is dropped unpolled still logs the loss. + let dispatched_auction = params.dispatched_auction.take().map(|dispatched| { + let telemetry = AuctionTelemetryCarry { + observation: params.auction_observation.take(), + auction_request: params.auction_request.take(), + }; + (DispatchedAuctionGuard::new(dispatched), telemetry) + }); + let stream = async_stream::try_stream! { + let compression = Compression::from_content_encoding(¶ms.content_encoding); + let max_body_bytes = settings.publisher.max_buffered_body_bytes; + let mut decoder = BodyStreamDecoder::new(compression, max_body_bytes); + let mut encoder = BodyStreamEncoder::new(compression); + let mut source = BodyChunkSource::new(body, STREAM_CHUNK_SIZE) + .with_max_bytes(max_body_bytes); + + // HTML rides the close-body hold so bids land before ``; + // non-HTML has no injection point, so its auction is collected + // before any byte streams (matching the buffered finalizer). + let mut hold_auction = None; + if let Some((mut guard, telemetry)) = dispatched_auction { + if is_html_content_type(¶ms.content_type) { + hold_auction = Some((guard, telemetry)); + } else if let Some(dispatched) = guard.take() { + collect_non_html_auction( + dispatched, + telemetry, + ¶ms, + &orchestrator, + &services, + &settings, + ) + .await; + } + } + + if let Some((guard, telemetry)) = hold_auction { + let mut state = AuctionHoldState::new(guard, telemetry); + let collect_refs = AuctionHoldCollectRefs { + price_granularity: params.price_granularity, + ad_bids_state: ¶ms.ad_bids_state, + orchestrator: &orchestrator, + services: &services, + settings: &settings, + }; + + while let Some(segments) = hold_step_next_chunk( + &mut source, + &mut decoder, + &mut encoder, + &mut processor, + &mut state, + &collect_refs, + ) + .await + .map_err(publisher_stream_error)? + { + for encoded in segments { + yield encoded; + } + } + + for encoded in hold_finish_segments( + &mut processor, + &mut decoder, + &mut encoder, + &mut state, + &collect_refs, + ) + .await + .map_err(publisher_stream_error)? + { + yield encoded; + } + } else { + while let Some(segments) = passthrough_step( + &mut source, + &mut decoder, + &mut encoder, + &mut processor, + ) + .await + .map_err(publisher_stream_error)? + { + for encoded in segments { + yield encoded; + } + } + for encoded in + passthrough_finish_segments(&mut processor, &mut decoder, &mut encoder) + .map_err(publisher_stream_error)? + { + yield encoded; + } + } + }; + *response.body_mut() = EdgeBody::from_stream::<_, std::io::Error>(stream); + Ok(response) + } + } +} + +/// Returns `true` when a buffered publisher response should carry a body and a +/// recomputed `Content-Length`. +/// +/// `HEAD` responses and bodiless statuses (204, 304) carry no body; rewriting +/// their `Content-Length` to the (empty) buffered length would mislead clients +/// and caches, so the origin metadata is preserved instead. +fn response_carries_body(method: &Method, status: StatusCode) -> bool { + *method != Method::HEAD + && status != StatusCode::NO_CONTENT + && status != StatusCode::NOT_MODIFIED +} + +/// A [`Write`] sink that buffers into a `Vec` but fails once the configured +/// byte limit would be exceeded. +/// +/// Used to bound in-WASM-heap buffering of decoded/re-written publisher bodies. +/// A highly-compressible origin response can sit under the platform raw-body cap +/// yet expand past a safe heap size after decode and post-processing; this writer +/// turns that into a recoverable error instead of an out-of-memory abort. +pub struct BoundedWriter { + inner: Vec, + limit: usize, +} + +impl BoundedWriter { + /// Creates a writer that accepts at most `limit` bytes before erroring. + #[must_use] + pub fn new(limit: usize) -> Self { + Self { + inner: Vec::new(), + limit, + } + } + + /// Consumes the writer and returns the buffered bytes. #[must_use] pub fn into_inner(self) -> Vec { self.inner @@ -652,7 +1387,18 @@ pub async fn stream_publisher_body_async( services: &RuntimeServices, ) -> Result<(), Report> { let Some(dispatched) = params.dispatched_auction.take() else { - // No auction — use the existing sync pipeline unchanged. + if body.is_stream() { + return process_response_streaming_async( + body, + output, + params, + settings, + integration_registry, + ) + .await; + } + + // No auction and already-buffered body — keep the existing sync pipeline. return stream_publisher_body(body, output, params, settings, integration_registry); }; let telemetry = AuctionTelemetryCarry { @@ -665,35 +1411,25 @@ pub async fn stream_publisher_body_async( if !is_html { // Non-HTML: collect auction first, then stream. There is no // to hold, so delaying the entire body until collection is acceptable. - let placeholder = mediator_placeholder_request(); - let result = orchestrator - .collect_dispatched_auction( - dispatched, - services, - &make_collect_context(settings, services, &placeholder), + collect_non_html_auction( + dispatched, + telemetry, + params, + orchestrator, + services, + settings, + ) + .await; + if body.is_stream() { + return process_response_streaming_async( + body, + output, + params, + settings, + integration_registry, ) .await; - if let (Some(observation), Some(auction_request)) = - (telemetry.observation, telemetry.auction_request.as_ref()) - { - emit_auction_events_best_effort_lazy(services, || { - build_auction_events( - observation, - AuctionTerminalOutcome::Completed { - request: auction_request, - result: &result, - }, - ) - }) - .await; } - - write_bids_to_state( - &result.winning_bids, - params.price_granularity, - ¶ms.ad_bids_state, - settings.debug.inject_adm_for_testing, - ); return stream_publisher_body(body, output, params, settings, integration_registry); } @@ -917,6 +1653,14 @@ struct AuctionCollectCtx<'a> { settings: &'a Settings, } +struct AuctionHoldCollectRefs<'a> { + price_granularity: PriceGranularity, + ad_bids_state: &'a Arc>>, + orchestrator: &'a AuctionOrchestrator, + services: &'a RuntimeServices, + settings: &'a Settings, +} + /// Run the close-body hold loop for HTML bodies, collecting the auction before /// the raw `( @@ -926,13 +1670,20 @@ async fn stream_html_with_auction_hold( compression: Compression, ctx: AuctionCollectCtx<'_>, ) -> Result<(), Report> { - use brotli::enc::writer::CompressorWriter; - use brotli::enc::BrotliEncoderParams; - use brotli::Decompressor; - use flate2::read::{GzDecoder, ZlibDecoder}; - use flate2::write::{GzEncoder, ZlibEncoder}; + if body.is_stream() { + let max_body_bytes = ctx.settings.publisher.max_buffered_body_bytes; + return body_close_hold_loop_stream( + body, + output, + processor, + compression, + ctx, + max_body_bytes, + ) + .await; + } - let body = body_as_reader(body); + let body = body_as_reader(body)?; match compression { Compression::None => body_close_hold_loop(body, output, processor, ctx).await, Compression::Gzip => { @@ -969,6 +1720,78 @@ async fn stream_html_with_auction_hold( } } +/// Async-pull variant of [`body_close_hold_loop`] for live origin streams. +/// +/// Shares [`hold_step_next_chunk`] and [`hold_finish_segments`] with the +/// lazy streaming body built by [`publisher_response_into_streaming_response`], +/// so the two async hold paths cannot drift apart. +/// +/// No production caller reaches this today: it is only entered through +/// [`buffer_publisher_response_async`], and the buffered adapters (Axum, +/// Cloudflare, Spin) never produce `Body::Stream` because the publisher fetch +/// is gated on `supports_streaming_responses()`. It is groundwork for those +/// adapters' streaming cutover; Fastly uses the lazy stream instead. +async fn body_close_hold_loop_stream( + body: EdgeBody, + writer: &mut W, + processor: &mut P, + compression: Compression, + ctx: AuctionCollectCtx<'_>, + max_body_bytes: usize, +) -> Result<(), Report> { + let AuctionCollectCtx { + dispatched, + telemetry, + price_granularity, + ad_bids_state, + orchestrator, + services, + settings, + } = ctx; + let mut decoder = BodyStreamDecoder::new(compression, max_body_bytes); + let mut encoder = BodyStreamEncoder::new(compression); + let mut source = BodyChunkSource::new(body, STREAM_CHUNK_SIZE).with_max_bytes(max_body_bytes); + let mut state = AuctionHoldState::new(DispatchedAuctionGuard::new(dispatched), telemetry); + let collect_refs = AuctionHoldCollectRefs { + price_granularity, + ad_bids_state, + orchestrator, + services, + settings, + }; + + while let Some(segments) = hold_step_next_chunk( + &mut source, + &mut decoder, + &mut encoder, + processor, + &mut state, + &collect_refs, + ) + .await? + { + for encoded in segments { + write_encoded_segment(writer, &encoded)?; + } + } + + for encoded in hold_finish_segments( + processor, + &mut decoder, + &mut encoder, + &mut state, + &collect_refs, + ) + .await? + { + write_encoded_segment(writer, &encoded)?; + } + writer.flush().change_context(TrustedServerError::Proxy { + message: "Failed to flush output".to_string(), + })?; + Ok(()) +} + const BODY_CLOSE_PREFIX: &[u8] = b"` to inject into, so bids are written to state up front and the +/// auction telemetry completes immediately. +async fn collect_non_html_auction( + dispatched: DispatchedAuction, + telemetry: AuctionTelemetryCarry, + params: &OwnedProcessResponseParams, + orchestrator: &AuctionOrchestrator, + services: &RuntimeServices, + settings: &Settings, +) { + let placeholder = mediator_placeholder_request(); + let result = orchestrator + .collect_dispatched_auction( + dispatched, + services, + &make_collect_context(settings, services, &placeholder), + ) + .await; + if let (Some(observation), Some(auction_request)) = + (telemetry.observation, telemetry.auction_request.as_ref()) + { + emit_auction_events_best_effort_lazy(services, || { + build_auction_events( + observation, + AuctionTerminalOutcome::Completed { + request: auction_request, + result: &result, + }, + ) + }) + .await; + } + write_bids_to_state( + &result.winning_bids, + params.price_granularity, + ¶ms.ad_bids_state, + settings.debug.inject_adm_for_testing, + ); +} + async fn collect_stream_auction( dispatched: DispatchedAuction, telemetry: AuctionTelemetryCarry, @@ -1600,11 +2464,17 @@ pub async fn handle_publisher_request( // SSP requests are already racing through the platform HTTP client, so // origin TTFB tracks origin latency rather than the auction timeout. - let mut response = match services - .http_client() - .send(PlatformHttpRequest::new(req, backend_name)) - .await - { + // + // Streaming is gated on the capability (unlike the asset-proxy path, which + // sets the flag unconditionally and tolerates buffered fallback): adapters + // without streaming support may reject the flag outright rather than + // silently buffering, which would fail every publisher fetch. + let mut platform_request = PlatformHttpRequest::new(req, backend_name); + if services.http_client().supports_streaming_responses() { + platform_request = platform_request.with_stream_response(); + } + + let mut response = match services.http_client().send(platform_request).await { Ok(platform_response) => platform_response.response, Err(err) => { if let Some(dispatched) = dispatched_auction.take() { @@ -2428,6 +3298,7 @@ pub async fn handle_page_bids( #[cfg(test)] mod tests { + use std::future::Future as _; use std::io::{self, Read as _, Write as _}; use std::sync::atomic::{AtomicUsize, Ordering}; @@ -2505,6 +3376,24 @@ mod tests { output } + fn deflate_encode(input: &[u8]) -> Vec { + let mut encoder = + flate2::write::ZlibEncoder::new(Vec::new(), flate2::Compression::default()); + encoder + .write_all(input) + .expect("should write deflate test input"); + encoder.finish().expect("should finish deflate encoding") + } + + fn deflate_decode(input: &[u8]) -> Vec { + let mut decoder = flate2::read::ZlibDecoder::new(input); + let mut output = Vec::new(); + decoder + .read_to_end(&mut output) + .expect("should decode deflate test output"); + output + } + fn brotli_encode(input: &[u8]) -> Vec { let mut encoder = CompressorWriter::new(Vec::new(), 4096, 5, 22); encoder @@ -2733,6 +3622,63 @@ mod tests { ); } + #[tokio::test] + async fn publisher_origin_fetch_leaves_stream_response_disabled_when_unsupported() { + let settings = create_test_settings(); + let stub = Arc::new(StubHttpClient::new()); + stub.push_response_with_headers( + 200, + b"origin".to_vec(), + vec![("content-type", "text/html; charset=utf-8")], + ); + let services = build_services_with_http_client( + Arc::clone(&stub) as Arc + ); + let req = HttpRequest::builder() + .method(Method::GET) + .uri("https://publisher.example/page") + .header(header::HOST, "publisher.example") + .body(EdgeBody::empty()) + .expect("should build request"); + + let _ = run_publisher_proxy(&settings, &services, req).await; + + assert_eq!( + stub.recorded_stream_response_flags(), + vec![false], + "publisher origin fetch must not request streams when the platform does not support them" + ); + } + + #[tokio::test] + async fn publisher_origin_fetch_sets_stream_response_when_supported() { + let settings = create_test_settings(); + let stub = Arc::new(StubHttpClient::new()); + stub.set_streaming_responses_supported(true); + stub.push_response_with_headers( + 200, + b"origin".to_vec(), + vec![("content-type", "text/html; charset=utf-8")], + ); + let services = build_services_with_http_client( + Arc::clone(&stub) as Arc + ); + let req = HttpRequest::builder() + .method(Method::GET) + .uri("https://publisher.example/page") + .header(header::HOST, "publisher.example") + .body(EdgeBody::empty()) + .expect("should build request"); + + let _ = run_publisher_proxy(&settings, &services, req).await; + + assert_eq!( + stub.recorded_stream_response_flags(), + vec![true], + "publisher origin fetch should request streams when the platform supports them" + ); + } + #[tokio::test] async fn handle_publisher_request_does_not_self_generate_ec() { // EC generation is the adapter's real-browser-gated responsibility. This @@ -3659,6 +4605,913 @@ mod tests { ); } + #[test] + fn stream_publisher_body_rejects_stream_body_in_sync_path() { + let settings = create_test_settings(); + let registry = + IntegrationRegistry::new(&settings).expect("should create integration registry"); + let params = OwnedProcessResponseParams { + content_encoding: String::new(), + origin_host: "origin.example.com".to_string(), + origin_url: "https://origin.example.com".to_string(), + request_host: "proxy.example.com".to_string(), + request_scheme: "https".to_string(), + content_type: "text/html; charset=utf-8".to_string(), + ad_slots_script: None, + ad_bids_state: Arc::new(Mutex::new(None)), + auction_observation: None, + auction_request: None, + dispatched_auction: None, + price_granularity: crate::price_bucket::PriceGranularity::default(), + }; + let body = EdgeBody::from_stream(futures::stream::iter(vec![Ok::<_, io::Error>( + bytes::Bytes::from_static(b"live"), + )])); + let mut output = Vec::new(); + + let err = stream_publisher_body(body, &mut output, ¶ms, &settings, ®istry) + .expect_err("should reject stream body in sync path"); + + assert!( + format!("{err:?}").contains("streaming body"), + "should explain that Body::Stream is not supported by the sync path: {err:?}" + ); + } + + #[test] + fn body_chunk_source_yields_once_body_in_chunks() { + futures::executor::block_on(async { + let body = EdgeBody::from_bytes(bytes::Bytes::from_static(b"abcdef")); + let mut source = BodyChunkSource::new(body, 3).with_max_bytes(16); + + assert_eq!( + source.next_chunk().await.expect("should read").as_deref(), + Some(&b"abc"[..]), + "should yield the first chunk" + ); + assert_eq!( + source.next_chunk().await.expect("should read").as_deref(), + Some(&b"def"[..]), + "should yield the second chunk" + ); + assert!( + source.next_chunk().await.expect("should read").is_none(), + "should end after buffered bytes are exhausted" + ); + }); + } + + #[test] + fn body_chunk_source_preserves_stream_chunks() { + futures::executor::block_on(async { + let body = EdgeBody::stream(futures::stream::iter(vec![ + bytes::Bytes::from_static(b"first"), + bytes::Bytes::from_static(b"second"), + ])); + let mut source = BodyChunkSource::new(body, 3).with_max_bytes(16); + + assert_eq!( + source.next_chunk().await.expect("should read").as_deref(), + Some(&b"first"[..]), + "stream chunks should pass through without re-chunking" + ); + assert_eq!( + source.next_chunk().await.expect("should read").as_deref(), + Some(&b"second"[..]), + "stream chunks should preserve upstream boundaries" + ); + assert!( + source.next_chunk().await.expect("should read").is_none(), + "should end after stream is exhausted" + ); + }); + } + + #[test] + fn body_chunk_source_enforces_cumulative_raw_cap() { + futures::executor::block_on(async { + let body = EdgeBody::stream(futures::stream::iter(vec![ + bytes::Bytes::from_static(b"1234"), + bytes::Bytes::from_static(b"5678"), + ])); + let mut source = BodyChunkSource::new(body, STREAM_CHUNK_SIZE).with_max_bytes(6); + + assert!( + source + .next_chunk() + .await + .expect("first chunk should pass") + .is_some(), + "first chunk should stay under cap" + ); + let err = source + .next_chunk() + .await + .expect_err("second chunk should exceed cap"); + + assert!( + format!("{err:?}").contains("publisher origin body exceeded"), + "should report cumulative cap: {err:?}" + ); + }); + } + + #[test] + fn stream_publisher_body_async_processes_stream_without_auction() { + futures::executor::block_on(async { + let settings = create_test_settings(); + let registry = + IntegrationRegistry::new(&settings).expect("should create integration registry"); + let orchestrator = AuctionOrchestrator::new(settings.auction.clone()); + let services = noop_services(); + let mut params = OwnedProcessResponseParams { + content_encoding: String::new(), + origin_host: "origin.example.com".to_string(), + origin_url: "https://origin.example.com".to_string(), + request_host: "proxy.example.com".to_string(), + request_scheme: "https".to_string(), + content_type: "text/css".to_string(), + ad_slots_script: None, + ad_bids_state: Arc::new(Mutex::new(None)), + auction_observation: None, + auction_request: None, + dispatched_auction: None, + price_granularity: crate::price_bucket::PriceGranularity::default(), + }; + let body = EdgeBody::stream(futures::stream::iter(vec![ + bytes::Bytes::from_static(b"body{background:url('https://origin.example.com/"), + bytes::Bytes::from_static(b"asset.png')}"), + ])); + let mut output = Vec::new(); + + stream_publisher_body_async( + body, + &mut output, + &mut params, + &settings, + ®istry, + &orchestrator, + &services, + ) + .await + .expect("stream body should process on async path"); + + let css = String::from_utf8(output).expect("should be valid UTF-8"); + assert!( + css.contains("proxy.example.com"), + "should rewrite origin host while streaming. Got: {css}" + ); + assert!( + !css.contains("origin.example.com"), + "should not leave origin host after rewrite. Got: {css}" + ); + }); + } + + #[test] + fn stream_publisher_body_async_processes_gzip_stream_without_auction() { + futures::executor::block_on(async { + let settings = create_test_settings(); + let registry = + IntegrationRegistry::new(&settings).expect("should create integration registry"); + let orchestrator = AuctionOrchestrator::new(settings.auction.clone()); + let services = noop_services(); + let mut params = OwnedProcessResponseParams { + content_encoding: "gzip".to_string(), + origin_host: "origin.example.com".to_string(), + origin_url: "https://origin.example.com".to_string(), + request_host: "proxy.example.com".to_string(), + request_scheme: "https".to_string(), + content_type: "text/css".to_string(), + ad_slots_script: None, + ad_bids_state: Arc::new(Mutex::new(None)), + auction_observation: None, + auction_request: None, + dispatched_auction: None, + price_granularity: crate::price_bucket::PriceGranularity::default(), + }; + let compressed = + gzip_encode(b"body{background:url('https://origin.example.com/asset.png')}"); + let split_at = compressed.len() / 2; + let body = EdgeBody::stream(futures::stream::iter(vec![ + bytes::Bytes::copy_from_slice(&compressed[..split_at]), + bytes::Bytes::copy_from_slice(&compressed[split_at..]), + ])); + let mut output = Vec::new(); + + stream_publisher_body_async( + body, + &mut output, + &mut params, + &settings, + ®istry, + &orchestrator, + &services, + ) + .await + .expect("gzip stream body should process on async path"); + + let css = String::from_utf8(gzip_decode(&output)).expect("should be valid UTF-8"); + assert!( + css.contains("proxy.example.com"), + "should rewrite origin host while streaming gzip. Got: {css}" + ); + assert!( + !css.contains("origin.example.com"), + "should not leave origin host after gzip rewrite. Got: {css}" + ); + }); + } + + #[test] + fn stream_publisher_body_async_processes_deflate_stream_without_auction() { + futures::executor::block_on(async { + let settings = create_test_settings(); + let registry = + IntegrationRegistry::new(&settings).expect("should create integration registry"); + let orchestrator = AuctionOrchestrator::new(settings.auction.clone()); + let services = noop_services(); + let mut params = OwnedProcessResponseParams { + content_encoding: "deflate".to_string(), + origin_host: "origin.example.com".to_string(), + origin_url: "https://origin.example.com".to_string(), + request_host: "proxy.example.com".to_string(), + request_scheme: "https".to_string(), + content_type: "text/css".to_string(), + ad_slots_script: None, + ad_bids_state: Arc::new(Mutex::new(None)), + auction_observation: None, + auction_request: None, + dispatched_auction: None, + price_granularity: crate::price_bucket::PriceGranularity::default(), + }; + let compressed = + deflate_encode(b"body{background:url('https://origin.example.com/asset.png')}"); + let split_at = compressed.len() / 2; + let body = EdgeBody::stream(futures::stream::iter(vec![ + bytes::Bytes::copy_from_slice(&compressed[..split_at]), + bytes::Bytes::copy_from_slice(&compressed[split_at..]), + ])); + let mut output = Vec::new(); + + stream_publisher_body_async( + body, + &mut output, + &mut params, + &settings, + ®istry, + &orchestrator, + &services, + ) + .await + .expect("deflate stream body should process on async path"); + + let css = String::from_utf8(deflate_decode(&output)).expect("should be valid UTF-8"); + assert!( + css.contains("proxy.example.com"), + "should rewrite origin host while streaming deflate. Got: {css}" + ); + assert!( + !css.contains("origin.example.com"), + "should not leave origin host after deflate rewrite. Got: {css}" + ); + }); + } + + #[test] + fn stream_publisher_body_async_processes_brotli_stream_without_auction() { + futures::executor::block_on(async { + let settings = create_test_settings(); + let registry = + IntegrationRegistry::new(&settings).expect("should create integration registry"); + let orchestrator = AuctionOrchestrator::new(settings.auction.clone()); + let services = noop_services(); + let mut params = OwnedProcessResponseParams { + content_encoding: "br".to_string(), + origin_host: "origin.example.com".to_string(), + origin_url: "https://origin.example.com".to_string(), + request_host: "proxy.example.com".to_string(), + request_scheme: "https".to_string(), + content_type: "text/css".to_string(), + ad_slots_script: None, + ad_bids_state: Arc::new(Mutex::new(None)), + auction_observation: None, + auction_request: None, + dispatched_auction: None, + price_granularity: crate::price_bucket::PriceGranularity::default(), + }; + let compressed = + brotli_encode(b"body{background:url('https://origin.example.com/asset.png')}"); + let split_at = compressed.len() / 2; + let body = EdgeBody::stream(futures::stream::iter(vec![ + bytes::Bytes::copy_from_slice(&compressed[..split_at]), + bytes::Bytes::copy_from_slice(&compressed[split_at..]), + ])); + let mut output = Vec::new(); + + stream_publisher_body_async( + body, + &mut output, + &mut params, + &settings, + ®istry, + &orchestrator, + &services, + ) + .await + .expect("brotli stream body should process on async path"); + + let css = String::from_utf8(brotli_decode(&output)).expect("should be valid UTF-8"); + assert!( + css.contains("proxy.example.com"), + "should rewrite origin host while streaming brotli. Got: {css}" + ); + assert!( + !css.contains("origin.example.com"), + "should not leave origin host after brotli rewrite. Got: {css}" + ); + }); + } + + #[test] + fn stream_publisher_body_async_rejects_truncated_brotli_stream() { + futures::executor::block_on(async { + let settings = create_test_settings(); + let registry = + IntegrationRegistry::new(&settings).expect("should create integration registry"); + let orchestrator = AuctionOrchestrator::new(settings.auction.clone()); + let services = noop_services(); + let mut params = OwnedProcessResponseParams { + content_encoding: "br".to_string(), + origin_host: "origin.example.com".to_string(), + origin_url: "https://origin.example.com".to_string(), + request_host: "proxy.example.com".to_string(), + request_scheme: "https".to_string(), + content_type: "text/css".to_string(), + ad_slots_script: None, + ad_bids_state: Arc::new(Mutex::new(None)), + auction_observation: None, + auction_request: None, + dispatched_auction: None, + price_granularity: crate::price_bucket::PriceGranularity::default(), + }; + let compressed = + brotli_encode(b"body{background:url('https://origin.example.com/asset.png')}"); + let truncated = &compressed[..compressed.len() - 3]; + let body = + EdgeBody::stream(futures::stream::iter(vec![bytes::Bytes::copy_from_slice( + truncated, + )])); + let mut output = Vec::new(); + + let err = stream_publisher_body_async( + body, + &mut output, + &mut params, + &settings, + ®istry, + &orchestrator, + &services, + ) + .await + .expect_err("truncated brotli stream must fail instead of truncating silently"); + + assert!( + format!("{err:?}").contains("brotli"), + "should surface the brotli finalization failure: {err:?}" + ); + }); + } + + fn non_html_stream_params(content_encoding: &str) -> OwnedProcessResponseParams { + OwnedProcessResponseParams { + content_encoding: content_encoding.to_string(), + origin_host: "origin.example.com".to_string(), + origin_url: "https://origin.example.com".to_string(), + request_host: "proxy.example.com".to_string(), + request_scheme: "https".to_string(), + content_type: "text/css".to_string(), + ad_slots_script: None, + ad_bids_state: Arc::new(Mutex::new(None)), + auction_observation: None, + auction_request: None, + dispatched_auction: None, + price_granularity: crate::price_bucket::PriceGranularity::default(), + } + } + + #[test] + fn stream_publisher_body_async_rejects_truncated_gzip_stream() { + futures::executor::block_on(async { + let settings = create_test_settings(); + let registry = + IntegrationRegistry::new(&settings).expect("should create integration registry"); + let orchestrator = AuctionOrchestrator::new(settings.auction.clone()); + let services = noop_services(); + let mut params = non_html_stream_params("gzip"); + let compressed = + gzip_encode(b"body{background:url('https://origin.example.com/asset.png')}"); + let truncated = &compressed[..compressed.len() - 3]; + let body = + EdgeBody::stream(futures::stream::iter(vec![bytes::Bytes::copy_from_slice( + truncated, + )])); + let mut output = Vec::new(); + + let err = stream_publisher_body_async( + body, + &mut output, + &mut params, + &settings, + ®istry, + &orchestrator, + &services, + ) + .await + .expect_err("truncated gzip stream must fail instead of truncating silently"); + + assert!( + format!("{err:?}").contains("gzip"), + "should surface the gzip finalization failure: {err:?}" + ); + }); + } + + #[test] + fn stream_publisher_body_async_rejects_truncated_deflate_stream() { + futures::executor::block_on(async { + let settings = create_test_settings(); + let registry = + IntegrationRegistry::new(&settings).expect("should create integration registry"); + let orchestrator = AuctionOrchestrator::new(settings.auction.clone()); + let services = noop_services(); + let mut params = non_html_stream_params("deflate"); + let compressed = + deflate_encode(b"body{background:url('https://origin.example.com/asset.png')}"); + // Cut into the deflate data itself, not just the adler32 trailer. + let truncated = &compressed[..compressed.len() / 2]; + let body = + EdgeBody::stream(futures::stream::iter(vec![bytes::Bytes::copy_from_slice( + truncated, + )])); + let mut output = Vec::new(); + + let err = stream_publisher_body_async( + body, + &mut output, + &mut params, + &settings, + ®istry, + &orchestrator, + &services, + ) + .await + .expect_err("truncated deflate stream must fail instead of truncating silently"); + + assert!( + format!("{err:?}").contains("deflate"), + "should surface the deflate finalization failure: {err:?}" + ); + }); + } + + #[test] + fn stream_publisher_body_async_enforces_decoded_byte_cap() { + futures::executor::block_on(async { + let mut settings = create_test_settings(); + // Raw compressed input stays tiny (well under the cap); only the + // decoded expansion exceeds it — the decompression-bomb case the + // raw-byte cap alone cannot catch. + settings.publisher.max_buffered_body_bytes = 1024; + let registry = + IntegrationRegistry::new(&settings).expect("should create integration registry"); + let orchestrator = AuctionOrchestrator::new(settings.auction.clone()); + let services = noop_services(); + let mut params = non_html_stream_params("gzip"); + let compressed = gzip_encode(&vec![b'a'; 64 * 1024]); + assert!( + compressed.len() < 1024, + "test precondition: compressed input must stay under the raw cap" + ); + let body = + EdgeBody::stream(futures::stream::iter(vec![bytes::Bytes::from(compressed)])); + let mut output = Vec::new(); + + let err = stream_publisher_body_async( + body, + &mut output, + &mut params, + &settings, + ®istry, + &orchestrator, + &services, + ) + .await + .expect_err("decoded expansion past the cap must fail"); + + assert!( + format!("{err:?}").contains("decoded size exceeded"), + "should report the cumulative decoded cap: {err:?}" + ); + }); + } + + #[test] + fn body_chunk_source_resumes_after_cancelled_poll() { + futures::executor::block_on(async { + let mut pending_once = true; + let mut yielded = false; + let stream = futures::stream::poll_fn(move |cx| { + if pending_once { + pending_once = false; + cx.waker().wake_by_ref(); + return std::task::Poll::Pending; + } + if yielded { + return std::task::Poll::Ready(None); + } + yielded = true; + std::task::Poll::Ready(Some(Ok::<_, io::Error>(bytes::Bytes::from_static( + b"chunk", + )))) + }); + let body = EdgeBody::from_stream(stream); + let mut source = BodyChunkSource::new(body, STREAM_CHUNK_SIZE); + + { + // Poll the pull future once (Pending), then drop it — + // simulating a cancelled await (select/timeout wrapper). + let mut pull = Box::pin(source.next_chunk()); + let waker = futures::task::noop_waker(); + let mut context = std::task::Context::from_waker(&waker); + assert!( + pull.as_mut().poll(&mut context).is_pending(), + "first poll should be pending" + ); + } + + let chunk = source + .next_chunk() + .await + .expect("should read after cancelled poll"); + assert_eq!( + chunk.as_deref(), + Some(&b"chunk"[..]), + "cancelled pull must not lose the origin stream" + ); + }); + } + + #[test] + fn stream_publisher_body_async_processes_stream_with_auction_hold() { + futures::executor::block_on(async { + let settings = create_test_settings(); + let registry = + IntegrationRegistry::new(&settings).expect("should create integration registry"); + let orchestrator = AuctionOrchestrator::new(settings.auction.clone()); + let services = noop_services(); + let state = Arc::new(Mutex::new(None)); + let mut params = OwnedProcessResponseParams { + content_encoding: String::new(), + origin_host: "origin.example.com".to_string(), + origin_url: "https://origin.example.com".to_string(), + request_host: "proxy.example.com".to_string(), + request_scheme: "https".to_string(), + content_type: "text/html; charset=utf-8".to_string(), + ad_slots_script: Some( + r#""# + .to_string(), + ), + ad_bids_state: state, + auction_observation: None, + auction_request: Some(test_auction_request()), + dispatched_auction: Some(DispatchedAuction::empty_for_test( + test_auction_request(), + 10, + )), + price_granularity: crate::price_bucket::PriceGranularity::default(), + }; + let body = EdgeBody::stream(futures::stream::iter(vec![ + bytes::Bytes::from_static(b"hello"), + bytes::Bytes::from_static(b""), + ])); + let mut output = Vec::new(); + + stream_publisher_body_async( + body, + &mut output, + &mut params, + &settings, + ®istry, + &orchestrator, + &services, + ) + .await + .expect("stream body with auction should process on async path"); + + let html = String::from_utf8(output).expect("should be valid UTF-8"); + assert!( + html.contains("hello"), + "should preserve streamed HTML content. Got: {html}" + ); + assert!( + html.contains(".adSlots=JSON.parse"), + "should still inject ad slots. Got: {html}" + ); + assert!( + html.contains(".bids=JSON.parse"), + "should collect auction and inject bids before body close. Got: {html}" + ); + }); + } + + #[test] + fn stream_publisher_body_async_processes_non_html_stream_after_auction_collect() { + futures::executor::block_on(async { + let settings = create_test_settings(); + let registry = + IntegrationRegistry::new(&settings).expect("should create integration registry"); + let orchestrator = AuctionOrchestrator::new(settings.auction.clone()); + let services = noop_services(); + let mut params = OwnedProcessResponseParams { + content_encoding: String::new(), + origin_host: "origin.example.com".to_string(), + origin_url: "https://origin.example.com".to_string(), + request_host: "proxy.example.com".to_string(), + request_scheme: "https".to_string(), + content_type: "text/css".to_string(), + ad_slots_script: None, + ad_bids_state: Arc::new(Mutex::new(None)), + auction_observation: None, + auction_request: Some(test_auction_request()), + dispatched_auction: Some(DispatchedAuction::empty_for_test( + test_auction_request(), + 10, + )), + price_granularity: crate::price_bucket::PriceGranularity::default(), + }; + let body = EdgeBody::stream(futures::stream::iter(vec![bytes::Bytes::from_static( + b"body{background:url('https://origin.example.com/asset.png')}", + )])); + let mut output = Vec::new(); + + stream_publisher_body_async( + body, + &mut output, + &mut params, + &settings, + ®istry, + &orchestrator, + &services, + ) + .await + .expect("non-html stream body should process after auction collection"); + + let css = String::from_utf8(output).expect("should be valid UTF-8"); + assert!( + css.contains("proxy.example.com"), + "should rewrite non-html stream after auction collection. Got: {css}" + ); + assert!( + !css.contains("origin.example.com"), + "should not leave origin host after rewrite. Got: {css}" + ); + }); + } + + fn drain_streaming_finalize_body(content_encoding: &str, body: EdgeBody) -> Vec { + let settings = Arc::new(create_test_settings()); + let registry = Arc::new( + IntegrationRegistry::new(&settings).expect("should create integration registry"), + ); + let orchestrator = Arc::new(AuctionOrchestrator::new(settings.auction.clone())); + let services = noop_services(); + let response = Response::builder() + .status(StatusCode::OK) + .header(header::CONTENT_TYPE, "text/css") + .body(EdgeBody::empty()) + .expect("should build response"); + let params = OwnedProcessResponseParams { + content_encoding: content_encoding.to_string(), + origin_host: "origin.example.com".to_string(), + origin_url: "https://origin.example.com".to_string(), + request_host: "proxy.example.com".to_string(), + request_scheme: "https".to_string(), + content_type: "text/css".to_string(), + ad_slots_script: None, + ad_bids_state: Arc::new(Mutex::new(None)), + auction_observation: None, + auction_request: None, + dispatched_auction: None, + price_granularity: crate::price_bucket::PriceGranularity::default(), + }; + let publisher_response = PublisherResponse::Stream { + response, + body, + params: Box::new(params), + }; + + let response = futures::executor::block_on(publisher_response_into_streaming_response( + publisher_response, + &Method::GET, + Arc::clone(&settings), + registry.as_ref(), + orchestrator, + services, + )) + .expect("should build streaming response"); + + assert!( + matches!(response.body(), EdgeBody::Stream(_)), + "streaming finalize should keep a lazy Body::Stream" + ); + + futures::executor::block_on( + response + .into_body() + .into_bytes_bounded(settings.publisher.max_buffered_body_bytes), + ) + .expect("streaming body should drain") + .to_vec() + } + + #[test] + fn publisher_response_streaming_finalize_keeps_stream_body_lazy() { + let body_bytes = drain_streaming_finalize_body( + "", + EdgeBody::stream(futures::stream::iter(vec![bytes::Bytes::from_static( + b"body{background:url('https://origin.example.com/asset.png')}", + )])), + ); + let css = String::from_utf8(body_bytes).expect("should be valid UTF-8"); + assert!( + css.contains("proxy.example.com"), + "streaming response body should still run publisher rewriting. Got: {css}" + ); + assert!( + !css.contains("origin.example.com"), + "streaming response body should not leave origin URLs unrewritten. Got: {css}" + ); + } + + #[test] + fn publisher_response_streaming_finalize_processes_gzip_stream() { + let compressed = + gzip_encode(b"body{background:url('https://origin.example.com/asset.png')}"); + let split_at = compressed.len() / 2; + let output = drain_streaming_finalize_body( + "gzip", + EdgeBody::stream(futures::stream::iter(vec![ + bytes::Bytes::copy_from_slice(&compressed[..split_at]), + bytes::Bytes::copy_from_slice(&compressed[split_at..]), + ])), + ); + + let css = String::from_utf8(gzip_decode(&output)).expect("should be valid UTF-8"); + assert!( + css.contains("proxy.example.com"), + "streaming response finalize should rewrite gzip body. Got: {css}" + ); + assert!( + !css.contains("origin.example.com"), + "streaming response finalize should not leave gzip origin URLs. Got: {css}" + ); + } + + #[test] + fn publisher_response_streaming_finalize_processes_deflate_stream() { + let compressed = + deflate_encode(b"body{background:url('https://origin.example.com/asset.png')}"); + let split_at = compressed.len() / 2; + let output = drain_streaming_finalize_body( + "deflate", + EdgeBody::stream(futures::stream::iter(vec![ + bytes::Bytes::copy_from_slice(&compressed[..split_at]), + bytes::Bytes::copy_from_slice(&compressed[split_at..]), + ])), + ); + + let css = String::from_utf8(deflate_decode(&output)).expect("should be valid UTF-8"); + assert!( + css.contains("proxy.example.com"), + "streaming response finalize should rewrite deflate body. Got: {css}" + ); + assert!( + !css.contains("origin.example.com"), + "streaming response finalize should not leave deflate origin URLs. Got: {css}" + ); + } + + #[test] + fn publisher_response_streaming_finalize_processes_brotli_stream() { + let compressed = + brotli_encode(b"body{background:url('https://origin.example.com/asset.png')}"); + let split_at = compressed.len() / 2; + let output = drain_streaming_finalize_body( + "br", + EdgeBody::stream(futures::stream::iter(vec![ + bytes::Bytes::copy_from_slice(&compressed[..split_at]), + bytes::Bytes::copy_from_slice(&compressed[split_at..]), + ])), + ); + + let css = String::from_utf8(brotli_decode(&output)).expect("should be valid UTF-8"); + assert!( + css.contains("proxy.example.com"), + "streaming response finalize should rewrite brotli body. Got: {css}" + ); + assert!( + !css.contains("origin.example.com"), + "streaming response finalize should not leave brotli origin URLs. Got: {css}" + ); + } + + #[test] + fn publisher_response_streaming_finalize_holds_auction_and_keeps_gzip_tail() { + let settings = Arc::new(create_test_settings()); + let registry = Arc::new( + IntegrationRegistry::new(&settings).expect("should create integration registry"), + ); + let orchestrator = Arc::new(AuctionOrchestrator::new(settings.auction.clone())); + let services = noop_services(); + let response = Response::builder() + .status(StatusCode::OK) + .header(header::CONTENT_TYPE, "text/html; charset=utf-8") + .body(EdgeBody::empty()) + .expect("should build response"); + // The trailing content after `` must exceed the flate2 write + // decoder's 32 KiB internal output buffer: the close-body tag then + // surfaces (and releases the auction hold) mid-stream, while the + // trailing markup only surfaces at decoder finalization. This guards + // against the EOF decoded tail being dropped once the hold is gone. + let trailing_comment = format!("", "trailing-content ".repeat(3 * 1024)); + let page = format!("hello{trailing_comment}"); + let compressed = gzip_encode(page.as_bytes()); + let chunks: Vec = compressed + .chunks(STREAM_CHUNK_SIZE) + .map(bytes::Bytes::copy_from_slice) + .collect(); + let params = OwnedProcessResponseParams { + content_encoding: "gzip".to_string(), + origin_host: "origin.example.com".to_string(), + origin_url: "https://origin.example.com".to_string(), + request_host: "proxy.example.com".to_string(), + request_scheme: "https".to_string(), + content_type: "text/html; charset=utf-8".to_string(), + ad_slots_script: Some( + r#""# + .to_string(), + ), + ad_bids_state: Arc::new(Mutex::new(None)), + auction_observation: None, + auction_request: Some(test_auction_request()), + dispatched_auction: Some(DispatchedAuction::empty_for_test( + test_auction_request(), + 10, + )), + price_granularity: crate::price_bucket::PriceGranularity::default(), + }; + let publisher_response = PublisherResponse::Stream { + response, + body: EdgeBody::stream(futures::stream::iter(chunks)), + params: Box::new(params), + }; + + let response = futures::executor::block_on(publisher_response_into_streaming_response( + publisher_response, + &Method::GET, + Arc::clone(&settings), + registry.as_ref(), + orchestrator, + services, + )) + .expect("should build streaming response"); + + let output = futures::executor::block_on( + response + .into_body() + .into_bytes_bounded(settings.publisher.max_buffered_body_bytes), + ) + .expect("streaming body should drain") + .to_vec(); + + let html = String::from_utf8(gzip_decode(&output)).expect("should be valid UTF-8"); + assert!( + html.contains(".bids=JSON.parse"), + "should collect the held auction and inject bids. Got tail: {}", + &html[html.len().saturating_sub(200)..] + ); + assert!( + html.contains("trailing-content"), + "should preserve content after the close-body tag" + ); + assert!( + html.trim_end().ends_with(""), + "should not drop the decoded tail once the auction hold is released. Got tail: {}", + &html[html.len().saturating_sub(200)..] + ); + } + #[test] fn stream_publisher_body_treats_mixed_case_html_as_html() { let settings = create_test_settings(); diff --git a/crates/trusted-server-core/src/settings.rs b/crates/trusted-server-core/src/settings.rs index 9cbb2a54..0a6178a0 100644 --- a/crates/trusted-server-core/src/settings.rs +++ b/crates/trusted-server-core/src/settings.rs @@ -50,15 +50,15 @@ pub struct Publisher { /// exceeding it fails the response rather than allocating past the cap. /// Defaults to 16 MiB — a conservative cap that prevents Wasm-heap OOM. /// - /// On Fastly the *effective* ceiling for a publisher page is lower: the - /// platform HTTP client rejects any origin response whose raw (still - /// compressed) body exceeds 10 MiB before this buffer is ever filled, so - /// raising this value only helps highly compressible pages whose decoded - /// size exceeds the 16 MiB default while their compressed origin body stays - /// under 10 MiB. Raising it above ~10 MiB does not lift the platform cap for - /// uncompressed pages. That platform limit is removed once true streaming - /// lands (tracked for PR 15, issue #495), after which this setting becomes - /// the sole ceiling. + /// Fastly origin bodies are preserved as streams on the publisher path, so + /// this setting also caps the streaming pipeline twice over: cumulative + /// raw (still compressed) bytes pulled from origin, and cumulative decoded + /// bytes emitted by the decompressor — the latter so a decompression bomb + /// cannot push an unbounded decoded volume through the rewrite pipeline. + /// Buffered adapters keep using it as the post-rewrite output buffer cap. + /// On the streaming path headers are already committed when either cap + /// trips, so the response is truncated mid-body (with the error logged) + /// rather than replaced with a 5xx. /// /// Must be at least 1: a zero-byte cap fails every non-empty buffered /// publisher response at request time, so it is rejected at config diff --git a/crates/trusted-server-core/src/streaming_processor.rs b/crates/trusted-server-core/src/streaming_processor.rs index 5692118a..963c69da 100644 --- a/crates/trusted-server-core/src/streaming_processor.rs +++ b/crates/trusted-server-core/src/streaming_processor.rs @@ -349,11 +349,361 @@ impl StreamProcessor for StreamingReplacer { } } +/// Read buffer size for streaming body processing and brotli internal buffers. +/// Both the `Decompressor` and `CompressorWriter` use this value so all +/// brotli I/O layers operate on consistently-sized chunks. +pub(crate) const STREAM_CHUNK_SIZE: usize = 8192; + +/// Incremental push-style decompressor for the async chunk pipeline. +/// +/// Compressed bytes go in via [`Self::decode_chunk`]; decoded bytes drain +/// out of the internal buffer after every push. Write-based decoders are +/// used because the async publisher path cannot wrap a blocking `Read`. +/// +/// Decoded output is capped cumulatively: the chunk source only bounds raw +/// (still compressed) bytes, and a decompression bomb can expand ~1000x past +/// that, so the decoder enforces its own ceiling on the total bytes it emits. +/// +/// Every codec validates end-of-stream at [`Self::finish`] so a truncated +/// origin body errors instead of silently truncating the page: gzip via its +/// trailer checksum, brotli via `close()`, and deflate via an explicit +/// [`flate2::Status::StreamEnd`] check (`write::ZlibDecoder` accepts +/// truncated input silently, so the deflate arm drives [`flate2::Decompress`] +/// directly). +pub(crate) struct BodyStreamDecoder { + codec: BodyStreamDecoderCodec, + decoded_bytes: usize, + max_decoded_bytes: usize, +} + +enum BodyStreamDecoderCodec { + None, + Gzip(flate2::write::GzDecoder>), + Deflate(DeflateStreamDecoder), + Brotli(Box>>), +} + +/// Streaming zlib decoder that tracks whether the stream reached its end +/// marker, so truncated deflate bodies fail at finalization. +struct DeflateStreamDecoder { + decompress: flate2::Decompress, + stream_ended: bool, +} + +impl DeflateStreamDecoder { + fn new() -> Self { + Self { + decompress: flate2::Decompress::new(true), + stream_ended: false, + } + } + + fn decode(&mut self, chunk: &[u8]) -> Result, Report> { + let mut output = Vec::with_capacity(STREAM_CHUNK_SIZE); + let mut offset = 0usize; + // Trailing bytes after the zlib end marker are ignored, matching the + // read-based decoder used by the buffered pipeline. + while offset < chunk.len() && !self.stream_ended { + if output.len() == output.capacity() { + output.reserve(STREAM_CHUNK_SIZE); + } + let before_in = self.decompress.total_in(); + let before_out = self.decompress.total_out(); + let status = self + .decompress + .decompress_vec(&chunk[offset..], &mut output, flate2::FlushDecompress::None) + .change_context(TrustedServerError::Proxy { + message: "Failed to decode deflate publisher body chunk".to_string(), + })?; + let consumed = (self.decompress.total_in() - before_in) as usize; + let produced = (self.decompress.total_out() - before_out) as usize; + offset += consumed; + match status { + flate2::Status::StreamEnd => self.stream_ended = true, + flate2::Status::Ok | flate2::Status::BufError => { + if consumed == 0 && produced == 0 && output.len() < output.capacity() { + return Err(Report::new(TrustedServerError::Proxy { + message: "deflate publisher body decoder made no progress".to_string(), + })); + } + } + } + } + Ok(output) + } +} + +impl BodyStreamDecoder { + pub(crate) fn new(compression: Compression, max_decoded_bytes: usize) -> Self { + let codec = match compression { + Compression::None => BodyStreamDecoderCodec::None, + Compression::Gzip => { + BodyStreamDecoderCodec::Gzip(flate2::write::GzDecoder::new(Vec::new())) + } + Compression::Deflate => BodyStreamDecoderCodec::Deflate(DeflateStreamDecoder::new()), + Compression::Brotli => BodyStreamDecoderCodec::Brotli(Box::new( + brotli::DecompressorWriter::new(Vec::new(), STREAM_CHUNK_SIZE), + )), + }; + Self { + codec, + decoded_bytes: 0, + max_decoded_bytes, + } + } + + pub(crate) fn decode_chunk( + &mut self, + chunk: bytes::Bytes, + ) -> Result> { + let decoded = match &mut self.codec { + BodyStreamDecoderCodec::None => chunk, + BodyStreamDecoderCodec::Gzip(decoder) => { + decoder + .write_all(&chunk) + .change_context(TrustedServerError::Proxy { + message: "Failed to decode gzip publisher body chunk".to_string(), + })?; + bytes::Bytes::from(std::mem::take(decoder.get_mut())) + } + BodyStreamDecoderCodec::Deflate(decoder) => bytes::Bytes::from(decoder.decode(&chunk)?), + BodyStreamDecoderCodec::Brotli(decoder) => { + decoder + .write_all(&chunk) + .change_context(TrustedServerError::Proxy { + message: "Failed to decode brotli publisher body chunk".to_string(), + })?; + bytes::Bytes::from(std::mem::take(decoder.get_mut())) + } + }; + self.track_decoded(decoded.len())?; + Ok(decoded) + } + + pub(crate) fn finish(&mut self) -> Result, Report> { + let tail = match &mut self.codec { + BodyStreamDecoderCodec::None => Vec::new(), + BodyStreamDecoderCodec::Gzip(decoder) => { + decoder + .try_finish() + .change_context(TrustedServerError::Proxy { + message: "Failed to finalize gzip publisher body decoder".to_string(), + })?; + std::mem::take(decoder.get_mut()) + } + BodyStreamDecoderCodec::Deflate(decoder) => { + if !decoder.stream_ended { + return Err(Report::new(TrustedServerError::Proxy { + message: + "Failed to finalize deflate publisher body decoder: truncated stream" + .to_string(), + })); + } + Vec::new() + } + BodyStreamDecoderCodec::Brotli(decoder) => { + // `close()` (not `flush()`): flush accepts a truncated brotli + // stream silently, while close validates end-of-stream and + // errors on incomplete input, matching the gzip/deflate arms. + decoder.close().change_context(TrustedServerError::Proxy { + message: "Failed to finalize brotli publisher body decoder".to_string(), + })?; + std::mem::take(decoder.get_mut()) + } + }; + self.track_decoded(tail.len())?; + Ok(tail) + } + + fn track_decoded(&mut self, len: usize) -> Result<(), Report> { + self.decoded_bytes = self.decoded_bytes.checked_add(len).ok_or_else(|| { + Report::new(TrustedServerError::Proxy { + message: "publisher origin body decoded byte count overflowed".to_string(), + }) + })?; + if self.decoded_bytes > self.max_decoded_bytes { + return Err(Report::new(TrustedServerError::Proxy { + message: format!( + "publisher origin body decoded size exceeded {}-byte streaming limit", + self.max_decoded_bytes + ), + })); + } + Ok(()) + } +} + +/// Incremental push-style compressor mirroring [`BodyStreamDecoder`]. +/// +/// Processed bytes go in via [`Self::encode_chunk`]; encoded bytes drain out +/// after every push, and [`Self::finish`] emits the stream trailer. +pub(crate) enum BodyStreamEncoder { + None, + Gzip(flate2::write::GzEncoder>), + Deflate(flate2::write::ZlibEncoder>), + Brotli(Box>>), +} + +fn new_brotli_vec_encoder() -> brotli::enc::writer::CompressorWriter> { + let params = brotli::enc::BrotliEncoderParams { + quality: 4, + lgwin: 22, + ..Default::default() + }; + brotli::enc::writer::CompressorWriter::with_params(Vec::new(), STREAM_CHUNK_SIZE, ¶ms) +} + +impl BodyStreamEncoder { + pub(crate) fn new(compression: Compression) -> Self { + match compression { + Compression::None => Self::None, + Compression::Gzip => Self::Gzip(flate2::write::GzEncoder::new( + Vec::new(), + flate2::Compression::default(), + )), + Compression::Deflate => Self::Deflate(flate2::write::ZlibEncoder::new( + Vec::new(), + flate2::Compression::default(), + )), + Compression::Brotli => Self::Brotli(Box::new(new_brotli_vec_encoder())), + } + } + + pub(crate) fn encode_chunk( + &mut self, + chunk: Vec, + ) -> Result, Report> { + match self { + // Identity encoding passes the processed chunk through untouched. + Self::None => Ok(chunk), + Self::Gzip(encoder) => { + encoder + .write_all(&chunk) + .change_context(TrustedServerError::Proxy { + message: "Failed to encode gzip publisher body chunk".to_string(), + })?; + Ok(std::mem::take(encoder.get_mut())) + } + Self::Deflate(encoder) => { + encoder + .write_all(&chunk) + .change_context(TrustedServerError::Proxy { + message: "Failed to encode deflate publisher body chunk".to_string(), + })?; + Ok(std::mem::take(encoder.get_mut())) + } + Self::Brotli(encoder) => { + encoder + .write_all(&chunk) + .change_context(TrustedServerError::Proxy { + message: "Failed to encode brotli publisher body chunk".to_string(), + })?; + Ok(std::mem::take(encoder.get_mut())) + } + } + } + + /// Emits the encoder trailer. Consumes the codec state (the encoder + /// becomes identity afterwards); terminal — call once at end of stream. + pub(crate) fn finish(&mut self) -> Result, Report> { + match std::mem::replace(self, Self::None) { + Self::None => Ok(Vec::new()), + Self::Gzip(encoder) => encoder.finish().change_context(TrustedServerError::Proxy { + message: "Failed to finalize gzip publisher body encoder".to_string(), + }), + Self::Deflate(encoder) => encoder.finish().change_context(TrustedServerError::Proxy { + message: "Failed to finalize deflate publisher body encoder".to_string(), + }), + Self::Brotli(encoder) => Ok((*encoder).into_inner()), + } + } +} + #[cfg(test)] mod tests { use super::*; use crate::streaming_replacer::{Replacement, StreamingReplacer}; + #[test] + fn body_stream_decoder_enforces_cumulative_decoded_cap() { + let compressed = { + let mut encoder = + flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default()); + encoder + .write_all(&vec![b'a'; 64 * 1024]) + .expect("should write gzip test input"); + encoder.finish().expect("should finish gzip encoding") + }; + assert!( + compressed.len() < 1024, + "test precondition: compressed input must stay small" + ); + let mut decoder = BodyStreamDecoder::new(Compression::Gzip, 1024); + + let err = decoder + .decode_chunk(bytes::Bytes::from(compressed)) + .expect_err("decoded expansion past the cap must fail"); + + assert!( + format!("{err:?}").contains("decoded size exceeded"), + "should report the cumulative decoded cap: {err:?}" + ); + } + + #[test] + fn body_stream_decoder_rejects_truncated_deflate_stream() { + let compressed = { + let mut encoder = + flate2::write::ZlibEncoder::new(Vec::new(), flate2::Compression::default()); + encoder + .write_all(b"deflate payload that spans more than one deflate block boundary") + .expect("should write deflate test input"); + encoder.finish().expect("should finish deflate encoding") + }; + let truncated = &compressed[..compressed.len() / 2]; + let mut decoder = BodyStreamDecoder::new(Compression::Deflate, usize::MAX); + decoder + .decode_chunk(bytes::Bytes::copy_from_slice(truncated)) + .expect("partial deflate input should decode incrementally"); + + let err = decoder + .finish() + .expect_err("truncated deflate stream must fail at finalization"); + + assert!( + format!("{err:?}").contains("truncated stream"), + "should report the missing deflate end marker: {err:?}" + ); + } + + #[test] + fn body_stream_decoder_ignores_deflate_trailing_bytes() { + let compressed = { + let mut encoder = + flate2::write::ZlibEncoder::new(Vec::new(), flate2::Compression::default()); + encoder + .write_all(b"deflate payload") + .expect("should write deflate test input"); + encoder.finish().expect("should finish deflate encoding") + }; + let mut with_trailing = compressed; + with_trailing.extend_from_slice(b"junk"); + let mut decoder = BodyStreamDecoder::new(Compression::Deflate, usize::MAX); + + let decoded = decoder + .decode_chunk(bytes::Bytes::from(with_trailing)) + .expect("complete deflate stream should decode"); + decoder + .finish() + .expect("trailing bytes after the end marker should be ignored"); + + assert_eq!( + decoded.as_ref(), + b"deflate payload", + "should decode the payload and drop trailing junk" + ); + } + /// Verify that `lol_html` fragments text nodes when input chunks split /// mid-text-node. Script rewriters must be fragment-safe — they accumulate /// text fragments internally until `is_last_in_text_node` is true. diff --git a/docs/superpowers/plans/2026-07-08-true-origin-streaming-fastly.md b/docs/superpowers/plans/2026-07-08-true-origin-streaming-fastly.md new file mode 100644 index 00000000..bc1983a9 --- /dev/null +++ b/docs/superpowers/plans/2026-07-08-true-origin-streaming-fastly.md @@ -0,0 +1,1039 @@ +# True Origin Streaming Fastly Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Fix issue #849 for the production Fastly path so publisher HTML origin bodies stream through the rewrite pipeline to the client, with auction collection outside TTFB except for the held `` tail. + +**Architecture:** Keep one cohesive Fastly PR because the core pipeline, Fastly origin fetch, and Fastly finalize path are incomplete in isolation. Convert publisher body processing from sync `Read` over buffered bodies to async chunk-pull over `edgezero_core::body::Body`, then enable Fastly `with_stream_response()` and return a lazy streaming body for publisher responses. Leave Cloudflare, Spin, and Axum streaming as follow-up work. + +**Tech Stack:** Rust 2024, `edgezero_core::body::Body`, `futures::StreamExt`, `error-stack`, `flate2`, `brotli`, Fastly Compute, Viceroy tests. + +--- + +## Scope + +In scope: + +- Publisher HTML and processable publisher responses on the Fastly adapter. +- Core publisher pipeline support for `Body::Stream`. +- Fastly platform capability signaling for streaming origin responses. +- Fastly EdgeZero response delivery that streams publisher bodies to clients. +- Tests proving stream-vs-buffer parity, bodiless handling, stream caps, and Fastly routing behavior. + +Out of scope: + +- Cloudflare origin streaming. Current adapter rejects `PlatformHttpRequest::stream_response`. +- Spin streaming. Current adapter and upstream EdgeZero Spin conversion are buffered/blocking issues. +- Axum client streaming. Axum is dev-only and has `LocalBoxStream`/`Send` constraints. +- Parser-context `` scan fix from issue #850. +- Origin template caching and transformed HTML caching from issue #852. + +## Current Failure Points + +- `crates/trusted-server-adapter-fastly/src/platform.rs`: `fastly_response_to_platform(..., stream_response: false)` uses `take_body_bytes()` and the 10 MiB platform cap for publisher origin responses. +- `crates/trusted-server-core/src/publisher.rs`: `body_as_reader()` calls `body.into_bytes().unwrap_or_default()`, so `Body::Stream` becomes an empty body. +- `crates/trusted-server-core/src/publisher.rs`: `stream_html_with_auction_hold()` and `body_close_hold_loop()` are sync-`Read` based. +- `crates/trusted-server-adapter-fastly/src/app.rs`: publisher route calls `buffer_publisher_response_async()`, buffering all processed output and awaiting auction before any client bytes are sent. +- `crates/trusted-server-adapter-fastly/src/main.rs`: `send_edgezero_response()` already streams `EdgeBody::Stream`, but currently only asset responses reach that arm. + +## File Structure + +- Modify `crates/trusted-server-core/src/platform/http.rs` + - Add `PlatformHttpClient::supports_streaming_responses()` with default `false`. +- Modify adapter platform implementations: + - `crates/trusted-server-adapter-fastly/src/platform.rs`: return `true` for `supports_streaming_responses()`. + - `crates/trusted-server-adapter-cloudflare/src/platform.rs`: inherit default `false`. + - `crates/trusted-server-adapter-spin/src/platform.rs`: inherit default `false`. + - `crates/trusted-server-adapter-axum/src/platform.rs`: inherit default `false`. + - `crates/trusted-server-core/src/platform/test_support.rs`: configurable test support if needed. +- Modify `crates/trusted-server-core/src/streaming_processor.rs` + - Add small push decoder/encoder helpers only if keeping them here reduces duplication. + - Keep the existing `StreamingPipeline::process(Read, Write)` API for existing call sites. +- Modify `crates/trusted-server-core/src/publisher.rs` + - Replace publisher async processing internals with async chunk-pull. + - Keep public `buffer_publisher_response_async()` for buffered adapters. + - Add a streaming response constructor/helper for Fastly to use. + - Make `body_as_reader()` reject `Body::Stream` loudly or remove its use from any stream-capable path. +- Modify `crates/trusted-server-adapter-fastly/src/app.rs` + - Replace publisher `buffer_publisher_response_async()` call with streaming finalize for streamable publisher responses. + - Preserve buffered behavior for `PublisherResponse::Buffered`, pass-through/bodiless responses, and error paths. +- Modify `crates/trusted-server-adapter-fastly/src/main.rs` + - Reuse existing `EdgeBody::Stream` delivery. + - If publisher streaming needs a different log message from asset streaming, split the helper name/log text without changing behavior. +- Tests: + - `crates/trusted-server-core/src/publisher.rs` unit tests. + - `crates/trusted-server-core/src/streaming_processor.rs` unit tests if push codec helpers are introduced there. + - `crates/trusted-server-core/src/platform/test_support.rs` tests for capability behavior. + - `crates/trusted-server-adapter-fastly/src/app.rs` route tests for publisher streaming response shape. + +## Design Decisions + +- Use one PR for the full Fastly production fix. Intermediate merged PRs would create incomplete behavior and review confusion. +- Use existing `publisher.max_buffered_body_bytes` as the publisher body ceiling after streaming. `settings.rs` already documents that this becomes the sole ceiling after true streaming removes the 10 MiB Fastly materialization cap. +- Keep `Content-Length` removed for rewritten stream responses. Streaming output can change size due to URL rewriting and bid injection. +- Preserve bodiless behavior for `HEAD`, `204`, and `304`: do not attach or drive a body stream, and log abandoned/wasted auctions as current code does. +- Do not build a sync `Read` bridge over `Body::Stream`; nested `block_on` can panic on Fastly because the router already runs under `futures::executor::block_on`. +- Avoid adding `async-stream` initially. Use `futures::stream::unfold` or a custom stream type so the PR does not add a dependency unless the implementation becomes materially clearer. + +## Task 1: Baseline Tests for Stream Input Safety + +**Files:** + +- Modify: `crates/trusted-server-core/src/publisher.rs` + +- [ ] **Step 1: Add a failing test for `Body::Stream` not becoming empty** + +Add a test near the existing `stream_publisher_body` tests: + +```rust +#[test] +fn stream_publisher_body_rejects_stream_body_in_sync_path() { + let settings = create_test_settings(); + let registry = IntegrationRegistry::new(&settings).expect("should build registry"); + let body = EdgeBody::from_stream(futures::stream::iter(vec![Ok(Bytes::from_static( + b"live", + ))])); + let params = test_process_params("text/html", ""); + let mut output = Vec::new(); + + let err = stream_publisher_body(body, &mut output, ¶ms, &settings, ®istry) + .expect_err("should reject stream body in sync path"); + + assert!( + format!("{err:?}").contains("streaming body"), + "should explain that Body::Stream is not supported by the sync path: {err:?}" + ); +} +``` + +- [ ] **Step 2: Run the targeted test and verify it fails** + +Run: + +```bash +cargo test-axum stream_publisher_body_rejects_stream_body_in_sync_path +``` + +Expected: FAIL because current `body_as_reader()` silently returns empty bytes. + +- [ ] **Step 3: Replace `body_as_reader()` with a fallible helper** + +Change `body_as_reader(body: EdgeBody) -> Cursor` to return `Result, Report>` and return a proxy error for `Body::Stream`. + +Minimal shape: + +```rust +fn body_as_reader(body: EdgeBody) -> Result, Report> { + let bytes = body.into_bytes().ok_or_else(|| { + Report::new(TrustedServerError::Proxy { + message: "streaming body cannot be processed by sync publisher pipeline".to_owned(), + }) + })?; + Ok(std::io::Cursor::new(bytes)) +} +``` + +Update existing sync call sites to use `body_as_reader(body)?`. + +- [ ] **Step 4: Run the targeted test and existing publisher sync tests** + +Run: + +```bash +cargo test-axum stream_publisher_body +``` + +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +git add crates/trusted-server-core/src/publisher.rs +git commit -m "Reject publisher stream bodies on sync path" +``` + +## Task 2: Async Chunk Source and Cumulative Cap + +**Files:** + +- Modify: `crates/trusted-server-core/src/publisher.rs` + +- [ ] **Step 1: Add tests for async chunk pulling** + +Add focused tests for a private helper that will pull chunks from both body variants: + +```rust +#[test] +fn body_chunk_source_yields_once_body_in_chunks() { + futures::executor::block_on(async { + let body = EdgeBody::from(Bytes::from_static(b"abcdef")); + let mut source = BodyChunkSource::new(body, 3); + + assert_eq!(source.next_chunk().await.expect("should read").as_deref(), Some(&b"abc"[..])); + assert_eq!(source.next_chunk().await.expect("should read").as_deref(), Some(&b"def"[..])); + assert!(source.next_chunk().await.expect("should read").is_none()); + }); +} +``` + +Add a separate test for `Body::Stream` preserving chunk boundaries and surfacing stream errors. + +- [ ] **Step 2: Add a failing test for the cumulative cap** + +Use a stream with two chunks whose total exceeds a small cap: + +```rust +#[test] +fn body_chunk_source_enforces_cumulative_raw_cap() { + futures::executor::block_on(async { + let body = EdgeBody::from_stream(futures::stream::iter(vec![ + Ok(Bytes::from_static(b"1234")), + Ok(Bytes::from_static(b"5678")), + ])); + let mut source = BodyChunkSource::new(body, STREAM_CHUNK_SIZE).with_max_bytes(6); + + assert!(source.next_chunk().await.expect("first chunk should pass").is_some()); + let err = source.next_chunk().await.expect_err("second chunk should exceed cap"); + assert!( + format!("{err:?}").contains("publisher origin body exceeded"), + "should report cumulative cap: {err:?}" + ); + }); +} +``` + +- [ ] **Step 3: Run the new tests and verify they fail** + +Run: + +```bash +cargo test-axum body_chunk_source +``` + +Expected: FAIL because helper does not exist. + +- [ ] **Step 4: Implement `BodyChunkSource`** + +Implement a private helper near `STREAM_CHUNK_SIZE`: + +- Owns `EdgeBody`. +- For `Body::Once`, yields `Bytes` slices up to `chunk_size` without copying more than necessary. +- For `Body::Stream`, awaits `stream.next()`. +- Tracks cumulative raw bytes and errors when total exceeds `max_bytes`. +- Maps stream errors to `TrustedServerError::Proxy`. + +Do not use `block_on` inside the helper. + +- [ ] **Step 5: Run helper tests** + +Run: + +```bash +cargo test-axum body_chunk_source +``` + +Expected: PASS. + +- [ ] **Step 6: Commit** + +```bash +git add crates/trusted-server-core/src/publisher.rs +git commit -m "Add async publisher body chunk source" +``` + +## Task 3: Push Compression Helpers + +**Files:** + +- Modify: `crates/trusted-server-core/src/streaming_processor.rs` +- Or modify: `crates/trusted-server-core/src/publisher.rs` if helpers are publisher-only + +- [ ] **Step 1: Add parity tests for compressed chunk processing** + +For each compression mode used by publisher HTML (`gzip`, `deflate`, `br`), add a test that feeds compressed HTML in multiple raw chunks through the future async path and verifies the decompressed/processed/recompressed output decodes to expected HTML. + +Start with gzip: + +```rust +#[test] +fn async_publisher_pipeline_preserves_gzip_html_across_stream_chunks() { + futures::executor::block_on(async { + let compressed = gzip_bytes(b"Hello"); + let body = EdgeBody::from_stream(bytes_to_two_chunk_stream(compressed)); + let output = process_test_body_async(body, "text/html", "gzip") + .await + .expect("should process gzip stream"); + + assert_eq!( + gunzip_bytes(&output), + b"Hello" + ); + }); +} +``` + +Use existing HTML processor expectations rather than inventing new behavior. + +- [ ] **Step 2: Run the gzip test and verify it fails** + +Run: + +```bash +cargo test-axum async_publisher_pipeline_preserves_gzip_html_across_stream_chunks +``` + +Expected: FAIL because async compressed processing does not exist. + +- [ ] **Step 3: Implement write-based push decoders** + +Use write-based APIs: + +- `flate2::write::GzDecoder` +- `flate2::write::ZlibDecoder` +- `brotli::DecompressorWriter` + +The helper should: + +- Accept raw compressed chunks. +- Write decoded bytes into an internal `Vec` sink. +- Return newly decoded bytes after each input chunk. +- Finalize at EOF and return any decoder tail bytes. +- Surface decoder errors as `TrustedServerError::Proxy`. + +Keep this helper private unless tests or other modules need it. + +- [ ] **Step 4: Implement output encoding wrapper** + +Continue to use existing write-based encoders: + +- `flate2::write::GzEncoder` +- `flate2::write::ZlibEncoder` +- `brotli::enc::writer::CompressorWriter` + +The async loop should write processed decoded chunks into the encoder and finalize once. + +- [ ] **Step 5: Add deflate and brotli tests** + +Run: + +```bash +cargo test-axum async_publisher_pipeline_preserves_ +``` + +Expected: gzip, deflate, and brotli async parity tests PASS. + +- [ ] **Step 6: Commit** + +```bash +git add crates/trusted-server-core/src/streaming_processor.rs crates/trusted-server-core/src/publisher.rs +git commit -m "Add push compression support for publisher streams" +``` + +## Task 4: Async Publisher Pipeline Without Auction + +**Files:** + +- Modify: `crates/trusted-server-core/src/publisher.rs` + +- [ ] **Step 1: Add stream-vs-once parity tests for no-auction paths** + +Cover: + +- HTML rewrite. +- RSC flight rewrite. +- Generic URL replacement. +- Unsupported stream body cannot reach sync path. + +Example: + +```rust +#[test] +fn stream_publisher_body_async_matches_buffered_html_without_auction() { + futures::executor::block_on(async { + let settings = create_test_settings(); + let registry = IntegrationRegistry::new(&settings).expect("should build registry"); + let html = Bytes::from_static(b"x"); + + let mut once_params = test_process_params("text/html", ""); + let mut once_output = Vec::new(); + stream_publisher_body_async( + EdgeBody::from(html.clone()), + &mut once_output, + &mut once_params, + &settings, + ®istry, + &AuctionOrchestrator::new(settings.auction.clone()), + &noop_services(), + ) + .await + .expect("once body should process"); + + let mut stream_params = test_process_params("text/html", ""); + let mut stream_output = Vec::new(); + stream_publisher_body_async( + EdgeBody::from_stream(futures::stream::iter(vec![Ok(html)])), + &mut stream_output, + &mut stream_params, + &settings, + ®istry, + &AuctionOrchestrator::new(settings.auction.clone()), + &noop_services(), + ) + .await + .expect("stream body should process"); + + assert_eq!(stream_output, once_output); + }); +} +``` + +- [ ] **Step 2: Run parity tests and verify they fail** + +Run: + +```bash +cargo test-axum stream_publisher_body_async_matches_buffered +``` + +Expected: FAIL because no-auction async path still delegates to sync processing. + +- [ ] **Step 3: Refactor `process_response_streaming` into reusable processor construction** + +Extract the shared routing logic into a helper such as: + +```rust +enum PublisherProcessor { + Html(HtmlRewriterAdapter), + Rsc(RscFlightUrlRewriter), + Url(StreamingReplacer), +} +``` + +Or use a generic closure/helper if that fits existing patterns better. The goal is to avoid duplicating content-type routing between sync and async paths. + +- [ ] **Step 4: Drive all `stream_publisher_body_async()` calls through async chunk-pull** + +Even when `params.dispatched_auction` is `None`, build the same processor and use `BodyChunkSource`. This prevents stream bodies from falling into the sync path. + +- [ ] **Step 5: Keep `stream_publisher_body()` for compatibility** + +The sync function should remain for old tests and any current non-stream callers, but it must not be used by the async path once this task is complete. + +- [ ] **Step 6: Run targeted tests** + +Run: + +```bash +cargo test-axum stream_publisher_body_async_matches_buffered +``` + +Expected: PASS. + +- [ ] **Step 7: Commit** + +```bash +git add crates/trusted-server-core/src/publisher.rs +git commit -m "Drive publisher async processing from body chunks" +``` + +## Task 5: Async Auction Hold Loop + +**Files:** + +- Modify: `crates/trusted-server-core/src/publisher.rs` + +- [ ] **Step 1: Replace reader-based hold-loop test with stream-based test** + +Update or add a test based on `body_close_hold_loop_processes_close_tail_before_reading_post_body_chunks()`: + +- Feed pre-`` chunk. +- Feed held `` chunk. +- Feed post-body chunk. +- Assert the loop collects auction immediately when `` test** + +Verify that auction collection happens at EOF and finalization still calls `processor.process_chunk(&[], true)`. + +- [ ] **Step 3: Add stream error abandonment test** + +Feed a stream error after dispatch and assert telemetry abandonment uses `stream_read_error` or the current expected reason. + +- [ ] **Step 4: Run tests and verify failures** + +Run: + +```bash +cargo test-axum body_close_hold_loop +``` + +Expected: FAIL until the loop consumes `BodyChunkSource`. + +- [ ] **Step 5: Change `body_close_hold_loop` to async chunk-pull** + +Replace: + +```rust +async fn body_close_hold_loop(...) +``` + +with a shape that accepts decoded chunks from an async driver, or accepts `BodyChunkSource` plus codec state. Keep the control flow: + +- Push decoded chunks into `BodyCloseHoldBuffer`. +- Write ready bytes immediately. +- On first ` bool { + false +} +``` + +In `FastlyPlatformHttpClient`: + +```rust +fn supports_streaming_responses(&self) -> bool { + true +} +``` + +In `StubHttpClient`, add a configurable flag if tests need both states. + +- [ ] **Step 4: Enable publisher origin streaming behind the gate** + +At the publisher origin fetch: + +```rust +let mut platform_request = PlatformHttpRequest::new(req, backend_name); +if services.http_client().supports_streaming_responses() { + platform_request = platform_request.with_stream_response(); +} +let mut response = services.http_client().send(platform_request).await?; +``` + +- [ ] **Step 5: Run capability tests** + +Run: + +```bash +cargo test-axum publisher_origin_fetch_sets_stream_response_when_supported +cargo test-axum publisher_origin_fetch_leaves_stream_response_disabled_when_unsupported +``` + +Expected: PASS. + +- [ ] **Step 6: Commit** + +```bash +git add crates/trusted-server-core/src/platform/http.rs crates/trusted-server-adapter-fastly/src/platform.rs crates/trusted-server-core/src/platform/test_support.rs crates/trusted-server-core/src/publisher.rs +git commit -m "Gate publisher origin streaming by platform capability" +``` + +## Task 8: Fastly Publisher Streaming Finalize + +**Files:** + +- Modify: `crates/trusted-server-adapter-fastly/src/app.rs` +- Modify: `crates/trusted-server-core/src/publisher.rs` if a helper is needed +- Possibly modify: `crates/trusted-server-adapter-fastly/src/main.rs` for logging/helper naming + +- [ ] **Step 1: Add Fastly route test for publisher response body shape** + +In Fastly app tests, configure a publisher HTML origin response and assert the router returns `Body::Stream` for processable publisher responses on `GET`. + +Expected assertion: + +```rust +assert!( + matches!(response.body(), Body::Stream(_)), + "processable publisher response should remain streaming on Fastly" +); +``` + +- [ ] **Step 2: Add bodiless route tests** + +Assert `HEAD`, `204`, and `304` publisher responses do not carry a stream body, preserving existing metadata. + +- [ ] **Step 3: Run tests and verify failure** + +Run: + +```bash +cargo test-fastly publisher_response_streams +``` + +Expected: FAIL because app still calls `buffer_publisher_response_async()`. + +- [ ] **Step 4: Add a core helper to convert `PublisherResponse` to streaming response** + +Preferred shape in `publisher.rs`: + +```rust +pub fn publisher_response_into_streaming_body( + publisher_response: PublisherResponse, + method: &Method, + settings: Arc, + integration_registry: Arc, + orchestrator: Arc, + services: RuntimeServices, +) -> Result, Report> +``` + +The helper should: + +- Return `PublisherResponse::Buffered` unchanged. +- Return `PublisherResponse::PassThrough` with body attached, except bodiless responses. +- For `PublisherResponse::Stream`, build `EdgeBody::from_stream(futures::stream::unfold(...))` or equivalent. +- Move `OwnedProcessResponseParams`, origin body, settings, registry, orchestrator, and services into the stream state. +- Yield processed chunks as they become available. +- On mid-stream processing error, log and end the stream. The client sees a truncated body, matching existing mid-stream error behavior. + +If borrowing/lifetime pressure is high, keep the helper in Fastly `app.rs` and call core `stream_publisher_body_async()` from inside the stream. Prefer core if it avoids Fastly-specific body processing logic. + +- [ ] **Step 5: Replace Fastly buffered finalize for publisher route** + +In `handle_publisher_route`, replace the `buffer_publisher_response_async()` call for Fastly with the streaming helper. Keep non-Fastly adapters on `buffer_publisher_response_async()`. + +- [ ] **Step 6: Preserve entry-point finalization** + +Verify the returned `Response` still carries extensions needed by `main.rs`: + +- `EcFinalizeState` +- `RequestFilterEffects` +- Final cache privacy guard + +Headers must be finalized before `send_edgezero_response()` splits the response and commits headers. + +- [ ] **Step 7: Run Fastly route tests** + +Run: + +```bash +cargo test-fastly publisher_response +``` + +Expected: PASS. + +- [ ] **Step 8: Commit** + +```bash +git add crates/trusted-server-adapter-fastly/src/app.rs crates/trusted-server-core/src/publisher.rs crates/trusted-server-adapter-fastly/src/main.rs +git commit -m "Stream Fastly publisher responses to clients" +``` + +## Task 9: Pass-Through Publisher Bodies + +**Files:** + +- Modify: `crates/trusted-server-core/src/publisher.rs` +- Modify: `crates/trusted-server-adapter-fastly/src/app.rs` + +- [ ] **Step 1: Add pass-through large body test** + +Verify a non-processable successful publisher response (`image/png`, font, video) is returned as `Body::Stream` when origin streaming is supported and body is not bodiless. + +- [ ] **Step 2: Add pass-through bodiless test** + +Verify `HEAD`, `204`, and `304` pass-through arms preserve headers but do not attach/drain the stream body. + +- [ ] **Step 3: Run targeted tests** + +Run: + +```bash +cargo test-fastly publisher_pass_through +``` + +Expected: FAIL if pass-through still buffers or attaches a body for bodiless responses. + +- [ ] **Step 4: Make pass-through use the same body-carrying guard** + +Mirror `asset_response_carries_body()` semantics in publisher finalize. If `response_carries_body(method, status)` is false, drop the body and return headers only. + +- [ ] **Step 5: Run targeted tests** + +Run: + +```bash +cargo test-fastly publisher_pass_through +``` + +Expected: PASS. + +- [ ] **Step 6: Commit** + +```bash +git add crates/trusted-server-core/src/publisher.rs crates/trusted-server-adapter-fastly/src/app.rs +git commit -m "Preserve publisher pass-through streaming semantics" +``` + +## Task 10: Headers, Length, and Error Semantics + +**Files:** + +- Modify: `crates/trusted-server-core/src/publisher.rs` +- Modify: `crates/trusted-server-adapter-fastly/src/app.rs` +- Modify: `crates/trusted-server-adapter-fastly/src/main.rs` if logs are misleading + +- [ ] **Step 1: Add header tests** + +Assert for streamed processed publisher responses: + +- `Content-Length` is absent. +- `Transfer-Encoding` is not manually set. +- `Content-Encoding` is preserved when recompression is used. +- Cache/privacy headers still downgrade when `Set-Cookie` is present. + +- [ ] **Step 2: Add mid-stream cap/error test** + +Use a body stream that exceeds `publisher.max_buffered_body_bytes` after headers would be committed. Assert the stream returns an error/truncates consistently with existing mid-stream asset behavior and logs enough context. + +- [ ] **Step 3: Run tests and verify failures** + +Run: + +```bash +cargo test-fastly publisher_stream +``` + +Expected: FAIL until header/error cleanup is complete. + +- [ ] **Step 4: Clean header handling** + +Ensure the existing `response.headers_mut().remove(header::CONTENT_LENGTH)` remains on `PublisherResponse::Stream`. Do not re-add content length for streaming finalize. + +- [ ] **Step 5: Improve log wording** + +If `main.rs` still logs "asset streaming" for all `EdgeBody::Stream` responses, rename log messages to "EdgeZero streaming body" or split publisher/asset helpers. + +- [ ] **Step 6: Run tests** + +Run: + +```bash +cargo test-fastly publisher_stream +``` + +Expected: PASS. + +- [ ] **Step 7: Commit** + +```bash +git add crates/trusted-server-core/src/publisher.rs crates/trusted-server-adapter-fastly/src/app.rs crates/trusted-server-adapter-fastly/src/main.rs +git commit -m "Tighten publisher streaming headers and errors" +``` + +## Task 11: End-to-End Regression Coverage + +**Files:** + +- Modify: `crates/trusted-server-adapter-fastly/src/app.rs` +- Modify: existing integration/parity tests if appropriate + +- [ ] **Step 1: Add a slow-origin behavior test if feasible in existing harness** + +Preferred test shape: + +- Origin response body is a stream with first chunk available immediately and second chunk delayed or instrumented. +- Router returns `Body::Stream` without collecting the whole body. +- Pulling the first output chunk does not require pulling the entire origin stream. + +If the harness cannot model time cleanly, use an instrumented stream that panics if polled past the first chunk before the returned response body is consumed. + +- [ ] **Step 2: Add auction timing test** + +Verify response construction does not await `collect_dispatched_auction`; collection happens when the body stream is pulled and reaches `` or EOF. + +- [ ] **Step 3: Run targeted tests** + +Run: + +```bash +cargo test-fastly publisher_streaming_does_not_buffer_origin_before_response +``` + +Expected: PASS after Fastly finalize is lazy. + +- [ ] **Step 4: Commit** + +```bash +git add crates/trusted-server-adapter-fastly/src/app.rs crates/trusted-server-core/src/publisher.rs +git commit -m "Cover lazy publisher streaming behavior" +``` + +## Task 12: Documentation Cleanup + +**Files:** + +- Modify: `crates/trusted-server-core/src/publisher.rs` +- Modify: `crates/trusted-server-core/src/settings.rs` +- Modify: `crates/trusted-server-adapter-fastly/src/app.rs` +- Optionally modify: issue/PR description only, not repo docs + +- [ ] **Step 1: Update stale interim comments** + +Remove or rewrite comments saying publisher stream bodies are already materialized into WASM heap. + +Targets: + +- `PublisherResponse::Stream` doc. +- `PublisherResponse::PassThrough` doc if Fastly now preserves stream bodies. +- `settings.rs` comments that reference future true streaming. +- Fastly app module comment that says publisher responses are buffered by `publisher.max_buffered_body_bytes`. + +- [ ] **Step 2: Run doc-related checks locally** + +Run: + +```bash +cargo fmt --all -- --check +``` + +Expected: PASS. + +- [ ] **Step 3: Commit** + +```bash +git add crates/trusted-server-core/src/publisher.rs crates/trusted-server-core/src/settings.rs crates/trusted-server-adapter-fastly/src/app.rs +git commit -m "Update publisher streaming documentation" +``` + +## Task 13: Full Verification + +**Files:** + +- No source edits unless failures reveal issues. + +- [ ] **Step 1: Run formatting** + +Run: + +```bash +cargo fmt --all -- --check +``` + +Expected: PASS. + +- [ ] **Step 2: Run target checks** + +Run: + +```bash +cargo check-fastly +cargo check-axum +cargo check-cloudflare +``` + +Expected: PASS. + +- [ ] **Step 3: Run target tests** + +Run: + +```bash +cargo test-fastly +cargo test-axum +cargo test-cloudflare +``` + +Expected: PASS. + +- [ ] **Step 4: Run Spin if touched by shared trait changes** + +Run: + +```bash +cargo test-spin +``` + +Expected: PASS. + +- [ ] **Step 5: Run clippy gates** + +Run: + +```bash +cargo clippy-fastly +cargo clippy-axum +cargo clippy-cloudflare +cargo clippy-cloudflare-wasm +cargo clippy-spin-native +cargo clippy-spin-wasm +``` + +Expected: PASS. + +- [ ] **Step 6: Run parity suite if available locally** + +Run: + +```bash +cargo test --manifest-path crates/trusted-server-integration-tests/Cargo.toml --test parity +``` + +Expected: PASS. + +- [ ] **Step 7: Optional local TTFB smoke** + +Run Fastly local serve against an artificially slow publisher origin: + +- Origin sends headers and first HTML chunk immediately. +- Origin delays later body chunks. +- Verify browser/curl receives response headers and first chunk before full origin drain. +- Verify bids still inject before `` when auction completes. + +Expected: TTFB tracks origin first byte, not full origin transfer or auction collection. + +## Review Checklist + +- [ ] No `block_on` inside stream body processing or `Read::read` equivalents. +- [ ] `Body::Stream` never falls through `into_bytes().unwrap_or_default()`. +- [ ] Fastly publisher origin fetch sets `with_stream_response()` only through capability gate. +- [ ] Cloudflare, Spin, and Axum do not start receiving stream-response requests. +- [ ] `HEAD`, `204`, and `304` do not drive or attach response bodies. +- [ ] `Content-Length` is absent on processed streaming responses. +- [ ] Existing buffered adapters still work through `buffer_publisher_response_async()`. +- [ ] Auction telemetry handles completed and abandoned stream cases. +- [ ] Mid-stream errors do not panic; they log and truncate consistently with current streaming behavior. +- [ ] Comments no longer describe publisher streaming as interim/in-memory cursor based. + +## PR Description Skeleton + +```markdown +## Summary + +- convert publisher response processing to async chunk-pull over `Body::Stream` +- enable Fastly publisher origin streaming behind a platform capability gate +- stream Fastly publisher responses to clients instead of buffering and awaiting auction before send + +## Scope + +Fastly production path for issue #849. Cloudflare, Spin, and Axum streaming remain follow-up work. + +## Tests + +- [ ] cargo fmt --all -- --check +- [ ] cargo check-fastly +- [ ] cargo check-axum +- [ ] cargo check-cloudflare +- [ ] cargo test-fastly +- [ ] cargo test-axum +- [ ] cargo test-cloudflare +- [ ] cargo test-spin +- [ ] cargo clippy-fastly +- [ ] cargo clippy-axum +- [ ] cargo clippy-cloudflare +- [ ] cargo clippy-cloudflare-wasm +- [ ] cargo clippy-spin-native +- [ ] cargo clippy-spin-wasm +``` + +## Known Follow-Ups + +- Cloudflare origin streaming once Worker `ReadableStream` is wrapped into `Body::Stream` and response header/set-cookie behavior is verified. +- Spin streaming after upstream EdgeZero Spin response conversion supports incremental body writes. +- Axum streaming only if the dev server needs it enough to justify a `Send` bridge. +- Issue #850 parser-context `` detection.