Bun.serve: send the whole HTTP/1.1 response head when a streamed body is not ready

packages/bun-uws/src/HttpResponse.h

@@ -171,6 +171,10 @@ struct HttpResponse : public AsyncSocket<SSL> {
                 data->state |= HttpResponseData<SSL>::HTTP_CLOSE_WHEN_IDLE;
                 idle = false;
             }
+        } else if (idle && !Super::hasFullyDrained()) {
+            /* Bun.serve: close() would drop the bytes of a finished response that are still in the backpressure buffer.
+             * Not idle until they are out: the graceful stop() marks the connection, and onWritable closes it. */
+            idle = false;
         }
         if (!idle) {
             return false;

src/runtime/server/RequestContext.rs

@@ -832,6 +832,8 @@ where
 
         // SAFETY: `response` is the live, protect()'d cell pointer.
         unsafe { self.render(response) };
+        #[cfg(debug_assertions)]
+        self.assert_no_parked_open_head();
     }
 
     #[inline]
@@ -2157,6 +2159,87 @@ where
         readable_ref.deinit();
     }
 
+    /// HTTP/1.1, and the sink has not ended the response (`resp` is stale after that).
+    fn takes_early_head(sink: &ResponseStream<SSL_ENABLED>, resp: uws::AnyResponse) -> bool {
+        !MUX && !sink.ended_response && !resp.state().is_ancient_request()
+    }
+
+    /// Commits the head after the microtask drain, so a body that ends in it keeps its framing.
+    fn defer_head_commit(&self) {
+        debug_assert!(!self.flags.has_deferred_head_commit());
+        self.flags.set_has_deferred_head_commit(true);
+        // Released by the task.
+        self.ref_();
+        let ctx = NonNull::new(self.as_ctx_ptr().cast::<c_void>());
+        let found_existing = self
+            .server()
+            .vm()
+            .as_mut()
+            .event_loop_mut()
+            .deferred_tasks
+            .post_task(ctx, Self::on_deferred_head_commit);
+        debug_assert!(!found_existing);
+    }
+
+    /// The task of [`Self::defer_head_commit`].
+    extern "C" fn on_deferred_head_commit(ctx: *mut c_void) -> bool {
+        // Adopts the ref `defer_head_commit` took.
+        let pinned = RequestContextRef::adopt(ctx.cast::<Self>());
+        let this = pinned.ctx();
+        this.flags.set_has_deferred_head_commit(false);
+
+        // Detached: the response ended, failed or was aborted inside the drain.
+        let Some(resp) = this.resp.get() else {
+            return false;
+        };
+        // No sink: the stream settled in the drain, and its teardown ends the response.
+        let Some(sink) = this.sink_mut() else {
+            return false;
+        };
+        if !sink.sends_head_itself() {
+            resp.commit_head();
+        }
+        false
+    }
+
+    /// `render_metadata` and the rest of the head, for a body that has no bytes yet.
+    fn render_committed_head_corked(this: *mut Self) {
+        // SAFETY: this is the live RequestContext threaded through cork user-data.
+        let this = unsafe { &*this };
+        this.render_metadata();
+        if let Some(resp) = this.resp.get() {
+            resp.commit_head();
+        }
+    }
+
+    /// A parked HTTP/1 response with a status has its whole head out, or a task that sends it.
+    #[cfg(debug_assertions)]
+    pub(crate) fn assert_no_parked_open_head(&self) {
+        // File bodies are exempt: `prepare_for_sendfile` ends the head without a state bit.
+        if MUX
+            || !self.flags.has_marked_pending()
+            || !self.flags.has_written_status()
+            || self.flags.has_sendfile_ctx()
+            || self.flags.has_deferred_head_commit()
+        {
+            return;
+        }
+        let Some(resp) = self.resp.get() else {
+            return;
+        };
+        if self.sink_mut().is_some_and(|sink| sink.sends_head_itself()) {
+            return;
+        }
+        let state = resp.state();
+        debug_assert!(
+            !state.is_response_pending()
+                || state.is_http_write_called()
+                || state.is_http_end_called()
+                || state.is_ancient_request(),
+            "a parked response left its head open"
+        );
+    }
+
     fn do_render_stream(pair: *mut StreamPair<'_, ThisServer, SSL_ENABLED, DEBUG_MODE, MUX>) {
         ctx_log!("doRenderStream");
         // SAFETY: pair is a stack local threaded through cork user-data.
@@ -2297,6 +2380,15 @@ where
                             this.render_metadata();
                         }
 
+                        if Self::takes_early_head(response_stream, resp) {
+                            if this.is_async() {
+                                // No drain ran above: the body's microtasks are still queued.
+                                this.defer_head_commit();
+                            } else if !response_stream.sends_head_itself() {
+                                resp.commit_head();
+                            }
+                        }
+
                         // TODO: should this timeout?
                         let body_value = this.response_mut().unwrap().get_body_value();
                         *body_value = Body::Value::Locked(Body::PendingValue {
@@ -3247,15 +3339,16 @@ where
                                 || matches!(
                                     byte_stream.parent_const().producer.get(),
                                     WebCore::streams::SourceHandle::HTMLRewriter(_)
+                                        | WebCore::streams::SourceHandle::S3DownloadBody(_)
                                 )
                             {
                                 // Defer status/headers to the first chunk/end
-                                // so a pre-first-byte handler failure can
+                                // so a failure before the first byte can
                                 // still reach `error()`.
                             } else {
-                                // if we only have metadata to send, send it now
+                                // No body bytes yet: send the whole head now.
                                 resp.run_corked_with_type(
-                                    Self::render_metadata_corked,
+                                    Self::render_committed_head_corked,
                                     this.as_ctx_ptr(),
                                 );
                             }
@@ -3318,7 +3411,7 @@ where
         }
         let resp = this.resp.get().expect("infallible: resp bound");
 
-        // A rewriter-produced `Source::Bytes` body defers metadata until the
+        // A rewriter or S3 `Source::Bytes` body defers metadata until the
         // first chunk so a pre-first-byte failure can still reach the
         // server's `error()` hook. Flush it now, corked.
         if !this.flags.has_written_status() {
@@ -4599,6 +4692,8 @@ bitflags::bitflags! {
         const REQUEST_BODY_PAUSED         = 1 << 16;
         /// `on_start_buffering` fired (`.text()` etc.); skip pre-stream backpressure.
         const REQUEST_BODY_BUFFER_ALL     = 1 << 17;
+        /// `defer_head_commit` posted its task and the task has not run yet.
+        const HAS_DEFERRED_HEAD_COMMIT    = 1 << 18;
     }
 }
 
@@ -4680,6 +4775,11 @@ impl<const DEBUG_MODE: bool> Flags<DEBUG_MODE> {
         set_request_body_buffer_all,
         REQUEST_BODY_BUFFER_ALL
     );
+    flag_accessor!(
+        has_deferred_head_commit,
+        set_has_deferred_head_commit,
+        HAS_DEFERRED_HEAD_COMMIT
+    );
 
     #[inline]
     pub(crate) fn is_web_browser_navigation(&self) -> bool {

src/runtime/server/mod.rs

@@ -1115,6 +1115,8 @@ impl<const SSL: bool, const DEBUG: bool> NewServer<SSL, DEBUG> {
             ctx_ref.render_missing();
             return;
         }
+        #[cfg(debug_assertions)]
+        ctx_ref.assert_no_parked_open_head();
 
         // The request is asynchronous, and all information from `req` must be copied
         // since the provided uws.Request will be re-used for future requests (stack allocated).
@@ -1177,6 +1179,8 @@ impl<const SSL: bool, const DEBUG: bool> NewServer<SSL, DEBUG> {
             ctx_ref.render_missing();
             return;
         }
+        #[cfg(debug_assertions)]
+        ctx_ref.assert_no_parked_open_head();
 
         // The request is asynchronous, and all information from `req` must be
         // copied since the provided uws.Request will be re-used for future

src/runtime/webcore/streams.rs

@@ -1806,6 +1806,13 @@ impl<const SSL: bool> HTTPServerWritable<SSL> {
         source.abort(CommonAbortReason::ConnectionClosed);
     }
 
+    /// The sink ended the response, parked an `end()`, or queued a flush: each completes the head.
+    pub(crate) fn sends_head_itself(&self) -> bool {
+        self.ended_response
+            || self.requested_end
+            || (self.auto_flusher.registered.get() && !self.readable_slice().is_empty())
+    }
+
     fn unregister_auto_flusher(&mut self) {
         if self.auto_flusher.registered.get() {
             let vm = self.global_this().bun_vm();

src/uws_sys/Response.rs

@@ -860,6 +860,24 @@ impl AnyResponse {
         any_dispatch!(self, |r| r.flush_headers(flush_immediately))
     }
 
+    /// Ends the header section of a live HTTP/1.1 response whose body has sent nothing yet.
+    pub fn commit_head(self) {
+        // HTTP/3 ends a failed body with a FIN, and early HTTP/2 HEADERS cost a socket write.
+        if matches!(self, AnyResponse::H3(_) | AnyResponse::H2(_)) {
+            return;
+        }
+        let open_head = State::HTTP_STATUS_CALLED | State::HTTP_RESPONSE_PENDING;
+        // A `try_end` parked on backpressure sets only END_CALLED: its head is out too.
+        let head_is_out = State::HTTP_WRITE_CALLED | State::HTTP_END_CALLED;
+        // HTTP/1.0: the close ends the body, so a cut behind a whole head reads as an empty body.
+        let close_ends_body = State::HTTP_ANCIENT_REQUEST;
+        let skip = head_is_out | close_ends_body;
+        if self.state().bits() & (open_head | skip).bits() != open_head.bits() {
+            return;
+        }
+        self.corked(|| self.flush_headers(false));
+    }
+
     pub fn is_corked(self) -> bool {
         any_dispatch!(self, |r| r.is_corked())
     }
@@ -1227,6 +1245,7 @@ bitflags::bitflags! {
         const HTTP_RESPONSE_PENDING            = 8;
         const HTTP_CONNECTION_CLOSE            = 16;
         const HTTP_WROTE_CONTENT_LENGTH_HEADER = 32;
+        const HTTP_ANCIENT_REQUEST             = 1 << 8;
         const HTTP_NODE_RECEIVED_FIN           = 1 << 15;
         const HTTP_NODE_CLOSE_AFTER_MESSAGE    = 1 << 20;
     }
@@ -1263,6 +1282,12 @@ impl State {
         self.bits() & State::HTTP_CONNECTION_CLOSE.bits() != 0
     }
 
+    /// The request was HTTP/1.0: a body with no `Content-Length` ends when the connection closes.
+    #[inline]
+    pub fn is_ancient_request(self) -> bool {
+        self.bits() & State::HTTP_ANCIENT_REQUEST.bits() != 0
+    }
+
     #[inline]
     pub fn is_node_received_fin(self) -> bool {
         self.bits() & State::HTTP_NODE_RECEIVED_FIN.bits() != 0

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

@@ -875,14 +875,20 @@ describe("a pull() that outlives its own end() does not hold the request", () =>
   // tick make a single tryEnd() of the whole body. The client is not reading,
   // so most of it stays unsent and end() parks on the drain. The request then
   // waits for that drain, and only for it.
-  describe("end() parked on transport backpressure", () => {
+  //
+  // A pull() that is still waiting when the handler's microtasks are done has
+  // its response head sent, and the head says chunked. An end() after that
+  // does not park: the body is one chunk, uWS keeps what the client has not
+  // read, and the request is released at once.
+  describe("end() under transport backpressure", () => {
     const payload = Buffer.alloc(32 * 1024 * 1024, "x");
+    const chunkFraming = `${payload.length.toString(16)}\r\n`.length + "\r\n0\r\n\r\n".length;
 
     // Windows takes the first write of a response whole, whatever its size
     // (250 MB to a client that is not reading), so one tryEnd() never parks.
     test.skipIf(isWindows).each([
-      ["in the first tick of pull()", false],
-      ["from a later task", true],
+      ["in the first tick of pull(), parks", false],
+      ["from a later task, after the head was sent, does not park", true],
     ] as const)("%s", async (_when, fromLaterTask) => {
       const ended = Promise.withResolvers<boolean>();
       await using server = Bun.serve({
@@ -906,21 +912,26 @@ describe("a pull() that outlives its own end() does not hold the request", () =>
       });
 
       const socket = net.connect(server.port!, "127.0.0.1");
-      const received = Promise.withResolvers<number>();
+      const received = Promise.withResolvers<{ chunked: boolean; bodyBytes: number }>();
       let head = "";
+      let chunked = false;
       let bodyBytes = -1;
+      let tail = "";
       socket.on("error", received.reject);
       socket.on("close", () => received.reject(new Error("closed before the whole body arrived")));
       socket.on("data", chunk => {
         if (bodyBytes < 0) {
           head += chunk.toString("latin1");
           const headEnd = head.indexOf("\r\n\r\n");
           if (headEnd < 0) return;
+          chunked = /^transfer-encoding: chunked$/im.test(head.slice(0, headEnd));
           bodyBytes = Buffer.byteLength(head, "latin1") - (headEnd + 4);
+          tail = head.slice(headEnd + 4).slice(-5);
         } else {
           bodyBytes += chunk.length;
+          tail = (tail + chunk.subarray(-5).toString("latin1")).slice(-5);
         }
-        if (bodyBytes >= payload.length) received.resolve(bodyBytes);
+        if (chunked ? tail === "0\r\n\r\n" : bodyBytes >= payload.length) received.resolve({ chunked, bodyBytes });
       });
       socket.on("connect", () => {
         socket.write("GET / HTTP/1.1\r\nHost: localhost\r\n\r\n");
@@ -931,14 +942,19 @@ describe("a pull() that outlives its own end() does not hold the request", () =>
         const parked = await ended.promise;
         const pendingWhileParked = server.pendingRequests;
         socket.resume();
-        const length = await received.promise;
+        const body = await received.promise;
         // The drain that completed the response and the release of its request share one event loop turn; the client sees the last bytes in a later one.
-        expect({ parked, pendingWhileParked, length, pendingRequests: server.pendingRequests }).toEqual({
-          parked: true,
-          pendingWhileParked: 1,
-          length: payload.length,
-          pendingRequests: 0,
-        });
+        expect({ parked, pendingWhileParked, ...body, pendingRequests: server.pendingRequests }).toEqual(
+          fromLaterTask
+            ? {
+                parked: false,
+                pendingWhileParked: 0,
+                chunked: true,
+                bodyBytes: payload.length + chunkFraming,
+                pendingRequests: 0,
+              }
+            : { parked: true, pendingWhileParked: 1, chunked: false, bodyBytes: payload.length, pendingRequests: 0 },
+        );
       } finally {
         socket.destroy();
       }