From aa18bf92ab70821223427cccf85b8b55c0cf3337 Mon Sep 17 00:00:00 2001 From: Marten Richter Date: Sun, 16 Aug 2026 20:11:11 +0200 Subject: [PATCH] quic: add option waitUntilAvailable to creating streams We add an option to createBidirectionalStream and createUnidirectionalStream to fail immediately, if the flow control's stream budget does not allow stream creation. The behavior matches W3C webtransport's behavior. Fixes #65321 Signed-off-by: Marten Richter --- doc/api/quic.md | 10 ++++ lib/internal/quic/quic.js | 3 +- src/quic/session.cc | 24 ++++++-- src/quic/session.h | 3 + .../test-quic-stream-limits-pending.mjs | 57 ++++++++++++++----- 5 files changed, 77 insertions(+), 20 deletions(-) diff --git a/doc/api/quic.md b/doc/api/quic.md index 7a52e4f72c1f..50d14617eb3e 100644 --- a/doc/api/quic.md +++ b/doc/api/quic.md @@ -1300,6 +1300,11 @@ added: v23.8.0 will buffer before `writeSync()` returns `false`. When the buffered data exceeds this limit, the caller should wait for drain before writing more. **Default:** `65536` (64 KB). + * `waitUntilAvailable` {boolean} When true the promise will wait until flow + control will allow to open the stream. If set to false, the function + will fail synchronously, if flow control will not allow to open the stream + immediately. + **Default:** `false` * `onheaders` {Function} Callback for received initial response headers. Called with `(headers)`. * `ontrailers` {Function} Callback for received trailing headers. @@ -1341,6 +1346,11 @@ added: v23.8.0 will buffer before `writeSync()` returns `false`. When the buffered data exceeds this limit, the caller should wait for drain before writing more. **Default:** `65536` (64 KB). + * `waitUntilAvailable` {boolean} When true the promise will wait until flow + control will allow to open the stream. If set to false, the function + will fail synchronously, if flow control will not allow to open the stream + immediately. + **Default:** `false` * `onheaders` {Function} Callback for received initial response headers. Called with `(headers)`. * `ontrailers` {Function} Callback for received trailing headers. diff --git a/lib/internal/quic/quic.js b/lib/internal/quic/quic.js index f645998e628d..bf128fbd8532 100644 --- a/lib/internal/quic/quic.js +++ b/lib/internal/quic/quic.js @@ -3347,6 +3347,7 @@ class QuicSession { incremental = false, budget = kDefaultBudget, headers, + waitUntilAvailable = false, onheaders, ontrailers, oninfo, @@ -3358,7 +3359,7 @@ class QuicSession { const validatedBody = validateBody(body); - const handle = this.#handle.openStream(direction, validatedBody); + const handle = this.#handle.openStream(direction, waitUntilAvailable, validatedBody); if (handle === undefined) { throw new ERR_QUIC_OPEN_STREAM_FAILED(); } diff --git a/src/quic/session.cc b/src/quic/session.cc index 1bea15fbadb4..6e2e1da78cd4 100644 --- a/src/quic/session.cc +++ b/src/quic/session.cc @@ -1167,17 +1167,24 @@ struct Session::Impl final : public MemoryRetainer { } DCHECK(args[0]->IsUint32()); + DCHECK(args[1]->IsBoolean()); + + auto direction = FromV8Value(args[0]); + if (!args[1].As()->Value() && false) { // This is waitUntilAvailable + if (!session->CanImmediatelyOpenStream(direction)) { + return THROW_ERR_INVALID_STATE(env, "No new stream available within flow control"); + } + } // GetDataQueueFromSource handles type validation. std::shared_ptr data_source; - if (!Stream::GetDataQueueFromSource(env, args[1]).To(&data_source)) + if (!Stream::GetDataQueueFromSource(env, args[2]).To(&data_source)) [[unlikely]] { return THROW_ERR_INVALID_ARG_VALUE(env, "Invalid data source"); } session->impl_->handshake_deferred_ = false; SendPendingDataScope send_scope(session); - auto direction = FromV8Value(args[0]); Local stream; if (session->OpenStream(direction, std::move(data_source)).ToLocal(&stream)) [[likely]] { @@ -3166,6 +3173,14 @@ BaseObjectPtr Session::CreateStream( return {}; } +bool Session::CanImmediatelyOpenStream(Direction direction) { + if (direction == Direction::BIDIRECTIONAL) { + return max_local_streams_bidi() > 0; + } else { + return max_local_streams_uni() > 0; + } +} + MaybeLocal Session::OpenStream(Direction direction, std::shared_ptr data_source) { // If can_create_streams() returns false, we are not able to open a stream @@ -3507,13 +3522,12 @@ void Session::SetApplicationError(error_code app_error_code) { uint64_t Session::max_local_streams_uni() const { DCHECK(!is_destroyed()); - return ngtcp2_conn_get_streams_uni_left(*this); + return ngtcp2_conn_get_streams_uni_left2(*this); } uint64_t Session::max_local_streams_bidi() const { DCHECK(!is_destroyed()); - return ngtcp2_conn_get_local_transport_params(*this) - ->initial_max_streams_bidi; + return ngtcp2_conn_get_streams_bidi_left2(*this); } void Session::set_wrapped() { diff --git a/src/quic/session.h b/src/quic/session.h index 3b2a9773a380..591a8cd007fb 100644 --- a/src/quic/session.h +++ b/src/quic/session.h @@ -522,6 +522,9 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source { size_t max_packet_size() const; void set_priority_supported(bool on = true); + // Check whether flow control permits opening another stream + bool CanImmediatelyOpenStream(Direction direction); + // Open a new locally-initialized stream with the specified directionality. // If the session is not yet in a state where the stream can be openen -- // such as when the handshake is not yet sufficiently far along and ORTT diff --git a/test/parallel/test-quic-stream-limits-pending.mjs b/test/parallel/test-quic-stream-limits-pending.mjs index bd7bcd51fff8..672ab92c0637 100644 --- a/test/parallel/test-quic-stream-limits-pending.mjs +++ b/test/parallel/test-quic-stream-limits-pending.mjs @@ -15,22 +15,30 @@ if (!hasQuic) { const { listen, connect } = await import('../common/quic.mjs'); const { bytes } = await import('stream/iter'); +const { setTimeout: sleep } = await import('timers/promises'); const encoder = new TextEncoder(); const allDone = Promise.withResolvers(); +const twoDone = Promise.withResolvers(); let serverStreamCount = 0; // Server allows only 1 bidi stream at a time. const serverEndpoint = await listen(mustCall((serverSession) => { serverSession.onstream = mustCall(async (stream) => { - await bytes(stream); + const streambytes = await bytes(stream); stream.writer.endSync(); await stream.closed; - if (++serverStreamCount === 2) { - serverSession.close(); + ++serverStreamCount; + if (serverStreamCount === 2) { + twoDone.resolve(); + } + if (serverStreamCount === 3) { allDone.resolve(); } - }, 2); + if (serverStreamCount === 4) { + serverSession.close(); + } + }, 4); }), { transportParams: { initialMaxStreamsBidi: 1 }, }); @@ -41,29 +49,50 @@ await clientSession.opened; // First stream opens immediately (within the limit). const s1 = await clientSession.createBidirectionalStream({ body: encoder.encode('stream 1'), + waitUntilAvailable: true }); -// Second stream is created but queued as pending because the +try { + // Second stream should not open, but throw. + const s2 = await clientSession.createBidirectionalStream({ + body: encoder.encode('stream 2'), + waitUntilAvailable: false + }); +} catch (error) { + assert.strictEqual(error.code, 'ERR_INVALID_STATE'); +} + +// Third stream is created but queued as pending because the // server only allows 1 concurrent bidi stream. -const s2 = await clientSession.createBidirectionalStream({ - body: encoder.encode('stream 2'), +const s3 = await clientSession.createBidirectionalStream({ + body: encoder.encode('stream 3'), + waitUntilAvailable: true }); -// s2 should be pending until s1 closes and the server grants +// s3 should be pending until s1 closes and the server grants // more stream credits. -assert.strictEqual(s2.pending, true); +assert.strictEqual(s3.pending, true); // Drain and close the first stream. for await (const _ of s1) { /* drain */ } // eslint-disable-line no-unused-vars await s1.closed; -// After s1 closes, the server sends MAX_STREAMS which opens s2. +// After s1 closes, the server sends MAX_STREAMS which opens s3. // Wait for the server to receive both streams. -await allDone.promise; +await twoDone.promise; +// s3 should no longer be pending. +for await (const _ of s3) { /* drain */ } // eslint-disable-line no-unused-vars +await s3.closed; -// s2 should no longer be pending. -for await (const _ of s2) { /* drain */ } // eslint-disable-line no-unused-vars -await s2.closed; +await sleep(10); // we wait a bit, as we do not have a callback exposed to js +// fourth stream should open immediately and not throw +console.log('before last stream') +const s4 = await clientSession.createBidirectionalStream({ + body: encoder.encode('stream 4'), + waitUntilAvailable: false +}); +await s4.closed; +await allDone.promise; await clientSession.close(); await serverEndpoint.close();