node:http: let server 'timeout' listeners veto the default socket destroy

src/js/internal/http.ts

@@ -5,19 +5,16 @@ const {
   setHeader,
   Headers,
   assignHeaders: assignHeadersFast,
-  setRequestTimeout,
   headersTuple,
   webRequestOrResponseHasBodyValue,
   setServerCustomOptions,
   getCompleteWebRequestOrResponseBodyValueAsArrayBuffer,
   drainMicrotasks,
-  setServerIdleTimeout,
 } = $cpp("NodeHTTP.cpp", "createNodeHTTPInternalBinding") as {
   getHeader: (headers: Headers, name: string) => string | undefined;
   setHeader: (headers: Headers, name: string, value: string) => void;
   Headers: (typeof globalThis)["Headers"];
   assignHeaders: (object: any, req: Request, headersTuple: any) => boolean;
-  setRequestTimeout: (req: Request, timeout: number) => boolean;
   headersTuple: any;
   webRequestOrResponseHasBodyValue: (arg: any) => boolean;
   setServerCustomOptions: (
@@ -29,7 +26,6 @@ const {
   ) => void;
   getCompleteWebRequestOrResponseBodyValueAsArrayBuffer: (arg: any) => ArrayBuffer | undefined;
   drainMicrotasks: () => void;
-  setServerIdleTimeout: (server: any, timeout: number) => void;
 };
 
 const getRawKeys = $newCppFunction("JSFetchHeaders.cpp", "jsFetchHeaders_getRawKeys", 0);
@@ -88,7 +84,6 @@ const serverSymbol = Symbol.for("::bunternal::");
 const kPendingCallbacks = Symbol("pendingCallbacks");
 const kRequest = Symbol("request");
 const kCloseCallback = Symbol("closeCallback");
-const kDeferredTimeouts = Symbol("deferredTimeouts");
 
 const kEmptyObject = Object.freeze(Object.create(null));
 
@@ -543,7 +538,6 @@ export {
   kBodyChunks,
   kClearTimeout,
   kCloseCallback,
-  kDeferredTimeouts,
   kDeprecatedReplySymbol,
   kEmitState,
   kEmptyObject,
@@ -584,9 +578,7 @@ export {
   setHeader,
   setIsNextIncomingMessageHTTPS,
   setMaxHTTPHeaderSize,
-  setRequestTimeout,
   setServerCustomOptions,
-  setServerIdleTimeout,
   statusCodeSymbol,
   statusMessageSymbol,
   timeoutTimerSymbol,

src/js/node/_http_incoming.ts

@@ -19,7 +19,6 @@ const {
   NodeHTTPBodyReadState,
   emitEOFIncomingMessage,
   NodeHTTPResponseAbortEvent,
-  setRequestTimeout,
   kAbortController,
 } = require("internal/http");
 
@@ -55,6 +54,7 @@ function onIncomingMessageResumeNodeHTTPResponse(this: IncomingMessage) {
   if (handle && !this.destroyed) {
     const resumed = handle.resume();
     if (resumed && resumed !== true) {
+      this.socket?._unrefTimer?.();
       const bodyReadState = handle.hasBody;
       if ((bodyReadState & NodeHTTPBodyReadState.done) !== 0) {
         emitEOFIncomingMessage(this);
@@ -302,13 +302,7 @@ ObjectDefineProperty(IncomingMessage.prototype, "signal", {
 
 IncomingMessage.prototype.setTimeout = function setTimeout(msecs, callback) {
   if (callback) this.on("timeout", callback);
-
-  const handle = this[kHandle];
-  if (handle) {
-    setRequestTimeout(handle, Math.ceil(msecs / 1000));
-  } else {
-    this.socket?.setTimeout(msecs);
-  }
+  this.socket?.setTimeout(msecs);
   return this;
 };
 
@@ -357,6 +351,7 @@ IncomingMessage.prototype._read = function _read(_n) {
   if ((bodyReadState & NodeHTTPBodyReadState.hasBufferedDataDuringPause) !== 0) {
     const drained = handle.drainRequestBody();
     if (drained && !this._dumped) {
+      socket?._unrefTimer?.();
       this.push(drained);
     }
   }
@@ -378,6 +373,8 @@ function onDataIncomingMessage(
     return;
   }
 
+  this.socket?._unrefTimer?.();
+
   if (chunk && !this._dumped) this.push(chunk);
 
   if (isLast) {

src/js/node/_http_server.ts

@@ -15,6 +15,7 @@ const {
   validateInteger,
   validateFunction,
 } = require("internal/validators");
+const { getTimerDuration } = require("internal/timers");
 const { ConnResetException, hasObserver, startPerf, stopPerf } = require("internal/shared");
 const kServerResponseStatistics = Symbol("ServerResponseStatistics");
 
@@ -27,7 +28,6 @@ const {
   kRealListen,
   tlsSymbol,
   optionsSymbol,
-  kDeferredTimeouts,
   kDeprecatedReplySymbol,
   headerStateSymbol,
   NodeHTTPHeaderState,
@@ -51,7 +51,6 @@ const {
   eofInProgress,
   runSymbol,
   drainMicrotasks,
-  setServerIdleTimeout,
   setServerCustomOptions,
   getMaxHTTPHeaderSize,
   fakeSocketSymbol,
@@ -254,6 +253,7 @@ function Server(options, callback): void {
 
   this.listening = false;
   this._unref = false;
+  this.timeout = 0;
   this.maxRequestsPerSocket = 0;
   this.maxHeadersCount = null;
   this[kInternalSocketData] = undefined;
@@ -609,6 +609,18 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort
           socket = new NodeHTTPServerSocket(server, socketHandle, !!tls);
         }
 
+        // Like Node.js's connectionListenerInternal: the server timeout is
+        // applied once per connection (before any request routing, so it also
+        // covers CONNECT and upgrade tunnels) and before 'connection' fires so
+        // a user listener can override it; subsequent keep-alive requests only
+        // refresh the existing timer.
+        if (isSocketNew) {
+          const serverTimeoutMs = server.timeout;
+          if (serverTimeoutMs) socket.setTimeout(serverTimeoutMs);
+        } else {
+          socket._unrefTimer();
+        }
+
         const http_req = new RequestClass(kHandle, url, method, headersObject, headersArray, handle, hasBody, socket);
         if (isAncientHTTP) {
           http_req.httpVersion = "1.0";
@@ -920,25 +932,16 @@ 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);
   }
 };
 
+// Like Node.js: the server records the value and applies it to each new
+// connection via socket.setTimeout(); the socket emits 'timeout' and a
+// listener on any of req/res/server vetoes the default destroy.
 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 (typeof callback === "function") this.on("timeout", callback);
   return this;
 };
 
@@ -1042,6 +1045,9 @@ const NodeHTTPServerSocket = class Socket extends Duplex {
   _httpMessage;
   _secureEstablished = false;
   #pendingCallback = null;
+  #timeoutTimer = null;
+  #boundOnTimeout = null;
+  #lastBufferedAmount = 0;
   constructor(server: Server, handle, encrypted) {
     super();
     this.server = server;
@@ -1086,6 +1092,7 @@ const NodeHTTPServerSocket = class Socket extends Duplex {
     }
   }
   #onDrain() {
+    this._unrefTimer();
     const handle = this[kHandle];
     this[kBytesWritten] = handle ? (handle.response?.getBytesWritten?.() ?? handle.bytesWritten ?? 0) : 0;
     const callback = this.#pendingCallback;
@@ -1096,6 +1103,7 @@ const NodeHTTPServerSocket = class Socket extends Duplex {
     this.emit("drain");
   }
   #onData(chunk, last) {
+    this._unrefTimer();
     if (chunk) {
       this.push(chunk);
     }
@@ -1122,6 +1130,11 @@ const NodeHTTPServerSocket = class Socket extends Duplex {
   }
   #onClose() {
     this[kHandle] = null;
+    const timer = this.#timeoutTimer;
+    if (timer) {
+      clearTimeout(timer);
+      this.#timeoutTimer = null;
+    }
     this.server?.[kTrackedConnections]?.delete(this);
 
     // Node.js's `socketOnClose` → `abortIncoming()` only destroys requests
@@ -1180,15 +1193,20 @@ const NodeHTTPServerSocket = class Socket extends Duplex {
   _onTimeout() {
     const handle = this[kHandle];
     const response = handle?.response;
-    // 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) {
+    // Like Node.js's net.Socket._onTimeout: only suppress when the buffered
+    // amount has changed since the last check (progress), so a stalled write
+    // still emits 'timeout' instead of looping forever.
+    const buffered = response ? response.bufferedAmount : 0;
+    if (buffered > 0 && buffered !== this.#lastBufferedAmount) {
+      this.#lastBufferedAmount = buffered;
+      this._unrefTimer();
       return;
     }
+    this.#lastBufferedAmount = buffered;
     this.emit("timeout");
   }
   _unrefTimer() {
-    // for compatibility
+    this.#timeoutTimer?.refresh();
   }
 
   address() {
@@ -1324,7 +1342,26 @@ const NodeHTTPServerSocket = class Socket extends Duplex {
     return this;
   }
 
-  setTimeout(_timeout, _callback) {
+  setTimeout(msecs, callback) {
+    if (this.destroyed) return this;
+    this.timeout = msecs;
+    msecs = getTimerDuration(msecs, "msecs");
+    const existing = this.#timeoutTimer;
+    if (existing) clearTimeout(existing);
+    if (msecs === 0) {
+      this.#timeoutTimer = null;
+      if (callback !== undefined) {
+        validateFunction(callback, "callback");
+        this.removeListener("timeout", callback);
+      }
+    } else {
+      const bound = (this.#boundOnTimeout ??= this._onTimeout.bind(this));
+      this.#timeoutTimer = setTimeout(bound, msecs).unref();
+      if (callback !== undefined) {
+        validateFunction(callback, "callback");
+        this.once("timeout", callback);
+      }
+    }
     return this;
   }
 
@@ -1339,6 +1376,7 @@ const NodeHTTPServerSocket = class Socket extends Duplex {
   }
 
   _write(_chunk, _encoding, _callback) {
+    this._unrefTimer();
     const handle = this[kHandle];
     let err;
     try {
@@ -1687,6 +1725,8 @@ function stopServerResponsePerf(this: any) {
 function endSocketOnFinishIfNeeded(socket, res) {
   if (res[kMustCloseConnection]) {
     socket?.end();
+  } else {
+    socket?._unrefTimer();
   }
 }
 
@@ -2189,6 +2229,8 @@ ServerResponse.prototype.write = function (chunk, encoding, callback) {
     result = handle.write(chunk, encoding, allowWritesToContinue.bind(this), strictContentLength(this));
   }
 
+  this[fakeSocketSymbol]?._unrefTimer();
+
   if (result < 0) {
     if (callback) {
       // The write was buffered due to backpressure.
@@ -2352,6 +2394,7 @@ ServerResponse.prototype._send = function (data, encoding, callback, _byteLength
   } else {
     handle.write(data, encoding, callback, strictContentLength(this));
   }
+  this[fakeSocketSymbol]?._unrefTimer();
 };
 
 const kSnapshotStatusCode = Symbol("kSnapshotStatusCode");
@@ -2501,6 +2544,7 @@ function callWriteHeadIfObservable(self, headerState) {
 }
 
 function allowWritesToContinue() {
+  this[fakeSocketSymbol]?._unrefTimer();
   this._callPendingCallbacks();
   this.emit("drain");
 }

src/jsc/bindings/NodeHTTP.cpp

@@ -35,9 +35,6 @@ BUN_DECLARE_HOST_FUNCTION(jsFunctionRequestOrResponseHasBodyValue);
 BUN_DECLARE_HOST_FUNCTION(jsFunctionGetCompleteRequestOrResponseBodyValueAsArrayBuffer);
 extern "C" uWS::HttpRequest* Request__getUWSRequest(void*);
 extern "C" void Request__setInternalEventCallback(void*, EncodedJSValue, JSC::JSGlobalObject*);
-extern "C" void Request__setTimeout(void*, EncodedJSValue, JSC::JSGlobalObject*);
-extern "C" bool NodeHTTPResponse__setTimeout(void*, EncodedJSValue, JSC::JSGlobalObject*);
-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__setMaxHTTPHeaderSize(JSC::JSGlobalObject*, EncodedJSValue, uint64_t);
@@ -972,43 +969,6 @@ JSC_DEFINE_HOST_FUNCTION(jsHTTPAssignEventCallback, (JSGlobalObject * globalObje
     return JSValue::encode(jsNull());
 }
 
-JSC_DEFINE_HOST_FUNCTION(jsHTTPSetTimeout, (JSGlobalObject * globalObject, CallFrame* callFrame))
-{
-    auto& vm = JSC::getVM(globalObject);
-    auto scope = DECLARE_THROW_SCOPE(vm);
-
-    // This is an internal binding.
-    JSValue requestValue = callFrame->uncheckedArgument(0);
-    JSValue seconds = callFrame->uncheckedArgument(1);
-
-    ASSERT(callFrame->argumentCount() == 2);
-
-    if (auto* jsRequest = dynamicDowncast<WebCore::JSRequest>(requestValue)) {
-        Request__setTimeout(jsRequest->wrapped(), JSValue::encode(seconds), globalObject);
-    }
-
-    if (auto* nodeHttpResponse = dynamicDowncast<WebCore::JSNodeHTTPResponse>(requestValue)) {
-        NodeHTTPResponse__setTimeout(nodeHttpResponse->wrapped(), JSValue::encode(seconds), globalObject);
-    }
-
-    return JSValue::encode(jsUndefined());
-}
-JSC_DEFINE_HOST_FUNCTION(jsHTTPSetServerIdleTimeout, (JSGlobalObject * globalObject, CallFrame* callFrame))
-{
-    auto& vm = JSC::getVM(globalObject);
-    auto scope = DECLARE_THROW_SCOPE(vm);
-
-    // This is an internal binding.
-    JSValue serverValue = callFrame->uncheckedArgument(0);
-    JSValue seconds = callFrame->uncheckedArgument(1);
-
-    ASSERT(callFrame->argumentCount() == 2);
-
-    Server__setIdleTimeout(JSValue::encode(serverValue), JSValue::encode(seconds), globalObject);
-
-    return JSValue::encode(jsUndefined());
-}
-
 JSC_DEFINE_HOST_FUNCTION(jsHTTPSetCustomOptions, (JSGlobalObject * globalObject, CallFrame* callFrame))
 {
     auto& vm = JSC::getVM(globalObject);
@@ -1180,14 +1140,6 @@ JSValue createNodeHTTPInternalBinding(Zig::GlobalObject* globalObject)
         vm, JSC::PropertyName(JSC::Identifier::fromString(vm, "assignEventCallback"_s)),
         JSC::JSFunction::create(vm, globalObject, 2, "assignEventCallback"_s, jsHTTPAssignEventCallback, ImplementationVisibility::Public), 0);
 
-    obj->putDirect(
-        vm, JSC::PropertyName(JSC::Identifier::fromString(vm, "setRequestTimeout"_s)),
-        JSC::JSFunction::create(vm, globalObject, 2, "setRequestTimeout"_s, jsHTTPSetTimeout, ImplementationVisibility::Public), 0);
-
-    obj->putDirect(
-        vm, JSC::PropertyName(JSC::Identifier::fromString(vm, "setServerIdleTimeout"_s)),
-        JSC::JSFunction::create(vm, globalObject, 2, "setServerIdleTimeout"_s, jsHTTPSetServerIdleTimeout, ImplementationVisibility::Public), 0);
-
     obj->putDirect(
         vm, JSC::PropertyName(JSC::Identifier::fromString(vm, "setServerCustomOptions"_s)),
         JSC::JSFunction::create(vm, globalObject, 2, "setServerCustomOptions"_s, jsHTTPSetCustomOptions, ImplementationVisibility::Public), 0);