diff --git a/lib/internal/quic/quic.js b/lib/internal/quic/quic.js index 8ecd5fc776a..91fff21e744 100644 --- a/lib/internal/quic/quic.js +++ b/lib/internal/quic/quic.js @@ -50,7 +50,9 @@ let debug = require('internal/util/debuglog').debuglog('quic', (fn) => { const { Endpoint: Endpoint_, + sendHeaders, setCallbacks, + setHeadersInterest, // The constants to be exposed to end users for various options. CC_ALGO_RENO_STR: CC_ALGO_RENO, @@ -1303,6 +1305,17 @@ function parseHeaderPairs(pairs) { return block; } +function updateHeaderInterest(handle, inner) { + if (handle === undefined) return; + setHeadersInterest( + handle, + inner.onheaders !== undefined || + inner.ontrailers !== undefined || + inner.oninfo !== undefined, + inner.onwanttrailers !== undefined || inner.pendingTrailers !== undefined, + ); +} + /** * Applies session and stream callbacks from an options object to a session. * @param {QuicSession} session @@ -1827,13 +1840,12 @@ class QuicStream { const inner = this.#inner; if (fn === undefined) { inner.onheaders = undefined; - inner.state.wantsHeaders = false; } else { validateFunction(fn, 'onheaders'); assertHeadersSupported(inner.session); inner.onheaders = FunctionPrototypeBind(fn, this); - inner.state.wantsHeaders = true; } + updateHeaderInterest(this.#handle, inner); } /** @type {Function|undefined} */ @@ -1852,6 +1864,7 @@ class QuicStream { assertHeadersSupported(inner.session); inner.oninfo = FunctionPrototypeBind(fn, this); } + updateHeaderInterest(this.#handle, inner); } /** @type {Function|undefined} */ @@ -1870,6 +1883,7 @@ class QuicStream { assertHeadersSupported(inner.session); inner.ontrailers = FunctionPrototypeBind(fn, this); } + updateHeaderInterest(this.#handle, inner); } /** @type {Function|undefined} */ @@ -1883,13 +1897,12 @@ class QuicStream { const inner = this.#inner; if (fn === undefined) { inner.onwanttrailers = undefined; - inner.state.wantsTrailers = false; } else { validateFunction(fn, 'onwanttrailers'); assertHeadersSupported(inner.session); inner.onwanttrailers = FunctionPrototypeBind(fn, this); - inner.state.wantsTrailers = true; } + updateHeaderInterest(this.#handle, inner); } /** @@ -1918,10 +1931,12 @@ class QuicStream { assertHeadersSupported(inner.session); if (headers === undefined) { inner.pendingTrailers = undefined; + updateHeaderInterest(this.#handle, inner); return; } validateObject(headers, 'headers'); inner.pendingTrailers = headers; + updateHeaderInterest(this.#handle, inner); } /** @@ -2104,7 +2119,8 @@ class QuicStream { const headerString = buildNgHeaderString( headers, assertValidPseudoHeader, true /* strictSingleValueFields */); const flags = terminal ? kHeadersFlagsTerminal : kHeadersFlagsNone; - return this.#handle.sendHeaders(kHeadersKindInitial, headerString, flags); + return sendHeaders( + this.#handle, kHeadersKindInitial, headerString, flags); } /** @@ -2123,8 +2139,8 @@ class QuicStream { validateObject(headers, 'headers'); const headerString = buildNgHeaderString( headers, assertValidPseudoHeader, true); - return this.#handle.sendHeaders( - kHeadersKindHints, headerString, kHeadersFlagsNone); + return sendHeaders( + this.#handle, kHeadersKindHints, headerString, kHeadersFlagsNone); } /** @@ -2143,8 +2159,8 @@ class QuicStream { } validateObject(headers, 'headers'); const headerString = buildNgHeaderString(headers); - return this.#handle.sendHeaders( - kHeadersKindTrailing, headerString, kHeadersFlagsNone); + return sendHeaders( + this.#handle, kHeadersKindTrailing, headerString, kHeadersFlagsNone); } /** @@ -2562,7 +2578,7 @@ class QuicStream { assertValidPseudoHeader, true, // This could become an option in future ); - return this.#handle.sendHeaders(kind, headerString, flags); + return sendHeaders(this.#handle, kind, headerString, flags); } [kFinishClose](error) { @@ -2670,7 +2686,6 @@ class QuicStream { switch (kindName) { case 'initial': - assert(inner.onheaders, 'Unexpected stream headers event'); inner.headers ??= block; if (onStreamHeadersChannel.hasSubscribers) { onStreamHeadersChannel.publish({ @@ -2680,7 +2695,8 @@ class QuicStream { headers: block, }); } - safeCallbackInvoke(inner.onheaders, this, block); + if (inner.onheaders) + safeCallbackInvoke(inner.onheaders, this, block); break; case 'trailing': if (onStreamTrailersChannel.hasSubscribers) { @@ -2716,8 +2732,20 @@ class QuicStream { // nghttp3 is asking us to provide trailers to send. // Check for pre-set pendingTrailers first, then the callback. if (inner.pendingTrailers) { - this.sendTrailers(inner.pendingTrailers); + let sent; + try { + sent = this.sendTrailers(inner.pendingTrailers); + } catch (error) { + this.destroy(error); + return; + } + if (!sent) { + this.destroy(new ERR_QUIC_STREAM_ABORTED( + 'Failed to submit trailing headers')); + return; + } inner.pendingTrailers = undefined; + updateHeaderInterest(this.#handle, inner); } else if (typeof inner.onwanttrailers === 'function') { safeCallbackInvoke(inner.onwanttrailers, this); } diff --git a/lib/internal/quic/state.js b/lib/internal/quic/state.js index 769fee68749..21db0f0e3d5 100644 --- a/lib/internal/quic/state.js +++ b/lib/internal/quic/state.js @@ -101,10 +101,8 @@ const { IDX_STATE_STREAM_HAS_OUTBOUND, IDX_STATE_STREAM_HAS_READER, IDX_STATE_STREAM_WANTS_BLOCK, - IDX_STATE_STREAM_WANTS_HEADERS, IDX_STATE_STREAM_WANTS_RESET, IDX_STATE_STREAM_WANTS_STOP_SENDING, - IDX_STATE_STREAM_WANTS_TRAILERS, IDX_STATE_STREAM_RECEIVED_EARLY_DATA, IDX_STATE_STREAM_WRITE_DESIRED_SIZE, IDX_STATE_STREAM_BUDGET, @@ -145,10 +143,8 @@ assert(IDX_STATE_STREAM_RESET !== undefined); assert(IDX_STATE_STREAM_HAS_OUTBOUND !== undefined); assert(IDX_STATE_STREAM_HAS_READER !== undefined); assert(IDX_STATE_STREAM_WANTS_BLOCK !== undefined); -assert(IDX_STATE_STREAM_WANTS_HEADERS !== undefined); assert(IDX_STATE_STREAM_WANTS_RESET !== undefined); assert(IDX_STATE_STREAM_WANTS_STOP_SENDING !== undefined); -assert(IDX_STATE_STREAM_WANTS_TRAILERS !== undefined); assert(IDX_STATE_STREAM_WRITE_DESIRED_SIZE !== undefined); assert(IDX_STATE_STREAM_RESET_CODE !== undefined); @@ -824,20 +820,6 @@ class QuicStreamState { DataViewPrototypeSetUint8(handle, this.#offset + IDX_STATE_STREAM_WANTS_BLOCK, val ? 1 : 0); } - /** @type {boolean} */ - get wantsHeaders() { - const handle = this.#handle; - if (handle === undefined) return undefined; - return DataViewPrototypeGetUint8(handle, this.#offset + IDX_STATE_STREAM_WANTS_HEADERS) !== 0; - } - - /** @type {boolean} */ - set wantsHeaders(val) { - const handle = this.#handle; - if (handle === undefined) return; - DataViewPrototypeSetUint8(handle, this.#offset + IDX_STATE_STREAM_WANTS_HEADERS, val ? 1 : 0); - } - /** @type {boolean} */ get wantsReset() { const handle = this.#handle; @@ -870,20 +852,6 @@ class QuicStreamState { val ? 1 : 0); } - /** @type {boolean} */ - get wantsTrailers() { - const handle = this.#handle; - if (handle === undefined) return undefined; - return DataViewPrototypeGetUint8(handle, this.#offset + IDX_STATE_STREAM_WANTS_TRAILERS) !== 0; - } - - /** @type {boolean} */ - set wantsTrailers(val) { - const handle = this.#handle; - if (handle === undefined) return; - DataViewPrototypeSetUint8(handle, this.#offset + IDX_STATE_STREAM_WANTS_TRAILERS, val ? 1 : 0); - } - /** @type {boolean} */ get early() { const handle = this.#handle; @@ -948,8 +916,6 @@ class QuicStreamState { wantsBlock, wantsReset, wantsStopSending, - wantsHeaders, - wantsTrailers, early, resetCode, writeDesiredSize, @@ -969,8 +935,6 @@ class QuicStreamState { wantsBlock, wantsReset, wantsStopSending, - wantsHeaders, - wantsTrailers, early, resetCode: `${resetCode}`, writeDesiredSize, @@ -1007,8 +971,6 @@ class QuicStreamState { wantsBlock, wantsReset, wantsStopSending, - wantsHeaders, - wantsTrailers, early, resetCode, writeDesiredSize, @@ -1028,8 +990,6 @@ class QuicStreamState { wantsBlock, wantsReset, wantsStopSending, - wantsHeaders, - wantsTrailers, early, resetCode, writeDesiredSize, diff --git a/src/quic/application.cc b/src/quic/application.cc index f83b64bad49..1e78b7931ce 100644 --- a/src/quic/application.cc +++ b/src/quic/application.cc @@ -117,7 +117,7 @@ Maybe Session::Application_Options::From( // Ensure the advertised max_field_section_size in SETTINGS is at least // as large as max_header_length. Otherwise the peer would be told to - // restrict headers to a smaller size than what CanAddHeader accepts. + // restrict headers to a smaller size than what the HTTP/3 stream accepts. if (options.max_field_section_size < options.max_header_length) { options.max_field_section_size = options.max_header_length; } diff --git a/src/quic/application.h b/src/quic/application.h index df472e0a9fd..0ae4ab53ab7 100644 --- a/src/quic/application.h +++ b/src/quic/application.h @@ -11,6 +11,17 @@ namespace node::quic { +enum class HeadersKind : uint8_t { + HINTS, + INITIAL, + TRAILING, +}; + +enum class HeadersFlags : uint8_t { + NONE, + TERMINAL, +}; + // An Application implements the ALPN-protocol specific semantics on behalf // of a QUIC Session. class Session::Application : public MemoryRetainer { @@ -95,13 +106,10 @@ class Session::Application : public MemoryRetainer { // Application. virtual bool AcknowledgeStreamData(stream_id id, size_t datalen); - // Called to determine if a Header can be added to this application. - // Applications that do not support headers will always return false. - virtual bool CanAddHeader(size_t current_count, - size_t current_headers_length, - size_t this_header_length) { - return false; - } + // Called when a pending transport stream receives its stream ID. Protocols + // can use this to flush operations that require an opened stream. Returns + // false if deferred application data could not be submitted. + virtual bool StreamOpened(Stream& stream) { return true; } // Called when ngtcp2 reports NGTCP2_ERR_STREAM_SHUT_WR for a stream. // Applications that manage their own framing (e.g., HTTP/3) must inform @@ -173,13 +181,19 @@ class Session::Application : public MemoryRetainer { // Submits an outbound block of headers for the given stream. Not all // Application types will support headers, in which case this function // should return false. - virtual bool SendHeaders(const Stream& stream, + virtual bool SendHeaders(Stream& stream, HeadersKind kind, const v8::Local& headers, HeadersFlags flags = HeadersFlags::NONE) { return false; } + // Updates JavaScript callback interest for an application's stream header + // events. Applications without header semantics ignore this. + virtual void SetHeadersInterest(Stream& stream, + bool wants_headers, + bool wants_trailers) {} + // Returns true if the application protocol supports sending and // receiving headers on streams (e.g. HTTP/3). Applications that // do not support headers should return false (the default). diff --git a/src/quic/bindingdata.cc b/src/quic/bindingdata.cc index 31467a8477a..d5711d88cb9 100644 --- a/src/quic/bindingdata.cc +++ b/src/quic/bindingdata.cc @@ -1,6 +1,7 @@ #if HAVE_OPENSSL && HAVE_QUIC #include "guard.h" #ifndef OPENSSL_NO_QUIC +#include #include #include #include @@ -13,13 +14,16 @@ #include #include #include +#include "application.h" #include "bindingdata.h" #include "session.h" #include "session_manager.h" +#include "streams.h" namespace node { using mem::kReserveSizeAndAlign; +using v8::Array; using v8::DictionaryTemplate; using v8::Function; using v8::FunctionTemplate; @@ -288,6 +292,26 @@ void nghttp3_debug_log(const char* fmt, va_list args) { void BindingData::InitPerContext(Realm* realm, Local target) { nghttp3_set_debug_vprintf_callback(nghttp3_debug_log); SetMethod(realm->context(), target, "setCallbacks", SetCallbacks); + SetMethod(realm->context(), target, "sendHeaders", SendHeaders); + SetMethod(realm->context(), target, "setHeadersInterest", SetHeadersInterest); + + constexpr int QUIC_STREAM_HEADERS_KIND_HINTS = + static_cast(HeadersKind::HINTS); + constexpr int QUIC_STREAM_HEADERS_KIND_INITIAL = + static_cast(HeadersKind::INITIAL); + constexpr int QUIC_STREAM_HEADERS_KIND_TRAILING = + static_cast(HeadersKind::TRAILING); + constexpr int QUIC_STREAM_HEADERS_FLAGS_NONE = + static_cast(HeadersFlags::NONE); + constexpr int QUIC_STREAM_HEADERS_FLAGS_TERMINAL = + static_cast(HeadersFlags::TERMINAL); + + NODE_DEFINE_CONSTANT(target, QUIC_STREAM_HEADERS_KIND_HINTS); + NODE_DEFINE_CONSTANT(target, QUIC_STREAM_HEADERS_KIND_INITIAL); + NODE_DEFINE_CONSTANT(target, QUIC_STREAM_HEADERS_KIND_TRAILING); + NODE_DEFINE_CONSTANT(target, QUIC_STREAM_HEADERS_FLAGS_NONE); + NODE_DEFINE_CONSTANT(target, QUIC_STREAM_HEADERS_FLAGS_TERMINAL); + Realm::GetCurrent(realm->context())->AddBindingData(target); } @@ -295,6 +319,32 @@ void BindingData::RegisterExternalReferences( ExternalReferenceRegistry* registry) { registry->Register(IllegalConstructor); registry->Register(SetCallbacks); + registry->Register(SendHeaders); + registry->Register(SetHeadersInterest); +} + +JS_METHOD_IMPL(BindingData::SendHeaders) { + Stream* stream; + ASSIGN_OR_RETURN_UNWRAP(&stream, args[0]); + CHECK(args[1]->IsUint32()); // Kind + CHECK(args[2]->IsArray()); // Headers + CHECK(args[3]->IsUint32()); // Flags + + HeadersKind kind = FromV8Value(args[1]); + Local headers = args[2].As(); + HeadersFlags flags = FromV8Value(args[3]); + + args.GetReturnValue().Set(stream->session().application().SendHeaders( + *stream, kind, headers, flags)); +} + +JS_METHOD_IMPL(BindingData::SetHeadersInterest) { + Stream* stream; + ASSIGN_OR_RETURN_UNWRAP(&stream, args[0]); + CHECK(args[1]->IsBoolean()); + CHECK(args[2]->IsBoolean()); + stream->session().application().SetHeadersInterest( + *stream, args[1]->IsTrue(), args[2]->IsTrue()); } BindingData::BindingData(Realm* realm, Local object) diff --git a/src/quic/bindingdata.h b/src/quic/bindingdata.h index e4763b5b0d3..e56c861261a 100644 --- a/src/quic/bindingdata.h +++ b/src/quic/bindingdata.h @@ -294,6 +294,8 @@ class BindingData final // Installs the set of JavaScript callback functions that are used to // bridge out to the JS API. JS_METHOD(SetCallbacks); + JS_METHOD(SendHeaders); + JS_METHOD(SetHeadersInterest); // Lazily-created per-Realm SessionManager. Centralizes CID -> Session // routing so that any endpoint can route packets to any session. diff --git a/src/quic/defs.h b/src/quic/defs.h index 5184288475b..32b3a10d7b8 100644 --- a/src/quic/defs.h +++ b/src/quic/defs.h @@ -300,17 +300,6 @@ enum class Direction : uint8_t { UNIDIRECTIONAL, }; -enum class HeadersKind : uint8_t { - HINTS, - INITIAL, - TRAILING, -}; - -enum class HeadersFlags : uint8_t { - NONE, - TERMINAL, -}; - enum class StreamPriority : uint8_t { DEFAULT = NGHTTP3_DEFAULT_URGENCY, LOW = NGHTTP3_URGENCY_LOW, diff --git a/src/quic/http3.cc b/src/quic/http3.cc index 6e1d0c44a04..43ead1b8164 100644 --- a/src/quic/http3.cc +++ b/src/quic/http3.cc @@ -22,7 +22,11 @@ namespace node { using v8::Array; +using v8::Global; +using v8::Integer; using v8::Local; +using v8::LocalVector; +using v8::Value; namespace quic { @@ -138,6 +142,25 @@ struct Http3HeaderTraits { using Http3Header = NgHeader; +struct Http3StreamState final : public StreamApplicationState { + struct PendingHeaders final { + HeadersKind kind; + Global headers; + HeadersFlags flags; + + PendingHeaders(HeadersKind kind, Global headers, HeadersFlags flags) + : kind(kind), headers(std::move(headers)), flags(flags) {} + DISALLOW_COPY_AND_MOVE(PendingHeaders) + }; + + std::vector> pending_headers; + std::vector> headers; + HeadersKind headers_kind = HeadersKind::INITIAL; + size_t headers_length = 0; + bool wants_headers = false; + bool wants_trailers = false; +}; + // Implements the low-level HTTP/3 Application semantics. class Http3ApplicationImpl final : public Session::Application { public: @@ -334,17 +357,6 @@ class Http3ApplicationImpl final : public Session::Application { return nghttp3_conn_add_ack_offset(*this, id, datalen) == 0; } - bool CanAddHeader(size_t current_count, - size_t current_headers_length, - size_t this_header_length) override { - // We cannot add the header if we've either reached - // * the max number of header pairs or - // * the max number of header bytes (name + value combined) - return (current_count < options_.max_header_pairs) && - (current_headers_length + this_header_length) <= - options_.max_header_length; - } - bool stream_fin_managed_by_application() const override { return true; } void StreamWriteShut(stream_id id) override { @@ -537,75 +549,49 @@ class Http3ApplicationImpl final : public Session::Application { Application::ReceiveStreamStopSending(stream, std::move(error)); } - bool SendHeaders(const Stream& stream, - HeadersKind kind, - const Local& headers, - HeadersFlags flags = HeadersFlags::NONE) override { - Session::SendPendingDataScope send_scope(&session()); - Http3Headers nva(env(), headers); + bool StreamOpened(Stream& stream) override { + auto* state = GetStreamState(stream); + if (state == nullptr || state->pending_headers.empty()) return true; - switch (kind) { - case HeadersKind::HINTS: { - if (!session().is_server()) { - // Client side cannot send hints - return false; - } - Debug(&session(), - "Submitting %" PRIu64 " early hints for stream %" PRIu64, - stream.id()); - return nghttp3_conn_submit_info( - *this, stream.id(), nva.data(), nva.length()) == 0; - break; + decltype(state->pending_headers) pending; + state->pending_headers.swap(pending); + Session::SendPendingDataScope send_scope(&session()); + for (auto& headers : pending) { + if (!SubmitHeaders(stream, + headers->kind, + headers->headers.Get(env()->isolate()), + headers->flags)) { + return false; } - case HeadersKind::INITIAL: { - static constexpr nghttp3_data_reader reader = {on_read_data_callback}; - const nghttp3_data_reader* reader_ptr = nullptr; - - // If the terminal flag is set, that means that we know we're only - // sending headers and no body and the stream writable side should be - // closed immediately because there is no nghttp3_data_reader provided. - if (flags != HeadersFlags::TERMINAL) { - reader_ptr = &reader; - } + } + return true; + } - if (session().is_server()) { - // If this is a server, we're submitting a response... - Debug(&session(), - "Submitting %" PRIu64 " response headers for stream %" PRIu64, - nva.length(), - stream.id()); - return nghttp3_conn_submit_response(*this, - stream.id(), - nva.data(), - nva.length(), - reader_ptr) == 0; - } else { - // Otherwise we're submitting a request... - Debug(&session(), - "Submitting %" PRIu64 " request headers for stream %" PRIu64, - nva.length(), - stream.id()); - return nghttp3_conn_submit_request(*this, - stream.id(), - nva.data(), - nva.length(), - reader_ptr, - const_cast(&stream)) == 0; - } - break; - } - case HeadersKind::TRAILING: { - Debug(&session(), - "Submitting %" PRIu64 " trailing headers for stream %" PRIu64, - nva.length(), - stream.id()); - return nghttp3_conn_submit_trailers( - *this, stream.id(), nva.data(), nva.length()) == 0; - break; - } + bool SendHeaders(Stream& stream, + HeadersKind kind, + const Local& headers, + HeadersFlags flags = HeadersFlags::NONE) override { + if (kind == HeadersKind::HINTS && !session().is_server()) return false; + + if (stream.is_pending()) { + Debug(&session(), "Enqueuing headers for pending HTTP/3 stream"); + auto& state = GetOrCreateStreamState(stream); + state.pending_headers.push_back( + std::make_unique( + kind, Global(env()->isolate(), headers), flags)); + return true; } - return false; + Session::SendPendingDataScope send_scope(&session()); + return SubmitHeaders(stream, kind, headers, flags); + } + + void SetHeadersInterest(Stream& stream, + bool wants_headers, + bool wants_trailers) override { + auto& state = GetOrCreateStreamState(stream); + state.wants_headers = wants_headers; + state.wants_trailers = wants_trailers; } void SetStreamPriority(const Stream& stream, @@ -642,8 +628,7 @@ class Http3ApplicationImpl final : public Session::Application { // PRIORITY_UPDATE frames). Client-side priority is tracked by the // Stream itself and returned directly from GetPriority in streams.cc. if (!session().is_server()) { - auto& stored = stream.stored_priority(); - return {stored.priority, stored.flags}; + return {stream.priority(), stream.priority_flags()}; } nghttp3_pri pri; if (nghttp3_conn_get_stream_priority(*this, &pri, stream.id()) == 0) { @@ -730,7 +715,7 @@ class Http3ApplicationImpl final : public Session::Application { // for the next writev_stream in the send loop. if (pending_trailers_stream_ == data->id) { pending_trailers_stream_ = -1; - if (data->stream) data->stream->EmitWantTrailers(); + if (data->stream) EmitWantTrailers(*data->stream); } return true; } @@ -750,6 +735,137 @@ class Http3ApplicationImpl final : public Session::Application { id == qpack_enc_stream_id_; } + bool SubmitHeaders(Stream& stream, + HeadersKind kind, + const Local& headers, + HeadersFlags flags) { + Http3Headers nva(env(), headers); + + switch (kind) { + case HeadersKind::HINTS: { + if (!session().is_server()) return false; + Debug(&session(), + "Submitting %" PRIu64 " early hints for stream %" PRIu64, + stream.id()); + return nghttp3_conn_submit_info( + *this, stream.id(), nva.data(), nva.length()) == 0; + } + case HeadersKind::INITIAL: { + static constexpr nghttp3_data_reader reader = {on_read_data_callback}; + const nghttp3_data_reader* reader_ptr = nullptr; + if (flags != HeadersFlags::TERMINAL) reader_ptr = &reader; + + if (session().is_server()) { + Debug(&session(), + "Submitting %" PRIu64 " response headers for stream %" PRIu64, + nva.length(), + stream.id()); + return nghttp3_conn_submit_response(*this, + stream.id(), + nva.data(), + nva.length(), + reader_ptr) == 0; + } + + Debug(&session(), + "Submitting %" PRIu64 " request headers for stream %" PRIu64, + nva.length(), + stream.id()); + return nghttp3_conn_submit_request(*this, + stream.id(), + nva.data(), + nva.length(), + reader_ptr, + &stream) == 0; + } + case HeadersKind::TRAILING: { + Debug(&session(), + "Submitting %" PRIu64 " trailing headers for stream %" PRIu64, + nva.length(), + stream.id()); + return nghttp3_conn_submit_trailers( + *this, stream.id(), nva.data(), nva.length()) == 0; + } + } + + return false; + } + + Http3StreamState* GetStreamState(Stream& stream) const { + return static_cast(stream.application_state()); + } + + Http3StreamState& GetOrCreateStreamState(Stream& stream) { + if (stream.application_state() == nullptr) { + stream.set_application_state(std::make_unique()); + } + return *GetStreamState(stream); + } + + void BeginHeaders(Stream& stream, HeadersKind kind) { + auto& state = GetOrCreateStreamState(stream); + state.headers_length = 0; + state.headers.clear(); + state.headers_kind = kind; + } + + bool AddHeader(Stream& stream, std::unique_ptr header) { + auto& state = GetOrCreateStreamState(stream); + size_t length = header->length(); + if (state.headers.size() >= options_.max_header_pairs || + state.headers_length + length > options_.max_header_length) { + return false; + } + state.headers_length += length; + state.headers.push_back(std::move(header)); + return true; + } + + void EmitHeaders(Stream& stream) { + auto& state = GetOrCreateStreamState(stream); + stream.RecordReceivedActivity(); + if (!env()->can_call_into_js() || !state.wants_headers) { + state.headers.clear(); + return; + } + + CallbackScope cb_scope(&stream); + auto& binding = BindingData::Get(env()); + size_t count = state.headers.size() * 2; + LocalVector values(env()->isolate(), count); + + for (size_t i = 0; i < state.headers.size(); i++) { + Local name; + Local value; + if (!state.headers[i]->GetName(&binding).ToLocal(&name) || + !state.headers[i]->GetValue(&binding).ToLocal(&value)) [[unlikely]] { + state.headers.clear(); + return; + } + values[i * 2] = name; + values[i * 2 + 1] = value; + } + + state.headers.clear(); + Local argv[] = { + Array::New(env()->isolate(), values.data(), count), + Integer::NewFromUnsigned(env()->isolate(), + static_cast(state.headers_kind))}; + stream.MakeCallback( + binding.stream_headers_callback(), arraysize(argv), argv); + } + + void EmitWantTrailers(Stream& stream) { + auto* state = GetStreamState(stream); + if (!env()->can_call_into_js() || state == nullptr || + !state->wants_trailers) { + return; + } + CallbackScope cb_scope(&stream); + stream.MakeCallback( + BindingData::Get(env()).stream_trailers_callback(), 0, nullptr); + } + void BuildOriginPayload() { // Build the serialized ORIGIN frame payload from the SNI configuration. // Each origin entry is: 2-byte BE length + origin string. @@ -831,7 +947,7 @@ class Http3ApplicationImpl final : public Session::Application { "HTTP/3 application beginning initial block of headers for stream " "%" PRIi64, id); - stream->BeginHeaders(HeadersKind::INITIAL); + BeginHeaders(*stream, HeadersKind::INITIAL); } void OnReceiveHeader(stream_id id, std::unique_ptr header) { @@ -843,7 +959,7 @@ class Http3ApplicationImpl final : public Session::Application { Debug(&session(), "HTTP/3 application switching to hints headers for stream %" PRIi64, stream->id()); - stream->set_headers_kind(HeadersKind::HINTS); + GetOrCreateStreamState(*stream).headers_kind = HeadersKind::HINTS; } IF_QUIC_DEBUG(env()) { Debug(&session(), @@ -851,7 +967,7 @@ class Http3ApplicationImpl final : public Session::Application { header->name(), header->value()); } - stream->AddHeader(std::move(header)); + AddHeader(*stream, std::move(header)); } void OnEndHeaders(stream_id id, int fin) { @@ -861,8 +977,8 @@ class Http3ApplicationImpl final : public Session::Application { Debug(&session(), "HTTP/3 application received end of headers for stream %" PRIi64, id); - stream->EmitHeaders(); - // EmitHeaders() calls into JavaScript, which can synchronously destroy the + EmitHeaders(*stream); + // EmitHeaders calls into JavaScript, which can synchronously destroy the // stream. Its arena-backed state is released by Destroy(), so do not touch // the stream again if that happened. if (stream->is_destroyed()) return; @@ -884,7 +1000,7 @@ class Http3ApplicationImpl final : public Session::Application { Debug(&session(), "HTTP/3 application beginning block of trailers for stream %" PRIi64, id); - stream->BeginHeaders(HeadersKind::TRAILING); + BeginHeaders(*stream, HeadersKind::TRAILING); } void OnReceiveTrailer(stream_id id, std::unique_ptr header) { @@ -897,7 +1013,7 @@ class Http3ApplicationImpl final : public Session::Application { header->name(), header->value()); } - stream->AddHeader(std::move(header)); + AddHeader(*stream, std::move(header)); } void OnEndTrailers(stream_id id, int fin) { @@ -907,8 +1023,8 @@ class Http3ApplicationImpl final : public Session::Application { Debug(&session(), "HTTP/3 application received end of trailers for stream %" PRIi64, id); - stream->EmitHeaders(); - // EmitHeaders() calls into JavaScript, which can synchronously destroy the + EmitHeaders(*stream); + // EmitHeaders calls into JavaScript, which can synchronously destroy the // stream. Its arena-backed state is released by Destroy(), so do not touch // the stream again if that happened. if (stream->is_destroyed()) return; @@ -1112,7 +1228,7 @@ class Http3ApplicationImpl final : public Session::Application { if (stream->is_eos()) { *pflags |= NGHTTP3_DATA_FLAG_EOF; - if (stream->wants_trailers()) { + if (app.GetOrCreateStreamState(*stream).wants_trailers) { *pflags |= NGHTTP3_DATA_FLAG_NO_END_STREAM; app.pending_trailers_stream_ = id; } @@ -1131,7 +1247,7 @@ class Http3ApplicationImpl final : public Session::Application { return; case bob::Status::STATUS_EOS: *pflags |= NGHTTP3_DATA_FLAG_EOF; - if (stream->wants_trailers()) { + if (app.GetOrCreateStreamState(*stream).wants_trailers) { *pflags |= NGHTTP3_DATA_FLAG_NO_END_STREAM; app.pending_trailers_stream_ = id; } diff --git a/src/quic/streams.cc b/src/quic/streams.cc index 1f8761c7504..9a11bb4d591 100644 --- a/src/quic/streams.cc +++ b/src/quic/streams.cc @@ -26,12 +26,10 @@ using v8::BackingStore; using v8::BackingStoreInitializationMode; using v8::BigInt; using v8::FunctionCallbackInfo; -using v8::Global; using v8::HandleScope; using v8::Integer; using v8::Just; using v8::Local; -using v8::LocalVector; using v8::Maybe; using v8::Nothing; using v8::Object; @@ -57,14 +55,10 @@ namespace quic { V(HAS_READER, has_reader, uint8_t) \ /* Set when the stream has a block event handler */ \ V(WANTS_BLOCK, wants_block, uint8_t) \ - /* Set when the stream has a headers event handler */ \ - V(WANTS_HEADERS, wants_headers, uint8_t) \ /* Set when the stream has a reset event handler */ \ V(WANTS_RESET, wants_reset, uint8_t) \ /* Set when the stream has a stop sending event handler */ \ V(WANTS_STOP_SENDING, wants_stop_sending, uint8_t) \ - /* Set when the stream has a trailers event handler */ \ - V(WANTS_TRAILERS, wants_trailers, uint8_t) \ /* True when 0-RTT early data was received */ \ V(RECEIVED_EARLY_DATA, received_early_data, uint8_t) \ V(WRITE_DESIRED_SIZE, write_desired_size, uint32_t) \ @@ -98,7 +92,6 @@ namespace quic { #define STREAM_JS_METHODS(V) \ V(AttachSource, attachSource, false) \ V(Destroy, destroy, false) \ - V(SendHeaders, sendHeaders, false) \ V(StopSending, stopSending, false) \ V(ResetStream, resetStream, false) \ V(SetPriority, setPriority, false) \ @@ -227,17 +220,6 @@ void PendingStream::reject(QuicError error) { stream_->Destroy(error); } -struct Stream::PendingHeaders { - HeadersKind kind; - Global headers; - HeadersFlags flags; - PendingHeaders(HeadersKind kind_, Global headers_, HeadersFlags flags_) - : kind(kind_), headers(std::move(headers_)), flags(flags_) {} - DISALLOW_COPY_AND_MOVE(PendingHeaders) -}; - -// ============================================================================ - struct Stream::State { #define V(_, name, type) type name; STREAM_STATE(V) @@ -445,37 +427,6 @@ struct Stream::Impl { } } - // Sends a block of headers to the peer. If the stream is not yet open, - // the headers will be queued and sent immediately when the stream is - // opened. Returns false if the application does not support headers. - JS_METHOD(SendHeaders) { - Stream* stream; - ASSIGN_OR_RETURN_UNWRAP(&stream, args.This()); - CHECK(args[0]->IsUint32()); // Kind - CHECK(args[1]->IsArray()); // Headers - CHECK(args[2]->IsUint32()); // Flags - - HeadersKind kind = FromV8Value(args[0]); - Local headers = args[1].As(); - HeadersFlags flags = FromV8Value(args[2]); - - // If the stream is pending, the headers will be queued until the - // stream is opened, at which time the queued header block will be - // immediately sent when the stream is opened. If we already know - // that the application does not support headers, return false - // immediately so the JS side can throw an appropriate error. - if (stream->is_pending()) { - if (!stream->session().application().SupportsHeaders()) { - return args.GetReturnValue().Set(false); - } - stream->EnqueuePendingHeaders(kind, headers, flags); - return args.GetReturnValue().Set(true); - } - - args.GetReturnValue().Set(stream->session().application().SendHeaders( - *stream, kind, headers, flags)); - } - // Tells the peer to stop sending data for this stream. This has the effect // of shutting down the readable side of the stream for this peer. Any data // that has already been received is still readable. @@ -1057,25 +1008,6 @@ void Stream::InitPerContext(Realm* realm, Local target) { #undef V NODE_DEFINE_CONSTANT(target, IDX_STATS_STREAM_COUNT); - - constexpr int QUIC_STREAM_HEADERS_KIND_HINTS = - static_cast(HeadersKind::HINTS); - constexpr int QUIC_STREAM_HEADERS_KIND_INITIAL = - static_cast(HeadersKind::INITIAL); - constexpr int QUIC_STREAM_HEADERS_KIND_TRAILING = - static_cast(HeadersKind::TRAILING); - - constexpr int QUIC_STREAM_HEADERS_FLAGS_NONE = - static_cast(HeadersFlags::NONE); - constexpr int QUIC_STREAM_HEADERS_FLAGS_TERMINAL = - static_cast(HeadersFlags::TERMINAL); - - NODE_DEFINE_CONSTANT(target, QUIC_STREAM_HEADERS_KIND_HINTS); - NODE_DEFINE_CONSTANT(target, QUIC_STREAM_HEADERS_KIND_INITIAL); - NODE_DEFINE_CONSTANT(target, QUIC_STREAM_HEADERS_KIND_TRAILING); - - NODE_DEFINE_CONSTANT(target, QUIC_STREAM_HEADERS_FLAGS_NONE); - NODE_DEFINE_CONSTANT(target, QUIC_STREAM_HEADERS_FLAGS_TERMINAL); } Stream* Stream::From(void* stream_user_data) { @@ -1243,25 +1175,6 @@ void Stream::NotifyStreamOpened(stream_id id) { *this, priority_.priority, priority_.flags); priority_.pending = false; } - if (!pending_headers_queue_.empty()) { - if (!session().application().SupportsHeaders()) { - // Headers were enqueued while the application was not yet known - // (headers_supported == 0), and the negotiated application does - // not support headers. This is a fatal mismatch. - Destroy(QuicError::ForApplication( - session().application().GetInternalErrorCode())); - return; - } - decltype(pending_headers_queue_) queue; - pending_headers_queue_.swap(queue); - for (auto& headers : queue) { - session().application().SendHeaders( - *this, - headers->kind, - headers->headers.Get(env()->isolate()), - headers->flags); - } - } // If the stream is not a local unidirectional stream and is_readable is // false, then we should shutdown the streams readable side now. if (!is_local_unidirectional() && !is_readable()) { @@ -1277,6 +1190,15 @@ void Stream::NotifyStreamOpened(stream_id id) { // since the stream likely hasn't had any opporunity to get blocked // yet, but just for completeness, let's make sure. if (outbound_) session().ResumeStream(id); + + // This may make application data sendable, so keep it as the final action: + // sending can eventually call into JavaScript and destroy the stream. + BaseObjectPtr self(this); + auto& application = session().application(); + error_code internal_error = application.GetInternalErrorCode(); + if (!application.StreamOpened(*this) && !is_destroyed()) { + Destroy(QuicError::ForApplication(internal_error)); + } } void Stream::NotifyReadableEnded(error_code code) { @@ -1291,14 +1213,6 @@ void Stream::NotifyWritableEnded(error_code code) { ngtcp2_conn_shutdown_stream_write(session(), 0, id(), code); } -void Stream::EnqueuePendingHeaders(HeadersKind kind, - Local headers, - HeadersFlags flags) { - Debug(this, "Enqueuing headers for pending stream"); - pending_headers_queue_.push_back(std::make_unique( - kind, Global(env()->isolate(), headers), flags)); -} - bool Stream::is_pending() const { return state()->pending; } @@ -1336,6 +1250,10 @@ uint64_t Stream::last_activity_timestamp() const { return ts != 0 ? ts : stats()->created_at; } +void Stream::RecordReceivedActivity() { + STAT_RECORD_TIMESTAMP(Stats, received_at); +} + bool Stream::is_local_unidirectional() const { return direction() == Direction::UNIDIRECTIONAL && ngtcp2_conn_is_local_stream(*session_, id()); @@ -1350,10 +1268,6 @@ bool Stream::is_eos() const { return state()->fin_sent; } -bool Stream::wants_trailers() const { - return state()->wants_trailers; -} - void Stream::set_early() { state()->received_early_data = 1; } @@ -1587,28 +1501,6 @@ int Stream::DoPull(bob::Next next, return outbound_->Pull(std::move(next), options, data, count, max_count_hint); } -void Stream::BeginHeaders(HeadersKind kind) { - headers_length_ = 0; - headers_.clear(); - set_headers_kind(kind); -} - -void Stream::set_headers_kind(HeadersKind kind) { - headers_kind_ = kind; -} - -bool Stream::AddHeader(std::unique_ptr
header) { - size_t len = header->length(); - if (!session_->application().CanAddHeader( - headers_.size(), headers_length_, len)) { - return false; - } - - headers_length_ += len; - headers_.push_back(std::move(header)); - return true; -} - void Stream::Acknowledge(size_t datalen) { if (outbound_ == nullptr) return; @@ -1680,6 +1572,7 @@ void Stream::Destroy(QuicError error) { // We are going to release our reference to the outbound_ queue here. outbound_.reset(); + application_state_.reset(); // EndReadable() above already flushed accumulated data. Just release // the ring buffer memory. @@ -1959,42 +1852,6 @@ void Stream::EmitClose(const QuicError& error) { MakeCallback(BindingData::Get(env()).stream_close_callback(), 1, &err); } -void Stream::EmitHeaders() { - STAT_RECORD_TIMESTAMP(Stats, received_at); - // state()->wants_headers will be set from the javascript side if the - // stream object has a handler for the headers event. - if (!env()->can_call_into_js() || !state()->wants_headers) { - headers_.clear(); - return; - } - CallbackScope cb_scope(this); - - auto& binding = BindingData::Get(env()); - size_t count = headers_.size() * 2; - LocalVector values(env()->isolate(), count); - - for (size_t i = 0; i < headers_.size(); i++) { - Local name; - Local value; - if (!headers_[i]->GetName(&binding).ToLocal(&name) || - !headers_[i]->GetValue(&binding).ToLocal(&value)) [[unlikely]] { - headers_.clear(); - return; - } - values[i * 2] = name; - values[i * 2 + 1] = value; - } - - headers_.clear(); - - Local argv[] = { - Array::New(env()->isolate(), values.data(), count), - Integer::NewFromUnsigned(env()->isolate(), - static_cast(headers_kind_))}; - - MakeCallback(binding.stream_headers_callback(), arraysize(argv), argv); -} - void Stream::EmitReset(const QuicError& error) { // state()->wants_reset will be set from the javascript side if the // stream object has a handler for the reset event. @@ -2019,16 +1876,6 @@ void Stream::EmitStopSending(const QuicError& error) { MakeCallback(BindingData::Get(env()).stream_stop_sending_callback(), 1, &err); } -void Stream::EmitWantTrailers() { - // state()->wants_trailers will be set from the javascript side if the - // stream object has a handler for the trailers event. - if (!env()->can_call_into_js() || !state()->wants_trailers) { - return; - } - CallbackScope cb_scope(this); - MakeCallback(BindingData::Get(env()).stream_trailers_callback(), 0, nullptr); -} - // ============================================================================ void Stream::Schedule(Queue* queue) { diff --git a/src/quic/streams.h b/src/quic/streams.h index cd4849aae27..1ea85f21652 100644 --- a/src/quic/streams.h +++ b/src/quic/streams.h @@ -10,7 +10,6 @@ #include #include #include -#include #include #include "bindingdata.h" #include "data.h" @@ -22,6 +21,14 @@ namespace node::quic { class Session; class Stream; +// Optional per-stream state owned by the negotiated application protocol. +// Stream deliberately treats this as opaque so protocol semantics do not +// become part of the transport stream abstraction. +class StreamApplicationState { + public: + virtual ~StreamApplicationState() = default; +}; + // An elastic ring buffer used by Stream to coalesce received data before // flushing it into the DataQueue. This avoids creating many small V8 // BackingStore allocations from per-QUIC-frame ngtcp2 callbacks. Data is @@ -179,11 +186,6 @@ class PendingStream final { // that the stream is gone. Any data that has already been received and is in // the inbound queue is preserved and may be read by the application. // -// QUIC streams in general do not have headers. Some QUIC applications, however, -// may associate headers with the stream (HTTP/3 for instance). As a -// convenience, the Stream class will hold onto these headers for the -// application. -// // Streams may be created in a pending state. This means that while the Stream // object is created, it has not yet been opened in ngtcp2 and therefore has // no official status yet. Certain operations can still be performed on the @@ -202,8 +204,6 @@ class Stream final : public AsyncWrap, public Ngtcp2Source, public DataQueue::BackpressureListener { public: - using Header = NgHeaderBase; - // Acquire a DataQueue from the given value if it is valid. The return // follows the typical V8 rules for Maybe types. If an error occurs, // the Maybe will be empty and an exception will be set on the isolate. @@ -263,6 +263,10 @@ class Stream final : public AsyncWrap, // otherwise falls back to created_at. Returns 0 if neither is set. uint64_t last_activity_timestamp() const; + // Records protocol-level receive activity that does not pass through + // ReceiveData(), such as application framing metadata. + void RecordReceivedActivity(); + // True if this stream was created in a pending state and is still waiting // to be created. bool is_pending() const; @@ -276,9 +280,6 @@ class Stream final : public AsyncWrap, // data to be acknowledged by the remote peer. bool is_eos() const; - // True if the stream wants to send trailing headers after the body. - bool wants_trailers() const; - // Marks this stream as having received 0-RTT early data. void set_early(); @@ -297,6 +298,17 @@ class Stream final : public AsyncWrap, // Returns the Blob::Reader for the inbound data, or nullptr. Blob::Reader* reader() const; + StreamApplicationState* application_state() const { + return application_state_.get(); + } + void set_application_state( + std::unique_ptr application_state) { + application_state_ = std::move(application_state); + } + + StreamPriority priority() const { return priority_.priority; } + StreamPriorityFlags priority_flags() const { return priority_.flags; } + // Called by the session/application to indicate that the specified number // of bytes have been acknowledged by the peer. void Acknowledge(size_t datalen); @@ -308,6 +320,10 @@ class Stream final : public AsyncWrap, // acknowledged to have been received by the peer. void Commit(size_t datalen, bool fin = false); + // Updates the write_desired_size state field based on current flow control + // and outbound buffer state. Emits drain if transitioning from 0 to > 0. + void UpdateWriteDesiredSize(); + void EndWritable(); void EndReadable(std::optional maybe_final_size = std::nullopt); void EntryRead(size_t amount) override; @@ -355,18 +371,8 @@ class Stream final : public AsyncWrap, // that has already been received is still readable. void SendStopSending(error_code code); - // Currently, only HTTP/3 streams support headers. These methods are here - // to support that. They are not used when using any other QUIC application. - - void BeginHeaders(HeadersKind kind); - void set_headers_kind(HeadersKind kind); - // Returns false if the header cannot be added. This will typically happen - // if the application does not support headers, a maximum number of headers - // have already been added, or the maximum total header length is reached. - bool AddHeader(std::unique_ptr
header); - // TODO(@jasnell): Implement MemoryInfo to track outbound_, inbound_, - // reader_, headers_, and pending_headers_queue_. + // reader_, and application_state_. SET_NO_MEMORY_INFO() SET_MEMORY_INFO_NAME(Stream) SET_SELF_SIZE(Stream) @@ -387,7 +393,6 @@ class Stream final : public AsyncWrap, private: struct Impl; - struct PendingHeaders; class Outbound; @@ -437,11 +442,6 @@ class Stream final : public AsyncWrap, // Notifies the JavaScript side that the peer asked it to stop sending. void EmitStopSending(const QuicError& error); - // Notifies the JavaScript side that the application is ready to receive - // trailing headers. Any trailing headers must be sent immediately, and - // synchronously when this callback is triggered. - void EmitWantTrailers(); - // Notifies the JavaScript side that sending data on the stream has been // blocked because of flow control restriction. void EmitBlocked(); @@ -450,23 +450,12 @@ class Stream final : public AsyncWrap, // for more data. Fires when write_desired_size transitions from 0 to > 0. void EmitDrain(); - // Updates the write_desired_size state field based on current flow control - // and outbound buffer state. Emits drain if transitioning from 0 to > 0. - void UpdateWriteDesiredSize(); - - // Delivers the set of inbound headers that have been collected. - void EmitHeaders(); - void NotifyReadableEnded(error_code code); void NotifyWritableEnded(error_code code); // When a pending stream is finally opened, the NotifyStreamOpened method // will be called and the id will be assigned. void NotifyStreamOpened(stream_id id); - void EnqueuePendingHeaders(HeadersKind kind, - v8::Local headers, - HeadersFlags flags); - ArenaSlotBase stats_slot_; ArenaSlotBase state_slot_; BaseObjectWeakPtr session_; @@ -474,6 +463,7 @@ class Stream final : public AsyncWrap, std::shared_ptr inbound_; BaseObjectWeakPtr reader_; std::unique_ptr recv_accumulator_; + std::unique_ptr application_state_; // Bytes delivered to ReceiveData() that still hold inbound flow control // credit. Returned incrementally as the consumer reads them, and in bulk @@ -488,7 +478,6 @@ class Stream final : public AsyncWrap, // and the stream id will be assigned. std::optional> maybe_pending_stream_ = std::nullopt; - std::vector> pending_headers_queue_; error_code pending_close_read_code_ = 0; error_code pending_close_write_code_ = 0; @@ -499,26 +488,8 @@ class Stream final : public AsyncWrap, }; StoredPriority priority_; - const StoredPriority& stored_priority() const { return priority_; } - - // The headers_ field holds a block of headers that have been received and - // are being buffered for delivery to the JavaScript side. Headers are - // stored as C++ objects during collection (AddHeader) and converted to - // V8 strings only when emitted (EmitHeaders), avoiding StrongRootAllocator - // mutex contention on the per-header hot path. - std::vector> headers_; - - // The headers_kind_ field indicates the kind of headers that are being - // buffered. - HeadersKind headers_kind_ = HeadersKind::INITIAL; - - // The headers_length_ field holds the total length of the headers that have - // been buffered. - size_t headers_length_ = 0; - friend struct Impl; friend class PendingStream; - friend class Http3ApplicationImpl; friend class DefaultApplication; public: diff --git a/test/parallel/test-quic-h3-header-interest.mjs b/test/parallel/test-quic-h3-header-interest.mjs new file mode 100644 index 00000000000..4fe05b6f6d3 --- /dev/null +++ b/test/parallel/test-quic-h3-header-interest.mjs @@ -0,0 +1,76 @@ +// Flags: --experimental-quic --experimental-stream-iter --no-warnings + +// Verify HTTP/3 header interest is tracked independently of onheaders and +// that pre-set trailing headers keep the response open until they are sent. + +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 { createPrivateKey } = await import('node:crypto'); +const { listen, connect } = await import('node:quic'); +const { bytes } = await import('stream/iter'); + +const key = createPrivateKey(fixtures.readKey('agent1-key.pem')); +const cert = fixtures.readKey('agent1-cert.pem'); +const encoder = new TextEncoder(); +const decoder = new TextDecoder(); +const serverDone = Promise.withResolvers(); + +const serverEndpoint = await listen(mustCall(async (serverSession) => { + serverSession.onstream = mustCall(async (stream) => { + await stream.closed; + serverSession.close(); + serverDone.resolve(); + }); +}), { + sni: { '*': { keys: [key], certs: [cert] } }, + onheaders: mustCall(function() { + this.sendInformationalHeaders({ + ':status': '103', + 'link': '; rel=preload', + }); + this.sendHeaders({ ':status': '200' }); + this.pendingTrailers = { 'x-checksum': 'abc123' }; + const writer = this.writer; + writer.writeSync(encoder.encode('body')); + writer.endSync(); + }), +}); + +const clientSession = await connect(serverEndpoint.address, { + servername: 'localhost', + verifyPeer: 'manual', +}); +await clientSession.opened; + +const infoReceived = Promise.withResolvers(); +const trailersReceived = Promise.withResolvers(); +const stream = await clientSession.createBidirectionalStream({ + headers: { + ':method': 'GET', + ':path': '/', + ':scheme': 'https', + ':authority': 'localhost', + }, + oninfo: mustCall((headers) => { + assert.strictEqual(headers[':status'], 103); + infoReceived.resolve(); + }), + ontrailers: mustCall((headers) => { + assert.strictEqual(headers['x-checksum'], 'abc123'); + trailersReceived.resolve(); + }), +}); + +assert.strictEqual(decoder.decode(await bytes(stream)), 'body'); +await Promise.all([infoReceived.promise, trailersReceived.promise]); +assert.strictEqual(stream.headers[':status'], 200); + +await Promise.all([stream.closed, serverDone.promise]); +await clientSession.close(); +await serverEndpoint.close();