quic: Implement webtransport close session stream

Author: martenrichter

lib/internal/quic/quic.js

@@ -212,6 +212,7 @@ const {
   kReset,
   kSendHeaders,
   kMakeWebtransportStream,
+  kCloseWebtransportSessionStream,
   kSessionApplication,
   kSessionTicket,
   kTrailers,
@@ -2040,6 +2041,33 @@ class QuicStream {
     return this.#inner.pendingClose.promise;
   }
 
+  /**
+   * Only for webtransport session streams
+   * Closes the webtransport session stream, and also closes
+   * connected datastream internally.
+   * @param {number|undefined} code optional error code
+   * @param {string|undefined} msg optional error message truncated to 1024 bytes
+   */
+  closeWebtransportSessionStream(code, msg) {
+    assertIsQuicStream(this);
+    const inner = this.#inner;
+    if (inner.destroying || this.destroyed) return;
+
+    if (msg !== undefined) {
+      validateString(msg, 'msg');
+    }
+    inner.destroying = true;
+    const handle = this.#handle;
+    const error = makeQuicError(
+          'ERR_QUIC_APPLICATION_ERROR',
+          'Webtransport error',
+          'application',
+          code ?? 0,
+          msg ?? '');
+    this[kFinishClose](error);
+    handle.destroy();
+  }
+
   /**
    * Immediately destroys the stream. Any queued data is discarded. If
    * an error is given, the closed promise will be rejected with that
@@ -2603,6 +2631,7 @@ class QuicStream {
   /** 
    * Attaches a webtransport session to the stream and sends initial bytes
    * indicating webtransport stream and the session id
+   * @param {QuicStream} session Webtransport session stream
    * @returns {boolean} true if it succeeded.
    */
   [kMakeWebtransportStream](session) {
@@ -2614,6 +2643,23 @@ class QuicStream {
     return this.#handle.makeWebtransportStream(session.#handle);
   }
 
+   /** 
+   * Closes a webtransport session stream and also closes associated data streams
+   * Passing optional error code and error message (limited to 1024 bytes).
+   * @param {number} code error code
+   * @param {string} msg error message limited to 1024 bytes
+   * @returns {boolean} true if it succeeded.
+   */
+  [kCloseWebtransportSessionStream](code, msg) {
+    if (this.pending) {
+      debug('pending stream enqueuing closeWebtransportSessionStream with code', code, ' and msg:', msg);
+    } else {
+      debug(`stream ${this.id} closeWebtransportSessionStream with code`, code,
+        ' and msg:', msg);
+    }
+    return this.#handle.closeWebtransportSessionStream(session.#handle);
+  }
+
   [kFinishClose](error) {
     const inner = this.#inner;
     inner.pendingClose ??= PromiseWithResolvers();

lib/internal/quic/symbols.js

@@ -58,6 +58,7 @@ const kRemoveStream = Symbol('kRemoveStream');
 const kReset = Symbol('kReset');
 const kSendHeaders = Symbol('kSendHeaders');
 const kMakeWebtransportStream = Symbol('kMakeWebtransportStream');
+const kCloseWebtransportSessionStream = Symbol('kCloseWebtransportSessionStream');
 const kSessionApplication = Symbol('kSessionApplication');
 const kSessionTicket = Symbol('kSessionTicket');
 const kTrailers = Symbol('kTrailers');
@@ -96,6 +97,7 @@ module.exports = {
   kReset,
   kSendHeaders,
   kMakeWebtransportStream,
+  kCloseWebtransportSessionStream,
   kSessionApplication,
   kSessionTicket,
   kTrailers,

src/quic/application.h

@@ -216,6 +216,17 @@ class Session::Application : public MemoryRetainer {
       return false;
   }
 
+  // closes the webtransort session stream,
+  // and also closes connect webtransport data streams
+  virtual bool CloseWebtransportSessionStream(
+      const Stream& stream,
+      uint32_t wt_error_code,
+      const uint8_t *msg,
+      size_t msglen
+    ) {
+      return false;
+  }
+
   // Signals to the Application that it should serialize and transmit any
   // pending session and stream packets it has accumulated.
   void SendPendingData();

src/quic/http3.cc

@@ -686,6 +686,27 @@ class Http3ApplicationImpl final : public Session::Application {
                                      == 0;
   }
 
+  // closes the webtransort session stream,
+  // and also closes connect webtransport data streams
+  // msg is optional
+  // msg length is maximum 1024
+  bool CloseWebtransportSessionStream(
+      const Stream& stream,
+      uint32_t wt_error_code,
+      const uint8_t *msg,
+      size_t msglen
+    ) override {
+    Session::SendPendingDataScope send_scope(&session());
+    Debug(&session(),
+          "Close webtransport session stream %" PRIu64,
+          stream.id());
+    return nghttp3_conn_close_wt_session(*this,
+                                         stream.id(),
+                                         wt_error_code,
+                                         msg,
+                                         msglen) == 0;
+  }
+
   void SetStreamPriority(const Stream& stream,
                          StreamPriority priority,
                          StreamPriorityFlags flags) override {

src/quic/streams.cc

@@ -103,6 +103,7 @@ namespace quic {
   V(Destroy, destroy, false)                                                   \
   V(SendHeaders, sendHeaders, false)                                           \
   V(MakeWebtransportStream, makeWebtransportStream, false)                     \
+  V(CloseWebtransportSessionStream, closeWebtransportSessionStream, false)     \
   V(StopSending, stopSending, false)                                           \
   V(ResetStream, resetStream, false)                                           \
   V(SetPriority, setPriority, false)                                           \
@@ -481,7 +482,7 @@ struct Stream::Impl {
   }
 
   // Connects a stream to a webtransport session stream,
-  // also sends the initial bytes of a stream to signel the wt stream
+  // also sends the initial bytes of a stream to signal the wt stream
   // also connects the readers
   JS_METHOD(MakeWebtransportStream) {
     Stream* stream;
@@ -500,6 +501,40 @@ struct Stream::Impl {
     ));
   }
 
+  // Closes a webtransport session stream,
+  // also closes connected data streams
+  JS_METHOD(CloseWebtransportSessionStream) {
+    Stream* stream;
+    ASSIGN_OR_RETURN_UNWRAP(&stream, args.This());
+    CHECK(args.Length() > 0);
+    CHECK(args[0]->IsObject());
+    uint32_t wt_error_code = 0;
+    if (args.Length() > 1) {
+      CHECK(args[1]->IsUint32());
+      wt_error_code = FromV8Value<uint32_t>(args[0]);
+    }
+    uint8_t * msg = nullptr;
+    size_t msglen = 0;
+    if (args.Length() > 2) {
+      CHECK(args[2]->IsString());
+      Local<String> msgstr = args[2].As<String>();
+      const size_t length = msgstr->Utf8LengthV2(args.GetIsolate());
+      msg = new  uint8_t[length];
+      msgstr->WriteUtf8V2(
+        args.GetIsolate(), reinterpret_cast<char*>(msg), length, String::WriteFlags::kNone);
+      msglen = std::min<size_t>(length, 1024);
+    }
+    args.GetReturnValue().Set(stream->session().application().CloseWebtransportSessionStream(
+      *stream,
+      wt_error_code,
+      msg,
+      msglen
+    ));
+    if (msg) {
+      delete[] msg;
+    }
+  }
+
   // Tells the peer to stop sending data for this stream. This has the effect
   // of shutting down the readable side of the stream for this peer. Any data
   // that has already been received is still readable.