quic: complete the internal implementation of QUIC

Signed-off-by: James M Snell <[email protected]>
Assisted-by: Opencode:Opus 4.6

Author: jasnell

deps/ngtcp2/ngtcp2.gyp

@@ -206,6 +206,7 @@
       'defines': [
         'BUILDING_NGHTTP3',
         'NGHTTP3_STATICLIB',
+        'DEBUGBUILD',
       ],
       'dependencies': [
         'ngtcp2'
@@ -247,7 +248,10 @@
     },
     {
       'target_name': 'ngtcp2_test_server',
-      'type': 'executable',
+      # Disabled: ngtcp2 examples now require C++23 (<print>, <expected>,
+      # std::println, std::expected) which is not yet supported on all
+      # Node.js platforms. Re-enable when C++23 is available.
+      'type': 'none',
       'cflags': [ '-Wno-everything' ],
       'include_dirs': [
         '',
@@ -305,7 +309,10 @@
     },
     {
       'target_name': 'ngtcp2_test_client',
-      'type': 'executable',
+      # Disabled: ngtcp2 examples now require C++23 (<print>, <expected>,
+      # std::println, std::expected) which is not yet supported on all
+      # Node.js platforms. Re-enable when C++23 is available.
+      'type': 'none',
       'cflags': [ '-Wno-everything' ],
       'include_dirs': [
         '',

node.gyp

@@ -346,7 +346,6 @@
       'src/quic/bindingdata.cc',
       'src/quic/cid.cc',
       'src/quic/data.cc',
-      'src/quic/logstream.cc',
       'src/quic/packet.cc',
       'src/quic/preferredaddress.cc',
       'src/quic/sessionticket.cc',
@@ -355,6 +354,7 @@
       'src/quic/endpoint.cc',
       'src/quic/http3.cc',
       'src/quic/session.cc',
+      'src/quic/session_manager.cc',
       'src/quic/streams.cc',
       'src/quic/tlscontext.cc',
       'src/quic/transportparams.cc',
@@ -364,7 +364,6 @@
       'src/quic/cid.h',
       'src/quic/data.h',
       'src/quic/defs.h',
-      'src/quic/logstream.h',
       'src/quic/packet.h',
       'src/quic/preferredaddress.h',
       'src/quic/sessionticket.h',
@@ -374,6 +373,7 @@
       'src/quic/endpoint.h',
       'src/quic/http3.h',
       'src/quic/session.h',
+      'src/quic/session_manager.h',
       'src/quic/streams.h',
       'src/quic/tlscontext.h',
       'src/quic/guard.h',

src/dataqueue/queue.cc

@@ -391,10 +391,11 @@ class NonIdempotentDataQueueReader final
     // If the collection of entries is empty, there's nothing currently left to
     // read. How we respond depends on whether the data queue has been capped
     // or not.
+
     if (data_queue_->entries_.empty()) {
       // If the data_queue_ is empty, and not capped, then we can reasonably
       // expect more data to be provided later, but we don't know exactly when
-      // that'll happe, so the proper response here is to return a blocked
+      // that'll happen, so the proper response here is to return a blocked
       // status.
       if (!data_queue_->is_capped()) {
         std::move(next)(bob::Status::STATUS_BLOCK, nullptr, 0, [](uint64_t) {});
@@ -437,22 +438,39 @@ class NonIdempotentDataQueueReader final
     CHECK(!pull_pending_);
     pull_pending_ = true;
     int status = current_reader->Pull(
-        [this, next = std::move(next)](
-            int status, const DataQueue::Vec* vecs, uint64_t count, Done done) {
+        [this, next = std::move(next), options, data, count, max_count_hint](
+            int status,
+            const DataQueue::Vec* vecs,
+            uint64_t vcount,
+            Done done) mutable {
           pull_pending_ = false;
 
           // In each of these cases, we do not expect that the source will
           // actually have provided any actual data.
           CHECK_IMPLIES(status == bob::Status::STATUS_BLOCK ||
                             status == bob::Status::STATUS_WAIT ||
                             status == bob::Status::STATUS_EOS,
-                        vecs == nullptr && count == 0);
+                        vecs == nullptr && vcount == 0);
           if (status == bob::Status::STATUS_EOS) {
             data_queue_->entries_.erase(data_queue_->entries_.begin());
-            ended_ = data_queue_->entries_.empty();
             current_reader_ = nullptr;
-            if (!ended_) status = bob::Status::STATUS_CONTINUE;
-            std::move(next)(status, nullptr, 0, [](uint64_t) {});
+            if (!data_queue_->entries_.empty()) {
+              // More entries remain. Pull from the next entry immediately
+              // rather than returning empty CONTINUE, which would leave
+              // callers with no data and no way to know they should retry.
+              Pull(std::move(next), options, data, count, max_count_hint);
+            } else if (!data_queue_->is_capped()) {
+              // The queue is empty but not capped — more data may arrive
+              // later. Return BLOCK so the consumer waits rather than
+              // falsely treating this as end-of-stream.
+              std::move(next)(
+                  bob::Status::STATUS_BLOCK, nullptr, 0, [](uint64_t) {});
+            } else {
+              // Empty and capped — truly done.
+              ended_ = true;
+              std::move(next)(
+                  bob::Status::STATUS_EOS, nullptr, 0, [](uint64_t) {});
+            }
             return;
           }
 
@@ -461,15 +479,15 @@ class NonIdempotentDataQueueReader final
           if (data_queue_->HasBackpressureListeners()) {
             // How much did we actually read?
             size_t read = 0;
-            for (uint64_t n = 0; n < count; n++) {
+            for (uint64_t n = 0; n < vcount; n++) {
               read += vecs[n].len;
             }
             data_queue_->NotifyBackpressure(read);
           }
 
           // Now that we have updated this readers state, we can forward
           // everything on to the outer next.
-          std::move(next)(status, vecs, count, std::move(done));
+          std::move(next)(status, vecs, vcount, std::move(done));
         },
         options,
         data,

src/debug_utils.h

@@ -54,7 +54,8 @@ void NODE_EXTERN_PRIVATE FWrite(FILE* file, const std::string& str);
   V(INSPECTOR_CLIENT)                                                          \
   V(INSPECTOR_PROFILER)                                                        \
   V(CODE_CACHE)                                                                \
-  V(NGTCP2_DEBUG)                                                              \
+  V(NGTCP2)                                                                    \
+  V(NGHTTP3)                                                                   \
   V(SEA)                                                                       \
   V(WASI)                                                                      \
   V(MODULE)                                                                    \

src/node_blob.cc

@@ -156,8 +156,7 @@ Local<FunctionTemplate> Blob::GetConstructorTemplate(Environment* env) {
     Isolate* isolate = env->isolate();
     tmpl = NewFunctionTemplate(isolate, nullptr);
     tmpl->InstanceTemplate()->SetInternalFieldCount(Blob::kInternalFieldCount);
-    tmpl->SetClassName(
-        FIXED_ONE_BYTE_STRING(env->isolate(), "Blob"));
+    tmpl->SetClassName(FIXED_ONE_BYTE_STRING(env->isolate(), "Blob"));
     SetProtoMethod(isolate, tmpl, "getReader", GetReader);
     SetProtoMethod(isolate, tmpl, "slice", ToSlice);
     env->set_blob_constructor_template(tmpl);
@@ -255,8 +254,7 @@ void Blob::New(const FunctionCallbackInfo<Value>& args) {
   }
 
   auto blob = Create(env, DataQueue::CreateIdempotent(std::move(entries)));
-  if (blob)
-    args.GetReturnValue().Set(blob->object());
+  if (blob) args.GetReturnValue().Set(blob->object());
 }
 
 void Blob::GetReader(const FunctionCallbackInfo<Value>& args) {
@@ -278,8 +276,7 @@ void Blob::ToSlice(const FunctionCallbackInfo<Value>& args) {
   size_t start = args[0].As<Uint32>()->Value();
   size_t end = args[1].As<Uint32>()->Value();
   BaseObjectPtr<Blob> slice = blob->Slice(env, start, end);
-  if (slice)
-    args.GetReturnValue().Set(slice->object());
+  if (slice) args.GetReturnValue().Set(slice->object());
 }
 
 void Blob::MemoryInfo(MemoryTracker* tracker) const {
@@ -343,6 +340,7 @@ void Blob::Reader::Pull(const FunctionCallbackInfo<Value>& args) {
   Environment* env = Environment::GetCurrent(args);
   Blob::Reader* reader;
   ASSIGN_OR_RETURN_UNWRAP(&reader, args.This());
+  reader->pull_pending_ = false;
 
   CHECK(args[0]->IsFunction());
   Local<Function> fn = args[0].As<Function>();
@@ -414,19 +412,31 @@ void Blob::Reader::Pull(const FunctionCallbackInfo<Value>& args) {
 void Blob::Reader::SetWakeup(const FunctionCallbackInfo<Value>& args) {
   Blob::Reader* reader;
   ASSIGN_OR_RETURN_UNWRAP(&reader, args.This());
+  if (args[0]->IsUndefined()) {
+    reader->wakeup_.Reset();
+    return;
+  }
   CHECK(args[0]->IsFunction());
   reader->wakeup_.Reset(args.GetIsolate(), args[0].As<Function>());
 }
 
-void Blob::Reader::NotifyPull() {
+void Blob::Reader::NotifyPull(bool fin) {
   if (wakeup_.IsEmpty() || !env()->can_call_into_js()) return;
+  // FIN notifications always fire — they must not be suppressed by
+  // pull_pending_ because there will be no further notifications to
+  // wake the iterator. Regular data notifications respect pull_pending_
+  // to coalesce multiple deliveries within a single packet.
+  if (!fin && pull_pending_) return;
+  pull_pending_ = true;
   HandleScope handle_scope(env()->isolate());
   Local<Function> fn = wakeup_.Get(env()->isolate());
-  MakeCallback(fn, 0, nullptr);
+  // Pass fin as the first argument so the JS iterator knows EOS is
+  // imminent and should pull again without waiting for another wakeup.
+  Local<Value> argv[] = {v8::Boolean::New(env()->isolate(), fin)};
+  MakeCallback(fn, 1, argv);
 }
 
-BaseObjectPtr<BaseObject>
-Blob::BlobTransferData::Deserialize(
+BaseObjectPtr<BaseObject> Blob::BlobTransferData::Deserialize(
     Environment* env,
     Local<Context> context,
     std::unique_ptr<worker::TransferData> self) {
@@ -448,10 +458,10 @@ std::unique_ptr<worker::TransferData> Blob::CloneForMessaging() const {
 void Blob::StoreDataObject(const FunctionCallbackInfo<Value>& args) {
   Realm* realm = Realm::GetCurrent(args);
 
-  CHECK(args[0]->IsString());  // ID key
+  CHECK(args[0]->IsString());                       // ID key
   CHECK(Blob::HasInstance(realm->env(), args[1]));  // Blob
-  CHECK(args[2]->IsUint32());  // Length
-  CHECK(args[3]->IsString());  // Type
+  CHECK(args[2]->IsUint32());                       // Length
+  CHECK(args[3]->IsString());                       // Type
 
   BlobBindingData* binding_data = realm->GetBindingData<BlobBindingData>();
   Isolate* isolate = realm->isolate();
@@ -531,12 +541,8 @@ void BlobBindingData::StoredDataObject::MemoryInfo(
 }
 
 BlobBindingData::StoredDataObject::StoredDataObject(
-    const BaseObjectPtr<Blob>& blob_,
-    size_t length_,
-    const std::string& type_)
-    : blob(blob_),
-      length(length_),
-      type(type_) {}
+    const BaseObjectPtr<Blob>& blob_, size_t length_, const std::string& type_)
+    : blob(blob_), length(length_), type(type_) {}
 
 BlobBindingData::BlobBindingData(Realm* realm, Local<Object> wrap)
     : SnapshotableObject(realm, wrap, type_int) {
@@ -550,8 +556,7 @@ void BlobBindingData::MemoryInfo(MemoryTracker* tracker) const {
 }
 
 void BlobBindingData::store_data_object(
-    const std::string& uuid,
-    const BlobBindingData::StoredDataObject& object) {
+    const std::string& uuid, const BlobBindingData::StoredDataObject& object) {
   data_objects_[uuid] = object;
 }
 
@@ -566,8 +571,7 @@ void BlobBindingData::revoke_data_object(const std::string& uuid) {
 BlobBindingData::StoredDataObject BlobBindingData::get_data_object(
     const std::string& uuid) {
   auto entry = data_objects_.find(uuid);
-  if (entry == data_objects_.end())
-    return BlobBindingData::StoredDataObject {};
+  if (entry == data_objects_.end()) return BlobBindingData::StoredDataObject{};
   return entry->second;
 }
 

src/node_blob.h

@@ -23,8 +23,7 @@ namespace node {
 
 class Blob : public BaseObject {
  public:
-  static void RegisterExternalReferences(
-      ExternalReferenceRegistry* registry);
+  static void RegisterExternalReferences(ExternalReferenceRegistry* registry);
 
   static void CreatePerIsolateProperties(IsolateData* isolate_data,
                                          v8::Local<v8::ObjectTemplate> target);
@@ -83,7 +82,7 @@ class Blob : public BaseObject {
                                         BaseObjectPtr<Blob> blob);
     static void Pull(const v8::FunctionCallbackInfo<v8::Value>& args);
     static void SetWakeup(const v8::FunctionCallbackInfo<v8::Value>& args);
-    void NotifyPull();
+    void NotifyPull(bool fin = false);
 
     explicit Reader(Environment* env,
                     v8::Local<v8::Object> obj,
@@ -97,6 +96,7 @@ class Blob : public BaseObject {
     std::shared_ptr<DataQueue::Reader> inner_;
     BaseObjectPtr<Blob> strong_ptr_;
     bool eos_ = false;
+    bool pull_pending_ = false;
     v8::Global<v8::Function> wakeup_;
   };
 
@@ -134,19 +134,17 @@ class BlobBindingData : public SnapshotableObject {
 
     StoredDataObject() = default;
 
-    StoredDataObject(
-        const BaseObjectPtr<Blob>& blob_,
-        size_t length_,
-        const std::string& type_);
+    StoredDataObject(const BaseObjectPtr<Blob>& blob_,
+                     size_t length_,
+                     const std::string& type_);
 
     void MemoryInfo(MemoryTracker* tracker) const override;
     SET_SELF_SIZE(StoredDataObject)
     SET_MEMORY_INFO_NAME(StoredDataObject)
   };
 
-  void store_data_object(
-      const std::string& uuid,
-      const StoredDataObject& object);
+  void store_data_object(const std::string& uuid,
+                         const StoredDataObject& object);
 
   void revoke_data_object(const std::string& uuid);
 

src/node_file.h

@@ -338,6 +338,7 @@ class FileHandle final : public AsyncWrap, public StreamBase {
   static void New(const v8::FunctionCallbackInfo<v8::Value>& args);
 
   int GetFD() override { return fd_; }
+  const std::string& original_name() const { return original_name_; }
 
   int Release();
 

src/node_perf_common.h

@@ -34,12 +34,13 @@ extern uint64_t performance_v8_start;
   V(LOOP_EXIT, "loopExit")                                                     \
   V(BOOTSTRAP_COMPLETE, "bootstrapComplete")
 
-#define NODE_PERFORMANCE_ENTRY_TYPES(V)                                       \
-  V(GC, "gc")                                                                 \
-  V(HTTP, "http")                                                             \
-  V(HTTP2, "http2")                                                           \
-  V(NET, "net")                                                               \
-  V(DNS, "dns")
+#define NODE_PERFORMANCE_ENTRY_TYPES(V)                                        \
+  V(GC, "gc")                                                                  \
+  V(HTTP, "http")                                                              \
+  V(HTTP2, "http2")                                                            \
+  V(NET, "net")                                                                \
+  V(DNS, "dns")                                                                \
+  V(QUIC, "quic")
 
 enum PerformanceMilestone {
 #define V(name, _) NODE_PERFORMANCE_MILESTONE_##name,