diff --git a/packages/bun-uws/src/HttpContext.h b/packages/bun-uws/src/HttpContext.h index 6e3bd96fe15d..9fa9bef0769a 100644 --- a/packages/bun-uws/src/HttpContext.h +++ b/packages/bun-uws/src/HttpContext.h @@ -472,6 +472,14 @@ struct HttpContext { ((HttpResponse *) s)->resetTimeout(); } + /* Bun.serve: an async handler left the response pending, so any + * pipelined request in this TCP read must be buffered and replayed + * once this response completes (markDone). node:http dispatches + * pipelined requests immediately and queues their responses above. */ + if constexpr (!IsNodeHttp) { + httpResponseData->deferPipeline = !((HttpResponse *) s)->hasResponded(); + } + /* Continue parsing */ return s; @@ -543,6 +551,16 @@ struct HttpContext { httpResponseData->inStream = nullptr; } } + + /* Bun.serve: the handler may have responded synchronously inside + * the body fin callback (e.g. `await req.text()` drains microtasks + * and renders), so re-derive from HTTP_RESPONSE_PENDING here. */ + if constexpr (!IsNodeHttp) { + if (fin) { + httpResponseData->deferPipeline = + (httpResponseData->state & HttpResponseData::HTTP_RESPONSE_PENDING) != 0; + } + } return user; }); @@ -599,6 +617,16 @@ struct HttpContext { } } + /* Bun.serve async pipelining: pipelined request bytes were buffered + * because the current response is still pending. Pause reads so the + * buffer stays bounded by this single recv; replayPipelinedRequests() + * resumes and replays once the response completes. */ + if constexpr (!IsNodeHttp) { + if (!httpResponseData->pipelinedBuffer.empty()) { + ((HttpResponse *) s)->pause(); + } + } + /* Timeout on uncork failure */ auto [written, failed] = ((AsyncSocket *) returnedData)->uncork(); if (written > 0 || failed) { @@ -700,6 +728,15 @@ struct HttpContext { * If write was never called, the developer should still return true so that we may drain. */ bool success = httpResponseData->callOnWritable(reinterpret_cast *>(asyncSocket), httpResponseData->offset); + /* The writable callback may have completed the response and replayed + * a buffered pipelined request whose dispatch closed or adopted this + * socket (parse error, Connection: close, WebSocket upgrade); every + * httpResponseData read below would then be on a destructed object. + * An in-place adopt leaves is_closed false, so also check kind. */ + if (reinterpret_cast *>(s)->isNoLongerHttp()) { + return s; + } + if constexpr (!IsNodeHttp) { /* Bun.serve: onEnd deferred close for a tryEnd tail (offset < total, * nothing in AsyncSocketData::buffer). A retry that moves zero bytes diff --git a/packages/bun-uws/src/HttpParser.h b/packages/bun-uws/src/HttpParser.h index 19d51f3db43a..a6c39bda0597 100644 --- a/packages/bun-uws/src/HttpParser.h +++ b/packages/bun-uws/src/HttpParser.h @@ -589,6 +589,18 @@ struct HttpResponseData; std::string fallback; /* This guy really has only 30 bits since we reserve two highest bits to chunked encoding parsing state */ uint64_t remainingStreamingBytes = 0; + + public: + /* Bun.serve async pipelining: while a response on this connection is + * still in flight (HTTP_RESPONSE_PENDING), the parse loop must stop + * BEFORE getHeaders mutates the next request's bytes; those bytes are + * held in pipelinedBuffer and replayed via replayPipelinedRequests() + * (from the uws_res_end* wrappers) once the + * in-flight response completes. node:http uses its own queue and + * never sets this. */ + bool deferPipeline = false; + std::string pipelinedBuffer; + private: /* node:http compat: a completed request on this connection forbade keep-alive * (Connection: close, or HTTP/1.0), so no further message may be dispatched * (llhttp parses nothing after such a message: HPE_CLOSED_CONNECTION). */ @@ -1102,6 +1114,12 @@ struct HttpResponseData; data[length + 1] = 'a'; /* Anything that is not \n, to trigger "invalid request" */ req->ancientHttp = false; for (;length;) { + /* A response on this connection is still in flight (async handler). + * Stop before getHeaders mutates the next request's bytes so the + * caller can buffer them verbatim for replay. */ + if (deferPipeline) { + break; + } /* node:http server compat: an accepted Upgrade request whose body just * finished parsing switched this connection into tunnel mode (the data * handler set isConnectRequest when it saw the body fin). Everything @@ -1536,7 +1554,16 @@ struct HttpResponseData; length -= consumedBytes; if (length) { - if (length < maxFallbackSize) { + if (deferPipeline) { + /* Pipelined request bytes held until the in-flight response + * completes; replayed via replayPipelinedRequests(). Reads are paused by the + * caller while this buffer is non-empty, so it is bounded by a + * single recv buffer's worth. Reserve post-padding so replay + * can hand this buffer straight back to getHeaders. */ + pipelinedBuffer.reserve(pipelinedBuffer.length() + length + + std::max(MINIMUM_HTTP_POST_PADDING, sizeof(std::string))); + pipelinedBuffer.append(data, length); + } else if (length < maxFallbackSize) { fallback.append(data, length); } else { return HttpParserResult::error(HTTP_ERROR_431_REQUEST_HEADER_FIELDS_TOO_LARGE, HTTP_PARSER_ERROR_REQUEST_HEADER_FIELDS_TOO_LARGE); diff --git a/packages/bun-uws/src/HttpResponse.h b/packages/bun-uws/src/HttpResponse.h index 7155dfd60015..265b0cc542d9 100644 --- a/packages/bun-uws/src/HttpResponse.h +++ b/packages/bun-uws/src/HttpResponse.h @@ -455,6 +455,71 @@ struct HttpResponse : public AsyncSocket { return this; } + /* True once this socket is no longer a valid HttpResponse: closed, shut + * down, or adopted (in-place or relocated) into a WebSocket. An in-place + * adopt leaves is_closed false, so also check the kind byte. */ + bool isNoLongerHttp() { + us_socket_t *s = (us_socket_t *) this; + return us_socket_is_closed(s) || us_socket_is_shut_down(s) + || us_socket_kind(s) != HttpContext::socketKind(); + } + + /* Dispatch pipelined request bytes that were buffered while the previous + * response was still in flight (Bun.serve's async-pipelining path). Called + * as the final action of each uws_res_end* C ABI wrapper: the replayed + * onData can synchronously close or adopt this socket (parse error, + * Connection: close, WebSocket upgrade), destructing HttpResponseData, so + * nothing may touch the response after this returns. node:http never + * buffers here (it dispatches immediately and queues responses), so replay + * is always the IsNodeHttp=false onData. */ + void replayPipelinedRequests() { + /* The caller's own close gate (internalEnd's shouldCloseConnection + * branch) may have already destructed HttpResponseData before we run; + * isNoLongerHttp reads only us_socket_t flags, so check it first. */ + if (isNoLongerHttp()) { + return; + } + HttpResponseData *httpResponseData = getHttpResponseData(); + if (httpResponseData->pipelinedBuffer.empty()) { + return; + } + /* Re-entering onData from inside onData would stomp the per-context + * isParsingHttp/upgradedWebSocket state; the pathological case is two + * connections whose handlers synchronously resolve each other. Leave + * the buffer and reads paused; the connection idles out rather than + * corrupting the outer parse. */ + HttpContextData *httpContextData = HttpContext::getSocketContextDataS((us_socket_t *) this); + if (httpContextData->flags.isParsingHttp) { + return; + } + /* Flush the just-completed response before dispatching the next + * request: onData's parse-error path closes with uncorkWithoutSending, + * which would otherwise drop the corked bytes. */ + Super::uncork(); + /* Connection is being torn down after this response; the buffered + * request is discarded. On the corked async path internalEnd()'s own + * close gate was skipped (isCorked() was true, then its uncork() + * released the slot so HttpResponse::cork() early-returns), so close + * here rather than leave the socket open until client FIN/timeout. */ + if (httpResponseData->shouldCloseConnection()) { + httpResponseData->pipelinedBuffer.clear(); + if (((AsyncSocket *) this)->hasFullyDrained()) { + ((AsyncSocket *) this)->shutdown(); + ((AsyncSocket *) this)->close(); + } + return; + } + + std::string buffer = std::move(httpResponseData->pipelinedBuffer); + httpResponseData->pipelinedBuffer.clear(); + /* Reads were paused when the buffer became non-empty. */ + this->resume(); + /* getHeaders writes post-padding bytes; the buffer reserved them at + * append time (see consumePostPadded). */ + buffer.reserve(buffer.length() + MINIMUM_HTTP_POST_PADDING); + HttpContext::template onData((us_socket_t *) this, buffer.data(), (int) buffer.length()); + } + /* Note: Headers are not checked in regards to timeout. * We only check when you actively push data or end the request */ diff --git a/packages/bun-uws/src/HttpResponseData.h b/packages/bun-uws/src/HttpResponseData.h index c27b94add094..60fcb4540f77 100644 --- a/packages/bun-uws/src/HttpResponseData.h +++ b/packages/bun-uws/src/HttpResponseData.h @@ -57,26 +57,48 @@ struct HttpResponseData : AsyncSocketData, HttpParser { /* We are done with this request */ this->state &= ~HttpResponseData::HTTP_RESPONSE_PENDING; + /* The parse loop may dispatch the next request on this connection. */ + this->deferPipeline = false; HttpResponseData *httpResponseData = uwsRes->getHttpResponseData(); httpResponseData->isIdle = true; + + /* Pipelined request bytes buffered while this response was in flight are + * NOT replayed here: every caller (internalEnd, uws_res_end_sendfile, + * uws_res_end_without_body) still reads our state afterwards, and a + * replay can close/adopt the socket and destruct this object. Callers + * invoke replayPipelinedRequests() themselves as their final action. */ } /* Caller of onWritable. It is possible onWritable calls markDone so we need to borrow it. */ bool callOnWritable(uWS::HttpResponse* response, uint64_t offset) { /* Borrow real onWritable */ - auto* borrowedOnWritable = std::move(onWritable); + auto* borrowedOnWritable = onWritable; - /* Set onWritable to placeholder */ - onWritable = [](uWS::HttpResponse*, uint64_t, void*) {return true;}; + /* Set onWritable to a placeholder we can identify afterwards: the + * borrowed callback may reach markDone() and replay the next pipelined + * request, whose handler can install its OWN onWritable. Restoring by + * non-null would stomp that with this response's stale callback. */ + static constexpr OnWritableCallback callOnWritablePlaceholder = + [](uWS::HttpResponse*, uint64_t, void*) { return true; }; + onWritable = callOnWritablePlaceholder; /* Run borrowed onWritable */ bool ret = borrowedOnWritable(response, offset, writableUserData); - /* If we still have onWritable (the placeholder) then move back the real one */ - if (onWritable) { - /* We haven't reset onWritable, so give it back */ - onWritable = std::move(borrowedOnWritable); + /* The callback may have completed the response and replayed a buffered + * pipelined request whose dispatch closed or adopted this socket, + * destructing us; every field access below would then be on a dead + * object (or on WebSocketData after an in-place adopt). */ + if (response->isNoLongerHttp()) { + return ret; + } + + /* Only restore if onWritable is still OUR placeholder; anything else + * (null from markDone, or a new callback from a replayed request) + * belongs to whoever set it. */ + if (onWritable == callOnWritablePlaceholder) { + onWritable = borrowedOnWritable; } return ret; diff --git a/src/runtime/server/FileRoute.rs b/src/runtime/server/FileRoute.rs index 88691374f5f4..fda15a9b99e7 100644 --- a/src/runtime/server/FileRoute.rs +++ b/src/runtime/server/FileRoute.rs @@ -604,9 +604,12 @@ impl FileRoute { } fn on_response_complete(this: *mut FileRoute, resp: AnyResponse) { - resp.clear_aborted(); - resp.clear_on_writable(); - resp.clear_timeout(); + // See StaticRoute::on_response_complete. + if let AnyResponse::H3(_) = resp { + resp.clear_aborted(); + resp.clear_on_writable(); + resp.clear_timeout(); + } // SAFETY: `this` is live (ref held by caller); `deref()` may free it. unsafe { if let Some(mut server) = (*this).server.get() { diff --git a/src/runtime/server/RequestContext.rs b/src/runtime/server/RequestContext.rs index f85ad8e6a6a9..d9aeaae5f811 100644 --- a/src/runtime/server/RequestContext.rs +++ b/src/runtime/server/RequestContext.rs @@ -1096,7 +1096,7 @@ where }; if try_end_ok { drop(bb); - self.detach_response(); + self.detach_response_after_end(); self.end_request_streaming_and_drain(); self.finalize_without_deinit(); self.deref(); @@ -1556,7 +1556,7 @@ where fn on_file_stream_complete(ctx: *mut c_void, _resp: uws::AnyResponse) { // SAFETY: ctx is a *RequestContext registered with FileResponseStream let this: &mut Self = unsafe { bun_ptr::callback_ctx::(ctx) }; - this.detach_response(); + this.detach_response_after_end(); this.end_request_streaming_and_drain(); this.deref(); } @@ -1638,7 +1638,7 @@ where let bytes = &bytes_[bytes_.len().min(write_offset)..]; // SAFETY: FFI handle if resp.try_end(bytes, bytes_.len(), self.should_close_connection()) { - self.detach_response(); + self.detach_response_after_end(); self.end_request_streaming_and_drain(); self.deref(); true @@ -1673,7 +1673,7 @@ where let done = resp.try_end(bytes, total_len, close_connection); if done { self.response_buf_owned.clear(); - self.detach_response(); + self.detach_response_after_end(); self.end_request_streaming_and_drain(); self.deref(); } else { @@ -2389,6 +2389,24 @@ where } } + /// `detach_response` without the HTTP/1 `resp.clear_*` FFI calls: H1's + /// `markDone()` already nulled them, and the `uws_res_end*` wrapper then + /// replayed any buffered pipelined request, so those handlers now belong + /// to the next request. HTTP/3's `markDone()` deliberately leaves + /// `onAborted` armed for `on_stream_close`, so fall through to the full + /// detach there. + fn detach_response_after_end(&mut self) { + if HTTP3 { + self.detach_response(); + return; + } + self.request_body_buf = Vec::new(); + self.resp.take(); + self.flags.set_is_waiting_for_request_body(false); + self.flags.set_has_abort_handler(false); + self.flags.set_has_timeout_handler(false); + } + pub fn is_aborted_or_ended(&self) -> bool { // resp == null or aborted or server.stop(true) self.resp.is_none() @@ -3807,7 +3825,7 @@ where return; } } - self.detach_response(); + self.detach_response_after_end(); self.end_request_streaming_and_drain(); self.deref(); } diff --git a/src/runtime/server/StaticRoute.rs b/src/runtime/server/StaticRoute.rs index 36c2d66b857e..4e9139238c0c 100644 --- a/src/runtime/server/StaticRoute.rs +++ b/src/runtime/server/StaticRoute.rs @@ -402,11 +402,17 @@ impl StaticRoute { /// `this` must be a live heap-allocated route with write provenance; may free /// `*this` via `deref_` when the refcount reaches zero. unsafe fn on_response_complete(this: *mut Self, resp: AnyResponse) { - // SAFETY: caller contract. - unsafe { + // HTTP/1: markDone() already nulled these and the wrapper then replayed + // any buffered pipelined request, so clearing would null THAT request's + // handlers. HTTP/3: Http3Response::markDone() leaves onAborted armed + // for on_stream_close, so clear it to avoid a second deref_ here. + if let AnyResponse::H3(_) = resp { resp.clear_aborted(); resp.clear_on_writable(); resp.clear_timeout(); + } + // SAFETY: caller contract. + unsafe { if let Some(mut server) = (*this).server.get() { server.on_static_request_complete(); } diff --git a/src/runtime/webcore/streams.rs b/src/runtime/webcore/streams.rs index d4dfdc47f1cf..a95c060b0bb1 100644 --- a/src/runtime/webcore/streams.rs +++ b/src/runtime/webcore/streams.rs @@ -1402,11 +1402,19 @@ impl HTTPServerWritable { total_written = chunk_len as u64; if self.requested_end { - if let Some(res) = self.any_res() { - res.clear_on_writable(); + // HTTP/1: `send_readable` drained the parked `try_end`/`end`, + // so `markDone()` nulled our `onWritable` and the wrapper then + // replayed any buffered pipelined request; the socket now + // belongs to THAT request. HTTP/3 has no replay. + if HTTP3 { + if let Some(res) = self.any_res() { + res.clear_on_writable(); + } + } else { + // finalize()'s `if !self.done` block would clear_on_writable + // and end_stream() the replayed request's state. + self.done = true; } - // `send_readable` drained the parked `try_end`, so uWS has - // `markDone()`d the response and dropped its `onAborted`. self.ended_response = true; self.signal.close(None); let _ = self.flush_promise(); // TODO: properly propagate exception upwards diff --git a/src/uws_sys/libuwsockets.cpp b/src/uws_sys/libuwsockets.cpp index e0525a110310..26805a4c73e4 100644 --- a/src/uws_sys/libuwsockets.cpp +++ b/src/uws_sys/libuwsockets.cpp @@ -1158,12 +1158,14 @@ extern "C" uWS::HttpResponse *uwsRes = (uWS::HttpResponse *)res; uwsRes->clearOnWritableAndAborted(); uwsRes->end(stringViewFromC(data, length), close_connection); + uwsRes->replayPipelinedRequests(); } else { uWS::HttpResponse *uwsRes = (uWS::HttpResponse *)res; uwsRes->clearOnWritableAndAborted(); uwsRes->end(stringViewFromC(data, length), close_connection); + uwsRes->replayPipelinedRequests(); } } @@ -1174,12 +1176,14 @@ extern "C" uWS::HttpResponse *uwsRes = (uWS::HttpResponse *)res; uwsRes->clearOnWritableAndAborted(); uwsRes->sendTerminatingChunk(close_connection); + uwsRes->replayPipelinedRequests(); } else { uWS::HttpResponse *uwsRes = (uWS::HttpResponse *)res; uwsRes->clearOnWritableAndAborted(); uwsRes->sendTerminatingChunk(close_connection); + uwsRes->replayPipelinedRequests(); } } @@ -1327,6 +1331,7 @@ extern "C" data->state |= uWS::HttpResponseData::HTTP_END_CALLED; data->markDone(uwsRes); uwsRes->resetTimeout(); + uwsRes->replayPipelinedRequests(); } else { @@ -1336,6 +1341,7 @@ extern "C" data->state |= uWS::HttpResponseData::HTTP_END_CALLED; data->markDone(uwsRes); uwsRes->resetTimeout(); + uwsRes->replayPipelinedRequests(); } } void uws_res_reset_timeout(int ssl, uws_res_r res) { @@ -1378,6 +1384,7 @@ extern "C" data->state |= uWS::HttpResponseData::HTTP_END_CALLED; data->markDone(uwsRes); uwsRes->resetTimeout(); + uwsRes->replayPipelinedRequests(); } else { @@ -1400,6 +1407,7 @@ extern "C" data->state |= uWS::HttpResponseData::HTTP_END_CALLED; data->markDone(uwsRes); uwsRes->resetTimeout(); + uwsRes->replayPipelinedRequests(); } } @@ -1837,7 +1845,9 @@ __attribute__((callback (corker, ctx))) if (pair.first) { uwsRes->clearOnWritableAndAborted(); } - + if (pair.second) { + uwsRes->replayPipelinedRequests(); + } return pair.first; } else @@ -1847,7 +1857,9 @@ __attribute__((callback (corker, ctx))) if (pair.first) { uwsRes->clearOnWritableAndAborted(); } - + if (pair.second) { + uwsRes->replayPipelinedRequests(); + } return pair.first; } } diff --git a/test/js/bun/http/serve.test.ts b/test/js/bun/http/serve.test.ts index d6534b96eae8..f8ab0090c970 100644 --- a/test/js/bun/http/serve.test.ts +++ b/test/js/bun/http/serve.test.ts @@ -3054,6 +3054,199 @@ server.listen(0, "127.0.0.1", () => { expect(exitCode).toBe(0); }); +// A pipelined request that arrives while the previous response is still in flight +// (async handler) must be buffered and dispatched once that response completes, +// not abort the connection. Previously uWS closed the socket the moment it saw a +// second request head while HTTP_RESPONSE_PENDING was set, so the first response +// was discarded and the second never dispatched. +describe("dispatches a pipelined request after the previous async response completes", () => { + // Count HTTP/1.1 responses in the raw wire bytes and pull each body by + // Content-Length so back-to-back responses with no interleaving newline + // (body "A" immediately followed by "HTTP/1.1 ...") are split correctly. + const wireResponses = (raw: string) => { + const out: { status: string; body: string }[] = []; + while (raw.startsWith("HTTP/1.1 ")) { + const headEnd = raw.indexOf("\r\n\r\n"); + const head = raw.slice(0, headEnd); + const cl = Number(/content-length: (\d+)/i.exec(head)?.[1] ?? 0); + out.push({ status: head.slice(0, head.indexOf("\r\n")), body: raw.slice(headEnd + 4, headEnd + 4 + cl) }); + raw = raw.slice(headEnd + 4 + cl); + } + return out; + }; + + const readPipeline = (port: number, payload: string, expected: number) => + new Promise<{ responses: { status: string; body: string }[] }>(resolve => { + let raw = ""; + const s = net.connect(port, "127.0.0.1", () => s.write(payload)); + s.on("data", d => { + raw += d.toString("latin1"); + // Resolve as soon as the expected number of responses have arrived, so the + // test waits on the condition rather than a timer. + if (wireResponses(raw).length >= expected) { + s.destroy(); + resolve({ responses: wireResponses(raw) }); + } + }); + s.on("error", () => {}); + s.on("close", () => resolve({ responses: wireResponses(raw) })); + }); + + it("request body forwarded to fetch(), pipelined GET behind it", async () => { + const events: string[] = []; + // Upstream that the relay forwards the body to; its result is irrelevant to + // the bug, but a real server avoids depending on connect-refused timing. + await using upstream = Bun.serve({ + port: 0, + hostname: "127.0.0.1", + async fetch(req) { + return new Response(await req.text()); + }, + }); + await using relay = Bun.serve({ + port: 0, + hostname: "127.0.0.1", + async fetch(req) { + const path = new URL(req.url).pathname; + events.push("handler " + path); + if (path === "/b") return new Response("B"); + const up = await fetch(upstream.url, { method: "POST", body: req.body, duplex: "half" }); + events.push("done " + path + " " + (await up.text())); + return new Response("A"); + }, + }); + + const { responses } = await readPipeline( + relay.port, + "POST /a HTTP/1.1\r\nHost: x\r\nContent-Length: 4\r\n\r\nabcd" + "GET /b HTTP/1.1\r\nHost: x\r\n\r\n", + 2, + ); + expect({ responses, events }).toEqual({ + responses: [ + { status: "HTTP/1.1 200 OK", body: "A" }, + { status: "HTTP/1.1 200 OK", body: "B" }, + ], + events: ["handler /a", "done /a abcd", "handler /b"], + }); + }); + + it("three async GETs pipelined in one write, served in order", async () => { + const events: string[] = []; + await using server = Bun.serve({ + port: 0, + hostname: "127.0.0.1", + async fetch(req) { + const path = new URL(req.url).pathname; + events.push("in " + path); + await Bun.sleep(10); + events.push("out " + path); + return new Response(path.slice(1)); + }, + }); + + const { responses } = await readPipeline( + server.port, + "GET /a HTTP/1.1\r\nHost: x\r\n\r\n" + + "GET /b HTTP/1.1\r\nHost: x\r\n\r\n" + + "GET /c HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n", + 3, + ); + // Each request is dispatched only after the previous response is written, so + // responses (and handler entry/exit) are strictly ordered. + expect({ responses, events }).toEqual({ + responses: [ + { status: "HTTP/1.1 200 OK", body: "a" }, + { status: "HTTP/1.1 200 OK", body: "b" }, + { status: "HTTP/1.1 200 OK", body: "c" }, + ], + events: ["in /a", "out /a", "in /b", "out /b", "in /c", "out /c"], + }); + }); + + it("pipelined request sent in a separate write while the first is pending", async () => { + const inflight = Promise.withResolvers(); + const release = Promise.withResolvers(); + await using server = Bun.serve({ + port: 0, + hostname: "127.0.0.1", + async fetch(req) { + const path = new URL(req.url).pathname; + if (path === "/b") return new Response("B"); + inflight.resolve(); + await release.promise; + return new Response("A"); + }, + }); + + const raw = await new Promise(resolve => { + let got = ""; + const s = net.connect(server.port, "127.0.0.1", () => { + s.write("GET /a HTTP/1.1\r\nHost: x\r\n\r\n"); + }); + s.on("data", d => { + got += d.toString("latin1"); + if (wireResponses(got).length >= 2) { + s.destroy(); + resolve(got); + } + }); + s.on("error", () => {}); + s.on("close", () => resolve(got)); + inflight.promise.then(async () => { + // /a's handler is parked; /b arrives now in its own TCP segment and must + // be buffered (not abort the connection) until /a responds. A fetch on a + // second connection round-trips through the same event loop poll, so once + // it resolves the server has definitely read /b from C1. + s.write("GET /b HTTP/1.1\r\nHost: x\r\n\r\n"); + await fetch(`http://127.0.0.1:${server.port}/b`).then(r => r.text()); + release.resolve(); + }); + }); + + expect(wireResponses(raw)).toEqual([ + { status: "HTTP/1.1 200 OK", body: "A" }, + { status: "HTTP/1.1 200 OK", body: "B" }, + ]); + }); + + it("delivers the first response when the pipelined bytes are a parse error", async () => { + // Replay runs onData on the buffered bytes after the first response + // completes; a parse error there closes the socket (writes 505, then + // us_socket_close). The first response must already be on the wire, and + // the post-markDone state reads in internalEnd must not touch a + // destructed HttpResponseData. + await using server = Bun.serve({ + port: 0, + hostname: "127.0.0.1", + async fetch() { + await new Promise(r => setImmediate(r)); + return new Response("A"); + }, + }); + + const raw = await new Promise(resolve => { + let got = ""; + const s = net.connect(server.port, "127.0.0.1", () => + s.write("GET /a HTTP/1.1\r\nHost: x\r\n\r\nGET / HTTP/9.9\r\nHost: x\r\n\r\n"), + ); + s.on("data", d => (got += d.toString("latin1"))); + s.on("error", () => {}); + s.on("close", () => resolve(got)); + }); + + const first = wireResponses(raw); + const headEnd = raw.indexOf("\r\n\r\n"); + const afterFirst = raw.slice(headEnd + 4 + (first[0]?.body.length ?? 0)); + expect({ + first: first[0], + followedBy505: afterFirst.startsWith("HTTP/1.1 505 "), + }).toEqual({ + first: { status: "HTTP/1.1 200 OK", body: "A" }, + followedBy505: true, + }); + }); +}); + it("only serves /bun:info to loopback clients in development mode", async () => { using server = Bun.serve({ port: 0,