From 769270520472d21c9ec27152a725e44d04748b41 Mon Sep 17 00:00:00 2001 From: Naman Trivedi Date: Thu, 28 May 2026 20:31:04 +0000 Subject: [PATCH] Fix drain deadlock on Node 24.16+: set customer highWaterMark on OutgoingMessage --- .../tests/stream/response-stream.test.ts | 133 +++++++++++++++++- src/stream/response-stream.ts | 20 +++ 2 files changed, 152 insertions(+), 1 deletion(-) diff --git a/integration-tests/tests/stream/response-stream.test.ts b/integration-tests/tests/stream/response-stream.test.ts index 565f13d..e9f6bbf 100644 --- a/integration-tests/tests/stream/response-stream.test.ts +++ b/integration-tests/tests/stream/response-stream.test.ts @@ -278,7 +278,7 @@ describe("response-stream", () => { hostname: "127.0.0.1", port, path: "/asd/efg", - highWaterMark: 16, + highWaterMark: 1024, }, }); @@ -811,4 +811,135 @@ describe("response-stream", () => { }, ); }); + + describe("highWaterMark enforcement", () => { + it("does not deadlock when highWaterMark > default and drain is awaited", async () => { + await withTestServer( + (req, res) => { + req.once("end", () => { + res.writeHead(200, { "Content-Type": "text/plain" }); + res.end(); + }); + }, + + async (port, recorded) => { + // GIVEN: highWaterMark 100KB, writing 64KB chunks + // This specifically tests the pre-socket-connect path (Path B). + // Without the fix, on Node 24.16+, this deadlocks because: + // - OM default (64KB) < chunk (64KB + overhead) → write() returns false + // - Socket HWM (100KB) > chunk → socket never backpressured + // - Socket never emits drain → customer hangs forever + const { request, responseDone } = createResponseStream({ + httpOptions: { + http, + agent: new http.Agent(), + method: "POST", + hostname: "127.0.0.1", + port, + path: "/asd/efg", + highWaterMark: 100000, + }, + }); + + // WHEN: write chunks and properly handle backpressure + const chunk = Buffer.alloc(64 * 1024, "x"); + for (let i = 0; i < 10; i++) { + const ok = request.write(chunk); + if (!ok) { + await new Promise((resolve) => + request.once("drain", resolve), + ); + } + } + + // THEN: should complete without deadlock + request.end(); + await responseDone.catch(() => {}); + + expect(recorded).toHaveLength(1); + // 10 chunks of 64KB = 640KB + chunked encoding overhead + expect(recorded[0].body.length).toBeGreaterThanOrEqual(640 * 1024); + }, + ); + }); + + it("write returns true for chunk below custom highWaterMark even before socket connects", async () => { + await withTestServer( + (req, res) => { + req.once("end", () => { + res.writeHead(200, { "Content-Type": "text/plain" }); + res.end(); + }); + }, + + async (port, recorded) => { + // GIVEN: highWaterMark 200KB, writing 80KB immediately (no await) + // The first write happens before socket connects (Path B in _writeRaw). + // Without the fix: OM uses default 64KB, 80KB > 64KB → returns false + // With the fix: OM uses 200KB, 80KB < 200KB → returns true + const { request, responseDone } = createResponseStream({ + httpOptions: { + http, + agent: new http.Agent(), + method: "POST", + hostname: "127.0.0.1", + port, + path: "/asd/efg", + highWaterMark: 200000, + }, + }); + + // WHEN: write 80KB synchronously (before socket connects) + const chunk = Buffer.alloc(80 * 1024, "x"); + const result = request.write(chunk); + + // THEN: should return true (80KB < 200KB custom HWM) + expect(result).toBe(true); + + request.end(); + await responseDone.catch(() => {}); + expect(recorded).toHaveLength(1); + }, + ); + }); + + it("write returns false for chunk above low custom highWaterMark", async () => { + await withTestServer( + (req, res) => { + req.once("end", () => { + res.writeHead(200, { "Content-Type": "text/plain" }); + res.end(); + }); + }, + + async (port, recorded) => { + // GIVEN: highWaterMark 512 bytes, writing 2KB + // Without the fix: OM uses default 64KB, 2KB < 64KB → returns true (HWM ignored) + // With the fix: OM uses 512, 2KB > 512 → returns false (HWM enforced) + const { request, responseDone } = createResponseStream({ + httpOptions: { + http, + agent: new http.Agent(), + method: "POST", + hostname: "127.0.0.1", + port, + path: "/asd/efg", + highWaterMark: 512, + }, + }); + + // WHEN: write 2KB synchronously + const chunk = Buffer.alloc(2048, "x"); + const result = request.write(chunk); + + // THEN: should return false (2KB > 512 custom HWM) + expect(result).toBe(false); + + request.end(); + await responseDone.catch(() => {}); + expect(recorded).toHaveLength(1); + }, + ); + }); + }); }); diff --git a/src/stream/response-stream.ts b/src/stream/response-stream.ts index b22e767..0d4d5c5 100644 --- a/src/stream/response-stream.ts +++ b/src/stream/response-stream.ts @@ -161,6 +161,26 @@ export function createResponseStream( req.destroy(err); }); + // Set the customer's highWaterMark on the OutgoingMessage's internal + // kHighWaterMark. Without this, the OM uses its default (64KB) for write() + // return values before the socket connects, causing the customer's HWM to + // be ignored. On Node 24.16+ (PR #62936), this mismatch causes a drain + // deadlock when socket HWM > OM HWM. + // + // We access kHighWaterMark via Object.getOwnPropertySymbols because Node + // does not expose a public API to set it on ClientRequest after construction. + // If the symbol is not found (e.g., future Node version changes internals), + // we silently fall back to the previous behavior — no crash, just the same + // pre-existing limitation that existed before this fix. + if (options.httpOptions.highWaterMark != null) { + const syms = Object.getOwnPropertySymbols(req); + const kHWM = syms.find((s) => s.toString() === "Symbol(kHighWaterMark)"); + if (kHWM) { + (req as unknown as Record)[kHWM] = + options.httpOptions.highWaterMark; + } + } + req.setContentType = (contentType: string) => { if (status !== STATUS_READY) { throw new InvalidStreamingOperation("Cannot set content-type, too late.");