diff --git a/doc/api/quic.md b/doc/api/quic.md
index 723c26c5bd0b99..563276e0863d03 100644
--- a/doc/api/quic.md
+++ b/doc/api/quic.md
@@ -475,7 +475,7 @@ added: v23.8.0
 -->
 
 * `options` {Object}
-  * `body` {ArrayBuffer | ArrayBufferView | Blob}
+  * `body` {ArrayBuffer | ArrayBufferView | Blob | ReadableStream<Uint8Array>}
   * `sendOrder` {number}
 * Returns: {Promise} for a {quic.QuicStream}
 
@@ -489,7 +489,7 @@ added: v23.8.0
 -->
 
 * `options` {Object}
-  * `body` {ArrayBuffer | ArrayBufferView | Blob}
+  * `body` {ArrayBuffer | ArrayBufferView | Blob | ReadableStream<Uint8Array>}
   * `sendOrder` {number}
 * Returns: {Promise} for a {quic.QuicStream}
 
@@ -820,7 +820,7 @@ The callback to invoke when the stream is reset. Read/write.
 added: v23.8.0
 -->
 
-* Type: {ReadableStream}
+* Type: {ReadableStream | undefined}
 
 ### `stream.session`
 
@@ -832,6 +832,16 @@ added: v23.8.0
 
 The session that created this stream. Read only.
 
+### `stream.setOutbound(outbound)`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* `outbound` {ArrayBuffer|SharedArrayBuffer|ArrayBufferView|Blob|ReadableStream}
+
+Set the outbound datasource.
+
 ### `stream.stats`
 
 <!-- YAML
diff --git a/lib/internal/blob.js b/lib/internal/blob.js
index 46abb4f793c465..0fa81ec1f971ec 100644
--- a/lib/internal/blob.js
+++ b/lib/internal/blob.js
@@ -475,6 +475,14 @@ function createBlobReaderStream(reader) {
           if (buffer !== undefined && buffer.byteLength !== 0) {
             c.enqueue(new Uint8Array(buffer));
           }
+          if (status === 3) {
+            // status wait!
+            // the source will call us (this function!) again !
+            // we should not schedule another readnext,
+            // as we end up in a busy loop, calling pull over and
+            // over and pilling up next callbacks in dataqueue
+            return;
+          }
           // We keep reading until we either reach EOS, some error, or we
           // hit the flow rate of the stream (c.desiredSize).
           // We use set immediate here because we have to allow the event
diff --git a/lib/internal/quic/quic.js b/lib/internal/quic/quic.js
index 459906d270dab6..0837861140b9e4 100644
--- a/lib/internal/quic/quic.js
+++ b/lib/internal/quic/quic.js
@@ -57,6 +57,10 @@ const {
   CLOSECONTEXT_START_FAILURE: kCloseContextStartFailure,
 } = internalBinding('quic');
 
+const {
+  DataQueueFeeder,
+} = internalBinding('dataqueuefeeder');
+
 const {
   isArrayBuffer,
   isArrayBufferView,
@@ -114,6 +118,10 @@ const {
   buildNgHeaderString,
 } = require('internal/http2/util');
 
+const {
+  isReadableStream,
+} = require('internal/webstreams/readablestream');
+
 const kEmptyObject = { __proto__: null };
 
 const {
@@ -546,6 +554,37 @@ setCallbacks({
 function validateBody(body) {
   // TODO(@jasnell): Support streaming sources
   if (body === undefined) return body;
+  if (isReadableStream(body)) {
+    const feeder = new DataQueueFeeder();
+    const reader = body.getReader();
+
+    const feeding = async () => {
+      await feeder.ready();
+      let cont = true;
+
+      while (cont) {
+        let read;
+        try {
+          read = await reader.read();
+        } catch (error) {
+          feeder.error(error);
+        }
+        const { value, done } = read;
+        try {
+          cont = await feeder.submit(value, done);
+        } catch (error) {
+          reader.cancel(error.toString());
+          break;
+        }
+      }
+      if (!cont) {
+        reader.releaseLock();
+      }
+
+    };
+    feeding();
+    return feeder;
+  }
   // Transfer ArrayBuffers...
   if (isArrayBuffer(body)) {
     return ArrayBufferPrototypeTransfer(body);
@@ -578,6 +617,7 @@ function validateBody(body) {
     'ArrayBuffer',
     'ArrayBufferView',
     'Blob',
+    'ReadableStream',
   ], body);
 }
 
@@ -802,10 +842,10 @@ class QuicStream {
   /**
    * Sets the outbound data source for the stream. This can only be called
    * once and must be called before any data will be sent. The body can be
-   * an ArrayBuffer, a TypedArray or DataView, or a Blob. If the stream
-   * is destroyed or already has an outbound data source, an error will
-   * be thrown.
-   * @param {ArrayBuffer|SharedArrayBuffer|ArrayBufferView|Blob} outbound
+   * an ArrayBuffer, a TypedArray or DataView, or a Blob, a ReadableStream.
+   * If the stream is destroyed or already has an outbound data source,
+   * an error will be thrown.
+   * @param {ArrayBuffer|SharedArrayBuffer|ArrayBufferView|Blob|ReadableStream} outbound
    */
   setOutbound(outbound) {
     QuicStream.#assertIsQuicStream(this);
diff --git a/src/dataqueue/queue.cc b/src/dataqueue/queue.cc
index 537844806d3087..0f28fbec4ba941 100644
--- a/src/dataqueue/queue.cc
+++ b/src/dataqueue/queue.cc
@@ -20,10 +20,18 @@
 
 namespace node {
 
+using v8::ArrayBuffer;
 using v8::ArrayBufferView;
 using v8::BackingStore;
+using v8::FunctionCallbackInfo;
+using v8::FunctionTemplate;
+using v8::Isolate;
 using v8::Local;
 using v8::Object;
+using v8::ObjectTemplate;
+using v8::Promise;
+using v8::TypedArray;
+using v8::Uint8Array;
 using v8::Value;
 
 namespace {
@@ -34,25 +42,34 @@ class NonIdempotentDataQueueReader;
 class EntryImpl : public DataQueue::Entry {
  public:
   virtual std::shared_ptr<DataQueue::Reader> get_reader() = 0;
+  virtual void clearPendingNext() = 0;
 };
 
 class DataQueueImpl final : public DataQueue,
                             public std::enable_shared_from_this<DataQueueImpl> {
  public:
   // Constructor for an idempotent, fixed sized DataQueue.
-  DataQueueImpl(std::vector<std::unique_ptr<Entry>>&& list, uint64_t size)
+  DataQueueImpl(Environment* env,
+                std::vector<std::unique_ptr<Entry>>&& list,
+                uint64_t size)
       : entries_(std::move(list)),
         idempotent_(true),
         size_(size),
-        capped_size_(0) {}
+        capped_size_(0),
+        notifier_(nullptr),
+        env_(env) {}
 
   // Constructor for a non-idempotent DataQueue. This kind of queue can have
   // entries added to it over time. The size is set to 0 initially. The queue
   // can be capped immediately on creation. Depending on the entries that are
   // added, the size can be cleared if any of the entries are not capable of
   // providing a size.
-  DataQueueImpl(std::optional<uint64_t> cap = std::nullopt)
-      : idempotent_(false), size_(0), capped_size_(cap) {}
+  DataQueueImpl(Environment* env, std::optional<uint64_t> cap = std::nullopt)
+      : idempotent_(false),
+        size_(0),
+        capped_size_(cap),
+        notifier_(nullptr),
+        env_(env) {}
 
   // Disallow moving and copying.
   DataQueueImpl(const DataQueueImpl&) = delete;
@@ -100,7 +117,7 @@ class DataQueueImpl final : public DataQueue,
       }
     }
 
-    return std::make_shared<DataQueueImpl>(std::move(slices), len);
+    return std::make_shared<DataQueueImpl>(env_, std::move(slices), len);
   }
 
   std::optional<uint64_t> size() const override { return size_; }
@@ -132,6 +149,12 @@ class DataQueueImpl final : public DataQueue,
     }
 
     entries_.push_back(std::move(entry));
+    if (notifier_) {
+      notifier_->newDataOrCloseAdded();
+    }
+    for (auto cur = readers_.begin(); cur != readers_.end(); cur++) {
+      (*cur)->newDataOrEnd();
+    }
     return true;
   }
 
@@ -141,11 +164,18 @@ class DataQueueImpl final : public DataQueue,
     // cap again with a smaller size.
     if (capped_size_.has_value()) {
       capped_size_ = std::min(limit, capped_size_.value());
+      if (notifier_) notifier_->newDataOrCloseAdded();
+      for (auto cur = readers_.begin(); cur != readers_.end(); cur++) {
+        (*cur)->newDataOrEnd();
+      }
       return;
     }
-
     // Otherwise just set the limit.
     capped_size_ = limit;
+    if (notifier_) notifier_->newDataOrCloseAdded();
+    for (auto cur = readers_.begin(); cur != readers_.end(); cur++) {
+      (*cur)->newDataOrEnd();
+    }
   }
 
   std::optional<uint64_t> maybeCapRemaining() const override {
@@ -157,11 +187,19 @@ class DataQueueImpl final : public DataQueue,
     return std::nullopt;
   }
 
+  void clearPendingNext() override {
+    for (const auto& entry : entries_) {
+      entry->clearPendingNext();
+    }
+  }
+
   void MemoryInfo(node::MemoryTracker* tracker) const override {
     tracker->TrackField(
         "entries", entries_, "std::vector<std::unique_ptr<Entry>>");
   }
 
+  void SetNotifier(Notifier* notifier) override { notifier_ = notifier; }
+
   void addBackpressureListener(BackpressureListener* listener) override {
     if (idempotent_) return;
     DCHECK_NOT_NULL(listener);
@@ -183,6 +221,17 @@ class DataQueueImpl final : public DataQueue,
     return !backpressure_listeners_.empty();
   }
 
+  Environment* env() { return env_; }
+
+  class NotifyReader {
+   public:
+    virtual void newDataOrEnd() = 0;
+  };
+
+  void reader_destructed(NotifyReader* reader) { readers_.erase(reader); }
+
+  void SetEnvironment(Environment* env) override { env_ = env; }
+
   std::shared_ptr<Reader> get_reader() override;
   SET_MEMORY_INFO_NAME(DataQueue)
   SET_SELF_SIZE(DataQueueImpl)
@@ -193,9 +242,13 @@ class DataQueueImpl final : public DataQueue,
   std::optional<uint64_t> size_ = std::nullopt;
   std::optional<uint64_t> capped_size_ = std::nullopt;
   bool locked_to_reader_ = false;
+  std::set<NotifyReader*> readers_;  // we may need to notify them
 
   std::unordered_set<BackpressureListener*> backpressure_listeners_;
 
+  Notifier* notifier_;
+  Environment* env_;
+
   friend class DataQueue;
   friend class IdempotentDataQueueReader;
   friend class NonIdempotentDataQueueReader;
@@ -207,6 +260,7 @@ class DataQueueImpl final : public DataQueue,
 // will not and cannot be changed.
 class IdempotentDataQueueReader final
     : public DataQueue::Reader,
+      public DataQueueImpl::NotifyReader,
       public std::enable_shared_from_this<IdempotentDataQueueReader> {
  public:
   IdempotentDataQueueReader(std::shared_ptr<DataQueueImpl> data_queue)
@@ -214,6 +268,8 @@ class IdempotentDataQueueReader final
     CHECK(data_queue_->is_idempotent());
   }
 
+  ~IdempotentDataQueueReader() { data_queue_->reader_destructed(this); }
+
   // Disallow moving and copying.
   IdempotentDataQueueReader(const IdempotentDataQueueReader&) = delete;
   IdempotentDataQueueReader(IdempotentDataQueueReader&&) = delete;
@@ -221,6 +277,10 @@ class IdempotentDataQueueReader final
       delete;
   IdempotentDataQueueReader& operator=(IdempotentDataQueueReader&&) = delete;
 
+  void newDataOrEnd() override {
+    // Currently nothing, but it may change
+  }
+
   int Pull(Next next,
            int options,
            DataQueue::Vec* data,
@@ -269,7 +329,10 @@ class IdempotentDataQueueReader final
     int status = current_reader->Pull(
         [this, next = std::move(next)](
             int status, const DataQueue::Vec* vecs, uint64_t count, Done done) {
-          pull_pending_ = false;
+          if (status != bob::Status::STATUS_WAIT) {
+            // do not set for an async call
+            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 ||
@@ -305,8 +368,11 @@ class IdempotentDataQueueReader final
 
     // The pull was handled synchronously. If we're not ended, we want to
     // make sure status returned is CONTINUE.
+    // But if the source blocks, we should let the caller know,
+    // otherwise, we may provoke a busy loop
     if (!pull_pending_) {
-      if (!ended_) return bob::Status::STATUS_CONTINUE;
+      if (!ended_ && status != bob::Status::STATUS_BLOCK)
+        return bob::Status::STATUS_CONTINUE;
       // For all other status, we just fall through and return it straightaway.
     }
 
@@ -317,6 +383,7 @@ class IdempotentDataQueueReader final
     //                              pull but it is not available yet. The
     //                              caller should not keep calling pull for
     //                              now but may check again later.
+    //                              Though we may call a notifier to signal.
     //  bob::Status::STATUS_WAIT - means that the entry has more data to
     //                             pull but it won't be provided
     //                             synchronously, instead the next() callback
@@ -359,6 +426,7 @@ class IdempotentDataQueueReader final
 // is mutated as the read proceeds.
 class NonIdempotentDataQueueReader final
     : public DataQueue::Reader,
+      public DataQueueImpl::NotifyReader,
       public std::enable_shared_from_this<NonIdempotentDataQueueReader> {
  public:
   NonIdempotentDataQueueReader(std::shared_ptr<DataQueueImpl> data_queue)
@@ -366,6 +434,8 @@ class NonIdempotentDataQueueReader final
     CHECK(!data_queue_->is_idempotent());
   }
 
+  ~NonIdempotentDataQueueReader() { data_queue_->reader_destructed(this); }
+
   // Disallow moving and copying.
   NonIdempotentDataQueueReader(const NonIdempotentDataQueueReader&) = delete;
   NonIdempotentDataQueueReader(NonIdempotentDataQueueReader&&) = delete;
@@ -374,12 +444,25 @@ class NonIdempotentDataQueueReader final
   NonIdempotentDataQueueReader& operator=(NonIdempotentDataQueueReader&&) =
       delete;
 
+  void newDataOrEnd() override {
+    if (waited_next_) {
+      auto next = std::move(waited_next_);
+      data_queue_->env()->SetImmediate(
+          [next, dropme = shared_from_this()](Environment* env) {
+            std::move(next)(
+                bob::Status::STATUS_CONTINUE, nullptr, 0, [](uint64_t) {});
+          },
+          CallbackFlags::kUnrefed);
+    }
+  }
+
   int Pull(Next next,
            int options,
            DataQueue::Vec* data,
            size_t count,
            size_t max_count_hint = bob::kMaxCountHint) override {
     std::shared_ptr<DataQueue::Reader> self = shared_from_this();
+    assert(!waited_next_);  // Do not call us with blocked next
 
     // If ended is true, this reader has already reached the end and cannot
     // provide any more data.
@@ -397,8 +480,17 @@ class NonIdempotentDataQueueReader final
       // that'll happe, 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) {});
-        return bob::STATUS_BLOCK;
+        // Do we have to call next?
+        if (!(options & bob::OPTIONS_SYNC)) {
+          assert(!waited_next_);
+          next(bob::Status::STATUS_WAIT, nullptr, 0, [](uint64_t) {});
+          waited_next_ = std::move(next);
+          return bob::STATUS_WAIT;
+        } else {
+          std::move(next)(
+              bob::Status::STATUS_BLOCK, nullptr, 0, [](uint64_t) {});
+          return bob::STATUS_BLOCK;
+        }
       }
 
       // However, if we are capped, the status will depend on whether the size
@@ -409,9 +501,16 @@ class NonIdempotentDataQueueReader final
         // still might get more data. We just don't know exactly when that'll
         // come, so let's return a blocked status.
         if (data_queue_->size().value() < data_queue_->capped_size_.value()) {
-          std::move(next)(
-              bob::Status::STATUS_BLOCK, nullptr, 0, [](uint64_t) {});
-          return bob::STATUS_BLOCK;
+          if (!(options & bob::OPTIONS_SYNC)) {
+            assert(!waited_next_);
+            next(bob::Status::STATUS_WAIT, nullptr, 0, [](uint64_t) {});
+            waited_next_ = std::move(next);
+            return bob::STATUS_WAIT;
+          } else {
+            std::move(next)(
+                bob::Status::STATUS_BLOCK, nullptr, 0, [](uint64_t) {});
+            return bob::STATUS_BLOCK;
+          }
         }
 
         // Otherwise, if size is equal to or greater than capped, we are done.
@@ -437,7 +536,7 @@ class NonIdempotentDataQueueReader final
     CHECK(!pull_pending_);
     pull_pending_ = true;
     int status = current_reader->Pull(
-        [this, next = std::move(next)](
+        [this, next = std::move(next), options](
             int status, const DataQueue::Vec* vecs, uint64_t count, Done done) {
           pull_pending_ = false;
 
@@ -448,10 +547,32 @@ class NonIdempotentDataQueueReader final
                             status == bob::Status::STATUS_EOS,
                         vecs == nullptr && count == 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;
+            // must be done before erasing the entries
+            // as FdEntry's and FeederEntry's reader hold
+            // a pointer to the entry! and FeederEntry
+            // invoke it in destructor
+            data_queue_->entries_.erase(data_queue_->entries_.begin());
+            // if the DataQueue is ended does not only depend
+            // if entries_ is empty, but also if it is capped!
+            if (data_queue_->entries_.empty()) {
+              if (data_queue_->is_capped()) {
+                ended_ = true;
+              } else {
+                if (!(options & bob::OPTIONS_SYNC)) {
+                  assert(!waited_next_);
+                  next(bob::Status::STATUS_WAIT, nullptr, 0, [](uint64_t) {});
+                  waited_next_ = std::move(next);
+                  return;
+                } else {
+                  std::move(next)(
+                      bob::Status::STATUS_BLOCK, nullptr, 0, [](uint64_t) {});
+                  return;
+                }
+              }
+            } else {
+              status = bob::Status::STATUS_CONTINUE;
+            }
             std::move(next)(status, nullptr, 0, [](uint64_t) {});
             return;
           }
@@ -478,7 +599,9 @@ class NonIdempotentDataQueueReader final
 
     if (!pull_pending_) {
       // The callback was resolved synchronously. Let's check our status.
-      if (!ended_) return bob::Status::STATUS_CONTINUE;
+      if (!ended_) {
+        return bob::Status::STATUS_CONTINUE;
+      }
       // For all other status, we just fall through and return it straightaway.
     }
 
@@ -489,6 +612,7 @@ class NonIdempotentDataQueueReader final
     //                              pull but it is not available yet. The
     //                              caller should not keep calling pull for
     //                              now but may check again later.
+    //                              Though we may call a notifier to signal.
     //  bob::Status::STATUS_WAIT - means that the entry has more data to
     //                             pull but it won't be provided
     //                             synchronously, instead the next() callback
@@ -519,17 +643,24 @@ class NonIdempotentDataQueueReader final
   std::shared_ptr<DataQueue::Reader> current_reader_ = nullptr;
   bool ended_ = false;
   bool pull_pending_ = false;
+  Next waited_next_;
 };
 
 std::shared_ptr<DataQueue::Reader> DataQueueImpl::get_reader() {
   if (is_idempotent()) {
-    return std::make_shared<IdempotentDataQueueReader>(shared_from_this());
+    auto reader =
+        std::make_shared<IdempotentDataQueueReader>(shared_from_this());
+    readers_.insert(reader.get());
+    return reader;
   }
 
   if (locked_to_reader_) return nullptr;
   locked_to_reader_ = true;
 
-  return std::make_shared<NonIdempotentDataQueueReader>(shared_from_this());
+  auto reader =
+      std::make_shared<NonIdempotentDataQueueReader>(shared_from_this());
+  readers_.insert(reader.get());
+  return reader;
 }
 
 // ============================================================================
@@ -584,6 +715,10 @@ class EmptyEntry final : public EntryImpl {
     return std::make_unique<EmptyEntry>();
   }
 
+  void clearPendingNext() override {
+    // this is a noop for an empty object
+  }
+
   std::optional<uint64_t> size() const override { return 0; }
 
   bool is_idempotent() const override { return true; }
@@ -696,6 +831,10 @@ class InMemoryEntry final : public EntryImpl {
     return makeEntry(start, byte_length_ - start);
   }
 
+  void clearPendingNext() override {
+    // this is a noop for an object, that immediately calls next.
+  }
+
   std::optional<uint64_t> size() const override { return byte_length_; }
 
   bool is_idempotent() const override { return true; }
@@ -744,6 +883,11 @@ class DataQueueEntry : public EntryImpl {
     return std::make_unique<DataQueueEntry>(std::move(sliced));
   }
 
+  void clearPendingNext() override {
+    // this is just calls clearPendingNext of the queue
+    data_queue_->clearPendingNext();
+  }
+
   // Returns the number of bytes represented by this Entry if it is
   // known. Certain types of entries, such as those backed by streams
   // might not know the size in advance and therefore cannot provide
@@ -862,6 +1006,8 @@ class FdEntry final : public EntryImpl {
     return std::make_unique<FdEntry>(env_, path_, stat_, new_start, new_end);
   }
 
+  void clearPendingNext() override { clearPendingNextImpl(); }
+
   std::optional<uint64_t> size() const override { return end_ - start_; }
 
   bool is_idempotent() const override { return true; }
@@ -879,6 +1025,16 @@ class FdEntry final : public EntryImpl {
   uint64_t start_ = 0;
   uint64_t end_ = 0;
 
+  class ReaderImpl;
+  std::unordered_set<ReaderImpl*> readers_;
+
+  void clearPendingNextImpl() {
+    // this should clear all pending pulls from the FD reader
+    for (ReaderImpl* p : readers_) {
+      p->ClearAllPendingPulls();
+    }
+  }
+
   bool is_modified(const uv_stat_t& other) {
     return other.st_size != stat_.st_size ||
            other.st_mtim.tv_nsec != stat_.st_mtim.tv_nsec;
@@ -926,12 +1082,14 @@ class FdEntry final : public EntryImpl {
         : env_(handle->env()), handle_(std::move(handle)), entry_(entry) {
       handle_->PushStreamListener(this);
       handle_->env()->AddCleanupHook(cleanup, this);
+      entry_->readers_.insert(this);
     }
 
     ~ReaderImpl() override {
       handle_->env()->RemoveCleanupHook(cleanup, this);
       DrainAndClose();
       handle_->RemoveStreamListener(this);
+      entry_->readers_.erase(this);
     }
 
     uv_buf_t OnStreamAlloc(size_t suggested_size) override {
@@ -996,6 +1154,12 @@ class FdEntry final : public EntryImpl {
         std::move(next)(UV_EINVAL, nullptr, 0, [](uint64_t) {});
         return UV_EINVAL;
       }
+      // Note: we do not support sync reading, so if it is requested
+      // we need to bail out and say we are blocked
+      if ((options & bob::OPTIONS_SYNC)) {
+        std::move(next)(bob::Status::STATUS_BLOCK, nullptr, 0, [](uint64_t) {});
+        return bob::STATUS_BLOCK;
+      }
 
       pending_pulls_.emplace_back(std::move(next), shared_from_this());
       if (!reading_) {
@@ -1056,18 +1220,18 @@ class FdEntry final : public EntryImpl {
       return std::move(pending_pulls_.front());
     }
 
+    void ClearAllPendingPulls() { pending_pulls_.clear(); }
+
     friend class FdEntry;
   };
 
   friend class ReaderImpl;
 };
-
-// ============================================================================
-
 }  // namespace
+// ============================================================================
 
 std::shared_ptr<DataQueue> DataQueue::CreateIdempotent(
-    std::vector<std::unique_ptr<Entry>> list) {
+    Environment* env, std::vector<std::unique_ptr<Entry>> list) {
   // Any entry is invalid for an idempotent DataQueue if any of the entries
   // are nullptr or is not idempotent.
   uint64_t size = 0;
@@ -1094,11 +1258,12 @@ std::shared_ptr<DataQueue> DataQueue::CreateIdempotent(
     return nullptr;
   }
 
-  return std::make_shared<DataQueueImpl>(std::move(list), size);
+  return std::make_shared<DataQueueImpl>(env, std::move(list), size);
 }
 
-std::shared_ptr<DataQueue> DataQueue::Create(std::optional<uint64_t> capped) {
-  return std::make_shared<DataQueueImpl>(capped);
+std::shared_ptr<DataQueue> DataQueue::Create(Environment* env,
+                                             std::optional<uint64_t> capped) {
+  return std::make_shared<DataQueueImpl>(env, capped);
 }
 
 std::unique_ptr<DataQueue::Entry> DataQueue::CreateInMemoryEntryFromView(
@@ -1147,4 +1312,210 @@ void DataQueue::RegisterExternalReferences(
   // Nothing to do here currently.
 }
 
+DataQueueFeeder::DataQueueFeeder(Environment* env, Local<Object> object)
+    : AsyncWrap(env, object), buffer_size_(0), bytecount(0) {
+  MakeWeak();
+}
+
+void DataQueueFeeder::setDataQueue(std::shared_ptr<DataQueue> queue) {
+  queue->SetEnvironment(env());
+  dataQueue_ = queue;
+
+  // Allows us to be notified when data is actually read from the
+  // queue so that we can limit the data hold in DataQueueFeeder
+  dataQueue_->addBackpressureListener(this);
+}
+
+void DataQueueFeeder::tryWakePulls() {
+  if (!readFinish_.IsEmpty()) {
+    v8::Local<v8::Promise::Resolver> resolver =
+        readFinish_.Get(env()->isolate());
+    // I do not think, that this can error...
+    [[maybe_unused]] v8::Maybe<bool> ignoredResult =
+        resolver->Resolve(env()->context(), v8::True(env()->isolate()));
+    readFinish_.Reset();
+  }
+}
+
+void DataQueueFeeder::EntryRead(size_t amount) {
+  buffer_size_ -= amount;
+  if (!Full() && !readFinish_.IsEmpty()) {
+    Local<v8::Promise::Resolver> resolver = readFinish_.Get(env()->isolate());
+    [[maybe_unused]] v8::Maybe<bool> ignoredResult =
+        resolver->Resolve(env()->context(), v8::True(env()->isolate()));
+    readFinish_.Reset();
+  }
+}
+
+void DataQueueFeeder::DrainAndClose() {
+  if (done) return;
+  done = true;
+  // do not do this several time, and note,
+  // it may be called several times.
+  if (dataQueue_) {
+    dataQueue_->cap();
+  }
+  if (!readFinish_.IsEmpty()) {
+    Local<v8::Promise::Resolver> resolver = readFinish_.Get(env()->isolate());
+    [[maybe_unused]] v8::Maybe<bool> ignoredResult =
+        resolver->Resolve(env()->context(), v8::False(env()->isolate()));
+    readFinish_.Reset();
+  }
+}
+
+void DataQueueFeeder::New(const FunctionCallbackInfo<Value>& args) {
+  DCHECK(args.IsConstructCall());
+  auto env = Environment::GetCurrent(args);
+  new DataQueueFeeder(env, args.This());
+}
+
+void DataQueueFeeder::Ready(const FunctionCallbackInfo<Value>& args) {
+  Environment* env = Environment::GetCurrent(args);
+  DataQueueFeeder* feeder;
+  ASSIGN_OR_RETURN_UNWRAP(&feeder, args.This());
+  if (feeder->Full()) {
+    // FIXME(martenrichter): may be use a queue for the promises,
+    // this is necessary, if used outside of quic
+    Local<Promise::Resolver> readFinish =
+        Promise::Resolver::New(env->context()).ToLocalChecked();
+    feeder->readFinish_.Reset(env->isolate(), readFinish);
+    args.GetReturnValue().Set(readFinish->GetPromise());
+  } else {
+    // may be not required
+    feeder->readFinish_.Reset();
+  }
+}
+
+void DataQueueFeeder::Submit(const FunctionCallbackInfo<Value>& args) {
+  Environment* env = Environment::GetCurrent(args);
+  DataQueueFeeder* feeder;
+  ASSIGN_OR_RETURN_UNWRAP(&feeder, args.This());
+
+  bool done = false;
+  if (args[1]->IsBoolean() && args[1].As<v8::Boolean>()->Value()) {
+    done = true;
+  }
+  if (!args[0].IsEmpty() && !args[0]->IsUndefined() && !args[0]->IsNull()) {
+    CHECK(!feeder->Full());
+    auto chunk = args[0];
+
+    if (chunk->IsArrayBuffer()) {
+      auto buffer = chunk.As<ArrayBuffer>();
+      chunk = Uint8Array::New(buffer, 0, buffer->ByteLength());
+    }
+    if (!chunk->IsTypedArray()) {
+      THROW_ERR_INVALID_ARG_TYPE(
+          env, "Invalid data must be Arraybuffer or TypedArray");
+      return;
+    }
+    Local<TypedArray> typedArray = chunk.As<TypedArray>();
+    // now we create a copy
+    // detaching, would not be a good idea for example, such
+    // a limitation is not given with W3C Webtransport
+    // if we do not do it here, a transform stream would
+    // be needed to do the copy in the Webtransport case.
+    // there may be also troubles, if multiple Uint8Array
+    // are derived in a parser from a single ArrayBuffer
+    size_t nread = typedArray->ByteLength();
+    auto backingUniq = v8::ArrayBuffer::NewBackingStore(
+        env->isolate(),
+        nread,
+        v8::BackingStoreInitializationMode::kUninitialized,
+        v8::BackingStoreOnFailureMode::kReturnNull);
+    if (!backingUniq) {
+      THROW_ERR_MEMORY_ALLOCATION_FAILED(env);
+      return;
+    }
+    std::shared_ptr<BackingStore> backing = std::move(backingUniq);
+
+    auto originalStore = typedArray->Buffer()->GetBackingStore();
+    const void* originalData =
+        static_cast<char*>(originalStore->Data()) + typedArray->ByteOffset();
+    memcpy(backing->Data(), originalData, nread);
+    feeder->buffer_size_ += nread;
+    feeder->dataQueue_->append(DataQueue::CreateInMemoryEntryFromBackingStore(
+        std::move(backing), 0, nread));
+    feeder->bytecount += nread;
+  }
+  if (done) {
+    feeder->DrainAndClose();
+    feeder->readFinish_.Reset();
+    args.GetReturnValue().Set(v8::False(env->isolate()));
+    return;
+  } else {
+    if (!feeder->Full()) {
+      feeder->readFinish_.Reset();
+      args.GetReturnValue().Set(v8::True(env->isolate()));
+      return;
+    } else {
+      Local<Promise::Resolver> readFinish =
+          Promise::Resolver::New(env->context()).ToLocalChecked();
+      feeder->readFinish_.Reset(env->isolate(), readFinish);
+      args.GetReturnValue().Set(readFinish->GetPromise());
+      return;
+    }
+  }
+}
+
+void DataQueueFeeder::Error(const FunctionCallbackInfo<Value>& args) {
+  DataQueueFeeder* feeder;
+  ASSIGN_OR_RETURN_UNWRAP(&feeder, args.This());
+  // FIXME, how should I pass on the error
+  // ResetStream must be send also
+  feeder->DrainAndClose();
+}
+
+bool DataQueueFeeder::HasInstance(Environment* env, Local<Value> object) {
+  return GetConstructorTemplate(env)->HasInstance(object);
+}
+
+Local<FunctionTemplate> DataQueueFeeder::GetConstructorTemplate(
+    Environment* env) {
+  Local<FunctionTemplate> tmpl = env->data_queue_feeder_constructor_template();
+  if (tmpl.IsEmpty()) {
+    Isolate* isolate = env->isolate();
+    tmpl = NewFunctionTemplate(isolate, DataQueueFeeder::New);
+    tmpl->InstanceTemplate()->SetInternalFieldCount(
+        BaseObject::kInternalFieldCount);
+    tmpl->SetClassName(
+        FIXED_ONE_BYTE_STRING(env->isolate(), "DataQueueFeeder"));
+    tmpl->Inherit(AsyncWrap::GetConstructorTemplate(env));
+    env->set_data_queue_feeder_constructor_template(tmpl);
+    SetProtoMethod(isolate, tmpl, "error", Error);
+    SetProtoMethod(isolate, tmpl, "submit", Submit);
+    SetProtoMethod(isolate, tmpl, "ready", Ready);
+  }
+  return tmpl;
+}
+
+void DataQueueFeeder::CreatePerIsolateProperties(
+    IsolateData* isolate_data, v8::Local<v8::ObjectTemplate> target) {}
+
+void DataQueueFeeder::CreatePerContextProperties(v8::Local<v8::Object> target,
+                                                 v8::Local<v8::Value> unused,
+                                                 v8::Local<v8::Context> context,
+                                                 void* priv) {
+  Environment* env = Environment::GetCurrent(context);
+  SetConstructorFunction(context,
+                         target,
+                         "DataQueueFeeder",
+                         GetConstructorTemplate(env),
+                         SetConstructorFunctionFlag::NONE);
+}
+
+void DataQueueFeeder::RegisterExternalReferences(
+    ExternalReferenceRegistry* registry) {
+  registry->Register(DataQueueFeeder::New);
+  registry->Register(DataQueueFeeder::Submit);
+  registry->Register(DataQueueFeeder::Error);
+  registry->Register(DataQueueFeeder::Ready);
+}
+
 }  // namespace node
+
+NODE_BINDING_CONTEXT_AWARE_INTERNAL(
+    dataqueuefeeder, node::DataQueueFeeder::CreatePerContextProperties)
+NODE_BINDING_PER_ISOLATE_INIT(dataqueuefeeder,
+                              node::DataQueueFeeder::CreatePerIsolateProperties)
+NODE_BINDING_EXTERNAL_REFERENCE(
+    dataqueuefeeder, node::DataQueueFeeder::RegisterExternalReferences)
diff --git a/src/dataqueue/queue.h b/src/dataqueue/queue.h
index a37bd27549986e..5654723a4a7a74 100644
--- a/src/dataqueue/queue.h
+++ b/src/dataqueue/queue.h
@@ -2,7 +2,9 @@
 
 #if defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS
 
+#include <async_wrap.h>
 #include <base_object.h>
+#include <env-inl.h>
 #include <memory_tracker.h>
 #include <node.h>
 #include <node_bob.h>
@@ -16,6 +18,8 @@
 #include <vector>
 
 namespace node {
+using v8::Local;
+using v8::Value;
 
 // Represents a sequenced collection of data sources that can be
 // consumed as a single logical stream of data. Sources can be
@@ -124,6 +128,8 @@ namespace node {
 // For non-idempotent DataQueues, only a single reader is ever allowed for
 // the DataQueue, and the data can only ever be read once.
 
+class DataQueueFeeder;
+
 class DataQueue : public MemoryRetainer {
  public:
   struct Vec {
@@ -149,6 +155,12 @@ class DataQueue : public MemoryRetainer {
     virtual void EntryRead(size_t amount) = 0;
   };
 
+  // A Notifier, that is called, when data is added or cap is called
+  class Notifier {
+   public:
+    virtual void newDataOrCloseAdded() = 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.
@@ -186,19 +198,23 @@ class DataQueue : public MemoryRetainer {
     // idempotent and cannot preserve that quality, subsequent reads
     // must fail with an error when a variance is detected.
     virtual bool is_idempotent() const = 0;
+
+    // calls if required readers to invalidate next function's
+    // which may hold captures to stream and session objects
+    virtual void clearPendingNext() = 0;
   };
 
   // Creates an idempotent DataQueue with a pre-established collection
   // of entries. All of the entries must also be idempotent otherwise
   // an empty std::unique_ptr will be returned.
   static std::shared_ptr<DataQueue> CreateIdempotent(
-      std::vector<std::unique_ptr<Entry>> list);
+      Environment* env, std::vector<std::unique_ptr<Entry>> list);
 
   // Creates a non-idempotent DataQueue. This kind of queue can be
   // mutated and updated such that multiple reads are not guaranteed
   // to produce the same result. The entries added can be of any type.
   static std::shared_ptr<DataQueue> Create(
-      std::optional<uint64_t> capped = std::nullopt);
+      Environment* env, std::optional<uint64_t> capped = std::nullopt);
 
   // Creates an idempotent Entry from a v8::ArrayBufferView. To help
   // ensure idempotency, the underlying ArrayBuffer is detached from
@@ -293,14 +309,76 @@ class DataQueue : public MemoryRetainer {
   // been set, maybeCapRemaining() will return std::nullopt.
   virtual std::optional<uint64_t> maybeCapRemaining() const = 0;
 
+  // calls all entries and readers to invalidate next function's
+  // which may hold captures to stream and session objects
+  virtual void clearPendingNext() = 0;
+
   // BackpressureListeners only work on non-idempotent DataQueues.
   virtual void addBackpressureListener(BackpressureListener* listener) = 0;
   virtual void removeBackpressureListener(BackpressureListener* listener) = 0;
 
+  // Set a notifier, e. g. to schedule packet sending
+  virtual void SetNotifier(Notifier* notfier) = 0;
+  // sets the environment in order to schedule functions
+  virtual void SetEnvironment(Environment* env) = 0;
+
   static void Initialize(Environment* env, v8::Local<v8::Object> target);
   static void RegisterExternalReferences(ExternalReferenceRegistry* registry);
 };
 
+class DataQueueFeeder final : public AsyncWrap,
+                              public DataQueue::BackpressureListener {
+ public:
+  using Next = bob::Next<DataQueue::Vec>;
+
+  DataQueueFeeder(Environment* env, v8::Local<v8::Object> object);
+
+  static void RegisterExternalReferences(ExternalReferenceRegistry* registry);
+
+  static BaseObjectPtr<DataQueueFeeder> Create();
+
+  void setDataQueue(std::shared_ptr<DataQueue> queue);
+
+  bool Done() { return done; }
+  bool Full() { return buffer_size_ >= max_buffer_size_; }
+
+  void DrainAndClose();
+  void tryWakePulls();
+
+  void EntryRead(size_t amount) override;
+
+  SET_NO_MEMORY_INFO()
+  SET_MEMORY_INFO_NAME(DataQueueFeeder)
+  SET_SELF_SIZE(DataQueueFeeder)
+
+  static void New(const v8::FunctionCallbackInfo<v8::Value>& args);
+  static void Submit(const v8::FunctionCallbackInfo<v8::Value>& args);
+  static void Error(const v8::FunctionCallbackInfo<v8::Value>& args);
+  static void Ready(const v8::FunctionCallbackInfo<v8::Value>& args);
+
+  static v8::Local<v8::FunctionTemplate> GetConstructorTemplate(
+      Environment* env);
+
+  static void CreatePerIsolateProperties(IsolateData* isolate_data,
+                                         v8::Local<v8::ObjectTemplate> target);
+  static void CreatePerContextProperties(v8::Local<v8::Object> target,
+                                         v8::Local<v8::Value> unused,
+                                         v8::Local<v8::Context> context,
+                                         void* priv);
+
+  static BaseObjectPtr<DataQueueFeeder> Create(Environment* env);
+
+  static bool HasInstance(Environment* env, v8::Local<v8::Value> object);
+
+ private:
+  const size_t max_buffer_size_ = 16384;
+  size_t buffer_size_;
+  std::shared_ptr<DataQueue> dataQueue_;
+  v8::Global<v8::Promise::Resolver> readFinish_;
+  size_t bytecount;
+  bool done = false;
+};
+
 }  // namespace node
 
 #endif  // defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS
diff --git a/src/env_properties.h b/src/env_properties.h
index 903158ebbdc2b7..ecdf9db4466761 100644
--- a/src/env_properties.h
+++ b/src/env_properties.h
@@ -395,6 +395,7 @@
   V(contextify_wrapper_template, v8::ObjectTemplate)                           \
   V(cpu_usage_template, v8::DictionaryTemplate)                                \
   V(crypto_key_object_handle_constructor, v8::FunctionTemplate)                \
+  V(data_queue_feeder_constructor_template, v8::FunctionTemplate)              \
   V(env_proxy_template, v8::ObjectTemplate)                                    \
   V(env_proxy_ctor_template, v8::FunctionTemplate)                             \
   V(ephemeral_key_template, v8::DictionaryTemplate)                            \
diff --git a/src/node_binding.cc b/src/node_binding.cc
index 5bd07e5253ae64..8787b4f9d494d3 100644
--- a/src/node_binding.cc
+++ b/src/node_binding.cc
@@ -47,6 +47,7 @@
   V(constants)                                                                 \
   V(contextify)                                                                \
   V(credentials)                                                               \
+  V(dataqueuefeeder)                                                           \
   V(encoding_binding)                                                          \
   V(errors)                                                                    \
   V(fs)                                                                        \
diff --git a/src/node_blob.cc b/src/node_blob.cc
index ab862bf93a411e..f28c8c484de05d 100644
--- a/src/node_blob.cc
+++ b/src/node_blob.cc
@@ -110,8 +110,8 @@ void BlobFromFilePath(const FunctionCallbackInfo<Value>& args) {
   std::vector<std::unique_ptr<DataQueue::Entry>> entries;
   entries.push_back(std::move(entry));
 
-  if (auto blob =
-          Blob::Create(env, DataQueue::CreateIdempotent(std::move(entries)))) {
+  if (auto blob = Blob::Create(
+          env, DataQueue::CreateIdempotent(env, std::move(entries)))) {
     Local<Value> vals[2]{
         blob->object(),
         Uint32::NewFromUnsigned(env->isolate(), blob->length()),
@@ -247,7 +247,7 @@ void Blob::New(const FunctionCallbackInfo<Value>& args) {
     }
   }
 
-  auto blob = Create(env, DataQueue::CreateIdempotent(std::move(entries)));
+  auto blob = Create(env, DataQueue::CreateIdempotent(env, std::move(entries)));
   if (blob)
     args.GetReturnValue().Set(blob->object());
 }
@@ -347,15 +347,24 @@ void Blob::Reader::Pull(const FunctionCallbackInfo<Value>& args) {
   }
 
   struct Impl {
-    BaseObjectPtr<Blob::Reader> reader;
+    BaseObjectWeakPtr<Blob::Reader> reader;
     Global<Function> callback;
     Environment* env;
+    std::vector<DataQueue::Vec> vecs;
+    std::list<bob::Done> dones;
+    size_t byte_count = 0;
+    bool innext = false;
   };
   // TODO(@jasnell): A unique_ptr is likely better here but making this a unique
   // pointer that is passed into the lambda causes the std::move(next) below to
   // complain about std::function needing to be copy-constructible.
-  Impl* impl = new Impl();
-  impl->reader = BaseObjectPtr<Blob::Reader>(reader);
+  // EDIT(martenrichter) We use a shared_ptr instead, as the previous
+  // implementation, with am ommrt unique_ptr did not allow to call next twice
+  // as impl is gone after the first call.
+  // Also the reference to reader should be a weak pointer, as if it is
+  // collected we should not be interested in the result of a call to next
+  std::shared_ptr<Impl> impl = std::make_shared<Impl>();
+  impl->reader = BaseObjectWeakPtr<Blob::Reader>(reader);
   impl->callback.Reset(env->isolate(), fn);
   impl->env = env;
 
@@ -363,34 +372,75 @@ void Blob::Reader::Pull(const FunctionCallbackInfo<Value>& args) {
                      const DataQueue::Vec* vecs,
                      size_t count,
                      bob::Done doneCb) mutable {
-    auto dropMe = std::unique_ptr<Impl>(impl);
+    assert(impl->reader);
     Environment* env = impl->env;
     HandleScope handleScope(env->isolate());
     Local<Function> fn = impl->callback.Get(env->isolate());
 
-    if (status == bob::STATUS_EOS) impl->reader->eos_ = true;
+    if (status == bob::STATUS_EOS) {
+      impl->reader->eos_ = true;
+    }
 
     if (count > 0) {
+      impl->vecs.insert(impl->vecs.end(), vecs, vecs + count);
+      impl->dones.push_back(std::move(doneCb));
+      for (size_t n = 0; n < count; n++) impl->byte_count += vecs[n].len;
+    }
+    if (impl->innext) {
+      CHECK(status != bob::STATUS_WAIT);
+      return;
+    }
+    if (status == bob::STATUS_CONTINUE && impl->byte_count < 16384) {
+      // We pull some more,
+      // but it must be sync as we
+      // merge it together
+      impl->innext = true;
+      while (status == bob::STATUS_CONTINUE && impl->byte_count < 16384) {
+        auto snext = [impl](int status,
+                            const DataQueue::Vec* vecs,
+                            size_t count,
+                            bob::Done doneCb) {
+          if (count > 0) {
+            impl->vecs.insert(impl->vecs.end(), vecs, vecs + count);
+            impl->dones.push_back(std::move(doneCb));
+            for (size_t n = 0; n < count; n++) impl->byte_count += vecs[n].len;
+          }
+        };
+        assert(impl->reader);
+        status = impl->reader->inner_->Pull(
+            std::move(snext), node::bob::OPTIONS_SYNC, nullptr, 0);
+      }
+    }
+    // otherwise we commit and call
+    if (impl->vecs.size() > 0) {
       // Copy the returns vectors into a single ArrayBuffer.
       size_t total = 0;
-      for (size_t n = 0; n < count; n++) total += vecs[n].len;
+      const size_t vecs_size = impl->vecs.size();
+      const DataQueue::Vec* ivecs = &(*impl->vecs.begin());
+      for (size_t n = 0; n < vecs_size; n++) total += ivecs[n].len;
 
       std::shared_ptr<BackingStore> store = ArrayBuffer::NewBackingStore(
           env->isolate(),
           total,
           BackingStoreInitializationMode::kUninitialized);
       auto ptr = static_cast<uint8_t*>(store->Data());
-      for (size_t n = 0; n < count; n++) {
-        std::copy(vecs[n].base, vecs[n].base + vecs[n].len, ptr);
-        ptr += vecs[n].len;
+      for (size_t n = 0; n < vecs_size; n++) {
+        std::copy(ivecs[n].base, ivecs[n].base + ivecs[n].len, ptr);
+        ptr += ivecs[n].len;
       }
       // Since we copied the data buffers, signal that we're done with them.
-      std::move(doneCb)(0);
-      Local<Value> argv[2] = {Uint32::New(env->isolate(), status),
+      std::for_each(impl->dones.begin(),
+                    impl->dones.end(),
+                    [](bob::Done& done) { std::move(done)(0); });
+      impl->dones.clear();
+      assert(impl->reader);
+      Local<Value> argv[2] = {Uint32::New(env->isolate(), bob::STATUS_CONTINUE),
                               ArrayBuffer::New(env->isolate(), store)};
       impl->reader->MakeCallback(fn, arraysize(argv), argv);
       return;
     }
+    impl->dones.clear();  // should not be necessary?
+    assert(impl->reader);
 
     Local<Value> argv[2] = {
         Int32::New(env->isolate(), status),
diff --git a/src/quic/application.cc b/src/quic/application.cc
index 00f8cf70d00cdb..3d9fa45a9cb761 100644
--- a/src/quic/application.cc
+++ b/src/quic/application.cc
@@ -428,8 +428,15 @@ void Session::Application::SendPendingData() {
     packet->Truncate(datalen);
     session_->Send(packet, path);
 
+    if (nwrite > 0 && stream_data.id >= 0) {
+      // We can not be sure, that the stream did not receive more
+      // data since the last call, so we need to check again
+      ResumeStream(stream_data.id);
+    }
+
     // If we have sent the maximum number of packets, we're done.
     if (++packet_send_count == max_packet_count) {
+      // question, who ensures, that SendPendingData is called again
       return;
     }
 
@@ -512,7 +519,9 @@ class DefaultApplication final : public Session::Application {
     Debug(&session(), "Default application getting stream data");
     DCHECK_NOT_NULL(stream_data);
     // If the queue is empty, there aren't any streams with data yet
-    if (stream_queue_.IsEmpty()) return 0;
+    if (stream_queue_.IsEmpty()) {
+      return 0;
+    }
 
     const auto get_length = [](auto vec, size_t count) {
       CHECK_NOT_NULL(vec);
diff --git a/src/quic/quic.cc b/src/quic/quic.cc
index edfb5dc9e66295..f5303d51c8906e 100644
--- a/src/quic/quic.cc
+++ b/src/quic/quic.cc
@@ -12,7 +12,7 @@
 #include "endpoint.h"
 #include "node_external_reference.h"
 
-namespace node {
+namespace node::quic {
 
 using v8::Context;
 using v8::Local;
@@ -20,8 +20,6 @@ using v8::Object;
 using v8::ObjectTemplate;
 using v8::Value;
 
-namespace quic {
-
 void CreatePerIsolateProperties(IsolateData* isolate_data,
                                 Local<ObjectTemplate> target) {
   Endpoint::InitPerIsolate(isolate_data, target);
@@ -47,8 +45,7 @@ void RegisterExternalReferences(ExternalReferenceRegistry* registry) {
   Stream::RegisterExternalReferences(registry);
 }
 
-}  // namespace quic
-}  // namespace node
+}  // namespace node::quic
 
 NODE_BINDING_CONTEXT_AWARE_INTERNAL(quic,
                                     node::quic::CreatePerContextProperties)
diff --git a/src/quic/session.cc b/src/quic/session.cc
index 882fd9bac0e54f..d044dca437e6f4 100644
--- a/src/quic/session.cc
+++ b/src/quic/session.cc
@@ -63,6 +63,7 @@ namespace quic {
   V(DATAGRAM, datagram, uint8_t)                                               \
   V(SESSION_TICKET, session_ticket, uint8_t)                                   \
   V(CLOSING, closing, uint8_t)                                                 \
+  V(FINISH_CLOSING, finish_closing, uint8_t)                                   \
   V(GRACEFUL_CLOSE, graceful_close, uint8_t)                                   \
   V(SILENT_CLOSE, silent_close, uint8_t)                                       \
   V(STATELESS_RESET, stateless_reset, uint8_t)                                 \
@@ -542,7 +543,13 @@ struct Session::Impl final : public MemoryRetainer {
         local_address_(config.local_address),
         remote_address_(config.remote_address),
         application_(SelectApplication(session, config_)),
-        timer_(session_->env(), [this] { session_->OnTimeout(); }) {
+        timer_(session_->env(), [this] {
+          BaseObjectPtr<Session> session = BaseObjectPtr<Session>(session_);
+          // we hold a reference to session,
+          // as the reference from session to us may go away otherwise
+          // while we call OnTimeout
+          session_->OnTimeout();
+        }) {
     timer_.Unref();
   }
   DISALLOW_COPY_AND_MOVE(Impl)
@@ -558,13 +565,18 @@ struct Session::Impl final : public MemoryRetainer {
     state_->closing = 1;
     STAT_RECORD_TIMESTAMP(Stats, closing_at);
 
+    // we hold a reference to our session,
+    // as the reference from session to us may go away otherwise
+    // while we destroy streams
+    BaseObjectPtr<Session> session = BaseObjectPtr<Session>(session_);
+
     // Iterate through all of the known streams and close them. The streams
     // will remove themselves from the Session as soon as they are closed.
     // Note: we create a copy because the streams will remove themselves
     // while they are cleaning up which will invalidate the iterator.
     StreamsMap streams = streams_;
     for (auto& stream : streams) stream.second->Destroy(last_error_);
-    DCHECK(streams.empty());
+    DCHECK(streams_.empty());  // do not check our local copy
 
     // Clear the pending streams.
     while (!pending_bidi_stream_queue_.IsEmpty()) {
@@ -1393,15 +1405,11 @@ bool Session::is_destroyed_or_closing() const {
 
 void Session::Close(CloseMethod method) {
   if (is_destroyed()) return;
-  auto& stats_ = impl_->stats_;
 
   if (impl_->last_error_) {
     Debug(this, "Closing with error: %s", impl_->last_error_);
   }
 
-  STAT_RECORD_TIMESTAMP(Stats, closing_at);
-  impl_->state_->closing = 1;
-
   // With both the DEFAULT and SILENT options, we will proceed to closing
   // the session immediately. All open streams will be immediately destroyed
   // with whatever is set as the last error. The session will then be destroyed
@@ -1448,10 +1456,13 @@ void Session::Close(CloseMethod method) {
 }
 
 void Session::FinishClose() {
-  // FinishClose() should be called only after, and as a result of, Close()
+  if (impl_->state_->finish_closing) return;
+  // we were already called, avoids calling it twice
+  impl_->state_->finish_closing = 1;
+  // FinishClose() should be called only after,
+  // and as a result of, Close()
   // being called first.
   DCHECK(!is_destroyed());
-  DCHECK(impl_->state_->closing);
 
   // If impl_->Close() returns true, then the session can be destroyed
   // immediately without round-tripping through JavaScript.
@@ -1465,12 +1476,12 @@ void Session::FinishClose() {
 }
 
 void Session::Destroy() {
-  // Destroy() should be called only after, and as a result of, Close()
+  // Destroy() should be called only after,
+  // and as a result of, Close()
   // being called first.
   DCHECK(impl_);
   DCHECK(impl_->state_->closing);
   Debug(this, "Session destroyed");
-  impl_.reset();
   if (qlog_stream_ || keylog_stream_) {
     env()->SetImmediate(
         [qlog = qlog_stream_, keylog = keylog_stream_](Environment*) {
@@ -1480,6 +1491,9 @@ void Session::Destroy() {
   }
   qlog_stream_.reset();
   keylog_stream_.reset();
+  impl_.reset();  // This can cause the session (so us) object to be garbage
+                  // collected, so the session object may not be valid after
+                  // this call.
 }
 
 PendingStream::PendingStreamQueue& Session::pending_bidi_stream_queue() const {
@@ -1942,6 +1956,8 @@ BaseObjectPtr<Stream> Session::CreateStream(
   if (auto stream = Stream::Create(this, id, std::move(data_source)))
       [[likely]] {
     AddStream(stream, option);
+    ResumeStream(id);  // ok, we need to resume, as the Resume before fails
+    // as the stream was not added yet
     return stream;
   }
   return {};
@@ -2091,7 +2107,8 @@ void Session::RemoveStream(stream_id id) {
   // returns.
   if (impl_->state_->closing && impl_->state_->graceful_close) {
     FinishClose();
-    CHECK(is_destroyed());
+    // CHECK(is_destroyed());
+    // this will not work, our this pointer may not be valid anymore!
   }
 }
 
@@ -2233,6 +2250,10 @@ void Session::ExtendOffset(size_t amount) {
 
 void Session::UpdateDataStats() {
   Debug(this, "Updating data stats");
+  if (!impl_) {
+    Debug(this, "Updating data stats failed, impl gone!");
+    return;
+  }
   auto& stats_ = impl_->stats_;
   ngtcp2_conn_info info;
   ngtcp2_conn_get_conn_info(*this, &info);
diff --git a/src/quic/streams.cc b/src/quic/streams.cc
index 8fe5b72ce1fe5b..93891751365e4a 100644
--- a/src/quic/streams.cc
+++ b/src/quic/streams.cc
@@ -167,7 +167,7 @@ Maybe<std::shared_ptr<DataQueue>> Stream::GetDataQueueFromSource(
     }
     entries.push_back(DataQueue::CreateInMemoryEntryFromBackingStore(
         std::move(backing), offset, length));
-    return Just(DataQueue::CreateIdempotent(std::move(entries)));
+    return Just(DataQueue::CreateIdempotent(env, std::move(entries)));
   } else if (value->IsSharedArrayBuffer()) {
     // We aren't going to allow use of SharedArrayBuffer as a data source.
     // The reason is that SharedArrayBuffer memory is possibly shared with
@@ -201,7 +201,7 @@ Maybe<std::shared_ptr<DataQueue>> Stream::GetDataQueueFromSource(
     auto length = view->ByteLength();
     entries.push_back(DataQueue::CreateInMemoryEntryFromBackingStore(
         std::move(backing), offset, length));
-    return Just(DataQueue::CreateIdempotent(std::move(entries)));
+    return Just(DataQueue::CreateIdempotent(env, std::move(entries)));
   } else if (Blob::HasInstance(env, value)) {
     Blob* blob;
     ASSIGN_OR_RETURN_UNWRAP(
@@ -214,9 +214,17 @@ Maybe<std::shared_ptr<DataQueue>> Stream::GetDataQueueFromSource(
     memcpy(backing->Data(), *str, str.length());
     entries.push_back(DataQueue::CreateInMemoryEntryFromBackingStore(
         std::move(backing), 0, backing->ByteLength()));
-    return Just(DataQueue::CreateIdempotent(std::move(entries)));
+    return Just(DataQueue::CreateIdempotent(env, std::move(entries)));
+  } else if (DataQueueFeeder::HasInstance(env, value)) {
+    // a DataQueueFeeder
+    DataQueueFeeder* dataQueueFeeder;
+    ASSIGN_OR_RETURN_UNWRAP(
+        &dataQueueFeeder, value, Nothing<std::shared_ptr<DataQueue>>());
+    std::shared_ptr<DataQueue> dataQueue = DataQueue::Create(env);
+    dataQueueFeeder->setDataQueue(dataQueue);
+    return Just(dataQueue);
   }
-  // TODO(jasnell): Add streaming sources...
+
   THROW_ERR_INVALID_ARG_TYPE(env, "Invalid data source type");
   return Nothing<std::shared_ptr<DataQueue>>();
 }
@@ -367,9 +375,13 @@ struct Stream::Impl {
 
   // Returns a Blob::Reader that can be used to read data that has been
   // received on the stream.
+  // returns undefined if local unidirectional stream
   JS_METHOD(GetReader) {
     Stream* stream;
     ASSIGN_OR_RETURN_UNWRAP(&stream, args.This());
+    if (stream->is_local_unidirectional()) {
+      return args.GetReturnValue().SetUndefined();
+    }
     BaseObjectPtr<Blob::Reader> reader = stream->get_reader();
     if (reader) return args.GetReturnValue().Set(reader->object());
     THROW_ERR_INVALID_STATE(Environment::GetCurrent(args),
@@ -386,6 +398,13 @@ class Stream::Outbound final : public MemoryRetainer {
         queue_(std::move(queue)),
         reader_(queue_->get_reader()) {}
 
+  ~Outbound() {
+    // we need to clear all pending next's from the queue, as it holds pointers
+    // to Stream and Session, which will be invalidated and may cause use after
+    // free otherwise
+    if (queue_) queue_->clearPendingNext();
+  }
+
   void Acknowledge(size_t amount) {
     size_t remaining = std::min(amount, total_ - uncommitted_);
     while (remaining > 0 && head_ != nullptr) {
@@ -500,19 +519,28 @@ class Stream::Outbound final : public MemoryRetainer {
     // that the pull is sync but allow for it to be async.
     int ret = reader_->Pull(
         [this](auto status, auto vecs, auto count, auto done) {
-          // Always make sure next_pending_ is false when we're done.
-          auto on_exit = OnScopeLeave([this] { next_pending_ = false; });
+          const bool last_next_pending_state = next_pending_;
+          next_pending_ = false;
+          // this ensures that next_pending is reset to false
+          // Note that for example ResumeStream below, may again call Pull
+          // so we have to erase next_pending_ to not block this call.
+          // Therefore the previous OnScopeLeave lead to a race condition.
+
+          // We need to hold a reference to stream and session
+          // so that it can not go away during the next calls.
+          BaseObjectPtr<Stream> stream = BaseObjectPtr<Stream>(stream_);
+          BaseObjectPtr<Session> session =
+              BaseObjectPtr<Session>(&stream_->session());
 
           // The status should never be wait here.
           DCHECK_NE(status, bob::Status::STATUS_WAIT);
 
           if (status < 0) {
-            // If next_pending_ is true then a pull from the reader ended up
+            // If next_pending_ was true then a pull from the reader ended up
             // being asynchronous, our stream is blocking waiting for the data,
             // but we have an error! oh no! We need to error the stream.
-            if (next_pending_) {
-              stream_->Destroy(
-                  QuicError::ForNgtcp2Error(NGTCP2_INTERNAL_ERROR));
+            if (last_next_pending_state) {
+              stream->Destroy(QuicError::ForNgtcp2Error(NGTCP2_INTERNAL_ERROR));
               // We do not need to worry about calling MarkErrored in this case
               // since we are immediately destroying the stream which will
               // release the outbound buffer anyway.
@@ -529,8 +557,8 @@ class Stream::Outbound final : public MemoryRetainer {
             // Here, there is no more data to read, but we will might have data
             // in the uncommitted queue. We'll resume the stream so that the
             // session will try to read from it again.
-            if (next_pending_) {
-              stream_->session().ResumeStream(stream_->id());
+            if (last_next_pending_state) {
+              session->ResumeStream(stream_->id());
             }
             return;
           }
@@ -549,12 +577,12 @@ class Stream::Outbound final : public MemoryRetainer {
           // bytes in the queue.
           Append(vecs, count, std::move(done));
 
-          // If next_pending_ is true, then a pull from the reader ended up
+          // If next_pending_ was true, then a pull from the reader ended up
           // being asynchronous, our stream is blocking waiting for the data.
           // Now that we have data, let's resume the stream so the session will
           // pull from it again.
-          if (next_pending_) {
-            stream_->session().ResumeStream(stream_->id());
+          if (last_next_pending_state) {
+            stream->session().ResumeStream(stream_->id());
           }
         },
         bob::OPTIONS_SYNC,
@@ -836,7 +864,7 @@ Stream::Stream(BaseObjectWeakPtr<Session> session,
       stats_(env()->isolate()),
       state_(env()->isolate()),
       session_(std::move(session)),
-      inbound_(DataQueue::Create()),
+      inbound_(DataQueue::Create(env())),
       headers_(env()->isolate()) {
   MakeWeak();
   DCHECK(id < kMaxStreamId);
@@ -866,7 +894,7 @@ Stream::Stream(BaseObjectWeakPtr<Session> session,
       stats_(env()->isolate()),
       state_(env()->isolate()),
       session_(std::move(session)),
-      inbound_(DataQueue::Create()),
+      inbound_(DataQueue::Create(env())),
       maybe_pending_stream_(
           std::make_unique<PendingStream>(direction, this, session_)),
       headers_(env()->isolate()) {
@@ -1038,6 +1066,8 @@ void Stream::set_outbound(std::shared_ptr<DataQueue> source) {
   if (!source || !is_writable()) return;
   Debug(this, "Setting the outbound data source");
   DCHECK_NULL(outbound_);
+  // sets us as notifier, to be alerted if new data is there
+  source->SetNotifier(this);
   outbound_ = std::make_unique<Outbound>(this, std::move(source));
   state_->has_outbound = 1;
   if (!is_pending()) session_->ResumeStream(id());
@@ -1112,10 +1142,7 @@ void Stream::Acknowledge(size_t datalen) {
 
   Debug(this, "Acknowledging %zu bytes", datalen);
 
-  // ngtcp2 guarantees that offset must always be greater than the previously
-  // received offset.
-  DCHECK_GE(datalen, STAT_GET(Stats, max_offset_ack));
-  STAT_SET(Stats, max_offset_ack, datalen);
+  STAT_SET(Stats, max_offset_ack, STAT_GET(Stats, max_offset_ack) + datalen);
 
   // Consumes the given number of bytes in the buffer.
   outbound_->Acknowledge(datalen);
@@ -1216,7 +1243,9 @@ void Stream::ReceiveData(const uint8_t* data,
   inbound_->append(DataQueue::CreateInMemoryEntryFromBackingStore(
       std::move(backing), 0, len));
 
-  if (flags.fin) EndReadable();
+  if (flags.fin) {
+    EndReadable();
+  }
 }
 
 void Stream::ReceiveStopSending(QuicError error) {
@@ -1307,6 +1336,10 @@ void Stream::EmitWantTrailers() {
 
 // ============================================================================
 
+void Stream::newDataOrCloseAdded() {
+  session_->ResumeStream(id());
+}
+
 void Stream::Schedule(Queue* queue) {
   // If this stream is not already in the queue to send data, add it.
   Debug(this, "Scheduled");
@@ -1319,6 +1352,7 @@ void Stream::Unschedule() {
 }
 
 }  // namespace quic
+
 }  // namespace node
 
 #endif  // OPENSSL_NO_QUIC
diff --git a/src/quic/streams.h b/src/quic/streams.h
index c230815d78e4be..f618dec48bcf76 100644
--- a/src/quic/streams.h
+++ b/src/quic/streams.h
@@ -150,7 +150,8 @@ class PendingStream final {
 // zero-length output.
 class Stream final : public AsyncWrap,
                      public Ngtcp2Source,
-                     public DataQueue::BackpressureListener {
+                     public DataQueue::BackpressureListener,
+                     public DataQueue::Notifier {
  public:
   using Header = NgHeaderBase<BindingData>;
 
@@ -250,6 +251,9 @@ class Stream final : public AsyncWrap,
              size_t count,
              size_t max_count_hint) override;
 
+  // Mechanism, so that the DataQueueFeeder can wake the stream
+  void newDataOrCloseAdded() override;
+
   // Forcefully close the stream immediately. Data already queued in the
   // inbound is preserved but new data will not be accepted. All pending
   // writes are abandoned, and the stream is immediately closed at the ngtcp2
diff --git a/test/cctest/test_dataqueue.cc b/test/cctest/test_dataqueue.cc
index 73488fcab0a4d1..1babbf82f08f92 100644
--- a/test/cctest/test_dataqueue.cc
+++ b/test/cctest/test_dataqueue.cc
@@ -101,7 +101,7 @@ TEST(DataQueue, IdempotentDataQueue) {
 
   // We can create an idempotent DataQueue from a list of entries.
   std::shared_ptr<DataQueue> data_queue =
-      DataQueue::CreateIdempotent(std::move(list));
+      DataQueue::CreateIdempotent(nullptr, std::move(list));
 
   CHECK_NOT_NULL(data_queue);
 
@@ -426,7 +426,7 @@ TEST(DataQueue, NonIdempotentDataQueue) {
       &buffer3, len3, [](void*, size_t, void*) {}, nullptr);
 
   // We can create an non-idempotent DataQueue from a list of entries.
-  std::shared_ptr<DataQueue> data_queue = DataQueue::Create();
+  std::shared_ptr<DataQueue> data_queue = DataQueue::Create(nullptr);
 
   CHECK(!data_queue->is_idempotent());
   CHECK_EQ(data_queue->size().value(), 0);
@@ -579,7 +579,7 @@ TEST(DataQueue, DataQueueEntry) {
 
   // We can create an idempotent DataQueue from a list of entries.
   std::shared_ptr<DataQueue> data_queue =
-      DataQueue::CreateIdempotent(std::move(list));
+      DataQueue::CreateIdempotent(nullptr, std::move(list));
 
   CHECK_NOT_NULL(data_queue);
 
@@ -603,12 +603,17 @@ TEST(DataQueue, DataQueueEntry) {
   // We can add it to another data queue, even if the new one is not
   // idempotent.
 
-  std::shared_ptr<DataQueue> data_queue2 = DataQueue::Create();
+  std::shared_ptr<DataQueue> data_queue2 = DataQueue::Create(nullptr);
   CHECK(data_queue2->append(std::move(slice)).value());
 
   // Our original data queue should have a use count of 2.
   CHECK_EQ(data_queue.use_count(), 2);
 
+  CHECK_EQ(data_queue2->is_capped(), false);
+  data_queue2->cap();  // The data queue needs to be capped, as data
+  // may be added later otherwise
+  CHECK_EQ(data_queue2->is_capped(), true);
+
   std::shared_ptr<DataQueue::Reader> reader = data_queue2->get_reader();
 
   bool pullIsPending = true;
diff --git a/test/common/quic/test-helpers.mjs b/test/common/quic/test-helpers.mjs
new file mode 100644
index 00000000000000..54d7c4f5e9ecf2
--- /dev/null
+++ b/test/common/quic/test-helpers.mjs
@@ -0,0 +1,67 @@
+import assert from 'node:assert';
+// start demo data
+// taken from @fails-components/webtransport tests
+// by the original author
+export function createBytesChunk(length) {
+  assert.strictEqual(length % 4, 0);
+  const workArray = new Array(length / 2);
+  for (let i = 0; i < length / 4; i++) {
+    workArray[2 * i + 1] = length % 0xffff;
+    workArray[2 * i] = i;
+  }
+  const helper = new Uint16Array(workArray);
+  const toreturn = new Uint8Array(
+    helper.buffer,
+    helper.byteOffset,
+    helper.byteLength,
+  );
+  return toreturn;
+}
+
+// The number in the comments, help you identify the chunk, as it is the length first two bytes
+// this is helpful, when debugging buffer passing
+export const KNOWN_BYTES_LONG = [
+  createBytesChunk(60000), // 96, 234
+  createBytesChunk(12), // 0, 12
+  createBytesChunk(50000), // 195, 80
+  createBytesChunk(1600).buffer, // 6, 64 we use buffer here to increae test coverage
+  createBytesChunk(20000), // 78, 32
+  createBytesChunk(30000), // 117, 48
+];
+
+// end demo data
+
+export function uint8concat(arrays) {
+  const length = arrays.reduce((acc, curr) => acc + curr.byteLength, 0);
+  const result = new Uint8Array(length);
+  let pos = 0;
+  let array = 0;
+  while (pos < length) {
+    const curArr = arrays[array];
+    const curLen = curArr.byteLength;
+    const dest = new Uint8Array(result.buffer, result.byteOffset + pos, curLen);
+    dest.set(new Uint8Array(curArr));
+    array++;
+    pos += curArr.byteLength;
+  }
+  return result;
+}
+
+export function equalUint8Arrays(arr1, arr2) {
+  assert.ok(arr1.byteLength === arr2.byteLength,
+            'Array content differs in length ' +
+            arr1.byteLength +
+            ' vs. ' +
+            arr2.byteLength +
+            ' diff: ' +
+            (arr1.byteLength - arr2.byteLength));
+  for (let i = 0; i < arr1.byteLength; i++) {
+    assert.ok(arr1[i] === arr2[i],
+              'Array content differs at ' +
+              i +
+              ' arr1: ' +
+              arr1[i] +
+              ' arr2: ' +
+              arr2[i]);
+  }
+}
diff --git a/test/parallel/test-blob.js b/test/parallel/test-blob.js
index 8ce1c53012ddcc..6f5cf591827d93 100644
--- a/test/parallel/test-blob.js
+++ b/test/parallel/test-blob.js
@@ -240,20 +240,20 @@ assert.throws(() => new Blob({}), {
 })().then(common.mustCall());
 
 (async () => {
-  const b = new Blob(Array(10).fill('hello'));
+  const b = new Blob(Array(10).fill('hello'.repeat(3000)));
   const reader = b.stream().getReader();
   const chunks = [];
   while (true) {
     const res = await reader.read();
     if (res.done) break;
-    assert.strictEqual(res.value.byteLength, 5);
+    assert.strictEqual(res.value.byteLength, 30000);
     chunks.push(res.value);
   }
-  assert.strictEqual(chunks.length, 10);
+  assert.strictEqual(chunks.length, 5);
 })().then(common.mustCall());
 
 (async () => {
-  const b = new Blob(Array(10).fill('hello'));
+  const b = new Blob(Array(10).fill('hello'.repeat(3000)));
   const reader = b.stream().getReader();
   const chunks = [];
   while (true) {
@@ -263,7 +263,7 @@ assert.throws(() => new Blob({}), {
       break;
     }
     if (res.done) break;
-    assert.strictEqual(res.value.byteLength, 5);
+    assert.strictEqual(res.value.byteLength, 30000);
     chunks.push(res.value);
   }
   assert.strictEqual(chunks.length, 5);
@@ -329,17 +329,19 @@ assert.throws(() => new Blob({}), {
 })().then(common.mustCall());
 
 (async () => {
-  const b = new Blob(Array(10).fill('hello'));
+  const b = new Blob(Array(10).fill('hello'.repeat(3000)));
   const stream = b.stream();
   const reader = stream.getReader();
   assert.strictEqual(stream[kState].controller.desiredSize, 0);
   const { value, done } = await reader.read();
-  assert.strictEqual(value.byteLength, 5);
+  assert.strictEqual(value.byteLength, 30000);
   assert(!done);
   setTimeout(common.mustCall(() => {
     // The blob stream is now a byte stream hence after the first read,
     // it should pull in the next 'hello' which is 5 bytes hence -5.
-    assert.strictEqual(stream[kState].controller.desiredSize, 0);
+    // but recently, we coalesce if possible adjacent memory
+    // the test code is adapted to the current limit for coalescing
+    assert.ok(stream[kState].controller.desiredSize <= 0);
   }), 0);
 })().then(common.mustCall());
 
@@ -358,20 +360,20 @@ assert.throws(() => new Blob({}), {
 })().then(common.mustCall());
 
 (async () => {
-  const b = new Blob(Array(10).fill('hello'));
+  const b = new Blob(Array(10).fill('hello'.repeat(3000)));
   const stream = b.stream();
   const reader = stream.getReader({ mode: 'byob' });
   assert.strictEqual(stream[kState].controller.desiredSize, 0);
-  const { value, done } = await reader.read(new Uint8Array(100));
-  assert.strictEqual(value.byteLength, 5);
+  const { value, done } = await reader.read(new Uint8Array(40000));
+  assert.strictEqual(value.byteLength, 30000);
   assert(!done);
   setTimeout(common.mustCall(() => {
-    assert.strictEqual(stream[kState].controller.desiredSize, 0);
+    assert.ok(stream[kState].controller.desiredSize <= 0);
   }), 0);
 })().then(common.mustCall());
 
 (async () => {
-  const b = new Blob(Array(10).fill('hello'));
+  const b = new Blob(Array(10).fill('hello'.repeat(3000)));
   const stream = b.stream();
   const reader = stream.getReader({ mode: 'byob' });
   assert.strictEqual(stream[kState].controller.desiredSize, 0);
@@ -379,7 +381,7 @@ assert.throws(() => new Blob({}), {
   assert.strictEqual(value.byteLength, 2);
   assert(!done);
   setTimeout(common.mustCall(() => {
-    assert.strictEqual(stream[kState].controller.desiredSize, -3);
+    assert.ok(stream[kState].controller.desiredSize <= -29998);
   }), 0);
 })().then(common.mustCall());
 
diff --git a/test/parallel/test-quic-client-to-server-bidirectional-echo.mjs b/test/parallel/test-quic-client-to-server-bidirectional-echo.mjs
new file mode 100644
index 00000000000000..0b6298a8898f55
--- /dev/null
+++ b/test/parallel/test-quic-client-to-server-bidirectional-echo.mjs
@@ -0,0 +1,84 @@
+// Flags: --experimental-quic --no-warnings
+
+import { hasQuic, skip, mustCall } from '../common/index.mjs';
+import assert from 'node:assert';
+import { readKey } from '../common/fixtures.mjs';
+import { KNOWN_BYTES_LONG, uint8concat, equalUint8Arrays } from '../common/quic/test-helpers.mjs';
+import { TransformStream } from 'node:stream/web';
+
+if (!hasQuic) {
+  skip('QUIC is not enabled');
+}
+
+// Import after the hasQuic check
+const { listen, connect } = await import('node:quic');
+const { createPrivateKey } = await import('node:crypto');
+
+const keys = createPrivateKey(readKey('agent1-key.pem'));
+const certs = readKey('agent1-cert.pem');
+
+const finished = Promise.withResolvers();
+
+const serverEndpoint = await listen(mustCall(async (serverSession) => {
+  serverSession.onstream = mustCall(async (stream) => {
+    assert.strictEqual(stream.direction, 'bidi');
+    stream.closed.catch(() => {
+      // ignore
+    });
+    const serverTransformStream = new TransformStream();
+    stream.setOutbound(serverTransformStream.readable);
+    await stream.readable.pipeTo(serverTransformStream.writable);
+
+  }, 1);
+
+  await finished.promise;
+  serverSession.closed.catch((err) => {
+    // ignore the error
+  });
+  serverSession.close();
+}), { keys, certs });
+
+// The server must have an address to connect to after listen resolves.
+assert.ok(serverEndpoint.address !== undefined);
+
+const clientSession = await connect(serverEndpoint.address);
+await clientSession.opened;
+
+
+const transformStream = new TransformStream();
+const sendStream = await clientSession.createBidirectionalStream({ body: transformStream.readable });
+sendStream.closed.catch(() => {
+  // ignore
+});
+assert.strictEqual(sendStream.direction, 'bidi');
+const writeToStream = async () => {
+  const clientWritable = transformStream.writable;
+  const writer = clientWritable.getWriter();
+  for (const chunk of KNOWN_BYTES_LONG) {
+    await writer.ready;
+    await writer.write(chunk);
+  }
+  await writer.ready;
+  await writer.close();
+};
+const readFromStream = mustCall(async () => {
+  const reader = sendStream.readable.getReader();
+  const readChunks = [];
+  while (true) {
+    const { done, value } = await reader.read();
+    if (value) {
+      assert.ok(value instanceof Uint8Array, 'Expects value to be a Uint8Array');
+      readChunks.push(value);
+    }
+    if (done) break;
+  }
+  // Now compare what we got
+  equalUint8Arrays(uint8concat(KNOWN_BYTES_LONG), uint8concat(readChunks));
+});
+
+await Promise.all([writeToStream(), readFromStream()]);
+clientSession.closed.catch((err) => {
+  // ignore the error
+});
+clientSession.close();
+finished.resolve();
diff --git a/test/parallel/test-quic-client-to-server-unidirectional.mjs b/test/parallel/test-quic-client-to-server-unidirectional.mjs
new file mode 100644
index 00000000000000..137b92947e9d4a
--- /dev/null
+++ b/test/parallel/test-quic-client-to-server-unidirectional.mjs
@@ -0,0 +1,77 @@
+// Flags: --experimental-quic --no-warnings
+
+import { hasQuic, skip, mustCall } from '../common/index.mjs';
+import assert from 'node:assert';
+import { readKey } from '../common/fixtures.mjs';
+import { KNOWN_BYTES_LONG, uint8concat, equalUint8Arrays } from '../common/quic/test-helpers.mjs';
+import { TransformStream } from 'node:stream/web';
+
+if (!hasQuic) {
+  skip('QUIC is not enabled');
+}
+
+// Import after the hasQuic check
+const { listen, connect } = await import('node:quic');
+const { createPrivateKey } = await import('node:crypto');
+
+const keys = createPrivateKey(readKey('agent1-key.pem'));
+const certs = readKey('agent1-cert.pem');
+
+// The opened promise should resolve when the client finished reading
+const serverFinished = Promise.withResolvers();
+
+const serverEndpoint = await listen(mustCall(async (serverSession) => {
+  serverSession.onstream = mustCall(async (stream) => {
+    assert.strictEqual(stream.direction, 'uni');
+    const reader = stream.readable.getReader();
+    const readChunks = [];
+    while (true) {
+      const { done, value } = await reader.read();
+      // if (readc > 20) throw new Error("after read " + readc);
+      if (value) {
+        assert.ok(value instanceof Uint8Array, 'Expects value to be a Uint8Array');
+        readChunks.push(value);
+      }
+      if (done) break;
+    }
+    stream.closed.catch(() => {
+      // ignore
+    });
+    // Now compare what we got
+    equalUint8Arrays(uint8concat(KNOWN_BYTES_LONG), uint8concat(readChunks));
+    serverFinished.resolve();
+  }, 1);
+
+  await serverFinished.promise;
+  serverSession.closed.catch((err) => {
+    // ignore the error
+  });
+  serverSession.close();
+}), { keys, certs });
+
+// The server must have an address to connect to after listen resolves.
+assert.ok(serverEndpoint.address !== undefined);
+
+const clientSession = await connect(serverEndpoint.address);
+await clientSession.opened;
+
+
+const transformStream = new TransformStream();
+const sendStream = await clientSession.createUnidirectionalStream({ body: transformStream.readable });
+sendStream.closed.catch(() => {
+  // ignore
+});
+assert.strictEqual(sendStream.direction, 'uni');
+const clientWritable = transformStream.writable;
+const writer = clientWritable.getWriter();
+for (const chunk of KNOWN_BYTES_LONG) {
+  await writer.ready;
+  await writer.write(chunk);
+}
+await writer.ready;
+await writer.close();
+clientSession.closed.catch((err) => {
+  // ignore the error
+});
+clientSession.close();
+await serverFinished.promise;
diff --git a/test/parallel/test-quic-server-to-client-bidirectional-echo.mjs b/test/parallel/test-quic-server-to-client-bidirectional-echo.mjs
new file mode 100644
index 00000000000000..ec896f205e1cc5
--- /dev/null
+++ b/test/parallel/test-quic-server-to-client-bidirectional-echo.mjs
@@ -0,0 +1,84 @@
+// Flags: --experimental-quic --no-warnings
+
+import { hasQuic, skip, mustCall } from '../common/index.mjs';
+import assert from 'node:assert';
+import { readKey } from '../common/fixtures.mjs';
+import { KNOWN_BYTES_LONG, uint8concat, equalUint8Arrays } from '../common/quic/test-helpers.mjs';
+import { TransformStream } from 'node:stream/web';
+
+if (!hasQuic) {
+  skip('QUIC is not enabled');
+}
+
+// Import after the hasQuic check
+const { listen, connect } = await import('node:quic');
+const { createPrivateKey } = await import('node:crypto');
+
+const keys = createPrivateKey(readKey('agent1-key.pem'));
+const certs = readKey('agent1-cert.pem');
+
+const finished = Promise.withResolvers();
+
+const serverEndpoint = await listen(mustCall(async (serverSession) => {
+  await serverSession.opened;
+  const transformStream = new TransformStream();
+  serverSession.closed.catch((err) => {
+    // ignore the error
+  });
+  const sendStream = await serverSession.createBidirectionalStream({ body: transformStream.readable });
+  // Now compare what we got
+
+  assert.strictEqual(sendStream.direction, 'bidi');
+  sendStream.closed.catch(() => {
+    // ignore
+  });
+  const writeToStream = async () => {
+    const clientWritable = transformStream.writable;
+    const writer = clientWritable.getWriter();
+    for (const chunk of KNOWN_BYTES_LONG) {
+      await writer.ready;
+      await writer.write(chunk);
+    }
+    await writer.ready;
+    await writer.close();
+  };
+  const readFromStream = mustCall(async () => {
+    const reader = sendStream.readable.getReader();
+    const readChunks = [];
+    while (true) {
+      const { done, value } = await reader.read();
+      if (value) {
+        assert.ok(value instanceof Uint8Array, 'Expects value to be a Uint8Array');
+        readChunks.push(value);
+      }
+      if (done) break;
+    }
+    equalUint8Arrays(uint8concat(KNOWN_BYTES_LONG), uint8concat(readChunks));
+  });
+  await Promise.all([writeToStream(), readFromStream()]);
+  serverSession.close();
+  finished.resolve();
+}), { keys, certs });
+
+// The server must have an address to connect to after listen resolves.
+assert.ok(serverEndpoint.address !== undefined);
+
+const clientSession = await connect(serverEndpoint.address);
+await clientSession.opened;
+
+
+clientSession.onstream = mustCall(async (stream) => {
+  assert.strictEqual(stream.direction, 'bidi');
+  stream.closed.catch(() => {
+    // ignore
+  });
+  const clientTransformStream = new TransformStream();
+  stream.setOutbound(clientTransformStream.readable);
+  await stream.readable.pipeTo(clientTransformStream.writable);
+}, 1);
+
+await finished.promise;
+clientSession.closed.catch((err) => {
+  // ignore the error
+});
+clientSession.close();
diff --git a/test/parallel/test-quic-server-to-client-unidirectional.mjs b/test/parallel/test-quic-server-to-client-unidirectional.mjs
new file mode 100644
index 00000000000000..fd04e907253b7d
--- /dev/null
+++ b/test/parallel/test-quic-server-to-client-unidirectional.mjs
@@ -0,0 +1,75 @@
+// Flags: --experimental-quic --no-warnings
+
+import { hasQuic, skip, mustCall } from '../common/index.mjs';
+import assert from 'node:assert';
+import { readKey } from '../common/fixtures.mjs';
+import { KNOWN_BYTES_LONG, uint8concat, equalUint8Arrays } from '../common/quic/test-helpers.mjs';
+import { TransformStream } from 'node:stream/web';
+
+if (!hasQuic) {
+  skip('QUIC is not enabled');
+}
+
+// Import after the hasQuic check
+const { listen, connect } = await import('node:quic');
+const { createPrivateKey } = await import('node:crypto');
+
+const keys = createPrivateKey(readKey('agent1-key.pem'));
+const certs = readKey('agent1-cert.pem');
+
+// The opened promise should resolve when the client finished reading
+const clientFinished = Promise.withResolvers();
+
+
+const serverEndpoint = await listen(mustCall(async (serverSession) => {
+  await serverSession.opened;
+  const transformStream = new TransformStream();
+  const sendStream = await serverSession.createUnidirectionalStream({ body: transformStream.readable });
+  sendStream.closed.catch(() => {
+    // ignore
+  });
+  assert.strictEqual(sendStream.direction, 'uni');
+  const serverWritable = transformStream.writable;
+  const writer = serverWritable.getWriter();
+  for (const chunk of KNOWN_BYTES_LONG) {
+    await writer.ready;
+    await writer.write(chunk);
+  }
+  await writer.ready;
+  await writer.close();
+  serverSession.closed.catch((err) => {
+    // ignore the error
+  });
+  serverSession.close();
+}), { keys, certs });
+
+// The server must have an address to connect to after listen resolves.
+assert.ok(serverEndpoint.address !== undefined);
+
+const clientSession = await connect(serverEndpoint.address);
+
+clientSession.onstream = mustCall(async (stream) => {
+  assert.strictEqual(stream.direction, 'uni');
+  const reader = stream.readable.getReader();
+  const readChunks = [];
+  while (true) {
+    const { done, value } = await reader.read();
+    if (value) {
+      assert.ok(value instanceof Uint8Array, 'Expects value to be a Uint8Array');
+      readChunks.push(value);
+    }
+    if (done) break;
+  }
+  stream.closed.catch(() => {
+    // ignore
+  });
+  // Now compare what we got
+  equalUint8Arrays(uint8concat(KNOWN_BYTES_LONG), uint8concat(readChunks));
+  clientFinished.resolve();
+}, 1);
+
+await clientFinished.promise;
+clientSession.closed.catch((err) => {
+  // ignore the error
+});
+clientSession.close();