githubmirror/node

src/quic/http3.cc
1 014 строк35 KB
James M Snell
quic: support multiple ALPN negotiation
07 май 2026, 19:14
07 май 2026, 19:14abb881e
#if HAVE_OPENSSL && HAVE_QUIC
#include "guard.h"
#ifndef OPENSSL_NO_QUIC
#include <async_wrap-inl.h>
#include <base_object-inl.h>
#include <debug_utils-inl.h>
#include <env-inl.h>
#include <memory_tracker-inl.h>
#include <nghttp3/nghttp3.h>
#include <ngtcp2/ngtcp2.h>
#include <node_http_common-inl.h>
#include <node_sockaddr-inl.h>
#include <util-inl.h>
#include "application.h"
#include "bindingdata.h"
#include "defs.h"
#include "http3.h"
#include "session.h"
#include "sessionticket.h"

namespace node {

using v8::Array;
using v8::Local;

namespace quic {

struct Http3HeadersTraits {
  using nv_t = nghttp3_nv;
};

struct Http3RcBufferPointerTraits {
  using rcbuf_t = nghttp3_rcbuf;
  using vector_t = nghttp3_vec;

  static void inc(rcbuf_t* buf) {
    CHECK_NOT_NULL(buf);
    nghttp3_rcbuf_incref(buf);
  }
  static void dec(rcbuf_t* buf) {
    CHECK_NOT_NULL(buf);
    nghttp3_rcbuf_decref(buf);
  }
  static vector_t get_vec(rcbuf_t* buf) {
    CHECK_NOT_NULL(buf);
    return nghttp3_rcbuf_get_buf(buf);
  }
  static bool is_static(const rcbuf_t* buf) {
    CHECK_NOT_NULL(buf);
    return nghttp3_rcbuf_is_static(buf);
  }
};

using Http3ConnectionPointer = DeleteFnPtr<nghttp3_conn, nghttp3_conn_del>;
using Http3Headers = NgHeaders<Http3HeadersTraits>;
using Http3RcBufferPointer = NgRcBufPointer<Http3RcBufferPointerTraits>;

struct Http3HeaderTraits {
  typedef Http3RcBufferPointer rcbufferpointer_t;
  typedef BindingData allocator_t;

  static const char* ToHttpHeaderName(int32_t token) {
    switch (token) {
      case -1:
        return nullptr;
#define V(key, name)                                                           \
  case NGHTTP3_QPACK_TOKEN__##key:                                             \
    return name;
        HTTP_SPECIAL_HEADERS(V)
#undef V
#define V(key, name)                                                           \
  case NGHTTP3_QPACK_TOKEN_##key:                                              \
    return name;
        HTTP_REGULAR_HEADERS(V)
#undef V
    }
    return nullptr;
  }
};

using Http3Header = NgHeader<Http3HeaderTraits>;

// Implements the low-level HTTP/3 Application semantics.
class Http3ApplicationImpl final : public Session::Application {
 public:
  Http3ApplicationImpl(Session* session, const Options& options)
      : Application(session, options),
        allocator_(BindingData::Get(env())),
        options_(options),
        conn_(InitializeConnection()) {
    session->set_priority_supported();
  }

  Session::Application::Type type() const override {
    return Session::Application::Type::HTTP3;
  }

  error_code GetNoErrorCode() const override { return NGHTTP3_H3_NO_ERROR; }

  bool Start() override {
    CHECK(!started_);
    started_ = true;
    Debug(&session(), "Starting HTTP/3 application.");

    auto params = ngtcp2_conn_get_remote_transport_params(session());
    if (params == nullptr) [[unlikely]] {
      // The params are not available yet. Cannot start.
      Debug(&session(),
            "Cannot start HTTP/3 application yet. No remote transport params");
      return false;
    }

    if (params->initial_max_streams_uni < 3) {
      // HTTP3 requires 3 unidirectional control streams to be opened in each
      // direction in additional to the bidirectional streams that are used to
      // actually carry request and response payload back and forth.
      // See:
      // https://nghttp2.org/nghttp3/programmers-guide.html#binding-control-streams
      Debug(&session(),
            "Cannot start HTTP/3 application. Initial max "
            "unidirectional streams [%zu] is too low. Must be at least 3",
            params->initial_max_streams_uni);
      return false;
    }

    // If this is a server session, then set the maximum number of
    // bidirectional streams that can be created. This determines the number
    // of requests that the client can actually created.
    if (session().is_server()) {
      nghttp3_conn_set_max_client_streams_bidi(
          *this, params->initial_max_streams_bidi);
    }

    Debug(&session(), "Creating and binding HTTP/3 control streams");
    bool ret =
        ngtcp2_conn_open_uni_stream(session(), &control_stream_id_, nullptr) ==
            0 &&
        ngtcp2_conn_open_uni_stream(
            session(), &qpack_enc_stream_id_, nullptr) == 0 &&
        ngtcp2_conn_open_uni_stream(
            session(), &qpack_dec_stream_id_, nullptr) == 0 &&
        nghttp3_conn_bind_control_stream(*this, control_stream_id_) == 0 &&
        nghttp3_conn_bind_qpack_streams(
            *this, qpack_enc_stream_id_, qpack_dec_stream_id_) == 0;

    if (env()->enabled_debug_list()->enabled(DebugCategory::QUIC) && ret) {
      Debug(&session(),
            "Created and bound control stream %" PRIi64,
            control_stream_id_);
      Debug(&session(),
            "Created and bound qpack enc stream %" PRIi64,
            qpack_enc_stream_id_);
      Debug(&session(),
            "Created and bound qpack dec streams %" PRIi64,
            qpack_dec_stream_id_);
    }

    return ret;
  }

  bool ReceiveStreamData(int64_t stream_id,
                         const uint8_t* data,
                         size_t datalen,
                         const Stream::ReceiveDataFlags& flags,
                         void* unused) override {
    Debug(&session(),
          "HTTP/3 application received %zu bytes of data "
          "on stream %" PRIi64 ". Is final? %d",
          datalen,
          stream_id,
          flags.fin);

    ssize_t nread = nghttp3_conn_read_stream(
        *this, stream_id, data, datalen, flags.fin ? 1 : 0);

    if (nread < 0) {
      Debug(&session(),
            "HTTP/3 application failed to read stream data: %s",
            nghttp3_strerror(nread));
      return false;
    }

    if (nread > 0) {
      Debug(&session(),
            "Extending stream and connection offset by %zd bytes",
            nread);
      session().ExtendStreamOffset(stream_id, nread);
      session().ExtendOffset(nread);
    }

    return true;
  }

  bool AcknowledgeStreamData(int64_t stream_id, size_t datalen) override {
    Debug(&session(),
          "HTTP/3 application received acknowledgement for %zu bytes of data "
          "on stream %" PRIi64,
          datalen,
          stream_id);
    return nghttp3_conn_add_ack_offset(*this, stream_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
    return (current_count < options_.max_header_pairs) &&
           (current_headers_length + this_header_length) <=
               options_.max_header_length;
  }

  void BlockStream(int64_t id) override {
    nghttp3_conn_block_stream(*this, id);
    Application::BlockStream(id);
  }

  void ResumeStream(int64_t id) override {
    nghttp3_conn_resume_stream(*this, id);
    Application::ResumeStream(id);
  }

  void ExtendMaxStreams(EndpointLabel label,
                        Direction direction,
                        uint64_t max_streams) override {
    switch (label) {
      case EndpointLabel::LOCAL:
        return;
      case EndpointLabel::REMOTE: {
        switch (direction) {
          case Direction::BIDIRECTIONAL: {
            Debug(&session(),
                  "HTTP/3 application extending max bidi streams by %" PRIu64,
                  max_streams);
            ngtcp2_conn_extend_max_streams_bidi(
                session(), static_cast<size_t>(max_streams));
            break;
          }
          case Direction::UNIDIRECTIONAL: {
            Debug(&session(),
                  "HTTP/3 application extending max uni streams by %" PRIu64,
                  max_streams);
            ngtcp2_conn_extend_max_streams_uni(
                session(), static_cast<size_t>(max_streams));
            break;
          }
        }
      }
    }
  }

  void ExtendMaxStreamData(Stream* stream, uint64_t max_data) override {
    Debug(&session(),
          "HTTP/3 application extending max stream data to %" PRIu64,
          max_data);
    nghttp3_conn_unblock_stream(*this, stream->id());
  }

  void CollectSessionTicketAppData(
      SessionTicket::AppData* app_data) const override {
    // TODO(@jasnell): When HTTP/3 settings become dynamic or
    // configurable per-connection, store them here so they can be
    // validated on 0-RTT resumption. Candidates include:
    // max_field_section_size, qpack_max_dtable_capacity,
    // qpack_encoder_max_dtable_capacity, qpack_blocked_streams,
    // enable_connect_protocol, and enable_datagrams. On extraction,
    // compare stored values against current settings and return
    // TICKET_IGNORE_RENEW if incompatible.
  }

  SessionTicket::AppData::Status ExtractSessionTicketAppData(
      const SessionTicket::AppData& app_data,
      SessionTicket::AppData::Source::Flag flag) override {
    // See CollectSessionTicketAppData above.
    return flag == SessionTicket::AppData::Source::Flag::STATUS_RENEW
               ? SessionTicket::AppData::Status::TICKET_USE_RENEW
               : SessionTicket::AppData::Status::TICKET_USE;
  }

  void StreamClose(Stream* stream, QuicError&& error = QuicError()) override {
    Debug(
        &session(), "HTTP/3 application closing stream %" PRIi64, stream->id());
    uint64_t code = NGHTTP3_H3_NO_ERROR;
    if (error) {
      CHECK_EQ(error.type(), QuicError::Type::APPLICATION);
      code = error.code();
    }

    int rv = nghttp3_conn_close_stream(*this, stream->id(), code);
    // 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;
    }

    session().SetLastError(
        QuicError::ForApplication(nghttp3_err_infer_quic_app_error_code(rv)));
    session().Close();
  }

  void StreamReset(Stream* stream,
                   uint64_t final_size,
                   QuicError&& error = QuicError()) override {
    // We are shutting down the readable side of the local stream here.
    Debug(&session(),
          "HTTP/3 application resetting stream %" PRIi64,
          stream->id());
    int rv = nghttp3_conn_shutdown_stream_read(*this, stream->id());
    if (rv == 0) {
      stream->ReceiveStreamReset(final_size, std::move(error));
      return;
    }

    session().SetLastError(
        QuicError::ForApplication(nghttp3_err_infer_quic_app_error_code(rv)));
    session().Close();
  }

  void StreamStopSending(Stream* stream,
                         QuicError&& error = QuicError()) override {
    Application::StreamStopSending(stream, std::move(error));
  }

  bool SendHeaders(const Stream& stream,
                   HeadersKind kind,
                   const Local<Array>& headers,
                   HeadersFlags flags = HeadersFlags::NONE) override {
    Session::SendPendingDataScope send_scope(&session());
    Http3Headers nva(env(), headers);

    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;
      }
      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;
        }

        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);
        } 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*>(&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;
      }
    }

    return false;
  }

  void SetStreamPriority(const Stream& stream,
                         StreamPriority priority,
                         StreamPriorityFlags flags) override {
    nghttp3_pri pri;
    pri.inc = (flags == StreamPriorityFlags::NON_INCREMENTAL) ? 0 : 1;
    switch (priority) {
      case StreamPriority::HIGH:
        pri.urgency = NGHTTP3_URGENCY_HIGH;
        break;
      case StreamPriority::LOW:
        pri.urgency = NGHTTP3_URGENCY_LOW;
        break;
      default:
        pri.urgency = NGHTTP3_DEFAULT_URGENCY;
        break;
    }
    if (session().is_server()) {
      nghttp3_conn_set_server_stream_priority(*this, stream.id(), &pri);
    }
    // Client-side priority is set at request submission time via
    // nghttp3_conn_submit_request and is not typically changed
    // after the fact. The client API takes a serialized RFC 9218
    // field value rather than an nghttp3_pri struct.
  }

  StreamPriority GetStreamPriority(const Stream& stream) override {
    nghttp3_pri pri;
    if (nghttp3_conn_get_stream_priority(*this, &pri, stream.id()) == 0) {
      // TODO(@jasnell): The nghttp3_pri.inc (incremental) flag is
      // not yet exposed. When priority-based stream scheduling is
      // implemented, GetStreamPriority should return both urgency
      // and the incremental flag (making get/set symmetrical).
      // The inc flag determines whether the server should interleave
      // data from this stream with others of the same urgency
      // (inc=1) or complete it first (inc=0).
      switch (pri.urgency) {
        case NGHTTP3_URGENCY_HIGH:
          return StreamPriority::HIGH;
        case NGHTTP3_URGENCY_LOW:
          return StreamPriority::LOW;
        default:
          return StreamPriority::DEFAULT;
      }
    }
    return StreamPriority::DEFAULT;
  }

  int GetStreamData(StreamData* data) override {
    data->count = kMaxVectorCount;
    ssize_t ret = 0;
    Debug(&session(), "HTTP/3 application getting stream data");
    if (conn_ && session().max_data_left()) {
      ret = nghttp3_conn_writev_stream(
          *this, &data->id, &data->fin, *data, data->count);
      // A negative return value indicates an error.
      if (ret < 0) {
        return static_cast<int>(ret);
      }

      data->count = static_cast<size_t>(ret);
      if (data->id > 0 && data->id != control_stream_id_ &&
          data->id != qpack_dec_stream_id_ &&
          data->id != qpack_enc_stream_id_) {
        data->stream = session().FindStream(data->id);
      }
    }

    return 0;
  }

  bool StreamCommit(StreamData* data, size_t datalen) override {
    Debug(&session(),
          "HTTP/3 application committing stream %" PRIi64 " data %zu",
          data->id,
          datalen);
    int err = nghttp3_conn_add_write_offset(*this, data->id, datalen);
    if (err != 0) {
      session().SetLastError(QuicError::ForApplication(
          nghttp3_err_infer_quic_app_error_code(err)));
      return false;
    }
    if (data->stream) data->stream->Commit(datalen, data->fin);
    return true;
  }

  SET_NO_MEMORY_INFO()
  SET_MEMORY_INFO_NAME(Http3ApplicationImpl)
  SET_SELF_SIZE(Http3ApplicationImpl)

 private:
  inline operator nghttp3_conn*() const {
    DCHECK_NOT_NULL(conn_.get());
    return conn_.get();
  }

  inline bool is_control_stream(int64_t id) const {
    return id == control_stream_id_ || id == qpack_dec_stream_id_ ||
           id == qpack_enc_stream_id_;
  }

  Http3ConnectionPointer InitializeConnection() {
    nghttp3_conn* conn = nullptr;
    nghttp3_settings settings = options_;
    if (session().is_server()) {
      CHECK_EQ(nghttp3_conn_server_new(
                   &conn, &kCallbacks, &settings, &allocator_, this),
               0);
    } else {
      CHECK_EQ(nghttp3_conn_client_new(
                   &conn, &kCallbacks, &settings, &allocator_, this),
               0);
    }
    return Http3ConnectionPointer(conn);
  }

  void OnStreamClose(Stream* stream, uint64_t app_error_code) {
    if (app_error_code != NGHTTP3_H3_NO_ERROR) {
      Debug(&session(),
            "HTTP/3 application received stream close for stream %" PRIi64
            " with code %" PRIu64,
            stream->id(),
            app_error_code);
    }
    auto direction = stream->direction();
    stream->Destroy(QuicError::ForApplication(app_error_code));
    ExtendMaxStreams(EndpointLabel::REMOTE, direction, 1);
  }

  void OnBeginHeaders(int64_t stream_id) {
    auto stream = session().FindStream(stream_id);
    // If the stream does not exist or is destroyed, ignore!
    if (!stream) [[unlikely]]
      return;
    Debug(&session(),
          "HTTP/3 application beginning initial block of headers for stream "
          "%" PRIi64,
          stream_id);
    stream->BeginHeaders(HeadersKind::INITIAL);
  }

  void OnReceiveHeader(int64_t stream_id, Http3Header&& header) {
    auto stream = session().FindStream(stream_id);

    if (!stream) [[unlikely]]
      return;
    if (header.name() == ":status" && header.value()[0] == '1') {
      Debug(&session(),
            "HTTP/3 application switching to hints headers for stream %" PRIi64,
            stream->id());
      stream->set_headers_kind(HeadersKind::HINTS);
    }
    IF_QUIC_DEBUG(env()) {
      Debug(&session(),
            "Received header \"%s: %s\"",
            header.name(),
            header.value());
    }
    stream->AddHeader(std::move(header));
  }

  void OnEndHeaders(int64_t stream_id, int fin) {
    auto stream = session().FindStream(stream_id);
    if (!stream) [[unlikely]]
      return;
    Debug(&session(),
          "HTTP/3 application received end of headers for stream %" PRIi64,
          stream_id);
    stream->EmitHeaders();
    if (fin) {
      // The stream is done. There's no more data to receive!
      Debug(&session(), "Headers are final for stream %" PRIi64, stream_id);
      Stream::ReceiveDataFlags flags{
          .fin = true,
          .early = false,
      };
      stream->ReceiveData(nullptr, 0, flags);
    }
  }

  void OnBeginTrailers(int64_t stream_id) {
    auto stream = session().FindStream(stream_id);
    if (!stream) [[unlikely]]
      return;
    Debug(&session(),
          "HTTP/3 application beginning block of trailers for stream %" PRIi64,
          stream_id);
    stream->BeginHeaders(HeadersKind::TRAILING);
  }

  void OnReceiveTrailer(int64_t stream_id, Http3Header&& header) {
    auto stream = session().FindStream(stream_id);
    if (!stream) [[unlikely]]
      return;
    IF_QUIC_DEBUG(env()) {
      Debug(&session(),
            "Received header \"%s: %s\"",
            header.name(),
            header.value());
    }
    stream->AddHeader(header);
  }

  void OnEndTrailers(int64_t stream_id, int fin) {
    auto stream = session().FindStream(stream_id);
    if (!stream) [[unlikely]]
      return;
    Debug(&session(),
          "HTTP/3 application received end of trailers for stream %" PRIi64,
          stream_id);
    stream->EmitHeaders();
    if (fin) {
      Debug(&session(), "Trailers are final for stream %" PRIi64, stream_id);
      Stream::ReceiveDataFlags flags{
          .fin = true,
          .early = false,
      };
      stream->ReceiveData(nullptr, 0, flags);
    }
  }

  void OnEndStream(int64_t stream_id) {
    auto stream = session().FindStream(stream_id);
    if (!stream) [[unlikely]]
      return;
    Debug(&session(),
          "HTTP/3 application received end of stream for stream %" PRIi64,
          stream_id);
    Stream::ReceiveDataFlags flags{
        .fin = true,
        .early = false,
    };
    stream->ReceiveData(nullptr, 0, flags);
  }

  void OnStopSending(int64_t stream_id, uint64_t app_error_code) {
    auto stream = session().FindStream(stream_id);
    if (!stream) [[unlikely]]
      return;
    Debug(&session(),
          "HTTP/3 application received stop sending for stream %" PRIi64,
          stream_id);
    stream->ReceiveStopSending(QuicError::ForApplication(app_error_code));
  }

  void OnResetStream(int64_t stream_id, uint64_t app_error_code) {
    auto stream = session().FindStream(stream_id);
    if (!stream) [[unlikely]]
      return;
    Debug(&session(),
          "HTTP/3 application received reset stream for stream %" PRIi64,
          stream_id);
    stream->ReceiveStreamReset(0, QuicError::ForApplication(app_error_code));
  }

  void OnShutdown(int64_t id) {
    // The peer has sent a GOAWAY frame initiating a graceful shutdown.
    // For a client, id is the stream ID beyond which the server will
    // not process requests. For a server, id is a push ID (server
    // push is not implemented). Streams/pushes with IDs >= id will
    // not be processed by the peer.
    //
    // When id equals NGHTTP3_SHUTDOWN_NOTICE_STREAM_ID (client) or
    // NGHTTP3_SHUTDOWN_NOTICE_PUSH_ID (server), this is a notice of
    // intent to shut down rather than an immediate refusal.
    //
    // This can be called multiple times with a decreasing id as the
    // peer progressively reduces the set of streams it will process.
    Debug(&session(), "HTTP/3 received GOAWAY (id=%" PRIi64 ")", id);
    session().Close(Session::CloseMethod::GRACEFUL);
  }

  void OnReceiveSettings(const nghttp3_settings* settings) {
    options_.enable_connect_protocol = settings->enable_connect_protocol;
    options_.enable_datagrams = settings->h3_datagram;
    options_.max_field_section_size = settings->max_field_section_size;
    options_.qpack_blocked_streams = settings->qpack_blocked_streams;
    options_.qpack_encoder_max_dtable_capacity =
        settings->qpack_encoder_max_dtable_capacity;
    options_.qpack_max_dtable_capacity = settings->qpack_max_dtable_capacity;
    Debug(&session(),
          "HTTP/3 application received updated settings: %s",
          options_);
  }

  bool started_ = false;
  nghttp3_mem allocator_;
  Options options_;
  Http3ConnectionPointer conn_;
  int64_t control_stream_id_ = -1;
  int64_t qpack_dec_stream_id_ = -1;
  int64_t qpack_enc_stream_id_ = -1;

  // ==========================================================================
  // Static callbacks

  static Http3ApplicationImpl* From(nghttp3_conn* conn, void* user_data) {
    DCHECK_NOT_NULL(user_data);
    auto app = static_cast<Http3ApplicationImpl*>(user_data);
    DCHECK_EQ(conn, app->conn_.get());
    return app;
  }

  static BaseObjectWeakPtr<Stream> FindOrCreateStream(nghttp3_conn* conn,
                                                      Session* session,
                                                      int64_t stream_id) {
    if (auto stream = session->FindStream(stream_id)) {
      return stream;
    }
    if (auto stream = session->CreateStream(stream_id)) {
      return stream;
    }
    return {};
  }

#define NGHTTP3_CALLBACK_SCOPE(name)                                           \
  auto ptr = From(conn, conn_user_data);                                       \
  CHECK_NOT_NULL(ptr);                                                         \
  auto& name = *ptr;                                                           \
  NgHttp3CallbackScope scope(name.env());

  static nghttp3_ssize on_read_data_callback(nghttp3_conn* conn,
                                             int64_t stream_id,
                                             nghttp3_vec* vec,
                                             size_t veccnt,
                                             uint32_t* pflags,
                                             void* conn_user_data,
                                             void* stream_user_data) {
    auto ptr = From(conn, conn_user_data);
    CHECK_NOT_NULL(ptr);
    auto& app = *ptr;
    NgHttp3CallbackScope scope(app.env());

    auto stream = app.session().FindStream(stream_id);
    if (!stream) return NGHTTP3_ERR_CALLBACK_FAILURE;

    if (stream->is_eos()) {
      *pflags |= NGHTTP3_DATA_FLAG_EOF;
      return 0;
    }

    size_t max_count = std::min(veccnt, static_cast<size_t>(kMaxVectorCount));
    nghttp3_ssize result = 0;

    auto next =
        [&](int status, const ngtcp2_vec* data, size_t count, bob::Done done) {
          switch (status) {
            case bob::Status::STATUS_BLOCK:
            case bob::Status::STATUS_WAIT:
              result = NGHTTP3_ERR_WOULDBLOCK;
              return;
            case bob::Status::STATUS_EOS:
              *pflags |= NGHTTP3_DATA_FLAG_EOF;
              break;
          }
          count = std::min(count, max_count);
          for (size_t n = 0; n < count; n++) {
            vec[n].base = data[n].base;
            vec[n].len = data[n].len;
          }
          result = static_cast<nghttp3_ssize>(count);
        };

    ngtcp2_vec data[kMaxVectorCount];
    stream->Pull(std::move(next),
                 bob::Options::OPTIONS_SYNC,
                 data,
                 max_count,
                 max_count);

    return result;
  }

  static int on_acked_stream_data(nghttp3_conn* conn,
                                  int64_t stream_id,
                                  uint64_t datalen,
                                  void* conn_user_data,
                                  void* stream_user_data) {
    NGHTTP3_CALLBACK_SCOPE(app);
    return app.AcknowledgeStreamData(stream_id, static_cast<size_t>(datalen))
               ? NGTCP2_SUCCESS
               : NGHTTP3_ERR_CALLBACK_FAILURE;
  }

  static int on_stream_close(nghttp3_conn* conn,
                             int64_t stream_id,
                             uint64_t app_error_code,
                             void* conn_user_data,
                             void* stream_user_data) {
    NGHTTP3_CALLBACK_SCOPE(app);
    if (auto stream = app.session().FindStream(stream_id)) {
      app.OnStreamClose(stream.get(), app_error_code);
    }
    return NGTCP2_SUCCESS;
  }

  static int on_receive_data(nghttp3_conn* conn,
                             int64_t stream_id,
                             const uint8_t* data,
                             size_t datalen,
                             void* conn_user_data,
                             void* stream_user_data) {
    NGHTTP3_CALLBACK_SCOPE(app);
    // The on_receive_data callback will never be called for control streams,
    // so we know that if we get here, the data received is for a stream that
    // we know is for an HTTP payload.
    if (app.is_control_stream(stream_id)) [[unlikely]] {
      return NGHTTP3_ERR_CALLBACK_FAILURE;
    }
    auto& session = app.session();
    if (auto stream = FindOrCreateStream(conn, &session, stream_id))
        [[likely]] {
      stream->ReceiveData(data, datalen, Stream::ReceiveDataFlags{});
      return NGTCP2_SUCCESS;
    }
    return NGHTTP3_ERR_CALLBACK_FAILURE;
  }

  static int on_deferred_consume(nghttp3_conn* conn,
                                 int64_t stream_id,
                                 size_t consumed,
                                 void* conn_user_data,
                                 void* stream_user_data) {
    NGHTTP3_CALLBACK_SCOPE(app);
    auto& session = app.session();
    Debug(&session, "HTTP/3 application deferred consume %zu bytes", consumed);
    session.ExtendStreamOffset(stream_id, consumed);
    session.ExtendOffset(consumed);
    return NGTCP2_SUCCESS;
  }

  static int on_begin_headers(nghttp3_conn* conn,
                              int64_t stream_id,
                              void* conn_user_data,
                              void* stream_user_data) {
    NGHTTP3_CALLBACK_SCOPE(app);
    if (app.is_control_stream(stream_id)) [[unlikely]] {
      return NGHTTP3_ERR_CALLBACK_FAILURE;
    }
    app.OnBeginHeaders(stream_id);
    return NGTCP2_SUCCESS;
  }

  static int on_receive_header(nghttp3_conn* conn,
                               int64_t stream_id,
                               int32_t token,
                               nghttp3_rcbuf* name,
                               nghttp3_rcbuf* value,
                               uint8_t flags,
                               void* conn_user_data,
                               void* stream_user_data) {
    NGHTTP3_CALLBACK_SCOPE(app);
    if (app.is_control_stream(stream_id)) [[unlikely]] {
      return NGHTTP3_ERR_CALLBACK_FAILURE;
    }
    if (Http3Header::IsZeroLength(token, name, value)) return NGTCP2_SUCCESS;
    app.OnReceiveHeader(stream_id,
                        Http3Header(app.env(), token, name, value, flags));
    return NGTCP2_SUCCESS;
  }

  static int on_end_headers(nghttp3_conn* conn,
                            int64_t stream_id,
                            int fin,
                            void* conn_user_data,
                            void* stream_user_data) {
    NGHTTP3_CALLBACK_SCOPE(app);
    if (app.is_control_stream(stream_id)) [[unlikely]] {
      return NGHTTP3_ERR_CALLBACK_FAILURE;
    }
    app.OnEndHeaders(stream_id, fin);
    return NGTCP2_SUCCESS;
  }

  static int on_begin_trailers(nghttp3_conn* conn,
                               int64_t stream_id,
                               void* conn_user_data,
                               void* stream_user_data) {
    NGHTTP3_CALLBACK_SCOPE(app);
    if (app.is_control_stream(stream_id)) [[unlikely]] {
      return NGHTTP3_ERR_CALLBACK_FAILURE;
    }
    app.OnBeginTrailers(stream_id);
    return NGTCP2_SUCCESS;
  }

  static int on_receive_trailer(nghttp3_conn* conn,
                                int64_t stream_id,
                                int32_t token,
                                nghttp3_rcbuf* name,
                                nghttp3_rcbuf* value,
                                uint8_t flags,
                                void* conn_user_data,
                                void* stream_user_data) {
    NGHTTP3_CALLBACK_SCOPE(app);
    if (app.is_control_stream(stream_id)) [[unlikely]] {
      return NGHTTP3_ERR_CALLBACK_FAILURE;
    }
    if (Http3Header::IsZeroLength(token, name, value)) return NGTCP2_SUCCESS;
    app.OnReceiveTrailer(stream_id,
                         Http3Header(app.env(), token, name, value, flags));
    return NGTCP2_SUCCESS;
  }

  static int on_end_trailers(nghttp3_conn* conn,
                             int64_t stream_id,
                             int fin,
                             void* conn_user_data,
                             void* stream_user_data) {
    NGHTTP3_CALLBACK_SCOPE(app);
    if (app.is_control_stream(stream_id)) [[unlikely]] {
      return NGHTTP3_ERR_CALLBACK_FAILURE;
    }
    app.OnEndTrailers(stream_id, fin);
    return NGTCP2_SUCCESS;
  }

  static int on_end_stream(nghttp3_conn* conn,
                           int64_t stream_id,
                           void* conn_user_data,
                           void* stream_user_data) {
    NGHTTP3_CALLBACK_SCOPE(app);
    if (app.is_control_stream(stream_id)) [[unlikely]] {
      return NGHTTP3_ERR_CALLBACK_FAILURE;
    }
    app.OnEndStream(stream_id);
    return NGTCP2_SUCCESS;
  }

  static int on_stop_sending(nghttp3_conn* conn,
                             int64_t stream_id,
                             uint64_t app_error_code,
                             void* conn_user_data,
                             void* stream_user_data) {
    NGHTTP3_CALLBACK_SCOPE(app);
    if (app.is_control_stream(stream_id)) [[unlikely]] {
      return NGHTTP3_ERR_CALLBACK_FAILURE;
    }
    app.OnStopSending(stream_id, app_error_code);
    return NGTCP2_SUCCESS;
  }

  static int on_reset_stream(nghttp3_conn* conn,
                             int64_t stream_id,
                             uint64_t app_error_code,
                             void* conn_user_data,
                             void* stream_user_data) {
    NGHTTP3_CALLBACK_SCOPE(app);
    if (app.is_control_stream(stream_id)) [[unlikely]] {
      return NGHTTP3_ERR_CALLBACK_FAILURE;
    }
    app.OnResetStream(stream_id, app_error_code);
    return NGTCP2_SUCCESS;
  }

  static int on_shutdown(nghttp3_conn* conn, int64_t id, void* conn_user_data) {
    NGHTTP3_CALLBACK_SCOPE(app);
    app.OnShutdown(id);
    return NGTCP2_SUCCESS;
  }

  static int on_receive_settings(nghttp3_conn* conn,
                                 const nghttp3_settings* settings,
                                 void* conn_user_data) {
    NGHTTP3_CALLBACK_SCOPE(app);
    app.OnReceiveSettings(settings);
    return NGTCP2_SUCCESS;
  }

  static int on_receive_origin(nghttp3_conn* conn,
                               const uint8_t* origin,
                               size_t originlen,
                               void* conn_user_data) {
    // ORIGIN frames (RFC 8336) are used for connection coalescing
    // across multiple origins. Not yet implemented u2014 requires
    // connection pooling and multi-origin reuse support.
    return NGTCP2_SUCCESS;
  }

  static int on_end_origin(nghttp3_conn* conn, void* conn_user_data) {
    // See on_receive_origin above.
    return NGTCP2_SUCCESS;
  }

  static void on_rand(uint8_t* dest, size_t destlen) {
    CHECK(ncrypto::CSPRNG(dest, destlen));
  }

  static constexpr nghttp3_callbacks kCallbacks = {on_acked_stream_data,
                                                   on_stream_close,
                                                   on_receive_data,
                                                   on_deferred_consume,
                                                   on_begin_headers,
                                                   on_receive_header,
                                                   on_end_headers,
                                                   on_begin_trailers,
                                                   on_receive_trailer,
                                                   on_end_trailers,
                                                   on_stop_sending,
                                                   on_end_stream,
                                                   on_reset_stream,
                                                   on_shutdown,
                                                   on_receive_settings,
                                                   on_receive_origin,
                                                   on_end_origin,
                                                   on_rand,
                                                   nullptr};
};

std::unique_ptr<Session::Application> CreateHttp3Application(
    Session* session, const Session::Application_Options& options) {
  Debug(session, "Selecting HTTP/3 application");
  return std::make_unique<Http3ApplicationImpl>(session, options);
}

}  // namespace quic
}  // namespace node
#endif  // OPENSSL_NO_QUIC
#endif  // HAVE_OPENSSL && HAVE_QUIC