From 94a47d6f15b2cf3e43871ba798e46d0e730bbbba Mon Sep 17 00:00:00 2001 From: Naman Trivedi Date: Mon, 3 Aug 2026 22:29:08 +0000 Subject: [PATCH] http: emit drain on socket takeover and avoid stale HWM reuse MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When OutgoingMessage transitions from pre-socket buffering (Path B) to socket-connected writing (Path A), the backpressure domain changes — subsequent writes go directly to the socket, which enforces its own backpressure via socket.write() return values. The OM should emit drain at this transition point to signal that its buffer is clear and the caller can resume writing under the socket backpressure regime. Previously, _flush() gated drain emission on writableLength === 0 which included socket.writableLength. This conflated two independent backpressure domains: the OM pre-socket buffer and the socket kernel write queue. When the socket had a higher writableHighWaterMark than the OM (e.g. agent-reused socket from a prior request), the socket was never backpressured and never emitted drain, causing a permanent deadlock. Additionally, avoid reusing a pooled socket in http.Agent when its writableHighWaterMark differs from the request highWaterMark, so that the user backpressure threshold is respected for the common case of the built-in Agent. Signed-off-by: Naman Trivedi Fixes: https://github.com/nodejs/node/issues/64680 Refs: https://github.com/nodejs/node/pull/64653 Refs: https://github.com/nodejs/node/pull/62936 --- lib/_http_agent.js | 10 ++++ lib/_http_outgoing.js | 6 +- .../test-http-agent-highwatermark-reuse.js | 56 ++++++++++++++++++ .../test-http-outgoing-flush-drain.js | 57 +++++++++++++++++++ 4 files changed, 127 insertions(+), 2 deletions(-) create mode 100644 test/parallel/test-http-agent-highwatermark-reuse.js create mode 100644 test/parallel/test-http-outgoing-flush-drain.js diff --git a/lib/_http_agent.js b/lib/_http_agent.js index edf988a046ae..f4da2ed246cd 100644 --- a/lib/_http_agent.js +++ b/lib/_http_agent.js @@ -394,6 +394,16 @@ Agent.prototype.addRequest = function addRequest(req, options, port/* legacy */, const sockLen = freeLen + this.sockets[name].length; // Reusing a socket from the pool. + // If the caller specified a highWaterMark that differs from the pooled + // socket's writableHighWaterMark, sync the socket's HWM so that + // backpressure semantics match what the caller requested. + if (socket && options.highWaterMark != null && + socket.writableHighWaterMark !== options.highWaterMark) { + debug('sync reused socket HWM (socket=%d, request=%d)', + socket.writableHighWaterMark, options.highWaterMark); + socket._writableState.highWaterMark = options.highWaterMark; + } + if (socket) { asyncResetHandle(socket); this.reuseSocket(socket, req); diff --git a/lib/_http_outgoing.js b/lib/_http_outgoing.js index 6bf7a1f9f68d..fcae7dbe2f21 100644 --- a/lib/_http_outgoing.js +++ b/lib/_http_outgoing.js @@ -1203,12 +1203,14 @@ OutgoingMessage.prototype._flush = function _flush() { if (socket?.writable) { // There might be remaining data in this.output; write it out - this._flushOutput(socket); + const ret = this._flushOutput(socket); if (this.finished) { // This is a queue to the server or client to bring in the next this. this._finish(); - } else if (this[kNeedDrain] && this.writableLength === 0) { + } else if (this[kNeedDrain] && ret !== false) { + // Socket accepted all data without backpressure - it won't emit + // drain, so we emit it since the OM buffer is now clear. this[kNeedDrain] = false; this.emit('drain'); } diff --git a/test/parallel/test-http-agent-highwatermark-reuse.js b/test/parallel/test-http-agent-highwatermark-reuse.js new file mode 100644 index 000000000000..b78b475ef6f1 --- /dev/null +++ b/test/parallel/test-http-agent-highwatermark-reuse.js @@ -0,0 +1,56 @@ +'use strict'; + +// Regression test: when a pooled socket's writableHighWaterMark differs from +// the new request's highWaterMark, the agent must sync the socket's HWM so +// that backpressure semantics match what the caller requested. +// +// See: https://github.com/nodejs/node/issues/64680 + +const common = require('../common'); +const assert = require('assert'); +const http = require('http'); + +const server = http.createServer(common.mustCall((req, res) => { + req.resume(); + req.on('end', () => res.end('ok')); +}, 2)); + +server.listen(0, common.mustCall(() => { + const port = server.address().port; + const agent = new http.Agent({ keepAlive: true }); + + // Request A: creates socket with HWM=1MB. + http.request({ + host: 'localhost', port, method: 'POST', agent, + highWaterMark: 1024 * 1024, + }, common.mustCall((res) => { + res.resume(); + res.on('end', common.mustCall(() => { + // Wait for socket to return to pool. + setTimeout(common.mustCall(requestB), 100); + })); + })).end('x'); + + function requestB() { + const freeCount = Object.values(agent.freeSockets).flat().length; + assert.strictEqual(freeCount, 1); + + // Request B: HWM=10KB — agent must sync the reused socket's HWM. + const reqB = http.request({ + host: 'localhost', port, method: 'POST', agent, + highWaterMark: 10 * 1024, + }, common.mustCall((res) => { + res.resume(); + res.on('end', common.mustCall(() => { + server.close(); + })); + })); + + reqB.on('socket', common.mustCall((socket) => { + // Socket HWM must be synced to the request's value. + assert.strictEqual(socket.writableHighWaterMark, 10 * 1024); + })); + + reqB.end('y'); + } +})); diff --git a/test/parallel/test-http-outgoing-flush-drain.js b/test/parallel/test-http-outgoing-flush-drain.js new file mode 100644 index 000000000000..12b5a5036cce --- /dev/null +++ b/test/parallel/test-http-outgoing-flush-drain.js @@ -0,0 +1,57 @@ +'use strict'; + +// Regression test: when _flush() hands buffered data to a socket whose +// writableHighWaterMark is higher than the OutgoingMessage's kHighWaterMark, +// drain must still fire. Previously, _flush() gated drain emission on +// writableLength === 0, which included socket.writableLength — but the +// socket was never backpressured (data < socket HWM), so drain never fired. +// +// See: https://github.com/nodejs/node/issues/64680 + +const common = require('../common'); +const assert = require('assert'); +const http = require('http'); + +// Server that delays reading to keep socket.writableLength > 0 during flush. +const server = http.createServer(common.mustCall((req, res) => { + setTimeout(() => { + req.resume(); + req.on('end', () => res.end('ok')); + }, 500); +}, 2)); + +server.listen(0, common.mustCall(() => { + const port = server.address().port; + const agent = new http.Agent({ keepAlive: true }); + + // Request A: creates socket with HWM=2MB. + http.request({ + host: 'localhost', port, method: 'POST', agent, + highWaterMark: 2 * 1024 * 1024, + }, common.mustCall((res) => { + res.resume(); + res.on('end', common.mustCall(() => { + // Wait for socket to return to pool. + setTimeout(common.mustCall(() => { + // Request B: default HWM (64KB), reuses socket (HWM=2MB). + // Write 500KB: above OM HWM (64KB), below socket HWM (2MB). + const reqB = http.request({ + host: 'localhost', port, method: 'POST', agent, + }, common.mustCall((res2) => { + res2.resume(); + res2.on('end', common.mustCall(() => { + server.close(); + })); + })); + + const result = reqB.write(Buffer.alloc(500 * 1024)); + assert.strictEqual(result, false); + + // Drain must fire — no deadlock. + reqB.on('drain', common.mustCall(() => { + reqB.end(); + })); + }), 100); + })); + })).end('x'); +}));