diff --git a/packages/bun-uws/src/App.h b/packages/bun-uws/src/App.h index 5209ab836cdc..0c6d62894085 100644 --- a/packages/bun-uws/src/App.h +++ b/packages/bun-uws/src/App.h @@ -723,6 +723,10 @@ struct TemplatedApp { void setOnSocketClosed(HttpContextData::OnSocketClosedCallback onClose) { httpContext->getSocketContextData()->onSocketClosed = onClose; } + void setOnSocketOpen(HttpContextData::OnSocketOpenCallback onOpen, void *userData) { + httpContext->getSocketContextData()->onSocketOpen = onOpen; + httpContext->getSocketContextData()->onSocketOpenUserData = userData; + } void setOnSocketDrain(HttpContextData::OnSocketDrainCallback onDrain) { httpContext->getSocketContextData()->onSocketDrain = onDrain; } diff --git a/packages/bun-uws/src/HttpContext.h b/packages/bun-uws/src/HttpContext.h index 75248bed3f5f..d8bc5c2845a5 100644 --- a/packages/bun-uws/src/HttpContext.h +++ b/packages/bun-uws/src/HttpContext.h @@ -179,6 +179,10 @@ struct HttpContext { for (auto &f : httpContextData->filterHandlers) { f((HttpResponse *) s, 1); } + + if (httpContextData->onSocketOpen) { + httpContextData->onSocketOpen(httpContextData->onSocketOpenUserData, SSL, s); + } } } @@ -194,6 +198,10 @@ struct HttpContext { for (auto &f : httpContextData->filterHandlers) { f((HttpResponse *) s, 1); } + + if (httpContextData->onSocketOpen) { + httpContextData->onSocketOpen(httpContextData->onSocketOpenUserData, SSL, s); + } } return s; diff --git a/packages/bun-uws/src/HttpContextData.h b/packages/bun-uws/src/HttpContextData.h index 538537c92c8f..18229dc7bd6d 100644 --- a/packages/bun-uws/src/HttpContextData.h +++ b/packages/bun-uws/src/HttpContextData.h @@ -48,6 +48,7 @@ struct alignas(16) HttpContextData { using OnSocketUpgradedCallback = void (*)(void* userData, int is_ssl, struct us_socket_t *rawSocket); using OnClientErrorCallback = MoveOnlyFunction; using OnSocketClosedCallback = void (*)(void* userData, int is_ssl, struct us_socket_t *rawSocket); + using OnSocketOpenCallback = void (*)(void* ctxUserData, int is_ssl, struct us_socket_t *rawSocket); MoveOnlyFunction missingServerNameHandler; @@ -68,6 +69,8 @@ struct alignas(16) HttpContextData { OnSocketDataCallback onSocketData = nullptr; OnSocketUpgradedCallback onSocketUpgraded = nullptr; OnClientErrorCallback onClientError = nullptr; + OnSocketOpenCallback onSocketOpen = nullptr; + void *onSocketOpenUserData = nullptr; uint64_t maxHeaderSize = 0; // 0 means no limit diff --git a/src/js/internal/http.ts b/src/js/internal/http.ts index a44142705651..dc69bfefdb6c 100644 --- a/src/js/internal/http.ts +++ b/src/js/internal/http.ts @@ -26,6 +26,7 @@ const { useStrictMethodValidation: boolean, maxHeaderSize: number, onClientError: (ssl: boolean, socket: any, errorCode: number, rawPacket: ArrayBuffer) => undefined, + onSocketOpen?: (ssl: boolean, socket: any) => undefined, ) => void; getCompleteWebRequestOrResponseBodyValueAsArrayBuffer: (arg: any) => ArrayBuffer | undefined; drainMicrotasks: () => void; diff --git a/src/js/node/_http_server.ts b/src/js/node/_http_server.ts index 21c67abb3d4b..9d88f584ca30 100644 --- a/src/js/node/_http_server.ts +++ b/src/js/node/_http_server.ts @@ -27,7 +27,6 @@ const { kRealListen, tlsSymbol, optionsSymbol, - kDeferredTimeouts, kDeprecatedReplySymbol, headerStateSymbol, NodeHTTPHeaderState, @@ -51,7 +50,6 @@ const { eofInProgress, runSymbol, drainMicrotasks, - setServerIdleTimeout, setServerCustomOptions, getMaxHTTPHeaderSize, fakeSocketSymbol, @@ -88,6 +86,7 @@ const kEmptyBuffer = Buffer.alloc(0); const ObjectKeys = Object.keys; const MathMin = Math.min; const MathFloor = Math.floor; +const DateNow = Date.now; let cluster; @@ -241,6 +240,7 @@ function Server(options, callback): void { this._unref = false; this.maxRequestsPerSocket = 0; this.maxHeadersCount = null; + this.timeout = 0; this[kInternalSocketData] = undefined; this[kTrackedConnections] = new Set(); this[tlsSymbol] = null; @@ -324,13 +324,12 @@ function rethrowUncaught(err) { // Like Node.js's setupConnectionsTracking: each 'listening' event replaces // the connections-checking interval timer (used by the headers/request // timeout machinery) and destroys the previous one. -function noopConnectionsCheck() {} function setupConnectionsTracking(this: any) { if (this[kConnectionsCheckingInterval]) { clearInterval(this[kConnectionsCheckingInterval]); } const delay = this.connectionsCheckingInterval || 30_000; - this[kConnectionsCheckingInterval] = setInterval(noopConnectionsCheck, delay); + this[kConnectionsCheckingInterval] = setInterval(checkConnections.bind(this), delay); this[kConnectionsCheckingInterval].unref(); } @@ -593,6 +592,17 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort if (!socket) { socket = new NodeHTTPServerSocket(server, socketHandle, !!tls); } + // A parsed request head is activity: headers just completed, the + // request-start anchor advances for a keep-alive reuse, and the + // inactivity timer switches back from keepAliveTimeout to + // server.timeout (like Node.js's parserOnIncoming). + socket[kHeadersCompleted] = true; + if (socket[kRequestStart] === 0) socket[kRequestStart] = DateNow(); + if (socket.timeout !== server.timeout) { + socket.setTimeout(server.timeout || 0); + } else { + socket._unrefTimer(); + } const http_req = new RequestClass(kHandle, url, method, headersObject, headersArray, handle, hasBody, socket); if (isAncientHTTP) { @@ -646,7 +656,7 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort if (!requestShouldKeepAlive(http_req)) { http_res[kMustCloseConnection] = true; } - http_res.once("finish", endSocketOnFinishIfNeeded.bind(undefined, socket, http_res)); + http_res.once("finish", resOnFinish.bind(undefined, server, socket, http_res)); if (hasObserver("http")) { startPerf(http_res, kServerResponseStatistics, { @@ -665,6 +675,14 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort setIsNextIncomingMessageHTTPS(prevIsNextIncomingMessageHTTPS); handle.onabort = onServerRequestEvent.bind(socket); + // Like Node.js, the requestTimeout clock runs until the incoming + // request (not the response) is complete: a body-less request is + // complete at the head, and a body stops the clock at its 'end'. + if (!hasBody) { + socket[kRequestStart] = 0; + } else { + http_req.once("end", clearIncomingRequestTimer.bind(undefined, socket)); + } // start buffering data if any, the user will need to resume() or .on("data") to read it if (hasBody) { handle.pause(); @@ -692,7 +710,8 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort } } - if (isSocketNew && !reachedRequestsLimit) { + if (!socket[kConnectionEmitted] && !reachedRequestsLimit) { + socket[kConnectionEmitted] = true; server.emit("connection", socket); } @@ -892,6 +911,7 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort true, typeof this.maxHeaderSize !== "undefined" ? this.maxHeaderSize : getMaxHTTPHeaderSize(), onServerClientError.bind(this), + onServerSocketOpen.bind(this), ); if (this?._unref) { @@ -902,25 +922,13 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort this.once("listening", onListen); } - if (this[kDeferredTimeouts]) { - for (const { msecs, callback } of this[kDeferredTimeouts]) { - this.setTimeout(msecs, callback); - } - delete this[kDeferredTimeouts]; - } - setTimeout(emitListeningNextTick, 1, this, this[serverSymbol]?.hostname, this[serverSymbol]?.port); } }; Server.prototype.setTimeout = function (msecs, callback) { - const server = this[serverSymbol]; - if (server) { - setServerIdleTimeout(server, Math.ceil(msecs / 1000)); - if (typeof callback === "function") this.once("timeout", callback); - } else { - (this[kDeferredTimeouts] ??= []).push({ msecs, callback }); - } + this.timeout = msecs; + if ($isCallable(callback)) this.on("timeout", callback); return this; }; @@ -953,6 +961,67 @@ enum HttpParserError { HTTP_PARSER_ERROR_INVALID_METHOD = 9, HTTP_PARSER_ERROR_INVALID_HEADER_TOKEN = 10, } +// Fired by the native side for every accepted connection (after the TLS +// handshake for https), before any bytes are parsed. This is what lets the +// headers/request/inactivity timeouts cover sockets that never send a full +// request, and it makes 'connection' fire on accept like Node.js. +function onServerSocketOpen(this: Server, ssl: boolean, handle: unknown) { + const self = this; + // The native side returns the existing wrapper for a handle it has seen + // before; a second Duplex would strand the first in kTrackedConnections. + const existingDuplex = (handle as any).duplex; + const socket = existingDuplex ?? new NodeHTTPServerSocket(self, handle, ssl); + // Anchor the headers/request timeout clock at accept. Node.js anchors at + // the first parsed byte, which we cannot observe; accept is never later. + socket[kRequestStart] = DateNow(); + socket[kHeadersCompleted] = false; + const serverTimeout = self.timeout; + if (serverTimeout) socket.setTimeout(serverTimeout); + if (!socket[kConnectionEmitted]) { + socket[kConnectionEmitted] = true; + self.emit("connection", socket); + } +} + +function clearIncomingRequestTimer(socket) { + socket[kRequestStart] = 0; +} + +const requestTimeoutResponse = Buffer.from("HTTP/1.1 408 Request Timeout\r\nConnection: close\r\n\r\n", "ascii"); +// Mirrors Node.js's default 'clientError' handling of ERR_HTTP_REQUEST_TIMEOUT: +// user listeners take over; otherwise answer 408 (only if nothing was +// written yet) and destroy the socket. +function onRequestTimeout(server: Server, socket) { + socket[kRequestStart] = 0; + const err = $ERR_HTTP_REQUEST_TIMEOUT("Request timeout"); + if (!server.emit("clientError", err, socket)) { + if (socket.writable && (socket.bytesWritten | 0) === 0) { + socket.write(requestTimeoutResponse); + } + socket.destroy(err); + } +} + +// Node.js's checkConnections sweep: every connectionsCheckingInterval ms, +// close connections whose request head (headersTimeout) or entire request +// (requestTimeout) has not arrived in time. +function checkConnections(this: Server) { + const headersTimeout = this.headersTimeout; + const requestTimeout = this.requestTimeout; + if (headersTimeout === 0 && requestTimeout === 0) return; + const now = DateNow(); + for (const socket of this[kTrackedConnections]) { + const start = socket[kRequestStart]; + if (!start) continue; + const elapsed = now - start; + if (headersTimeout > 0 && !socket[kHeadersCompleted] && elapsed > headersTimeout) { + onRequestTimeout(this, socket); + } else if (requestTimeout > 0 && elapsed > requestTimeout) { + onRequestTimeout(this, socket); + } + } +} + function onServerClientError(ssl: boolean, socket: unknown, errorCode: number, rawPacket: ArrayBuffer) { const self = this as Server; let err; @@ -989,7 +1058,8 @@ function onServerClientError(ssl: boolean, socket: unknown, errorCode: number, r // connections - the existing duplex already had its 'connection' event. const existingDuplex = (socket as any).duplex; const nodeSocket = existingDuplex ?? new NodeHTTPServerSocket(self, socket, ssl); - if (!existingDuplex) { + if (!nodeSocket[kConnectionEmitted]) { + nodeSocket[kConnectionEmitted] = true; self.emit("connection", nodeSocket); } self.emit("clientError", err, nodeSocket); @@ -1000,6 +1070,15 @@ function onServerClientError(ssl: boolean, socket: unknown, errorCode: number, r const kBytesWritten = Symbol("kBytesWritten"); const kEnableStreaming = Symbol("kEnableStreaming"); +const kTimeoutHandle = Symbol("kTimeoutHandle"); +const kRequestStart = Symbol("kRequestStart"); +const kHeadersCompleted = Symbol("kHeadersCompleted"); +const kConnectionEmitted = Symbol("kConnectionEmitted"); + +function socketOnTimeoutFire(socket) { + socket[kTimeoutHandle] = null; + socket._onTimeout(); +} function onServerSocketError(this: any, _err) { // Default 'error' listener so socket-level errors (e.g. res.destroy(err) // forwarding the error to the socket) do not crash the process as @@ -1021,6 +1100,10 @@ const NodeHTTPServerSocket = class Socket extends Duplex { server: Server; _httpMessage; _secureEstablished = false; + [kTimeoutHandle] = null; + [kRequestStart] = 0; + [kHeadersCompleted] = false; + [kConnectionEmitted] = false; #pendingCallback = null; constructor(server: Server, handle, encrypted) { super(); @@ -1098,6 +1181,11 @@ const NodeHTTPServerSocket = class Socket extends Duplex { } #onClose() { this[kHandle] = null; + const timeoutHandle = this[kTimeoutHandle]; + if (timeoutHandle !== null) { + clearTimeout(timeoutHandle); + this[kTimeoutHandle] = null; + } this.server?.[kTrackedConnections]?.delete(this); // Node.js's `socketOnClose` → `abortIncoming()` only destroys requests @@ -1152,12 +1240,20 @@ const NodeHTTPServerSocket = class Socket extends Duplex { // If there is a response, and it has pending data, // we suppress the timeout because a write is in progress. if (response && response.writableLength > 0) { + this._unrefTimer(); return; } this.emit("timeout"); } _unrefTimer() { - // for compatibility + const msecs = this.timeout; + const handle = this[kTimeoutHandle]; + if (handle !== null) clearTimeout(handle); + if (msecs > 0) { + this[kTimeoutHandle] = setTimeout(socketOnTimeoutFire, msecs, this).unref(); + } else { + this[kTimeoutHandle] = null; + } } address() { @@ -1293,7 +1389,10 @@ const NodeHTTPServerSocket = class Socket extends Duplex { return this; } - setTimeout(_timeout, _callback) { + setTimeout(msecs, callback) { + this.timeout = msecs | 0; + if ($isCallable(callback)) this.once("timeout", callback); + this._unrefTimer(); return this; } @@ -1653,9 +1752,22 @@ function stopServerResponsePerf(this: any) { } } -function endSocketOnFinishIfNeeded(socket, res) { +function resOnFinish(server, socket, res) { if (res[kMustCloseConnection]) { socket?.end(); + return; + } + if (!socket || socket.destroyed) return; + // Keep-alive idle between requests: the headers/request timeouts only + // apply while a request is in flight; the keep-alive inactivity timer + // takes over until the next request head arrives (Node.js's resOnFinish). + socket[kRequestStart] = 0; + socket[kHeadersCompleted] = false; + const keepAliveTimeout = server.keepAliveTimeout; + if (keepAliveTimeout) { + socket.setTimeout(keepAliveTimeout); + } else { + socket._unrefTimer(); } } diff --git a/src/jsc/bindings/ErrorCode.ts b/src/jsc/bindings/ErrorCode.ts index e0c9acf2b9e4..782b00ff68a7 100644 --- a/src/jsc/bindings/ErrorCode.ts +++ b/src/jsc/bindings/ErrorCode.ts @@ -340,5 +340,6 @@ const errors: ErrorCodeMapping = [ ["ERR_FS_CP_SYMLINK_TO_SUBDIRECTORY", Error], ["ERR_DIR_CONCURRENT_OPERATION", Error], ["ERR_INVALID_BUFFER_SIZE", RangeError], + ["ERR_HTTP_REQUEST_TIMEOUT", Error], ]; export default errors; diff --git a/src/jsc/bindings/NodeHTTP.cpp b/src/jsc/bindings/NodeHTTP.cpp index 9ad2c767dd11..c75df89bff0b 100644 --- a/src/jsc/bindings/NodeHTTP.cpp +++ b/src/jsc/bindings/NodeHTTP.cpp @@ -40,6 +40,7 @@ extern "C" bool NodeHTTPResponse__setTimeout(void*, EncodedJSValue, JSC::JSGloba extern "C" void Server__setIdleTimeout(EncodedJSValue, EncodedJSValue, JSC::JSGlobalObject*); extern "C" EncodedJSValue Server__setAppFlags(JSC::JSGlobalObject*, EncodedJSValue, bool require_host_header, bool use_strict_method_validation); extern "C" EncodedJSValue Server__setOnClientError(JSC::JSGlobalObject*, EncodedJSValue, EncodedJSValue); +extern "C" EncodedJSValue Server__setOnNodeSocketOpen(JSC::JSGlobalObject*, EncodedJSValue, EncodedJSValue); extern "C" EncodedJSValue Server__setMaxHTTPHeaderSize(JSC::JSGlobalObject*, EncodedJSValue, uint64_t); static EncodedJSValue assignHeadersFromFetchHeaders(FetchHeaders& impl, JSObject* prototype, JSObject* objectValue, JSC::InternalFieldTuple* tuple, JSC::JSGlobalObject* globalObject, JSC::VM& vm) @@ -988,13 +989,14 @@ JSC_DEFINE_HOST_FUNCTION(jsHTTPSetCustomOptions, (JSGlobalObject * globalObject, { auto& vm = JSC::getVM(globalObject); auto scope = DECLARE_THROW_SCOPE(vm); - ASSERT(callFrame->argumentCount() == 5); + ASSERT(callFrame->argumentCount() >= 5); // This is an internal binding. JSValue serverValue = callFrame->uncheckedArgument(0); JSValue requireHostHeader = callFrame->uncheckedArgument(1); JSValue useStrictMethodValidation = callFrame->uncheckedArgument(2); JSValue maxHeaderSize = callFrame->uncheckedArgument(3); JSValue callback = callFrame->uncheckedArgument(4); + JSValue socketOpenCallback = callFrame->argument(5); double maxHeaderSizeNumber = maxHeaderSize.toNumber(globalObject); RETURN_IF_EXCEPTION(scope, {}); @@ -1008,6 +1010,11 @@ JSC_DEFINE_HOST_FUNCTION(jsHTTPSetCustomOptions, (JSGlobalObject * globalObject, Server__setOnClientError(globalObject, JSValue::encode(serverValue), JSValue::encode(callback)); RETURN_IF_EXCEPTION(scope, {}); + if (socketOpenCallback.isCallable()) { + Server__setOnNodeSocketOpen(globalObject, JSValue::encode(serverValue), JSValue::encode(socketOpenCallback)); + RETURN_IF_EXCEPTION(scope, {}); + } + return JSValue::encode(jsUndefined()); } diff --git a/src/runtime/server/mod.rs b/src/runtime/server/mod.rs index f989e71f399b..b8a069cb96ff 100644 --- a/src/runtime/server/mod.rs +++ b/src/runtime/server/mod.rs @@ -289,6 +289,7 @@ pub struct NewServer { pub user_routes: Vec>, pub on_clienterror: jsc::StrongOptional, + pub on_node_socket_open: jsc::StrongOptional, pub inspector_server_id: jsc::DebuggerId, } @@ -1913,6 +1914,7 @@ impl NewServer { plugins: None, user_routes: Vec::new(), on_clienterror: jsc::StrongOptional::empty(), + on_node_socket_open: jsc::StrongOptional::empty(), inspector_server_id: jsc::DebuggerId::init(0), })); diff --git a/src/runtime/server/server_body.rs b/src/runtime/server/server_body.rs index c7f7a4825559..865107c1d6fc 100644 --- a/src/runtime/server/server_body.rs +++ b/src/runtime/server/server_body.rs @@ -3522,6 +3522,37 @@ where } } + pub fn on_node_socket_open_callback(&mut self, socket: &mut uws::Socket) { + let Some(callback) = self.on_node_socket_open.get() else { + return; + }; + let is_ssl = SSL; + let global = self.global(); + let node_socket = match jsc::from_js_host_call(&global, || { + Bun__createNodeHTTPServerSocketForClientError( + is_ssl, + std::ptr::from_mut(socket).cast::(), + &global, + ) + }) { + Ok(v) => v, + Err(_) => return, + }; + if node_socket.is_undefined_or_null() { + return; + } + // SAFETY: event_loop() returns a live raw pointer tied to the global. + let _scope = + unsafe { jsc::event_loop::EventLoop::enter_scope(global.bun_vm().event_loop()) }; + if let Err(err) = callback.call( + &global, + JSValue::UNDEFINED, + &[JSValue::from(is_ssl), node_socket], + ) { + global.report_active_exception_as_unhandled(err); + } + } + // `js_gc_route_list_set` / `ptr_to_js` live on the unbounded // `impl NewServer` in mod.rs; do not redefine them here. } @@ -3666,6 +3697,50 @@ pub(super) fn server_set_on_client_error_( Ok(JSValue::UNDEFINED) } +pub(super) fn server_set_on_node_socket_open_( + global: &JSGlobalObject, + server: JSValue, + callback: JSValue, +) -> JsResult { + if !server.is_object() || !callback.is_function() { + return Ok(JSValue::UNDEFINED); + } + + macro_rules! handle { + ($T:ty) => { + if let Some(this) = server.as_::<$T>() { + // SAFETY: as_ returned a non-null *mut to a live server. + let this = unsafe { &mut *this }; + if let Some(app) = this.app { + this.on_node_socket_open.deinit(); + this.on_node_socket_open = StrongOptional::create(callback, global); + extern "C" fn thunk( + user_data: *mut c_void, + _ssl: c_int, + socket: *mut uws_sys::us_socket_t, + ) { + // SAFETY: user_data is the `*mut Self` registered below; + // socket is a live uWS socket. + let this = unsafe { &mut *user_data.cast::<$T>() }; + // S008: `us_socket_t` is an `opaque_ffi!` ZST — safe deref. + this.on_node_socket_open_callback(bun_opaque::opaque_deref_mut(socket)); + } + // S008: `NewApp` is a ZST opaque — safe `*mut → &mut` deref. + bun_opaque::opaque_deref_mut(app) + .on_socket_open(thunk, core::ptr::from_mut::<$T>(this).cast::()); + } + return Ok(JSValue::UNDEFINED); + } + }; + } + handle!(HTTPServer); + handle!(HTTPSServer); + handle!(DebugHTTPServer); + handle!(DebugHTTPSServer); + debug_assert!(false); + Ok(JSValue::UNDEFINED) +} + pub(super) fn server_set_app_flags_( global: &JSGlobalObject, server: JSValue, @@ -3769,6 +3844,18 @@ extern "C" fn server_set_on_client_error_shim( ) } +#[unsafe(export_name = "Server__setOnNodeSocketOpen")] +extern "C" fn server_set_on_node_socket_open_shim( + global: &JSGlobalObject, + server: JSValue, + callback: JSValue, +) -> JSValue { + host_fn::to_js_host_fn_result( + global, + server_set_on_node_socket_open_(global, server, callback), + ) +} + #[unsafe(export_name = "Server__setMaxHTTPHeaderSize")] extern "C" fn server_set_max_http_header_size_shim( global: &JSGlobalObject, diff --git a/src/uws_sys/App.rs b/src/uws_sys/App.rs index c5d79e47d173..6e6ede93d24d 100644 --- a/src/uws_sys/App.rs +++ b/src/uws_sys/App.rs @@ -260,6 +260,14 @@ impl App { c::uws_app_set_on_clienterror(Self::SSL_FLAG, self.as_raw(), handler, user_data) } + pub fn on_socket_open( + &mut self, + handler: extern "C" fn(*mut c_void, c_int, *mut us_socket_t), + user_data: *mut c_void, + ) { + c::uws_app_set_on_socket_open(Self::SSL_FLAG, self.as_raw(), handler, user_data) + } + pub fn listen_with_config( &mut self, handler: c::uws_listen_handler, @@ -505,6 +513,12 @@ pub mod c { handler: extern "C" fn(*mut c_void, c_int, *mut us_socket_t, u8, *mut u8, c_int), user_data: *mut c_void, ); + pub(crate) safe fn uws_app_set_on_socket_open( + ssl: c_int, + app: &mut uws_app_s, + handler: extern "C" fn(*mut c_void, c_int, *mut us_socket_t), + user_data: *mut c_void, + ); pub(crate) fn uws_create_app(ssl: i32, options: BunSocketContextOptions) -> *mut uws_app_t; pub(crate) fn uws_app_destroy(ssl: i32, app: *mut uws_app_t); pub(crate) safe fn uws_app_set_flags( diff --git a/src/uws_sys/libuwsockets.cpp b/src/uws_sys/libuwsockets.cpp index c481e6513581..de70e504008c 100644 --- a/src/uws_sys/libuwsockets.cpp +++ b/src/uws_sys/libuwsockets.cpp @@ -401,6 +401,18 @@ extern "C" } } + void uws_app_set_on_socket_open(int ssl, uws_app_t *app, void (*handler)(void *user_data, int is_ssl, struct us_socket_t *rawSocket), void *user_data) + { + if (ssl) + { + ((uWS::SSLApp *)app)->setOnSocketOpen(handler, user_data); + } + else + { + ((uWS::App *)app)->setOnSocketOpen(handler, user_data); + } + } + void uws_app_set_on_clienterror(int ssl, uws_app_t *app, void (*handler)(void *user_data, int is_ssl, struct us_socket_t *rawSocket, uint8_t errorCode, char *rawPacket, int rawPacketLength), void *user_data) { if (ssl) diff --git a/test/js/node/http/node-http-server-timeouts.test.ts b/test/js/node/http/node-http-server-timeouts.test.ts new file mode 100644 index 000000000000..98700bbdbec0 --- /dev/null +++ b/test/js/node/http/node-http-server-timeouts.test.ts @@ -0,0 +1,220 @@ +import { describe, expect, test } from "bun:test"; +import { once } from "node:events"; +import http from "node:http"; +import net from "node:net"; + +// Each test opens a raw TCP socket against a server whose timeout knob is a +// few hundred ms and waits for the server to close the connection. A small +// connectionsCheckingInterval makes the headers/request sweep run well inside +// the probe window. Every probe awaits the 'close' event rather than a fixed +// delay, so on a build that does not enforce the knob the test times out. + +async function listen(server: http.Server) { + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + return (server.address() as net.AddressInfo).port; +} + +describe("node:http server timeout enforcement", () => { + test("headersTimeout closes a connection that never completes its request headers", async () => { + const server = http.createServer({ connectionsCheckingInterval: 50 }, (req, res) => { + req.resume(); + req.on("end", () => res.end("ok")); + }); + server.headersTimeout = 200; + server.requestTimeout = 800; + let clientErrorCode: string | undefined; + server.on("clientError", (err: any, socket) => { + clientErrorCode = err.code; + socket.destroy(); + }); + const port = await listen(server); + try { + const { promise: closed, resolve: onClosed } = Promise.withResolvers(); + const socket = net.connect(port, "127.0.0.1"); + const t0 = Date.now(); + socket.setNoDelay(true); + socket.on("error", () => {}); + socket.resume(); + socket.on("connect", () => { + // A valid but incomplete request head: no terminating CRLF. + socket.write("GET / HTTP/1.1\r\nHost: a\r\n"); + }); + socket.on("close", () => onClosed(Date.now() - t0)); + const elapsed = await closed; + expect({ clientErrorCode, closedPromptly: elapsed < 3000 }).toEqual({ + clientErrorCode: "ERR_HTTP_REQUEST_TIMEOUT", + closedPromptly: true, + }); + } finally { + server.closeAllConnections(); + server.close(); + } + }); + + test("requestTimeout closes a connection that stalls mid-body", async () => { + const server = http.createServer({ connectionsCheckingInterval: 50 }, (req, res) => { + req.resume(); + req.on("end", () => res.end("ok")); + }); + server.headersTimeout = 150; + server.requestTimeout = 300; + let clientErrorCode: string | undefined; + server.on("clientError", (err: any, socket) => { + clientErrorCode = err.code; + socket.destroy(); + }); + const port = await listen(server); + try { + const { promise: closed, resolve: onClosed } = Promise.withResolvers(); + const socket = net.connect(port, "127.0.0.1"); + const t0 = Date.now(); + socket.setNoDelay(true); + socket.on("error", () => {}); + socket.resume(); + socket.on("connect", () => { + // Complete headers, then only 2 of the promised 50 body bytes. + socket.write("POST / HTTP/1.1\r\nHost: a\r\nContent-Length: 50\r\n\r\nab"); + }); + socket.on("close", () => onClosed(Date.now() - t0)); + const elapsed = await closed; + expect({ clientErrorCode, closedPromptly: elapsed < 3000 }).toEqual({ + clientErrorCode: "ERR_HTTP_REQUEST_TIMEOUT", + closedPromptly: true, + }); + } finally { + server.closeAllConnections(); + server.close(); + } + }); + + test("server.setTimeout() fires the 'timeout' event for an inactive connection", async () => { + const server = http.createServer((req, res) => { + req.resume(); + req.on("end", () => res.end("ok")); + }); + let timeoutFired = false; + server.setTimeout(200, socket => { + timeoutFired = true; + socket.destroy(); + }); + expect(server.timeout).toBe(200); + const port = await listen(server); + try { + const { promise: closed, resolve: onClosed } = Promise.withResolvers(); + const socket = net.connect(port, "127.0.0.1"); + const t0 = Date.now(); + socket.setNoDelay(true); + socket.on("error", () => {}); + socket.resume(); + // A valid-but-incomplete request head, then silence. headersTimeout + // and requestTimeout keep their large defaults, so only the + // server.setTimeout inactivity timer can close this connection. + socket.on("connect", () => socket.write("GET / HTTP/1.1\r\nHost: a\r\n")); + socket.on("close", () => onClosed(Date.now() - t0)); + const elapsed = await closed; + expect({ timeoutFired, closedPromptly: elapsed < 3000 }).toEqual({ + timeoutFired: true, + closedPromptly: true, + }); + } finally { + server.closeAllConnections(); + server.close(); + } + }); + + test("keepAliveTimeout closes an idle keep-alive connection after the response", async () => { + const server = http.createServer((req, res) => { + req.resume(); + res.end("ok"); + }); + server.keepAliveTimeout = 200; + const port = await listen(server); + try { + const { promise: closed, resolve: onClosed } = Promise.withResolvers(); + const { promise: gotResponse, resolve: onResponse } = Promise.withResolvers(); + const socket = net.connect(port, "127.0.0.1"); + let t0 = 0; + socket.setNoDelay(true); + socket.on("error", () => {}); + socket.resume(); + socket.on("connect", () => socket.write("GET / HTTP/1.1\r\nHost: a\r\n\r\n")); + socket.once("data", () => { + t0 = Date.now(); + onResponse(); + }); + socket.on("close", () => onClosed(t0 ? Date.now() - t0 : -1)); + await gotResponse; + const elapsed = await closed; + expect(elapsed).toBeGreaterThanOrEqual(0); + expect(elapsed).toBeLessThan(3000); + } finally { + server.closeAllConnections(); + server.close(); + } + }); + + test("headersTimeout answers 408 when there is no 'clientError' listener", async () => { + const server = http.createServer({ connectionsCheckingInterval: 50 }, (req, res) => res.end("ok")); + server.headersTimeout = 200; + server.requestTimeout = 800; + const port = await listen(server); + try { + const { promise: done, resolve } = Promise.withResolvers(); + const socket = net.connect(port, "127.0.0.1"); + socket.setNoDelay(true); + let received = ""; + socket.on("data", chunk => { + received += chunk.toString("latin1"); + }); + socket.on("error", () => {}); + socket.on("close", () => resolve(received)); + socket.on("connect", () => socket.write("GET / HTTP/1.1\r\nHost: a\r\n")); + const response = await done; + expect(response).toContain("408 Request Timeout"); + expect(response).toContain("Connection: close"); + } finally { + server.closeAllConnections(); + server.close(); + } + }); + + test("requestTimeout does not fire while a slow handler streams a response", async () => { + // The request (a body-less GET) is complete as soon as its head is + // parsed, so requestTimeout must stop ticking even though the handler + // holds the response open well past it. + const server = http.createServer({ connectionsCheckingInterval: 25 }, (req, res) => { + res.writeHead(200, { "Content-Type": "text/plain" }); + res.write("started\n"); + setTimeout(() => res.end("done\n"), 400); + }); + server.headersTimeout = 100; + server.requestTimeout = 100; + let sawClientError = false; + server.on("clientError", (_err, socket) => { + sawClientError = true; + socket.destroy(); + }); + const port = await listen(server); + try { + const { promise: done, resolve } = Promise.withResolvers(); + const socket = net.connect(port, "127.0.0.1"); + socket.setNoDelay(true); + let received = ""; + socket.on("data", chunk => { + received += chunk.toString("latin1"); + }); + socket.on("error", () => {}); + socket.on("connect", () => socket.write("GET / HTTP/1.1\r\nHost: a\r\nConnection: close\r\n\r\n")); + socket.on("close", () => resolve(received)); + const response = await done; + expect({ sawClientError, ok: response.includes("started") && response.includes("done") }).toEqual({ + sawClientError: false, + ok: true, + }); + } finally { + server.closeAllConnections(); + server.close(); + } + }); +});