Skip to content

Commit afb656c

Browse files
committed
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
1 parent a008632 commit afb656c

5 files changed

Lines changed: 153 additions & 5 deletions

‎lib/_http_agent.js‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -392,6 +392,17 @@ Agent.prototype.addRequest = function addRequest(req, options, port/* legacy */,
392392
const sockLen = freeLen + this.sockets[name].length;
393393

394394
// Reusing a socket from the pool.
395+
// If the caller specified a highWaterMark that differs from the pooled
396+
// socket's writableHighWaterMark, skip reuse and create a fresh socket
397+
// so that backpressure semantics match what the caller requested.
398+
if (socket && options.highWaterMark != null &&
399+
socket.writableHighWaterMark !== options.highWaterMark) {
400+
debug('skip reuse, HWM mismatch (socket=%d, request=%d)',
401+
socket.writableHighWaterMark, options.highWaterMark);
402+
socket.destroy();
403+
socket = null;
404+
}
405+
395406
if (socket) {
396407
asyncResetHandle(socket);
397408
this.reuseSocket(socket, req);

‎lib/_http_outgoing.js‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1201,7 +1201,10 @@ OutgoingMessage.prototype._flush = function _flush() {
12011201
if (this.finished) {
12021202
// This is a queue to the server or client to bring in the next this.
12031203
this._finish();
1204-
} else if (this[kNeedDrain] && this.writableLength === 0) {
1204+
} else if (this[kNeedDrain]) {
1205+
// _flushOutput() handed all buffered data to the socket; the OM's
1206+
// backpressure concern is resolved. Subsequent writes go directly
1207+
// to the socket where socket-level backpressure takes over.
12051208
this[kNeedDrain] = false;
12061209
this.emit('drain');
12071210
}
Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
'use strict';
2+
3+
// Regression test: when a pooled socket's writableHighWaterMark differs from
4+
// the new request's highWaterMark, the agent must not reuse that socket.
5+
// Reusing it causes the user's backpressure threshold to be silently ignored
6+
// (and can deadlock under TCP backpressure conditions).
7+
//
8+
// See: https://github.com/nodejs/node/pull/64653#issuecomment-5047543911
9+
10+
const common = require('../common');
11+
const assert = require('assert');
12+
const http = require('http');
13+
14+
const server = http.createServer(common.mustCall((req, res) => {
15+
req.resume();
16+
req.on('end', () => res.end('ok'));
17+
}, 3));
18+
19+
server.listen(0, common.mustCall(() => {
20+
const port = server.address().port;
21+
const agent = new http.Agent({ keepAlive: true });
22+
23+
// Request A: creates socket with HWM=1MB.
24+
http.request({
25+
host: 'localhost', port, method: 'POST', agent,
26+
highWaterMark: 1024 * 1024,
27+
}, common.mustCall((res) => {
28+
res.resume();
29+
res.on('end', common.mustCall(() => {
30+
// Wait for socket to return to pool.
31+
setTimeout(requestB, 100);
32+
}));
33+
})).end('x');
34+
35+
function requestB() {
36+
const freeCount = Object.values(agent.freeSockets).flat().length;
37+
assert.strictEqual(freeCount, 1);
38+
39+
// Request B: HWM=10KB — must get a fresh socket, not the 1MB one.
40+
const reqB = http.request({
41+
host: 'localhost', port, method: 'POST', agent,
42+
highWaterMark: 10 * 1024,
43+
}, common.mustCall((res) => {
44+
res.resume();
45+
res.on('end', common.mustCall(() => {
46+
setTimeout(requestC, 100);
47+
}));
48+
}));
49+
50+
reqB.on('socket', common.mustCall((socket) => {
51+
assert.strictEqual(socket.writableHighWaterMark, 10 * 1024);
52+
}));
53+
54+
reqB.end('y');
55+
}
56+
57+
function requestC() {
58+
// Request C: same HWM=10KB — should reuse the socket from Request B.
59+
const reqC = http.request({
60+
host: 'localhost', port, method: 'POST', agent,
61+
highWaterMark: 10 * 1024,
62+
}, common.mustCall((res) => {
63+
res.resume();
64+
res.on('end', common.mustCall(() => {
65+
server.close();
66+
}));
67+
}));
68+
69+
reqC.on('socket', common.mustCall((socket) => {
70+
assert.strictEqual(socket.writableHighWaterMark, 10 * 1024);
71+
}));
72+
73+
reqC.end('z');
74+
}
75+
}));

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

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -37,10 +37,12 @@ const server = http.createServer(common.mustCall((req, res) => {
3737
assert.strictEqual(res.writableNeedDrain, true);
3838

3939
res.on('drain', common.mustCall(() => {
40-
assert.strictEqual(
41-
res.writableLength, 0,
42-
`'drain' fired with writableLength=${res.writableLength}`,
43-
);
40+
// After the changeover from pre-socket buffering to socket-connected
41+
// writing, drain fires once the OM's own buffer (outputData) has been
42+
// handed off to the socket. The socket may still have data queued
43+
// in libuv — that's the socket's backpressure domain, handled by
44+
// subsequent write() calls returning false via Path A.
45+
assert.strictEqual(res.outputSize, 0);
4446
res.end();
4547
server.close();
4648
}));
Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,57 @@
1+
'use strict';
2+
3+
// Regression test: when _flush() hands buffered data to a socket whose
4+
// writableHighWaterMark is higher than the OutgoingMessage's kHighWaterMark,
5+
// drain must still fire. Previously, _flush() gated drain emission on
6+
// writableLength === 0, which included socket.writableLength — but the
7+
// socket was never backpressured (data < socket HWM), so drain never fired.
8+
//
9+
// See: https://github.com/nodejs/node/issues/64680
10+
11+
const common = require('../common');
12+
const assert = require('assert');
13+
const http = require('http');
14+
15+
// Server that delays reading to keep socket.writableLength > 0 during flush.
16+
const server = http.createServer(common.mustCall((req, res) => {
17+
setTimeout(() => {
18+
req.resume();
19+
req.on('end', () => res.end('ok'));
20+
}, 500);
21+
}, 2));
22+
23+
server.listen(0, common.mustCall(() => {
24+
const port = server.address().port;
25+
const agent = new http.Agent({ keepAlive: true });
26+
27+
// Request A: creates socket with HWM=2MB.
28+
http.request({
29+
host: 'localhost', port, method: 'POST', agent,
30+
highWaterMark: 2 * 1024 * 1024,
31+
}, common.mustCall((res) => {
32+
res.resume();
33+
res.on('end', common.mustCall(() => {
34+
// Wait for socket to return to pool.
35+
setTimeout(common.mustCall(() => {
36+
// Request B: default HWM (64KB), reuses socket (HWM=2MB).
37+
// Write 500KB: above OM HWM (64KB), below socket HWM (2MB).
38+
const reqB = http.request({
39+
host: 'localhost', port, method: 'POST', agent,
40+
}, common.mustCall((res2) => {
41+
res2.resume();
42+
res2.on('end', common.mustCall(() => {
43+
server.close();
44+
}));
45+
}));
46+
47+
const result = reqB.write(Buffer.alloc(500 * 1024));
48+
assert.strictEqual(result, false);
49+
50+
// Drain must fire — no deadlock.
51+
reqB.on('drain', common.mustCall(() => {
52+
reqB.end();
53+
}));
54+
}), 100);
55+
}));
56+
})).end('x');
57+
}));

0 commit comments

Comments
 (0)