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

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);

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');
     }

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');
+  }
+}));

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');
+}));