From 46498569d2606076c3c4b66b9d4dc0ee84b35300 Mon Sep 17 00:00:00 2001 From: Tim Perry Date: Thu, 20 Aug 2026 08:38:31 +0100 Subject: [PATCH 1/3] http2: fix write deadlock exposed by larger window sizes This removes a guard (no reads while write pending) that creates this deadlock, which was added as a security mechanism. This guard is redundant given then other existing mechanisms, and a test is added to demonstrate that. Signed-off-by: Tim Perry --- benchmark/http2/full-duplex.js | 71 ++++++++++++ src/node_http2.cc | 11 +- ...test-http2-bidirectional-write-deadlock.js | 107 ++++++++++++++++++ .../test-http2-session-memory-unread-peer.js | 100 ++++++++++++++++ test/parallel/test-stream-pipeline-http2.js | 3 +- 5 files changed, 280 insertions(+), 12 deletions(-) create mode 100644 benchmark/http2/full-duplex.js create mode 100644 test/parallel/test-http2-bidirectional-write-deadlock.js create mode 100644 test/parallel/test-http2-session-memory-unread-peer.js diff --git a/benchmark/http2/full-duplex.js b/benchmark/http2/full-duplex.js new file mode 100644 index 000000000000..d2fec719f95a --- /dev/null +++ b/benchmark/http2/full-duplex.js @@ -0,0 +1,71 @@ +'use strict'; + +const common = require('../common.js'); +const fixtures = require('../../test/common/fixtures'); + +const bench = common.createBenchmark(main, { + n: [100], + streams: [2], + size: [4 * 1024 * 1024], + // Use the HTTP/2 protocol default. + window: [65535], +}, { + test: { size: 128 * 1024, window: 65535 }, +}); + +function main({ n, streams, size, window }) { + const http2 = require('http2'); + const payload = Buffer.alloc(size); + const server = http2.createSecureServer({ + key: fixtures.readKey('agent1-key.pem'), + cert: fixtures.readKey('agent1-cert.pem'), + settings: { initialWindowSize: window }, + }); + + let completed = 0; + let batches = 0; + + function onTransferComplete() { + if (++completed !== streams * 2) + return; + + if (++batches === n) { + // Report combined upload and download throughput in MiB/s. + bench.end(n * streams * size * 2 / (1024 * 1024)); + client.close(); + server.close(); + return; + } + + startBatch(); + } + + server.on('stream', (stream) => { + stream.resume(); + stream.on('end', onTransferComplete); + stream.respond(); + stream.end(payload); + }); + + let client; + function startBatch() { + completed = 0; + for (let i = 0; i < streams; i++) { + const request = client.request({ ':method': 'POST' }); + request.resume(); + request.on('end', onTransferComplete); + request.end(payload); + } + } + + server.listen(0, () => { + client = http2.connect(`https://localhost:${server.address().port}`, { + rejectUnauthorized: false, + settings: { initialWindowSize: window }, + }); + client.on('connect', () => { + bench.start(); + startBatch(); + }); + }); +} diff --git a/src/node_http2.cc b/src/node_http2.cc index 58f12c5561b5..45c4ed1530f9 100644 --- a/src/node_http2.cc +++ b/src/node_http2.cc @@ -1485,15 +1485,6 @@ int Http2Session::OnDataChunkReceived(nghttp2_session* handle, } } while (len != 0); - // If we are currently waiting for a write operation to finish, we should - // tell nghttp2 that we want to wait before we process more input data. - if (session->is_write_in_progress()) { - CHECK(session->is_reading_stopped()); - session->set_receive_paused(); - Debug(session, "receive paused"); - return NGHTTP2_ERR_PAUSE; - } - return 0; } @@ -1942,7 +1933,7 @@ void Http2Session::MaybeStopReading() { if (is_reading_stopped() || is_closing()) return; int want_read = nghttp2_session_want_read(session_.get()); Debug(this, "wants read? %d", want_read); - if (want_read == 0 || is_write_in_progress()) { + if (want_read == 0) { set_reading_stopped(); stream_->ReadStop(); } diff --git a/test/parallel/test-http2-bidirectional-write-deadlock.js b/test/parallel/test-http2-bidirectional-write-deadlock.js new file mode 100644 index 000000000000..9dd4b91c5234 --- /dev/null +++ b/test/parallel/test-http2-bidirectional-write-deadlock.js @@ -0,0 +1,107 @@ +'use strict'; + +// Regression test against deadlocks between two HTTP/2 peers that are both +// writing at the same time. +// +// To bound how much it buffered while output was backed up, an Http2Session +// used to stop reading from its socket whenever a write was in flight, and +// resume only once that write completed. When the peer was itself blocked +// writing, that write never completed, so the session never read again and +// the connection hung forever with no error and no timeout. +// +// Rather than relying on kernel socket buffers filling up - which depends on +// the platform and configured window sizes - this models one half of that +// cycle directly. The client's socket forwards a write but does not report it +// as complete, substituting for a write blocked because the peer is not +// reading. Only after that write is stalled does the server send its response +// body. A session that stops reading while writing never sees it. + +const common = require('../common'); +if (!common.hasCrypto) + common.skip('missing crypto'); +const assert = require('assert'); +const http2 = require('http2'); +const net = require('net'); +const { Duplex } = require('stream'); + +const BODY = 'the response body'; + +const heldCallbacks = []; +let serverStream; + +let stallWrites = false; + +// Client-side socket that forwards writes to a real connection, but can leave +// their completion callbacks pending to model a transport-blocked write. +class StalledClientSocket extends Duplex { + constructor(port) { + super(); + this.inner = net.connect(port, common.localhostIPv4); + this.inner.on('data', (chunk) => this.push(chunk)); + } + _read() { + // Incoming data is pushed as it arrives. + } + _write(chunk, encoding, callback) { + this.inner.write(chunk, encoding); + if (stallWrites) { + heldCallbacks.push(callback); + // Avoid writing from the server re-entrantly inside _write(). The + // ordering is still explicit: this callback is already held. + setImmediate(() => serverStream.end(BODY)); + return; + } + callback(); + } + _final(callback) { + callback(); + } + _destroy(err, callback) { + this.inner.destroy(); + callback(err); + } +} + +const server = http2.createServer(); + +server.on('stream', common.mustCall((stream) => { + // Send headers first. Their response event starts the stalled client write. + stream.respond(); + serverStream = stream; +})); + +server.listen(0, common.mustCall(() => { + const port = server.address().port; + + const client = http2.connect(`http://${common.localhostIPv4}:${port}`, { + createConnection: () => new StalledClientSocket(port), + }); + + const req = client.request({ ':method': 'POST' }); + + let received = ''; + + req.on('response', common.mustCall(() => { + // _write() will schedule the response body only after it has retained the + // callback, guaranteeing that the native write is still in progress. + stallWrites = true; + req.write(Buffer.alloc(256)); + })); + + req.on('data', (chunk) => { + received += chunk; + }); + + req.on('end', common.mustCall(() => { + assert.strictEqual(received, BODY); + assert.ok(heldCallbacks.length > 0, + 'test did not actually stall a socket write'); + + // Let the stalled writes complete so that everything can shut down. + stallWrites = false; + for (const callback of heldCallbacks) callback(); + + client.destroy(); + server.close(); + })); +})); diff --git a/test/parallel/test-http2-session-memory-unread-peer.js b/test/parallel/test-http2-session-memory-unread-peer.js new file mode 100644 index 000000000000..7f8358824eda --- /dev/null +++ b/test/parallel/test-http2-session-memory-unread-peer.js @@ -0,0 +1,100 @@ +'use strict'; + +// CVE-2019-9517 describes a peer that advertises HTTP/2 flow-control credit +// but does not drain its TCP socket, leaving the server holding responses it +// cannot write out. +// +// Node bounds this with maxSessionMemory. Once the budget is spent, further +// streams are refused rather than queued, so a client cannot make the server +// buffer a response for every request it opens. + +const common = require('../common'); +if (!common.hasCrypto) + common.skip('missing crypto'); +const assert = require('assert'); +const http2 = require('http2'); +const net = require('net'); +const { Duplex } = require('stream'); + +const TOTAL_REQUESTS = 200; +// Accepted count depends on kernel buffer sizes but this seems to +// cover most likely scenarios: +const MAX_ACCEPTED = 20; +const body = Buffer.alloc(1_000_000); + +let accepted = 0; +let client; + +// Client-side socket that forwards writes but never consumes incoming data. +class UnreadClientSocket extends Duplex { + constructor(port) { + super(); + this.inner = net.connect(port, common.localhostIPv4); + this.inner.pause(); + } + _read() { + // Deliberately never pull from the underlying socket. + } + _write(chunk, encoding, callback) { + this.inner.write(chunk, encoding, callback); + } + _final(callback) { + callback(); + } + _destroy(err, callback) { + this.inner.destroy(); + callback(err); + } +} + +const onStreamsRefused = common.mustCall(() => { + assert.ok(accepted >= 1, 'client never reached the server'); + client.destroy(); + server.close(); +}); + +const server = http2.createServer({ + maxSessionMemory: 1, + // Turn the rejected streams into an observable server-session failure. + maxSessionRejectedStreams: 0, +}); + +server.on('session', common.mustCall((session) => { + session.on('error', common.mustCall((err) => { + assert.strictEqual(err.code, 'ERR_HTTP2_ERROR'); + onStreamsRefused(); + })); +})); + +// At least one stream must be accepted, otherwise the assertion above could +// pass without the client ever having reached the server. +server.on('stream', common.mustCallAtLeast((stream) => { + accepted++; + // Assert here rather than once the session fails: if the bound does not + // hold there may be no session failure at all. + assert.ok(accepted <= MAX_ACCEPTED, + `server accepted ${accepted} of ${TOTAL_REQUESTS} streams ` + + 'from a peer that never reads; maxSessionMemory did not ' + + 'bound its outbound buffering'); + // The fatal session error also destroys every accepted stream. + stream.on('error', common.mustCall((err) => { + assert.strictEqual(err.code, 'ERR_HTTP2_ERROR'); + })); + stream.respond(); + stream.end(body); +})); + +server.listen(0, common.mustCall(() => { + const port = server.address().port; + + client = http2.connect(`http://${common.localhostIPv4}:${port}`, { + createConnection: () => new UnreadClientSocket(port), + }); + + client.on('connect', common.mustCall(() => { + for (let i = 0; i < TOTAL_REQUESTS; i++) { + const req = client.request(); + req.end(); + } + })); +})); diff --git a/test/parallel/test-stream-pipeline-http2.js b/test/parallel/test-stream-pipeline-http2.js index 8ffee7786838..0b26c95b3a4e 100644 --- a/test/parallel/test-stream-pipeline-http2.js +++ b/test/parallel/test-stream-pipeline-http2.js @@ -30,8 +30,7 @@ const http2 = require('http2'); let received = 0; req.on('data', (data) => { received += data.length; - // Bound the data that flows before teardown - bytes per data event vary - // by platform, and letting this run longer hangs on macOS. + // Use a byte threshold because data event chunking varies by platform. if (received >= 32 * 1024) rs.destroy(); }); })); From a46d64f055e3d06a223f93689395157158e3e478 Mon Sep 17 00:00:00 2001 From: Tim Perry Date: Thu, 20 Aug 2026 18:15:58 +0100 Subject: [PATCH 2/3] Cleanup unused code - no longer reachable without pausing --- src/node_http2.cc | 81 +++++++---------------------------------------- src/node_http2.h | 7 ++-- 2 files changed, 14 insertions(+), 74 deletions(-) diff --git a/src/node_http2.cc b/src/node_http2.cc index 45c4ed1530f9..4ce35dc37c23 100644 --- a/src/node_http2.cc +++ b/src/node_http2.cc @@ -981,45 +981,24 @@ ssize_t Http2Session::OnMaxFrameSizePadding(size_t frameLen, // quite expensive. This is a potential performance optimization target later. void Http2Session::ConsumeHTTP2Data() { CHECK_NOT_NULL(stream_buf_.base); - CHECK_LE(stream_buf_offset_, stream_buf_.len); - size_t read_len = stream_buf_.len - stream_buf_offset_; // multiple side effects. - Debug(this, "receiving %d bytes [wants data? %d]", - read_len, + Debug(this, + "receiving %d bytes [wants data? %d]", + stream_buf_.len, nghttp2_session_want_read(session_.get())); - set_receive_paused(false); custom_recv_error_code_ = nullptr; set_receiving(); ssize_t ret = - nghttp2_session_mem_recv(session_.get(), - reinterpret_cast(stream_buf_.base) + - stream_buf_offset_, - read_len); + nghttp2_session_mem_recv(session_.get(), + reinterpret_cast(stream_buf_.base), + stream_buf_.len); set_receiving(false); CHECK_NE(ret, NGHTTP2_ERR_NOMEM); CHECK_IMPLIES(custom_recv_error_code_ != nullptr, ret < 0); - if (is_receive_paused()) { - CHECK(is_reading_stopped()); - - CHECK_GT(ret, 0); - CHECK_LE(static_cast(ret), read_len); - - // Mark the remainder of the data as available for later consumption. - // Even if all bytes were received, a paused stream may delay the - // nghttp2_on_frame_recv_callback which may have an END_STREAM flag. - stream_buf_offset_ += ret; - // Still complete a Close() deferred during mem_recv; do not fall through - // to SendPendingData() here (paused receives historically skip that flush - // because a write may already be in progress). - MaybeFinishPendingClose(); - goto done; - } - // We are done processing the current input chunk. DecrementCurrentSessionMemory(stream_buf_.len); - stream_buf_offset_ = 0; stream_buf_ab_.Reset(); stream_buf_allocation_.reset(); stream_buf_ = uv_buf_init(nullptr, 0); @@ -1028,14 +1007,6 @@ void Http2Session::ConsumeHTTP2Data() { // not written after pending RST_STREAM frames. MaybeFinishPendingClose(); -done: - // Finish a Close() deferred above before flushing, so GOAWAY is not written - // after pending RST_STREAM frames. - if (is_close_pending() && !is_destroyed()) { - set_close_pending(false); - FinishClose(pending_close_code_, pending_close_socket_closed_); - } - // Send any data that was queued up while processing the received data. if (ret >= 0 && !is_destroyed()) { SendPendingData(); @@ -1567,7 +1538,6 @@ void Http2StreamListener::OnStreamRead(ssize_t nread, const uv_buf_t& buf) { size_t offset = buf.base - session->stream_buf_.base; // Verify that the data offset is inside the current read buffer. - CHECK_GE(offset, session->stream_buf_offset_); CHECK_LE(offset, session->stream_buf_.len); CHECK_LE(offset + buf.len, session->stream_buf_.len); @@ -1882,11 +1852,6 @@ void Http2Session::OnStreamAfterWrite(WriteWrap* w, int status) { return; } - // If there is more incoming data queued up, consume it. - if (stream_buf_offset_ > 0) { - ConsumeHTTP2Data(); - } - if (!is_write_scheduled() && !is_destroyed()) { // Schedule a new write if nghttp2 wants to send data. MaybeScheduleWrite(); @@ -2198,7 +2163,7 @@ void Http2Session::OnStreamRead(ssize_t nread, const uv_buf_t& buf_) { Context::Scope context_scope(env()->context()); Http2Scope h2scope(this); CHECK_NOT_NULL(stream_); - Debug(this, "receiving %d bytes, offset %d", nread, stream_buf_offset_); + Debug(this, "receiving %d bytes", nread); std::unique_ptr bs = env()->release_managed_buffer(buf_); // Only pass data on if nread > 0 @@ -2213,8 +2178,11 @@ void Http2Session::OnStreamRead(ssize_t nread, const uv_buf_t& buf_) { statistics_.data_received += nread; - if (stream_buf_offset_ == 0 && static_cast(nread) != bs->ByteLength()) - [[likely]] { + // ConsumeHTTP2Data() always consumes the whole chunk, so there is never a + // partially processed buffer left over from a previous read. + DCHECK_NULL(stream_buf_.base); + + if (static_cast(nread) != bs->ByteLength()) [[likely]] { // Shrink to the actual amount of used data. std::unique_ptr old_bs = std::move(bs); bs = ArrayBuffer::NewBackingStore( @@ -2222,31 +2190,6 @@ void Http2Session::OnStreamRead(ssize_t nread, const uv_buf_t& buf_) { nread, BackingStoreInitializationMode::kUninitialized); memcpy(bs->Data(), old_bs->Data(), nread); - } else { - // This is a very unlikely case, and should only happen if the ReadStart() - // call in OnStreamAfterWrite() immediately provides data. If that does - // happen, we concatenate the data we received with the already-stored - // pending input data, slicing off the already processed part. - size_t pending_len = stream_buf_.len - stream_buf_offset_; - std::unique_ptr new_bs = ArrayBuffer::NewBackingStore( - env()->isolate(), - pending_len + nread, - BackingStoreInitializationMode::kUninitialized); - memcpy(static_cast(new_bs->Data()), - stream_buf_.base + stream_buf_offset_, - pending_len); - memcpy(static_cast(new_bs->Data()) + pending_len, - bs->Data(), - nread); - - bs = std::move(new_bs); - nread = bs->ByteLength(); - stream_buf_offset_ = 0; - stream_buf_ab_.Reset(); - - // We have now fully processed the stream_buf_ input chunk (by moving the - // remaining part into buf, which will be accounted for below). - DecrementCurrentSessionMemory(stream_buf_.len); } IncrementCurrentSessionMemory(nread); diff --git a/src/node_http2.h b/src/node_http2.h index b81862b27bce..f3edcf658f80 100644 --- a/src/node_http2.h +++ b/src/node_http2.h @@ -85,9 +85,8 @@ constexpr int kSessionStateClosing = 0x8; constexpr int kSessionStateSending = 0x10; constexpr int kSessionStateWriteInProgress = 0x20; constexpr int kSessionStateReadingStopped = 0x40; -constexpr int kSessionStateReceivePaused = 0x80; -constexpr int kSessionStateReceiving = 0x100; -constexpr int kSessionStateClosePending = 0x200; +constexpr int kSessionStateReceiving = 0x80; +constexpr int kSessionStateClosePending = 0x100; // The Padding Strategy determines the method by which extra padding is // selected for HEADERS and DATA frames. These are configurable via the @@ -698,7 +697,6 @@ class Http2Session : public AsyncWrap, IS_FLAG(sending, kSessionStateSending) IS_FLAG(write_in_progress, kSessionStateWriteInProgress) IS_FLAG(reading_stopped, kSessionStateReadingStopped) - IS_FLAG(receive_paused, kSessionStateReceivePaused) IS_FLAG(receiving, kSessionStateReceiving) IS_FLAG(close_pending, kSessionStateClosePending) @@ -979,7 +977,6 @@ class Http2Session : public AsyncWrap, // will be set. stream_buf_ab_ is lazily created from stream_buf_allocation_. v8::Global stream_buf_ab_; std::unique_ptr stream_buf_allocation_; - size_t stream_buf_offset_ = 0; // Custom error code for errors that originated inside one of the callbacks // called by nghttp2_session_mem_recv. const char* custom_recv_error_code_ = nullptr; From 364edc00758ce5b0b50ed643f3e2d3d893840329 Mon Sep 17 00:00:00 2001 From: Tim Perry Date: Fri, 21 Aug 2026 14:42:03 +0100 Subject: [PATCH 3/3] Make security test reliable across platforms This rewrites it to focus on the key case: single stream backpressure is still applied correctly when writes block even if the remote peer H2 window allows more data. The previous case covered multiple streams which is protected by maxSessionMemory, but kernel buffering makes the behaviour variable on other platforms (not breaking the security guarantees, but failing the test) and this isn't the clearest representation of the key issue we need to guard against (CVE-2019-9517). --- .../test-http2-session-memory-unread-peer.js | 100 ------------- ...t-http2-stream-backpressure-unread-peer.js | 133 ++++++++++++++++++ 2 files changed, 133 insertions(+), 100 deletions(-) delete mode 100644 test/parallel/test-http2-session-memory-unread-peer.js create mode 100644 test/parallel/test-http2-stream-backpressure-unread-peer.js diff --git a/test/parallel/test-http2-session-memory-unread-peer.js b/test/parallel/test-http2-session-memory-unread-peer.js deleted file mode 100644 index 7f8358824eda..000000000000 --- a/test/parallel/test-http2-session-memory-unread-peer.js +++ /dev/null @@ -1,100 +0,0 @@ -'use strict'; - -// CVE-2019-9517 describes a peer that advertises HTTP/2 flow-control credit -// but does not drain its TCP socket, leaving the server holding responses it -// cannot write out. -// -// Node bounds this with maxSessionMemory. Once the budget is spent, further -// streams are refused rather than queued, so a client cannot make the server -// buffer a response for every request it opens. - -const common = require('../common'); -if (!common.hasCrypto) - common.skip('missing crypto'); -const assert = require('assert'); -const http2 = require('http2'); -const net = require('net'); -const { Duplex } = require('stream'); - -const TOTAL_REQUESTS = 200; -// Accepted count depends on kernel buffer sizes but this seems to -// cover most likely scenarios: -const MAX_ACCEPTED = 20; -const body = Buffer.alloc(1_000_000); - -let accepted = 0; -let client; - -// Client-side socket that forwards writes but never consumes incoming data. -class UnreadClientSocket extends Duplex { - constructor(port) { - super(); - this.inner = net.connect(port, common.localhostIPv4); - this.inner.pause(); - } - _read() { - // Deliberately never pull from the underlying socket. - } - _write(chunk, encoding, callback) { - this.inner.write(chunk, encoding, callback); - } - _final(callback) { - callback(); - } - _destroy(err, callback) { - this.inner.destroy(); - callback(err); - } -} - -const onStreamsRefused = common.mustCall(() => { - assert.ok(accepted >= 1, 'client never reached the server'); - client.destroy(); - server.close(); -}); - -const server = http2.createServer({ - maxSessionMemory: 1, - // Turn the rejected streams into an observable server-session failure. - maxSessionRejectedStreams: 0, -}); - -server.on('session', common.mustCall((session) => { - session.on('error', common.mustCall((err) => { - assert.strictEqual(err.code, 'ERR_HTTP2_ERROR'); - onStreamsRefused(); - })); -})); - -// At least one stream must be accepted, otherwise the assertion above could -// pass without the client ever having reached the server. -server.on('stream', common.mustCallAtLeast((stream) => { - accepted++; - // Assert here rather than once the session fails: if the bound does not - // hold there may be no session failure at all. - assert.ok(accepted <= MAX_ACCEPTED, - `server accepted ${accepted} of ${TOTAL_REQUESTS} streams ` + - 'from a peer that never reads; maxSessionMemory did not ' + - 'bound its outbound buffering'); - // The fatal session error also destroys every accepted stream. - stream.on('error', common.mustCall((err) => { - assert.strictEqual(err.code, 'ERR_HTTP2_ERROR'); - })); - stream.respond(); - stream.end(body); -})); - -server.listen(0, common.mustCall(() => { - const port = server.address().port; - - client = http2.connect(`http://${common.localhostIPv4}:${port}`, { - createConnection: () => new UnreadClientSocket(port), - }); - - client.on('connect', common.mustCall(() => { - for (let i = 0; i < TOTAL_REQUESTS; i++) { - const req = client.request(); - req.end(); - } - })); -})); diff --git a/test/parallel/test-http2-stream-backpressure-unread-peer.js b/test/parallel/test-http2-stream-backpressure-unread-peer.js new file mode 100644 index 000000000000..9667c1611b00 --- /dev/null +++ b/test/parallel/test-http2-stream-backpressure-unread-peer.js @@ -0,0 +1,133 @@ +'use strict'; + +// CVE-2019-9517 describes a peer that offers HTTP/2 flow-control credit +// but then doesn't read from its TCP connection (blocking TCP flow). +// A streaming response must stop producing data when the resulting server +// write becomes blocked by the transport. + +// We use a raw client here to directly control the flow of requests and +// stall responses to deterministically reproduce this on all platforms. + +const common = require('../common'); +if (!common.hasCrypto) + common.skip('missing crypto'); +const assert = require('assert'); +const http2 = require('http2'); +const { Duplex, Readable } = require('stream'); + +const { MAX_INITIAL_WINDOW_SIZE } = http2.constants; + +const FRAME_HEADERS = 1; +const FRAME_SETTINGS = 4; +const FRAME_WINDOW_UPDATE = 8; +const FLAG_END_STREAM = 1; +const FLAG_END_HEADERS = 4; +const INITIAL_CONNECTION_WINDOW_SIZE = 65_535; + +const preface = Buffer.from('PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n'); + +function frame(type, flags, streamID, payload = Buffer.alloc(0)) { + const header = Buffer.alloc(9); + header.writeUIntBE(payload.length, 0, 3); + header[3] = type; + header[4] = flags; + header.writeUInt32BE(streamID, 5); + return Buffer.concat([header, payload]); +} + +function settings() { + const payload = http2.getPackedSettings({ + initialWindowSize: MAX_INITIAL_WINDOW_SIZE, + }); + return frame(FRAME_SETTINGS, 0, 0, payload); +} + +function connectionWindowUpdate() { + const payload = Buffer.alloc(4); + payload.writeUInt32BE( + MAX_INITIAL_WINDOW_SIZE - INITIAL_CONNECTION_WINDOW_SIZE, 0); + return frame(FRAME_WINDOW_UPDATE, 0, 0, payload); +} + +function request() { + return frame( + FRAME_HEADERS, + FLAG_END_HEADERS | FLAG_END_STREAM, + 1, + Buffer.from([ + 0x82, // :method: GET + 0x86, // :scheme: http + 0x84, // :path: / + 0x41, 0x01, 0x78, // :authority: x + ])); +} + +let onWriteStalled; +const socket = new Duplex({ + read() {}, + write(_data, _encoding, callback) { + if (onWriteStalled === undefined) { + callback(); + return; + } + + // Leave the write pending to model a peer that has stopped reading. + onWriteStalled(); + }, +}); + +// The 1MB body is well within the client's stream and connection windows +// (2^31-1), so HTTP/2 flow control will not limit the response. +const CHUNK_SIZE = 16 * 1024; +const RESPONSE_SIZE = 1024 * 1024; +const chunk = Buffer.alloc(CHUNK_SIZE); + +const serverSession = http2.performServerHandshake(socket); +serverSession.on('stream', common.mustCall((stream) => { + let produced = 0; + + // A datasource that produces chunks on demand + const source = new Readable({ + highWaterMark: CHUNK_SIZE, + read() { + if (produced === RESPONSE_SIZE) { + this.push(null); + return; + } + produced += CHUNK_SIZE; + this.push(chunk); + }, + }); + + // We set this once the stream opens, to stall all future writes once the + // initial preface & SETTINGS dance is completed. + onWriteStalled = common.mustCall(() => { + setImmediate(common.mustCall(() => { + // Backpressure should have been applied to the source when the + // socket writable writes stalled, even though there is still + // H2 flow-control credit available: + assert(source.isPaused()); + assert(stream.writableNeedDrain); + + // Four chunks fill the stream's 64KB writable buffer and one more is + // prefetched into the source's 16KB readable buffer. Only 80KB of the + // 1MB source is produced before TCP backpressure stops buffering. + assert.strictEqual(produced, CHUNK_SIZE * 5); + + serverSession.destroy(); + socket.destroy(); + source.destroy(); + })); + }); + + stream.respond(); + source.pipe(stream); +})); + +// Manually send a single request with all required preamble: +socket.push(Buffer.concat([ + preface, + settings(), + connectionWindowUpdate(), + request(), +]));