Stream publisher origin bodies end-to-end on Fastly

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"

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

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);
                 }
             }

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,

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 }

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

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<Vec<Option<PlatformImageOptimizerOptions>>>,
     stream_response_flags: Mutex<Vec<bool>>,
     request_methods: Mutex<Vec<String>>,
@@ -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<u8>) {
         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,

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<W: Write>(
             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<W: Write>(
                 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(),
                     })?;
             }
         }