From 28daf5924f60c14110e12a412ddde009191338d2 Mon Sep 17 00:00:00 2001 From: Tim Perry Date: Wed, 9 Sep 2026 16:14:34 +0200 Subject: [PATCH] quic: improve stream cleanup & lookup Signed-off-by: Tim Perry --- src/quic/application.cc | 5 +- src/quic/application.h | 16 +- src/quic/defs.h | 5 - src/quic/http3.cc | 154 +++++++------------ src/quic/session.cc | 35 ++--- src/quic/session.h | 4 +- test/parallel/test-quic-h3-stream-credit.mjs | 87 +++++++++++ 7 files changed, 172 insertions(+), 134 deletions(-) create mode 100644 test/parallel/test-quic-h3-stream-credit.mjs diff --git a/src/quic/application.cc b/src/quic/application.cc index 1e78b7931ce1..9116b138ee64 100644 --- a/src/quic/application.cc +++ b/src/quic/application.cc @@ -207,9 +207,10 @@ Session::Application::ExtractSessionTicketAppData( : SessionTicket::AppData::Status::TICKET_USE; } -void Session::Application::ReceiveStreamClose(Stream* stream, +void Session::Application::ReceiveStreamClose(stream_id id, + Stream* stream, QuicError&& error) { - DCHECK_NOT_NULL(stream); + if (stream == nullptr) return; stream->Destroy(std::move(error)); } diff --git a/src/quic/application.h b/src/quic/application.h index 0ae4ab53ab7e..4fd748763742 100644 --- a/src/quic/application.h +++ b/src/quic/application.h @@ -126,14 +126,6 @@ class Session::Application : public MemoryRetainer { // to send for the given stream. virtual void ResumeStream(stream_id id) {} - // Called when the Session determines that the maximum number of - // remotely-initiated unidirectional streams has been extended. Not all - // Application types will require this notification so the default is to do - // nothing. - virtual void ExtendMaxStreams(EndpointLabel label, - Direction direction, - uint64_t max_streams) {} - // Returns true if the application manages stream FIN internally (e.g., // HTTP/3 uses nghttp3 which sends FIN via the fin flag in writev_stream). // When true, the stream infrastructure must NOT call @@ -166,9 +158,15 @@ class Session::Application : public MemoryRetainer { SessionTicket::AppData::Source::Flag flag); // Notifies the Application that the identified stream has been closed. - virtual void ReceiveStreamClose(Stream* stream, + virtual void ReceiveStreamClose(stream_id id, + Stream* stream, QuicError&& error = QuicError()); + // Notifies the Application that the Stream for the identified stream has + // been removed from the session and may be freed immediately afterwards. + // Applications caching the Stream pointer must drop it here. + virtual void StreamRemoved(stream_id id) {} + // Notifies the Application that the identified stream has been reset. virtual void ReceiveStreamReset(Stream* stream, uint64_t final_size, diff --git a/src/quic/defs.h b/src/quic/defs.h index 32b3a10d7b80..45b4c77d1584 100644 --- a/src/quic/defs.h +++ b/src/quic/defs.h @@ -290,11 +290,6 @@ enum class Side : uint8_t { SERVER, }; -enum class EndpointLabel : uint8_t { - LOCAL, - REMOTE, -}; - enum class Direction : uint8_t { BIDIRECTIONAL, UNIDIRECTIONAL, diff --git a/src/quic/http3.cc b/src/quic/http3.cc index 43ead1b8164e..a1a7ce11a62b 100644 --- a/src/quic/http3.cc +++ b/src/quic/http3.cc @@ -373,22 +373,6 @@ class Http3ApplicationImpl final : public Session::Application { Application::ResumeStream(id); } - void ExtendMaxStreams(EndpointLabel label, - Direction direction, - uint64_t max_streams) override { - switch (label) { - case EndpointLabel::LOCAL: - return; - case EndpointLabel::REMOTE: { - Debug(&session(), - "HTTP/3 application extending max %s streams by %" PRIu64, - direction == Direction::BIDIRECTIONAL ? "bidi" : "uni", - max_streams); - session().ExtendMaxStreams(direction, max_streams); - } - } - } - void ExtendMaxStreamData(Stream* stream, uint64_t max_data) override { Debug(&session(), "HTTP/3 application extending max stream data to %" PRIu64, @@ -498,33 +482,26 @@ class Http3ApplicationImpl final : public Session::Application { : SessionTicket::AppData::Status::TICKET_USE; } - void ReceiveStreamClose(Stream* stream, + void ReceiveStreamClose(stream_id id, + Stream* stream, QuicError&& error = QuicError()) override { - Debug( - &session(), "HTTP/3 application closing stream %" PRIi64, stream->id()); - error_code code = NGHTTP3_H3_NO_ERROR; - if (error.type() == QuicError::Type::APPLICATION) { - code = error.code(); - } - - int rv = nghttp3_conn_close_stream2( - *this, - NGHTTP3_STREAM_CLOSE_FLAG_RX_APP_ERROR_CODE_SET, - stream->id(), - code, - 0); - // If the call is successful, Http3Application::OnStreamClose callback will - // be invoked when the stream is ready to be closed. We'll handle destroying - // the actual Stream object there. - if (rv == 0) return; - - if (rv == NGHTTP3_ERR_STREAM_NOT_FOUND) { - ExtendMaxStreams(EndpointLabel::REMOTE, stream->direction(), 1); - return; + Debug(&session(), "HTTP/3 application closing stream %" PRIi64, id); + + // Clean up nghttp3's state first. N.b. destroying the Stream calls into + // JS, so this can tear down the session. Skip unidirectional streams + // (control/QPACK) as nghttp3 handles this and would reject if we try. + if (conn_ && ngtcp2_is_bidi_stream(id)) { + int rv = nghttp3_conn_close_stream2( + *this, NGHTTP3_STREAM_CLOSE_FLAG_NONE, id, 0, 0); + if (rv != 0 && rv != NGHTTP3_ERR_STREAM_NOT_FOUND) { + session().SetApplicationError( + nghttp3_err_infer_quic_app_error_code(rv)); + session().Close(); + return; + } } - session().SetApplicationError(nghttp3_err_infer_quic_app_error_code(rv)); - session().Close(); + Application::ReceiveStreamClose(id, stream, std::move(error)); } void ReceiveStreamReset(Stream* stream, @@ -567,6 +544,10 @@ class Http3ApplicationImpl final : public Session::Application { return true; } + void StreamRemoved(stream_id id) override { + if (conn_) nghttp3_conn_set_stream_user_data(*this, id, nullptr); + } + bool SendHeaders(Stream& stream, HeadersKind kind, const Local& headers, @@ -910,37 +891,8 @@ class Http3ApplicationImpl final : public Session::Application { return Http3ConnectionPointer(conn); } - void OnStreamClose(Stream* stream, - uint32_t flags, - error_code rx_app_error_code, - error_code tx_app_error_code) { - if (flags & NGHTTP3_STREAM_CLOSE_FLAG_RX_APP_ERROR_CODE_SET) { - Debug(&session(), - "HTTP/3 application received stream close for stream %" PRIi64 - " with remote error code %" PRIu64, - stream->id(), - rx_app_error_code); - } - if (flags & NGHTTP3_STREAM_CLOSE_FLAG_TX_APP_ERROR_CODE_SET) { - Debug(&session(), - "HTTP/3 application send stream close for stream %" PRIi64 - " with error code %" PRIu64, - stream->id(), - tx_app_error_code); - } - auto direction = stream->direction(); - if (flags & NGHTTP3_STREAM_CLOSE_FLAG_RX_APP_ERROR_CODE_SET) { - stream->Destroy(QuicError::ForApplication(rx_app_error_code)); - } else if (flags & NGHTTP3_STREAM_CLOSE_FLAG_TX_APP_ERROR_CODE_SET) { - stream->Destroy(QuicError::ForApplication(tx_app_error_code)); - } else { - stream->Destroy(); - } - ExtendMaxStreams(EndpointLabel::REMOTE, direction, 1); - } - void OnBeginHeaders(stream_id id) { - auto stream = FindOrCreateStream(conn_.get(), &session(), id); + auto stream = FindOrCreateStream(id); if (!stream) [[unlikely]] return; Debug(&session(), @@ -994,7 +946,7 @@ class Http3ApplicationImpl final : public Session::Application { } void OnBeginTrailers(stream_id id) { - auto stream = FindOrCreateStream(conn_.get(), &session(), id); + auto stream = FindOrCreateStream(id); if (!stream) [[unlikely]] return; Debug(&session(), @@ -1186,20 +1138,26 @@ class Http3ApplicationImpl final : public Session::Application { return app; } - static BaseObjectWeakPtr FindOrCreateStream(nghttp3_conn* conn, - Session* session, - stream_id id) { - if (auto stream = session->FindStream(id)) { + // Cache the Stream* in nghttp3 so we can quickly get it later: + void BindStreamUserData(stream_id id, Stream* stream) { + if (conn_) nghttp3_conn_set_stream_user_data(*this, id, stream); + } + + BaseObjectWeakPtr FindOrCreateStream(stream_id id) { + if (auto stream = session().FindStream(id)) { + BindStreamUserData(id, stream.get()); return stream; } // No record of a locally-initiated stream means we already destroyed it, // and frames still in flight must not bring it back to life. See // DefaultApplication::ReceiveStreamData for the same guard on the raw // QUIC path. - if (!session->is_destroyed() && ngtcp2_conn_is_local_stream(*session, id)) { + if (!session().is_destroyed() && + ngtcp2_conn_is_local_stream(session(), id)) { return {}; } - if (auto stream = session->CreateStream(id)) { + if (auto stream = session().CreateStream(id)) { + if (!stream->is_destroyed()) BindStreamUserData(id, stream.get()); return stream; } return {}; @@ -1223,8 +1181,11 @@ class Http3ApplicationImpl final : public Session::Application { auto& app = *ptr; NgHttp3CallbackScope scope(&app.session()); - auto stream = app.session().FindStream(id); - if (!stream) return NGHTTP3_ERR_CALLBACK_FAILURE; + BaseObjectPtr stream(static_cast(stream_user_data)); + if (!stream) [[unlikely]] { + stream = app.session().FindStream(id); + if (!stream) return NGHTTP3_ERR_CALLBACK_FAILURE; + } if (stream->is_eos()) { *pflags |= NGHTTP3_DATA_FLAG_EOF; @@ -1305,23 +1266,12 @@ class Http3ApplicationImpl final : public Session::Application { auto ptr = From(conn, conn_user_data); CHECK_NOT_NULL(ptr); auto& app = *ptr; - if (auto stream = app.session().FindStream(id)) { - stream->Acknowledge(static_cast(datalen)); + BaseObjectPtr stream(static_cast(stream_user_data)); + if (!stream) [[unlikely]] { + stream = app.session().FindStream(id); } - return NGTCP2_SUCCESS; - } - - static int on_stream_close(nghttp3_conn* conn, - uint32_t flags, - stream_id id, - error_code rx_app_error_code, - error_code tx_app_error_code, - void* conn_user_data, - void* stream_user_data) { - NGHTTP3_CALLBACK_SCOPE(app); - if (auto stream = app.session().FindStream(id)) { - app.OnStreamClose( - stream.get(), flags, rx_app_error_code, tx_app_error_code); + if (stream) { + stream->Acknowledge(static_cast(datalen)); } return NGTCP2_SUCCESS; } @@ -1339,6 +1289,14 @@ class Http3ApplicationImpl final : public Session::Application { if (app.is_control_stream(id)) [[unlikely]] { return NGHTTP3_ERR_CALLBACK_FAILURE; } + // A cached Stream* is cleared before the Stream is removed from the + // session, non-null here means the stream is good to go. + if (auto* cached = static_cast(stream_user_data)) [[likely]] { + BaseObjectPtr stream(cached); + stream->ReceiveData(data, datalen, Stream::ReceiveDataFlags{}); + return NGTCP2_SUCCESS; + } + auto& session = app.session(); // DATA frames for a request stream the application already destroyed can @@ -1357,7 +1315,7 @@ class Http3ApplicationImpl final : public Session::Application { return NGTCP2_SUCCESS; } - if (auto stream = FindOrCreateStream(conn, &session, id)) [[likely]] { + if (auto stream = app.FindOrCreateStream(id)) { stream->ReceiveData(data, datalen, Stream::ReceiveDataFlags{}); return NGTCP2_SUCCESS; } @@ -1564,7 +1522,9 @@ class Http3ApplicationImpl final : public Session::Application { on_end_origin, on_rand, on_receive_settings, - on_stream_close}; + // We don't have to listen for stream_close - nghttp3 only closes when + // ReceiveStreamClose requests it, when we've already handled this. + nullptr}; }; std::unique_ptr CreateHttp3Application( diff --git a/src/quic/session.cc b/src/quic/session.cc index 1c159d763521..c6cd1734f9af 100644 --- a/src/quic/session.cc +++ b/src/quic/session.cc @@ -1579,13 +1579,15 @@ struct Session::Impl final : public MemoryRetainer { void* user_data, void* stream_user_data) { NGTCP2_CALLBACK_SCOPE(session) + // If the peer closes a stream, we return the credit to allow a new one: + session->ExtendMaxStreams(stream_id); + if (!session->has_application()) return NGTCP2_SUCCESS; auto* stream = Stream::From(stream_user_data); - if (stream == nullptr) return NGTCP2_SUCCESS; if (flags & NGTCP2_STREAM_CLOSE_FLAG_APP_ERROR_CODE_SET) { session->application().ReceiveStreamClose( - stream, QuicError::ForApplication(app_error_code)); + stream_id, stream, QuicError::ForApplication(app_error_code)); } else { - session->application().ReceiveStreamClose(stream); + session->application().ReceiveStreamClose(stream_id, stream); } return NGTCP2_SUCCESS; } @@ -3320,17 +3322,11 @@ void Session::AddStream(BaseObjectPtr stream, void Session::RemoveStream(stream_id id) { DCHECK(!is_destroyed()); Debug(this, "Removing stream %" PRIi64 " from session", id); - if (!is_in_draining_period() && !is_in_closing_period() && - !ngtcp2_conn_is_local_stream(*this, id)) { - if (ngtcp2_is_bidi_stream(id)) { - ngtcp2_conn_extend_max_streams_bidi(*this, 1); - } else { - ngtcp2_conn_extend_max_streams_uni(*this, 1); - } - } ngtcp2_conn_set_stream_user_data(*this, id, nullptr); + if (has_application()) application().StreamRemoved(id); + // Note that removing the stream from the streams map likely releases // the last BaseObjectPtr holding onto the Stream instance, at which // point it will be freed. If there are other BaseObjectPtr instances @@ -3479,14 +3475,15 @@ bool Session::OpenUnidirectionalStream(stream_id* id) { return ngtcp2_conn_open_uni_stream(*this, id, nullptr) == 0; } -void Session::ExtendMaxStreams(Direction direction, uint64_t max) { - switch (direction) { - case Direction::BIDIRECTIONAL: - ngtcp2_conn_extend_max_streams_bidi(*this, static_cast(max)); - break; - case Direction::UNIDIRECTIONAL: - ngtcp2_conn_extend_max_streams_uni(*this, static_cast(max)); - break; +void Session::ExtendMaxStreams(stream_id id) { + // MAX_STREAMS only limits what the peer opens, and there is nothing to + // grant once the connection is going away. + if (is_in_draining_period() || is_in_closing_period()) return; + if (ngtcp2_conn_is_local_stream(*this, id)) return; + if (ngtcp2_is_bidi_stream(id)) { + ngtcp2_conn_extend_max_streams_bidi(*this, 1); + } else { + ngtcp2_conn_extend_max_streams_uni(*this, 1); } } diff --git a/src/quic/session.h b/src/quic/session.h index 6cc3db4c86f4..936c9e68db6e 100644 --- a/src/quic/session.h +++ b/src/quic/session.h @@ -518,6 +518,8 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source { void AddStream(BaseObjectPtr stream, CreateStreamOption option = CreateStreamOption::NOTIFY); void RemoveStream(stream_id id); + + void ExtendMaxStreams(stream_id id); void ResumeStream(stream_id id); void StreamDataBlocked(stream_id id); void ShutdownStream(stream_id id, QuicError error = QuicError()); @@ -552,8 +554,6 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source { // Open a unidirectional stream, setting *id on success, or returning false bool OpenUnidirectionalStream(stream_id* id); - void ExtendMaxStreams(Direction direction, uint64_t max); - // Signal that we've consumed `len` bytes on stream `id` to update flow // control void Consume(stream_id id, size_t len); diff --git a/test/parallel/test-quic-h3-stream-credit.mjs b/test/parallel/test-quic-h3-stream-credit.mjs new file mode 100644 index 000000000000..276a6ede247a --- /dev/null +++ b/test/parallel/test-quic-h3-stream-credit.mjs @@ -0,0 +1,87 @@ +// Flags: --experimental-quic --experimental-stream-iter --no-warnings + +// Test: completing an HTTP/3 request returns exactly one unit of stream +// credit. +// +// initialMaxStreamsBidi = 1 lets the client hold one request stream open at +// a time. Every stream that finishes on the wire returns credit for exactly +// one more, so a batch of requests issued at once stays serialised for the +// life of the connection: the server must never see two live at the same +// time, and must not stall by losing credit unexpectedly. + +import { hasQuic, skip, mustCall } from '../common/index.mjs'; +import assert from 'node:assert'; +import * as fixtures from '../common/fixtures.mjs'; + +if (!hasQuic) { + skip('QUIC is not enabled'); +} + +const { listen, connect } = await import('node:quic'); +const { createPrivateKey } = await import('node:crypto'); +const { bytes } = await import('stream/iter'); + +const key = createPrivateKey(fixtures.readKey('agent1-key.pem')); +const cert = fixtures.readKey('agent1-cert.pem'); + +const kRequests = 6; + +let liveServerStreams = 0; +let peakLiveServerStreams = 0; + +const serverEndpoint = await listen(mustCall((serverSession) => { + serverSession.onstream = mustCall((stream) => { + liveServerStreams++; + peakLiveServerStreams = Math.max(peakLiveServerStreams, liveServerStreams); + stream.closed.then(mustCall(() => { liveServerStreams--; })); + }, kRequests); +}), { + sni: { '*': { keys: [key], certs: [cert] } }, + // Only one client-initiated bidi stream may be open at a time. + transportParams: { initialMaxStreamsBidi: 1 }, + onheaders: mustCall(function() { + this.sendHeaders({ ':status': '200' }); + const w = this.writer; + w.writeSync('ok'); + w.endSync(); + }, kRequests), +}); + +const clientSession = await connect(serverEndpoint.address, { + servername: 'localhost', + verifyPeer: 'manual', +}); + +const info = await clientSession.opened; +assert.strictEqual(info.protocol, 'h3'); + +// Issue all requests up front. The 1st will open, the others will be left +// pending and fire as the max-stream credit is returned. +const streams = []; +for (let i = 0; i < kRequests; i++) { + streams.push(await clientSession.createBidirectionalStream({ + headers: { + ':method': 'GET', + ':path': `/${i}`, + ':scheme': 'https', + ':authority': 'localhost', + }, + })); +} +assert.strictEqual(streams[0].pending, false); +assert.ok(streams.slice(1).every((s) => s.pending), + 'only one stream may open while the limit is 1'); + +const unexpectedStreamCount = () => + `server saw ${peakLiveServerStreams} streams open at once with a limit of 1`; + +for (const stream of streams) { + assert.ok(peakLiveServerStreams <= 1, unexpectedStreamCount()); + await bytes(stream); + await stream.closed; +} + +assert.strictEqual(peakLiveServerStreams, 1, unexpectedStreamCount()); + +await clientSession.close(); +await serverEndpoint.close();