Bun.serve: send the buffered tail when a direct stream controller is close()d

src/runtime/webcore/streams.rs

@@ -1051,8 +1051,8 @@ pub struct HTTPServerWritable<const SSL: bool, const HTTP3: bool> {
     /// `flush_promise()` → `pending.run()`.
     pub(crate) pending: WritablePending,
     pub(crate) wrote_at_start_of_flush: BlobSizeType,
-    // JSC_BORROW: process-lifetime VM global; `None` until `flush_from_js`/
-    // `end_from_js` install it. Safe `Deref` via `BackRef`.
+    // JSC_BORROW: process-lifetime VM global, installed by
+    // `RequestContext::do_render_stream` at construction. Safe `Deref` via `BackRef`.
     pub global_this: Option<BackRef<JSGlobalObject>>,
     pub(crate) high_water_mark: BlobSizeType,
 
@@ -1643,13 +1643,7 @@ impl<const SSL: bool, const HTTP3: bool> HTTPServerWritable<SSL, HTTP3> {
             }
         }
         self.wrote_at_start_of_flush = self.wrote;
-        self.pending_flush = Some(JSPromise::create(global_this));
-        self.global_this = Some(BackRef::new(global_this));
-        // S008: `JSPromise` is an `opaque_ffi!` ZST — safe `*const → &` deref.
-        let promise_value = JSPromise::opaque_ref(self.pending_flush.unwrap()).to_js();
-        promise_value.protect();
-
-        bun_sys::Result::Ok(promise_value)
+        bun_sys::Result::Ok(self.park_pending_flush(global_this))
     }
 
     pub fn flush(&mut self) -> bun_sys::Result<()> {
@@ -1806,7 +1800,9 @@ impl<const SSL: bool, const HTTP3: bool> HTTPServerWritable<SSL, HTTP3> {
         self.unregister_auto_flusher();
     }
 
-    /// In this case, it's always an error
+    /// Controller `close()`. The buffered tail is sent here rather than left
+    /// to the auto-flusher: a `pull()` that closes synchronously has its sink
+    /// torn down by `do_render_stream` before any deferred task runs.
     pub(crate) fn end(&mut self, err: Option<SysError>) -> bun_sys::Result<()> {
         bun_core::scoped_log!(HTTPServerWritableLog, "end({:?})", err);
 
@@ -1833,6 +1829,13 @@ impl<const SSL: bool, const HTTP3: bool> HTTPServerWritable<SSL, HTTP3> {
             self.finalize();
             return bun_sys::Result::Ok(());
         }
+
+        if self.send_readable(0) {
+            self.handle_ended_response(err);
+        } else {
+            let global_this = BackRef::new(self.global_this());
+            self.park_pending_flush(&global_this);
+        }
         bun_sys::Result::Ok(())
     }
 
@@ -1857,32 +1860,46 @@ impl<const SSL: bool, const HTTP3: bool> HTTPServerWritable<SSL, HTTP3> {
 
         if readable_len > 0 {
             if !self.send_readable(0) {
-                self.pending_flush = Some(JSPromise::create(global_this));
-                self.global_this = Some(BackRef::new(global_this));
-                // S008: `JSPromise` is an `opaque_ffi!` ZST — safe `*const → &` deref.
-                let value = JSPromise::opaque_ref(self.pending_flush.unwrap()).to_js();
-                value.protect();
-                return bun_sys::Result::Ok(value);
+                return bun_sys::Result::Ok(self.park_pending_flush(global_this));
             }
         } else {
             if let Some(res) = self.any_res() {
                 res.end(b"", false);
             }
         }
 
+        self.handle_ended_response(None);
+
+        bun_sys::Result::Ok(JSValue::from(self.wrote))
+    }
+
+    /// uWS could not drain what was handed to it: `on_writable` settles this
+    /// promise via `flush_promise` once it has (resending the `try_end` tail
+    /// still in `buffer` first). For an ended sink the request waits on it
+    /// instead of tearing the sink down (`do_render_stream`,
+    /// `handle_resolve_stream`).
+    fn park_pending_flush(&mut self, global_this: &JSGlobalObject) -> JSValue {
+        let promise = JSPromise::create(global_this);
+        self.pending_flush = Some(promise);
+        self.global_this = Some(BackRef::new(global_this));
+        // S008: `JSPromise` is an `opaque_ffi!` ZST — safe `*const → &` deref.
+        let value = JSPromise::opaque_ref(promise).to_js();
+        value.protect();
+        value
+    }
+
+    /// `end()`/`end_from_js()` fully ended the response through uWS, which
+    /// `markDone()`s it and drops its `onAborted` (see `ended_response`).
+    fn handle_ended_response(&mut self, err: Option<SysError>) {
         if let Some(res) = self.any_res() {
             // Release any request-body pause while `res` is live (see `end_already_responded_stream`).
             res.resume();
         }
-        // Both branches above fully ended the response through uWS, which
-        // `markDone()`s it and drops its `onAborted`.
         self.ended_response = true;
         self.mark_done();
-        let _ = self.flush_promise(); // TODO: properly propagate exception upwards
-        self.source.close(None);
+        let _ = self.flush_promise();
+        self.source.close(err);
         self.finalize();
-
-        bun_sys::Result::Ok(JSValue::from(self.wrote))
     }
 
     /// Takes `*mut Self`, not `&mut self`: closing the signal runs the controller's
@@ -1942,6 +1959,9 @@ impl<const SSL: bool, const HTTP3: bool> HTTPServerWritable<SSL, HTTP3> {
             self.auto_flusher.registered.set(false);
             return false;
         }
+        // `end()`/`end_from_js()` send their own tail (or park it for
+        // `on_writable`) and unregister this flusher on the way.
+        debug_assert!(!self.requested_end);
 
         let readable_len = self.readable_slice().len();
 
@@ -1955,20 +1975,6 @@ impl<const SSL: bool, const HTTP3: bool> HTTPServerWritable<SSL, HTTP3> {
             return true;
         }
         self.auto_flusher.registered.set(false);
-
-        if self.requested_end {
-            if let Some(res) = self.any_res() {
-                res.clear_on_writable();
-                // Release any request-body pause while `res` is live (see `end_already_responded_stream`).
-                res.resume();
-            }
-            // `send_readable` drained the parked `try_end`/`end`, so uWS has
-            // `markDone()`d the response and dropped its `onAborted`.
-            self.ended_response = true;
-            self.source.close(None);
-            let _ = self.flush_promise();
-            self.finalize();
-        }
         false
     }
 

test/js/bun/http/serve-direct-readable-stream.test.ts

@@ -617,6 +617,40 @@ describe("end() under transport backpressure over h3", () => {
     const res = await h3fetch(server);
     expect(await res.text()).toBe("hey");
   });
+
+  // close() must park the same pending flush as end(). It used to leave the
+  // tail to the auto-flusher instead, and when that try_end hit backpressure
+  // nothing was parked for the request to wait on, so the sink was torn down
+  // and the body came back empty even from an async pull().
+  test("async pull() that closes synchronously", async () => {
+    using server = serveH3(
+      () =>
+        new ReadableStream({
+          type: "direct",
+          async pull(c: any) {
+            c.write("hey");
+            c.close();
+          },
+        } as any),
+    );
+    const res = await h3fetch(server);
+    expect(await res.text()).toBe("hey");
+  });
+
+  test("sync pull() that closes synchronously", async () => {
+    using server = serveH3(
+      () =>
+        new ReadableStream({
+          type: "direct",
+          pull(c: any) {
+            c.write("hey");
+            c.close();
+          },
+        } as any),
+    );
+    const res = await h3fetch(server);
+    expect(await res.text()).toBe("hey");
+  });
 });
 
 // The controller's detach() used to skip the close callback when it was
@@ -729,3 +763,81 @@ test("close() with unflushed data writes the chunked terminator exactly once", a
   expect(afterTerminator.slice(0, 12)).toBe("HTTP/1.1 200");
   expect(afterTerminator).toEndWith("ok");
 });
+
+// close() with bytes still buffered below the high-water mark left them for
+// the auto-flusher. A pull() that closes synchronously never gets there: the
+// request finalizes the sink as soon as pull() returns, so the buffered tail
+// was dropped and the client got Content-Length: 0 (or, after a mid-stream
+// flush(), a chunked body missing its tail). end() in the same position sent
+// everything; close() must too.
+describe("buffered bytes are sent when the controller finishes", () => {
+  type Framing = { body: string; contentLength: string | null; transferEncoding: string | null };
+  const unflushed: Framing = { body: "helloworld", contentLength: "10", transferEncoding: null };
+  const flushedMidway: Framing = { body: "helloworld", contentLength: null, transferEncoding: "chunked" };
+  const nothingWritten: Framing = { body: "", contentLength: "0", transferEncoding: null };
+  // A write of at least the sink's high-water mark (2048 bytes by default)
+  // goes straight to the socket instead of the buffer, so close() finds
+  // nothing left to send and only has to terminate the chunked response.
+  const highWaterMarkBody = Buffer.alloc(2048, "x").toString();
+  const sentByWrite: Framing = { body: highWaterMarkBody, contentLength: null, transferEncoding: "chunked" };
+
+  // The writes stay below the sink's high-water mark, so whatever follows the
+  // last flush() is still buffered when the controller is finished.
+  const writeThen = (finish: "close" | "end", flushMidway: boolean) => (c: any) => {
+    c.write("hello");
+    if (flushMidway) c.flush();
+    c.write("world");
+    c[finish]();
+  };
+
+  const cases: [name: string, pull: (c: any) => unknown, expected: Framing][] = [
+    ["sync pull, close()", writeThen("close", false), unflushed],
+    ["sync pull, end()", writeThen("end", false), unflushed],
+    ["sync pull, flush() midway, close()", writeThen("close", true), flushedMidway],
+    ["sync pull, flush() midway, end()", writeThen("end", true), flushedMidway],
+    ["async pull, close()", async c => writeThen("close", false)(c), unflushed],
+    ["async pull, flush() midway, close()", async c => writeThen("close", true)(c), flushedMidway],
+    [
+      "sync pull, single write, close()",
+      c => {
+        c.write("helloworld");
+        c.close();
+      },
+      unflushed,
+    ],
+    // Nothing is buffered in these two, so close() takes its empty-buffer
+    // path; they pin down that it still ends the response.
+    ["sync pull, close() without writing", c => c.close(), nothingWritten],
+    [
+      "sync pull, write at the high-water mark, close()",
+      c => {
+        c.write(highWaterMarkBody);
+        c.close();
+      },
+      sentByWrite,
+    ],
+  ];
+
+  // The TLS server uses a separate instantiation of the sink, so the matrix
+  // runs over both.
+  describe.each([
+    ["http", {}, {}],
+    ["https", { tls }, { tls: { rejectUnauthorized: false } }],
+  ])("over %s", (_protocol, serveOptions, fetchOptions) => {
+    test.concurrent.each(cases)("%s", async (_name, pull, expected) => {
+      using server = Bun.serve({
+        port: 0,
+        ...serveOptions,
+        fetch: () => new Response(new ReadableStream({ type: "direct", pull } as any)),
+      });
+
+      const response = await fetch(server.url, fetchOptions);
+      expect({
+        body: await response.text(),
+        contentLength: response.headers.get("content-length"),
+        transferEncoding: response.headers.get("transfer-encoding"),
+      }).toEqual(expected);
+      expect(response.status).toBe(200);
+    });
+  });
+});