quic: improve stream cleanup & lookup

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

Author: pimterry

src/quic/application.cc

@@ -207,9 +207,11 @@ Session::Application::ExtractSessionTicketAppData(
              : SessionTicket::AppData::Status::TICKET_USE;
 }
 
-void Session::Application::ReceiveStreamClose(Stream* stream,
+void Session::Application::ReceiveStreamClose(stream_id id,
+                                              Stream* stream,
                                               QuicError&& error) {
-  DCHECK_NOT_NULL(stream);
+  // Stream may be nullptr if our side is already gone
+  if (stream == nullptr) return;
   stream->Destroy(std::move(error));
 }
 

src/quic/application.h

@@ -126,14 +126,6 @@ class Session::Application : public MemoryRetainer {
   // to send for the given stream.
   virtual void ResumeStream(stream_id id) {}
 
-  // Called when the Session determines that the maximum number of
-  // remotely-initiated unidirectional streams has been extended. Not all
-  // Application types will require this notification so the default is to do
-  // nothing.
-  virtual void ExtendMaxStreams(EndpointLabel label,
-                                Direction direction,
-                                uint64_t max_streams) {}
-
   // Returns true if the application manages stream FIN internally (e.g.,
   // HTTP/3 uses nghttp3 which sends FIN via the fin flag in writev_stream).
   // When true, the stream infrastructure must NOT call
@@ -166,9 +158,15 @@ class Session::Application : public MemoryRetainer {
       SessionTicket::AppData::Source::Flag flag);
 
   // Notifies the Application that the identified stream has been closed.
-  virtual void ReceiveStreamClose(Stream* stream,
+  virtual void ReceiveStreamClose(stream_id id,
+                                  Stream* stream,
                                   QuicError&& error = QuicError());
 
+  // Notifies the Application that the Stream for the identified stream has
+  // been removed from the session and may be freed immediately afterwards.
+  // Applications caching the Stream pointer must drop it here.
+  virtual void StreamRemoved(stream_id id) {}
+
   // Notifies the Application that the identified stream has been reset.
   virtual void ReceiveStreamReset(Stream* stream,
                                   uint64_t final_size,

src/quic/defs.h

@@ -290,11 +290,6 @@ enum class Side : uint8_t {
   SERVER,
 };
 
-enum class EndpointLabel : uint8_t {
-  LOCAL,
-  REMOTE,
-};
-
 enum class Direction : uint8_t {
   BIDIRECTIONAL,
   UNIDIRECTIONAL,

src/quic/http3.cc

@@ -373,22 +373,6 @@ class Http3ApplicationImpl final : public Session::Application {
     Application::ResumeStream(id);
   }
 
-  void ExtendMaxStreams(EndpointLabel label,
-                        Direction direction,
-                        uint64_t max_streams) override {
-    switch (label) {
-      case EndpointLabel::LOCAL:
-        return;
-      case EndpointLabel::REMOTE: {
-        Debug(&session(),
-              "HTTP/3 application extending max %s streams by %" PRIu64,
-              direction == Direction::BIDIRECTIONAL ? "bidi" : "uni",
-              max_streams);
-        session().ExtendMaxStreams(direction, max_streams);
-      }
-    }
-  }
-
   void ExtendMaxStreamData(Stream* stream, uint64_t max_data) override {
     Debug(&session(),
           "HTTP/3 application extending max stream data to %" PRIu64,
@@ -498,33 +482,26 @@ class Http3ApplicationImpl final : public Session::Application {
                : SessionTicket::AppData::Status::TICKET_USE;
   }
 
-  void ReceiveStreamClose(Stream* stream,
+  void ReceiveStreamClose(stream_id id,
+                          Stream* stream,
                           QuicError&& error = QuicError()) override {
-    Debug(
-        &session(), "HTTP/3 application closing stream %" PRIi64, stream->id());
-    error_code code = NGHTTP3_H3_NO_ERROR;
-    if (error.type() == QuicError::Type::APPLICATION) {
-      code = error.code();
-    }
-
-    int rv = nghttp3_conn_close_stream2(
-        *this,
-        NGHTTP3_STREAM_CLOSE_FLAG_RX_APP_ERROR_CODE_SET,
-        stream->id(),
-        code,
-        0);
-    // 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;
+    Debug(&session(), "HTTP/3 application closing stream %" PRIi64, id);
+
+    // Clean up nghttp3's state first. N.b. destroying the Stream calls into
+    // JS, so this can tear down the session. Skip unidirectional streams
+    // (control/QPACK) as nghttp3 handles this and would reject if we try.
+    if (conn_ && ngtcp2_is_bidi_stream(id)) {
+      int rv = nghttp3_conn_close_stream2(
+          *this, NGHTTP3_STREAM_CLOSE_FLAG_NONE, id, 0, 0);
+      if (rv != 0 && rv != NGHTTP3_ERR_STREAM_NOT_FOUND) {
+        session().SetApplicationError(
+            nghttp3_err_infer_quic_app_error_code(rv));
+        session().Close();
+        return;
+      }
     }
 
-    session().SetApplicationError(nghttp3_err_infer_quic_app_error_code(rv));
-    session().Close();
+    Application::ReceiveStreamClose(id, stream, std::move(error));
   }
 
   void ReceiveStreamReset(Stream* stream,
@@ -567,6 +544,10 @@ class Http3ApplicationImpl final : public Session::Application {
     return true;
   }
 
+  void StreamRemoved(stream_id id) override {
+    if (conn_) nghttp3_conn_set_stream_user_data(*this, id, nullptr);
+  }
+
   bool SendHeaders(Stream& stream,
                    HeadersKind kind,
                    const Local<Array>& headers,
@@ -910,37 +891,8 @@ class Http3ApplicationImpl final : public Session::Application {
     return Http3ConnectionPointer(conn);
   }
 
-  void OnStreamClose(Stream* stream,
-                     uint32_t flags,
-                     error_code rx_app_error_code,
-                     error_code tx_app_error_code) {
-    if (flags & NGHTTP3_STREAM_CLOSE_FLAG_RX_APP_ERROR_CODE_SET) {
-      Debug(&session(),
-            "HTTP/3 application received stream close for stream %" PRIi64
-            " with remote error code %" PRIu64,
-            stream->id(),
-            rx_app_error_code);
-    }
-    if (flags & NGHTTP3_STREAM_CLOSE_FLAG_TX_APP_ERROR_CODE_SET) {
-      Debug(&session(),
-            "HTTP/3 application send stream close for stream %" PRIi64
-            " with error code %" PRIu64,
-            stream->id(),
-            tx_app_error_code);
-    }
-    auto direction = stream->direction();
-    if (flags & NGHTTP3_STREAM_CLOSE_FLAG_RX_APP_ERROR_CODE_SET) {
-      stream->Destroy(QuicError::ForApplication(rx_app_error_code));
-    } else if (flags & NGHTTP3_STREAM_CLOSE_FLAG_TX_APP_ERROR_CODE_SET) {
-      stream->Destroy(QuicError::ForApplication(tx_app_error_code));
-    } else {
-      stream->Destroy();
-    }
-    ExtendMaxStreams(EndpointLabel::REMOTE, direction, 1);
-  }
-
   void OnBeginHeaders(stream_id id) {
-    auto stream = FindOrCreateStream(conn_.get(), &session(), id);
+    auto stream = FindOrCreateStream(id);
     if (!stream) [[unlikely]]
       return;
     Debug(&session(),
@@ -994,7 +946,7 @@ class Http3ApplicationImpl final : public Session::Application {
   }
 
   void OnBeginTrailers(stream_id id) {
-    auto stream = FindOrCreateStream(conn_.get(), &session(), id);
+    auto stream = FindOrCreateStream(id);
     if (!stream) [[unlikely]]
       return;
     Debug(&session(),
@@ -1186,20 +1138,26 @@ class Http3ApplicationImpl final : public Session::Application {
     return app;
   }
 
-  static BaseObjectWeakPtr<Stream> FindOrCreateStream(nghttp3_conn* conn,
-                                                      Session* session,
-                                                      stream_id id) {
-    if (auto stream = session->FindStream(id)) {
+  // Cache the Stream* in nghttp3 so we can quickly get it later:
+  void BindStreamUserData(stream_id id, Stream* stream) {
+    if (conn_) nghttp3_conn_set_stream_user_data(*this, id, stream);
+  }
+
+  BaseObjectWeakPtr<Stream> FindOrCreateStream(stream_id id) {
+    if (auto stream = session().FindStream(id)) {
+      BindStreamUserData(id, stream.get());
       return stream;
     }
     // No record of a locally-initiated stream means we already destroyed it,
     // and frames still in flight must not bring it back to life. See
     // DefaultApplication::ReceiveStreamData for the same guard on the raw
     // QUIC path.
-    if (!session->is_destroyed() && ngtcp2_conn_is_local_stream(*session, id)) {
+    if (!session().is_destroyed() &&
+        ngtcp2_conn_is_local_stream(session(), id)) {
       return {};
     }
-    if (auto stream = session->CreateStream(id)) {
+    if (auto stream = session().CreateStream(id)) {
+      if (!stream->is_destroyed()) BindStreamUserData(id, stream.get());
       return stream;
     }
     return {};
@@ -1223,8 +1181,11 @@ class Http3ApplicationImpl final : public Session::Application {
     auto& app = *ptr;
     NgHttp3CallbackScope scope(&app.session());
 
-    auto stream = app.session().FindStream(id);
-    if (!stream) return NGHTTP3_ERR_CALLBACK_FAILURE;
+    BaseObjectPtr<Stream> stream(static_cast<Stream*>(stream_user_data));
+    if (!stream) [[unlikely]] {
+      stream = app.session().FindStream(id);
+      if (!stream) return NGHTTP3_ERR_CALLBACK_FAILURE;
+    }
 
     if (stream->is_eos()) {
       *pflags |= NGHTTP3_DATA_FLAG_EOF;
@@ -1305,23 +1266,12 @@ class Http3ApplicationImpl final : public Session::Application {
     auto ptr = From(conn, conn_user_data);
     CHECK_NOT_NULL(ptr);
     auto& app = *ptr;
-    if (auto stream = app.session().FindStream(id)) {
-      stream->Acknowledge(static_cast<size_t>(datalen));
+    BaseObjectPtr<Stream> stream(static_cast<Stream*>(stream_user_data));
+    if (!stream) [[unlikely]] {
+      stream = app.session().FindStream(id);
     }
-    return NGTCP2_SUCCESS;
-  }
-
-  static int on_stream_close(nghttp3_conn* conn,
-                             uint32_t flags,
-                             stream_id id,
-                             error_code rx_app_error_code,
-                             error_code tx_app_error_code,
-                             void* conn_user_data,
-                             void* stream_user_data) {
-    NGHTTP3_CALLBACK_SCOPE(app);
-    if (auto stream = app.session().FindStream(id)) {
-      app.OnStreamClose(
-          stream.get(), flags, rx_app_error_code, tx_app_error_code);
+    if (stream) {
+      stream->Acknowledge(static_cast<size_t>(datalen));
     }
     return NGTCP2_SUCCESS;
   }
@@ -1339,6 +1289,14 @@ class Http3ApplicationImpl final : public Session::Application {
     if (app.is_control_stream(id)) [[unlikely]] {
       return NGHTTP3_ERR_CALLBACK_FAILURE;
     }
+    // A cached Stream* is cleared before the Stream is removed from the
+    // session, non-null here means the stream is good to go.
+    if (auto* cached = static_cast<Stream*>(stream_user_data)) [[likely]] {
+      BaseObjectPtr<Stream> stream(cached);
+      stream->ReceiveData(data, datalen, Stream::ReceiveDataFlags{});
+      return NGTCP2_SUCCESS;
+    }
+
     auto& session = app.session();
 
     // DATA frames for a request stream the application already destroyed can
@@ -1357,7 +1315,7 @@ class Http3ApplicationImpl final : public Session::Application {
       return NGTCP2_SUCCESS;
     }
 
-    if (auto stream = FindOrCreateStream(conn, &session, id)) [[likely]] {
+    if (auto stream = app.FindOrCreateStream(id)) {
       stream->ReceiveData(data, datalen, Stream::ReceiveDataFlags{});
       return NGTCP2_SUCCESS;
     }
@@ -1564,7 +1522,9 @@ class Http3ApplicationImpl final : public Session::Application {
       on_end_origin,
       on_rand,
       on_receive_settings,
-      on_stream_close};
+      // We don't have to listen for stream_close - nghttp3 only closes when
+      // ReceiveStreamClose requests it, when we've already handled this.
+      nullptr};
 };
 
 std::unique_ptr<Session::Application> CreateHttp3Application(

src/quic/session.cc

@@ -1579,13 +1579,19 @@ struct Session::Impl final : public MemoryRetainer {
                              void* user_data,
                              void* stream_user_data) {
     NGTCP2_CALLBACK_SCOPE(session)
+    // If the peer closes a stream, we return the credit to allow a new one:
+    session->ExtendMaxStreams(stream_id);
+    if (!session->has_application()) return NGTCP2_SUCCESS;
+
+    // Stream here may be null (e.g. if already destroyed locally) so we also
+    // pass stream_id to the calls below as well.
     auto* stream = Stream::From(stream_user_data);
-    if (stream == nullptr) return NGTCP2_SUCCESS;
+
     if (flags & NGTCP2_STREAM_CLOSE_FLAG_APP_ERROR_CODE_SET) {
       session->application().ReceiveStreamClose(
-          stream, QuicError::ForApplication(app_error_code));
+          stream_id, stream, QuicError::ForApplication(app_error_code));
     } else {
-      session->application().ReceiveStreamClose(stream);
+      session->application().ReceiveStreamClose(stream_id, stream);
     }
     return NGTCP2_SUCCESS;
   }
@@ -3320,17 +3326,11 @@ void Session::AddStream(BaseObjectPtr<Stream> stream,
 void Session::RemoveStream(stream_id id) {
   DCHECK(!is_destroyed());
   Debug(this, "Removing stream %" PRIi64 " from session", id);
-  if (!is_in_draining_period() && !is_in_closing_period() &&
-      !ngtcp2_conn_is_local_stream(*this, id)) {
-    if (ngtcp2_is_bidi_stream(id)) {
-      ngtcp2_conn_extend_max_streams_bidi(*this, 1);
-    } else {
-      ngtcp2_conn_extend_max_streams_uni(*this, 1);
-    }
-  }
 
   ngtcp2_conn_set_stream_user_data(*this, id, nullptr);
 
+  if (has_application()) application().StreamRemoved(id);
+
   // Note that removing the stream from the streams map likely releases
   // the last BaseObjectPtr holding onto the Stream instance, at which
   // point it will be freed. If there are other BaseObjectPtr instances
@@ -3479,14 +3479,15 @@ bool Session::OpenUnidirectionalStream(stream_id* id) {
   return ngtcp2_conn_open_uni_stream(*this, id, nullptr) == 0;
 }
 
-void Session::ExtendMaxStreams(Direction direction, uint64_t max) {
-  switch (direction) {
-    case Direction::BIDIRECTIONAL:
-      ngtcp2_conn_extend_max_streams_bidi(*this, static_cast<size_t>(max));
-      break;
-    case Direction::UNIDIRECTIONAL:
-      ngtcp2_conn_extend_max_streams_uni(*this, static_cast<size_t>(max));
-      break;
+void Session::ExtendMaxStreams(stream_id id) {
+  // MAX_STREAMS only limits what the peer opens, and there is nothing to
+  // grant once the connection is going away.
+  if (is_in_draining_period() || is_in_closing_period()) return;
+  if (ngtcp2_conn_is_local_stream(*this, id)) return;
+  if (ngtcp2_is_bidi_stream(id)) {
+    ngtcp2_conn_extend_max_streams_bidi(*this, 1);
+  } else {
+    ngtcp2_conn_extend_max_streams_uni(*this, 1);
   }
 }
 

src/quic/session.h

@@ -518,6 +518,8 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
   void AddStream(BaseObjectPtr<Stream> stream,
                  CreateStreamOption option = CreateStreamOption::NOTIFY);
   void RemoveStream(stream_id id);
+
+  void ExtendMaxStreams(stream_id id);
   void ResumeStream(stream_id id);
   void StreamDataBlocked(stream_id id);
   void ShutdownStream(stream_id id, QuicError error = QuicError());
@@ -552,8 +554,6 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
   // Open a unidirectional stream, setting *id on success, or returning false
   bool OpenUnidirectionalStream(stream_id* id);
 
-  void ExtendMaxStreams(Direction direction, uint64_t max);
-
   // Signal that we've consumed `len` bytes on stream `id` to update flow
   // control
   void Consume(stream_id id, size_t len);