add _onTimeout

src/bun.js/api/bun/socket.zig

@@ -2860,6 +2860,13 @@ fn NewSocket(comptime ssl: bool) type {
             return JSC.JSValue.jsNumber(this.bytes_written + this.buffered_data_for_node_net.len);
         }
 
+        pub fn getBufferedAmount(
+            this: *This,
+            _: *JSC.JSGlobalObject,
+        ) JSValue {
+            return JSC.JSValue.jsNumber(this.buffered_data_for_node_net.len);
+        }
+
         pub fn getALPNProtocol(
             this: *This,
             globalObject: *JSC.JSGlobalObject,

src/bun.js/api/server/NodeHTTPResponse.zig

@@ -369,7 +369,7 @@ pub fn getHasBody(this: *const NodeHTTPResponse, _: *JSC.JSGlobalObject) JSC.JSV
 
 pub fn getBufferedAmount(this: *const NodeHTTPResponse, _: *JSC.JSGlobalObject) JSC.JSValue {
     if (this.flags.request_has_completed or this.flags.socket_closed) {
-        return JSC.JSValue.jsNull();
+        return JSC.JSValue.jsNumber(0);
     }
 
     return JSC.JSValue.jsNumber(this.raw_response.getBufferedAmount());

src/bun.js/api/sockets.classes.ts

@@ -91,6 +91,9 @@ function generate(ssl) {
       bytesWritten: {
         getter: "getBytesWritten",
       },
+      bufferedAmount: {
+        getter: "getBufferedAmount",
+      },
       setNoDelay: {
         fn: "setNoDelay",
         length: 1,

src/js/node/http.ts

@@ -231,6 +231,10 @@ var FakeSocket = class Socket extends Duplex {
     return this;
   }
 
+  _onTimeout = function () {
+    this.emit("timeout");
+  };
+
   _destroy(err, callback) {
     const socketData = this[kInternalSocketData];
     if (!socketData) return; // sometimes 'this' is Socket not FakeSocket
@@ -375,6 +379,20 @@ const NodeHTTPServerSocket = class Socket extends Duplex {
     $isCallable(closeCallback) && closeCallback();
   }
 
+  _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.getBufferedAmount() > 0) {
+      return;
+    }
+    this.emit("timeout");
+  }
+  _unrefTimer() {
+    // for compatibility
+  }
+
   address() {
     return this[kHandle]?.remoteAddress || null;
   }

src/js/node/net.ts

@@ -587,6 +587,22 @@ Socket.prototype.address = function address() {
   };
 };
 
+Socket.prototype._onTimeout = function () {
+  // if there is pending data, write is in progress
+  // so we suppress the timeout
+  if (this._pendingData) {
+    return;
+  }
+
+  const handle = this._handle;
+  // if there is a handle, and it has pending data,
+  // we suppress the timeout because a write is in progress
+  if (handle && handle.bufferedAmount > 0) {
+    return;
+  }
+  this.emit("timeout");
+};
+
 Object.defineProperty(Socket.prototype, "bufferSize", {
   get: function () {
     return this.writableLength;

test/js/node/test/sequential/test-http-keep-alive-large-write.js

@@ -0,0 +1,47 @@
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+const http = require('http');
+
+// This test assesses whether long-running writes can complete
+// or timeout because the socket is not aware that the backing
+// stream is still writing.
+
+const writeSize = 3000000;
+let socket;
+
+const server = http.createServer(common.mustCall((req, res) => {
+  server.close();
+  const content = Buffer.alloc(writeSize, 0x44);
+
+  res.writeHead(200, {
+    'Content-Type': 'application/octet-stream',
+    'Content-Length': content.length.toString(),
+    'Vary': 'Accept-Encoding'
+  });
+
+  socket = res.socket;
+  const onTimeout = socket._onTimeout;
+  socket._onTimeout = common.mustCallAtLeast(() => onTimeout.call(socket), 1);
+  res.write(content);
+  res.end();
+}));
+server.on('timeout', () => {
+  // TODO(apapirovski): This test is faulty on certain Windows systems
+  // as no queue is ever created
+  assert(!socket._handle || socket._handle.writeQueueSize === 0,
+         'Should not timeout');
+});
+
+server.listen(0, common.mustCall(() => {
+  http.get({
+    path: '/',
+    port: server.address().port
+  }, (res) => {
+    res.once('data', () => {
+      socket._onTimeout();
+      res.on('data', () => {});
+    });
+    res.on('end', () => server.close());
+  });
+}));