Enable streaming output by flushing during processing

Cargo.toml

@@ -11,7 +11,7 @@ members = [
 resolver = "2"
 
 [workspace.package]
-version = "0.7.0-beta.4"
+version = "0.7.0"
 authors = [
     "Kailan Blanks <kblanks@fastly.com>",
     "Vadim Getmanshchuk <vadim@fastly.com>",

esi/Cargo.toml

@@ -13,7 +13,7 @@ expose-internals = []
 
 [dependencies]
 thiserror = "2.0.6"
-fastly = "^0.11"
+fastly = "^0.12"
 log = "^0.4"
 regex = "1.11.1"
 html-escape = "0.2.13"

esi/src/lib.rs

@@ -797,17 +797,24 @@ impl Processor {
 
             match parse_result {
                 Ok((remaining, elements)) => {
-                    let mut handler = DocumentHandler {
-                        processor: self,
-                        output: output_writer,
-                        dispatch_fragment_request: dispatcher,
-                        fragment_response_handler: process_fragment_response,
-                    };
-                    for element in elements {
-                        handler.process(&element)?;
-                        handler.process_queue()?;
+                    {
+                        let mut handler = DocumentHandler {
+                            processor: self,
+                            output: output_writer,
+                            dispatch_fragment_request: dispatcher,
+                            fragment_response_handler: process_fragment_response,
+                        };
+                        for element in elements {
+                            handler.process(&element)?;
+                            handler.process_queue()?;
+                        }
                     }
 
+                    // Flush any content written during this parse batch so it
+                    // reaches the client immediately (progressive streaming)
+                    // rather than buffering until drain_queue completes.
+                    output_writer.flush()?;
+
                     if eof {
                         // Nothing left to read — we're done
                         break;
@@ -1151,14 +1158,41 @@ impl Processor {
                             ready => {
                                 // CompletedRequest or NoContent: process now.
                                 fragment.pending_fragment = ready;
-                                let mut slot_buf = Vec::new();
-                                self.process_include(
-                                    *fragment,
-                                    &mut slot_buf,
-                                    dispatch_fragment_request,
-                                    process_fragment_response,
-                                )?;
-                                buf[slot] = Some(Bytes::from(slot_buf));
+
+                                if slot == next_out {
+                                    // Head-of-line: stream directly to the
+                                    // client rather than buffering.
+                                    self.process_include(
+                                        *fragment,
+                                        output_writer,
+                                        dispatch_fragment_request,
+                                        process_fragment_response,
+                                    )?;
+                                    buf[slot] = Some(Bytes::new());
+                                    next_out += 1;
+
+                                    // Flush any subsequent ready slots.
+                                    while next_out < buf.len() {
+                                        match &buf[next_out] {
+                                            Some(bytes) => {
+                                                output_writer.write_all(bytes)?;
+                                                buf[next_out] = Some(Bytes::new());
+                                                next_out += 1;
+                                            }
+                                            None => break,
+                                        }
+                                    }
+                                    output_writer.flush()?;
+                                } else {
+                                    let mut slot_buf = Vec::new();
+                                    self.process_include(
+                                        *fragment,
+                                        &mut slot_buf,
+                                        dispatch_fragment_request,
+                                        process_fragment_response,
+                                    )?;
+                                    buf[slot] = Some(Bytes::from(slot_buf));
+                                }
                                 // dca="esi" may push new items onto self.queue;
                                 // the outer while picks them up next iteration.
                             }
@@ -1325,6 +1359,11 @@ impl Processor {
                 }
             }
 
+            // Flush written content to the client before potentially blocking
+            // on select().  Without this, data sits in the writer's buffer
+            // while we wait for slow includes, defeating streaming.
+            output_writer.flush()?;
+
             // ------------------------------------------------------------------
             // Step 3: done when nothing is pending.
             // ------------------------------------------------------------------
@@ -1399,16 +1438,47 @@ impl Processor {
                 // -------------------------------------------------------
                 None => {
                     fragment.pending_fragment = completed_content;
-                    let mut slot_buf = Vec::new();
-                    self.process_include(
-                        *fragment,
-                        &mut slot_buf,
-                        dispatch_fragment_request,
-                        process_fragment_response,
-                    )?;
-                    buf[buf_slot] = Some(Bytes::from(slot_buf));
-                    // dca="esi" may push new QueuedElements onto self.queue.
-                    // Loop back to Step 1 to assign them slots.
+
+                    if buf_slot == next_out {
+                        // Head-of-line: write directly to the streaming
+                        // output instead of buffering into a slot.  This
+                        // lets nested dca="esi" fragments stream
+                        // incrementally — their isolated drain_queue
+                        // writes (and flushes) to the real client writer
+                        // rather than to a Vec that only surfaces later.
+                        self.process_include(
+                            *fragment,
+                            output_writer,
+                            dispatch_fragment_request,
+                            process_fragment_response,
+                        )?;
+                        buf[buf_slot] = Some(Bytes::new()); // mark consumed
+                        next_out += 1;
+
+                        // Advance past any subsequent ready slots that
+                        // can now be flushed contiguously.
+                        while next_out < buf.len() {
+                            match &buf[next_out] {
+                                Some(bytes) => {
+                                    output_writer.write_all(bytes)?;
+                                    buf[next_out] = Some(Bytes::new());
+                                    next_out += 1;
+                                }
+                                None => break,
+                            }
+                        }
+                        output_writer.flush()?;
+                    } else {
+                        let mut slot_buf = Vec::new();
+                        self.process_include(
+                            *fragment,
+                            &mut slot_buf,
+                            dispatch_fragment_request,
+                            process_fragment_response,
+                        )?;
+                        buf[buf_slot] = Some(Bytes::from(slot_buf));
+                    }
+                    // Loop back to Step 1 to pick up any new queue items.
                 }
 
                 // -------------------------------------------------------

esi/tests/esi_tests.rs

@@ -202,23 +202,19 @@ fn test_esi_choose_compatibility_not_equal() {
 // Test for nested variable expansion - INVALID ESI SYNTAX
 // The construct $($(outer){param}) is NOT valid Akamai ESI syntax.
 // Akamai's ESI does not support nested variable expansion like this.
-// This test was checking that it doesn't work, but the syntax is so invalid
-// that different parsers may handle it differently (error vs. pass-through).
 #[test]
-#[ignore] // Invalid ESI syntax - $($(var){key}) is not supported by Akamai ESI spec
-fn test_nested_subfields() {
+fn test_nested_subfields_is_invalid() {
     let input = r#"
         <esi:assign name="outer" value="'QUERY_STRING'" />
         <esi:vars>
             $($(outer){param})
         </esi:vars>
     "#;
     let req = Request::get("http://example.com?param=value");
-    let result = process_esi_document(input, req).expect("Processing should succeed");
-    assert_ne!(
-        result.trim(),
-        "value",
-        "Nested variable expansion is not valid ESI syntax and should not work"
+    let result = process_esi_document(input, req);
+    assert!(
+        result.is_err(),
+        "Nested variable expansion $($(var){{key}}) is not valid ESI syntax and should fail"
     );
 }