From cb2e8ddeb5cf17c7dc81da24a259fe971a34bb56 Mon Sep 17 00:00:00 2001 From: Marten Richter Date: Sat, 15 Aug 2026 09:29:07 +0200 Subject: [PATCH] quic: update writedesiredsize on maxdataframe This commit will react on a callback from ngtcp2, when a maxdata frame arrives and update the scheduled streams accordingly. A PR to get this callback is pending. Fixes: #64835 Signed-off-by: Marten Richter --- src/quic/application.cc | 11 ++- src/quic/application.h | 8 ++ src/quic/http3.cc | 10 +++ src/quic/session.cc | 14 ++++ ...uic-h3-maxdata-external-buffer-failure.mjs | 76 +++++++++++++++++++ 5 files changed, 118 insertions(+), 1 deletion(-) create mode 100644 test/parallel/test-quic-h3-maxdata-external-buffer-failure.mjs diff --git a/src/quic/application.cc b/src/quic/application.cc index 79a3263b8537..69ee401927bf 100644 --- a/src/quic/application.cc +++ b/src/quic/application.cc @@ -419,7 +419,7 @@ class DefaultApplication final : public Session::Application { void BlockStream(stream_id id) override { if (auto stream = session().FindStream(id)) [[likely]] { // Remove the stream from the send queue. It will be re-scheduled - // via ExtendMaxStreamData when the peer grants more flow control. + // via ExtendMax(Stream)Data when the peer grants more flow control. // Without this, SendPendingData would repeatedly pop and retry // the same blocked stream in an infinite loop. stream->Unschedule(); @@ -434,6 +434,15 @@ class DefaultApplication final : public Session::Application { stream->Schedule(&stream_queue_); } + void ExtendMaxData(uint64_t max_data) override { + // The peer granted more flow control for session. Re-schedule + // all streams so SendPendingData will resume writing. + for (auto& [id, stream] : session().streams()) { + stream->Schedule(&stream_queue_); + } + } + + bool StreamCommit(Session::StreamData* stream_data, size_t datalen) override { DCHECK_NOT_NULL(stream_data); CHECK(stream_data->stream); diff --git a/src/quic/application.h b/src/quic/application.h index 619b41dd8d0b..a6904556b926 100644 --- a/src/quic/application.h +++ b/src/quic/application.h @@ -149,6 +149,14 @@ class Session::Application : public MemoryRetainer { // By default do nothing. } + // Called when the Session determines that the flow control window for the + // session has been expanded. Not all Application types will require + // this notification so the default is to do nothing. + virtual void ExtendMaxData(uint64_t max_data) { + Debug(session_, "Application extending max data"); + // By default do nothing. + } + // Different Applications may wish to set some application data in the // session ticket (e.g. http/3 would set server settings in the application // data). The first byte written MUST be the Application::Type enum value. diff --git a/src/quic/http3.cc b/src/quic/http3.cc index b6d876af60f6..07a4d2e55cf4 100644 --- a/src/quic/http3.cc +++ b/src/quic/http3.cc @@ -378,6 +378,16 @@ class Http3ApplicationImpl final : public Session::Application { nghttp3_conn_unblock_stream(*this, stream->id()); } + void ExtendMaxData(uint64_t max_data) override { + Debug(&session(), + "HTTP/3 application extending max data to %" PRIu64, + max_data); + for (auto& [id, stream] : session().streams()) { + stream->UpdateWriteDesiredSize(); // the stream might be blocked on js side + // is unblock stream also required? + } + } + void CollectSessionTicketAppData( SessionTicket::AppData* app_data) const override { uint8_t buf[kSessionTicketAppDataSize]; diff --git a/src/quic/session.cc b/src/quic/session.cc index 1bea15fbadb4..76eb7afc9ce4 100644 --- a/src/quic/session.cc +++ b/src/quic/session.cc @@ -1358,6 +1358,14 @@ struct Session::Impl final : public MemoryRetainer { return NGTCP2_SUCCESS; } + static int on_extend_max_data(ngtcp2_conn* conn, + uint64_t max_data, + void* user_data) { + NGTCP2_CALLBACK_SCOPE(session) + session->application().ExtendMaxData(max_data); + return NGTCP2_SUCCESS; + } + static int on_get_new_cid(ngtcp2_conn* conn, ngtcp2_cid* cid, ngtcp2_stateless_reset_token* token, @@ -1700,6 +1708,9 @@ struct Session::Impl final : public MemoryRetainer { on_receive_stream_stop_sending, #ifdef NGTCP2_CALLBACKS_V5 nullptr, +#ifdef NGTCP2_CALLBACKS_V6 + on_extend_max_data, +#endif #endif // NGTCP2_CALLBACKS_V5 #endif // NGTCP2_CALLBACKS_V4 }; @@ -1754,6 +1765,9 @@ struct Session::Impl final : public MemoryRetainer { on_receive_stream_stop_sending, #ifdef NGTCP2_CALLBACKS_V5 nullptr, +#ifdef NGTCP2_CALLBACKS_V6 + on_extend_max_data, +#endif #endif // NGTCP2_CALLBACKS_V5 #endif // NGTCP2_CALLBACKS_V4 }; diff --git a/test/parallel/test-quic-h3-maxdata-external-buffer-failure.mjs b/test/parallel/test-quic-h3-maxdata-external-buffer-failure.mjs new file mode 100644 index 000000000000..e93a2cf173a8 --- /dev/null +++ b/test/parallel/test-quic-h3-maxdata-external-buffer-failure.mjs @@ -0,0 +1,76 @@ +// Flags: --experimental-quic --experimental-stream-iter --no-warnings + +// Test: Quic maxdata updates on http/3 +// Client sends a body that precisely fills the session window size, +// and verifies that it is data transfer is not stalled. + +import { hasQuic, skip } from '../common/index.mjs'; +import { readFile } from 'node:fs/promises'; +import { setTimeout as sleep } from 'node:timers/promises'; + +if (!hasQuic) { + skip('QUIC is not enabled'); +} +const { listen, connect } = await import('node:quic'); +const { createPrivateKey } = await import('node:crypto'); +const { drainableProtocol } = await import('stream/iter'); + +const keys = 'test/fixtures/keys'; +const key = createPrivateKey(await readFile(`${keys}/agent1-key.pem`)); +const cert = await readFile(`${keys}/agent1-cert.pem`); + +const WINDOW = 4096; +// Fills the window exactly: +// considers all framing including some initial session capsules +const BODY = WINDOW - 38; + +let letServerRead; +const serverMayRead = new Promise((resolve) => { letServerRead = resolve; }); + +const endpoint = await listen((session) => { + session.onstream = async (stream) => { + await serverMayRead; + // eslint-disable-next-line no-unused-vars + for await (const _ of stream) { /* reading extends the window */ } + }; +}, { + sni: { '*': { keys: [key], certs: [cert] } }, + transportParams: { + initialMaxStreamDataBidiRemote: 1024 * 1024, // Make sure maxstreamdata does not block + initialMaxData: WINDOW, + }, + onheaders() { this.sendHeaders({ ':status': '200' }); }, +}); + +const session = await connect(endpoint.address, { + servername: 'localhost', + verifyPeer: 'manual', +}); +await session.opened; + +// Budget well above the window, so the window is what stops the writer. +const stream = await session.createBidirectionalStream({ budget: 1024 * 1024 }); +stream.sendHeaders({ + ':method': 'POST', + ':path': '/', + ':scheme': 'https', + ':authority': 'localhost', +}, { terminal: false }); + +const writer = stream.writer; +writer.writeSync(new Uint8Array(BODY)); + +// Long enough for every byte to be acked. The peer acks as data arrives, +// whether or not its application has read any of it, so by now the window is +// exhausted, the send buffer is empty, and no further ACK can arrive. +await sleep(500); + +const watchdog = setTimeout(() => { + console.error('STALLED: no drain after MAX_STREAM_DATA'); + process.exit(1); +}, 5000); +letServerRead(); // Extend the window, with no ack attached +await writer[drainableProtocol](); + +clearTimeout(watchdog); +process.exit(0);