quic: move HTTP/3 event emit logic into Http3Application

Signed-off-by: Tim Perry <pimterry@gmail.com>

Author: pimterry

src/quic/application.h

@@ -36,7 +36,7 @@ class Session::Application : public MemoryRetainer {
   // data owned by the matching application's ticket hooks.
   enum class Type : uint8_t {
     DEFAULT = 1,  // Native opaque byte-match data (no application)
-    HTTP3 = 2,    // Http3Conn typed settings data
+    HTTP3 = 2,    // Http3Application typed settings data
   };
   virtual Type type() const = 0;
 

src/quic/http3.cc

@@ -151,7 +151,7 @@ Http3Settings::operator const nghttp3_settings() const {
       .enable_connect_protocol = enable_connect_protocol,
       .h3_datagram = enable_datagrams,
       // origin_list is nullptr here because it is set directly on the
-      // nghttp3_settings in Http3ApplicationImpl::InitializeConnection()
+      // nghttp3_settings in Http3Application::InitializeConnection()
       // from the SNI configuration.
       .origin_list = nullptr,
       .glitch_ratelim_burst = 1000,
@@ -320,9 +320,9 @@ using Http3Header = NgHeader<Http3HeaderTraits>;
 
 // The Session::Application implementation for HTTP/3, which owns the nghttp3
 // connection and all HTTP/3 protocol logic and state.
-class Http3ApplicationImpl final : public Session::Application {
+class Http3Application final : public Session::Application {
  public:
-  Http3ApplicationImpl(Session* session, const Http3Settings& settings)
+  Http3Application(Session* session, const Http3Settings& settings)
       : Application(session),
         allocator_(BindingData::Get(session->env()).nghttp3_allocator()),
         options_(settings),
@@ -337,6 +337,58 @@ class Http3ApplicationImpl final : public Session::Application {
     session->set_priority_supported();
   }
 
+  // ==========================================================================
+  // HTTP/3 events reported up to the session/JS layer. Each routes through
+  // Session::DeferOrRun, so an event raised before the session is surfaced to
+  // JS (in 0-RTT first-flight frames) is held and replayed after attach.
+
+  void EmitGoaway(stream_id last_stream_id) {
+    session().DeferOrRun([this, last_stream_id]() {
+      Session& s = session();
+      if (s.is_destroyed() || !s.env()->can_call_into_js()) return;
+      CallbackScope<Session> cb_scope(&s);
+      Local<Value> argv[] = {BigInt::New(s.env()->isolate(), last_stream_id)};
+      s.MakeCallback(BindingData::Get(s.env()).session_goaway_callback(),
+                     arraysize(argv),
+                     argv);
+    });
+  }
+
+  void EmitOrigins(std::vector<std::string>&& origins) {
+    auto held = std::make_shared<std::vector<std::string>>(std::move(origins));
+    session().DeferOrRun([this, held]() {
+      Session& s = session();
+      if (s.is_destroyed() || !s.env()->can_call_into_js()) return;
+      if (!s.has_origin_listener()) return;
+      CallbackScope<Session> cb_scope(&s);
+      auto* isolate = s.env()->isolate();
+      LocalVector<Value> elements(isolate, held->size());
+      for (size_t i = 0; i < held->size(); i++) {
+        Local<Value> str;
+        if (!ToV8Value(s.env()->context(), (*held)[i]).ToLocal(&str))
+            [[unlikely]] {
+          return;
+        }
+        elements[i] = str;
+      }
+      Local<Value> argv[] = {
+          Array::New(isolate, elements.data(), elements.size())};
+      s.MakeCallback(BindingData::Get(s.env()).session_origin_callback(),
+                     arraysize(argv),
+                     argv);
+    });
+  }
+
+  void EmitApplicationSettings() {
+    session().DeferOrRun([this]() {
+      Session& s = session();
+      if (s.is_destroyed() || !s.env()->can_call_into_js()) return;
+      CallbackScope<Session> cb_scope(&s);
+      s.MakeCallback(
+          BindingData::Get(s.env()).session_application_callback(), 0, nullptr);
+    });
+  }
+
   // ==========================================================================
   // Session::Application
 
@@ -471,9 +523,9 @@ class Http3ApplicationImpl final : public Session::Application {
     stream_id emit_id = is_notice ? -1 : goaway_id;
 
     if (!is_notice) {
-      // Final GOAWAY: destroy client-initiated bidi streams with
-      // IDs > goaway_id. These were not processed by the peer and
-      // can be retried. Copy the map because Destroy modifies it.
+      // Final GOAWAY: destroy client-initiated bidi streams with IDs >
+      // goaway_id. These were not processed by the peer and can be retried.
+      // Copy the map because Destroy() modifies it.
       auto streams = session().streams();
       for (auto& [id, stream] : streams) {
         if (session().is_destroyed()) return;
@@ -495,7 +547,7 @@ class Http3ApplicationImpl final : public Session::Application {
     // when the peer sends CONNECTION_CLOSE after all streams finish.
     // Calling Close(GRACEFUL) would send a GOAWAY back and trigger
     // BeginShutdown, which can interfere with in-progress streams.
-    session().EmitGoaway(emit_id);
+    EmitGoaway(emit_id);
   }
 
   bool ReceiveStreamData(stream_id id,
@@ -635,7 +687,7 @@ class Http3ApplicationImpl final : public Session::Application {
     }
 
     int rv = nghttp3_conn_close_stream(*this, stream->id(), code);
-    // If the call is successful, the Http3ApplicationImpl::OnStreamClose
+    // If the call is successful, the 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;
@@ -880,8 +932,8 @@ class Http3ApplicationImpl final : public Session::Application {
   }
 
   SET_NO_MEMORY_INFO()
-  SET_MEMORY_INFO_NAME(Http3ApplicationImpl)
-  SET_SELF_SIZE(Http3ApplicationImpl)
+  SET_MEMORY_INFO_NAME(Http3Application)
+  SET_SELF_SIZE(Http3Application)
 
  private:
   inline operator nghttp3_conn*() const {
@@ -1168,8 +1220,8 @@ class Http3ApplicationImpl final : public Session::Application {
     Debug(&session(),
           "HTTP/3 application received updated settings: %s",
           options_);
-    // Report the negotiated settings up to the JS application layer.
-    session().EmitApplication();
+    // Report the negotiated settings up to the session/JS layer.
+    EmitApplicationSettings();
   }
 
   // Inbound header-block accumulation, keyed by stream id. Entries are
@@ -1210,9 +1262,9 @@ class Http3ApplicationImpl final : public Session::Application {
   // ==========================================================================
   // Static callbacks
 
-  static Http3ApplicationImpl* From(nghttp3_conn* conn, void* user_data) {
+  static Http3Application* From(nghttp3_conn* conn, void* user_data) {
     DCHECK_NOT_NULL(user_data);
-    auto app = static_cast<Http3ApplicationImpl*>(user_data);
+    auto app = static_cast<Http3Application*>(user_data);
     DCHECK_EQ(conn, app->conn_.get());
     return app;
   }
@@ -1329,7 +1381,7 @@ class Http3ApplicationImpl final : public Session::Application {
                                   void* conn_user_data,
                                   void* stream_user_data) {
     // This callback is invoked by nghttp3_conn_add_ack_offset() (called
-    // from Http3ApplicationImpl::AcknowledgeStreamData). We must NOT call
+    // from Http3Application::AcknowledgeStreamData). We must NOT call
     // AcknowledgeStreamData here — that would re-enter nghttp3 via
     // nghttp3_conn_add_ack_offset, triggering the NgHttp3CallbackScope
     // re-entrancy assertion. Instead, directly notify the stream that data
@@ -1555,7 +1607,7 @@ class Http3ApplicationImpl final : public Session::Application {
   static int on_end_origin(nghttp3_conn* conn, void* conn_user_data) {
     NGHTTP3_CALLBACK_SCOPE(app);
     if (!app.received_origins_.empty()) {
-      app.session().EmitOrigins(std::move(app.received_origins_));
+      app.EmitOrigins(std::move(app.received_origins_));
       app.received_origins_.clear();
     }
     return NGTCP2_SUCCESS;
@@ -1604,7 +1656,7 @@ Http3Settings ResolveHttp3Settings(const Session::Options& options) {
 
 std::unique_ptr<Session::Application> CreateHttp3Application(Session* session) {
   Debug(session, "Installing HTTP/3 application");
-  return std::make_unique<Http3ApplicationImpl>(
+  return std::make_unique<Http3Application>(
       session, ResolveHttp3Settings(std::as_const(*session).config().options));
 }
 

src/quic/session.cc

@@ -3818,6 +3818,11 @@ void Session::ReplayDeferredEmits() {
   }
 }
 
+bool Session::has_origin_listener() const {
+  return HasListenerFlag(impl_->state()->listener_flags,
+                         SessionListenerFlags::ORIGIN);
+}
+
 void Session::set_priority_supported(bool on) {
   DCHECK(!is_destroyed());
   impl_->state()->priority_supported = on ? 1 : 0;
@@ -4327,21 +4332,6 @@ void Session::set_max_datagram_size(uint16_t size) {
   }
 }
 
-void Session::EmitGoaway(stream_id last_stream_id) {
-  if (is_destroyed()) return;
-  if (DeferEmit([this, last_stream_id] { EmitGoaway(last_stream_id); })) return;
-  if (!env()->can_call_into_js()) return;
-
-  CallbackScope<Session> cb_scope(this);
-
-  Local<Value> argv[] = {
-      BigInt::New(env()->isolate(), last_stream_id),
-  };
-
-  MakeCallback(
-      BindingData::Get(env()).session_goaway_callback(), arraysize(argv), argv);
-}
-
 void Session::EmitDatagram(Store&& datagram, DatagramReceivedFlags flag) {
   DCHECK(!is_destroyed());
 
@@ -4528,20 +4518,6 @@ void Session::EmitSessionTicket(Store&& ticket) {
   }
 }
 
-void Session::EmitApplication() {
-  if (is_destroyed()) return;
-  if (DeferEmit([this] { EmitApplication(); })) return;
-  if (!env()->can_call_into_js()) return;
-
-  // A bare notification that the installed application's negotiated options
-  // have been updated (e.g. an HTTP/3 SETTINGS frame arrived). The consumer
-  // reads the current values back through the application settings binding;
-  // the transport layer carries no protocol-specific data here.
-  CallbackScope<Session> cb_scope(this);
-  MakeCallback(
-      BindingData::Get(env()).session_application_callback(), 0, nullptr);
-}
-
 void Session::DestroyAllStreams(const QuicError& error) {
   DCHECK(!is_destroyed());
   // Copy the streams map since streams remove themselves during
@@ -4644,39 +4620,6 @@ void Session::EmitVersionNegotiation(const ngtcp2_pkt_hd& hd,
                argv);
 }
 
-void Session::EmitOrigins(std::vector<std::string>&& origins) {
-  if (is_destroyed()) return;
-
-  if (must_defer_emits()) {
-    QueueDeferredEmit([this, origins = std::move(origins)]() mutable {
-      EmitOrigins(std::move(origins));
-    });
-    return;
-  }
-
-  if (!HasListenerFlag(impl_->state()->listener_flags,
-                       SessionListenerFlags::ORIGIN))
-    return;
-  if (!env()->can_call_into_js()) return;
-
-  CallbackScope<Session> cb_scope(this);
-
-  auto isolate = env()->isolate();
-
-  LocalVector<Value> elements(env()->isolate(), origins.size());
-  for (size_t i = 0; i < origins.size(); i++) {
-    Local<Value> str;
-    if (!ToV8Value(env()->context(), origins[i]).ToLocal(&str)) [[unlikely]] {
-      return;
-    }
-    elements[i] = str;
-  }
-
-  Local<Value> argv[] = {Array::New(isolate, elements.data(), elements.size())};
-  MakeCallback(
-      BindingData::Get(env()).session_origin_callback(), arraysize(argv), argv);
-}
-
 void Session::EmitKeylog(const char* line) {
   DCHECK(!is_destroyed());
 

src/quic/session.h

@@ -10,8 +10,10 @@
 #include <node_sockaddr.h>
 #include <timer_wrap.h>
 #include <util.h>
+#include <functional>
 #include <memory>
 #include <optional>
+#include <utility>
 #include "bindingdata.h"
 #include "cid.h"
 #include "data.h"
@@ -82,7 +84,6 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
     std::string ToString() const;
   };
 
-
   // Select the Application implementation: the factory registered under
   // options.application when set (see application.h), otherwise nullptr.
   // With no application requested the session runs the native raw-stream
@@ -628,6 +629,19 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
     return true;
   }
 
+  // Runs fn now, or queues it for replay if emits are currently being
+  // deferred (must_defer_emits()) before the session is ready.
+  template <typename F>
+  void DeferOrRun(F&& fn) {
+    if (must_defer_emits()) {
+      QueueDeferredEmit(std::forward<F>(fn));
+    } else {
+      fn();
+    }
+  }
+
+  bool has_origin_listener() const;
+
   enum class CloseMethod : uint8_t {
     // Immediate close with a roundtrip through JavaScript, causing all
     // currently opened streams to be closed. An attempt will be made to
@@ -673,17 +687,15 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
   // JavaScript callouts
 
   void EmitClose(const QuicError& error = QuicError());
-  void EmitGoaway(stream_id last_stream_id);
 
   // Sets the max datagram payload size in the shared state. Used by
-  // Http3Conn to block datagram sends when the peer's
+  // Http3Application to block datagram sends when the peer's
   // SETTINGS_H3_DATAGRAM=0 (RFC 9297 §3).
   void set_max_datagram_size(uint16_t size);
   void EmitDatagram(Store&& datagram, DatagramReceivedFlags flag);
   void EmitDatagramStatus(datagram_id id, DatagramStatus status);
   void EmitHandshakeComplete();
   void EmitKeylog(const char* line);
-  void EmitOrigins(std::vector<std::string>&& origins);
 
   struct ValidatedPath {
     std::shared_ptr<SocketAddress> local;
@@ -702,7 +714,6 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
   void EmitVersionNegotiation(const ngtcp2_pkt_hd& hd,
                               const uint32_t* sv,
                               size_t nsv);
-  void EmitApplication();
   void DatagramStatus(datagram_id datagramId, DatagramStatus status);
   void DatagramReceived(const uint8_t* data,
                         size_t datalen,
@@ -753,7 +764,7 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
   friend struct NgHttp3CallbackScope;
   friend class Application;
   friend class BindingData;
-  friend class Http3ApplicationImpl;
+  friend class Http3Application;
   friend class Endpoint;
   friend class SessionManager;
   friend class Stream;

src/quic/streams.h

@@ -462,7 +462,7 @@ class Stream final : public AsyncWrap,
 
   friend struct Impl;
   friend class PendingStream;
-  friend class Http3ApplicationImpl;
+  friend class Http3Application;
   friend class Session;
 
  public: