From 6094a6a6922b2c681497a6e4b50e9cb7e7b60242 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Wed, 7 Oct 2026 06:09:07 +0000 Subject: [PATCH 1/6] Bun.serve: end a server.timeout(req, N) override with its response On HTTP/1 one per-socket byte held both the server's idleTimeout and the value of server.timeout(req, N). Nothing restored the server's value when the response completed, so the value of a finished request ruled the kept connection: server.timeout(req, 0) left a connection the server never closed, and any other N replaced idleTimeout until the next request. The server's value now lives on the uWS HTTP context (HttpContextData::idleTimeout, set once at listen). Every response starts under it: resetResponseState() copies it into the per-socket byte, now named idleTimeoutInForce. HttpResponse::setTimeout() is the override only. It tags the response (HTTP_TIMEOUT_OVERRIDDEN) and uWS goes back to the context's value when that response is complete and the socket has drained (markDone, or onWritable for a response that ended under backpressure). The routes no longer carry the value. They arm the timer (reset_timeout / start_timeout), so FileResponseStream no longer writes the server's value over an override while a Bun.file body is in flight, on HTTP/1 and on HTTP/2. The docs and the bun-types JSDoc now say that idleTimeout also bounds an idle keep-alive connection, and that server.timeout() lasts until the response to its request is complete. --- docs/runtime/http/server.mdx | 2 + packages/bun-types/serve.d.ts | 6 + packages/bun-uws/src/App.h | 5 + packages/bun-uws/src/HttpContext.h | 8 +- packages/bun-uws/src/HttpContextData.h | 4 + packages/bun-uws/src/HttpResponse.h | 34 +- packages/bun-uws/src/HttpResponseData.h | 18 +- .../bindings/node/JSNodeHTTPServerSocket.cpp | 2 +- src/runtime/server/DirectoryRoute.rs | 3 +- src/runtime/server/FileResponseStream.rs | 9 +- src/runtime/server/FileRoute.rs | 3 +- src/runtime/server/RequestContext.rs | 1 - src/runtime/server/StaticRoute.rs | 10 +- src/runtime/server/mod.rs | 5 +- src/runtime/server/server_body.rs | 17 +- src/uws_sys/App.rs | 6 + src/uws_sys/Response.rs | 15 +- src/uws_sys/h2.rs | 5 + src/uws_sys/libuwsockets.cpp | 9 + src/uws_sys/libuwsockets_h2.cpp | 6 + test/js/bun/http/serve-http2-fixture.ts | 4 + .../js/bun/http/serve-http2-lifecycle.test.ts | 69 +- test/js/bun/http/serve.test.ts | 589 +++++++++++++++++- 23 files changed, 788 insertions(+), 42 deletions(-) diff --git a/docs/runtime/http/server.mdx b/docs/runtime/http/server.mdx index 2c375f63faeb..1bfa3af492f2 100644 --- a/docs/runtime/http/server.mdx +++ b/docs/runtime/http/server.mdx @@ -259,6 +259,8 @@ Bun.serve({ }); ``` +`idleTimeout` is also how long Bun keeps an idle keep-alive connection open between requests. Behind a load balancer or a reverse proxy, set `idleTimeout` above the keep-alive time the proxy uses for its upstream connections, or set it to `0`. [`server.timeout(req, seconds)`](#server-timeout-request-seconds) lasts until the response to that request is complete. After that, the connection is under `idleTimeout` again. + **Streaming & Server-Sent Events** — The idle timer applies while a response is being streamed. If your stream goes quiet for longer than `idleTimeout`, Bun closes the connection mid-response. For long-lived streams, disable the diff --git a/packages/bun-types/serve.d.ts b/packages/bun-types/serve.d.ts index 356c15491724..412682593233 100644 --- a/packages/bun-types/serve.d.ts +++ b/packages/bun-types/serve.d.ts @@ -872,6 +872,9 @@ declare module "bun" { * Sets the number of seconds to wait before timing out a connection * due to inactivity. * + * This is also how long an idle keep-alive connection stays open + * between requests. `0` disables the timeout. The maximum is `255`. + * * @default 10 */ idleTimeout?: number; @@ -1139,6 +1142,9 @@ declare module "bun" { /** * Reset the idle timeout of the given Request to the given number of seconds. `0` means no timeout. * + * The value lasts until the response to that request is complete. After + * that, the connection is under `idleTimeout` again. + * * @example * ```js * export default { diff --git a/packages/bun-uws/src/App.h b/packages/bun-uws/src/App.h index 6396a840d0d3..01ecd0b1b5b8 100644 --- a/packages/bun-uws/src/App.h +++ b/packages/bun-uws/src/App.h @@ -803,6 +803,11 @@ struct TemplatedApp { return std::move(*this); } + TemplatedApp &&setIdleTimeout(uint8_t seconds) { + httpContext->getSocketContextData()->idleTimeout = seconds; + return std::move(*this); + } + TemplatedApp &&setMaxHeadersCount(uint32_t maxHeadersCount) { httpContext->getSocketContextData()->maxHeadersCount = maxHeadersCount; return std::move(*this); diff --git a/packages/bun-uws/src/HttpContext.h b/packages/bun-uws/src/HttpContext.h index 9c041bdf611c..6b8e509d16c4 100644 --- a/packages/bun-uws/src/HttpContext.h +++ b/packages/bun-uws/src/HttpContext.h @@ -536,7 +536,7 @@ struct HttpContext { * response's per-request framing bits (204/304, close-delimited, * trailers), which writeHead only ever sets: a stale one would * strip the next response's body framing. */ - httpResponseData->resetResponseState(); + httpResponseData->resetResponseState(httpContextData->idleTimeout); /* An ancient (HTTP/1.0) request gets no keep-alive and no chunked * framing; so does an explicit `Connection: close`. */ @@ -667,7 +667,7 @@ struct HttpContext { /* We still have some more data coming in later, so reset timeout */ /* Only reset timeout if we got enough bytes (16kb/sec) since last time we reset here */ httpResponseData->received_bytes_per_timeout += (unsigned int) data.length(); - if (httpResponseData->received_bytes_per_timeout >= HTTP_RECEIVE_THROUGHPUT_BYTES * httpResponseData->idleTimeout) { + if (httpResponseData->received_bytes_per_timeout >= HTTP_RECEIVE_THROUGHPUT_BYTES * httpResponseData->idleTimeoutInForce) { ((HttpResponse *) user)->resetTimeout(); httpResponseData->received_bytes_per_timeout = 0; } @@ -967,6 +967,10 @@ struct HttpContext { /* Drain any socket buffer, this might empty our backpressure and thus finish the request */ asyncSocket->flush(); + /* The tail of a response that ended under backpressure may be out now; + * the resetTimeout() below arms what this leaves. */ + reinterpret_cast *>(s)->endTimeoutOverrideIfDrained(); + /* node:http compat: reads were paused while pipelined responses were * queued and stayed paused because the socket still had outgoing * backpressure when the queue drained; now that it has flushed, read diff --git a/packages/bun-uws/src/HttpContextData.h b/packages/bun-uws/src/HttpContextData.h index 4c28067a9c8c..40b296affa59 100644 --- a/packages/bun-uws/src/HttpContextData.h +++ b/packages/bun-uws/src/HttpContextData.h @@ -114,6 +114,10 @@ struct alignas(16) HttpContextData { public: HttpFlags flags; + /* The server's idle timeout in seconds, 0 = never. Every response starts + * under it (resetResponseState); HttpResponse::setTimeout() replaces it + * for one response. */ + uint8_t idleTimeout = 10; bool isParsing(struct us_socket_t *s) const { return flags.isParsingHttp && parsingSocket == s; } }; diff --git a/packages/bun-uws/src/HttpResponse.h b/packages/bun-uws/src/HttpResponse.h index 89d1d6f762ac..8e42b5b68f6f 100644 --- a/packages/bun-uws/src/HttpResponse.h +++ b/packages/bun-uws/src/HttpResponse.h @@ -53,16 +53,36 @@ struct HttpResponse : public AsyncSocket { return (HttpResponseData *) Super::getAsyncSocketData(); } + /* Replaces the server's idle timeout (HttpContextData::idleTimeout) for + * the response in flight, and arms it. */ void setTimeout(uint8_t seconds) { auto* data = getHttpResponseData(); - data->idleTimeout = seconds; - Super::timeout(data->idleTimeout); + if (!(data->state & HttpResponseData::HTTP_RESPONSE_PENDING)) { + return; + } + data->state |= HttpResponseData::HTTP_TIMEOUT_OVERRIDDEN; + data->idleTimeoutInForce = seconds; + Super::timeout(data->idleTimeoutInForce); + } + + /* A setTimeout() override ends when its response is complete and + * hasFullyDrained(): a response that ended under backpressure keeps it + * until onWritable has drained the tail. Bytes still in the cork buffer + * do not count. Does not arm. */ + void endTimeoutOverrideIfDrained() { + auto* data = getHttpResponseData(); + if ((data->state & (HttpResponseData::HTTP_TIMEOUT_OVERRIDDEN | HttpResponseData::HTTP_RESPONSE_PENDING)) == HttpResponseData::HTTP_TIMEOUT_OVERRIDDEN) [[unlikely]] { + if (Super::hasFullyDrained()) { + data->state &= ~HttpResponseData::HTTP_TIMEOUT_OVERRIDDEN; + data->idleTimeoutInForce = HttpContext::getSocketContextDataS((us_socket_t *) this)->idleTimeout; + } + } } void resetTimeout() { auto* data = getHttpResponseData(); - Super::timeout(data->idleTimeout); + Super::timeout(data->idleTimeoutInForce); } /* The chunk-size line of a chunk. Returns its length. */ static constexpr size_t CHUNK_HEAD_MAX = 10; @@ -361,17 +381,21 @@ struct HttpResponse : public AsyncSocket { /* Success is when we wrote the entire thing without any failures */ bool success = written == data.length() && !failed; - /* Reset the timeout on each tryEnd */ - this->resetTimeout(); /* Remove onAborted function if we reach the end */ if (httpResponseData->offset == totalSize) { httpResponseData->markDone(this); + /* After markDone(), which can end a setTimeout() override, and + * before the close gate, which can destruct the ext. */ + this->resetTimeout(); /* We need to check if we should close this socket here now */ if (uncorkCompletedResponse()) { closeIfDoneAndMarked(httpResponseData); } + } else { + /* Reset the timeout on each tryEnd */ + this->resetTimeout(); } return success; diff --git a/packages/bun-uws/src/HttpResponseData.h b/packages/bun-uws/src/HttpResponseData.h index 454793c07481..7a859130ca77 100644 --- a/packages/bun-uws/src/HttpResponseData.h +++ b/packages/bun-uws/src/HttpResponseData.h @@ -58,6 +58,8 @@ struct HttpResponseData : AsyncSocketData, HttpParser { /* We are done with this request */ this->state &= ~HttpResponseData::HTTP_RESPONSE_PENDING; + /* Every caller arms the timer after markDone(), so it arms the value this leaves. */ + uwsRes->endTimeoutOverrideIfDrained(); HttpResponseData *httpResponseData = uwsRes->getHttpResponseData(); /* A queued pipelined response (node:http) still owes output on this @@ -169,6 +171,10 @@ struct HttpResponseData : AsyncSocketData, HttpParser { /* node:http: the peer sent its FIN first (HTTP_NODE_RECEIVED_FIN only covers a * deferred close). onSocketClosed reports it so the JS socket emits 'end'. */ HTTP_NODE_PEER_ENDED = 1 << 22, + /* HttpResponse::setTimeout() replaced the server's idle timeout in + * idleTimeoutInForce for the response in flight. Not derivable from the + * byte: before the first request it is 10, whatever the server's value. */ + HTTP_TIMEOUT_OVERRIDDEN = 1 << 27, /* Bits that describe the connection rather than the response in flight. * There is one HttpResponseData per socket, reused by every request on a @@ -185,9 +191,13 @@ struct HttpResponseData : AsyncSocketData, HttpParser { /* Begin a new response on this connection. Clearing the word in one go is * what keeps a 204/304 (HTTP_NO_BODY_STATUS), a close-delimited body or a * previous response's trailers from leaking into the next request on a - * keep-alive socket; only the connection-scoped bits are carried over. */ - void resetResponseState() { + * keep-alive socket; only the connection-scoped bits are carried over. + * The idle timeout starts over as well: serverIdleTimeout is + * HttpContextData::idleTimeout, so a setTimeout() override that its own + * response could not hand back yet ends here. */ + void resetResponseState(uint8_t serverIdleTimeout) { state = (state & HTTP_CONNECTION_SCOPED) | HTTP_RESPONSE_PENDING; + idleTimeoutInForce = serverIdleTimeout; /* A response is in flight again (a new request dispatched, or a queued * pipelined response activated), so the connection is not idle. */ this->isIdle = false; @@ -226,7 +236,9 @@ struct HttpResponseData : AsyncSocketData, HttpParser { /* Current state (content-length sent, status sent, write called, etc */ uint32_t state = 0; - uint8_t idleTimeout = 10; // default HTTP_TIMEOUT 10 seconds + /* Seconds resetTimeout() arms: 10 until the first request, then the + * server's value (resetResponseState) or a setTimeout() override. */ + uint8_t idleTimeoutInForce = 10; /* The parser writes this through a bool& (getHeaders / consumePostPadded), * so it cannot live in `state`. */ bool isConnectRequest = false; diff --git a/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp b/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp index 67990a061c7f..ca8bb943e6bc 100644 --- a/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp +++ b/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp @@ -662,7 +662,7 @@ static void startPipelinedResponseImpl(us_socket_t* socket, bool isAncient, bool httpResponseData->offset = 0; // Clears the finished response's framing bits and keeps the connection-scoped // ones (notably HTTP_NODE_READS_PAUSED, read again below). - httpResponseData->resetResponseState(); + httpResponseData->resetResponseState(uWS::HttpContext::getSocketContextDataS(socket)->idleTimeout); if (connectionClose) { httpResponseData->state |= uWS::HttpResponseData::HTTP_CONNECTION_CLOSE; } diff --git a/src/runtime/server/DirectoryRoute.rs b/src/runtime/server/DirectoryRoute.rs index 2b322ff2751b..be0f238de13f 100644 --- a/src/runtime/server/DirectoryRoute.rs +++ b/src/runtime/server/DirectoryRoute.rs @@ -118,7 +118,7 @@ impl DirectoryRoute { }; if let Some(mut server) = this.server.get() { server.on_pending_request(); - resp.timeout(server.config().idle_timeout); + resp.start_timeout(); } let mut path_buf = bun_paths::path_buffer_pool::get(); @@ -248,7 +248,6 @@ impl DirectoryRoute { pollable: false, offset: body_offset, length: Some(body_len), - idle_timeout: server.config().idle_timeout, owner: StreamOwner::DirectoryRoute(guard.into_route()), }); } diff --git a/src/runtime/server/FileResponseStream.rs b/src/runtime/server/FileResponseStream.rs index 6c092fcfb2c0..bfb31eafb1ff 100644 --- a/src/runtime/server/FileResponseStream.rs +++ b/src/runtime/server/FileResponseStream.rs @@ -39,7 +39,6 @@ pub(crate) struct FileResponseStream { event_loop_handle: Cell, fd: Cell, auto_close: Cell, - idle_timeout: Cell, /// Taken by whichever of complete / abort / error fires first. owner: Cell>, @@ -109,7 +108,6 @@ pub(crate) struct StartOptions { /// Maximum bytes to send; `None` reads to EOF. For regular files this /// should be `stat.size - offset` (after Range/slice clamping). pub length: Option, - pub idle_timeout: u8, pub owner: StreamOwner, } @@ -171,7 +169,6 @@ impl FileResponseStream { )), fd: Cell::new(opts.fd), auto_close: Cell::new(opts.auto_close), - idle_timeout: Cell::new(opts.idle_timeout), owner: Cell::new(Some(opts.owner)), mode: Cell::new(if use_sendfile { Mode::Sendfile @@ -189,7 +186,7 @@ impl FileResponseStream { let this_ref = unsafe { &*this }; let resp = this_ref.resp.get(); - resp.timeout(opts.idle_timeout); + resp.reset_timeout(); resp.on_aborted( |p: *mut FileResponseStream, r| { // SAFETY: uWS hands back the userdata pointer set below; the @@ -313,7 +310,7 @@ impl FileResponseStream { } let resp = self.resp.get(); - resp.timeout(self.idle_timeout.get()); + resp.reset_timeout(); if state == ReadState::Eof { self.insert_state(State::RESPONSE_DONE); @@ -390,7 +387,7 @@ impl FileResponseStream { self.finish(); return true; } - self.resp.get().timeout(self.idle_timeout.get()); + self.resp.get().reset_timeout(); self.hold_read_ref(); // A paused POSIX reader ignores `read()`. self.reader_mut().unpause(); diff --git a/src/runtime/server/FileRoute.rs b/src/runtime/server/FileRoute.rs index 09dad42332e9..498bf880cd01 100644 --- a/src/runtime/server/FileRoute.rs +++ b/src/runtime/server/FileRoute.rs @@ -246,7 +246,7 @@ impl FileRoute { let route = RefPtr::from_this(this); if let Some(mut server) = route.server.get() { server.on_pending_request(); - resp.timeout(server.config().idle_timeout); + resp.start_timeout(); } let store = route.blob.store().unwrap().clone(); let Some(path) = store.get_path() else { @@ -310,7 +310,6 @@ impl FileRoute { pollable, offset, length, - idle_timeout: server.config().idle_timeout, owner: StreamOwner::FileRoute(route), }); } diff --git a/src/runtime/server/RequestContext.rs b/src/runtime/server/RequestContext.rs index 813b8502a282..22893d090060 100644 --- a/src/runtime/server/RequestContext.rs +++ b/src/runtime/server/RequestContext.rs @@ -2032,7 +2032,6 @@ where } else { None }, - idle_timeout: server.config().idle_timeout, owner: file_response_stream::StreamOwner::Ctx { ctx: self.as_ctx_ptr().cast::(), on_complete: Self::on_file_stream_complete, diff --git a/src/runtime/server/StaticRoute.rs b/src/runtime/server/StaticRoute.rs index cae0a042df00..72736a744967 100644 --- a/src/runtime/server/StaticRoute.rs +++ b/src/runtime/server/StaticRoute.rs @@ -265,7 +265,7 @@ impl StaticRoute { Self::retain_for_response(this); if let Some(mut server) = this.server.get() { server.on_pending_request(); - resp.timeout(server.config().idle_timeout); + resp.start_timeout(); } resp.corked(|| this.render_metadata_and_end(resp)); Self::on_response_complete(this, resp); @@ -320,7 +320,7 @@ impl StaticRoute { Self::retain_for_response(this); if let Some(mut server) = this.server.get() { server.on_pending_request(); - resp.timeout(server.config().idle_timeout); + resp.start_timeout(); } let mut finished = false; this.do_render_blob(resp, &mut finished); @@ -400,9 +400,7 @@ impl StaticRoute { } fn on_writable(this: ThisPtr, write_offset: u64, resp: AnyResponse) -> bool { - if let Some(server) = this.server.get() { - resp.timeout(server.config().idle_timeout); - } + resp.reset_timeout(); if !this.on_writable_bytes(write_offset, resp) { return false; @@ -551,7 +549,7 @@ impl StaticRoute { Self::retain_for_response(this); if let Some(mut server) = this.server.get() { server.on_pending_request(); - resp.timeout(server.config().idle_timeout); + resp.start_timeout(); } this.do_write_status(status, resp); this.do_write_headers(resp); diff --git a/src/runtime/server/mod.rs b/src/runtime/server/mod.rs index 7284a0a2b9c7..7fd1865242dd 100644 --- a/src/runtime/server/mod.rs +++ b/src/runtime/server/mod.rs @@ -785,7 +785,7 @@ impl NewServer { server.on_pending_request(); req.set_yield(false); - resp_ref.timeout(server.config.idle_timeout); + resp_ref.reset_timeout(); // Since we do timeouts by default, we should tell the user when // this happens - but limit it to only warn once. @@ -1302,7 +1302,7 @@ impl NewServer { // The listener and what it starts continue the script that made the server. let _context = this_ref.vm().enter_context(this_ref.context.get()); req.set_yield(false); - resp.timeout(this_ref.config.idle_timeout); + resp.reset_timeout(); let global = this_ref.global_this(); let this_object = this_ref.js_value.try_get().unwrap_or(JSValue::UNDEFINED); @@ -3104,6 +3104,7 @@ impl NewServer { // S012: `NewApp` is a ZST opaque — safe `*mut → &mut` deref. bun_opaque::opaque_deref_mut(app).filter(Self::on_connection_filter, this.cast::()); + bun_opaque::opaque_deref_mut(app).set_idle_timeout(this_ref.config.idle_timeout); if this_ref.config.is_node_http_server { // SAFETY: `this` is the live boxed server from `init()`; no other diff --git a/src/runtime/server/server_body.rs b/src/runtime/server/server_body.rs index 102a1c6ca94e..1a3c2965124f 100644 --- a/src/runtime/server/server_body.rs +++ b/src/runtime/server/server_body.rs @@ -336,7 +336,8 @@ pub(super) trait RespLike { const IS_MUX: bool; fn write_status(&mut self, status: &[u8]); fn end_without_body(&mut self, close_connection: bool); - fn timeout(&mut self, seconds: u8); + /// A request starts under the server's idle timeout. + fn start_timeout(&mut self); fn on_timeout_warn(&mut self, ud: *mut c_void); fn to_any_response(&mut self) -> uws::AnyResponse; /// HTTP/2 only: END_STREAM on the HEADERS frame, or `content-length: 0`. @@ -357,8 +358,8 @@ impl RespLike for uws_sys::NewAppResponse { uws_sys::NewAppResponse::::end_without_body(self, c) } #[inline] - fn timeout(&mut self, s: u8) { - uws_sys::NewAppResponse::::timeout(self, s) + fn start_timeout(&mut self) { + uws_sys::NewAppResponse::::reset_timeout(self) } #[inline] fn on_timeout_warn(&mut self, ud: *mut c_void) { @@ -402,9 +403,7 @@ impl RespLike for uws_sys::h3::Response { uws_sys::h3::Response::end_without_body(self, c) } #[inline] - fn timeout(&mut self, s: u8) { - uws_sys::h3::Response::timeout(self, s) - } + fn start_timeout(&mut self) {} #[inline] fn on_timeout_warn(&mut self, ud: *mut c_void) { uws_sys::h3::Response::on_timeout( @@ -433,8 +432,8 @@ impl RespLike for uws_sys::h2::Response { uws_sys::h2::Response::end_without_body(self, c) } #[inline] - fn timeout(&mut self, s: u8) { - uws_sys::h2::Response::timeout(self, s) + fn start_timeout(&mut self) { + uws_sys::h2::Response::start_timeout(self) } #[inline] fn on_timeout_warn(&mut self, ud: *mut c_void) { @@ -2926,7 +2925,7 @@ where server.on_pending_request(); ReqLike::set_yield(req, false); - RespLike::timeout(resp, server.config.idle_timeout); + RespLike::start_timeout(resp); // Since we do timeouts by default, we should tell the user when // this happens - but limit it to only warn once. diff --git a/src/uws_sys/App.rs b/src/uws_sys/App.rs index 9ec9afb62143..596339003299 100644 --- a/src/uws_sys/App.rs +++ b/src/uws_sys/App.rs @@ -146,6 +146,11 @@ impl App { c::uws_app_set_max_http_header_size(Self::SSL_FLAG, self.as_raw(), max_header_size) } + /// HTTP/1 only; seconds, 0 = never (`HttpContextData::idleTimeout`). + pub fn set_idle_timeout(&mut self, seconds: u8) { + c::uws_app_set_idle_timeout(Self::SSL_FLAG, self.as_raw(), seconds) + } + pub fn set_max_headers_count(&mut self, max_headers_count: u32) { c::uws_app_set_max_headers_count(Self::SSL_FLAG, self.as_raw(), max_headers_count) } @@ -528,6 +533,7 @@ pub mod c { app: &mut uws_app_t, max_header_size: u64, ); + pub(crate) safe fn uws_app_set_idle_timeout(ssl: i32, app: &mut uws_app_t, seconds: u8); pub(crate) safe fn uws_app_set_max_headers_count( ssl: i32, app: &mut uws_app_t, diff --git a/src/uws_sys/Response.rs b/src/uws_sys/Response.rs index 453eae7c61eb..4db9e3c4f830 100644 --- a/src/uws_sys/Response.rs +++ b/src/uws_sys/Response.rs @@ -287,11 +287,13 @@ impl Response { c::uws_res_close_if_done_and_marked(Self::ssl_flag(), self.as_raw()) } + /// Overrides the server's idle timeout until the response in flight completes. pub fn timeout(&mut self, seconds: u8) { c::uws_res_timeout(Self::ssl_flag(), self.as_raw(), seconds) } - pub(crate) fn reset_timeout(&mut self) { + /// Re-arms the idle timer; does not change its value. + pub fn reset_timeout(&mut self) { c::uws_res_reset_timeout(Self::ssl_flag(), self.as_raw()) } @@ -852,6 +854,7 @@ impl AnyResponse { any_dispatch!(self, |r| r.state()) } + /// `server.timeout(req, seconds)`: ends with the response (HTTP/1) or the stream (HTTP/2). pub fn timeout(self, seconds: u8) { any_dispatch!(self, |r| r.timeout(seconds)) } @@ -1061,6 +1064,16 @@ impl AnyResponse { any_dispatch!(self, |r| r.reset_timeout()) } + /// A request starts under the server's idle timeout. + pub fn start_timeout(self) { + match self { + AnyResponse::SSL(ptr) => TLSResponse::as_handle(ptr).reset_timeout(), + AnyResponse::TCP(ptr) => TCPResponse::as_handle(ptr).reset_timeout(), + AnyResponse::H3(_) => {} + AnyResponse::H2(ptr) => H2Response::as_handle(ptr).start_timeout(), + } + } + pub fn clear_on_data(self) { any_dispatch!(self, |r| r.clear_on_data()) } diff --git a/src/uws_sys/h2.rs b/src/uws_sys/h2.rs index d0270c3c8d1a..f5faa00af4f1 100644 --- a/src/uws_sys/h2.rs +++ b/src/uws_sys/h2.rs @@ -107,6 +107,10 @@ impl Response { pub(crate) fn reset_timeout(&mut self) { c::uws_h2_res_reset_timeout(self) } + /// This stream's idle budget becomes the server's idle timeout. + pub fn start_timeout(&mut self) { + c::uws_h2_res_start_timeout(self) + } pub(crate) fn get_buffered_amount(&mut self) -> u64 { c::uws_h2_res_get_buffered_amount(self) } @@ -633,6 +637,7 @@ mod c { pub(super) safe fn uws_h2_res_get_buffered_amount(res: &mut Response) -> u64; pub(super) safe fn uws_h2_res_reset_timeout(res: &mut Response); pub(super) safe fn uws_h2_res_timeout(res: &mut Response, seconds: u8); + pub(super) safe fn uws_h2_res_start_timeout(res: &mut Response); pub(super) safe fn uws_h2_res_end_sendfile(res: &mut Response, off: u64, close: bool); // safe: `&mut Response` is ABI-identical to a non-null `*mut`; // `cb`/`ud` are stored opaquely (never dereferenced by the C++ shim diff --git a/src/uws_sys/libuwsockets.cpp b/src/uws_sys/libuwsockets.cpp index d9dc129a943d..49966ddd9492 100644 --- a/src/uws_sys/libuwsockets.cpp +++ b/src/uws_sys/libuwsockets.cpp @@ -502,6 +502,15 @@ extern "C" uwsApp->setMaxHeadersCount(max_headers_count); } } + void uws_app_set_idle_timeout(int ssl, uws_app_t *app, uint8_t seconds) { + if (ssl) { + uWS::SSLApp *uwsApp = (uWS::SSLApp *)app; + uwsApp->setIdleTimeout(seconds); + } else { + uWS::App *uwsApp = (uWS::App *)app; + uwsApp->setIdleTimeout(seconds); + } + } void uws_app_set_flags(int ssl, uws_app_t *app, bool require_host_header, bool use_strict_method_validation, uint8_t lenient_http_flags, bool http_allow_half_open) { if (ssl) { uWS::SSLApp *uwsApp = (uWS::SSLApp *)app; diff --git a/src/uws_sys/libuwsockets_h2.cpp b/src/uws_sys/libuwsockets_h2.cpp index 59a9173e956f..23ebd7ab8dc6 100644 --- a/src/uws_sys/libuwsockets_h2.cpp +++ b/src/uws_sys/libuwsockets_h2.cpp @@ -162,6 +162,12 @@ size_t uws_h2_res_get_buffered_amount(uws_h2_res_t* res) { return ((Http2Respons void uws_h2_res_reset_timeout(uws_h2_res_t* res) { ((Http2Response*)res)->resetTimeout(); } void uws_h2_res_timeout(uws_h2_res_t* res, uint8_t seconds) { ((Http2Response*)res)->setTimeout(seconds); } +/* The stream takes the server's idle timeout as an explicit budget: retireStream() only recomputes the connection's for a stream that has one. */ +void uws_h2_res_start_timeout(uws_h2_res_t* res) +{ + auto* stream = (Http2Response*)res; + if (!stream->dead) stream->setTimeout((uint8_t)stream->conn->ctx->idleTimeoutS); +} void uws_h2_res_end_sendfile(uws_h2_res_t* res, uint64_t, bool close) { ((Http2Response*)res)->sendTerminatingChunk(close); diff --git a/test/js/bun/http/serve-http2-fixture.ts b/test/js/bun/http/serve-http2-fixture.ts index fa546836c2c9..9c129ce75da1 100644 --- a/test/js/bun/http/serve-http2-fixture.ts +++ b/test/js/bun/http/serve-http2-fixture.ts @@ -301,6 +301,10 @@ async function handler(req: Request, server: Bun.Server): Promise { expect((await closed) - tDone).toBeLessThan(15000); }, 40000); + // A FIFO with an open writer is the one file body that can stay quiet. Windows has no FIFO. + test.skipIf(isWindows)( + "server.timeout(req, 0) holds while a Bun.file body is quiet", + async () => { + using dir = tempDir("serve-http2-quiet-file", {}); + const fifo = join(String(dir), "body.fifo"); + mkfifo(fifo); + // A writer that stays open: the body is quiet, not at its end. + const writer = openSync(fifo, "r+"); + let writerOpen = true; + const fx = await startFixture({ tls: false, idleTimeout: 2 }); + try { + writeSync(writer, "first"); + const session = await connectH2(fx.port, false); + const stream = session.request({ ":path": "/file-no-timeout?path=" + encodeURIComponent(fifo) }); + let body = ""; + let closedByServer = false; + let wake = Promise.withResolvers(); + const streamClosed = Promise.withResolvers(); + stream.on("error", () => {}); + stream.on("data", chunk => { + body += chunk; + wake.resolve(); + }); + stream.on("close", () => { + closedByServer = true; + streamClosed.resolve(); + wake.resolve(); + }); + const sees = async (text: string) => { + while (!body.includes(text) && !closedByServer) { + wake = Promise.withResolvers(); + await wake.promise; + } + return body.includes(text); + }; + // The clock: the server closes a silent connection in its next timer + // sweep, and that sweep also ends a stream that is under idleTimeout. + const sweep = async () => { + const silent = await connectH2(fx.port, false); + await new Promise(resolve => silent.once("close", () => resolve())); + }; + const first = await sees("first"); + await sweep(); + writeSync(writer, "late"); + const late = await sees("late"); + expect({ first, late, closedByServer }).toEqual({ first: true, late: true, closedByServer: false }); + // The end of the FIFO ends the body. A fixture that exits with the + // body open leaves the stream allocated, and the leak check reports it. + closeSync(writer); + writerOpen = false; + let waiting = true; + await Promise.race([streamClosed.promise, sweep().then(() => waiting && sweep())]); + waiting = false; + session.destroy(); + } finally { + if (writerOpen) closeSync(writer); + await fx[Symbol.asyncDispose](); + } + expect(fx.proc.exitCode).toBe(0); + }, + 40000, + ); + test("h2 is not negotiated below TLS 1.2; a TLS 1.2 client offering only ECDHE-RSA-AES128-GCM-SHA256 works", async () => { await using fx = await startFixture({ tls: true }); const old = await new Promise(resolve => { diff --git a/test/js/bun/http/serve.test.ts b/test/js/bun/http/serve.test.ts index bad56ba2379b..98cb8880bd0a 100644 --- a/test/js/bun/http/serve.test.ts +++ b/test/js/bun/http/serve.test.ts @@ -1,6 +1,6 @@ import { file, gc, Serve, serve, Server } from "bun"; import { afterAll, afterEach, describe, expect, it, mock } from "bun:test"; -import { readFileSync, writeFileSync } from "fs"; +import { closeSync, openSync, readFileSync, writeFileSync, writeSync } from "fs"; import { bunEnv, bunExe, @@ -29,6 +29,7 @@ import { join, resolve } from "path"; // import app_jsx from "./app.jsx"; import { heapStats } from "bun:jsc"; import { spawn } from "child_process"; +import { mkfifo } from "mkfifo"; import { on, once } from "node:events"; import { createServer as createHttpServer } from "node:http"; import net, { type AddressInfo } from "node:net"; @@ -3439,6 +3440,592 @@ it.concurrent( 20_000, ); +describe("server.timeout(req, seconds) ends with its response", () => { + const GET = (path: string, headers = "") => `GET ${path} HTTP/1.1\r\nHost: localhost\r\n${headers}\r\n`; + const upgrade = + "Connection: Upgrade\r\nUpgrade: websocket\r\nSec-WebSocket-Version: 13\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n"; + const rawSocket = (port: number, secure = false): net.Socket => + secure ? nodeTls.connect({ port, host: "127.0.0.1", rejectUnauthorized: false }) : net.connect(port, "127.0.0.1"); + + // A raw client that keeps its connection and holds what it receives. + function keepAliveClient(port: number, secure = false) { + const socket = rawSocket(port, secure); + let wire = ""; + let isClosed = false; + let wake = Promise.withResolvers(); + const closed = Promise.withResolvers(); + socket.on("error", () => {}); + socket.on("data", chunk => { + wire += chunk.toString("latin1"); + wake.resolve(); + }); + socket.on("close", () => { + isClosed = true; + closed.resolve(); + wake.resolve(); + }); + + // Resolves with the first value `take` finds on the wire, or with undefined + // when the server closes the connection first. + async function next(take: () => T | undefined) { + while (true) { + const value = take(); + if (value !== undefined) return value; + if (isClosed) return undefined; + wake = Promise.withResolvers(); + await wake.promise; + } + } + + // Takes one complete response off the wire and returns its status. + function takeResponse(toHead: boolean) { + const headEnd = wire.indexOf("\r\n\r\n"); + if (headEnd === -1) return undefined; + const status = Number(wire.slice(9, 12)); + const length = /\r\ncontent-length: (\d+)/i.exec(wire.slice(0, headEnd)); + let end: number; + if (toHead || status === 304) { + end = headEnd + 4; + } else if (length) { + end = headEnd + 4 + Number(length[1]); + if (wire.length < end) return undefined; + } else { + // Chunked: the body ends at the zero-length chunk. + const lastChunk = wire.indexOf("0\r\n\r\n", headEnd + 4); + if (lastChunk === -1) return undefined; + end = lastChunk + 5; + } + wire = wire.slice(end); + return status; + } + + return { + closed: closed.promise, + get isClosed() { + return isClosed; + }, + send(bytes: string) { + socket.write(bytes); + }, + response() { + return next(() => takeResponse(false)); + }, + request(head: string) { + socket.write(head); + return next(() => takeResponse(head.startsWith("HEAD "))); + }, + sees(text: string) { + return next(() => (wire.includes(text) ? true : undefined)); + }, + [Symbol.dispose]() { + socket.destroy(); + }, + }; + } + + // A raw client for a large body: it counts what it receives and can stop reading. + function countingClient(port: number, path: string, secure = false) { + const socket = rawSocket(port, secure); + const state = { received: 0, tail: "", closedByServer: false }; + let wake = Promise.withResolvers(); + const closed = Promise.withResolvers(); + socket.on("error", () => {}); + socket.on("close", () => { + state.closedByServer = true; + closed.resolve(); + wake.resolve(); + }); + socket.on("data", chunk => { + state.received += chunk.length; + state.tail = (state.tail + chunk.subarray(-16).toString("latin1")).slice(-16); + wake.resolve(); + }); + socket.write(GET(path)); + return { + state, + socket, + closed: closed.promise, + // Resolves true when `done` holds, false when the server closes first. + async until(done: () => boolean) { + while (!done() && !state.closedByServer) { + wake = Promise.withResolvers(); + await wake.promise; + } + return done(); + }, + [Symbol.dispose]() { + socket.destroy(); + }, + }; + } + + // uWS runs the idle timers of every server of the process in one sweep, + // every 4 s. The kept connection of an `idleTimeout: 1` server is closed by + // the first sweep after its response, so it is a clock with no wall time in + // it. The sweep that closes the clock's connection has also closed, on the + // server side, every connection that was one tick from its timeout before + // the clock's request went out. Only the 'close' event of such a connection + // can reach its client later. The next sweep is the bound for that. + function idleClock() { + const server = Bun.serve({ port: 0, idleTimeout: 1, fetch: () => new Response("tick") }); + let stopped = false; + const sweeps = async (count: number) => { + // Only an answered round trip counts: a sweep closed that connection. + for (let i = 0; i < count && !stopped; ) { + using idle = keepAliveClient(server.port); + if ((await idle.request(GET("/"))) === 200) i++; + await idle.closed; + } + }; + return { + sweeps, + // Resolves when the server has closed every client, and after two sweeps at the latest. + closedOrTwoSweeps(clients: { closed: Promise }[]) { + return Promise.race([Promise.all(clients.map(client => client.closed)), sweeps(2)]); + }, + [Symbol.dispose]() { + stopped = true; + server.stop(true); + }, + }; + } + + it.concurrent( + "the kept connection is under idleTimeout again, however the response ended", + async () => { + using clock = idleClock(); + using dir = tempDir("serve-timeout-bodies", { + "small.txt": "ok", + // Large enough for sendfile on Linux. + "large.bin": Buffer.alloc(2 * 1024 * 1024, "f"), + }); + // Larger than a socket buffer, so that the response can complete from a writable event. + const blob = new Blob([Buffer.alloc(4 * 1024 * 1024, "b")]); + const seconds = (req: Request) => Number(new URL(req.url).searchParams.get("seconds")); + const options = { + port: 0, + idleTimeout: 1, + websocket: { message() {} }, + routes: { + "/route": (req: Request, server: Server) => { + server.timeout(req, seconds(req)); + return new Response("ok"); + }, + }, + fetch(req: Request, server: Server) { + server.timeout(req, seconds(req)); + switch (new URL(req.url).pathname) { + case "/stream": + return new Response( + new ReadableStream({ + async pull(controller) { + controller.enqueue("o"); + await new Promise(resolve => setImmediate(resolve)); + controller.enqueue("k"); + controller.close(); + }, + }), + ); + case "/async": + return new Promise(resolve => setImmediate(() => resolve(new Response("ok")))); + case "/not-modified": + return new Response(null, { status: 304 }); + case "/blob": + return new Response(blob); + case "/file": + return new Response(Bun.file(join(String(dir), "small.txt"))); + case "/large-file": + return new Response(Bun.file(join(String(dir), "large.bin"))); + } + return new Response("ok"); + }, + }; + using plain = Bun.serve(options); + using secure = Bun.serve({ ...options, tls }); + // The handler answers before the rest of the request body arrives. + const upload = (value: number, framing: string) => + `POST /?seconds=${value} HTTP/1.1\r\nHost: localhost\r\n${framing}`; + // 0 never ends the connection and 60 ends it after a minute: idleTimeout + // is 1, so neither value may still be in force after the response. + const legs = [0, 60] + .flatMap(value => [ + { leg: `string body, ${value}`, status: 200, head: GET(`/?seconds=${value}`) }, + { leg: `HEAD, ${value}`, status: 200, head: `HEAD /?seconds=${value} HTTP/1.1\r\nHost: localhost\r\n\r\n` }, + { leg: `304, ${value}`, status: 304, head: GET(`/not-modified?seconds=${value}`) }, + { leg: `stream body, ${value}`, status: 200, head: GET(`/stream?seconds=${value}`) }, + { leg: `async handler, ${value}`, status: 200, head: GET(`/async?seconds=${value}`) }, + { leg: `route handler, ${value}`, status: 200, head: GET(`/route?seconds=${value}`) }, + { + leg: `WebSocket request that fetch answers, ${value}`, + status: 200, + head: GET(`/?seconds=${value}`, upgrade), + }, + { leg: `4 MiB Blob, ${value}`, status: 200, head: GET(`/blob?seconds=${value}`) }, + { leg: `Bun.file, ${value}`, status: 200, head: GET(`/file?seconds=${value}`) }, + { leg: `Bun.file of 2 MiB, ${value}`, status: 200, head: GET(`/large-file?seconds=${value}`) }, + { leg: `unfinished upload, ${value}`, status: 200, head: upload(value, "Content-Length: 10\r\n\r\nhello") }, + { + leg: `unfinished chunked upload, ${value}`, + status: 200, + head: upload(value, "Transfer-Encoding: chunked\r\n\r\n5\r\nhello\r\n"), + }, + { leg: `TLS, ${value}`, status: 200, head: GET(`/?seconds=${value}`), secure: true }, + { + leg: `TLS, Bun.file of 2 MiB, ${value}`, + status: 200, + head: GET(`/large-file?seconds=${value}`), + secure: true, + }, + ]) + .map(leg => ({ ...leg, client: keepAliveClient(leg.secure ? secure.port : plain.port, leg.secure) })); + try { + const statuses = await Promise.all(legs.map(({ client, head }) => client.request(head))); + await clock.closedOrTwoSweeps(legs.map(({ client }) => client)); + expect( + legs.map(({ leg, client }, i) => ({ leg, status: statuses[i], closedByServer: client.isClosed })), + ).toEqual(legs.map(({ leg, status }) => ({ leg, status, closedByServer: true }))); + } finally { + for (const { client } of legs) client[Symbol.dispose](); + } + }, + 30_000, + ); + + it.concurrent( + "a value below idleTimeout does not end the kept connection early", + async () => { + using clock = idleClock(); + // 30 s is eight sweeps away. 0 never ends an idle connection. + const legs = [30, 0].map(idleTimeout => { + const server = Bun.serve({ + port: 0, + idleTimeout, + fetch(req, server) { + if (req.url.endsWith("/short")) server.timeout(req, 1); + return new Response("ok"); + }, + }); + return { idleTimeout, server, client: keepAliveClient(server.port) }; + }); + try { + const first = await Promise.all(legs.map(({ client }) => client.request(GET("/short")))); + await clock.sweeps(1); + // A second request on the kept connection shows that the server did not close it. + const second = await Promise.all(legs.map(({ client }) => client.request(GET("/")))); + expect(legs.map(({ idleTimeout }, i) => ({ idleTimeout, first: first[i], second: second[i] }))).toEqual( + legs.map(({ idleTimeout }) => ({ idleTimeout, first: 200, second: 200 })), + ); + } finally { + for (const { client, server } of legs) { + client[Symbol.dispose](); + server.stop(true); + } + } + }, + 30_000, + ); + + it.concurrent( + "a request that no route arms starts under idleTimeout as well", + async () => { + using clock = idleClock(); + // No fetch handler: the 404 for an unknown path runs no handler. + using routesOnly = Bun.serve({ + port: 0, + idleTimeout: 1, + routes: { + "/forever": (req, server) => { + server.timeout(req, 0); + return new Response("ok"); + }, + }, + }); + // A WebSocket request that fetch answers does not pass the route start either. + using withWebSocket = Bun.serve({ + port: 0, + idleTimeout: 1, + websocket: { message() {} }, + fetch(req) { + if (!req.url.endsWith("/quiet")) return new Response("ok"); + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue("hi"); + }, + }), + ); + }, + }); + using afterOverride = keepAliveClient(routesOnly.port); + using unknown = keepAliveClient(routesOnly.port); + using answered = keepAliveClient(withWebSocket.port); + using quietBody = keepAliveClient(withWebSocket.port); + // One write for both requests: between two writes a sweep can end the + // kept connection, which is under idleTimeout again after the first. + afterOverride.send(GET("/forever") + GET("/unknown")); + unknown.send(GET("/unknown")); + answered.send(GET("/", upgrade)); + quietBody.send(GET("/quiet", upgrade)); + const seen = { + afterOverride: [await afterOverride.response(), await afterOverride.response()], + unknown: await unknown.response(), + answered: await answered.response(), + quietBody: await quietBody.sees("hi"), + }; + const clients = [afterOverride, unknown, answered, quietBody]; + await clock.closedOrTwoSweeps(clients); + expect({ ...seen, closedByServer: clients.map(client => client.isClosed) }).toEqual({ + afterOverride: [200, 404], + unknown: 404, + answered: 200, + quietBody: true, + closedByServer: [true, true, true, true], + }); + }, + 30_000, + ); + + it.concurrent( + "a pending request keeps its idle timer, and the value holds while its response is pending", + async () => { + using clock = idleClock(); + const release = Promise.withResolvers(); + const handler = async (req: Request, server: Server) => { + // Never answers. + if (!req.url.endsWith("/held")) return new Promise(() => {}); + server.timeout(req, 0); + await release.promise; + return new Response("ok"); + }; + using server = Bun.serve({ port: 0, idleTimeout: 1, routes: { "/route/:kind": handler }, fetch: handler }); + using never = keepAliveClient(server.port); + using routeNever = keepAliveClient(server.port); + using held = keepAliveClient(server.port); + using routeHeld = keepAliveClient(server.port); + never.send(GET("/never")); + routeNever.send(GET("/route/never")); + const answers = [held.request(GET("/held")), routeHeld.request(GET("/route/held"))]; + await clock.sweeps(1); + release.resolve(); + const heldStatuses = await Promise.all(answers); + await clock.closedOrTwoSweeps([never, routeNever]); + expect({ held: heldStatuses, neverClosedByServer: [never.isClosed, routeNever.isClosed] }).toEqual({ + held: [200, 200], + neverClosedByServer: [true, true], + }); + }, + 30_000, + ); + + // A FIFO with an open writer is the one file body that can stay quiet. Windows has no FIFO. + it.concurrent.skipIf(isWindows)( + "it holds while a Bun.file body is quiet", + async () => { + using clock = idleClock(); + using dir = tempDir("serve-timeout-quiet-file", {}); + const fifo = join(String(dir), "body.fifo"); + mkfifo(fifo); + // A writer that stays open: the body is quiet, not at its end. + const writer = openSync(fifo, "r+"); + try { + writeSync(writer, "first"); + using server = Bun.serve({ + port: 0, + idleTimeout: 1, + fetch(req, server) { + server.timeout(req, 0); + return new Response(Bun.file(fifo)); + }, + }); + using client = keepAliveClient(server.port); + client.send(GET("/")); + const first = await client.sees("first"); + await clock.sweeps(1); + writeSync(writer, "late"); + const late = await client.sees("late"); + expect({ first, late, closedByServer: client.isClosed }).toEqual({ + first: true, + late: true, + closedByServer: false, + }); + } finally { + closeSync(writer); + } + }, + 30_000, + ); + + it.concurrent( + "it holds while its response is pending, under backpressure and after a drain", + async () => { + using clock = idleClock(); + // More than the kernel buffers between the two ends. + const size = 32 * 1024 * 1024; + using dir = tempDir("serve-timeout-pending-body", { "body.bin": Buffer.alloc(size, "f") }); + const burst = Buffer.alloc(size, "s"); + const quiet = Promise.withResolvers(); + const options = { + port: 0, + idleTimeout: 1, + fetch(req: Request, server: Server) { + server.timeout(req, 0); + if (req.url.endsWith("/file")) return new Response(Bun.file(join(String(dir), "body.bin"))); + return new Response( + new ReadableStream({ + type: "direct", + async pull(controller) { + controller.write(burst); + await controller.flush(); + await quiet.promise; + controller.write("late"); + controller.close(); + }, + }), + ); + }, + }; + using plain = Bun.serve(options); + using secure = Bun.serve({ ...options, tls }); + using file = countingClient(plain.port, "/file"); + using tlsFile = countingClient(secure.port, "/file", true); + using stream = countingClient(plain.port, "/stream"); + // The files: stop reading with a quarter of the body across. + const stalledMidBody = await Promise.all( + [file, tlsFile].map(async client => { + const stalled = await client.until(() => client.state.received > size / 4); + client.socket.pause(); + return stalled; + }), + ); + // The stream: the client holds the whole burst, so the server has drained it. + const drained = await stream.until(() => stream.state.received > size); + await clock.sweeps(1); + file.socket.resume(); + tlsFile.socket.resume(); + quiet.resolve(); + const [deliveredWhole, tlsDeliveredWhole, late] = await Promise.all([ + file.until(() => file.state.received > size), + tlsFile.until(() => tlsFile.state.received > size), + stream.until(() => stream.state.tail.includes("late")), + ]); + expect({ stalledMidBody, deliveredWhole, tlsDeliveredWhole, drained, late }).toEqual({ + stalledMidBody: [true, true], + deliveredWhole: true, + tlsDeliveredWhole: true, + drained: true, + late: true, + }); + }, + 40_000, + ); + + it.concurrent( + "a value below idleTimeout ends a stalled Bun.file body", + async () => { + using clock = idleClock(); + // More than the kernel buffers between the two ends. + const size = 32 * 1024 * 1024; + using dir = tempDir("serve-timeout-short-file", { "body.bin": Buffer.alloc(size, "f") }); + const cut = Promise.withResolvers(); + using server = Bun.serve({ + port: 0, + idleTimeout: 30, + fetch(req, server) { + server.timeout(req, 1); + req.signal.addEventListener("abort", () => cut.resolve()); + return new Response(Bun.file(join(String(dir), "body.bin"))); + }, + }); + using client = countingClient(server.port, "/"); + const stalled = await client.until(() => client.state.received > 0); + client.socket.pause(); + // The server ends the request in its next sweep. 30 s is eight sweeps away. + const cutInTwoSweeps = await Promise.race([cut.promise.then(() => true), clock.sweeps(2).then(() => false)]); + // A client that does not read does not see the close. Read what is left. + client.socket.resume(); + const deliveredWhole = await client.until(() => client.state.received > size); + expect({ stalled, cutInTwoSweeps, deliveredWhole, closedByServer: client.state.closedByServer }).toEqual({ + stalled: true, + cutInTwoSweeps: true, + deliveredWhole: false, + closedByServer: true, + }); + }, + 30_000, + ); + + it.concurrent( + "it holds until a response that ended under backpressure has drained", + async () => { + using clock = idleClock(); + // More than the kernel buffers between the two ends. + const chunk = Buffer.alloc(32 * 1024 * 1024, "a"); + const written: Record = {}; + const options = { + port: 0, + idleTimeout: 1, + fetch(req: Request, server: Server) { + server.timeout(req, 60); + const response = (written[new URL(req.url).pathname.slice(1)] = { bytes: 0, backedUp: false }); + return new Response( + new ReadableStream({ + type: "direct", + pull(controller) { + // Write until the socket pushes back, then close without a wait: + // the stream ends while the server still holds the rest. + while (!response.backedUp && response.bytes < 4 * chunk.length) { + response.backedUp = controller.write(chunk) instanceof Promise; + response.bytes += chunk.length; + } + controller.close(); + }, + }), + ); + }, + }; + using plain = Bun.serve(options); + using secure = Bun.serve({ ...options, tls }); + const legs = [ + { leg: "TCP", client: countingClient(plain.port, "/TCP") }, + { leg: "TLS", client: countingClient(secure.port, "/TLS", true) }, + ]; + try { + // Stop reading for one sweep, with most of the body still to come. + const stalled = await Promise.all( + legs.map(async ({ client }) => { + const gotFirstBytes = await client.until(() => client.state.received > 0); + client.socket.pause(); + return gotFirstBytes; + }), + ); + await clock.sweeps(1); + for (const { client } of legs) client.socket.resume(); + const deliveredWhole = await Promise.all( + legs.map(({ leg, client }) => + client.until(() => client.state.received > written[leg].bytes && client.state.tail.endsWith("0\r\n\r\n")), + ), + ); + // Drained: the kept connection is under idleTimeout again. + await clock.closedOrTwoSweeps(legs.map(({ client }) => client)); + expect( + legs.map(({ leg, client }, i) => ({ + leg, + stalled: stalled[i], + backedUp: written[leg].backedUp, + deliveredWhole: deliveredWhole[i], + closedByServer: client.state.closedByServer, + })), + ).toEqual( + legs.map(({ leg }) => ({ leg, stalled: true, backedUp: true, deliveredWhole: true, closedByServer: true })), + ); + } finally { + for (const { client } of legs) client[Symbol.dispose](); + } + }, + 40_000, + ); +}); + it.concurrent( "TLS: reaps every zero-byte pre-handshake connection in a burst", async () => { From 92a92f08f68c085d7a018460fc9949e5d19611f0 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Thu, 8 Oct 2026 03:45:11 +0000 Subject: [PATCH 2/6] Bun.serve: keep the override for a corked tail, and test the sweep without waits A response that ended while its last bytes were in the cork buffer gave the server's idleTimeout back before the uncork. When the socket did not take those bytes, the tail waited under idleTimeout and not under the server.timeout(req, N) of its request. endTimeoutOverrideIfDrained() now sends the corked bytes first (sendCorked), so such a tail is backpressure and keeps the override until onWritable drains it. bun:internal-for-testing gets runSocketTimeoutSweepSoon(). It makes the 4 s usockets timeout sweep of the calling thread's loop run in its next iteration. The new tests use it: they wait for no wall-clock time and set no timeouts. They are serial, because a forced sweep moves every connection of the process one tick closer to its timeout. The HTTP/2 test runs its server in the test process. The first HTTP/1 test has fewer legs, and TLS has a test of its own. --- packages/bun-usockets/src/eventing/libuv.c | 11 + packages/bun-usockets/src/internal/internal.h | 3 + packages/bun-usockets/src/loop.c | 6 + packages/bun-uws/src/HttpResponse.h | 6 +- src/js/internal-for-testing.ts | 7 + src/runtime/socket/socket_body.rs | 11 + src/uws_sys/Loop.rs | 3 +- test/js/bun/http/serve-http2-fixture.ts | 4 - .../js/bun/http/serve-http2-lifecycle.test.ts | 132 +-- test/js/bun/http/serve-syscall-fault.test.ts | 45 + test/js/bun/http/serve.test.ts | 831 +++++++++--------- 11 files changed, 573 insertions(+), 486 deletions(-) diff --git a/packages/bun-usockets/src/eventing/libuv.c b/packages/bun-usockets/src/eventing/libuv.c index a76d0e73f48d..18cda0700b5b 100644 --- a/packages/bun-usockets/src/eventing/libuv.c +++ b/packages/bun-usockets/src/eventing/libuv.c @@ -503,6 +503,17 @@ struct us_loop_t *us_timer_loop(struct us_timer_t *t) { return internal_cb->loop; } +void us_internal_sweep_soon(struct us_loop_t *loop) { + struct us_internal_callback_t *internal_cb = + (struct us_internal_callback_t *)loop->data.sweep_timer; + + /* Not us_timer_set(): it does not restart a sweep timer that runs. */ + if (internal_cb->has_added_timer_to_event_loop) { + uv_timer_start((uv_timer_t *)(internal_cb + 1), timer_cb, 1, + LIBUS_TIMEOUT_GRANULARITY * 1000); + } +} + // async (internal only) struct us_internal_async *us_internal_create_async(struct us_loop_t *loop, int fallthrough, diff --git a/packages/bun-usockets/src/internal/internal.h b/packages/bun-usockets/src/internal/internal.h index d0364eb665a5..3549476846f6 100644 --- a/packages/bun-usockets/src/internal/internal.h +++ b/packages/bun-usockets/src/internal/internal.h @@ -177,6 +177,9 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in void us_internal_timer_sweep(us_loop_r loop); void us_internal_enable_sweep_timer(struct us_loop_t *loop); void us_internal_disable_sweep_timer(struct us_loop_t *loop); +/* Testing only: the next loop iteration runs the sweep, as if + * LIBUS_TIMEOUT_GRANULARITY seconds had passed. */ +void us_internal_sweep_soon(struct us_loop_t *loop); #ifndef LIBUS_USE_LIBUV /* CLOCK_MONOTONIC in ns. The clock every deadline on the loop is measured * against, so anything comparing against one must read it and not another. */ diff --git a/packages/bun-usockets/src/loop.c b/packages/bun-usockets/src/loop.c index 135ec4c879f4..e97ae418b9a1 100644 --- a/packages/bun-usockets/src/loop.c +++ b/packages/bun-usockets/src/loop.c @@ -118,6 +118,12 @@ void us_internal_sweep_if_due(struct us_loop_t *loop) { us_internal_timer_sweep(loop); } +void us_internal_sweep_soon(struct us_loop_t *loop) { + if (loop->data.sweep_next_tick_ns >= 0) { + loop->data.sweep_next_tick_ns = 0; + } +} + #endif diff --git a/packages/bun-uws/src/HttpResponse.h b/packages/bun-uws/src/HttpResponse.h index 8e42b5b68f6f..f4a47eceef04 100644 --- a/packages/bun-uws/src/HttpResponse.h +++ b/packages/bun-uws/src/HttpResponse.h @@ -67,11 +67,13 @@ struct HttpResponse : public AsyncSocket { /* A setTimeout() override ends when its response is complete and * hasFullyDrained(): a response that ended under backpressure keeps it - * until onWritable has drained the tail. Bytes still in the cork buffer - * do not count. Does not arm. */ + * until onWritable has drained the tail. Does not arm. */ void endTimeoutOverrideIfDrained() { auto* data = getHttpResponseData(); if ((data->state & (HttpResponseData::HTTP_TIMEOUT_OVERRIDDEN | HttpResponseData::HTTP_RESPONSE_PENDING)) == HttpResponseData::HTTP_TIMEOUT_OVERRIDDEN) [[unlikely]] { + /* The cork buffer can hold the end of the response. Send it + * first: a tail the socket does not take is backpressure too. */ + Super::sendCorked(); if (Super::hasFullyDrained()) { data->state &= ~HttpResponseData::HTTP_TIMEOUT_OVERRIDDEN; data->idleTimeoutInForce = HttpContext::getSocketContextDataS((us_socket_t *) this)->idleTimeout; diff --git a/src/js/internal-for-testing.ts b/src/js/internal-for-testing.ts index 68183d47f35e..4a0ccc9595db 100644 --- a/src/js/internal-for-testing.ts +++ b/src/js/internal-for-testing.ts @@ -651,6 +651,13 @@ export const socketFaultInjection = { clear: $newRustFunction("runtime/socket/socket.rs", "TestingAPIs.jsClearSocketFaults", 0) as () => void, }; +// Runs the 4 s usockets timeout sweep of this thread in its next loop iteration: every socket moves one step closer to its timeout, so not beside tests that need real idle time. +export const runSocketTimeoutSweepSoon: () => void = $newRustFunction( + "runtime/socket/socket.rs", + "TestingAPIs.jsRunSocketTimeoutSweepSoon", + 0, +); + export const namedPipeInternals = { /** * Live native contexts behind sockets over Windows named pipes: one per diff --git a/src/runtime/socket/socket_body.rs b/src/runtime/socket/socket_body.rs index 84dfc2309afc..8aca564e75d3 100644 --- a/src/runtime/socket/socket_body.rs +++ b/src/runtime/socket/socket_body.rs @@ -5125,6 +5125,17 @@ pub(crate) mod testing_apis { Ok(JSValue::js_number(count as f64)) } + /// `runSocketTimeoutSweepSoon` in `bun:internal-for-testing`. + #[bun_jsc::host_fn] + pub(crate) fn js_run_socket_timeout_sweep_soon( + _global: &JSGlobalObject, + _frame: &CallFrame, + ) -> JsResult { + // SAFETY: this thread's loop; the call only moves its sweep deadline. + unsafe { bun_uws_sys::loop_::us_internal_sweep_soon(bun_uws_sys::Loop::get()) }; + Ok(JSValue::UNDEFINED) + } + #[bun_jsc::host_fn] pub(crate) fn js_clear_socket_faults( global: &JSGlobalObject, diff --git a/src/uws_sys/Loop.rs b/src/uws_sys/Loop.rs index 42ebb296a441..025fcd71068d 100644 --- a/src/uws_sys/Loop.rs +++ b/src/uws_sys/Loop.rs @@ -515,6 +515,7 @@ mod c { now_ns: u64, ); pub(super) fn us_internal_free_closed_sockets(loop_: *mut Loop); + pub fn us_internal_sweep_soon(loop_: *mut Loop); pub(super) fn us_loop_close_all_groups(loop_: *mut Loop) -> c_int; #[cfg(not(windows))] pub(super) safe fn uws_get_loop() -> *mut Loop; @@ -530,7 +531,7 @@ mod c { // event-loop thread parks inside it while worker threads call // `us_wakeup_loop` concurrently; routing either through a `&mut self` // receiver would create two live `&mut Loop` to the same singleton (UB). -pub use c::{us_loop_run, us_wakeup_loop}; +pub use c::{us_internal_sweep_soon, us_loop_run, us_wakeup_loop}; unsafe extern "C" { // safe: no args; frees this thread's lazily-created uws loop if it exists. diff --git a/test/js/bun/http/serve-http2-fixture.ts b/test/js/bun/http/serve-http2-fixture.ts index 9c129ce75da1..fa546836c2c9 100644 --- a/test/js/bun/http/serve-http2-fixture.ts +++ b/test/js/bun/http/serve-http2-fixture.ts @@ -301,10 +301,6 @@ async function handler(req: Request, server: Bun.Server): Promise { }, 40000); // A FIFO with an open writer is the one file body that can stay quiet. Windows has no FIFO. - test.skipIf(isWindows)( - "server.timeout(req, 0) holds while a Bun.file body is quiet", - async () => { - using dir = tempDir("serve-http2-quiet-file", {}); - const fifo = join(String(dir), "body.fifo"); - mkfifo(fifo); - // A writer that stays open: the body is quiet, not at its end. - const writer = openSync(fifo, "r+"); - let writerOpen = true; - const fx = await startFixture({ tls: false, idleTimeout: 2 }); - try { - writeSync(writer, "first"); - const session = await connectH2(fx.port, false); - const stream = session.request({ ":path": "/file-no-timeout?path=" + encodeURIComponent(fifo) }); - let body = ""; - let closedByServer = false; - let wake = Promise.withResolvers(); - const streamClosed = Promise.withResolvers(); - stream.on("error", () => {}); - stream.on("data", chunk => { - body += chunk; - wake.resolve(); - }); - stream.on("close", () => { - closedByServer = true; - streamClosed.resolve(); - wake.resolve(); - }); - const sees = async (text: string) => { - while (!body.includes(text) && !closedByServer) { - wake = Promise.withResolvers(); - await wake.promise; - } - return body.includes(text); - }; - // The clock: the server closes a silent connection in its next timer - // sweep, and that sweep also ends a stream that is under idleTimeout. - const sweep = async () => { - const silent = await connectH2(fx.port, false); - await new Promise(resolve => silent.once("close", () => resolve())); - }; - const first = await sees("first"); - await sweep(); - writeSync(writer, "late"); - const late = await sees("late"); - expect({ first, late, closedByServer }).toEqual({ first: true, late: true, closedByServer: false }); - // The end of the FIFO ends the body. A fixture that exits with the - // body open leaves the stream allocated, and the leak check reports it. - closeSync(writer); - writerOpen = false; - let waiting = true; - await Promise.race([streamClosed.promise, sweep().then(() => waiting && sweep())]); - waiting = false; - session.destroy(); - } finally { - if (writerOpen) closeSync(writer); - await fx[Symbol.asyncDispose](); - } - expect(fx.proc.exitCode).toBe(0); - }, - 40000, - ); + test.skipIf(isWindows)("server.timeout(req, 0) holds while a Bun.file body is quiet", async () => { + using dir = tempDir("serve-http2-quiet-file", {}); + const fifo = join(String(dir), "body.fifo"); + mkfifo(fifo); + // A writer that stays open: the body is quiet, not at its end. + const writer = openSync(fifo, "r+"); + let writerOpen = true; + try { + writeSync(writer, "first"); + using server = Bun.serve({ + port: 0, + http2: true, + idleTimeout: 1, + fetch(req, server) { + if (!req.url.endsWith("/quiet")) return new Response("tick"); + server.timeout(req, 0); + return new Response(Bun.file(fifo)); + }, + }); + const session = await connectH2(server.port, false); + const stream = session.request({ ":path": "/quiet" }); + let body = ""; + let closedByServer = false; + let wake = Promise.withResolvers(); + const streamClosed = Promise.withResolvers(); + stream.on("error", () => {}); + stream.on("data", chunk => { + body += chunk; + wake.resolve(); + }); + stream.on("close", () => { + closedByServer = true; + streamClosed.resolve(); + wake.resolve(); + }); + const sees = async (text: string) => { + while (!body.includes(text) && !closedByServer) { + wake = Promise.withResolvers(); + await wake.promise; + } + return body.includes(text); + }; + // The clock: a connection whose one request is answered is under idleTimeout, + // and the sweep that closes it also ends a stream that is under idleTimeout. + // runSocketTimeoutSweepSoon() makes that sweep run now. + const sweep = async () => { + const idle = await connectH2(server.port, false); + idle.on("error", () => {}); + const closed = new Promise(resolve => idle.once("close", () => resolve())); + await request(idle, { ":path": "/tick" }); + runSocketTimeoutSweepSoon(); + await closed; + }; + const first = await sees("first"); + await sweep(); + writeSync(writer, "late"); + const late = await sees("late"); + expect({ first, late, closedByServer }).toEqual({ first: true, late: true, closedByServer: false }); + // The end of the FIFO ends the body, so that no open body outlives the test. + closeSync(writer); + writerOpen = false; + let waiting = true; + await Promise.race([streamClosed.promise, sweep().then(() => waiting && sweep())]); + waiting = false; + session.destroy(); + } finally { + if (writerOpen) closeSync(writer); + } + }); test("h2 is not negotiated below TLS 1.2; a TLS 1.2 client offering only ECDHE-RSA-AES128-GCM-SHA256 works", async () => { await using fx = await startFixture({ tls: true }); diff --git a/test/js/bun/http/serve-syscall-fault.test.ts b/test/js/bun/http/serve-syscall-fault.test.ts index ac1a02895c08..f6083a7a11b1 100644 --- a/test/js/bun/http/serve-syscall-fault.test.ts +++ b/test/js/bun/http/serve-syscall-fault.test.ts @@ -1,6 +1,7 @@ import { socketFaultInjection as fault } from "bun:internal-for-testing"; import { describe, expect, test } from "bun:test"; import { bunEnv, bunExe, tls as certs, isWindows } from "harness"; +import net from "node:net"; const skip = !fault.available() || isWindows; @@ -120,6 +121,50 @@ describe.skipIf(skip)("Bun.serve under injected syscall faults", () => { expect(proc.exitCode).toBe(0); }); + test("server.timeout(req, N) holds when the socket takes none of a response that ended in the cork buffer", async () => { + const { proc, port } = await spawnServer(/* js */ ` + const { socketFaultInjection: fault, runSocketTimeoutSweepSoon } = require("bun:internal-for-testing"); + const s = Bun.serve({ port: 0, hostname: "127.0.0.1", idleTimeout: 1, + fetch(req, server) { + server.timeout(req, 60); + // The response ends with all of its bytes in the cork buffer, and no send moves one. + fault.set({ syscall: "send", action: "zero", repeat: -1 }); + fault.set({ syscall: "writev", action: "zero", repeat: -1 }); + // One idle sweep runs over the connection, then the sends work again. + setImmediate(() => { + runSocketTimeoutSweepSoon(); + setImmediate(() => setImmediate(() => fault.clear())); + }); + return new Response("tail"); + } }); + console.log(s.port); + process.on("SIGTERM", () => { fault.clear(); s.stop(true); process.exit(0); }); + `); + try { + // The wire up to the end of the response, or up to the close if the server cut it. + const wire = Promise.withResolvers(); + const socket = net.connect(port, "127.0.0.1"); + let received = ""; + socket.on("error", () => {}); + socket.on("data", chunk => { + received += chunk; + if (received.endsWith("\r\n\r\ntail")) wire.resolve(received); + }); + socket.on("close", () => wire.resolve(received)); + socket.write("GET / HTTP/1.1\r\nHost: x\r\n\r\n"); + try { + expect(await wire.promise).toEndWith("\r\n\r\ntail"); + } finally { + socket.destroy(); + } + } finally { + proc.kill("SIGTERM"); + await proc.exited; + } + expect(proc.signalCode).toBeNull(); + expect(proc.exitCode).toBe(0); + }); + test("client abort under server-side 1-byte sends: every response reaches a terminal state", async () => { const { proc, port } = await spawnServer(/* js */ ` const { socketFaultInjection: fault } = require("bun:internal-for-testing"); diff --git a/test/js/bun/http/serve.test.ts b/test/js/bun/http/serve.test.ts index 98cb8880bd0a..39115abe6eec 100644 --- a/test/js/bun/http/serve.test.ts +++ b/test/js/bun/http/serve.test.ts @@ -27,6 +27,7 @@ import { connect } from "net"; import { join, resolve } from "path"; // import { renderToReadableStream } from "react-dom/server"; // import app_jsx from "./app.jsx"; +import { runSocketTimeoutSweepSoon } from "bun:internal-for-testing"; import { heapStats } from "bun:jsc"; import { spawn } from "child_process"; import { mkfifo } from "mkfifo"; @@ -3440,6 +3441,7 @@ it.concurrent( 20_000, ); +// Not concurrent: a forced sweep moves every connection of the process one tick closer to its timeout. describe("server.timeout(req, seconds) ends with its response", () => { const GET = (path: string, headers = "") => `GET ${path} HTTP/1.1\r\nHost: localhost\r\n${headers}\r\n`; const upgrade = @@ -3560,12 +3562,13 @@ describe("server.timeout(req, seconds) ends with its response", () => { } // uWS runs the idle timers of every server of the process in one sweep, - // every 4 s. The kept connection of an `idleTimeout: 1` server is closed by - // the first sweep after its response, so it is a clock with no wall time in - // it. The sweep that closes the clock's connection has also closed, on the - // server side, every connection that was one tick from its timeout before - // the clock's request went out. Only the 'close' event of such a connection - // can reach its client later. The next sweep is the bound for that. + // every 4 s. runSocketTimeoutSweepSoon() makes that sweep run now, so no + // test waits for wall time. The kept connection of an `idleTimeout: 1` + // server is closed by the first sweep after its response: its 'close' shows + // that a sweep ran. That sweep has also closed, on the server side, every + // connection that was one tick from its timeout before the clock's request + // went out. Only the 'close' event of such a connection can reach its + // client later. The next sweep is the bound for that. function idleClock() { const server = Bun.serve({ port: 0, idleTimeout: 1, fetch: () => new Response("tick") }); let stopped = false; @@ -3573,7 +3576,10 @@ describe("server.timeout(req, seconds) ends with its response", () => { // Only an answered round trip counts: a sweep closed that connection. for (let i = 0; i < count && !stopped; ) { using idle = keepAliveClient(server.port); - if ((await idle.request(GET("/"))) === 200) i++; + if ((await idle.request(GET("/"))) === 200 && !stopped) { + i++; + runSocketTimeoutSweepSoon(); + } await idle.closed; } }; @@ -3590,440 +3596,431 @@ describe("server.timeout(req, seconds) ends with its response", () => { }; } - it.concurrent( - "the kept connection is under idleTimeout again, however the response ended", - async () => { - using clock = idleClock(); - using dir = tempDir("serve-timeout-bodies", { - "small.txt": "ok", - // Large enough for sendfile on Linux. - "large.bin": Buffer.alloc(2 * 1024 * 1024, "f"), - }); - // Larger than a socket buffer, so that the response can complete from a writable event. - const blob = new Blob([Buffer.alloc(4 * 1024 * 1024, "b")]); - const seconds = (req: Request) => Number(new URL(req.url).searchParams.get("seconds")); - const options = { - port: 0, - idleTimeout: 1, - websocket: { message() {} }, - routes: { - "/route": (req: Request, server: Server) => { - server.timeout(req, seconds(req)); - return new Response("ok"); - }, - }, - fetch(req: Request, server: Server) { + // Answers in every way a response can end. `seconds` in the query goes to server.timeout(). + function serverOfEveryEnding(options: { tls?: typeof tls } = {}) { + const dir = tempDir("serve-timeout-bodies", { + "small.txt": "ok", + // Large enough for sendfile on Linux. + "large.bin": Buffer.alloc(2 * 1024 * 1024, "f"), + }); + // Larger than a socket buffer, so that the response can complete from a writable event. + const blob = new Blob([Buffer.alloc(4 * 1024 * 1024, "b")]); + const seconds = (req: Request) => Number(new URL(req.url).searchParams.get("seconds")); + const server = Bun.serve({ + ...options, + port: 0, + idleTimeout: 1, + websocket: { message() {} }, + routes: { + "/route": (req: Request, server: Server) => { server.timeout(req, seconds(req)); - switch (new URL(req.url).pathname) { - case "/stream": - return new Response( - new ReadableStream({ - async pull(controller) { - controller.enqueue("o"); - await new Promise(resolve => setImmediate(resolve)); - controller.enqueue("k"); - controller.close(); - }, - }), - ); - case "/async": - return new Promise(resolve => setImmediate(() => resolve(new Response("ok")))); - case "/not-modified": - return new Response(null, { status: 304 }); - case "/blob": - return new Response(blob); - case "/file": - return new Response(Bun.file(join(String(dir), "small.txt"))); - case "/large-file": - return new Response(Bun.file(join(String(dir), "large.bin"))); - } return new Response("ok"); }, - }; - using plain = Bun.serve(options); - using secure = Bun.serve({ ...options, tls }); - // The handler answers before the rest of the request body arrives. - const upload = (value: number, framing: string) => - `POST /?seconds=${value} HTTP/1.1\r\nHost: localhost\r\n${framing}`; - // 0 never ends the connection and 60 ends it after a minute: idleTimeout - // is 1, so neither value may still be in force after the response. - const legs = [0, 60] - .flatMap(value => [ - { leg: `string body, ${value}`, status: 200, head: GET(`/?seconds=${value}`) }, - { leg: `HEAD, ${value}`, status: 200, head: `HEAD /?seconds=${value} HTTP/1.1\r\nHost: localhost\r\n\r\n` }, - { leg: `304, ${value}`, status: 304, head: GET(`/not-modified?seconds=${value}`) }, - { leg: `stream body, ${value}`, status: 200, head: GET(`/stream?seconds=${value}`) }, - { leg: `async handler, ${value}`, status: 200, head: GET(`/async?seconds=${value}`) }, - { leg: `route handler, ${value}`, status: 200, head: GET(`/route?seconds=${value}`) }, - { - leg: `WebSocket request that fetch answers, ${value}`, - status: 200, - head: GET(`/?seconds=${value}`, upgrade), - }, - { leg: `4 MiB Blob, ${value}`, status: 200, head: GET(`/blob?seconds=${value}`) }, - { leg: `Bun.file, ${value}`, status: 200, head: GET(`/file?seconds=${value}`) }, - { leg: `Bun.file of 2 MiB, ${value}`, status: 200, head: GET(`/large-file?seconds=${value}`) }, - { leg: `unfinished upload, ${value}`, status: 200, head: upload(value, "Content-Length: 10\r\n\r\nhello") }, - { - leg: `unfinished chunked upload, ${value}`, - status: 200, - head: upload(value, "Transfer-Encoding: chunked\r\n\r\n5\r\nhello\r\n"), - }, - { leg: `TLS, ${value}`, status: 200, head: GET(`/?seconds=${value}`), secure: true }, - { - leg: `TLS, Bun.file of 2 MiB, ${value}`, - status: 200, - head: GET(`/large-file?seconds=${value}`), - secure: true, - }, - ]) - .map(leg => ({ ...leg, client: keepAliveClient(leg.secure ? secure.port : plain.port, leg.secure) })); - try { - const statuses = await Promise.all(legs.map(({ client, head }) => client.request(head))); - await clock.closedOrTwoSweeps(legs.map(({ client }) => client)); - expect( - legs.map(({ leg, client }, i) => ({ leg, status: statuses[i], closedByServer: client.isClosed })), - ).toEqual(legs.map(({ leg, status }) => ({ leg, status, closedByServer: true }))); - } finally { - for (const { client } of legs) client[Symbol.dispose](); - } - }, - 30_000, - ); - - it.concurrent( - "a value below idleTimeout does not end the kept connection early", - async () => { - using clock = idleClock(); - // 30 s is eight sweeps away. 0 never ends an idle connection. - const legs = [30, 0].map(idleTimeout => { - const server = Bun.serve({ - port: 0, - idleTimeout, - fetch(req, server) { - if (req.url.endsWith("/short")) server.timeout(req, 1); - return new Response("ok"); - }, - }); - return { idleTimeout, server, client: keepAliveClient(server.port) }; - }); - try { - const first = await Promise.all(legs.map(({ client }) => client.request(GET("/short")))); - await clock.sweeps(1); - // A second request on the kept connection shows that the server did not close it. - const second = await Promise.all(legs.map(({ client }) => client.request(GET("/")))); - expect(legs.map(({ idleTimeout }, i) => ({ idleTimeout, first: first[i], second: second[i] }))).toEqual( - legs.map(({ idleTimeout }) => ({ idleTimeout, first: 200, second: 200 })), - ); - } finally { - for (const { client, server } of legs) { - client[Symbol.dispose](); - server.stop(true); + }, + fetch(req: Request, server: Server) { + server.timeout(req, seconds(req)); + switch (new URL(req.url).pathname) { + case "/stream": + return new Response( + new ReadableStream({ + async pull(controller) { + controller.enqueue("o"); + await new Promise(resolve => setImmediate(resolve)); + controller.enqueue("k"); + controller.close(); + }, + }), + ); + case "/async": + return new Promise(resolve => setImmediate(() => resolve(new Response("ok")))); + case "/not-modified": + return new Response(null, { status: 304 }); + case "/blob": + return new Response(blob); + case "/file": + return new Response(Bun.file(join(String(dir), "small.txt"))); + case "/large-file": + return new Response(Bun.file(join(String(dir), "large.bin"))); } - } - }, - 30_000, - ); + return new Response("ok"); + }, + }); + return { + port: server.port, + [Symbol.dispose]() { + server.stop(true); + dir[Symbol.dispose](); + }, + }; + } - it.concurrent( - "a request that no route arms starts under idleTimeout as well", - async () => { - using clock = idleClock(); - // No fetch handler: the 404 for an unknown path runs no handler. - using routesOnly = Bun.serve({ - port: 0, - idleTimeout: 1, - routes: { - "/forever": (req, server) => { - server.timeout(req, 0); - return new Response("ok"); - }, - }, - }); - // A WebSocket request that fetch answers does not pass the route start either. - using withWebSocket = Bun.serve({ + // One connection for each request: its status, then the close of the kept connection. + async function expectClosedAfterResponse( + port: number, + secure: boolean, + legs: { leg: string; status: number; head: string }[], + ) { + using clock = idleClock(); + const clients = legs.map(() => keepAliveClient(port, secure)); + try { + const statuses = await Promise.all(legs.map(({ head }, i) => clients[i].request(head))); + await clock.closedOrTwoSweeps(clients); + expect(legs.map(({ leg }, i) => ({ leg, status: statuses[i], closedByServer: clients[i].isClosed }))).toEqual( + legs.map(({ leg, status }) => ({ leg, status, closedByServer: true })), + ); + } finally { + for (const client of clients) client[Symbol.dispose](); + } + } + + it("the kept connection is under idleTimeout again, however the response ended", async () => { + using server = serverOfEveryEnding(); + // 0 never ends the connection and 60 ends it after a minute: idleTimeout + // is 1, so neither value may still be in force after the response. + await expectClosedAfterResponse(server.port, false, [ + { leg: "string body", status: 200, head: GET("/?seconds=0") }, + { leg: "string body, 60 s", status: 200, head: GET("/?seconds=60") }, + { leg: "HEAD", status: 200, head: "HEAD /?seconds=0 HTTP/1.1\r\nHost: localhost\r\n\r\n" }, + { leg: "304", status: 304, head: GET("/not-modified?seconds=0") }, + { leg: "stream body", status: 200, head: GET("/stream?seconds=0") }, + { leg: "stream body, 60 s", status: 200, head: GET("/stream?seconds=60") }, + { leg: "async handler", status: 200, head: GET("/async?seconds=0") }, + { leg: "route handler", status: 200, head: GET("/route?seconds=0") }, + { leg: "WebSocket request that fetch answers", status: 200, head: GET("/?seconds=0", upgrade) }, + ]); + }); + + it("the kept connection is under idleTimeout again after a large body, a file and an early answer", async () => { + using server = serverOfEveryEnding(); + // The handler answers before the rest of the request body arrives. + const upload = (framing: string) => `POST /?seconds=0 HTTP/1.1\r\nHost: localhost\r\n${framing}`; + await expectClosedAfterResponse(server.port, false, [ + { leg: "4 MiB Blob", status: 200, head: GET("/blob?seconds=0") }, + { leg: "Bun.file", status: 200, head: GET("/file?seconds=0") }, + { leg: "Bun.file of 2 MiB", status: 200, head: GET("/large-file?seconds=0") }, + { leg: "unfinished upload", status: 200, head: upload("Content-Length: 10\r\n\r\nhello") }, + { + leg: "unfinished chunked upload", + status: 200, + head: upload("Transfer-Encoding: chunked\r\n\r\n5\r\nhello\r\n"), + }, + ]); + }); + + it("the kept connection is under idleTimeout again over TLS", async () => { + using server = serverOfEveryEnding({ tls }); + await expectClosedAfterResponse(server.port, true, [ + { leg: "string body", status: 200, head: GET("/?seconds=0") }, + { leg: "Bun.file of 2 MiB", status: 200, head: GET("/large-file?seconds=0") }, + ]); + }); + + it("a value below idleTimeout does not end the kept connection early", async () => { + using clock = idleClock(); + // 30 s is eight sweeps away. 0 never ends an idle connection. + const legs = [30, 0].map(idleTimeout => { + const server = Bun.serve({ port: 0, - idleTimeout: 1, - websocket: { message() {} }, - fetch(req) { - if (!req.url.endsWith("/quiet")) return new Response("ok"); - return new Response( - new ReadableStream({ - start(controller) { - controller.enqueue("hi"); - }, - }), - ); + idleTimeout, + fetch(req, server) { + if (req.url.endsWith("/short")) server.timeout(req, 1); + return new Response("ok"); }, }); - using afterOverride = keepAliveClient(routesOnly.port); - using unknown = keepAliveClient(routesOnly.port); - using answered = keepAliveClient(withWebSocket.port); - using quietBody = keepAliveClient(withWebSocket.port); - // One write for both requests: between two writes a sweep can end the - // kept connection, which is under idleTimeout again after the first. - afterOverride.send(GET("/forever") + GET("/unknown")); - unknown.send(GET("/unknown")); - answered.send(GET("/", upgrade)); - quietBody.send(GET("/quiet", upgrade)); - const seen = { - afterOverride: [await afterOverride.response(), await afterOverride.response()], - unknown: await unknown.response(), - answered: await answered.response(), - quietBody: await quietBody.sees("hi"), - }; - const clients = [afterOverride, unknown, answered, quietBody]; - await clock.closedOrTwoSweeps(clients); - expect({ ...seen, closedByServer: clients.map(client => client.isClosed) }).toEqual({ - afterOverride: [200, 404], - unknown: 404, - answered: 200, - quietBody: true, - closedByServer: [true, true, true, true], - }); - }, - 30_000, - ); - - it.concurrent( - "a pending request keeps its idle timer, and the value holds while its response is pending", - async () => { - using clock = idleClock(); - const release = Promise.withResolvers(); - const handler = async (req: Request, server: Server) => { - // Never answers. - if (!req.url.endsWith("/held")) return new Promise(() => {}); - server.timeout(req, 0); - await release.promise; - return new Response("ok"); - }; - using server = Bun.serve({ port: 0, idleTimeout: 1, routes: { "/route/:kind": handler }, fetch: handler }); - using never = keepAliveClient(server.port); - using routeNever = keepAliveClient(server.port); - using held = keepAliveClient(server.port); - using routeHeld = keepAliveClient(server.port); - never.send(GET("/never")); - routeNever.send(GET("/route/never")); - const answers = [held.request(GET("/held")), routeHeld.request(GET("/route/held"))]; + return { idleTimeout, server, client: keepAliveClient(server.port) }; + }); + try { + const first = await Promise.all(legs.map(({ client }) => client.request(GET("/short")))); await clock.sweeps(1); - release.resolve(); - const heldStatuses = await Promise.all(answers); - await clock.closedOrTwoSweeps([never, routeNever]); - expect({ held: heldStatuses, neverClosedByServer: [never.isClosed, routeNever.isClosed] }).toEqual({ - held: [200, 200], - neverClosedByServer: [true, true], - }); - }, - 30_000, - ); - - // A FIFO with an open writer is the one file body that can stay quiet. Windows has no FIFO. - it.concurrent.skipIf(isWindows)( - "it holds while a Bun.file body is quiet", - async () => { - using clock = idleClock(); - using dir = tempDir("serve-timeout-quiet-file", {}); - const fifo = join(String(dir), "body.fifo"); - mkfifo(fifo); - // A writer that stays open: the body is quiet, not at its end. - const writer = openSync(fifo, "r+"); - try { - writeSync(writer, "first"); - using server = Bun.serve({ - port: 0, - idleTimeout: 1, - fetch(req, server) { - server.timeout(req, 0); - return new Response(Bun.file(fifo)); - }, - }); - using client = keepAliveClient(server.port); - client.send(GET("/")); - const first = await client.sees("first"); - await clock.sweeps(1); - writeSync(writer, "late"); - const late = await client.sees("late"); - expect({ first, late, closedByServer: client.isClosed }).toEqual({ - first: true, - late: true, - closedByServer: false, - }); - } finally { - closeSync(writer); + // A second request on the kept connection shows that the server did not close it. + const second = await Promise.all(legs.map(({ client }) => client.request(GET("/")))); + expect(legs.map(({ idleTimeout }, i) => ({ idleTimeout, first: first[i], second: second[i] }))).toEqual( + legs.map(({ idleTimeout }) => ({ idleTimeout, first: 200, second: 200 })), + ); + } finally { + for (const { client, server } of legs) { + client[Symbol.dispose](); + server.stop(true); } - }, - 30_000, - ); + } + }); - it.concurrent( - "it holds while its response is pending, under backpressure and after a drain", - async () => { - using clock = idleClock(); - // More than the kernel buffers between the two ends. - const size = 32 * 1024 * 1024; - using dir = tempDir("serve-timeout-pending-body", { "body.bin": Buffer.alloc(size, "f") }); - const burst = Buffer.alloc(size, "s"); - const quiet = Promise.withResolvers(); - const options = { - port: 0, - idleTimeout: 1, - fetch(req: Request, server: Server) { + it("a request that no route arms starts under idleTimeout as well", async () => { + using clock = idleClock(); + // No fetch handler: the 404 for an unknown path runs no handler. + using routesOnly = Bun.serve({ + port: 0, + idleTimeout: 1, + routes: { + "/forever": (req, server) => { server.timeout(req, 0); - if (req.url.endsWith("/file")) return new Response(Bun.file(join(String(dir), "body.bin"))); - return new Response( - new ReadableStream({ - type: "direct", - async pull(controller) { - controller.write(burst); - await controller.flush(); - await quiet.promise; - controller.write("late"); - controller.close(); - }, - }), - ); + return new Response("ok"); }, - }; - using plain = Bun.serve(options); - using secure = Bun.serve({ ...options, tls }); - using file = countingClient(plain.port, "/file"); - using tlsFile = countingClient(secure.port, "/file", true); - using stream = countingClient(plain.port, "/stream"); - // The files: stop reading with a quarter of the body across. - const stalledMidBody = await Promise.all( - [file, tlsFile].map(async client => { - const stalled = await client.until(() => client.state.received > size / 4); - client.socket.pause(); - return stalled; - }), - ); - // The stream: the client holds the whole burst, so the server has drained it. - const drained = await stream.until(() => stream.state.received > size); - await clock.sweeps(1); - file.socket.resume(); - tlsFile.socket.resume(); - quiet.resolve(); - const [deliveredWhole, tlsDeliveredWhole, late] = await Promise.all([ - file.until(() => file.state.received > size), - tlsFile.until(() => tlsFile.state.received > size), - stream.until(() => stream.state.tail.includes("late")), - ]); - expect({ stalledMidBody, deliveredWhole, tlsDeliveredWhole, drained, late }).toEqual({ - stalledMidBody: [true, true], - deliveredWhole: true, - tlsDeliveredWhole: true, - drained: true, - late: true, - }); - }, - 40_000, - ); + }, + }); + // A WebSocket request that fetch answers does not pass the route start either. + using withWebSocket = Bun.serve({ + port: 0, + idleTimeout: 1, + websocket: { message() {} }, + fetch(req) { + if (!req.url.endsWith("/quiet")) return new Response("ok"); + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue("hi"); + }, + }), + ); + }, + }); + using afterOverride = keepAliveClient(routesOnly.port); + using unknown = keepAliveClient(routesOnly.port); + using answered = keepAliveClient(withWebSocket.port); + using quietBody = keepAliveClient(withWebSocket.port); + // One write for both requests: between two writes a sweep can end the + // kept connection, which is under idleTimeout again after the first. + afterOverride.send(GET("/forever") + GET("/unknown")); + unknown.send(GET("/unknown")); + answered.send(GET("/", upgrade)); + quietBody.send(GET("/quiet", upgrade)); + const seen = { + afterOverride: [await afterOverride.response(), await afterOverride.response()], + unknown: await unknown.response(), + answered: await answered.response(), + quietBody: await quietBody.sees("hi"), + }; + const clients = [afterOverride, unknown, answered, quietBody]; + await clock.closedOrTwoSweeps(clients); + expect({ ...seen, closedByServer: clients.map(client => client.isClosed) }).toEqual({ + afterOverride: [200, 404], + unknown: 404, + answered: 200, + quietBody: true, + closedByServer: [true, true, true, true], + }); + }); + + it("a pending request keeps its idle timer, and the value holds while its response is pending", async () => { + using clock = idleClock(); + const release = Promise.withResolvers(); + const handler = async (req: Request, server: Server) => { + // Never answers. + if (!req.url.endsWith("/held")) return new Promise(() => {}); + server.timeout(req, 0); + await release.promise; + return new Response("ok"); + }; + using server = Bun.serve({ port: 0, idleTimeout: 1, routes: { "/route/:kind": handler }, fetch: handler }); + using never = keepAliveClient(server.port); + using routeNever = keepAliveClient(server.port); + using held = keepAliveClient(server.port); + using routeHeld = keepAliveClient(server.port); + never.send(GET("/never")); + routeNever.send(GET("/route/never")); + const answers = [held.request(GET("/held")), routeHeld.request(GET("/route/held"))]; + await clock.sweeps(1); + release.resolve(); + const heldStatuses = await Promise.all(answers); + await clock.closedOrTwoSweeps([never, routeNever]); + expect({ held: heldStatuses, neverClosedByServer: [never.isClosed, routeNever.isClosed] }).toEqual({ + held: [200, 200], + neverClosedByServer: [true, true], + }); + }); - it.concurrent( - "a value below idleTimeout ends a stalled Bun.file body", - async () => { - using clock = idleClock(); - // More than the kernel buffers between the two ends. - const size = 32 * 1024 * 1024; - using dir = tempDir("serve-timeout-short-file", { "body.bin": Buffer.alloc(size, "f") }); - const cut = Promise.withResolvers(); + // A FIFO with an open writer is the one file body that can stay quiet. Windows has no FIFO. + it.skipIf(isWindows)("it holds while a Bun.file body is quiet", async () => { + using clock = idleClock(); + using dir = tempDir("serve-timeout-quiet-file", {}); + const fifo = join(String(dir), "body.fifo"); + mkfifo(fifo); + // A writer that stays open: the body is quiet, not at its end. + const writer = openSync(fifo, "r+"); + try { + writeSync(writer, "first"); using server = Bun.serve({ port: 0, - idleTimeout: 30, + idleTimeout: 1, fetch(req, server) { - server.timeout(req, 1); - req.signal.addEventListener("abort", () => cut.resolve()); - return new Response(Bun.file(join(String(dir), "body.bin"))); + server.timeout(req, 0); + return new Response(Bun.file(fifo)); }, }); - using client = countingClient(server.port, "/"); - const stalled = await client.until(() => client.state.received > 0); - client.socket.pause(); - // The server ends the request in its next sweep. 30 s is eight sweeps away. - const cutInTwoSweeps = await Promise.race([cut.promise.then(() => true), clock.sweeps(2).then(() => false)]); - // A client that does not read does not see the close. Read what is left. - client.socket.resume(); - const deliveredWhole = await client.until(() => client.state.received > size); - expect({ stalled, cutInTwoSweeps, deliveredWhole, closedByServer: client.state.closedByServer }).toEqual({ - stalled: true, - cutInTwoSweeps: true, - deliveredWhole: false, - closedByServer: true, + using client = keepAliveClient(server.port); + client.send(GET("/")); + const first = await client.sees("first"); + await clock.sweeps(1); + writeSync(writer, "late"); + const late = await client.sees("late"); + expect({ first, late, closedByServer: client.isClosed }).toEqual({ + first: true, + late: true, + closedByServer: false, }); - }, - 30_000, - ); + } finally { + closeSync(writer); + } + }); - it.concurrent( - "it holds until a response that ended under backpressure has drained", - async () => { - using clock = idleClock(); - // More than the kernel buffers between the two ends. - const chunk = Buffer.alloc(32 * 1024 * 1024, "a"); - const written: Record = {}; - const options = { - port: 0, - idleTimeout: 1, - fetch(req: Request, server: Server) { - server.timeout(req, 60); - const response = (written[new URL(req.url).pathname.slice(1)] = { bytes: 0, backedUp: false }); - return new Response( - new ReadableStream({ - type: "direct", - pull(controller) { - // Write until the socket pushes back, then close without a wait: - // the stream ends while the server still holds the rest. - while (!response.backedUp && response.bytes < 4 * chunk.length) { - response.backedUp = controller.write(chunk) instanceof Promise; - response.bytes += chunk.length; - } - controller.close(); - }, - }), - ); - }, - }; - using plain = Bun.serve(options); - using secure = Bun.serve({ ...options, tls }); - const legs = [ - { leg: "TCP", client: countingClient(plain.port, "/TCP") }, - { leg: "TLS", client: countingClient(secure.port, "/TLS", true) }, - ]; - try { - // Stop reading for one sweep, with most of the body still to come. - const stalled = await Promise.all( - legs.map(async ({ client }) => { - const gotFirstBytes = await client.until(() => client.state.received > 0); - client.socket.pause(); - return gotFirstBytes; + it("it holds while its response is pending, under backpressure and after a drain", async () => { + using clock = idleClock(); + // More than the kernel buffers between the two ends. + const size = 16 * 1024 * 1024; + using dir = tempDir("serve-timeout-pending-body", { "body.bin": Buffer.alloc(size, "f") }); + // More than the socket takes in one write. + const burst = Buffer.alloc(size / 2, "s"); + const quiet = Promise.withResolvers(); + const options = { + port: 0, + idleTimeout: 1, + fetch(req: Request, server: Server) { + server.timeout(req, 0); + if (req.url.endsWith("/file")) return new Response(Bun.file(join(String(dir), "body.bin"))); + return new Response( + new ReadableStream({ + type: "direct", + async pull(controller) { + controller.write(burst); + await controller.flush(); + await quiet.promise; + controller.write("late"); + controller.close(); + }, }), ); - await clock.sweeps(1); - for (const { client } of legs) client.socket.resume(); - const deliveredWhole = await Promise.all( - legs.map(({ leg, client }) => - client.until(() => client.state.received > written[leg].bytes && client.state.tail.endsWith("0\r\n\r\n")), - ), - ); - // Drained: the kept connection is under idleTimeout again. - await clock.closedOrTwoSweeps(legs.map(({ client }) => client)); - expect( - legs.map(({ leg, client }, i) => ({ - leg, - stalled: stalled[i], - backedUp: written[leg].backedUp, - deliveredWhole: deliveredWhole[i], - closedByServer: client.state.closedByServer, - })), - ).toEqual( - legs.map(({ leg }) => ({ leg, stalled: true, backedUp: true, deliveredWhole: true, closedByServer: true })), + }, + }; + using plain = Bun.serve(options); + using secure = Bun.serve({ ...options, tls }); + using file = countingClient(plain.port, "/file"); + using tlsFile = countingClient(secure.port, "/file", true); + using stream = countingClient(plain.port, "/stream"); + // The files: stop reading part-way, with most of the body still to come. + const stalledMidBody = await Promise.all( + [file, tlsFile].map(async client => { + const stalled = await client.until(() => client.state.received > size / 8); + client.socket.pause(); + return stalled; + }), + ); + // The stream: the client holds the whole burst, so the server has drained it. + const drained = await stream.until(() => stream.state.received > burst.length); + await clock.sweeps(1); + file.socket.resume(); + tlsFile.socket.resume(); + quiet.resolve(); + const [deliveredWhole, tlsDeliveredWhole, late] = await Promise.all([ + file.until(() => file.state.received > size), + tlsFile.until(() => tlsFile.state.received > size), + stream.until(() => stream.state.tail.includes("late")), + ]); + expect({ stalledMidBody, deliveredWhole, tlsDeliveredWhole, drained, late }).toEqual({ + stalledMidBody: [true, true], + deliveredWhole: true, + tlsDeliveredWhole: true, + drained: true, + late: true, + }); + }); + + it("a value below idleTimeout ends a stalled Bun.file body", async () => { + using clock = idleClock(); + // More than the kernel buffers between the two ends. + const size = 16 * 1024 * 1024; + using dir = tempDir("serve-timeout-short-file", { "body.bin": Buffer.alloc(size, "f") }); + const cut = Promise.withResolvers(); + using server = Bun.serve({ + port: 0, + idleTimeout: 30, + fetch(req, server) { + server.timeout(req, 1); + req.signal.addEventListener("abort", () => cut.resolve()); + return new Response(Bun.file(join(String(dir), "body.bin"))); + }, + }); + using client = countingClient(server.port, "/"); + const stalled = await client.until(() => client.state.received > 0); + client.socket.pause(); + // The server ends the request in its next sweep. 30 s is eight sweeps away. + const cutInTwoSweeps = await Promise.race([cut.promise.then(() => true), clock.sweeps(2).then(() => false)]); + // A client that does not read does not see the close. Read what is left. + client.socket.resume(); + const deliveredWhole = await client.until(() => client.state.received > size); + expect({ stalled, cutInTwoSweeps, deliveredWhole, closedByServer: client.state.closedByServer }).toEqual({ + stalled: true, + cutInTwoSweeps: true, + deliveredWhole: false, + closedByServer: true, + }); + }); + + it("it holds until a response that ended under backpressure has drained", async () => { + using clock = idleClock(); + // More than the kernel buffers between the two ends. + const chunk = Buffer.alloc(16 * 1024 * 1024, "a"); + const written: Record = {}; + const options = { + port: 0, + idleTimeout: 1, + fetch(req: Request, server: Server) { + server.timeout(req, 60); + const response = (written[new URL(req.url).pathname.slice(1)] = { bytes: 0, backedUp: false }); + return new Response( + new ReadableStream({ + type: "direct", + pull(controller) { + // Write until the socket pushes back, then close without a wait: + // the stream ends while the server still holds the rest. + while (!response.backedUp && response.bytes < 4 * chunk.length) { + response.backedUp = controller.write(chunk) instanceof Promise; + response.bytes += chunk.length; + } + controller.close(); + }, + }), ); - } finally { - for (const { client } of legs) client[Symbol.dispose](); - } - }, - 40_000, - ); + }, + }; + using plain = Bun.serve(options); + using secure = Bun.serve({ ...options, tls }); + const legs = [ + { leg: "TCP", client: countingClient(plain.port, "/TCP") }, + { leg: "TLS", client: countingClient(secure.port, "/TLS", true) }, + ]; + try { + // Stop reading for one sweep, with most of the body still to come. + const stalled = await Promise.all( + legs.map(async ({ client }) => { + const gotFirstBytes = await client.until(() => client.state.received > 0); + client.socket.pause(); + return gotFirstBytes; + }), + ); + await clock.sweeps(1); + for (const { client } of legs) client.socket.resume(); + const deliveredWhole = await Promise.all( + legs.map(({ leg, client }) => + client.until(() => client.state.received > written[leg].bytes && client.state.tail.endsWith("0\r\n\r\n")), + ), + ); + // Drained: the kept connection is under idleTimeout again. + await clock.closedOrTwoSweeps(legs.map(({ client }) => client)); + expect( + legs.map(({ leg, client }, i) => ({ + leg, + stalled: stalled[i], + backedUp: written[leg].backedUp, + deliveredWhole: deliveredWhole[i], + closedByServer: client.state.closedByServer, + })), + ).toEqual( + legs.map(({ leg }) => ({ leg, stalled: true, backedUp: true, deliveredWhole: true, closedByServer: true })), + ); + } finally { + for (const { client } of legs) client[Symbol.dispose](); + } + }); }); it.concurrent( From bc80c1274030fc731598631980aadc9b83bce8b4 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Thu, 8 Oct 2026 07:02:31 +0000 Subject: [PATCH 3/6] Bun.serve: drop a server.timeout override when its request upgrades endUpgradeHandshake() marks the 101 done, and the hand-back of a server.timeout(req, N) override in markDone() sends the cork. The 101 then left in a send of its own, and the first frames that open() writes needed a second one. The socket leaves HTTP at that point and the WebSocket arms its own timeout, so the override is dropped there and nothing is handed back. The 101 and the first frame share one send again. --- packages/bun-uws/src/HttpResponse.h | 6 ++- test/js/bun/http/serve-syscall-fault.test.ts | 56 ++++++++++++++++++++ 2 files changed, 60 insertions(+), 2 deletions(-) diff --git a/packages/bun-uws/src/HttpResponse.h b/packages/bun-uws/src/HttpResponse.h index f4a47eceef04..6e9dc4dcdf8f 100644 --- a/packages/bun-uws/src/HttpResponse.h +++ b/packages/bun-uws/src/HttpResponse.h @@ -229,12 +229,14 @@ struct HttpResponse : public AsyncSocket { * after: the connection close gate does not apply (Connection: close, * HTTP/1.0 and close-when-idle describe the HTTP connection, not the * WebSocket that takes over the socket), and the cork stays so the - * handshake batches with the first frames written from open(). */ + * handshake batches with the first frames written from open(). A + * setTimeout() override is dropped for that reason too: its hand-back + * in markDone() sends the cork, and the WebSocket arms its own timeout. */ void endUpgradeHandshake() { HttpResponseData *httpResponseData = getHttpResponseData(); writeMark(); Super::write("\r\n", 2); - httpResponseData->state |= HttpResponseData::HTTP_END_CALLED; + httpResponseData->state = (httpResponseData->state | HttpResponseData::HTTP_END_CALLED) & ~HttpResponseData::HTTP_TIMEOUT_OVERRIDDEN; httpResponseData->markDone(this); } diff --git a/test/js/bun/http/serve-syscall-fault.test.ts b/test/js/bun/http/serve-syscall-fault.test.ts index f6083a7a11b1..fa86bcd1adb0 100644 --- a/test/js/bun/http/serve-syscall-fault.test.ts +++ b/test/js/bun/http/serve-syscall-fault.test.ts @@ -165,6 +165,62 @@ describe.skipIf(skip)("Bun.serve under injected syscall faults", () => { expect(proc.exitCode).toBe(0); }); + test("server.timeout(req, N) before server.upgrade(req) leaves the 101 and the first frame in one send", async () => { + const { proc, port } = await spawnServer(/* js */ ` + const { socketFaultInjection: fault } = require("bun:internal-for-testing"); + const s = Bun.serve({ port: 0, hostname: "127.0.0.1", + websocket: { + open(ws) { + ws.send("first-frame"); + // Two loop turns later every send of this turn was tried. Then the connection goes. + setImmediate(() => setImmediate(() => ws.terminate())); + }, + message() {}, + }, + fetch(req, server) { + server.timeout(req, 60); + // Only the first send from here on moves bytes: a first frame that is not in it never leaves. + fault.set({ syscall: "send", action: "zero", after: 1, repeat: -1 }); + if (server.upgrade(req)) return; + return new Response("no upgrade", { status: 400 }); + } }); + console.log(s.port); + process.on("SIGTERM", () => { fault.clear(); s.stop(true); process.exit(0); }); + `); + try { + // An unmasked text frame of 11 bytes. + const frame = "\x81\x0bfirst-frame"; + // The wire up to the frame, or up to the close if the frame did not come. + const wire = Promise.withResolvers(); + const socket = net.connect(port, "127.0.0.1"); + let received = ""; + socket.on("error", () => {}); + socket.on("data", chunk => { + received += chunk.toString("latin1"); + if (received.endsWith(frame)) wire.resolve(received); + }); + socket.on("close", () => wire.resolve(received)); + socket.write( + "GET / HTTP/1.1\r\nHost: x\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n" + + "Sec-WebSocket-Version: 13\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n\r\n", + ); + try { + const got = await wire.promise; + expect({ switched: got.startsWith("HTTP/1.1 101 "), firstFrame: got.endsWith(frame) }).toEqual({ + switched: true, + firstFrame: true, + }); + } finally { + socket.destroy(); + } + } finally { + proc.kill("SIGTERM"); + await proc.exited; + } + expect(proc.signalCode).toBeNull(); + expect(proc.exitCode).toBe(0); + }); + test("client abort under server-side 1-byte sends: every response reaches a terminal state", async () => { const { proc, port } = await spawnServer(/* js */ ` const { socketFaultInjection: fault } = require("bun:internal-for-testing"); From 4c0c1d4004a6b08dbff5291f6e9407a7db17c1c8 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Thu, 8 Oct 2026 08:28:08 +0000 Subject: [PATCH 4/6] test(serve): end the fault-injection legs on a condition, not on loop turns The cork leg clears the fault when a socket of the same loop, armed with a 1 s timeout just before the forced sweep, has timed out. The upgrade leg ends the connection when the client, which has the 101 by then, sends a frame. --- test/js/bun/http/serve-syscall-fault.test.ts | 50 ++++++++++++-------- 1 file changed, 29 insertions(+), 21 deletions(-) diff --git a/test/js/bun/http/serve-syscall-fault.test.ts b/test/js/bun/http/serve-syscall-fault.test.ts index fa86bcd1adb0..a5a70be4574f 100644 --- a/test/js/bun/http/serve-syscall-fault.test.ts +++ b/test/js/bun/http/serve-syscall-fault.test.ts @@ -124,21 +124,26 @@ describe.skipIf(skip)("Bun.serve under injected syscall faults", () => { test("server.timeout(req, N) holds when the socket takes none of a response that ended in the cork buffer", async () => { const { proc, port } = await spawnServer(/* js */ ` const { socketFaultInjection: fault, runSocketTimeoutSweepSoon } = require("bun:internal-for-testing"); - const s = Bun.serve({ port: 0, hostname: "127.0.0.1", idleTimeout: 1, - fetch(req, server) { - server.timeout(req, 60); - // The response ends with all of its bytes in the cork buffer, and no send moves one. - fault.set({ syscall: "send", action: "zero", repeat: -1 }); - fault.set({ syscall: "writev", action: "zero", repeat: -1 }); - // One idle sweep runs over the connection, then the sends work again. - setImmediate(() => { + // A socket of this process that sends nothing. Its 1 s timeout fires in the next sweep of the loop. + const listener = Bun.listen({ port: 0, hostname: "127.0.0.1", socket: { data() {} } }); + let afterSweep = () => {}; + Bun.connect({ port: listener.port, hostname: "127.0.0.1", + socket: { data() {}, timeout() { afterSweep(); } } }).then(witness => { + const s = Bun.serve({ port: 0, hostname: "127.0.0.1", idleTimeout: 1, + fetch(req, server) { + server.timeout(req, 60); + // The response ends with all of its bytes in the cork buffer, and no send moves one. + fault.set({ syscall: "send", action: "zero", repeat: -1 }); + fault.set({ syscall: "writev", action: "zero", repeat: -1 }); + // One sweep runs over both connections. The sends work again when it has run. + afterSweep = () => fault.clear(); + witness.timeout(1); runSocketTimeoutSweepSoon(); - setImmediate(() => setImmediate(() => fault.clear())); - }); - return new Response("tail"); - } }); - console.log(s.port); - process.on("SIGTERM", () => { fault.clear(); s.stop(true); process.exit(0); }); + return new Response("tail"); + } }); + console.log(s.port); + process.on("SIGTERM", () => { fault.clear(); s.stop(true); process.exit(0); }); + }); `); try { // The wire up to the end of the response, or up to the close if the server cut it. @@ -170,12 +175,9 @@ describe.skipIf(skip)("Bun.serve under injected syscall faults", () => { const { socketFaultInjection: fault } = require("bun:internal-for-testing"); const s = Bun.serve({ port: 0, hostname: "127.0.0.1", websocket: { - open(ws) { - ws.send("first-frame"); - // Two loop turns later every send of this turn was tried. Then the connection goes. - setImmediate(() => setImmediate(() => ws.terminate())); - }, - message() {}, + open(ws) { ws.send("first-frame"); }, + // The client sends a frame when it has the 101. A first frame that is not out by then never leaves. + message(ws) { ws.terminate(); }, }, fetch(req, server) { server.timeout(req, 60); @@ -194,10 +196,16 @@ describe.skipIf(skip)("Bun.serve under injected syscall faults", () => { const wire = Promise.withResolvers(); const socket = net.connect(port, "127.0.0.1"); let received = ""; + let asked = false; socket.on("error", () => {}); socket.on("data", chunk => { received += chunk.toString("latin1"); - if (received.endsWith(frame)) wire.resolve(received); + if (received.endsWith(frame)) return wire.resolve(received); + if (!asked && received.includes("\r\n\r\n")) { + asked = true; + // A masked text frame "done" with a zero mask: the server ends the connection. + socket.write(Buffer.from([0x81, 0x84, 0, 0, 0, 0, 0x64, 0x6f, 0x6e, 0x65])); + } }); socket.on("close", () => wire.resolve(received)); socket.write( From 109b8c05c13ffba528012fce3f53aded7a55e5a2 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sun, 18 Oct 2026 19:55:53 +0000 Subject: [PATCH 5/6] Bun.serve: leave the timer arms of a file body to uWS, and test the WebSocket-handshake entry FileResponseStream and StaticRoute::on_writable armed the idle timer before each write. uWS arms it in write(), in tryEnd() and at the end of onWritable, so those three calls go. The file stream still arms once at its start, where the timer can be off. A GET with WebSocket handshake headers that fetch answers with a Response reaches fetch through an entry that stores no idleTimeout. The response ran under the 10 s of a new connection. Since the request head now starts every response under the server's value, it follows idleTimeout. A test holds that for idleTimeout 0. --- src/runtime/server/FileResponseStream.rs | 3 +-- src/runtime/server/StaticRoute.rs | 2 -- test/js/bun/http/serve.test.ts | 32 +++++++++++++++++++++++- 3 files changed, 32 insertions(+), 5 deletions(-) diff --git a/src/runtime/server/FileResponseStream.rs b/src/runtime/server/FileResponseStream.rs index bfb31eafb1ff..a9dadf27236e 100644 --- a/src/runtime/server/FileResponseStream.rs +++ b/src/runtime/server/FileResponseStream.rs @@ -186,6 +186,7 @@ impl FileResponseStream { let this_ref = unsafe { &*this }; let resp = this_ref.resp.get(); + // Arm only: the value is the response's (`server.timeout()` may have set it). The timer can be off here: uWS clears it at the last request-body chunk, the WebSocket-upgrade fetch entry never arms it, and sendfile(2) makes no uWS write that would arm it. resp.reset_timeout(); resp.on_aborted( |p: *mut FileResponseStream, r| { @@ -310,7 +311,6 @@ impl FileResponseStream { } let resp = self.resp.get(); - resp.reset_timeout(); if state == ReadState::Eof { self.insert_state(State::RESPONSE_DONE); @@ -387,7 +387,6 @@ impl FileResponseStream { self.finish(); return true; } - self.resp.get().reset_timeout(); self.hold_read_ref(); // A paused POSIX reader ignores `read()`. self.reader_mut().unpause(); diff --git a/src/runtime/server/StaticRoute.rs b/src/runtime/server/StaticRoute.rs index 72736a744967..c222855e6981 100644 --- a/src/runtime/server/StaticRoute.rs +++ b/src/runtime/server/StaticRoute.rs @@ -400,8 +400,6 @@ impl StaticRoute { } fn on_writable(this: ThisPtr, write_offset: u64, resp: AnyResponse) -> bool { - resp.reset_timeout(); - if !this.on_writable_bytes(write_offset, resp) { return false; } diff --git a/test/js/bun/http/serve.test.ts b/test/js/bun/http/serve.test.ts index e355e2d0e1f4..67a4be5ff319 100644 --- a/test/js/bun/http/serve.test.ts +++ b/test/js/bun/http/serve.test.ts @@ -3618,7 +3618,7 @@ describe("server.timeout(req, seconds) ends with its response", () => { let stopped = false; const sweeps = async (count: number) => { // Only an answered round trip counts: a sweep closed that connection. - for (let i = 0; i < count && !stopped; ) { + for (let i = 0; i < count && !stopped;) { using idle = keepAliveClient(server.port); if ((await idle.request(GET("/"))) === 200 && !stopped) { i++; @@ -3845,6 +3845,36 @@ describe("server.timeout(req, seconds) ends with its response", () => { }); }); + it("a WebSocket request that fetch answers is not under the 10 s of a new connection", async () => { + using clock = idleClock(); + const release = Promise.withResolvers(); + let pulls = 0; + // 0 never ends the connection. 10 s is three sweeps. + using server = Bun.serve({ + port: 0, + idleTimeout: 0, + websocket: { message() {} }, + fetch: () => + new Response( + new ReadableStream({ + async pull(controller) { + if (pulls++ === 0) return controller.enqueue("first"); + await release.promise; + controller.enqueue("late"); + controller.close(); + }, + }), + ), + }); + using client = keepAliveClient(server.port); + client.send(GET("/", upgrade)); + const first = await client.sees("first"); + await clock.sweeps(3); + release.resolve(); + const late = await client.sees("late"); + expect({ first, late }).toEqual({ first: true, late: true }); + }); + it("a pending request keeps its idle timer, and the value holds while its response is pending", async () => { using clock = idleClock(); const release = Promise.withResolvers(); From 2955e4d3473c17ffc7724f3f25c775c48ca231d3 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Thu, 5 Nov 2026 12:05:31 +0000 Subject: [PATCH 6/6] uws: the timeout hand-back arms the server's idle timeout itself endTimeoutOverrideIfDrained() left the arm to its caller, so every response end path had to arm after markDone(). finishResponseInCorkBuffer() arms before it. The hand-back now arms the value it restores, so the order in an end path does not matter, and the Content-Length path of internalEnd() is as on main again. --- packages/bun-uws/src/HttpContext.h | 3 +-- packages/bun-uws/src/HttpResponse.h | 12 +++++------- packages/bun-uws/src/HttpResponseData.h | 1 - 3 files changed, 6 insertions(+), 10 deletions(-) diff --git a/packages/bun-uws/src/HttpContext.h b/packages/bun-uws/src/HttpContext.h index f7ee0968df79..b2613c8cf83d 100644 --- a/packages/bun-uws/src/HttpContext.h +++ b/packages/bun-uws/src/HttpContext.h @@ -973,8 +973,7 @@ struct HttpContext { /* Drain any socket buffer, this might empty our backpressure and thus finish the request */ asyncSocket->flush(); - /* The tail of a response that ended under backpressure may be out now; - * the resetTimeout() below arms what this leaves. */ + /* The tail of a response that ended under backpressure may be out now. */ reinterpret_cast *>(s)->endTimeoutOverrideIfDrained(); /* node:http compat: reads were paused while pipelined responses were diff --git a/packages/bun-uws/src/HttpResponse.h b/packages/bun-uws/src/HttpResponse.h index 8f880159c524..cea2ec15fe7e 100644 --- a/packages/bun-uws/src/HttpResponse.h +++ b/packages/bun-uws/src/HttpResponse.h @@ -67,7 +67,8 @@ struct HttpResponse : public AsyncSocket { /* A setTimeout() override ends when its response is complete and * hasFullyDrained(): a response that ended under backpressure keeps it - * until onWritable has drained the tail. Does not arm. */ + * until onWritable has drained the tail. Arms the server's value when it + * ends the override, so an end path's own arm may come before or after. */ void endTimeoutOverrideIfDrained() { auto* data = getHttpResponseData(); if ((data->state & (HttpResponseData::HTTP_TIMEOUT_OVERRIDDEN | HttpResponseData::HTTP_RESPONSE_PENDING)) == HttpResponseData::HTTP_TIMEOUT_OVERRIDDEN) [[unlikely]] { @@ -77,6 +78,7 @@ struct HttpResponse : public AsyncSocket { if (Super::hasFullyDrained()) { data->state &= ~HttpResponseData::HTTP_TIMEOUT_OVERRIDDEN; data->idleTimeoutInForce = HttpContext::getSocketContextDataS((us_socket_t *) this)->idleTimeout; + Super::timeout(data->idleTimeoutInForce); } } } @@ -365,21 +367,17 @@ struct HttpResponse : public AsyncSocket { /* Success is when we wrote the entire thing without any failures */ bool success = written == data.length() && !failed; + /* Reset the timeout on each tryEnd */ + this->resetTimeout(); /* Remove onAborted function if we reach the end */ if (httpResponseData->offset == totalSize) { httpResponseData->markDone(this); - /* After markDone(), which can end a setTimeout() override, and - * before the close gate, which can destruct the ext. */ - this->resetTimeout(); /* We need to check if we should close this socket here now */ if (uncorkCompletedResponse()) { closeIfDoneAndMarked(httpResponseData); } - } else { - /* Reset the timeout on each tryEnd */ - this->resetTimeout(); } return success; diff --git a/packages/bun-uws/src/HttpResponseData.h b/packages/bun-uws/src/HttpResponseData.h index b9d627d99b5d..80ae9c849bb5 100644 --- a/packages/bun-uws/src/HttpResponseData.h +++ b/packages/bun-uws/src/HttpResponseData.h @@ -58,7 +58,6 @@ struct HttpResponseData : AsyncSocketData, HttpParser { /* We are done with this request */ this->state &= ~HttpResponseData::HTTP_RESPONSE_PENDING; - /* Every caller arms the timer after markDone(), so it arms the value this leaves. */ uwsRes->endTimeoutOverrideIfDrained(); HttpResponseData *httpResponseData = uwsRes->getHttpResponseData();