fixup! quic: do not destroy incoming streams that have a consumer

Signed-off-by: Naman Trivedi <trivenay@amazon.com>

Author: trivenay

lib/internal/quic/quic.js

@@ -4139,7 +4139,7 @@ class QuicSession {
   #hasStreamConsumer() {
     if (typeof this.#inner.onstream === 'function') return true;
     if (this[kStreamCallbacks] == null) return false;
-    return getQuicSessionState(this).headersSupported === 1;
+    return getQuicSessionState(this).streamCallbacksSupported === 1;
   }
 
   /**

lib/internal/quic/state.js

@@ -72,6 +72,7 @@ const {
   IDX_STATE_SESSION_STREAM_OPEN_ALLOWED,
   IDX_STATE_SESSION_PRIORITY_SUPPORTED,
   IDX_STATE_SESSION_HEADERS_SUPPORTED,
+  IDX_STATE_SESSION_STREAM_CALLBACKS_SUPPORTED,
   IDX_STATE_SESSION_WRAPPED,
   IDX_STATE_SESSION_APPLICATION_TYPE,
   IDX_STATE_SESSION_NO_ERROR_CODE,
@@ -119,6 +120,7 @@ assert(IDX_STATE_SESSION_HANDSHAKE_CONFIRMED !== undefined);
 assert(IDX_STATE_SESSION_STREAM_OPEN_ALLOWED !== undefined);
 assert(IDX_STATE_SESSION_PRIORITY_SUPPORTED !== undefined);
 assert(IDX_STATE_SESSION_HEADERS_SUPPORTED !== undefined);
+assert(IDX_STATE_SESSION_STREAM_CALLBACKS_SUPPORTED !== undefined);
 assert(IDX_STATE_SESSION_WRAPPED !== undefined);
 assert(IDX_STATE_SESSION_APPLICATION_TYPE !== undefined);
 assert(IDX_STATE_SESSION_NO_ERROR_CODE !== undefined);
@@ -493,6 +495,19 @@ class QuicSessionState {
     return DataViewPrototypeGetUint8(handle, this.#offset + IDX_STATE_SESSION_HEADERS_SUPPORTED);
   }
 
+  /**
+   * Whether the negotiated application dispatches the session-level
+   * stream callbacks (onheaders et al) for incoming streams.
+   * Returns 0 (unknown), 1 (supported), or 2 (not supported).
+   * @type {number}
+   */
+  get streamCallbacksSupported() {
+    const handle = this.#handle;
+    if (handle === undefined) return undefined;
+    return DataViewPrototypeGetUint8(
+      handle, this.#offset + IDX_STATE_SESSION_STREAM_CALLBACKS_SUPPORTED);
+  }
+
   /** @type {boolean} */
   get isWrapped() {
     const handle = this.#handle;

src/quic/application.h

@@ -210,6 +210,11 @@ class Session::Application : public MemoryRetainer {
   // do not support headers should return false (the default).
   virtual bool SupportsHeaders() const { return false; }
 
+  // True if this application dispatches the session-level stream
+  // callbacks (onheaders et al) for incoming streams when they are
+  // registered on the session.
+  virtual bool SupportsStreamCallbacks() const { return false; }
+
   // Initiates application-level graceful shutdown signaling (e.g.,
   // HTTP/3 GOAWAY). Called when Session::Close(GRACEFUL) is invoked.
   virtual void BeginShutdown() {}

src/quic/defs.h

@@ -328,6 +328,12 @@ enum class HeadersSupportState : uint8_t {
   UNSUPPORTED,
 };
 
+enum class StreamCallbacksSupportState : uint8_t {
+  UNKNOWN,
+  SUPPORTED,
+  UNSUPPORTED,
+};
+
 enum class PathValidationResult : uint8_t {
   SUCCESS = NGTCP2_PATH_VALIDATION_RESULT_SUCCESS,
   FAILURE = NGTCP2_PATH_VALIDATION_RESULT_FAILURE,

src/quic/http3.cc

@@ -202,6 +202,8 @@ class Http3ApplicationImpl final : public Session::Application {
 
   bool SupportsHeaders() const override { return true; }
 
+  bool SupportsStreamCallbacks() const override { return true; }
+
   bool is_started() const override { return started_; }
 
   bool Start() override {

src/quic/session.cc

@@ -136,6 +136,7 @@ uint64_t MaxDatagramPayload(uint64_t max_frame_size) {
   V(STREAM_OPEN_ALLOWED, stream_open_allowed, uint8_t)                         \
   V(PRIORITY_SUPPORTED, priority_supported, uint8_t)                           \
   V(HEADERS_SUPPORTED, headers_supported, uint8_t)                             \
+  V(STREAM_CALLBACKS_SUPPORTED, stream_callbacks_supported, uint8_t)           \
   V(WRAPPED, wrapped, uint8_t)                                                 \
   V(APPLICATION_TYPE, application_type, uint8_t)                               \
   V(NO_ERROR_CODE, no_error_code, error_code)                                  \
@@ -2649,6 +2650,10 @@ void Session::SetApplication(std::unique_ptr<Application> app) {
   impl_->state()->headers_supported = static_cast<uint8_t>(
       app->SupportsHeaders() ? HeadersSupportState::SUPPORTED
                              : HeadersSupportState::UNSUPPORTED);
+  impl_->state()->stream_callbacks_supported =
+      static_cast<uint8_t>(app->SupportsStreamCallbacks()
+                               ? StreamCallbacksSupportState::SUPPORTED
+                               : StreamCallbacksSupportState::UNSUPPORTED);
   // Surface the application's "no error" and "internal error" codes via
   // session state so that JS-side code (e.g. the stream writer's fail()
   // path) can resolve the right wire code for the negotiated ALPN