http: emit 'drain' on OutgoingMessage only after buffers drain

Previously, socketOnDrain could be invoked synchronously from
_flushOutput (via _onPendingData -> updateOutgoingData) while the bytes
just handed to the socket were still buffered and while outputSize had
not yet been reset on the OutgoingMessage. The 'drain' event fired even
though res.writableLength was non-zero, breaking the invariant a user
would reasonably expect after `while (!res.write(...));`.

Gate the emission in socketOnDrain on msg.writableLength === 0 (which
also covers outputSize + chunked buffer + socket.writableLength), and
apply the same check in OutgoingMessage._flush so that 'drain' is only
emitted when the response is genuinely drained. The socket's own
'drain' event will otherwise propagate through socketOnDrain when the
socket buffer actually empties.

Signed-off-by: Robert Nagy <[email protected]>
Assisted-by: Claude Opus 4.6 (1M context) <[email protected]>
PR-URL: #62936
Reviewed-By: Matteo Collina <[email protected]>
Reviewed-By: Ethan Arrowood <[email protected]>
Reviewed-By: Gürgün Dayıoğlu <[email protected]>

Author: ronag

lib/_http_outgoing.js

@@ -1148,12 +1148,12 @@ OutgoingMessage.prototype._flush = function _flush() {
 
   if (socket?.writable) {
     // There might be remaining data in this.output; write it out
-    const ret = this._flushOutput(socket);
+    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 (ret && this[kNeedDrain]) {
+    } else if (this[kNeedDrain] && this.writableLength === 0) {
       this[kNeedDrain] = false;
       this.emit('drain');
     }

lib/_http_server.js

@@ -820,7 +820,13 @@ function socketOnDrain(socket, state) {
   }
 
   const msg = socket._httpMessage;
-  if (msg && !msg.finished && msg[kNeedDrain]) {
+  // Only emit 'drain' once the message has no data pending anywhere, so that
+  // msg.writableLength === 0 when the event fires. socketOnDrain is called
+  // synchronously from updateOutgoingData during _flushOutput, at which point
+  // the bytes we just handed to the socket (or the stale outputSize) mean
+  // the message is not actually drained yet - we wait for the socket's
+  // own 'drain' event instead.
+  if (msg && !msg.finished && msg[kNeedDrain] && msg.writableLength === 0) {
     msg[kNeedDrain] = false;
     msg.emit('drain');
   }

test/parallel/test-http-outgoing-drain-writable-length.js

@@ -0,0 +1,58 @@
+'use strict';
+// Regression test: when a pipelined ServerResponse (whose writes were
+// buffered in outputData while the socket belonged to a previous response)
+// is finally assigned its socket and flushed, 'drain' must not be emitted
+// until the socket's own buffer has actually drained. Previously,
+// socketOnDrain was called synchronously from _flushOutput via _onPendingData
+// and emitted 'drain' even though the bytes we just wrote were still sitting
+// in the socket's writable buffer, so res.writableLength was non-zero.
+
+const common = require('../common');
+const http = require('http');
+const net = require('net');
+const assert = require('assert');
+
+let step = 0;
+
+const server = http.createServer(common.mustCall((req, res) => {
+  step++;
+
+  if (step === 1) {
+    // Keep the first response open briefly so the second is queued with
+    // res.socket === null.
+    res.writeHead(200, { 'Content-Type': 'text/plain' });
+    setTimeout(() => res.end('ok'), 50);
+    return;
+  }
+
+  // Second (pipelined) response - queued in state.outgoing, no socket yet.
+  assert.strictEqual(res.socket, null);
+
+  res.writeHead(200, { 'Content-Type': 'text/plain' });
+
+  // Write past the response's highWaterMark so res.write() returns false
+  // and kNeedDrain is set. Data is buffered in outputData.
+  const chunk = Buffer.alloc(16 * 1024, 'x');
+  while (res.write(chunk));
+  assert.strictEqual(res.writableNeedDrain, true);
+
+  res.on('drain', common.mustCall(() => {
+    assert.strictEqual(
+      res.writableLength, 0,
+      `'drain' fired with writableLength=${res.writableLength}`,
+    );
+    res.end();
+    server.close();
+  }));
+}, 2));
+
+server.listen(0, common.mustCall(function() {
+  const port = this.address().port;
+  const client = net.connect(port);
+  client.write(
+    `GET /1 HTTP/1.1\r\nHost: localhost:${port}\r\n\r\n` +
+    `GET /2 HTTP/1.1\r\nHost: localhost:${port}\r\n\r\n`,
+  );
+  client.resume();
+  client.on('error', () => {});
+}));