quic: extract transport logic from Application to Session

One small fix notably included:

- Check is_destroyed() after StreamCommit, since it calls JS callbacks
  which could destroy the session.

Signed-off-by: Tim Perry <pimterry@gmail.com>
PR-URL: #64127
Reviewed-By: James M Snell <jasnell@gmail.com>

Author: pimterry

src/quic/README.md

@@ -77,7 +77,7 @@ data channels that carry application data.
 
 Every entry point that may generate outbound data creates a
 `SendPendingDataScope`. Scopes nest — an internal depth counter ensures
-`Application::SendPendingData()` is called exactly once, when the outermost
+`Session::SendPendingData()` is called exactly once, when the outermost
 scope exits:
 
 ```cpp
@@ -218,13 +218,13 @@ Session::Receive()
 
 ```text
 SendPendingDataScope::~SendPendingDataScope()
- → Application::SendPendingData()
+ → Session::SendPendingData()
     Loop (up to max_packet_count):
-     ├── GetStreamData()           // pull data from next stream
-     │   └── stream->Pull()        // bob pull from Outbound→DataQueue
-     ├── WriteVStream()            // ngtcp2_conn_writev_stream()
+     ├── application().GetStreamData()  // pull data from next stream
+     │   └── stream->Pull()             // bob pull from Outbound→DataQueue
+     ├── WriteVStream()                 // ngtcp2_conn_writev_stream()
      │                              encrypts, frames, paces
-     ├── if ndatalen > 0: StreamCommit()
+     ├── if ndatalen > 0: application().StreamCommit()
      │                              stream->Commit(datalen, fin)
      ├── if nwrite > 0:  Send()    // uv_udp_send()
      ├── if WRITE_MORE:  continue  // room for more in this packet

src/quic/application.cc

@@ -205,10 +205,6 @@ class Session::Application : public MemoryRetainer {
     return false;
   }
 
-  // Signals to the Application that it should serialize and transmit any
-  // pending session and stream packets it has accumulated.
-  void SendPendingData();
-
   // 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).
@@ -243,10 +239,6 @@ class Session::Application : public MemoryRetainer {
     return {StreamPriority::DEFAULT, StreamPriorityFlags::NON_INCREMENTAL};
   }
 
-  // The StreamData struct is used by the application to pass pending stream
-  // data to the session for transmission.
-  struct StreamData;
-
   virtual int GetStreamData(StreamData* data) = 0;
   virtual bool StreamCommit(StreamData* data, size_t datalen) = 0;
 
@@ -262,57 +254,9 @@ class Session::Application : public MemoryRetainer {
   }
 
  private:
-  Packet::Ptr CreateStreamDataPacket();
-
-  // Tries to pack a pending datagram into the current packet buffer.
-  // If < 0 is returned, either NGTCP2_ERR_WRITE_MORE or a fatal error is
-  // returned; the caller must check. If > 0 is returned, the packet is done
-  // and the value is the size of the finalized packet. If 0 is returned,
-  // the datagram is either congestion limited or was abandoned
-  ssize_t TryWritePendingDatagram(PathStorage* path,
-                                  uint8_t* dest,
-                                  size_t destlen,
-                                  uint64_t ts);
-
-  // Write the given stream_data into the buffer. The PacketInfo out-param
-  // is populated by ngtcp2 with per-packet metadata (e.g., ECN codepoint)
-  // that should be applied when sending the packet.
-  ssize_t WriteVStream(PathStorage* path,
-                       PacketInfo* pi,
-                       uint8_t* buf,
-                       ssize_t* ndatalen,
-                       size_t max_packet_size,
-                       const StreamData& stream_data,
-                       uint64_t ts);
-
   Session* session_ = nullptr;
 };
 
-struct Session::Application::StreamData final {
-  // The actual number of vectors in the struct, up to kMaxVectorCount.
-  size_t count = 0;
-  // The stream identifier. If this is a negative value then no stream is
-  // identified.
-  stream_id id = -1;
-  int fin = 0;
-  ngtcp2_vec data[kMaxVectorCount]{};
-  BaseObjectPtr<Stream> stream;
-
-  static_assert(sizeof(ngtcp2_vec) == sizeof(nghttp3_vec) &&
-                    alignof(ngtcp2_vec) == alignof(nghttp3_vec) &&
-                    offsetof(ngtcp2_vec, base) == offsetof(nghttp3_vec, base) &&
-                    offsetof(ngtcp2_vec, len) == offsetof(nghttp3_vec, len),
-                "ngtcp2_vec and nghttp3_vec must have identical layout");
-  inline operator nghttp3_vec*() {
-    return reinterpret_cast<nghttp3_vec*>(data);
-  }
-
-  inline operator const ngtcp2_vec*() const { return data; }
-  inline operator ngtcp2_vec*() { return data; }
-
-  std::string ToString() const;
-};
-
 // Create a DefaultApplication for the given session.
 std::unique_ptr<Session::Application> CreateDefaultApplication(
     Session* session, const Session::Application_Options& options);

src/quic/application.h

@@ -209,24 +209,25 @@ class Http3ApplicationImpl final : public Session::Application {
     started_ = true;
     Debug(&session(), "Starting HTTP/3 application.");
 
-    auto params = ngtcp2_conn_get_remote_transport_params(session());
-    if (params == nullptr) [[unlikely]] {
+    const auto params = session().remote_transport_params();
+    if (!params) [[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) {
+    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);
+            "unidirectional streams [%" PRIu64
+            "] is too low. Must be at least 3",
+            params.initial_max_streams_uni());
       return false;
     }
 
@@ -235,17 +236,14 @@ class Http3ApplicationImpl final : public Session::Application {
     // 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);
+          *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 &&
+        session().OpenUnidirectionalStream(&control_stream_id_) &&
+        session().OpenUnidirectionalStream(&qpack_enc_stream_id_) &&
+        session().OpenUnidirectionalStream(&qpack_dec_stream_id_) &&
         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;
@@ -306,8 +304,7 @@ class Http3ApplicationImpl final : public Session::Application {
       Debug(&session(),
             "Extending stream and connection offset by %zd bytes",
             nread);
-      session().ExtendStreamOffset(id, nread);
-      session().ExtendOffset(nread);
+      session().Consume(id, nread);
     }
 
     // If this data arrived as 0-RTT, mark the stream. We set it after
@@ -365,24 +362,11 @@ class Http3ApplicationImpl final : public Session::Application {
       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;
-          }
-        }
+        Debug(&session(),
+              "HTTP/3 application extending max %s streams by %" PRIu64,
+              direction == Direction::BIDIRECTIONAL ? "bidi" : "uni",
+              max_streams);
+        session().ExtendMaxStreams(direction, max_streams);
       }
     }
   }
@@ -530,8 +514,7 @@ class Http3ApplicationImpl final : public Session::Application {
       return;
     }
 
-    session().SetLastError(
-        QuicError::ForApplication(nghttp3_err_infer_quic_app_error_code(rv)));
+    session().SetApplicationError(nghttp3_err_infer_quic_app_error_code(rv));
     session().Close();
   }
 
@@ -548,8 +531,7 @@ class Http3ApplicationImpl final : public Session::Application {
       return;
     }
 
-    session().SetLastError(
-        QuicError::ForApplication(nghttp3_err_infer_quic_app_error_code(rv)));
+    session().SetApplicationError(nghttp3_err_infer_quic_app_error_code(rv));
     session().Close();
   }
 
@@ -687,17 +669,30 @@ class Http3ApplicationImpl final : public Session::Application {
     return {StreamPriority::DEFAULT, StreamPriorityFlags::NON_INCREMENTAL};
   }
 
-  int GetStreamData(StreamData* data) override {
+  int GetStreamData(Session::StreamData* data) override {
+    static_assert(
+        sizeof(ngtcp2_vec) == sizeof(nghttp3_vec) &&
+            alignof(ngtcp2_vec) == alignof(nghttp3_vec) &&
+            offsetof(ngtcp2_vec, base) == offsetof(nghttp3_vec, base) &&
+            offsetof(ngtcp2_vec, len) == offsetof(nghttp3_vec, len),
+        "ngtcp2_vec and nghttp3_vec must have identical layout");
     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);
+      // nghttp3 reports fin through an int out-param; bridge it to the bool.
+      int fin = 0;
+      ret =
+          nghttp3_conn_writev_stream(*this,
+                                     &data->id,
+                                     &fin,
+                                     reinterpret_cast<nghttp3_vec*>(data->data),
+                                     data->count);
       // A negative return value indicates an error.
       if (ret < 0) {
         return static_cast<int>(ret);
       }
+      data->fin = fin != 0;
 
       data->count = static_cast<size_t>(ret);
       if (data->id >= 0 && data->id != control_stream_id_ &&
@@ -710,7 +705,7 @@ class Http3ApplicationImpl final : public Session::Application {
     return 0;
   }
 
-  bool StreamCommit(StreamData* data, size_t datalen) override {
+  bool StreamCommit(Session::StreamData* data, size_t datalen) override {
     Debug(&session(),
           "HTTP/3 application committing stream %" PRIi64 " data %zu",
           data->id,
@@ -720,8 +715,7 @@ class Http3ApplicationImpl final : public Session::Application {
     // nghttp3 tracks its own offset via add_write_offset.
     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)));
+      session().SetApplicationError(nghttp3_err_infer_quic_app_error_code(err));
       return false;
     }
     // Raw application bytes are committed to the stream's outbound
@@ -1212,10 +1206,10 @@ class Http3ApplicationImpl final : public Session::Application {
                                  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(id, consumed);
-    session.ExtendOffset(consumed);
+    Debug(&app.session(),
+          "HTTP/3 application deferred consume %zu bytes",
+          consumed);
+    app.session().Consume(id, consumed);
     return NGTCP2_SUCCESS;
   }