src, quic: refine more of the quic implementation

Signed-off-by: James M Snell <[email protected]>

Author: jasnell

lib/internal/quic/quic.js

@@ -23,6 +23,7 @@ const {
 } = require('util/types');
 
 const { inspect } = require('internal/util/inspect');
+const assert = require('internal/assert');
 
 const {
   kFinishClose,
@@ -77,6 +78,41 @@ const {
   IDX_STATE_STREAM_WANTS_TRAILERS,
 } = internalBinding('quic');
 
+assert(IDX_STATE_SESSION_PATH_VALIDATION !== undefined);
+assert(IDX_STATE_SESSION_VERSION_NEGOTIATION !== undefined);
+assert(IDX_STATE_SESSION_DATAGRAM !== undefined);
+assert(IDX_STATE_SESSION_SESSION_TICKET !== undefined);
+assert(IDX_STATE_SESSION_CLOSING !== undefined);
+assert(IDX_STATE_SESSION_GRACEFUL_CLOSE !== undefined);
+assert(IDX_STATE_SESSION_SILENT_CLOSE !== undefined);
+assert(IDX_STATE_SESSION_STATELESS_RESET !== undefined);
+assert(IDX_STATE_SESSION_DESTROYED !== undefined);
+assert(IDX_STATE_SESSION_HANDSHAKE_COMPLETED !== undefined);
+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_WRAPPED !== undefined);
+assert(IDX_STATE_SESSION_LAST_DATAGRAM_ID !== undefined);
+assert(IDX_STATE_ENDPOINT_BOUND !== undefined);
+assert(IDX_STATE_ENDPOINT_RECEIVING !== undefined);
+assert(IDX_STATE_ENDPOINT_LISTENING !== undefined);
+assert(IDX_STATE_ENDPOINT_CLOSING !== undefined);
+assert(IDX_STATE_ENDPOINT_BUSY !== undefined);
+assert(IDX_STATE_ENDPOINT_PENDING_CALLBACKS !== undefined);
+assert(IDX_STATE_STREAM_ID !== undefined);
+assert(IDX_STATE_STREAM_FIN_SENT !== undefined);
+assert(IDX_STATE_STREAM_FIN_RECEIVED !== undefined);
+assert(IDX_STATE_STREAM_READ_ENDED !== undefined);
+assert(IDX_STATE_STREAM_WRITE_ENDED !== undefined);
+assert(IDX_STATE_STREAM_DESTROYED !== undefined);
+assert(IDX_STATE_STREAM_PAUSED !== undefined);
+assert(IDX_STATE_STREAM_RESET !== undefined);
+assert(IDX_STATE_STREAM_HAS_READER !== undefined);
+assert(IDX_STATE_STREAM_WANTS_BLOCK !== undefined);
+assert(IDX_STATE_STREAM_WANTS_HEADERS !== undefined);
+assert(IDX_STATE_STREAM_WANTS_RESET !== undefined);
+assert(IDX_STATE_STREAM_WANTS_TRAILERS !== undefined);
+
 class QuicEndpointState {
   /** @type {DataView} */
   #handle;

lib/internal/quic/state.js

@@ -93,17 +93,13 @@ bool NgHeader<T>::IsZeroLength(
 }
 
 template <typename T>
-bool NgHeader<T>::IsZeroLength(
-    int32_t token,
-    NgHeader<T>::rcbuf_t* name,
-    NgHeader<T>::rcbuf_t* value) {
-
+bool NgHeader<T>::IsZeroLength(int32_t token,
+                               NgHeader<T>::rcbuf_t* name,
+                               NgHeader<T>::rcbuf_t* value) {
   if (NgHeader<T>::rcbufferpointer_t::IsZeroLength(value))
     return true;
 
-  const char* header_name = T::ToHttpHeaderName(token);
-  return header_name != nullptr ||
-      NgHeader<T>::rcbufferpointer_t::IsZeroLength(name);
+  return NgHeader<T>::rcbufferpointer_t::IsZeroLength(name);
 }
 
 template <typename T>

src/node_http_common-inl.h

@@ -27,22 +27,25 @@ namespace quic {
 
 // ============================================================================
 // Session::Application_Options
-const Session::Application_Options Session::Application_Options::kDefault = {};
+const Session::Application::Options Session::Application::Options::kDefault =
+    {};
 
-Session::Application_Options::operator const nghttp3_settings() const {
-  // In theory, Application_Options might contain options for more than just
+Session::Application::Options::operator const nghttp3_settings() const {
+  // In theory, Application::Options might contain options for more than just
   // HTTP/3. Here we extract only the properties that are relevant to HTTP/3.
   return nghttp3_settings{
-      max_field_section_size,
-      static_cast<size_t>(qpack_max_dtable_capacity),
-      static_cast<size_t>(qpack_encoder_max_dtable_capacity),
-      static_cast<size_t>(qpack_blocked_streams),
-      enable_connect_protocol,
-      enable_datagrams,
+      .max_field_section_size = max_field_section_size,
+      .qpack_max_dtable_capacity =
+          static_cast<size_t>(qpack_max_dtable_capacity),
+      .qpack_encoder_max_dtable_capacity =
+          static_cast<size_t>(qpack_encoder_max_dtable_capacity),
+      .qpack_blocked_streams = static_cast<size_t>(qpack_blocked_streams),
+      .enable_connect_protocol = enable_connect_protocol,
+      .h3_datagram = enable_datagrams,
   };
 }
 
-std::string Session::Application_Options::ToString() const {
+std::string Session::Application::Options::ToString() const {
   DebugIndentScope indent;
   auto prefix = indent.Prefix();
   std::string res("{");
@@ -64,48 +67,58 @@ std::string Session::Application_Options::ToString() const {
   return res;
 }
 
-Maybe<Session::Application_Options> Session::Application_Options::From(
+Maybe<Session::Application::Options> Session::Application::Options::From(
     Environment* env, Local<Value> value) {
-  if (value.IsEmpty() || (!value->IsUndefined() && !value->IsObject())) {
+  if (value.IsEmpty()) [[unlikely]] {
     THROW_ERR_INVALID_ARG_TYPE(env, "options must be an object");
-    return Nothing<Application_Options>();
+    return Nothing<Application::Options>();
   }
 
-  Application_Options options;
+  Application::Options options;
   auto& state = BindingData::Get(env);
-  if (value->IsUndefined()) {
-    return Just<Application_Options>(options);
-  }
-
-  auto params = value.As<Object>();
 
 #define SET(name)                                                              \
-  SetOption<Session::Application_Options,                                      \
-            &Session::Application_Options::name>(                              \
+  SetOption<Session::Application::Options,                                     \
+            &Session::Application::Options::name>(                             \
       env, &options, params, state.name##_string())
 
-  if (!SET(max_header_pairs) || !SET(max_header_length) ||
-      !SET(max_field_section_size) || !SET(qpack_max_dtable_capacity) ||
-      !SET(qpack_encoder_max_dtable_capacity) || !SET(qpack_blocked_streams) ||
-      !SET(enable_connect_protocol) || !SET(enable_datagrams)) {
-    return Nothing<Application_Options>();
+  if (!value->IsUndefined()) {
+    if (!value->IsObject()) {
+      THROW_ERR_INVALID_ARG_TYPE(env, "options must be an object");
+      return Nothing<Application::Options>();
+    }
+    auto params = value.As<Object>();
+    if (!SET(max_header_pairs) || !SET(max_header_length) ||
+        !SET(max_field_section_size) || !SET(qpack_max_dtable_capacity) ||
+        !SET(qpack_encoder_max_dtable_capacity) ||
+        !SET(qpack_blocked_streams) || !SET(enable_connect_protocol) ||
+        !SET(enable_datagrams)) {
+      // The call to SetOption should have scheduled an exception to be thrown.
+      return Nothing<Application::Options>();
+    }
   }
 
 #undef SET
 
-  return Just<Application_Options>(options);
+  return Just<Application::Options>(options);
 }
 
 // ============================================================================
 
 std::string Session::Application::StreamData::ToString() const {
   DebugIndentScope indent;
+
+  size_t total_bytes = 0;
+  for (size_t n = 0; n < count; n++) {
+    total_bytes += data[n].len;
+  }
+
   auto prefix = indent.Prefix();
   std::string res("{");
   res += prefix + "count: " + std::to_string(count);
-  res += prefix + "remaining: " + std::to_string(remaining);
   res += prefix + "id: " + std::to_string(id);
   res += prefix + "fin: " + std::to_string(fin);
+  res += prefix + "total: " + std::to_string(total_bytes);
   res += indent.Close();
   return res;
 }
@@ -120,14 +133,16 @@ bool Session::Application::Start() {
   return true;
 }
 
-void Session::Application::AcknowledgeStreamData(Stream* stream,
+bool Session::Application::AcknowledgeStreamData(int64_t stream_id,
                                                  size_t datalen) {
   Debug(session_,
         "Application acknowledging stream %" PRIi64 " data: %zu",
-        stream->id(),
+        stream_id,
         datalen);
-  DCHECK_NOT_NULL(stream);
+  auto stream = session().FindStream(stream_id);
+  if (!stream) return false;
   stream->Acknowledge(datalen);
+  return true;
 }
 
 void Session::Application::BlockStream(int64_t id) {
@@ -241,6 +256,14 @@ void Session::Application::SendPendingData() {
   PathStorage path;
   StreamData stream_data;
 
+  auto update_stats = OnScopeLeave([&] {
+    auto& s = session();
+    s.UpdateDataStats();
+    if (!s.is_destroyed()) {
+      s.UpdateTimer();
+    }
+  });
+
   // The maximum size of packet to create.
   const size_t max_packet_size = session_->max_packet_size();
 
@@ -296,7 +319,15 @@ void Session::Application::SendPendingData() {
     // Awesome, let's write our packet!
     ssize_t nwrite =
         WriteVStream(&path, pos, &ndatalen, max_packet_size, stream_data);
-    Debug(session_, "Application accepted %zu bytes into packet", ndatalen);
+
+    if (ndatalen > 0) {
+      Debug(session_,
+            "Application accepted %zu bytes from stream into packet",
+            ndatalen);
+    } else {
+      Debug(session_,
+            "Application did not accept any bytes from stream into packet");
+    }
 
     // A negative nwrite value indicates either an error or that there is more
     // data to write into the packet.
@@ -309,7 +340,6 @@ void Session::Application::SendPendingData() {
           // ndatalen = -1 means that no stream data was accepted into the
           // packet, which is what we want here.
           DCHECK_EQ(ndatalen, -1);
-          DCHECK(stream_data.stream);
           session_->StreamDataBlocked(stream_data.id);
           continue;
         }
@@ -323,8 +353,7 @@ void Session::Application::SendPendingData() {
           // ndatalen = -1 means that no stream data was accepted into the
           // packet, which is what we want here.
           DCHECK_EQ(ndatalen, -1);
-          DCHECK(stream_data.stream);
-          stream_data.stream->EndWritable();
+          if (stream_data.stream) stream_data.stream->EndWritable();
           continue;
         }
         case NGTCP2_ERR_WRITE_MORE: {
@@ -406,16 +435,16 @@ ssize_t Session::Application::WriteVStream(PathStorage* path,
   DCHECK_LE(stream_data.count, kMaxVectorCount);
   uint32_t flags = NGTCP2_WRITE_STREAM_FLAG_MORE;
   if (stream_data.fin) flags |= NGTCP2_WRITE_STREAM_FLAG_FIN;
-  ngtcp2_pkt_info pi;
+
   return ngtcp2_conn_writev_stream(*session_,
                                    &path->path,
-                                   &pi,
+                                   nullptr,
                                    dest,
                                    max_packet_size,
                                    ndatalen,
                                    flags,
                                    stream_data.id,
-                                   stream_data.buf,
+                                   stream_data,
                                    stream_data.count,
                                    uv_hrtime());
 }
@@ -536,39 +565,20 @@ class DefaultApplication final : public Session::Application {
   }
 
   bool ShouldSetFin(const StreamData& stream_data) override {
-    auto const is_empty = [](auto vec, size_t cnt) {
-      size_t i;
-      for (i = 0; i < cnt && vec[i].len == 0; ++i) {
-      }
-      return i == cnt;
+    auto const is_empty = [](const ngtcp2_vec* vec, size_t cnt) {
+      size_t i = 0;
+      for (size_t n = 0; n < cnt; n++) i += vec[n].len;
+      return i > 0;
     };
 
-    return stream_data.stream && is_empty(stream_data.buf, stream_data.count);
+    return stream_data.stream && is_empty(stream_data, stream_data.count);
   }
 
   bool StreamCommit(StreamData* stream_data, size_t datalen) override {
     Debug(&session(), "Default application committing stream data");
     DCHECK_NOT_NULL(stream_data);
-    const auto consume = [](ngtcp2_vec** pvec, size_t* pcnt, size_t len) {
-      ngtcp2_vec* v = *pvec;
-      size_t cnt = *pcnt;
-
-      for (; cnt > 0; --cnt, ++v) {
-        if (v->len > len) {
-          v->len -= len;
-          v->base += len;
-          break;
-        }
-        len -= v->len;
-      }
-
-      *pvec = v;
-      *pcnt = cnt;
-    };
 
     CHECK(stream_data->stream);
-    stream_data->remaining -= datalen;
-    consume(&stream_data->buf, &stream_data->count, datalen);
     stream_data->stream->Commit(datalen);
     return true;
   }

src/quic/application.cc

@@ -35,7 +35,7 @@ class Session::Application : public MemoryRetainer {
 
   // Session will forward all data acknowledgements for a stream to the
   // Application.
-  virtual void AcknowledgeStreamData(Stream* stream, size_t datalen);
+  virtual bool AcknowledgeStreamData(int64_t stream_id, size_t datalen);
 
   // Called to determine if a Header can be added to this application.
   // Applications that do not support headers will always return false.
@@ -146,10 +146,14 @@ struct Session::Application::StreamData final {
   int64_t id = -1;
   int fin = 0;
   ngtcp2_vec data[kMaxVectorCount]{};
-  ngtcp2_vec* buf = data;
   BaseObjectPtr<Stream> stream;
 
-  inline operator nghttp3_vec() const { return {data[0].base, data[0].len}; }
+  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;
 };

src/quic/application.h

@@ -257,6 +257,12 @@ std::optional<int> QuicError::crypto_error() const {
 }
 
 MaybeLocal<Value> QuicError::ToV8Value(Environment* env) const {
+  if ((type() == QuicError::Type::TRANSPORT && code() == NGTCP2_NO_ERROR) ||
+      (type() == QuicError::Type::APPLICATION && code() == NGTCP2_APP_NOERROR) ||
+      (type() == QuicError::Type::APPLICATION && code() == NGHTTP3_H3_NO_ERROR)) {
+    return Undefined(env->isolate());
+  }
+
   Local<Value> argv[] = {
       Integer::New(env->isolate(), static_cast<int>(type())),
       BigInt::NewFromUnsigned(env->isolate(), code()),