http: emit drain on socket takeover and avoid stale HWM reuse

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 <trivenay@amazon.com>
Fixes: #64680
Refs: #64653
Refs: #62936
PR-URL: #64991
Reviewed-By: Robert Nagy <ronagy@icloud.com>
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day>

Author: trivenay

lib/_http_agent.js

@@ -392,6 +392,16 @@ Agent.prototype.addRequest = function addRequest(req, options, port/* legacy */,
 constsockLen=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);

lib/_http_outgoing.js

@@ -1230,12 +1230,14 @@ OutgoingMessage.prototype._flush = function _flush() {
 
 if(socket?.writable){
 // There might be remaining data in this.output; write it out
-this._flushOutput(socket);
+constret=this._flushOutput(socket);
 
 if(this.finished){
 // This is a queue to the server or client to bring in the next this.
 this._finish();
-}elseif(this[kNeedDrain]&&this.writableLength===0){
+}elseif(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');
 }

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
+
+constcommon=require('../common');
+constassert=require('assert');
+consthttp=require('http');
+
+constserver=http.createServer(common.mustCall((req,res)=>{
+req.resume();
+req.on('end',()=>res.end('ok'));
+},2));
+
+server.listen(0,common.mustCall(()=>{
+constport=server.address().port;
+constagent=newhttp.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');
+
+functionrequestB(){
+constfreeCount=Object.values(agent.freeSockets).flat().length;
+assert.strictEqual(freeCount,1);
+
+// Request B: HWM=10KB — agent must sync the reused socket's HWM.
+constreqB=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');
+}
+}));

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
+
+constcommon=require('../common');
+constassert=require('assert');
+consthttp=require('http');
+
+// Server that delays reading to keep socket.writableLength > 0 during flush.
+constserver=http.createServer(common.mustCall((req,res)=>{
+setTimeout(()=>{
+req.resume();
+req.on('end',()=>res.end('ok'));
+},500);
+},2));
+
+server.listen(0,common.mustCall(()=>{
+constport=server.address().port;
+constagent=newhttp.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).
+constreqB=http.request({
+host: 'localhost', port,method: 'POST', agent,
+},common.mustCall((res2)=>{
+res2.resume();
+res2.on('end',common.mustCall(()=>{
+server.close();
+}));
+}));
+
+constresult=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');
+}));

lib/_http_agent.js

@@ -392,6 +392,16 @@ Agent.prototype.addRequest = function addRequest(req, options, port/* legacy */,
 constsockLen=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);

lib/_http_outgoing.js

@@ -1230,12 +1230,14 @@ OutgoingMessage.prototype._flush = function _flush() {
 
 if(socket?.writable){
 // There might be remaining data in this.output; write it out
-this._flushOutput(socket);
+constret=this._flushOutput(socket);
 
 if(this.finished){
 // This is a queue to the server or client to bring in the next this.
 this._finish();
-}elseif(this[kNeedDrain]&&this.writableLength===0){
+}elseif(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');
 }

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
+
+constcommon=require('../common');
+constassert=require('assert');
+consthttp=require('http');
+
+constserver=http.createServer(common.mustCall((req,res)=>{
+req.resume();
+req.on('end',()=>res.end('ok'));
+},2));
+
+server.listen(0,common.mustCall(()=>{
+constport=server.address().port;
+constagent=newhttp.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');
+
+functionrequestB(){
+constfreeCount=Object.values(agent.freeSockets).flat().length;
+assert.strictEqual(freeCount,1);
+
+// Request B: HWM=10KB — agent must sync the reused socket's HWM.
+constreqB=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');
+}
+}));

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
+
+constcommon=require('../common');
+constassert=require('assert');
+consthttp=require('http');
+
+// Server that delays reading to keep socket.writableLength > 0 during flush.
+constserver=http.createServer(common.mustCall((req,res)=>{
+setTimeout(()=>{
+req.resume();
+req.on('end',()=>res.end('ok'));
+},500);
+},2));
+
+server.listen(0,common.mustCall(()=>{
+constport=server.address().port;
+constagent=newhttp.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).
+constreqB=http.request({
+host: 'localhost', port,method: 'POST', agent,
+},common.mustCall((res2)=>{
+res2.resume();
+res2.on('end',common.mustCall(()=>{
+server.close();
+}));
+}));
+
+constresult=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');
+}));

lib/_http_agent.js

@@ -392,6 +392,16 @@ Agent.prototype.addRequest = function addRequest(req, options, port/* legacy */,
 constsockLen=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);

lib/_http_outgoing.js

@@ -1230,12 +1230,14 @@ OutgoingMessage.prototype._flush = function _flush() {
 
 if(socket?.writable){
 // There might be remaining data in this.output; write it out
-this._flushOutput(socket);
+constret=this._flushOutput(socket);
 
 if(this.finished){
 // This is a queue to the server or client to bring in the next this.
 this._finish();
-}elseif(this[kNeedDrain]&&this.writableLength===0){
+}elseif(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');
 }

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
+
+constcommon=require('../common');
+constassert=require('assert');
+consthttp=require('http');
+
+constserver=http.createServer(common.mustCall((req,res)=>{
+req.resume();
+req.on('end',()=>res.end('ok'));
+},2));
+
+server.listen(0,common.mustCall(()=>{
+constport=server.address().port;
+constagent=newhttp.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');
+
+functionrequestB(){
+constfreeCount=Object.values(agent.freeSockets).flat().length;
+assert.strictEqual(freeCount,1);
+
+// Request B: HWM=10KB — agent must sync the reused socket's HWM.
+constreqB=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');
+}
+}));

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
+
+constcommon=require('../common');
+constassert=require('assert');
+consthttp=require('http');
+
+// Server that delays reading to keep socket.writableLength > 0 during flush.
+constserver=http.createServer(common.mustCall((req,res)=>{
+setTimeout(()=>{
+req.resume();
+req.on('end',()=>res.end('ok'));
+},500);
+},2));
+
+server.listen(0,common.mustCall(()=>{
+constport=server.address().port;
+constagent=newhttp.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).
+constreqB=http.request({
+host: 'localhost', port,method: 'POST', agent,
+},common.mustCall((res2)=>{
+res2.resume();
+res2.on('end',common.mustCall(()=>{
+server.close();
+}));
+}));
+
+constresult=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');
+}));

lib/_http_agent.js

@@ -392,6 +392,16 @@ Agent.prototype.addRequest = function addRequest(req, options, port/* legacy */,
 constsockLen=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);

lib/_http_outgoing.js

@@ -1230,12 +1230,14 @@ OutgoingMessage.prototype._flush = function _flush() {
 
 if(socket?.writable){
 // There might be remaining data in this.output; write it out
-this._flushOutput(socket);
+constret=this._flushOutput(socket);
 
 if(this.finished){
 // This is a queue to the server or client to bring in the next this.
 this._finish();
-}elseif(this[kNeedDrain]&&this.writableLength===0){
+}elseif(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');
 }

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
+
+constcommon=require('../common');
+constassert=require('assert');
+consthttp=require('http');
+
+constserver=http.createServer(common.mustCall((req,res)=>{
+req.resume();
+req.on('end',()=>res.end('ok'));
+},2));
+
+server.listen(0,common.mustCall(()=>{
+constport=server.address().port;
+constagent=newhttp.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');
+
+functionrequestB(){
+constfreeCount=Object.values(agent.freeSockets).flat().length;
+assert.strictEqual(freeCount,1);
+
+// Request B: HWM=10KB — agent must sync the reused socket's HWM.
+constreqB=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');
+}
+}));

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
+
+constcommon=require('../common');
+constassert=require('assert');
+consthttp=require('http');
+
+// Server that delays reading to keep socket.writableLength > 0 during flush.
+constserver=http.createServer(common.mustCall((req,res)=>{
+setTimeout(()=>{
+req.resume();
+req.on('end',()=>res.end('ok'));
+},500);
+},2));
+
+server.listen(0,common.mustCall(()=>{
+constport=server.address().port;
+constagent=newhttp.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).
+constreqB=http.request({
+host: 'localhost', port,method: 'POST', agent,
+},common.mustCall((res2)=>{
+res2.resume();
+res2.on('end',common.mustCall(()=>{
+server.close();
+}));
+}));
+
+constresult=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');
+}));

lib/_http_agent.js

@@ -392,6 +392,16 @@ Agent.prototype.addRequest = function addRequest(req, options, port/* legacy */,
 constsockLen=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);

lib/_http_outgoing.js

@@ -1230,12 +1230,14 @@ OutgoingMessage.prototype._flush = function _flush() {
 
 if(socket?.writable){
 // There might be remaining data in this.output; write it out
-this._flushOutput(socket);
+constret=this._flushOutput(socket);
 
 if(this.finished){
 // This is a queue to the server or client to bring in the next this.
 this._finish();
-}elseif(this[kNeedDrain]&&this.writableLength===0){
+}elseif(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');
 }

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
+
+constcommon=require('../common');
+constassert=require('assert');
+consthttp=require('http');
+
+constserver=http.createServer(common.mustCall((req,res)=>{
+req.resume();
+req.on('end',()=>res.end('ok'));
+},2));
+
+server.listen(0,common.mustCall(()=>{
+constport=server.address().port;
+constagent=newhttp.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');
+
+functionrequestB(){
+constfreeCount=Object.values(agent.freeSockets).flat().length;
+assert.strictEqual(freeCount,1);
+
+// Request B: HWM=10KB — agent must sync the reused socket's HWM.
+constreqB=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');
+}
+}));

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
+
+constcommon=require('../common');
+constassert=require('assert');
+consthttp=require('http');
+
+// Server that delays reading to keep socket.writableLength > 0 during flush.
+constserver=http.createServer(common.mustCall((req,res)=>{
+setTimeout(()=>{
+req.resume();
+req.on('end',()=>res.end('ok'));
+},500);
+},2));
+
+server.listen(0,common.mustCall(()=>{
+constport=server.address().port;
+constagent=newhttp.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).
+constreqB=http.request({
+host: 'localhost', port,method: 'POST', agent,
+},common.mustCall((res2)=>{
+res2.resume();
+res2.on('end',common.mustCall(()=>{
+server.close();
+}));
+}));
+
+constresult=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');
+}));

lib/_http_agent.js

@@ -392,6 +392,16 @@ Agent.prototype.addRequest = function addRequest(req, options, port/* legacy */,
 constsockLen=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);

lib/_http_outgoing.js

@@ -1230,12 +1230,14 @@ OutgoingMessage.prototype._flush = function _flush() {
 
 if(socket?.writable){
 // There might be remaining data in this.output; write it out
-this._flushOutput(socket);
+constret=this._flushOutput(socket);
 
 if(this.finished){
 // This is a queue to the server or client to bring in the next this.
 this._finish();
-}elseif(this[kNeedDrain]&&this.writableLength===0){
+}elseif(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');
 }

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
+
+constcommon=require('../common');
+constassert=require('assert');
+consthttp=require('http');
+
+constserver=http.createServer(common.mustCall((req,res)=>{
+req.resume();
+req.on('end',()=>res.end('ok'));
+},2));
+
+server.listen(0,common.mustCall(()=>{
+constport=server.address().port;
+constagent=newhttp.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');
+
+functionrequestB(){
+constfreeCount=Object.values(agent.freeSockets).flat().length;
+assert.strictEqual(freeCount,1);
+
+// Request B: HWM=10KB — agent must sync the reused socket's HWM.
+constreqB=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');
+}
+}));

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
+
+constcommon=require('../common');
+constassert=require('assert');
+consthttp=require('http');
+
+// Server that delays reading to keep socket.writableLength > 0 during flush.
+constserver=http.createServer(common.mustCall((req,res)=>{
+setTimeout(()=>{
+req.resume();
+req.on('end',()=>res.end('ok'));
+},500);
+},2));
+
+server.listen(0,common.mustCall(()=>{
+constport=server.address().port;
+constagent=newhttp.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).
+constreqB=http.request({
+host: 'localhost', port,method: 'POST', agent,
+},common.mustCall((res2)=>{
+res2.resume();
+res2.on('end',common.mustCall(()=>{
+server.close();
+}));
+}));
+
+constresult=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');
+}));

lib/_http_agent.js

@@ -392,6 +392,16 @@ Agent.prototype.addRequest = function addRequest(req, options, port/* legacy */,
 constsockLen=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);

lib/_http_outgoing.js

@@ -1230,12 +1230,14 @@ OutgoingMessage.prototype._flush = function _flush() {
 
 if(socket?.writable){
 // There might be remaining data in this.output; write it out
-this._flushOutput(socket);
+constret=this._flushOutput(socket);
 
 if(this.finished){
 // This is a queue to the server or client to bring in the next this.
 this._finish();
-}elseif(this[kNeedDrain]&&this.writableLength===0){
+}elseif(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');
 }

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
+
+constcommon=require('../common');
+constassert=require('assert');
+consthttp=require('http');
+
+constserver=http.createServer(common.mustCall((req,res)=>{
+req.resume();
+req.on('end',()=>res.end('ok'));
+},2));
+
+server.listen(0,common.mustCall(()=>{
+constport=server.address().port;
+constagent=newhttp.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');
+
+functionrequestB(){
+constfreeCount=Object.values(agent.freeSockets).flat().length;
+assert.strictEqual(freeCount,1);
+
+// Request B: HWM=10KB — agent must sync the reused socket's HWM.
+constreqB=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');
+}
+}));

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
+
+constcommon=require('../common');
+constassert=require('assert');
+consthttp=require('http');
+
+// Server that delays reading to keep socket.writableLength > 0 during flush.
+constserver=http.createServer(common.mustCall((req,res)=>{
+setTimeout(()=>{
+req.resume();
+req.on('end',()=>res.end('ok'));
+},500);
+},2));
+
+server.listen(0,common.mustCall(()=>{
+constport=server.address().port;
+constagent=newhttp.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).
+constreqB=http.request({
+host: 'localhost', port,method: 'POST', agent,
+},common.mustCall((res2)=>{
+res2.resume();
+res2.on('end',common.mustCall(()=>{
+server.close();
+}));
+}));
+
+constresult=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');
+}));

lib/_http_agent.js

@@ -392,6 +392,16 @@ Agent.prototype.addRequest = function addRequest(req, options, port/* legacy */,
 constsockLen=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);

lib/_http_outgoing.js

@@ -1230,12 +1230,14 @@ OutgoingMessage.prototype._flush = function _flush() {
 
 if(socket?.writable){
 // There might be remaining data in this.output; write it out
-this._flushOutput(socket);
+constret=this._flushOutput(socket);
 
 if(this.finished){
 // This is a queue to the server or client to bring in the next this.
 this._finish();
-}elseif(this[kNeedDrain]&&this.writableLength===0){
+}elseif(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');
 }

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
+
+constcommon=require('../common');
+constassert=require('assert');
+consthttp=require('http');
+
+constserver=http.createServer(common.mustCall((req,res)=>{
+req.resume();
+req.on('end',()=>res.end('ok'));
+},2));
+
+server.listen(0,common.mustCall(()=>{
+constport=server.address().port;
+constagent=newhttp.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');
+
+functionrequestB(){
+constfreeCount=Object.values(agent.freeSockets).flat().length;
+assert.strictEqual(freeCount,1);
+
+// Request B: HWM=10KB — agent must sync the reused socket's HWM.
+constreqB=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');
+}
+}));

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
+
+constcommon=require('../common');
+constassert=require('assert');
+consthttp=require('http');
+
+// Server that delays reading to keep socket.writableLength > 0 during flush.
+constserver=http.createServer(common.mustCall((req,res)=>{
+setTimeout(()=>{
+req.resume();
+req.on('end',()=>res.end('ok'));
+},500);
+},2));
+
+server.listen(0,common.mustCall(()=>{
+constport=server.address().port;
+constagent=newhttp.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).
+constreqB=http.request({
+host: 'localhost', port,method: 'POST', agent,
+},common.mustCall((res2)=>{
+res2.resume();
+res2.on('end',common.mustCall(()=>{
+server.close();
+}));
+}));
+
+constresult=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');
+}));