quic: make Applications optional, with raw stream support in sessions

node:quic no longer requires an Application. When none is set, the
session schedules streams on its own send queue and pulls/commits
their data directly. This lets us drop DefaultApplication, to
support the imminent HTTP 3 app changes coming.

The Session also gains the data-plane dispatchers
(GetStreamData/StreamCommit/ReceiveStream*/ScheduleStream) that route
to an installed application of present or a native path otherwise,
with the ngtcp2 callbacks guarding on whether an application is
installed. HTTP/3 is unchanged (for now).

Author: pimterry

src/quic/application.cc

@@ -295,221 +295,6 @@ void Session::Application::ReceiveStreamReset(Stream* stream,
   stream->ReceiveStreamReset(final_size, std::move(error));
 }
 
-// ============================================================================
-// The DefaultApplication is the default implementation of Session::Application
-// that is used for all unrecognized ALPN identifiers.
-class DefaultApplication final : public Session::Application {
- public:
-  // Marked NOLINT because the cpp linter gets confused about this using
-  // statement not being sorted with the using v8 statements at the top
-  // of the namespace.
-  DefaultApplication(Session* session, const Options& options)
-      : Session::Application(session, options), options_(options) {}
-
-  const Options& options() const override { return options_; }
-
-  Session::Application::Type type() const override {
-    return Session::Application::Type::DEFAULT;
-  }
-
-  error_code GetNoErrorCode() const override { return 0; }
-
-  // Raw QUIC has no application-defined "general failure" code, so
-  // fall back to the QUIC transport-level INTERNAL_ERROR (0x1) used
-  // by ngtcp2 for unspecified failures.
-  error_code GetInternalErrorCode() const override {
-    return NGTCP2_INTERNAL_ERROR;
-  }
-
-  void EarlyDataRejected() override {
-    // Destroy all open streams — ngtcp2 has already discarded their
-    // internal state when it rejected the early data. Use the
-    // application's internal error code since this is an error
-    // condition (code 0 would be treated as a clean close).
-    session().DestroyAllStreams(
-        QuicError::ForApplication(GetInternalErrorCode()));
-    if (!session().is_destroyed()) {
-      session().EmitEarlyDataRejected();
-    }
-  }
-
-  void CollectSessionTicketAppData(
-      SessionTicket::AppData* app_data) const override {
-    const auto& atd = session().config().options.app_ticket_data;
-    if (!atd.has_value() || atd->length() == 0) {
-      // No app data configured — write just the type byte (base behaviour).
-      Session::Application::CollectSessionTicketAppData(app_data);
-      return;
-    }
-    // Layout: [type byte][opaque app data].
-    uv_buf_t bytes = *atd;
-    std::vector<uint8_t> buf;
-    buf.reserve(1 + bytes.len);
-    buf.push_back(static_cast<uint8_t>(type()));  // Type::DEFAULT
-    const auto* p = reinterpret_cast<const uint8_t*>(bytes.base);
-    buf.insert(buf.end(), p, p + bytes.len);
-    app_data->Set(
-        uv_buf_init(reinterpret_cast<char*>(buf.data()), buf.size()));
-  }
-
-  bool ApplySessionTicketData(const PendingTicketAppData& data) override {
-    return std::holds_alternative<DefaultTicketData>(data);
-  }
-
-  bool ReceiveStreamOpen(stream_id id) override {
-    auto stream = session().CreateStream(id);
-    if (!stream || session().is_destroyed()) [[unlikely]] {
-      return !session().is_destroyed();
-    }
-    return true;
-  }
-
-  bool ReceiveStreamData(stream_id id,
-                         const uint8_t* data,
-                         size_t datalen,
-                         const Stream::ReceiveDataFlags& flags,
-                         void* stream_user_data) override {
-    BaseObjectPtr<Stream> stream;
-    if (stream_user_data == nullptr) {
-      // This is the first time we're seeing this stream. Implicitly create it.
-      stream = session().CreateStream(id);
-      if (!stream || session().is_destroyed()) [[unlikely]] {
-        // We couldn't create the stream, or the session was destroyed
-        // during the onstream callback (via MakeCallback re-entrancy).
-        return false;
-      }
-    } else {
-      stream = BaseObjectPtr<Stream>(Stream::From(stream_user_data));
-      if (!stream) {
-        Debug(&session(),
-              "Default application failed to get existing stream "
-              "from user data");
-        return false;
-      }
-    }
-
-    CHECK(stream);
-
-    // Now we can actually receive the data! Woo!
-    stream->ReceiveData(data, datalen, flags);
-    return true;
-  }
-
-  int GetStreamData(StreamData* stream_data) override {
-    // Reset the state of stream_data before proceeding...
-    stream_data->id = -1;
-    stream_data->count = 0;
-    stream_data->fin = 0;
-    stream_data->stream.reset();
-    Debug(&session(), "Default application getting stream data");
-    DCHECK_NOT_NULL(stream_data);
-    // If the queue is empty, there aren't any streams with data yet
-
-    // If the connection-level flow control window is exhausted,
-    // there is no point in pulling stream data.
-    if (!session().max_data_left()) return 0;
-    if (stream_queue_.IsEmpty()) return 0;
-
-    Stream* stream = stream_queue_.PopFront();
-    CHECK_NOT_NULL(stream);
-    stream_data->stream.reset(stream);
-    stream_data->id = stream->id();
-    auto next =
-        [&](int status, const ngtcp2_vec* data, size_t count, bob::Done done) {
-          switch (status) {
-            case bob::Status::STATUS_BLOCK:
-              // Fall through
-            case bob::Status::STATUS_WAIT:
-              return;
-            case bob::Status::STATUS_EOS:
-              stream_data->fin = 1;
-          }
-
-          // It is possible that the data pointers returned are not actually
-          // the data pointers in the stream_data. If that's the case, we need
-          // to copy over the pointers.
-          count = std::min(count, kMaxVectorCount);
-          ngtcp2_vec* dest = *stream_data;
-          if (dest != data) {
-            for (size_t n = 0; n < count; n++) {
-              dest[n] = data[n];
-            }
-          }
-
-          stream_data->count = count;
-
-          if (count > 0) {
-            stream->Schedule(&stream_queue_);
-          }
-
-          // Not calling done here because we defer committing
-          // the data until after we're sure it's written.
-        };
-
-    if (!stream->is_eos()) [[likely]] {
-      int ret = stream->Pull(std::move(next),
-                             bob::Options::OPTIONS_SYNC,
-                             stream_data->data,
-                             arraysize(stream_data->data),
-                             kMaxVectorCount);
-      if (ret == bob::Status::STATUS_EOS) {
-        stream_data->fin = 1;
-      }
-    } else {
-      stream_data->fin = 1;
-    }
-
-    return 0;
-  }
-
-  void ResumeStream(stream_id id) override { ScheduleStream(id); }
-
-  void BlockStream(stream_id id) override {
-    if (auto stream = session().FindStream(id)) [[likely]] {
-      // Remove the stream from the send queue. It will be re-scheduled
-      // via ExtendMaxStreamData when the peer grants more flow control.
-      // Without this, SendPendingData would repeatedly pop and retry
-      // the same blocked stream in an infinite loop.
-      stream->Unschedule();
-      stream->EmitBlocked();
-    }
-  }
-
-  void ExtendMaxStreamData(Stream* stream, uint64_t max_data) override {
-    // The peer granted more flow control for this stream. Re-schedule
-    // it so SendPendingData will resume writing.
-    DCHECK_NOT_NULL(stream);
-    stream->Schedule(&stream_queue_);
-  }
-
-  bool StreamCommit(StreamData* stream_data, size_t datalen) override {
-    DCHECK_NOT_NULL(stream_data);
-    CHECK(stream_data->stream);
-    stream_data->stream->Commit(datalen, stream_data->fin);
-    return true;
-  }
-
-  SET_SELF_SIZE(DefaultApplication)
-  SET_MEMORY_INFO_NAME(DefaultApplication)
-  SET_NO_MEMORY_INFO()
-
- private:
-  void ScheduleStream(stream_id id) {
-    if (auto stream = session().FindStream(id)) [[likely]] {
-      stream->Schedule(&stream_queue_);
-    }
-  }
-
-  Options options_;
-
-  Stream::Queue stream_queue_;
-};
-
-std::unique_ptr<Session::Application> CreateDefaultApplication(
-    Session* session, const Session::Application_Options& options) {
-  return std::make_unique<DefaultApplication>(session, options);
-}
-
 }  // namespace quic
 }  // namespace node