Skip to content

quic: improve stream cleanup & lookup #65944

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
pimterry wants to merge 1 commit into
base: main
Choose a base branch
from
Open

quic: improve stream cleanup & lookup #65944

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions src/quic/application.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

@jasnell jasnell Sep 9, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: let's add a comment here explaining briefly the conditions in whch stream can be nullptr

stream->Destroy(std::move(error));
}

Expand Down
16 changes: 7 additions & 9 deletions src/quic/application.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand Down
5 changes: 0 additions & 5 deletions src/quic/defs.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -290,11 +290,6 @@ enum class Side : uint8_t {
SERVER,
};

enum class EndpointLabel : uint8_t {
LOCAL,
REMOTE,
};

enum class Direction : uint8_t {
BIDIRECTIONAL,
UNIDIRECTIONAL,
Expand Down
154 changes: 57 additions & 97 deletions src/quic/http3.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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<Array>& headers,
Expand Down Expand Up @@ -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(),
Expand Down Expand Up @@ -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(),
Expand Down Expand Up @@ -1186,20 +1138,26 @@ class Http3ApplicationImpl final : public Session::Application {
return app;
}

static BaseObjectWeakPtr<Stream> 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<Stream> 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 {};
Expand All @@ -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> stream(static_cast<Stream*>(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;
Expand Down Expand Up @@ -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<size_t>(datalen));
BaseObjectPtr<Stream> stream(static_cast<Stream*>(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<size_t>(datalen));
}
return NGTCP2_SUCCESS;
}
Expand All @@ -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*>(stream_user_data)) [[likely]] {
BaseObjectPtr<Stream> 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
Expand All @@ -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;
}
Expand Down Expand Up @@ -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<Session::Application> CreateHttp3Application(
Expand Down
35 changes: 16 additions & 19 deletions src/quic/session.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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));

@jasnell jasnell Sep 9, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A comment here explaining that we're passing stream_id independently from stream because stream might be nullptr would be good.

} else {
session->application().ReceiveStreamClose(stream);
session->application().ReceiveStreamClose(stream_id, stream);
}
return NGTCP2_SUCCESS;
}
Expand Down Expand Up @@ -3320,17 +3322,11 @@ void Session::AddStream(BaseObjectPtr<Stream> 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
Expand Down Expand Up @@ -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<size_t>(max));
break;
case Direction::UNIDIRECTIONAL:
ngtcp2_conn_extend_max_streams_uni(*this, static_cast<size_t>(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);
}
}

Expand Down
4 changes: 2 additions & 2 deletions src/quic/session.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -518,6 +518,8 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
void AddStream(BaseObjectPtr<Stream> 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());
Expand Down Expand Up @@ -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);
Expand Down
Loading
Loading