quic: further implementation details

PR-URL: https://github.com/nodejs/node/pull/48244
Reviewed-By: Yagiz Nizipli <yagiz.nizipli@sentry.io>
Reviewed-By: Stephen Belanger <admin@stephenbelanger.com>

src/dataqueue/queue.cc

@@ -162,6 +162,27 @@ class DataQueueImpl final : public DataQueue,
         "entries", entries_, "std::vector<std::unique_ptr<Entry>>");
   }
 
+  void addBackpressureListener(BackpressureListener* listener) override {
+    if (idempotent_) return;
+    DCHECK_NOT_NULL(listener);
+    backpressure_listeners_.insert(listener);
+  }
+
+  void removeBackpressureListener(BackpressureListener* listener) override {
+    if (idempotent_) return;
+    DCHECK_NOT_NULL(listener);
+    backpressure_listeners_.erase(listener);
+  }
+
+  void NotifyBackpressure(size_t amount) {
+    if (idempotent_) return;
+    for (auto& listener : backpressure_listeners_) listener->EntryRead(amount);
+  }
+
+  bool HasBackpressureListeners() const noexcept {
+    return !backpressure_listeners_.empty();
+  }
+
   std::shared_ptr<Reader> get_reader() override;
   SET_MEMORY_INFO_NAME(DataQueue)
   SET_SELF_SIZE(DataQueueImpl)
@@ -173,6 +194,8 @@ class DataQueueImpl final : public DataQueue,
   std::optional<uint64_t> capped_size_ = std::nullopt;
   bool locked_to_reader_ = false;
 
+  std::unordered_set<BackpressureListener*> backpressure_listeners_;
+
   friend class DataQueue;
   friend class IdempotentDataQueueReader;
   friend class NonIdempotentDataQueueReader;
@@ -433,6 +456,17 @@ class NonIdempotentDataQueueReader final
             return;
           }
 
+          // If there is a backpressure listener, lets report on how much data
+          // was actually read.
+          if (data_queue_->HasBackpressureListeners()) {
+            // How much did we actually read?
+            size_t read = 0;
+            for (uint64_t n = 0; n < count; 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));

src/dataqueue/queue.h

@@ -141,6 +141,14 @@ class DataQueue : public MemoryRetainer {
     using Done = bob::Done;
   };
 
+  // A BackpressureListener can be used to receive notifications
+  // when a non-idempotent DataQueue releases entries as they
+  // are consumed.
+  class BackpressureListener {
+   public:
+    virtual void EntryRead(size_t amount) = 0;
+  };
+
   // A DataQueue::Entry represents a logical chunk of data in the queue.
   // The entry may or may not represent memory-resident data. It may
   // or may not be consumable more than once.
@@ -285,6 +293,10 @@ class DataQueue : public MemoryRetainer {
   // been set, maybeCapRemaining() will return std::nullopt.
   virtual std::optional<uint64_t> maybeCapRemaining() const = 0;
 
+  // BackpressureListeners only work on non-idempotent DataQueues.
+  virtual void addBackpressureListener(BackpressureListener* listener) = 0;
+  virtual void removeBackpressureListener(BackpressureListener* listener) = 0;
+
   static void Initialize(Environment* env, v8::Local<v8::Object> target);
   static void RegisterExternalReferences(ExternalReferenceRegistry* registry);
 };

src/quic/application.cc

@@ -1,3 +1,4 @@
+#include "node_bob.h"
 #include "uv.h"
 #if HAVE_OPENSSL && NODE_OPENSSL_HAS_QUIC
 
@@ -79,7 +80,7 @@ void Session::Application::AcknowledgeStreamData(Stream* stream,
 
 void Session::Application::BlockStream(int64_t id) {
   auto stream = session().FindStream(id);
-  if (stream) stream->Blocked();
+  if (stream) stream->EmitBlocked();
 }
 
 bool Session::Application::CanAddHeader(size_t current_count,
@@ -233,7 +234,7 @@ void Session::Application::SendPendingData() {
           // and no more outbound data can be sent.
           CHECK_LE(ndatalen, 0);
           auto stream = session_->FindStream(stream_data.id);
-          if (stream) stream->End();
+          if (stream) stream->EndWritable();
           continue;
         }
         case NGTCP2_ERR_WRITE_MORE: {
@@ -360,10 +361,8 @@ class DefaultApplication final : public Session::Application {
                              stream_data->data,
                              arraysize(stream_data->data),
                              kMaxVectorCount);
-      switch (ret) {
-        case bob::Status::STATUS_EOS:
-          stream_data->fin = 1;
-          break;
+      if (ret == bob::Status::STATUS_EOS) {
+        stream_data->fin = 1;
       }
     } else {
       stream_data->fin = 1;

src/quic/bindingdata.cc

@@ -203,8 +203,12 @@ CallbackScopeBase::CallbackScopeBase(Environment* env)
     : env(env), context_scope(env->context()), try_catch(env->isolate()) {}
 
 CallbackScopeBase::~CallbackScopeBase() {
-  if (try_catch.HasCaught() && !try_catch.HasTerminated()) {
-    errors::TriggerUncaughtException(env->isolate(), try_catch);
+  if (try_catch.HasCaught()) {
+    if (!try_catch.HasTerminated() && env->can_call_into_js()) {
+      errors::TriggerUncaughtException(env->isolate(), try_catch);
+    } else {
+      try_catch.ReThrow();
+    }
   }
 }
 

src/quic/bindingdata.h

@@ -103,7 +103,6 @@ constexpr size_t kMaxVectorCount = 16;
   V(session_version_negotiation, SessionVersionNegotiation)                    \
   V(session_path_validation, SessionPathValidation)                            \
   V(stream_close, StreamClose)                                                 \
-  V(stream_error, StreamError)                                                 \
   V(stream_created, StreamCreated)                                             \
   V(stream_reset, StreamReset)                                                 \
   V(stream_headers, StreamHeaders)                                             \
@@ -304,6 +303,8 @@ struct CallbackScopeBase {
   ~CallbackScopeBase();
 };
 
+// Maintains a strong reference to BaseObject type ptr to keep it alive during
+// a MakeCallback during which it might be destroyed.
 template <typename T>
 struct CallbackScope final : public CallbackScopeBase {
   BaseObjectPtr<T> ref;

src/quic/streams.h

@@ -1,12 +1,18 @@
 #pragma once
 
+#include <cstdint>
 #if defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS
 #if HAVE_OPENSSL && NODE_OPENSSL_HAS_QUIC
 
+#include <aliased_struct.h>
 #include <async_wrap.h>
 #include <base_object.h>
+#include <dataqueue/queue.h>
 #include <env.h>
+#include <memory_tracker.h>
+#include <node_blob.h>
 #include <node_bob.h>
+#include <node_http_common.h>
 #include "bindingdata.h"
 #include "data.h"
 
@@ -17,27 +23,122 @@ class Session;
 
 using Ngtcp2Source = bob::SourceImpl<ngtcp2_vec>;
 
-// TODO(@jasnell): This is currently a placeholder for the actual definition.
-class Stream : public AsyncWrap, public Ngtcp2Source {
+// QUIC Stream's are simple data flows that may be:
+//
+// * Bidirectional (both sides can send) or Unidirectional (one side can send)
+// * Server or Client Initiated
+//
+// The flow direction and origin of the stream are important in determining the
+// write and read state (Open or Closed). Specifically:
+//
+// Bidirectional Stream States:
+// +--------+--------------+----------+----------+
+// |   ON   | Initiated By | Readable | Writable |
+// +--------+--------------+----------+----------+
+// | Server |   Server     |    Y     |    Y     |
+// +--------+--------------+----------+----------+
+// | Server |   Client     |    Y     |    Y     |
+// +--------+--------------+----------+----------+
+// | Client |   Server     |    Y     |    Y     |
+// +--------+--------------+----------+----------+
+// | Client |   Client     |    Y     |    Y     |
+// +--------+--------------+----------+----------+
+//
+// Unidirectional Stream States
+// +--------+--------------+----------+----------+
+// |   ON   | Initiated By | Readable | Writable |
+// +--------+--------------+----------+----------+
+// | Server |   Server     |    N     |    Y     |
+// +--------+--------------+----------+----------+
+// | Server |   Client     |    Y     |    N     |
+// +--------+--------------+----------+----------+
+// | Client |   Server     |    Y     |    N     |
+// +--------+--------------+----------+----------+
+// | Client |   Client     |    N     |    Y     |
+// +--------+--------------+----------+----------+
+//
+// All data sent via the Stream is buffered internally until either receipt is
+// acknowledged from the peer or attempts to send are abandoned. The fact that
+// data is buffered in memory makes it essential that the flow control for the
+// session and the stream are properly handled. For now, we are largely relying
+// on ngtcp2's default flow control mechanisms which generally should be doing
+// the right thing.
+//
+// A Stream may be in a fully closed state (No longer readable nor writable)
+// state but still have unacknowledged data in it's inbound and outbound
+// queues.
+//
+// A Stream is gracefully closed when (a) both read and write states are closed,
+// (b) all queued data has been acknowledged.
+//
+// The Stream may be forcefully closed immediately using destroy(err). This
+// causes all queued outbound data and pending JavaScript writes are abandoned,
+// and causes the Stream to be immediately closed at the ngtcp2 level without
+// waiting for any outstanding acknowledgements. Keep in mind, however, that the
+// peer is not notified that the stream is destroyed and may attempt to continue
+// sending data and acknowledgements.
+//
+// QUIC streams in general do not have headers. Some QUIC applications, however,
+// may associate headers with the stream (HTTP/3 for instance).
+class Stream : public AsyncWrap,
+               public Ngtcp2Source,
+               public DataQueue::BackpressureListener {
  public:
+  using Header = NgHeaderBase<BindingData>;
+
   static Stream* From(void* stream_user_data);
 
-  static BaseObjectPtr<Stream> Create(Session* session, int64_t id);
+  static bool HasInstance(Environment* env, v8::Local<v8::Value> value);
+  static v8::Local<v8::FunctionTemplate> GetConstructorTemplate(
+      Environment* env);
+  static void Initialize(Environment* env, v8::Local<v8::Object> target);
+  static void RegisterExternalReferences(ExternalReferenceRegistry* registry);
 
-  Stream(BaseObjectPtr<Session> session, v8::Local<v8::Object> obj);
+  static BaseObjectPtr<Stream> Create(
+      Session* session,
+      int64_t id,
+      std::shared_ptr<DataQueue> source = nullptr);
+
+  // The constructor is only public to be visible by MakeDetachedBaseObject.
+  // Call Create to create new instances of Stream.
+  Stream(BaseObjectWeakPtr<Session> session,
+         v8::Local<v8::Object> obj,
+         int64_t id,
+         std::shared_ptr<DataQueue> source);
+  ~Stream() override;
 
   int64_t id() const;
   Side origin() const;
   Direction direction() const;
+  Session& session() const;
 
   bool is_destroyed() const;
+
+  // True if we've completely sent all outbound data for this stream.
   bool is_eos() const;
 
+  bool is_readable() const;
+  bool is_writable() const;
+
+  // Called by the session/application to indicate that the specified number
+  // of bytes have been acknowledged by the peer.
   void Acknowledge(size_t datalen);
-  void Blocked();
   void Commit(size_t datalen);
-  void End();
-  void Destroy(QuicError error);
+  void EndWritable();
+  void EndReadable(std::optional<uint64_t> maybe_final_size = std::nullopt);
+  void EntryRead(size_t amount) override;
+
+  // Pulls data from the internal outbound DataQueue configured for this stream.
+  int DoPull(bob::Next<ngtcp2_vec> next,
+             int options,
+             ngtcp2_vec* data,
+             size_t count,
+             size_t max_count_hint) override;
+
+  // Forcefully close the stream immediately. All queued data and pending
+  // writes are abandoned, and the stream is immediately closed at the ngtcp2
+  // level without waiting for any outstanding acknowledgements.
+  void Destroy(QuicError error = QuicError());
 
   struct ReceiveDataFlags final {
     // Identifies the final chunk of data that the peer will send for the
@@ -53,11 +154,61 @@ class Stream : public AsyncWrap, public Ngtcp2Source {
   void ReceiveStopSending(QuicError error);
   void ReceiveStreamReset(uint64_t final_size, QuicError error);
 
+  void BeginHeaders(HeadersKind kind);
+  // Returns false if the header cannot be added. This will typically happen
+  // if the application does not support headers, a maximimum number of headers
+  // have already been added, or the maximum total header length is reached.
+  bool AddHeader(const Header& header);
+
   SET_NO_MEMORY_INFO()
   SET_MEMORY_INFO_NAME(Stream)
   SET_SELF_SIZE(Stream)
 
-  ListNode<Stream> stream_queue_;
+  struct State;
+  struct Stats;
+
+  // Notifies the JavaScript side that sending data on the stream has been
+  // blocked because of flow control restriction.
+  void EmitBlocked();
+
+ private:
+  struct Impl;
+  class Outbound;
+
+  // Gets a reader for the data received for this stream from the peer,
+  BaseObjectPtr<Blob::Reader> get_reader();
+
+  void set_final_size(uint64_t amount);
+  void set_outbound(std::shared_ptr<DataQueue> source);
+
+  // JavaScript callouts
+
+  // Notifies the JavaScript side that the stream has been destroyed.
+  void EmitClose(const QuicError& error);
+
+  // Delivers the set of inbound headers that have been collected.
+  void EmitHeaders();
+
+  // Notifies the JavaScript side that the stream has been reset.
+  void EmitReset(const QuicError& error);
+
+  // Notifies the JavaScript side that the application is ready to receive
+  // trailing headers.
+  void EmitWantTrailers();
+
+  AliasedStruct<Stats> stats_;
+  AliasedStruct<State> state_;
+  BaseObjectWeakPtr<Session> session_;
+  const Side origin_;
+  const Direction direction_;
+  std::unique_ptr<Outbound> outbound_;
+  std::shared_ptr<DataQueue> inbound_;
+
+  std::vector<v8::Local<v8::Value>> headers_;
+  HeadersKind headers_kind_ = HeadersKind::INITIAL;
+  size_t headers_length_ = 0;
+
+  friend struct Impl;
 
  public:
   // The Queue/Schedule/Unschedule here are part of the mechanism used to
@@ -69,6 +220,7 @@ class Stream : public AsyncWrap, public Ngtcp2Source {
   // data to send (such as when it is initially created or is using an async
   // source that is still waiting for data to be pushed) it will not appear in
   // the queue.
+  ListNode<Stream> stream_queue_;
   using Queue = ListHead<Stream, &Stream::stream_queue_>;
 
   void Schedule(Queue* queue);