Harden the streaming publisher pipeline after review

Address the deep-review findings on the streaming cutover:

- Cap cumulative decoded bytes in BodyStreamDecoder against
  publisher.max_buffered_body_bytes: the chunk source only bounds raw
  compressed bytes, so a decompression bomb could expand ~1000x past it
  and push unbounded decoded volume through the rewrite pipeline
- Detect truncated deflate streams: write::ZlibDecoder::try_finish
  accepts truncated input silently, so the deflate arm now drives
  flate2::Decompress directly and requires Status::StreamEnd at
  finalization; trailing bytes after the end marker stay ignored.
  Add truncated-gzip and truncated-deflate regression tests
- Make BodyChunkSource::next_chunk cancellation-safe by polling the
  body in place instead of moving it out across an await; a cancelled
  pull no longer turns into a silent EOF
- Log dispatched auctions dropped uncollected (client disconnect
  mid-stream or never-polled body) via DispatchedAuctionGuard; the
  guard is created before the lazy stream so unpolled drops log too
- Share the pull+decode step between the lazy publisher body and the
  write-sink drivers (hold_step_next_chunk / passthrough_step),
  removing the unreachable!() error plumbing and the triplicated
  processor selection; document body_close_hold_loop_stream as
  groundwork for the buffered adapters' streaming cutover
- Pass identity-encoded chunks through zero-copy and finish encoders
  by consuming them instead of allocating a throwaway replacement
- Add a Fastly dispatch test asserting the publisher fallback returns
  Body::Stream without a stale Content-Length, plus a comment on why
  the publisher fetch gates streaming on capability while the asset
  path does not

Behavior note: gzip bodies with trailing garbage after the trailer now
error mid-stream; the old read-path decoder ignored them.

Author: prk-Jr

crates/trusted-server-adapter-fastly/src/app.rs

@@ -2150,6 +2150,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,
@@ -2260,6 +2264,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