diff --git a/doc/api/fs.md b/doc/api/fs.md
index 865813069e16b1..5fd4cfeefb504d 100644
--- a/doc/api/fs.md
+++ b/doc/api/fs.md
@@ -3324,6 +3324,45 @@ a colon, Node.js will open a file system stream, as described by
 Functions based on `fs.open()` exhibit this behavior as well:
 `fs.writeFile()`, `fs.readFile()`, etc.
 
+### `fs.openAsBlob(path[, options])`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+* `path` {string|Buffer|URL}
+* `options` {Object}
+  * `type` {string} An optional mime type for the blob.
+* Return: {Promise} containing {Blob}
+
+Returns a {Blob} whose data is backed by the given file.
+
+The file must not be modified after the {Blob} is created. Any modifications
+will cause reading the {Blob} data to fail with a `DOMException`.
+error. Synchronous stat operations on the file when the `Blob` is created, and
+before each read in order to detect whether the file data has been modified
+on disk.
+
+```mjs
+import { openAsBlob } from 'node:fs';
+
+const blob = await openAsBlob('the.file.txt');
+const ab = await blob.arrayBuffer();
+blob.stream();
+```
+
+```cjs
+const { openAsBlob } = require('node:fs');
+
+(async () => {
+  const blob = await openAsBlob('the.file.txt');
+  const ab = await blob.arrayBuffer();
+  blob.stream();
+})();
+```
+
 ### `fs.opendir(path[, options], callback)`
 
 <!-- YAML
diff --git a/lib/fs.js b/lib/fs.js
index 300abd73fada23..847c43de7596dc 100644
--- a/lib/fs.js
+++ b/lib/fs.js
@@ -32,6 +32,7 @@ const {
   ObjectDefineProperties,
   ObjectDefineProperty,
   Promise,
+  PromiseResolve,
   ReflectApply,
   SafeMap,
   SafeSet,
@@ -62,6 +63,9 @@ const { isArrayBufferView } = require('internal/util/types');
 // it's re-initialized after deserialization.
 
 const binding = internalBinding('fs');
+
+const { createBlobFromFilePath } = require('internal/blob');
+
 const { Buffer } = require('buffer');
 const {
   aggregateTwoErrors,
@@ -586,6 +590,20 @@ function openSync(path, flags, mode) {
   return result;
 }
 
+/**
+ * @param {string | Buffer | URL } path
+ * @returns {Promise<Blob>}
+ */
+function openAsBlob(path, options = kEmptyObject) {
+  validateObject(options, 'options');
+  const type = options.type || '';
+  validateString(type, 'options.type');
+  // The underlying implementation here returns the Blob synchronously for now.
+  // To give ourselves flexibility to maybe return the Blob asynchronously,
+  // this API returns a Promise.
+  return PromiseResolve(createBlobFromFilePath(getValidatedPath(path), { type }));
+}
+
 /**
  * Reads file from the specified `fd` (file descriptor).
  * @param {number} fd
@@ -3022,6 +3040,7 @@ module.exports = fs = {
   mkdtempSync,
   open,
   openSync,
+  openAsBlob,
   readdir,
   readdirSync,
   read,
diff --git a/lib/internal/blob.js b/lib/internal/blob.js
index 832ecd41e076ae..6761ccae230816 100644
--- a/lib/internal/blob.js
+++ b/lib/internal/blob.js
@@ -1,14 +1,14 @@
 'use strict';
 
 const {
+  ArrayBuffer,
   ArrayFrom,
   MathMax,
   MathMin,
   ObjectDefineProperties,
   ObjectDefineProperty,
-  PromiseResolve,
   PromiseReject,
-  SafePromisePrototypeFinally,
+  PromiseResolve,
   ReflectConstruct,
   RegExpPrototypeExec,
   RegExpPrototypeSymbolReplace,
@@ -22,7 +22,8 @@ const {
 
 const {
   createBlob: _createBlob,
-  FixedSizeBlobCopyJob,
+  createBlobFromFilePath: _createBlobFromFilePath,
+  concat,
   getDataObject,
 } = internalBinding('blob');
 
@@ -48,17 +49,17 @@ const {
   customInspectSymbol: kInspect,
   kEmptyObject,
   kEnumerableProperty,
+  lazyDOMException,
 } = require('internal/util');
 const { inspect } = require('internal/util/inspect');
 
 const {
-  AbortError,
   codes: {
     ERR_INVALID_ARG_TYPE,
     ERR_INVALID_ARG_VALUE,
     ERR_INVALID_THIS,
     ERR_BUFFER_TOO_LARGE,
-  }
+  },
 } = require('internal/errors');
 
 const {
@@ -66,14 +67,13 @@ const {
   validateDictionary,
 } = require('internal/validators');
 
+const {
+  CountQueuingStrategy,
+} = require('internal/webstreams/queuingstrategies');
+
 const kHandle = Symbol('kHandle');
-const kState = Symbol('kState');
-const kIndex = Symbol('kIndex');
 const kType = Symbol('kType');
 const kLength = Symbol('kLength');
-const kArrayBufferPromise = Symbol('kArrayBufferPromise');
-
-const kMaxChunkSize = 65536;
 
 const disallowedTypeCharacters = /[^\u{0020}-\u{007E}]/u;
 
@@ -266,40 +266,35 @@ class Blob {
     if (!isBlob(this))
       return PromiseReject(new ERR_INVALID_THIS('Blob'));
 
-    // If there's already a promise in flight for the content,
-    // reuse it, but only while it's in flight. After the cached
-    // promise resolves it will be cleared, allowing it to be
-    // garbage collected as soon as possible.
-    if (this[kArrayBufferPromise])
-      return this[kArrayBufferPromise];
-
-    const job = new FixedSizeBlobCopyJob(this[kHandle]);
-
-    const ret = job.run();
-
-    // If the job returns a value immediately, the ArrayBuffer
-    // was generated synchronously and should just be returned
-    // directly.
-    if (ret !== undefined)
-      return PromiseResolve(ret);
+    if (this.size === 0) {
+      return PromiseResolve(new ArrayBuffer(0));
+    }
 
-    const {
-      promise,
-      resolve,
-      reject,
-    } = createDeferredPromise();
-
-    job.ondone = (err, ab) => {
-      if (err !== undefined)
-        return reject(new AbortError(undefined, { cause: err }));
-      resolve(ab);
+    const { promise, resolve, reject } = createDeferredPromise();
+    const reader = this[kHandle].getReader();
+    const buffers = [];
+    const readNext = () => {
+      reader.pull((status, buffer) => {
+        if (status === 0) {
+          // EOS, concat & resolve
+          // buffer should be undefined here
+          resolve(concat(buffers));
+          return;
+        } else if (status < 0) {
+          // The read could fail for many different reasons when reading
+          // from a non-memory resident blob part (e.g. file-backed blob).
+          // The error details the system error code.
+          const error = lazyDOMException('The blob could not be read', 'NotReadableError');
+          reject(error);
+          return;
+        }
+        if (buffer !== undefined)
+          buffers.push(buffer);
+        readNext();
+      });
     };
-    this[kArrayBufferPromise] =
-    SafePromisePrototypeFinally(
-      promise,
-      () => this[kArrayBufferPromise] = undefined);
-
-    return this[kArrayBufferPromise];
+    readNext();
+    return promise;
   }
 
   /**
@@ -321,24 +316,63 @@ class Blob {
     if (!isBlob(this))
       throw new ERR_INVALID_THIS('Blob');
 
-    const self = this;
+    if (this.size === 0) {
+      return new lazyReadableStream({
+        start(c) { c.close(); }
+      });
+    }
+
+    const reader = this[kHandle].getReader();
     return new lazyReadableStream({
-      async start() {
-        this[kState] = await self.arrayBuffer();
-        this[kIndex] = 0;
+      start(c) {
+        // There really should only be one read at a time so using an
+        // array here is purely defensive.
+        this.pendingPulls = [];
       },
-
-      pull(controller) {
-        if (this[kState].byteLength - this[kIndex] <= kMaxChunkSize) {
-          controller.enqueue(new Uint8Array(this[kState], this[kIndex]));
-          controller.close();
-          this[kState] = undefined;
-        } else {
-          controller.enqueue(new Uint8Array(this[kState], this[kIndex], kMaxChunkSize));
-          this[kIndex] += kMaxChunkSize;
+      pull(c) {
+        const { promise, resolve, reject } = createDeferredPromise();
+        this.pendingPulls.push({ resolve, reject });
+        reader.pull((status, buffer) => {
+          // If pendingPulls is empty here, the stream had to have
+          // been canceled, and we don't really care about the result.
+          // we can simply exit.
+          if (this.pendingPulls.length === 0) {
+            return;
+          }
+          const pending = this.pendingPulls.shift();
+          if (status === 0) {
+            // EOS
+            c.close();
+            pending.resolve();
+            return;
+          } else if (status < 0) {
+            // The read could fail for many different reasons when reading
+            // from a non-memory resident blob part (e.g. file-backed blob).
+            // The error details the system error code.
+            const error = lazyDOMException('The blob could not be read', 'NotReadableError');
+
+            c.error(error);
+            pending.reject(error);
+            return;
+          }
+          if (buffer !== undefined) {
+            c.enqueue(new Uint8Array(buffer));
+          }
+          pending.resolve();
+        });
+        return promise;
+      },
+      cancel(reason) {
+        // Reject any currently pending pulls here.
+        for (const pending of this.pendingPulls) {
+          pending.reject(reason);
         }
+        this.pendingPulls = [];
       }
-    });
+    // We set the highWaterMark to 0 because we do not want the stream to
+    // start reading immediately on creation. We want it to wait until read
+    // is called.
+    }, new CountQueuingStrategy({ highWaterMark: 0 }));
   }
 }
 
@@ -406,10 +440,22 @@ function resolveObjectURL(url) {
   }
 }
 
+// TODO(@jasnell): Now that the File class exists, we might consider having
+// this return a `File` instead of a `Blob`.
+function createBlobFromFilePath(path, options) {
+  const maybeBlob = _createBlobFromFilePath(path);
+  if (maybeBlob === undefined) {
+    return lazyDOMException('The blob could not be read', 'NotReadableError');
+  }
+  const { 0: blob, 1: length } = maybeBlob;
+  return createBlob(blob, length, options?.type);
+}
+
 module.exports = {
   Blob,
   ClonedBlob,
   createBlob,
+  createBlobFromFilePath,
   isBlob,
   kHandle,
   resolveObjectURL,
diff --git a/node.gyp b/node.gyp
index 6d1b2bf36902cf..f89091774721b7 100644
--- a/node.gyp
+++ b/node.gyp
@@ -477,6 +477,7 @@
         'src/cleanup_queue.cc',
         'src/connect_wrap.cc',
         'src/connection_wrap.cc',
+        'src/dataqueue/queue.cc',
         'src/debug_utils.cc',
         'src/env.cc',
         'src/fs_event_wrap.cc',
@@ -580,6 +581,7 @@
         'src/cleanup_queue-inl.h',
         'src/connect_wrap.h',
         'src/connection_wrap.h',
+        'src/dataqueue/queue.h',
         'src/debug_utils.h',
         'src/debug_utils-inl.h',
         'src/env_properties.h',
@@ -991,6 +993,7 @@
         'test/cctest/test_sockaddr.cc',
         'test/cctest/test_traced_value.cc',
         'test/cctest/test_util.cc',
+        'test/cctest/test_dataqueue.cc',
       ],
 
       'conditions': [
diff --git a/src/async_wrap.h b/src/async_wrap.h
index 71a205376d2767..121216579a8ad7 100644
--- a/src/async_wrap.h
+++ b/src/async_wrap.h
@@ -31,49 +31,49 @@
 
 namespace node {
 
-#define NODE_ASYNC_NON_CRYPTO_PROVIDER_TYPES(V)                               \
-  V(NONE)                                                                     \
-  V(DIRHANDLE)                                                                \
-  V(DNSCHANNEL)                                                               \
-  V(ELDHISTOGRAM)                                                             \
-  V(FILEHANDLE)                                                               \
-  V(FILEHANDLECLOSEREQ)                                                       \
-  V(FIXEDSIZEBLOBCOPY)                                                        \
-  V(FSEVENTWRAP)                                                              \
-  V(FSREQCALLBACK)                                                            \
-  V(FSREQPROMISE)                                                             \
-  V(GETADDRINFOREQWRAP)                                                       \
-  V(GETNAMEINFOREQWRAP)                                                       \
-  V(HEAPSNAPSHOT)                                                             \
-  V(HTTP2SESSION)                                                             \
-  V(HTTP2STREAM)                                                              \
-  V(HTTP2PING)                                                                \
-  V(HTTP2SETTINGS)                                                            \
-  V(HTTPINCOMINGMESSAGE)                                                      \
-  V(HTTPCLIENTREQUEST)                                                        \
-  V(JSSTREAM)                                                                 \
-  V(JSUDPWRAP)                                                                \
-  V(MESSAGEPORT)                                                              \
-  V(PIPECONNECTWRAP)                                                          \
-  V(PIPESERVERWRAP)                                                           \
-  V(PIPEWRAP)                                                                 \
-  V(PROCESSWRAP)                                                              \
-  V(PROMISE)                                                                  \
-  V(QUERYWRAP)                                                                \
-  V(SHUTDOWNWRAP)                                                             \
-  V(SIGNALWRAP)                                                               \
-  V(STATWATCHER)                                                              \
-  V(STREAMPIPE)                                                               \
-  V(TCPCONNECTWRAP)                                                           \
-  V(TCPSERVERWRAP)                                                            \
-  V(TCPWRAP)                                                                  \
-  V(TTYWRAP)                                                                  \
-  V(UDPSENDWRAP)                                                              \
-  V(UDPWRAP)                                                                  \
-  V(SIGINTWATCHDOG)                                                           \
-  V(WORKER)                                                                   \
-  V(WORKERHEAPSNAPSHOT)                                                       \
-  V(WRITEWRAP)                                                                \
+#define NODE_ASYNC_NON_CRYPTO_PROVIDER_TYPES(V)                                \
+  V(NONE)                                                                      \
+  V(DIRHANDLE)                                                                 \
+  V(DNSCHANNEL)                                                                \
+  V(ELDHISTOGRAM)                                                              \
+  V(FILEHANDLE)                                                                \
+  V(FILEHANDLECLOSEREQ)                                                        \
+  V(BLOBREADER)                                                                \
+  V(FSEVENTWRAP)                                                               \
+  V(FSREQCALLBACK)                                                             \
+  V(FSREQPROMISE)                                                              \
+  V(GETADDRINFOREQWRAP)                                                        \
+  V(GETNAMEINFOREQWRAP)                                                        \
+  V(HEAPSNAPSHOT)                                                              \
+  V(HTTP2SESSION)                                                              \
+  V(HTTP2STREAM)                                                               \
+  V(HTTP2PING)                                                                 \
+  V(HTTP2SETTINGS)                                                             \
+  V(HTTPINCOMINGMESSAGE)                                                       \
+  V(HTTPCLIENTREQUEST)                                                         \
+  V(JSSTREAM)                                                                  \
+  V(JSUDPWRAP)                                                                 \
+  V(MESSAGEPORT)                                                               \
+  V(PIPECONNECTWRAP)                                                           \
+  V(PIPESERVERWRAP)                                                            \
+  V(PIPEWRAP)                                                                  \
+  V(PROCESSWRAP)                                                               \
+  V(PROMISE)                                                                   \
+  V(QUERYWRAP)                                                                 \
+  V(SHUTDOWNWRAP)                                                              \
+  V(SIGNALWRAP)                                                                \
+  V(STATWATCHER)                                                               \
+  V(STREAMPIPE)                                                                \
+  V(TCPCONNECTWRAP)                                                            \
+  V(TCPSERVERWRAP)                                                             \
+  V(TCPWRAP)                                                                   \
+  V(TTYWRAP)                                                                   \
+  V(UDPSENDWRAP)                                                               \
+  V(UDPWRAP)                                                                   \
+  V(SIGINTWATCHDOG)                                                            \
+  V(WORKER)                                                                    \
+  V(WORKERHEAPSNAPSHOT)                                                        \
+  V(WRITEWRAP)                                                                 \
   V(ZLIB)
 
 #if HAVE_OPENSSL
diff --git a/src/dataqueue/queue.cc b/src/dataqueue/queue.cc
new file mode 100644
index 00000000000000..a39314e23e13bf
--- /dev/null
+++ b/src/dataqueue/queue.cc
@@ -0,0 +1,1113 @@
+#include "queue.h"
+#include <async_wrap-inl.h>
+#include <base_object-inl.h>
+#include <env-inl.h>
+#include <memory_tracker-inl.h>
+#include <node.h>
+#include <node_bob-inl.h>
+#include <node_errors.h>
+#include <node_external_reference.h>
+#include <node_file-inl.h>
+#include <stream_base-inl.h>
+#include <util-inl.h>
+#include <uv.h>
+#include <v8.h>
+#include <algorithm>
+#include <deque>
+#include <initializer_list>
+#include <memory>
+#include <vector>
+
+namespace node {
+
+using v8::ArrayBufferView;
+using v8::BackingStore;
+using v8::Local;
+using v8::Object;
+using v8::Value;
+
+namespace {
+// ============================================================================
+class IdempotentDataQueueReader;
+class NonIdempotentDataQueueReader;
+
+class EntryImpl : public DataQueue::Entry {
+ public:
+  virtual std::shared_ptr<DataQueue::Reader> get_reader() = 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)
+      : entries_(std::move(list)),
+        idempotent_(true),
+        size_(size),
+        capped_size_(0) {}
+
+  // 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) {}
+
+  // Disallow moving and copying.
+  DataQueueImpl(const DataQueueImpl&) = delete;
+  DataQueueImpl(DataQueueImpl&&) = delete;
+  DataQueueImpl& operator=(const DataQueueImpl&) = delete;
+  DataQueueImpl& operator=(DataQueueImpl&&) = delete;
+
+  std::shared_ptr<DataQueue> slice(
+      uint64_t start,
+      std::optional<uint64_t> maybeEnd = std::nullopt) override {
+    // If the data queue is not idempotent, or the size cannot be determined,
+    // we cannot reasonably create a slice. Therefore, return nothing.
+    if (!idempotent_ || !size_.has_value()) return nullptr;
+
+    uint64_t size = size_.value();
+
+    // start cannot be greater than the size.
+    start = std::min(start, size);
+
+    uint64_t end = std::max(start, std::min(maybeEnd.value_or(size), size));
+
+    DCHECK_LE(start, end);
+
+    uint64_t len = end - start;
+    uint64_t remaining = end - start;
+    std::vector<std::unique_ptr<Entry>> slices;
+
+    if (remaining > 0) {
+      for (const auto& entry : entries_) {
+        // The size of every entry should be known since this is an
+        // idempotent queue.
+        uint64_t entrySize = entry->size().value();
+        if (start > entrySize) {
+          start -= entrySize;
+          continue;
+        }
+
+        uint64_t chunkStart = start;
+        uint64_t len = std::min(remaining, entrySize - chunkStart);
+        slices.emplace_back(entry->slice(chunkStart, chunkStart + len));
+        remaining -= len;
+        start = 0;
+
+        if (remaining == 0) break;
+      }
+    }
+
+    return std::make_shared<DataQueueImpl>(std::move(slices), len);
+  }
+
+  std::optional<uint64_t> size() const override { return size_; }
+
+  bool is_idempotent() const override { return idempotent_; }
+
+  bool is_capped() const override { return capped_size_.has_value(); }
+
+  std::optional<bool> append(std::unique_ptr<Entry> entry) override {
+    if (idempotent_) return std::nullopt;
+    if (!entry) return false;
+
+    // If this entry successfully provides a size, we can add it to our size_
+    // if that has a value, otherwise, we keep uint64_t empty.
+    if (entry->size().has_value() && size_.has_value()) {
+      uint64_t entrySize = entry->size().value();
+      uint64_t size = size_.value();
+      // If capped_size_ is set, size + entrySize cannot exceed capped_size_
+      // or the entry cannot be added.
+      if (capped_size_.has_value() &&
+          (capped_size_.value() < entrySize + size)) {
+        return false;
+      }
+      size_ = size + entrySize;
+    } else {
+      // This entry cannot provide a size. We can still add it but we have to
+      // clear the known size.
+      size_ = std::nullopt;
+    }
+
+    entries_.push_back(std::move(entry));
+    return true;
+  }
+
+  void cap(uint64_t limit = 0) override {
+    if (is_idempotent()) return;
+    // If the data queue is already capped, it is possible to call
+    // cap again with a smaller size.
+    if (capped_size_.has_value()) {
+      capped_size_ = std::min(limit, capped_size_.value());
+      return;
+    }
+
+    // Otherwise just set the limit.
+    capped_size_ = limit;
+  }
+
+  std::optional<uint64_t> maybeCapRemaining() const override {
+    if (capped_size_.has_value() && size_.has_value()) {
+      uint64_t capped_size = capped_size_.value();
+      uint64_t size = size_.value();
+      return capped_size > size ? capped_size - size : 0UL;
+    }
+    return std::nullopt;
+  }
+
+  void MemoryInfo(node::MemoryTracker* tracker) const override {
+    tracker->TrackField(
+        "entries", entries_, "std::vector<std::unique_ptr<Entry>>");
+  }
+
+  std::shared_ptr<Reader> get_reader() override;
+  SET_MEMORY_INFO_NAME(DataQueue)
+  SET_SELF_SIZE(DataQueueImpl)
+
+ private:
+  std::vector<std::unique_ptr<Entry>> entries_;
+  bool idempotent_;
+  std::optional<uint64_t> size_ = std::nullopt;
+  std::optional<uint64_t> capped_size_ = std::nullopt;
+  bool locked_to_reader_ = false;
+
+  friend class DataQueue;
+  friend class IdempotentDataQueueReader;
+  friend class NonIdempotentDataQueueReader;
+};
+
+// An IdempotentDataQueueReader always reads the entire content of the
+// DataQueue with which it is associated, and always from the beginning.
+// Reads are non-destructive, meaning that the state of the DataQueue
+// will not and cannot be changed.
+class IdempotentDataQueueReader final
+    : public DataQueue::Reader,
+      public std::enable_shared_from_this<IdempotentDataQueueReader> {
+ public:
+  IdempotentDataQueueReader(std::shared_ptr<DataQueueImpl> data_queue)
+      : data_queue_(std::move(data_queue)) {
+    CHECK(data_queue_->is_idempotent());
+  }
+
+  // Disallow moving and copying.
+  IdempotentDataQueueReader(const IdempotentDataQueueReader&) = delete;
+  IdempotentDataQueueReader(IdempotentDataQueueReader&&) = delete;
+  IdempotentDataQueueReader& operator=(const IdempotentDataQueueReader&) =
+      delete;
+  IdempotentDataQueueReader& operator=(IdempotentDataQueueReader&&) = delete;
+
+  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();
+
+    // If ended is true, this reader has already reached the end and cannot
+    // provide any more data.
+    if (ended_) {
+      std::move(next)(bob::Status::STATUS_EOS, nullptr, 0, [](uint64_t) {});
+      return bob::Status::STATUS_EOS;
+    }
+
+    // If this is the first pull from this reader, we are first going to
+    // check to see if there is anything at all to actually do.
+    if (!current_index_.has_value()) {
+      // First, let's check the number of entries. If there are no entries,
+      // we've reached the end and have nothing to do.
+      // Because this is an idempotent dataqueue, we should always know the
+      // size...
+      if (data_queue_->entries_.empty()) {
+        ended_ = true;
+        std::move(next)(bob::Status::STATUS_EOS, nullptr, 0, [](uint64_t) {});
+        return bob::Status::STATUS_EOS;
+      }
+
+      current_index_ = 0;
+    }
+
+    // We have current_index_, awesome, we are going to keep reading from
+    // it until we receive and end.
+
+    auto current_reader = getCurrentReader();
+    if (current_reader == nullptr) {
+      // Getting the current reader for an entry could fail for several
+      // reasons. For an FdEntry, for instance, getting the reader may
+      // fail if the file has been modified since the FdEntry was created.
+      // We handle the case simply by erroring.
+      std::move(next)(UV_EINVAL, nullptr, 0, [](uint64_t) {});
+      return UV_EINVAL;
+    }
+
+    CHECK(!pull_pending_);
+    pull_pending_ = true;
+    int status = current_reader->Pull(
+        [this, next = std::move(next)](
+            int status, const DataQueue::Vec* vecs, uint64_t count, Done done) {
+          pull_pending_ = false;
+          // In each of these cases, we do not expect that the source will
+          // actually have provided any actual data.
+          CHECK_IMPLIES(status == bob::Status::STATUS_BLOCK ||
+                            status == bob::Status::STATUS_WAIT ||
+                            status == bob::Status::STATUS_EOS,
+                        vecs == nullptr && count == 0);
+          if (status == bob::Status::STATUS_EOS) {
+            uint32_t current = current_index_.value() + 1;
+            current_reader_ = nullptr;
+            // We have reached the end of this entry. If this is the last entry,
+            // then we are done. Otherwise, we advance the current_index_, clear
+            // the current_reader_ and wait for the next read.
+
+            if (current == data_queue_->entries_.size()) {
+              // Yes, this was the final entry. We're all done.
+              ended_ = true;
+            } else {
+              // This was not the final entry, so we update the index and
+              // continue on by performing another read.
+              current_index_ = current;
+              status = bob::STATUS_CONTINUE;
+            }
+            std::move(next)(status, nullptr, 0, [](uint64_t) {});
+            return;
+          }
+
+          std::move(next)(status, vecs, count, std::move(done));
+        },
+        options,
+        data,
+        count,
+        max_count_hint);
+
+    // The pull was handled synchronously. If we're not ended, we want to
+    // make sure status returned is CONTINUE.
+    if (!pull_pending_) {
+      if (!ended_) return bob::Status::STATUS_CONTINUE;
+      // For all other status, we just fall through and return it straightaway.
+    }
+
+    // The other statuses that can be returned by the pull are:
+    //  bob::Status::STATUS_CONTINUE - means that the entry has more data
+    //                                 to pull.
+    //  bob::Status::STATUS_BLOCK - means that the entry has more data to
+    //                              pull but it is not available yet. The
+    //                              caller should not keep calling pull for
+    //                              now but may check again later.
+    //  bob::Status::STATUS_WAIT - means that the entry has more data to
+    //                             pull but it won't be provided
+    //                             synchronously, instead the next() callback
+    //                             will be called when the data is available.
+    //
+    // For any of these statuses, we want to keep the current index and
+    // current_reader_ set for the next pull.
+
+    return status;
+  }
+
+  DataQueue::Reader* getCurrentReader() {
+    CHECK(!ended_);
+    CHECK(current_index_.has_value());
+    if (current_reader_ == nullptr) {
+      auto& entry = data_queue_->entries_[current_index_.value()];
+      // Because this is an idempotent reader, let's just be sure to
+      // doublecheck that the entry itself is actually idempotent
+      DCHECK(entry->is_idempotent());
+      current_reader_ = static_cast<EntryImpl&>(*entry).get_reader();
+    }
+    return current_reader_.get();
+  }
+
+  SET_NO_MEMORY_INFO()
+  SET_MEMORY_INFO_NAME(IdempotentDataQueueReader)
+  SET_SELF_SIZE(IdempotentDataQueueReader)
+
+ private:
+  std::shared_ptr<DataQueueImpl> data_queue_;
+  std::optional<uint32_t> current_index_ = std::nullopt;
+  std::shared_ptr<DataQueue::Reader> current_reader_ = nullptr;
+  bool ended_ = false;
+  bool pull_pending_ = false;
+};
+
+// A NonIdempotentDataQueueReader reads entries from the DataEnqueue
+// and removes those entries from the queue as they are fully consumed.
+// This means that reads are destructive and the state of the DataQueue
+// is mutated as the read proceeds.
+class NonIdempotentDataQueueReader final
+    : public DataQueue::Reader,
+      public std::enable_shared_from_this<NonIdempotentDataQueueReader> {
+ public:
+  NonIdempotentDataQueueReader(std::shared_ptr<DataQueueImpl> data_queue)
+      : data_queue_(std::move(data_queue)) {
+    CHECK(!data_queue_->is_idempotent());
+  }
+
+  // Disallow moving and copying.
+  NonIdempotentDataQueueReader(const NonIdempotentDataQueueReader&) = delete;
+  NonIdempotentDataQueueReader(NonIdempotentDataQueueReader&&) = delete;
+  NonIdempotentDataQueueReader& operator=(const NonIdempotentDataQueueReader&) =
+      delete;
+  NonIdempotentDataQueueReader& operator=(NonIdempotentDataQueueReader&&) =
+      delete;
+
+  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();
+
+    // If ended is true, this reader has already reached the end and cannot
+    // provide any more data.
+    if (ended_) {
+      std::move(next)(bob::Status::STATUS_EOS, nullptr, 0, [](uint64_t) {});
+      return bob::Status::STATUS_EOS;
+    }
+
+    // If the collection of entries is empty, there's nothing currently left to
+    // read. How we respond depends on whether the data queue has been capped
+    // or not.
+    if (data_queue_->entries_.empty()) {
+      // If the data_queue_ is empty, and not capped, then we can reasonably
+      // expect more data to be provided later, but we don't know exactly when
+      // that'll happe, so the proper response here is to return a blocked
+      // status.
+      if (!data_queue_->is_capped()) {
+        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
+      // of the data_queue_ is known or not.
+
+      if (data_queue_->size().has_value()) {
+        // If the size is known, and it is still less than the cap, then we
+        // 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;
+        }
+
+        // Otherwise, if size is equal to or greater than capped, we are done.
+        // Fall through to allow the end handling to run.
+      }
+
+      // If the size is not known, and the data queue is capped, no additional
+      // entries are going to be added to the queue. Since we are all out of
+      // entries, we're done. There's nothing left to read.
+      current_reader_ = nullptr;
+      ended_ = true;
+      std::move(next)(bob::Status::STATUS_EOS, nullptr, 0, [](uint64_t) {});
+      return bob::STATUS_EOS;
+    }
+
+    auto current_reader = getCurrentReader();
+    if (current_reader == nullptr) {
+      std::move(next)(UV_EINVAL, nullptr, 0, [](uint64_t) {});
+      return UV_EINVAL;
+    }
+
+    // If we got here, we have an entry to read from.
+    CHECK(!pull_pending_);
+    pull_pending_ = true;
+    int status = current_reader->Pull(
+        [this, next = std::move(next)](
+            int status, const DataQueue::Vec* vecs, uint64_t count, Done done) {
+          pull_pending_ = false;
+
+          // In each of these cases, we do not expect that the source will
+          // actually have provided any actual data.
+          CHECK_IMPLIES(status == bob::Status::STATUS_BLOCK ||
+                            status == bob::Status::STATUS_WAIT ||
+                            status == bob::Status::STATUS_EOS,
+                        vecs == nullptr && count == 0);
+          if (status == bob::Status::STATUS_EOS) {
+            data_queue_->entries_.erase(data_queue_->entries_.begin());
+            ended_ = data_queue_->entries_.empty();
+            current_reader_ = nullptr;
+            if (!ended_) status = bob::Status::STATUS_CONTINUE;
+            std::move(next)(status, nullptr, 0, [](uint64_t) {});
+            return;
+          }
+
+          // 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));
+        },
+        options,
+        data,
+        count,
+        max_count_hint);
+
+    if (!pull_pending_) {
+      // The callback was resolved synchronously. Let's check our status.
+      if (!ended_) return bob::Status::STATUS_CONTINUE;
+      // For all other status, we just fall through and return it straightaway.
+    }
+
+    // The other statuses that can be returned by the pull are:
+    //  bob::Status::STATUS_CONTINUE - means that the entry has more data
+    //                                 to pull.
+    //  bob::Status::STATUS_BLOCK - means that the entry has more data to
+    //                              pull but it is not available yet. The
+    //                              caller should not keep calling pull for
+    //                              now but may check again later.
+    //  bob::Status::STATUS_WAIT - means that the entry has more data to
+    //                             pull but it won't be provided
+    //                             synchronously, instead the next() callback
+    //                             will be called when the data is available.
+    //
+    // For any of these statuses, we want to keep the current index and
+    // current_reader_ set for the next pull.
+
+    return status;
+  }
+
+  DataQueue::Reader* getCurrentReader() {
+    CHECK(!ended_);
+    CHECK(!data_queue_->entries_.empty());
+    if (current_reader_ == nullptr) {
+      auto& entry = data_queue_->entries_.front();
+      current_reader_ = static_cast<EntryImpl&>(*entry).get_reader();
+    }
+    return current_reader_.get();
+  }
+
+  SET_NO_MEMORY_INFO()
+  SET_MEMORY_INFO_NAME(NonIdempotentDataQueueReader)
+  SET_SELF_SIZE(NonIdempotentDataQueueReader)
+
+ private:
+  std::shared_ptr<DataQueueImpl> data_queue_;
+  std::shared_ptr<DataQueue::Reader> current_reader_ = nullptr;
+  bool ended_ = false;
+  bool pull_pending_ = false;
+};
+
+std::shared_ptr<DataQueue::Reader> DataQueueImpl::get_reader() {
+  if (is_idempotent()) {
+    return std::make_shared<IdempotentDataQueueReader>(shared_from_this());
+  }
+
+  if (locked_to_reader_) return nullptr;
+  locked_to_reader_ = true;
+
+  return std::make_shared<NonIdempotentDataQueueReader>(shared_from_this());
+}
+
+// ============================================================================
+
+// An empty, always idempotent entry.
+class EmptyEntry final : public EntryImpl {
+ public:
+  class EmptyReader final : public DataQueue::Reader,
+                            public std::enable_shared_from_this<EmptyReader> {
+   public:
+    int Pull(Next next,
+             int options,
+             DataQueue::Vec* data,
+             size_t count,
+             size_t max_count_hint = bob::kMaxCountHint) override {
+      auto self = shared_from_this();
+      if (ended_) {
+        std::move(next)(bob::Status::STATUS_EOS, nullptr, 0, [](uint64_t) {});
+        return bob::Status::STATUS_EOS;
+      }
+
+      ended_ = true;
+      std::move(next)(
+          bob::Status::STATUS_CONTINUE, nullptr, 0, [](uint64_t) {});
+      return bob::Status::STATUS_CONTINUE;
+    }
+
+    SET_NO_MEMORY_INFO()
+    SET_MEMORY_INFO_NAME(EmptyReader)
+    SET_SELF_SIZE(EmptyReader)
+
+   private:
+    bool ended_ = false;
+  };
+
+  EmptyEntry() = default;
+
+  // Disallow moving and copying.
+  EmptyEntry(const EmptyEntry&) = delete;
+  EmptyEntry(EmptyEntry&&) = delete;
+  EmptyEntry& operator=(const EmptyEntry&) = delete;
+  EmptyEntry& operator=(EmptyEntry&&) = delete;
+
+  std::shared_ptr<DataQueue::Reader> get_reader() override {
+    return std::make_shared<EmptyReader>();
+  }
+
+  std::unique_ptr<Entry> slice(
+      uint64_t start,
+      std::optional<uint64_t> maybeEnd = std::nullopt) override {
+    if (start != 0) return nullptr;
+    return std::make_unique<EmptyEntry>();
+  }
+
+  std::optional<uint64_t> size() const override { return 0; }
+
+  bool is_idempotent() const override { return true; }
+
+  SET_NO_MEMORY_INFO()
+  SET_MEMORY_INFO_NAME(EmptyEntry)
+  SET_SELF_SIZE(EmptyEntry)
+};
+
+// ============================================================================
+
+// An entry that consists of a single memory resident v8::BackingStore.
+// These are always idempotent and always a fixed, known size.
+class InMemoryEntry final : public EntryImpl {
+ public:
+  struct InMemoryFunctor final {
+    std::shared_ptr<BackingStore> backing_store;
+    void operator()(uint64_t) { backing_store = nullptr; }
+  };
+
+  class InMemoryReader final
+      : public DataQueue::Reader,
+        public std::enable_shared_from_this<InMemoryReader> {
+   public:
+    InMemoryReader(InMemoryEntry& entry) : entry_(entry) {}
+
+    int Pull(Next next,
+             int options,
+             DataQueue::Vec* data,
+             size_t count,
+             size_t max_count_hint = bob::kMaxCountHint) override {
+      auto self = shared_from_this();
+      if (ended_) {
+        std::move(next)(bob::Status::STATUS_EOS, nullptr, 0, [](uint64_t) {});
+        return bob::Status::STATUS_EOS;
+      }
+
+      ended_ = true;
+      DataQueue::Vec vec{
+          reinterpret_cast<uint8_t*>(entry_.backing_store_->Data()) +
+              entry_.offset_,
+          entry_.byte_length_,
+      };
+
+      std::move(next)(bob::Status::STATUS_CONTINUE,
+                      &vec,
+                      1,
+                      InMemoryFunctor({entry_.backing_store_}));
+      return bob::Status::STATUS_CONTINUE;
+    }
+
+    SET_NO_MEMORY_INFO()
+    SET_MEMORY_INFO_NAME(InMemoryReader)
+    SET_SELF_SIZE(InMemoryReader)
+
+   private:
+    InMemoryEntry& entry_;
+    bool ended_ = false;
+  };
+
+  InMemoryEntry(std::shared_ptr<BackingStore> backing_store,
+                uint64_t offset,
+                uint64_t byte_length)
+      : backing_store_(std::move(backing_store)),
+        offset_(offset),
+        byte_length_(byte_length) {
+    // The offset_ + byte_length_ cannot extend beyond the size of the
+    // backing store, because that would just be silly.
+    CHECK_LE(offset_ + byte_length_, backing_store_->ByteLength());
+  }
+
+  // Disallow moving and copying.
+  InMemoryEntry(const InMemoryEntry&) = delete;
+  InMemoryEntry(InMemoryEntry&&) = delete;
+  InMemoryEntry& operator=(const InMemoryEntry&) = delete;
+  InMemoryEntry& operator=(InMemoryEntry&&) = delete;
+
+  std::shared_ptr<DataQueue::Reader> get_reader() override {
+    return std::make_shared<InMemoryReader>(*this);
+  }
+
+  std::unique_ptr<Entry> slice(
+      uint64_t start,
+      std::optional<uint64_t> maybeEnd = std::nullopt) override {
+    const auto makeEntry = [&](uint64_t start,
+                               uint64_t len) -> std::unique_ptr<Entry> {
+      if (len == 0) {
+        return std::make_unique<EmptyEntry>();
+      }
+
+      return std::make_unique<InMemoryEntry>(backing_store_, start, len);
+    };
+
+    start += offset_;
+
+    // The start cannot extend beyond the maximum end point of this entry.
+    start = std::min(start, offset_ + byte_length_);
+
+    if (maybeEnd.has_value()) {
+      uint64_t end = maybeEnd.value();
+      // The end cannot extend beyond the maximum end point of this entry,
+      // and the end must be equal to or greater than the start.
+      end = std::max(start, std::min(offset_ + end, offset_ + byte_length_));
+
+      return makeEntry(start, end - start);
+    }
+
+    // If no end is given, then the new length is the current length
+    // minus the adjusted start.
+    return makeEntry(start, byte_length_ - start);
+  }
+
+  std::optional<uint64_t> size() const override { return byte_length_; }
+
+  bool is_idempotent() const override { return true; }
+
+  void MemoryInfo(node::MemoryTracker* tracker) const override {
+    tracker->TrackField(
+        "store", backing_store_, "std::shared_ptr<v8::BackingStore>");
+  }
+  SET_MEMORY_INFO_NAME(InMemoryEntry)
+  SET_SELF_SIZE(InMemoryEntry)
+
+ private:
+  std::shared_ptr<BackingStore> backing_store_;
+  uint64_t offset_;
+  uint64_t byte_length_;
+
+  friend class InMemoryReader;
+};
+
+// ============================================================================
+
+// An entry that wraps a DataQueue. The entry takes on the characteristics
+// of the wrapped dataqueue.
+class DataQueueEntry : public EntryImpl {
+ public:
+  explicit DataQueueEntry(std::shared_ptr<DataQueue> data_queue)
+      : data_queue_(std::move(data_queue)) {
+    CHECK(data_queue_);
+  }
+
+  // Disallow moving and copying.
+  DataQueueEntry(const DataQueueEntry&) = delete;
+  DataQueueEntry(DataQueueEntry&&) = delete;
+  DataQueueEntry& operator=(const DataQueueEntry&) = delete;
+  DataQueueEntry& operator=(DataQueueEntry&&) = delete;
+
+  std::shared_ptr<DataQueue::Reader> get_reader() override {
+    return std::make_shared<ReaderImpl>(data_queue_->get_reader());
+  }
+
+  std::unique_ptr<Entry> slice(
+      uint64_t start, std::optional<uint64_t> end = std::nullopt) override {
+    std::shared_ptr<DataQueue> sliced = data_queue_->slice(start, end);
+    if (!sliced) return nullptr;
+
+    return std::make_unique<DataQueueEntry>(std::move(sliced));
+  }
+
+  // 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
+  // a value. In such cases, size() must return std::nullopt.
+  //
+  // If the entry is idempotent, a size should always be available.
+  std::optional<uint64_t> size() const override { return data_queue_->size(); }
+
+  // When true, multiple reads on the object must produce the exact
+  // same data or the reads will fail. Some sources of entry data,
+  // such as streams, may not be capable of preserving idempotency
+  // and therefore must not claim to be. If an entry claims to be
+  // idempotent and cannot preserve that quality, subsequent reads
+  // must fail with an error when a variance is detected.
+  bool is_idempotent() const override { return data_queue_->is_idempotent(); }
+
+  void MemoryInfo(node::MemoryTracker* tracker) const override {
+    tracker->TrackField(
+        "data_queue", data_queue_, "std::shared_ptr<DataQueue>");
+  }
+
+  DataQueue& getDataQueue() { return *data_queue_; }
+
+  SET_MEMORY_INFO_NAME(DataQueueEntry)
+  SET_SELF_SIZE(DataQueueEntry)
+
+ private:
+  std::shared_ptr<DataQueue> data_queue_;
+
+  class ReaderImpl : public DataQueue::Reader,
+                     public std::enable_shared_from_this<ReaderImpl> {
+   public:
+    explicit ReaderImpl(std::shared_ptr<DataQueue::Reader> inner)
+        : inner_(std::move(inner)) {}
+
+    int Pull(DataQueue::Reader::Next next,
+             int options,
+             DataQueue::Vec* data,
+             size_t count,
+             size_t max_count_hint) override {
+      auto self = shared_from_this();
+      return inner_->Pull(
+          std::move(next), options, data, count, max_count_hint);
+    }
+
+    SET_NO_MEMORY_INFO()
+    SET_MEMORY_INFO_NAME(ReaderImpl)
+    SET_SELF_SIZE(ReaderImpl)
+
+   private:
+    std::shared_ptr<DataQueue::Reader> inner_;
+  };
+};
+
+// ============================================================================
+
+// An FdEntry reads from a file descriptor. A check is made before each read
+// to determine if the fd has changed on disc. This is a best-effort check
+// that only looks at file size, creation, and modification times. The stat
+// check is also async, so there's a natural race condition there where the
+// file could be modified between the stat and actual read calls. That's
+// a tolerable risk here. While FdEntry is considered idempotent, this race
+// means that it is indeed possible for multiple reads to return different
+// results if the file just happens to get modified.
+class FdEntry final : public EntryImpl {
+  // TODO(@jasnell, @flakey5):
+  // * This should only allow reading from regular files. No directories, no
+  // pipes, etc.
+  // * The reader should support accepting the buffer(s) from the pull, if any.
+  // It should
+  //   only allocate a managed buffer if the pull doesn't provide any.
+  // * We might want to consider making the stat on each read sync to eliminate
+  // the race
+  //   condition described in the comment above.
+ public:
+  static std::unique_ptr<FdEntry> Create(Environment* env, Local<Value> path) {
+    // We're only going to create the FdEntry if the file exists.
+    uv_fs_t req;
+    auto cleanup = OnScopeLeave([&] { uv_fs_req_cleanup(&req); });
+
+    auto buf = std::make_shared<BufferValue>(env->isolate(), path);
+    if (uv_fs_stat(nullptr, &req, buf->out(), nullptr) < 0) return nullptr;
+
+    return std::make_unique<FdEntry>(
+        env, std::move(buf), req.statbuf, 0, req.statbuf.st_size);
+  }
+
+  FdEntry(Environment* env,
+          std::shared_ptr<BufferValue> path_,
+          uv_stat_t stat,
+          uint64_t start,
+          uint64_t end)
+      : env_(env),
+        path_(std::move(path_)),
+        stat_(stat),
+        start_(start),
+        end_(end) {}
+
+  std::shared_ptr<DataQueue::Reader> get_reader() override {
+    return ReaderImpl::Create(this);
+  }
+
+  std::unique_ptr<Entry> slice(
+      uint64_t start, std::optional<uint64_t> end = std::nullopt) override {
+    uint64_t new_start = start_ + start;
+    uint64_t new_end = end_;
+    if (end.has_value()) {
+      new_end = std::min(end.value() + start, new_end);
+    }
+
+    CHECK(new_start >= start_);
+    CHECK(new_end <= end_);
+
+    return std::make_unique<FdEntry>(env_, path_, stat_, new_start, new_end);
+  }
+
+  std::optional<uint64_t> size() const override { return end_ - start_; }
+
+  bool is_idempotent() const override { return true; }
+
+  Environment* env() const { return env_; }
+
+  SET_NO_MEMORY_INFO()
+  SET_MEMORY_INFO_NAME(FdEntry)
+  SET_SELF_SIZE(FdEntry)
+
+ private:
+  Environment* env_;
+  std::shared_ptr<BufferValue> path_;
+  uv_stat_t stat_;
+  uint64_t start_ = 0;
+  uint64_t end_ = 0;
+
+  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;
+  }
+
+  static bool CheckModified(FdEntry* entry, int fd) {
+    uv_fs_t req;
+    auto cleanup = OnScopeLeave([&] { uv_fs_req_cleanup(&req); });
+    // TODO(jasnell): Note the use of a sync fs call here is a bit unfortunate.
+    // Doing this asynchronously creates a bit of a race condition tho, a file
+    // could be unmodified when we call the operation but then by the time the
+    // async callback is triggered to give us that answer the file is modified.
+    // While such silliness is still possible here, the sync call at least makes
+    // it less likely to hit the race.
+    if (uv_fs_fstat(nullptr, &req, fd, nullptr) < 0) return true;
+    return entry->is_modified(req.statbuf);
+  }
+
+  class ReaderImpl final : public DataQueue::Reader,
+                           public StreamListener,
+                           public std::enable_shared_from_this<ReaderImpl> {
+   public:
+    static std::shared_ptr<ReaderImpl> Create(FdEntry* entry) {
+      uv_fs_t req;
+      auto cleanup = OnScopeLeave([&] { uv_fs_req_cleanup(&req); });
+      int file =
+          uv_fs_open(nullptr, &req, entry->path_->out(), O_RDONLY, 0, nullptr);
+      if (file < 0 || FdEntry::CheckModified(entry, file)) {
+        uv_fs_close(nullptr, &req, file, nullptr);
+        return nullptr;
+      }
+      return std::make_shared<ReaderImpl>(
+          BaseObjectPtr<fs::FileHandle>(
+              fs::FileHandle::New(entry->env()->GetBindingData<fs::BindingData>(
+                                      entry->env()->context()),
+                                  file,
+                                  Local<Object>(),
+                                  entry->start_,
+                                  entry->end_)),
+          entry);
+    }
+
+    explicit ReaderImpl(BaseObjectPtr<fs::FileHandle> handle, FdEntry* entry)
+        : env_(handle->env()), handle_(std::move(handle)), entry_(entry) {
+      handle_->PushStreamListener(this);
+      handle_->env()->AddCleanupHook(cleanup, this);
+    }
+
+    ~ReaderImpl() override {
+      handle_->env()->RemoveCleanupHook(cleanup, this);
+      DrainAndClose();
+      handle_->RemoveStreamListener(this);
+    }
+
+    uv_buf_t OnStreamAlloc(size_t suggested_size) override {
+      return env_->allocate_managed_buffer(suggested_size);
+    }
+
+    void OnStreamRead(ssize_t nread, const uv_buf_t& buf) override {
+      std::shared_ptr<v8::BackingStore> store =
+          env_->release_managed_buffer(buf);
+
+      if (ended_) {
+        // If we got here and ended_ is true, it means we ended and drained
+        // while the read was pending. We're just going to do nothing.
+        CHECK(pending_pulls_.empty());
+        return;
+      }
+
+      CHECK(reading_);
+      auto pending = DequeuePendingPull();
+
+      if (CheckModified(entry_, handle_->GetFD())) {
+        DrainAndClose();
+        // The file was modified while the read was pending. We need to error.
+        std::move(pending.next)(UV_EINVAL, nullptr, 0, [](uint64_t) {});
+        return;
+      }
+
+      if (nread < 0) {
+        if (nread == UV_EOF) {
+          std::move(pending.next)(bob::STATUS_EOS, nullptr, 0, [](uint64_t) {});
+        } else {
+          std::move(pending.next)(nread, nullptr, 0, [](uint64_t) {});
+        }
+
+        return DrainAndClose();
+      }
+
+      DataQueue::Vec vec;
+      vec.base = static_cast<uint8_t*>(store->Data());
+      vec.len = static_cast<uint64_t>(nread);
+      std::move(pending.next)(
+          bob::STATUS_CONTINUE, &vec, 1, [store](uint64_t) {});
+
+      if (pending_pulls_.empty()) {
+        reading_ = false;
+        if (handle_->IsAlive()) handle_->ReadStop();
+      }
+    }
+
+    int Pull(Next next,
+             int options,
+             DataQueue::Vec* data,
+             size_t count,
+             size_t max_count_hint = bob::kMaxCountHint) override {
+      if (ended_ || !handle_->IsAlive()) {
+        std::move(next)(bob::STATUS_EOS, nullptr, 0, [](uint64_t) {});
+        return bob::STATUS_EOS;
+      }
+
+      if (FdEntry::CheckModified(entry_, handle_->GetFD())) {
+        DrainAndClose();
+        std::move(next)(UV_EINVAL, nullptr, 0, [](uint64_t) {});
+        return UV_EINVAL;
+      }
+
+      pending_pulls_.emplace_back(std::move(next), shared_from_this());
+      if (!reading_) {
+        reading_ = true;
+        handle_->ReadStart();
+      }
+      return bob::STATUS_WAIT;
+    }
+
+    SET_NO_MEMORY_INFO()
+    SET_MEMORY_INFO_NAME(FdEntry::Reader)
+    SET_SELF_SIZE(ReaderImpl)
+
+   private:
+    struct PendingPull {
+      Next next;
+      std::shared_ptr<ReaderImpl> self;
+      PendingPull(Next next, std::shared_ptr<ReaderImpl> self)
+          : next(std::move(next)), self(std::move(self)) {}
+    };
+
+    Environment* env_;
+    BaseObjectPtr<fs::FileHandle> handle_;
+    FdEntry* entry_;
+    std::deque<PendingPull> pending_pulls_;
+    bool reading_ = false;
+    bool ended_ = false;
+
+    static void cleanup(void* self) {
+      auto ptr = static_cast<ReaderImpl*>(self);
+      ptr->DrainAndClose();
+    }
+
+    void DrainAndClose() {
+      if (ended_) return;
+      ended_ = true;
+      while (!pending_pulls_.empty()) {
+        auto pending = DequeuePendingPull();
+        std::move(pending.next)(bob::STATUS_EOS, nullptr, 0, [](uint64_t) {});
+      }
+      handle_->ReadStop();
+
+      // We fallback to a sync close on the raw fd here because it is the
+      // easiest, simplest thing to do. All of FileHandle's close mechanisms
+      // assume async close and cleanup, while DrainAndClose might be running
+      // in the destructor during GC, for instance. As a todo, FileHandle could
+      // provide a sync mechanism for closing the FD but, for now, this
+      // approach works.
+      int fd = handle_->Release();
+      uv_fs_t req;
+      uv_fs_close(nullptr, &req, fd, nullptr);
+      uv_fs_req_cleanup(&req);
+    }
+
+    PendingPull DequeuePendingPull() {
+      CHECK(!pending_pulls_.empty());
+      auto pop = OnScopeLeave([this] { pending_pulls_.pop_front(); });
+      return std::move(pending_pulls_.front());
+    }
+
+    friend class FdEntry;
+  };
+
+  friend class ReaderImpl;
+};
+
+// ============================================================================
+
+}  // namespace
+
+std::shared_ptr<DataQueue> DataQueue::CreateIdempotent(
+    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;
+  const auto isInvalid = [&size](auto& item) {
+    if (item == nullptr || !item->is_idempotent()) {
+      return true;  // true means the entry is not valid here.
+    }
+
+    // To keep from having to iterate over the entries
+    // again, we'll try calculating the size. If any
+    // of the entries are unable to provide a size, then
+    // we assume we cannot safely treat this entry as
+    // idempotent even if it claims to be.
+    if (item->size().has_value()) {
+      size += item->size().value();
+    } else {
+      return true;  // true means the entry is not valid here.
+    }
+
+    return false;
+  };
+
+  if (std::any_of(list.begin(), list.end(), isInvalid)) {
+    return nullptr;
+  }
+
+  return std::make_shared<DataQueueImpl>(std::move(list), size);
+}
+
+std::shared_ptr<DataQueue> DataQueue::Create(std::optional<uint64_t> capped) {
+  return std::make_shared<DataQueueImpl>(capped);
+}
+
+std::unique_ptr<DataQueue::Entry> DataQueue::CreateInMemoryEntryFromView(
+    Local<ArrayBufferView> view) {
+  // If the view is not detachable, we do not want to create an InMemoryEntry
+  // from it. Why? Because if we're not able to detach the backing store from
+  // the underlying buffer, something else could modify the buffer while we're
+  // holding the reference, which means we cannot guarantee that reads will be
+  // idempotent.
+  if (!view->Buffer()->IsDetachable()) {
+    return nullptr;
+  }
+  auto store = view->Buffer()->GetBackingStore();
+  auto offset = view->ByteOffset();
+  auto length = view->ByteLength();
+  USE(view->Buffer()->Detach(Local<Value>()));
+  return CreateInMemoryEntryFromBackingStore(std::move(store), offset, length);
+}
+
+std::unique_ptr<DataQueue::Entry>
+DataQueue::CreateInMemoryEntryFromBackingStore(
+    std::shared_ptr<BackingStore> store, uint64_t offset, uint64_t length) {
+  CHECK(store);
+  if (offset + length > store->ByteLength()) {
+    return nullptr;
+  }
+  return std::make_unique<InMemoryEntry>(std::move(store), offset, length);
+}
+
+std::unique_ptr<DataQueue::Entry> DataQueue::CreateDataQueueEntry(
+    std::shared_ptr<DataQueue> data_queue) {
+  return std::make_unique<DataQueueEntry>(std::move(data_queue));
+}
+
+std::unique_ptr<DataQueue::Entry> DataQueue::CreateFdEntry(Environment* env,
+                                                           Local<Value> path) {
+  return FdEntry::Create(env, path);
+}
+
+void DataQueue::Initialize(Environment* env, v8::Local<v8::Object> target) {
+  // Nothing to do here currently.
+}
+
+void DataQueue::RegisterExternalReferences(
+    ExternalReferenceRegistry* registry) {
+  // Nothing to do here currently.
+}
+
+}  // namespace node
diff --git a/src/dataqueue/queue.h b/src/dataqueue/queue.h
new file mode 100644
index 00000000000000..a1a297a8fc2911
--- /dev/null
+++ b/src/dataqueue/queue.h
@@ -0,0 +1,294 @@
+#pragma once
+
+#if defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS
+
+#include <base_object.h>
+#include <memory_tracker.h>
+#include <node.h>
+#include <node_bob.h>
+#include <node_file.h>
+#include <stream_base.h>
+#include <uv.h>
+#include <v8.h>
+
+#include <memory>
+#include <optional>
+#include <vector>
+
+namespace node {
+
+// Represents a sequenced collection of data sources that can be
+// consumed as a single logical stream of data. Sources can be
+// memory-resident or streaming.
+//
+// There are two essential kinds of DataQueue:
+//
+// * Idempotent - Multiple reads always produce the same result.
+//                This is even the case if individual sources
+//                are not memory-resident. Reads never change
+//                the state of the DataQueue. Every entry in
+//                an Idempotent DataQueue must also be idempotent.
+//
+// * Non-idempotent - Reads are destructive of the internal state.
+//                    A non-idempotent DataQueue can be read at
+//                    most once and only by a single reader.
+//                    Entries in a non-idempotent DataQueue can
+//                    be a mix of idempotent and non-idempotent
+//                    entries.
+//
+// The DataQueue is essentially a collection of DataQueue::Entry
+// instances. A DataQueue::Entry is a single logical source of
+// data. The data may be memory-resident or streaming. The entry
+// can be idempotent or non-idempotent. An entry cannot be read
+// by itself, it must be part of a DataQueue to be consumed.
+//
+// Example of creating an idempotent DataQueue:
+//
+//   std::shared_ptr<v8::BackingStore> store1 = getBackingStoreSomehow();
+//   std::shared_ptr<v8::BackingStore> store2 = getBackingStoreSomehow();
+//
+//   std::vector<std::unique_ptr<DataQueue::Entry>> list;
+//   list.push_back(DataQueue::CreateInMemoryEntryFromBackingStore(
+//       store1, 0, len1));
+//   list.push_back(DataQueue::CreateInMemoryEntryFromBackingStore(
+//       store2, 0, len2));
+//
+//   std::shared_ptr<DataQueue> data_queue =
+//       DataQueue::CreateIdempotent(std::move(list));
+//
+// Importantly, idempotent DataQueue's are immutable and all entries
+// must be provided when the DataQueue is constructed. Every entry
+// must be idempotent with known sizes. The entries may be memory
+// resident or streaming. Streaming entries must be capable of
+// being read multiple times.
+//
+// Because idempotent DataQueue's will always produce the same results
+// when read, they can be sliced. Slices yield a new DataQueue instance
+// that is a subset view over the original:
+//
+//   std::shared_ptr<DataQueue> slice = data_queue.slice(
+//       5, v8::Just(10UL));
+//
+// Example of creating a non-idempotent DataQueue:
+//
+//   std::shared_ptr<v8::BackingStore> store1 = getBackingStoreSomehow();
+//   std::shared_ptr<v8::BackingStore> store2 = getBackingStoreSomehow();
+//
+//   std::shared_ptr<DataQueue> data_queue = DataQueue::Create();
+//
+//   data_queue->append(DataQueue::CreateInMemoryEntryFromBackingStore(
+//       store1, 0, len1));
+//
+//   data_queue->append(DataQueue::CreateInMemoryEntryFromBackingStore(
+//       store2, 0, len2));
+//
+// These data-queues can have new entries appended to them. Entries can
+// be memory-resident or streaming. Streaming entries might not have
+// a known size. Entries may not be capable of being read multiple
+// times.
+//
+// A non-idempotent data queue will, by default, allow any amount of
+// entries to be appended to it. To limit the size of the DataQueue,
+// or the close the DataQueue (preventing new entries from being
+// appending), use the cap() method. The DataQueue can be capped
+// at a specific size or whatever size it currently it.
+//
+// It might not be possible for a non-idempotent DataQueue to provide
+// a size because it might not know how much data a streaming entry
+// will ultimately provide.
+//
+// Non-idempotent DataQueues cannot be sliced.
+//
+// To read from a DataQueue, we use the node::bob::Source API
+// (see src/node_bob.h).
+//
+//   std::shared_ptr<DataQueue::Reader> reader = data_queue->get_reader();
+//
+//   reader->Pull(
+//        [](int status, const DataQueue::Vec* vecs,
+//           uint64_t count, Done done) {
+//     // status is one of node::bob::Status
+//     // vecs is zero or more data buffers containing the read data
+//     // count is the number of vecs
+//     // done is a callback to be invoked when done processing the data
+//   }, options, nullptr, 0, 16);
+//
+// Keep calling Pull() until status is equal to node::bob::Status::STATUS_EOS.
+//
+// For idempotent DataQueues, any number of readers can be created and
+// pull concurrently from the same DataQueue. The DataQueue can be read
+// multiple times. Succesful reads should always produce the same result.
+// If, for whatever reason, the implementation cannot ensure that the
+// data read will remain the same, the read must fail with an error status.
+//
+// For non-idempotent DataQueues, only a single reader is ever allowed for
+// the DataQueue, and the data can only ever be read once.
+
+class DataQueue : public MemoryRetainer {
+ public:
+  struct Vec {
+    uint8_t* base;
+    uint64_t len;
+  };
+
+  // A DataQueue::Reader consumes the DataQueue. If the data queue is
+  // idempotent, multiple Readers can be attached to the DataQueue at
+  // any given time, all guaranteed to yield the same result when the
+  // data is read. Otherwise, only a single Reader can be attached.
+  class Reader : public MemoryRetainer, public bob::Source<Vec> {
+   public:
+    using Next = bob::Next<Vec>;
+    using Done = bob::Done;
+  };
+
+  // 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.
+  class Entry : public MemoryRetainer {
+   public:
+    // Returns a new Entry that is a view over this entries data
+    // from the start offset to the ending offset. If the end
+    // offset is omitted, the slice extends to the end of the
+    // data.
+    //
+    // Creating a slice is only possible if is_idempotent() returns
+    // true. This is because consuming either the original entry or
+    // the new entry would change the state of the other in non-
+    // deterministic ways. When is_idempotent() returns false, slice()
+    // must return a nulled unique_ptr.
+    //
+    // Creating a slice is also only possible if the size of the
+    // entry is known. If size() returns std::nullopt, slice()
+    // must return a nulled unique_ptr.
+    virtual std::unique_ptr<Entry> slice(
+        uint64_t start, std::optional<uint64_t> end = std::nullopt) = 0;
+
+    // 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
+    // a value. In such cases, size() must return v8::Nothing<uint64_t>.
+    //
+    // If the entry is idempotent, a size should always be available.
+    virtual std::optional<uint64_t> size() const = 0;
+
+    // When true, multiple reads on the object must produce the exact
+    // same data or the reads will fail. Some sources of entry data,
+    // such as streams, may not be capable of preserving idempotency
+    // and therefore must not claim to be. If an entry claims to be
+    // idempotent and cannot preserve that quality, subsequent reads
+    // must fail with an error when a variance is detected.
+    virtual bool is_idempotent() const = 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);
+
+  // 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);
+
+  // Creates an idempotent Entry from a v8::ArrayBufferView. To help
+  // ensure idempotency, the underlying ArrayBuffer is detached from
+  // the BackingStore. It is the callers responsibility to ensure that
+  // the BackingStore is not otherwise modified through any other
+  // means. If the ArrayBuffer is not detachable, nullptr will be
+  // returned.
+  static std::unique_ptr<Entry> CreateInMemoryEntryFromView(
+      v8::Local<v8::ArrayBufferView> view);
+
+  // Creates an idempotent Entry from a v8::BackingStore. It is the
+  // callers responsibility to ensure that the BackingStore is not
+  // otherwise modified through any other means. If the ArrayBuffer
+  // is not detachable, nullptr will be returned.
+  static std::unique_ptr<Entry> CreateInMemoryEntryFromBackingStore(
+      std::shared_ptr<v8::BackingStore> store,
+      uint64_t offset,
+      uint64_t length);
+
+  static std::unique_ptr<Entry> CreateDataQueueEntry(
+      std::shared_ptr<DataQueue> data_queue);
+
+  static std::unique_ptr<Entry> CreateFdEntry(Environment* env,
+                                              v8::Local<v8::Value> path);
+
+  // Creates a Reader for the given queue. If the queue is idempotent,
+  // any number of readers can be created, all of which are guaranteed
+  // to provide the same data. Otherwise, only a single reader is
+  // permitted.
+  virtual std::shared_ptr<Reader> get_reader() = 0;
+
+  // Append a single new entry to the queue. Appending is only allowed
+  // when is_idempotent() is false. std::nullopt will be returned
+  // if is_idempotent() is true. std::optional(false) will be returned if the
+  // data queue is not idempotent but the entry otherwise cannot be added.
+  virtual std::optional<bool> append(std::unique_ptr<Entry> entry) = 0;
+
+  // Caps the size of this DataQueue preventing additional entries to
+  // be added if those cause the size to extend beyond the specified
+  // limit.
+  //
+  // If limit is zero, or is less than the known current size of the
+  // data queue, the limit is set to the current known size, meaning
+  // that no additional entries can be added at all.
+  //
+  // If the size of the data queue is not known, the limit will be
+  // ignored and no additional entries will be allowed at all.
+  //
+  // If is_idempotent is true capping is unnecessary because the data
+  // queue cannot be appended to. In that case, cap() is a non-op.
+  //
+  // If the data queue has already been capped, cap can be called
+  // again with a smaller size.
+  virtual void cap(uint64_t limit = 0) = 0;
+
+  // Returns a new DataQueue that is a view over this queues data
+  // from the start offset to the ending offset. If the end offset
+  // is omitted, the slice extends to the end of the data.
+  //
+  // The slice will coverage a range from start up to, but excluding, end.
+  //
+  // Creating a slice is only possible if is_idempotent() returns
+  // true. This is because consuming either the original DataQueue or
+  // the new queue would change the state of the other in non-
+  // deterministic ways. When is_idempotent() returns false, slice()
+  // must return a nulled unique_ptr.
+  //
+  // Creating a slice is also only possible if the size of the
+  // DataQueue is known. If size() returns std::nullopt, slice()
+  // must return a null unique_ptr.
+  virtual std::shared_ptr<DataQueue> slice(
+      uint64_t start, std::optional<uint64_t> end = std::nullopt) = 0;
+
+  // The size of DataQueue is the total size of all of its member entries.
+  // If any of the entries is not able to specify a size, the DataQueue
+  // will also be incapable of doing so, in which case size() must return
+  // std::nullopt.
+  virtual std::optional<uint64_t> size() const = 0;
+
+  // A DataQueue is idempotent only if all of its member entries are
+  // idempotent.
+  virtual bool is_idempotent() const = 0;
+
+  // True only if cap is called or the data queue is a limited to a
+  // fixed size.
+  virtual bool is_capped() const = 0;
+
+  // If the data queue has been capped, and the size of the data queue
+  // is known, maybeCapRemaining will return the number of additional
+  // bytes the data queue can receive before reaching the cap limit.
+  // If the size of the queue cannot be known, or the cap has not
+  // been set, maybeCapRemaining() will return std::nullopt.
+  virtual std::optional<uint64_t> maybeCapRemaining() const = 0;
+
+  static void Initialize(Environment* env, v8::Local<v8::Object> target);
+  static void RegisterExternalReferences(ExternalReferenceRegistry* registry);
+};
+
+}  // namespace node
+
+#endif  // defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS
diff --git a/src/env_properties.h b/src/env_properties.h
index c4c30eb5e416ad..fcae9d8887483f 100644
--- a/src/env_properties.h
+++ b/src/env_properties.h
@@ -329,6 +329,7 @@
   V(base_object_ctor_template, v8::FunctionTemplate)                           \
   V(binding_data_ctor_template, v8::FunctionTemplate)                          \
   V(blob_constructor_template, v8::FunctionTemplate)                           \
+  V(blob_reader_constructor_template, v8::FunctionTemplate)                    \
   V(blocklist_constructor_template, v8::FunctionTemplate)                      \
   V(contextify_global_template, v8::ObjectTemplate)                            \
   V(contextify_wrapper_template, v8::ObjectTemplate)                           \
@@ -339,6 +340,7 @@
   V(dir_instance_template, v8::ObjectTemplate)                                 \
   V(fd_constructor_template, v8::ObjectTemplate)                               \
   V(fdclose_constructor_template, v8::ObjectTemplate)                          \
+  V(fdentry_constructor_template, v8::FunctionTemplate)                        \
   V(filehandlereadwrap_template, v8::ObjectTemplate)                           \
   V(fsreqpromise_constructor_template, v8::ObjectTemplate)                     \
   V(handle_wrap_ctor_template, v8::FunctionTemplate)                           \
@@ -358,7 +360,10 @@
   V(secure_context_constructor_template, v8::FunctionTemplate)                 \
   V(shutdown_wrap_template, v8::ObjectTemplate)                                \
   V(socketaddress_constructor_template, v8::FunctionTemplate)                  \
+  V(streambaseentry_ctor_template, v8::FunctionTemplate)                       \
   V(streambaseoutputstream_constructor_template, v8::ObjectTemplate)           \
+  V(streamentry_ctor_template, v8::FunctionTemplate)                           \
+  V(streamentry_opaque_ctor_template, v8::FunctionTemplate)                    \
   V(qlogoutputstream_constructor_template, v8::ObjectTemplate)                 \
   V(tcp_constructor_template, v8::FunctionTemplate)                            \
   V(tty_constructor_template, v8::FunctionTemplate)                            \
diff --git a/src/node_blob.cc b/src/node_blob.cc
index f35a44c6b7958d..884e9d61c9b95c 100644
--- a/src/node_blob.cc
+++ b/src/node_blob.cc
@@ -3,9 +3,10 @@
 #include "base_object-inl.h"
 #include "env-inl.h"
 #include "memory_tracker-inl.h"
+#include "node_bob-inl.h"
 #include "node_errors.h"
 #include "node_external_reference.h"
-#include "threadpoolwork-inl.h"
+#include "node_file.h"
 #include "v8.h"
 
 #include <algorithm>
@@ -17,21 +18,93 @@ using v8::ArrayBuffer;
 using v8::ArrayBufferView;
 using v8::BackingStore;
 using v8::Context;
-using v8::EscapableHandleScope;
 using v8::Function;
 using v8::FunctionCallbackInfo;
 using v8::FunctionTemplate;
+using v8::Global;
 using v8::HandleScope;
+using v8::Int32;
 using v8::Isolate;
 using v8::Local;
-using v8::MaybeLocal;
-using v8::Number;
 using v8::Object;
 using v8::String;
 using v8::Uint32;
 using v8::Undefined;
 using v8::Value;
 
+namespace {
+
+// Concatenate multiple ArrayBufferView/ArrayBuffers into a single ArrayBuffer.
+// This method treats all ArrayBufferView types the same.
+void Concat(const FunctionCallbackInfo<Value>& args) {
+  Environment* env = Environment::GetCurrent(args);
+  CHECK(args[0]->IsArray());
+  Local<Array> array = args[0].As<Array>();
+
+  struct View {
+    std::shared_ptr<BackingStore> store;
+    size_t length;
+    size_t offset = 0;
+  };
+
+  std::vector<View> views;
+  size_t total = 0;
+
+  for (uint32_t n = 0; n < array->Length(); n++) {
+    Local<Value> val;
+    if (!array->Get(env->context(), n).ToLocal(&val)) return;
+    if (val->IsArrayBuffer()) {
+      auto ab = val.As<ArrayBuffer>();
+      views.push_back(View{ab->GetBackingStore(), ab->ByteLength(), 0});
+      total += ab->ByteLength();
+    } else {
+      CHECK(val->IsArrayBufferView());
+      auto view = val.As<ArrayBufferView>();
+      views.push_back(View{view->Buffer()->GetBackingStore(),
+                           view->ByteLength(),
+                           view->ByteOffset()});
+      total += view->ByteLength();
+    }
+  }
+
+  std::shared_ptr<BackingStore> store =
+      ArrayBuffer::NewBackingStore(env->isolate(), total);
+  uint8_t* ptr = static_cast<uint8_t*>(store->Data());
+  for (size_t n = 0; n < views.size(); n++) {
+    uint8_t* from =
+        static_cast<uint8_t*>(views[n].store->Data()) + views[n].offset;
+    std::copy(from, from + views[n].length, ptr);
+    ptr += views[n].length;
+  }
+
+  args.GetReturnValue().Set(ArrayBuffer::New(env->isolate(), std::move(store)));
+}
+
+void BlobFromFilePath(const FunctionCallbackInfo<Value>& args) {
+  Environment* env = Environment::GetCurrent(args);
+  auto entry = DataQueue::CreateFdEntry(env, args[0]);
+  if (entry == nullptr) {
+    return THROW_ERR_INVALID_ARG_VALUE(env, "Unabled to open file as blob");
+  }
+
+  std::vector<std::unique_ptr<DataQueue::Entry>> entries;
+  entries.push_back(std::move(entry));
+
+  auto blob =
+      Blob::Create(env, DataQueue::CreateIdempotent(std::move(entries)));
+
+  if (blob) {
+    auto array = Array::New(env->isolate(), 2);
+    USE(array->Set(env->context(), 0, blob->object()));
+    USE(array->Set(env->context(),
+                   1,
+                   Uint32::NewFromUnsigned(env->isolate(), blob->length())));
+
+    args.GetReturnValue().Set(array);
+  }
+}
+}  // namespace
+
 void Blob::Initialize(
     Local<Object> target,
     Local<Value> unused,
@@ -47,7 +120,8 @@ void Blob::Initialize(
   SetMethod(context, target, "storeDataObject", StoreDataObject);
   SetMethod(context, target, "getDataObject", GetDataObject);
   SetMethod(context, target, "revokeDataObject", RevokeDataObject);
-  FixedSizeBlobCopyJob::Initialize(env, target);
+  SetMethod(context, target, "concat", Concat);
+  SetMethod(context, target, "createBlobFromFilePath", BlobFromFilePath);
 }
 
 Local<FunctionTemplate> Blob::GetConstructorTemplate(Environment* env) {
@@ -60,7 +134,7 @@ Local<FunctionTemplate> Blob::GetConstructorTemplate(Environment* env) {
     tmpl->Inherit(BaseObject::GetConstructorTemplate(env));
     tmpl->SetClassName(
         FIXED_ONE_BYTE_STRING(env->isolate(), "Blob"));
-    SetProtoMethod(isolate, tmpl, "toArrayBuffer", ToArrayBuffer);
+    SetProtoMethod(isolate, tmpl, "getReader", GetReader);
     SetProtoMethod(isolate, tmpl, "slice", ToSlice);
     env->set_blob_constructor_template(tmpl);
   }
@@ -72,8 +146,7 @@ bool Blob::HasInstance(Environment* env, v8::Local<v8::Value> object) {
 }
 
 BaseObjectPtr<Blob> Blob::Create(Environment* env,
-                                 const std::vector<BlobEntry>& store,
-                                 size_t length) {
+                                 std::shared_ptr<DataQueue> data_queue) {
   HandleScope scope(env->isolate());
 
   Local<Function> ctor;
@@ -84,56 +157,79 @@ BaseObjectPtr<Blob> Blob::Create(Environment* env,
   if (!ctor->NewInstance(env->context()).ToLocal(&obj))
     return BaseObjectPtr<Blob>();
 
-  return MakeBaseObject<Blob>(env, obj, store, length);
+  return MakeBaseObject<Blob>(env, obj, data_queue);
 }
 
 void Blob::New(const FunctionCallbackInfo<Value>& args) {
   Environment* env = Environment::GetCurrent(args);
   CHECK(args[0]->IsArray());  // sources
-  CHECK(args[1]->IsUint32());  // length
 
-  std::vector<BlobEntry> entries;
+  Local<Array> array = args[0].As<Array>();
+  std::vector<std::unique_ptr<DataQueue::Entry>> entries(array->Length());
 
-  size_t length = args[1].As<Uint32>()->Value();
-  size_t len = 0;
-  Local<Array> ary = args[0].As<Array>();
-  for (size_t n = 0; n < ary->Length(); n++) {
+  for (size_t i = 0; i < array->Length(); i++) {
     Local<Value> entry;
-    if (!ary->Get(env->context(), n).ToLocal(&entry))
+    if (!array->Get(env->context(), i).ToLocal(&entry)) {
       return;
-    CHECK(entry->IsArrayBufferView() || Blob::HasInstance(env, entry));
-    if (entry->IsArrayBufferView()) {
+    }
+
+    const auto entryFromArrayBuffer = [env](v8::Local<v8::ArrayBuffer> buf,
+                                            size_t byte_length,
+                                            size_t byte_offset = 0) {
+      if (buf->IsDetachable()) {
+        std::shared_ptr<BackingStore> store = buf->GetBackingStore();
+        USE(buf->Detach(Local<Value>()));
+        return DataQueue::CreateInMemoryEntryFromBackingStore(
+            store, byte_offset, byte_length);
+      }
+
+      // If the ArrayBuffer is not detachable, we will copy from it instead.
+      std::shared_ptr<BackingStore> store =
+          ArrayBuffer::NewBackingStore(env->isolate(), byte_length);
+      uint8_t* ptr = static_cast<uint8_t*>(buf->Data()) + byte_offset;
+      std::copy(ptr, ptr + byte_length, static_cast<uint8_t*>(store->Data()));
+      return DataQueue::CreateInMemoryEntryFromBackingStore(
+          store, 0, byte_length);
+    };
+
+    // Every entry should be either an ArrayBuffer, ArrayBufferView, or Blob.
+    // If the input to the Blob constructor in JavaScript was a string, then
+    // it will be decoded into an ArrayBufferView there before being passed
+    // in.
+    //
+    // Importantly, here we also assume that the ArrayBuffer/ArrayBufferView
+    // is not going to be modified here so we will detach them. It is up to
+    // the JavaScript side to do the right thing with regards to copying and
+    // ensuring appropriate spec compliance.
+    if (entry->IsArrayBuffer()) {
+      Local<ArrayBuffer> buf = entry.As<ArrayBuffer>();
+      entries[i] = entryFromArrayBuffer(buf, buf->ByteLength());
+    } else if (entry->IsArrayBufferView()) {
       Local<ArrayBufferView> view = entry.As<ArrayBufferView>();
-      CHECK_EQ(view->ByteOffset(), 0);
-      std::shared_ptr<BackingStore> store = view->Buffer()->GetBackingStore();
-      size_t byte_length = view->ByteLength();
-      view->Buffer()
-          ->Detach(Local<Value>())
-          .Check();  // The Blob will own the backing store now.
-      entries.emplace_back(BlobEntry{std::move(store), byte_length, 0});
-      len += byte_length;
-    } else {
+      entries[i] = entryFromArrayBuffer(
+          view->Buffer(), view->ByteLength(), view->ByteOffset());
+    } else if (Blob::HasInstance(env, entry)) {
       Blob* blob;
       ASSIGN_OR_RETURN_UNWRAP(&blob, entry);
-      auto source = blob->entries();
-      entries.insert(entries.end(), source.begin(), source.end());
-      len += blob->length();
+      entries[i] = DataQueue::CreateDataQueueEntry(blob->data_queue_);
+    } else {
+      UNREACHABLE("Incorrect Blob initialization type");
     }
   }
-  CHECK_EQ(length, len);
 
-  BaseObjectPtr<Blob> blob = Create(env, entries, length);
+  auto blob = Create(env, DataQueue::CreateIdempotent(std::move(entries)));
   if (blob)
     args.GetReturnValue().Set(blob->object());
 }
 
-void Blob::ToArrayBuffer(const FunctionCallbackInfo<Value>& args) {
+void Blob::GetReader(const FunctionCallbackInfo<Value>& args) {
   Environment* env = Environment::GetCurrent(args);
   Blob* blob;
   ASSIGN_OR_RETURN_UNWRAP(&blob, args.Holder());
-  Local<Value> ret;
-  if (blob->GetArrayBuffer(env).ToLocal(&ret))
-    args.GetReturnValue().Set(ret);
+
+  BaseObjectPtr<Blob::Reader> reader =
+      Blob::Reader::Create(env, BaseObjectPtr<Blob>(blob));
+  if (reader) args.GetReturnValue().Set(reader->object());
 }
 
 void Blob::ToSlice(const FunctionCallbackInfo<Value>& args) {
@@ -150,70 +246,130 @@ void Blob::ToSlice(const FunctionCallbackInfo<Value>& args) {
 }
 
 void Blob::MemoryInfo(MemoryTracker* tracker) const {
-  tracker->TrackFieldWithSize("store", length_);
+  tracker->TrackField("data_queue_", data_queue_, "std::shared_ptr<DataQueue>");
 }
 
-MaybeLocal<Value> Blob::GetArrayBuffer(Environment* env) {
-  EscapableHandleScope scope(env->isolate());
-  size_t len = length();
-  std::shared_ptr<BackingStore> store =
-      ArrayBuffer::NewBackingStore(env->isolate(), len);
-  if (len > 0) {
-    unsigned char* dest = static_cast<unsigned char*>(store->Data());
-    size_t total = 0;
-    for (const auto& entry : entries()) {
-      unsigned char* src = static_cast<unsigned char*>(entry.store->Data());
-      src += entry.offset;
-      memcpy(dest, src, entry.length);
-      dest += entry.length;
-      total += entry.length;
-      CHECK_LE(total, len);
-    }
-  }
+BaseObjectPtr<Blob> Blob::Slice(Environment* env, size_t start, size_t end) {
+  return Create(env,
+                this->data_queue_->slice(start, static_cast<uint64_t>(end)));
+}
 
-  return scope.Escape(ArrayBuffer::New(env->isolate(), store));
+Blob::Blob(Environment* env,
+           v8::Local<v8::Object> obj,
+           std::shared_ptr<DataQueue> data_queue)
+    : BaseObject(env, obj), data_queue_(data_queue) {
+  MakeWeak();
 }
 
-BaseObjectPtr<Blob> Blob::Slice(Environment* env, size_t start, size_t end) {
-  CHECK_LE(start, length());
-  CHECK_LE(end, length());
-  CHECK_LE(start, end);
+Blob::Reader::Reader(Environment* env,
+                     v8::Local<v8::Object> obj,
+                     BaseObjectPtr<Blob> strong_ptr)
+    : AsyncWrap(env, obj, AsyncWrap::PROVIDER_BLOBREADER),
+      inner_(strong_ptr->data_queue_->get_reader()),
+      strong_ptr_(std::move(strong_ptr)) {
+  MakeWeak();
+}
+
+bool Blob::Reader::HasInstance(Environment* env, v8::Local<v8::Value> value) {
+  return GetConstructorTemplate(env)->HasInstance(value);
+}
 
-  std::vector<BlobEntry> slices;
-  size_t total = end - start;
-  size_t remaining = total;
+Local<FunctionTemplate> Blob::Reader::GetConstructorTemplate(Environment* env) {
+  Local<FunctionTemplate> tmpl = env->blob_reader_constructor_template();
+  if (tmpl.IsEmpty()) {
+    Isolate* isolate = env->isolate();
+    tmpl = NewFunctionTemplate(isolate, nullptr);
+    tmpl->InstanceTemplate()->SetInternalFieldCount(
+        BaseObject::kInternalFieldCount);
+    tmpl->Inherit(BaseObject::GetConstructorTemplate(env));
+    tmpl->SetClassName(FIXED_ONE_BYTE_STRING(env->isolate(), "BlobReader"));
+    SetProtoMethod(env->isolate(), tmpl, "pull", Pull);
+    env->set_blob_reader_constructor_template(tmpl);
+  }
+  return tmpl;
+}
 
-  if (total == 0) return Create(env, slices, 0);
+BaseObjectPtr<Blob::Reader> Blob::Reader::Create(Environment* env,
+                                                 BaseObjectPtr<Blob> blob) {
+  Local<Object> obj;
+  if (!GetConstructorTemplate(env)
+           ->InstanceTemplate()
+           ->NewInstance(env->context())
+           .ToLocal(&obj)) {
+    return BaseObjectPtr<Blob::Reader>();
+  }
 
-  for (const auto& entry : entries()) {
-    if (start + entry.offset > entry.store->ByteLength()) {
-      start -= entry.length;
-      continue;
-    }
+  return MakeBaseObject<Blob::Reader>(env, obj, std::move(blob));
+}
 
-    size_t offset = entry.offset + start;
-    size_t len = std::min(remaining, entry.store->ByteLength() - offset);
-    slices.emplace_back(BlobEntry{entry.store, len, offset});
+void Blob::Reader::Pull(const FunctionCallbackInfo<Value>& args) {
+  Environment* env = Environment::GetCurrent(args);
+  Blob::Reader* reader;
+  ASSIGN_OR_RETURN_UNWRAP(&reader, args.Holder());
 
-    remaining -= len;
-    start = 0;
+  CHECK(args[0]->IsFunction());
+  Local<Function> fn = args[0].As<Function>();
+  CHECK(!fn->IsConstructor());
 
-    if (remaining == 0)
-      break;
+  if (reader->eos_) {
+    Local<Value> arg = Int32::New(env->isolate(), bob::STATUS_EOS);
+    reader->MakeCallback(fn, 1, &arg);
+    return args.GetReturnValue().Set(bob::STATUS_EOS);
   }
 
-  return Create(env, slices, total);
-}
+  struct Impl {
+    BaseObjectPtr<Blob::Reader> reader;
+    Global<Function> callback;
+    Environment* env;
+  };
+  // 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);
+  impl->callback.Reset(env->isolate(), fn);
+  impl->env = env;
+
+  auto next = [impl](int status,
+                     const DataQueue::Vec* vecs,
+                     size_t count,
+                     bob::Done doneCb) mutable {
+    auto dropMe = std::unique_ptr<Impl>(impl);
+    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 (count > 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;
+
+      std::shared_ptr<BackingStore> store =
+          v8::ArrayBuffer::NewBackingStore(env->isolate(), total);
+      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;
+      }
+      // 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),
+                              ArrayBuffer::New(env->isolate(), store)};
+      impl->reader->MakeCallback(fn, arraysize(argv), argv);
+      return;
+    }
 
-Blob::Blob(
-    Environment* env,
-    v8::Local<v8::Object> obj,
-    const std::vector<BlobEntry>& store,
-    size_t length)
-    : BaseObject(env, obj),
-      store_(store),
-      length_(length) {
-  MakeWeak();
+    Local<Value> argv[2] = {
+        Int32::New(env->isolate(), status),
+        Undefined(env->isolate()),
+    };
+    impl->reader->MakeCallback(fn, arraysize(argv), argv);
+  };
+
+  args.GetReturnValue().Set(reader->inner_->Pull(
+      std::move(next), node::bob::OPTIONS_END, nullptr, 0));
 }
 
 BaseObjectPtr<BaseObject>
@@ -225,7 +381,7 @@ Blob::BlobTransferData::Deserialize(
     THROW_ERR_MESSAGE_TARGET_CONTEXT_UNAVAILABLE(env);
     return {};
   }
-  return Blob::Create(env, store_, length_);
+  return Blob::Create(env, data_queue);
 }
 
 BaseObject::TransferMode Blob::GetTransferMode() const {
@@ -233,7 +389,7 @@ BaseObject::TransferMode Blob::GetTransferMode() const {
 }
 
 std::unique_ptr<worker::TransferData> Blob::CloneForMessaging() const {
-  return std::make_unique<BlobTransferData>(store_, length_);
+  return std::make_unique<BlobTransferData>(data_queue_);
 }
 
 void Blob::StoreDataObject(const v8::FunctionCallbackInfo<v8::Value>& args) {
@@ -308,118 +464,9 @@ void Blob::GetDataObject(const v8::FunctionCallbackInfo<v8::Value>& args) {
   }
 }
 
-FixedSizeBlobCopyJob::FixedSizeBlobCopyJob(Environment* env,
-                                           Local<Object> object,
-                                           Blob* blob,
-                                           FixedSizeBlobCopyJob::Mode mode)
-    : AsyncWrap(env, object, AsyncWrap::PROVIDER_FIXEDSIZEBLOBCOPY),
-      ThreadPoolWork(env, "blob"),
-      mode_(mode) {
-  if (mode == FixedSizeBlobCopyJob::Mode::SYNC) MakeWeak();
-  source_ = blob->entries();
-  length_ = blob->length();
-}
-
-void FixedSizeBlobCopyJob::AfterThreadPoolWork(int status) {
-  Environment* env = AsyncWrap::env();
-  CHECK_EQ(mode_, Mode::ASYNC);
-  CHECK(status == 0 || status == UV_ECANCELED);
-  std::unique_ptr<FixedSizeBlobCopyJob> ptr(this);
-  HandleScope handle_scope(env->isolate());
-  Context::Scope context_scope(env->context());
-  Local<Value> args[2];
-
-  if (status == UV_ECANCELED) {
-    args[0] = Number::New(env->isolate(), status),
-    args[1] = Undefined(env->isolate());
-  } else {
-    args[0] = Undefined(env->isolate());
-    args[1] = ArrayBuffer::New(env->isolate(), destination_);
-  }
-
-  ptr->MakeCallback(env->ondone_string(), arraysize(args), args);
-}
-
-void FixedSizeBlobCopyJob::DoThreadPoolWork() {
-  unsigned char* dest = static_cast<unsigned char*>(destination_->Data());
-  if (length_ > 0) {
-    size_t total = 0;
-    for (const auto& entry : source_) {
-      unsigned char* src = static_cast<unsigned char*>(entry.store->Data());
-      src += entry.offset;
-      memcpy(dest, src, entry.length);
-      dest += entry.length;
-      total += entry.length;
-      CHECK_LE(total, length_);
-    }
-  }
-}
-
-void FixedSizeBlobCopyJob::MemoryInfo(MemoryTracker* tracker) const {
-  tracker->TrackFieldWithSize("source", length_);
-  tracker->TrackFieldWithSize(
-      "destination",
-      destination_ ? destination_->ByteLength() : 0);
-}
-
-void FixedSizeBlobCopyJob::Initialize(Environment* env, Local<Object> target) {
-  Isolate* isolate = env->isolate();
-  v8::Local<v8::FunctionTemplate> job = NewFunctionTemplate(isolate, New);
-  job->Inherit(AsyncWrap::GetConstructorTemplate(env));
-  job->InstanceTemplate()->SetInternalFieldCount(
-      AsyncWrap::kInternalFieldCount);
-  SetProtoMethod(isolate, job, "run", Run);
-  SetConstructorFunction(env->context(), target, "FixedSizeBlobCopyJob", job);
-}
-
-void FixedSizeBlobCopyJob::New(const FunctionCallbackInfo<Value>& args) {
-  static constexpr size_t kMaxSyncLength = 4096;
-  static constexpr size_t kMaxEntryCount = 4;
-
-  Environment* env = Environment::GetCurrent(args);
-  CHECK(args.IsConstructCall());
-  CHECK(args[0]->IsObject());
-  CHECK(Blob::HasInstance(env, args[0]));
-
-  Blob* blob;
-  ASSIGN_OR_RETURN_UNWRAP(&blob, args[0]);
-
-  // This is a fairly arbitrary heuristic. We want to avoid deferring to
-  // the threadpool if the amount of data being copied is small and there
-  // aren't that many entries to copy.
-  FixedSizeBlobCopyJob::Mode mode =
-      (blob->length() < kMaxSyncLength &&
-       blob->entries().size() < kMaxEntryCount) ?
-          FixedSizeBlobCopyJob::Mode::SYNC :
-          FixedSizeBlobCopyJob::Mode::ASYNC;
-
-  new FixedSizeBlobCopyJob(env, args.This(), blob, mode);
-}
-
-void FixedSizeBlobCopyJob::Run(const FunctionCallbackInfo<Value>& args) {
-  Environment* env = Environment::GetCurrent(args);
-  FixedSizeBlobCopyJob* job;
-  ASSIGN_OR_RETURN_UNWRAP(&job, args.Holder());
-  job->destination_ =
-      ArrayBuffer::NewBackingStore(env->isolate(), job->length_);
-  if (job->mode() == FixedSizeBlobCopyJob::Mode::ASYNC)
-    return job->ScheduleWork();
-
-  job->DoThreadPoolWork();
-  args.GetReturnValue().Set(
-      ArrayBuffer::New(env->isolate(), job->destination_));
-}
-
-void FixedSizeBlobCopyJob::RegisterExternalReferences(
-    ExternalReferenceRegistry* registry) {
-  registry->Register(New);
-  registry->Register(Run);
-}
-
 void BlobBindingData::StoredDataObject::MemoryInfo(
     MemoryTracker* tracker) const {
-  tracker->TrackField("blob", blob);
-  tracker->TrackFieldWithSize("type", type.length());
+  tracker->TrackField("blob", blob, "BaseObjectPtr<Blob>");
 }
 
 BlobBindingData::StoredDataObject::StoredDataObject(
@@ -436,7 +483,9 @@ BlobBindingData::BlobBindingData(Environment* env, Local<Object> wrap)
 }
 
 void BlobBindingData::MemoryInfo(MemoryTracker* tracker) const {
-  tracker->TrackField("data_objects", data_objects_);
+  tracker->TrackField("data_objects_",
+                      data_objects_,
+                      "std::unordered_map<std::string, StoredDataObject>");
 }
 
 void BlobBindingData::store_data_object(
@@ -490,13 +539,14 @@ InternalFieldInfoBase* BlobBindingData::Serialize(int index) {
 
 void Blob::RegisterExternalReferences(ExternalReferenceRegistry* registry) {
   registry->Register(Blob::New);
-  registry->Register(Blob::ToArrayBuffer);
+  registry->Register(Blob::GetReader);
   registry->Register(Blob::ToSlice);
   registry->Register(Blob::StoreDataObject);
   registry->Register(Blob::GetDataObject);
   registry->Register(Blob::RevokeDataObject);
-
-  FixedSizeBlobCopyJob::RegisterExternalReferences(registry);
+  registry->Register(Blob::Reader::Pull);
+  registry->Register(Concat);
+  registry->Register(BlobFromFilePath);
 }
 
 }  // namespace node
diff --git a/src/node_blob.h b/src/node_blob.h
index f0340f313bde6f..e56627cf35cf76 100644
--- a/src/node_blob.h
+++ b/src/node_blob.h
@@ -1,10 +1,13 @@
 #ifndef SRC_NODE_BLOB_H_
 #define SRC_NODE_BLOB_H_
 
+#include "v8-function-callback.h"
+#include "v8-template.h"
 #if defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS
 
 #include "async_wrap.h"
 #include "base_object.h"
+#include "dataqueue/queue.h"
 #include "env.h"
 #include "memory_tracker.h"
 #include "node_internals.h"
@@ -18,12 +21,6 @@
 
 namespace node {
 
-struct BlobEntry {
-  std::shared_ptr<v8::BackingStore> store;
-  size_t length;
-  size_t offset;
-};
-
 class Blob : public BaseObject {
  public:
   static void RegisterExternalReferences(
@@ -36,7 +33,7 @@ class Blob : public BaseObject {
       void* priv);
 
   static void New(const v8::FunctionCallbackInfo<v8::Value>& args);
-  static void ToArrayBuffer(const v8::FunctionCallbackInfo<v8::Value>& args);
+  static void GetReader(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void ToSlice(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void StoreDataObject(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void GetDataObject(const v8::FunctionCallbackInfo<v8::Value>& args);
@@ -46,31 +43,22 @@ class Blob : public BaseObject {
       Environment* env);
 
   static BaseObjectPtr<Blob> Create(Environment* env,
-                                    const std::vector<BlobEntry>& store,
-                                    size_t length);
+                                    std::shared_ptr<DataQueue> data_queue);
 
   static bool HasInstance(Environment* env, v8::Local<v8::Value> object);
 
-  const std::vector<BlobEntry>& entries() const { return store_; }
-
   void MemoryInfo(MemoryTracker* tracker) const override;
   SET_MEMORY_INFO_NAME(Blob)
   SET_SELF_SIZE(Blob)
 
-  // Copies the contents of the Blob into an ArrayBuffer.
-  v8::MaybeLocal<v8::Value> GetArrayBuffer(Environment* env);
-
   BaseObjectPtr<Blob> Slice(Environment* env, size_t start, size_t end);
 
-  inline size_t length() const { return length_; }
+  inline size_t length() const { return this->data_queue_->size().value(); }
 
   class BlobTransferData : public worker::TransferData {
    public:
-    explicit BlobTransferData(
-        const std::vector<BlobEntry>& store,
-      size_t length)
-        : store_(store),
-          length_(length) {}
+    explicit BlobTransferData(std::shared_ptr<DataQueue> data_queue)
+        : data_queue(data_queue) {}
 
     BaseObjectPtr<BaseObject> Deserialize(
         Environment* env,
@@ -82,61 +70,43 @@ class Blob : public BaseObject {
     SET_NO_MEMORY_INFO()
 
    private:
-    std::vector<BlobEntry> store_;
-    size_t length_ = 0;
+    std::shared_ptr<DataQueue> data_queue;
   };
 
-  BaseObject::TransferMode GetTransferMode() const override;
-  std::unique_ptr<worker::TransferData> CloneForMessaging() const override;
+  class Reader final : public AsyncWrap {
+   public:
+    static bool HasInstance(Environment* env, v8::Local<v8::Value> value);
+    static v8::Local<v8::FunctionTemplate> GetConstructorTemplate(
+        Environment* env);
+    static BaseObjectPtr<Reader> Create(Environment* env,
+                                        BaseObjectPtr<Blob> blob);
+    static void Pull(const v8::FunctionCallbackInfo<v8::Value>& args);
 
-  Blob(
-      Environment* env,
-      v8::Local<v8::Object> obj,
-      const std::vector<BlobEntry>& store,
-      size_t length);
+    explicit Reader(Environment* env,
+                    v8::Local<v8::Object> obj,
+                    BaseObjectPtr<Blob> strong_ptr);
 
- private:
-  std::vector<BlobEntry> store_;
-  size_t length_ = 0;
-};
+    SET_NO_MEMORY_INFO()
+    SET_MEMORY_INFO_NAME(Blob::Reader)
+    SET_SELF_SIZE(Reader)
 
-class FixedSizeBlobCopyJob : public AsyncWrap, public ThreadPoolWork {
- public:
-  enum class Mode {
-    SYNC,
-    ASYNC
+   private:
+    std::shared_ptr<DataQueue::Reader> inner_;
+    BaseObjectPtr<Blob> strong_ptr_;
+    bool eos_ = false;
   };
 
-  static void RegisterExternalReferences(
-      ExternalReferenceRegistry* registry);
-  static void Initialize(Environment* env, v8::Local<v8::Object> target);
-  static void New(const v8::FunctionCallbackInfo<v8::Value>& args);
-  static void Run(const v8::FunctionCallbackInfo<v8::Value>& args);
-
-  bool IsNotIndicativeOfMemoryLeakAtExit() const override {
-    return true;
-  }
-
-  void DoThreadPoolWork() override;
-  void AfterThreadPoolWork(int status) override;
+  BaseObject::TransferMode GetTransferMode() const override;
+  std::unique_ptr<worker::TransferData> CloneForMessaging() const override;
 
-  Mode mode() const { return mode_; }
+  Blob(Environment* env,
+       v8::Local<v8::Object> obj,
+       std::shared_ptr<DataQueue> data_queue);
 
-  void MemoryInfo(MemoryTracker* tracker) const override;
-  SET_MEMORY_INFO_NAME(FixedSizeBlobCopyJob)
-  SET_SELF_SIZE(FixedSizeBlobCopyJob)
+  DataQueue& getDataQueue() const { return *data_queue_; }
 
  private:
-  FixedSizeBlobCopyJob(
-    Environment* env,
-    v8::Local<v8::Object> object,
-    Blob* blob,
-    Mode mode = Mode::ASYNC);
-
-  Mode mode_;
-  std::vector<BlobEntry> source_;
-  std::shared_ptr<v8::BackingStore> destination_;
-  size_t length_ = 0;
+  std::shared_ptr<DataQueue> data_queue_;
 };
 
 class BlobBindingData : public SnapshotableObject {
diff --git a/src/node_bob-inl.h b/src/node_bob-inl.h
index 578e45e9ff6149..5dec96282ad305 100644
--- a/src/node_bob-inl.h
+++ b/src/node_bob-inl.h
@@ -25,8 +25,7 @@ int SourceImpl<T>::Pull(
 
   status = DoPull(std::move(next), options, data, count, max_count_hint);
 
-  if (status == bob::Status::STATUS_END)
-    eos_ = true;
+  if (status == bob::Status::STATUS_EOS) eos_ = true;
 
   return status;
 }
diff --git a/src/node_bob.h b/src/node_bob.h
index 74571608f3ae56..b56c5f02cc70d6 100644
--- a/src/node_bob.h
+++ b/src/node_bob.h
@@ -11,12 +11,7 @@ constexpr size_t kMaxCountHint = 16;
 // Negative status codes indicate error conditions.
 enum Status : int {
   // Indicates that an attempt was made to pull after end.
-  STATUS_EOS = -1,
-
-  // Indicates the end of the stream. No additional
-  // data will be available and the consumer should stop
-  // pulling.
-  STATUS_END = 0,
+  STATUS_EOS = 0,
 
   // Indicates that there is additional data available
   // and the consumer may continue to pull.
@@ -72,6 +67,7 @@ using Next = std::function<void(int, const T*, size_t count, Done done)>;
 template <typename T>
 class Source {
  public:
+  virtual ~Source() = default;
   virtual int Pull(
       Next<T> next,
       int options,
diff --git a/src/node_errors.h b/src/node_errors.h
index 3f17a80a6278cc..1c0c342df36cfc 100644
--- a/src/node_errors.h
+++ b/src/node_errors.h
@@ -156,6 +156,7 @@ ERRORS_WITH_CODE(V)
     "Context not associated with Node.js environment")                         \
   V(ERR_INVALID_ADDRESS, "Invalid socket address")                             \
   V(ERR_INVALID_MODULE, "No such module")                                      \
+  V(ERR_INVALID_STATE, "Invalid state")                                        \
   V(ERR_INVALID_THIS, "Value of \"this\" is the wrong type")                   \
   V(ERR_INVALID_TRANSFER_OBJECT, "Found invalid object in transferList")       \
   V(ERR_MEMORY_ALLOCATION_FAILED, "Failed to allocate memory")                 \
diff --git a/src/node_file.cc b/src/node_file.cc
index 49ed96eca05416..673d67e2d3c010 100644
--- a/src/node_file.cc
+++ b/src/node_file.cc
@@ -242,14 +242,20 @@ FileHandle::FileHandle(BindingData* binding_data,
 }
 
 FileHandle* FileHandle::New(BindingData* binding_data,
-                            int fd, Local<Object> obj) {
+                            int fd,
+                            Local<Object> obj,
+                            std::optional<int64_t> maybeOffset,
+                            std::optional<int64_t> maybeLength) {
   Environment* env = binding_data->env();
   if (obj.IsEmpty() && !env->fd_constructor_template()
                             ->NewInstance(env->context())
                             .ToLocal(&obj)) {
     return nullptr;
   }
-  return new FileHandle(binding_data, obj, fd);
+  auto handle = new FileHandle(binding_data, obj, fd);
+  if (maybeOffset.has_value()) handle->read_offset_ = maybeOffset.value();
+  if (maybeLength.has_value()) handle->read_length_ = maybeLength.value();
+  return handle;
 }
 
 void FileHandle::New(const FunctionCallbackInfo<Value>& args) {
@@ -258,13 +264,18 @@ void FileHandle::New(const FunctionCallbackInfo<Value>& args) {
   CHECK(args.IsConstructCall());
   CHECK(args[0]->IsInt32());
 
-  FileHandle* handle =
-      FileHandle::New(binding_data, args[0].As<Int32>()->Value(), args.This());
-  if (handle == nullptr) return;
+  std::optional<int64_t> maybeOffset = std::nullopt;
+  std::optional<int64_t> maybeLength = std::nullopt;
   if (args[1]->IsNumber())
-    handle->read_offset_ = args[1]->IntegerValue(env->context()).FromJust();
+    maybeOffset = args[1]->IntegerValue(env->context()).FromJust();
   if (args[2]->IsNumber())
-    handle->read_length_ = args[2]->IntegerValue(env->context()).FromJust();
+    maybeLength = args[2]->IntegerValue(env->context()).FromJust();
+
+  FileHandle::New(binding_data,
+                  args[0].As<Int32>()->Value(),
+                  args.This(),
+                  maybeOffset,
+                  maybeLength);
 }
 
 FileHandle::~FileHandle() {
@@ -503,10 +514,15 @@ void FileHandle::Close(const FunctionCallbackInfo<Value>& args) {
 void FileHandle::ReleaseFD(const FunctionCallbackInfo<Value>& args) {
   FileHandle* fd;
   ASSIGN_OR_RETURN_UNWRAP(&fd, args.Holder());
-  // Just act as if this FileHandle has been closed.
-  fd->AfterClose();
+  fd->Release();
 }
 
+int FileHandle::Release() {
+  int fd = GetFD();
+  // Just pretend that Close was called and we're all done.
+  AfterClose();
+  return fd;
+}
 
 void FileHandle::AfterClose() {
   closing_ = false;
diff --git a/src/node_file.h b/src/node_file.h
index 5554ba7de51a2a..28c62bef799f81 100644
--- a/src/node_file.h
+++ b/src/node_file.h
@@ -3,6 +3,7 @@
 
 #if defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS
 
+#include <optional>
 #include "aliased_buffer.h"
 #include "node_messaging.h"
 #include "node_snapshotable.h"
@@ -302,13 +303,17 @@ class FileHandle final : public AsyncWrap, public StreamBase {
 
   static FileHandle* New(BindingData* binding_data,
                          int fd,
-                         v8::Local<v8::Object> obj = v8::Local<v8::Object>());
+                         v8::Local<v8::Object> obj = v8::Local<v8::Object>(),
+                         std::optional<int64_t> maybeOffset = std::nullopt,
+                         std::optional<int64_t> maybeLength = std::nullopt);
   ~FileHandle() override;
 
   static void New(const v8::FunctionCallbackInfo<v8::Value>& args);
 
   int GetFD() override { return fd_; }
 
+  int Release();
+
   // Will asynchronously close the FD and return a Promise that will
   // be resolved once closing is complete.
   static void Close(const v8::FunctionCallbackInfo<v8::Value>& args);
diff --git a/test/cctest/test_dataqueue.cc b/test/cctest/test_dataqueue.cc
new file mode 100644
index 00000000000000..73488fcab0a4d1
--- /dev/null
+++ b/test/cctest/test_dataqueue.cc
@@ -0,0 +1,660 @@
+#include <dataqueue/queue.h>
+#include <gtest/gtest.h>
+#include <node_bob-inl.h>
+#include <util-inl.h>
+#include <v8.h>
+#include <memory>
+#include <vector>
+
+using node::DataQueue;
+using v8::ArrayBuffer;
+using v8::BackingStore;
+
+TEST(DataQueue, InMemoryEntry) {
+  char buffer[] = "hello world";
+  size_t len = strlen(buffer);
+
+  std::shared_ptr<BackingStore> store = ArrayBuffer::NewBackingStore(
+      &buffer, len, [](void*, size_t, void*) {}, nullptr);
+
+  // We can create an InMemoryEntry from a v8::BackingStore.
+  std::unique_ptr<DataQueue::Entry> entry =
+      DataQueue::CreateInMemoryEntryFromBackingStore(store, 0, len);
+
+  // The entry is idempotent.
+  CHECK(entry->is_idempotent());
+
+  // The size is known.
+  CHECK_EQ(entry->size().value(), len);
+
+  // We can slice it.
+  // slice: "llo world"
+  std::unique_ptr<DataQueue::Entry> slice1 = entry->slice(2);
+
+  // The slice is idempotent.
+  CHECK(slice1->is_idempotent());
+
+  // The slice size is known.
+  CHECK_EQ(slice1->size().value(), len - 2);
+
+  // We can slice the slice with a length.
+  // slice: "o w"
+  uint64_t end = 5;
+  std::unique_ptr<DataQueue::Entry> slice2 = slice1->slice(2, end);
+
+  // That slice is idempotent.
+  CHECK(slice2->is_idempotent());
+
+  // That slice size is known.
+  CHECK_EQ(slice2->size().value(), 3);
+
+  // The slice end can extend beyond the actual size and will be adjusted.
+  // slice: "orld"
+  end = 100;
+  std::unique_ptr<DataQueue::Entry> slice3 = slice1->slice(5, end);
+  CHECK_NOT_NULL(slice3);
+
+  // The slice size is known.
+  CHECK_EQ(slice3->size().value(), 4);
+
+  // If the slice start is greater than the length, we get a zero length slice.
+  std::unique_ptr<DataQueue::Entry> slice4 = entry->slice(100);
+  CHECK_NOT_NULL(slice4);
+  CHECK_EQ(slice4->size().value(), 0);
+
+  // If the slice end is less than the start, we get a zero length slice.
+  end = 1;
+  std::unique_ptr<DataQueue::Entry> slice5 = entry->slice(2, end);
+  CHECK_NOT_NULL(slice5);
+  CHECK_EQ(slice5->size().value(), 0);
+
+  // If the slice end equal to the start, we get a zero length slice.
+  end = 2;
+  std::unique_ptr<DataQueue::Entry> slice6 = entry->slice(2, end);
+  CHECK_NOT_NULL(slice6);
+  CHECK_EQ(slice6->size().value(), 0);
+
+  // The shared_ptr for the BackingStore should show only 5 uses because
+  // the zero-length slices do not maintain a reference to it.
+  CHECK_EQ(store.use_count(), 5);
+}
+
+TEST(DataQueue, IdempotentDataQueue) {
+  char buffer1[] = "hello world";
+  char buffer2[] = "what fun this is";
+  char buffer3[] = "not added";
+  size_t len1 = strlen(buffer1);
+  size_t len2 = strlen(buffer2);
+  size_t len3 = strlen(buffer3);
+
+  std::shared_ptr<BackingStore> store1 = ArrayBuffer::NewBackingStore(
+      &buffer1, len1, [](void*, size_t, void*) {}, nullptr);
+
+  std::shared_ptr<BackingStore> store2 = ArrayBuffer::NewBackingStore(
+      &buffer2, len2, [](void*, size_t, void*) {}, nullptr);
+
+  std::vector<std::unique_ptr<DataQueue::Entry>> list;
+  list.push_back(
+      DataQueue::CreateInMemoryEntryFromBackingStore(store1, 0, len1));
+  list.push_back(
+      DataQueue::CreateInMemoryEntryFromBackingStore(store2, 0, len2));
+
+  // We can create an idempotent DataQueue from a list of entries.
+  std::shared_ptr<DataQueue> data_queue =
+      DataQueue::CreateIdempotent(std::move(list));
+
+  CHECK_NOT_NULL(data_queue);
+
+  // The data_queue is idempotent.
+  CHECK(data_queue->is_idempotent());
+
+  // The data_queue is capped.
+  CHECK(data_queue->is_capped());
+
+  // maybeCapRemaining() returns zero.
+  CHECK_EQ(data_queue->maybeCapRemaining().value(), 0);
+
+  // Calling cap() is a nonop but doesn't crash or error.
+  data_queue->cap();
+  data_queue->cap(100);
+
+  // maybeCapRemaining() still returns zero.
+  CHECK_EQ(data_queue->maybeCapRemaining().value(), 0);
+
+  // The size is known to be the sum of the in memory-entries.
+  CHECK_EQ(data_queue->size().value(), len1 + len2);
+
+  std::shared_ptr<BackingStore> store3 = ArrayBuffer::NewBackingStore(
+      &buffer3, len3, [](void*, size_t, void*) {}, nullptr);
+
+  // Trying to append a new entry does not crash, but returns std::nullopt.
+  CHECK(!data_queue
+             ->append(DataQueue::CreateInMemoryEntryFromBackingStore(
+                 store3, 0, len3))
+             .has_value());
+
+  // The size has not changed after the append.
+  CHECK_EQ(data_queue->size().value(), len1 + len2);
+
+  // We can acquire multiple readers from the data_queue.
+  std::shared_ptr<DataQueue::Reader> reader1 = data_queue->get_reader();
+  std::shared_ptr<DataQueue::Reader> reader2 = data_queue->get_reader();
+
+  CHECK_NOT_NULL(reader1);
+  CHECK_NOT_NULL(reader2);
+
+  const auto testRead = [&](auto& reader) {
+    // We can read the expected data from reader. Because the entries are
+    // InMemoryEntry instances, reads will be fully synchronous here.
+    bool waitingForPull = true;
+
+    // The first read produces buffer1
+    int status = reader->Pull(
+        [&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {
+          waitingForPull = false;
+          CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+          CHECK_EQ(count, 1);
+          CHECK_EQ(vecs[0].len, len1);
+          CHECK_EQ(memcmp(vecs[0].base, buffer1, len1), 0);
+          std::move(done)(0);
+        },
+        node::bob::OPTIONS_SYNC,
+        nullptr,
+        0,
+        node::bob::kMaxCountHint);
+
+    CHECK(!waitingForPull);
+    CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+
+    // We can read the expected data from reader1. Because the entries are
+    // InMemoryEntry instances, reads will be fully synchronous here.
+    waitingForPull = true;
+
+    // The second read should have status CONTINUE but no buffer.
+    status = reader->Pull(
+        [&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {
+          waitingForPull = false;
+          CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+          CHECK_EQ(count, 0);
+        },
+        node::bob::OPTIONS_SYNC,
+        nullptr,
+        0,
+        node::bob::kMaxCountHint);
+
+    CHECK(!waitingForPull);
+    CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+
+    // The third read produces buffer2, and should be the end.
+    status = reader->Pull(
+        [&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {
+          waitingForPull = false;
+          CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+          CHECK_EQ(count, 1);
+          CHECK_EQ(vecs[0].len, len2);
+          CHECK_EQ(memcmp(vecs[0].base, buffer2, len2), 0);
+          std::move(done)(0);
+        },
+        node::bob::OPTIONS_SYNC,
+        nullptr,
+        0,
+        node::bob::kMaxCountHint);
+
+    CHECK(!waitingForPull);
+    CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+
+    status = reader->Pull(
+        [&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {
+          waitingForPull = false;
+          CHECK_EQ(status, node::bob::STATUS_EOS);
+          CHECK_EQ(count, 0);
+        },
+        node::bob::OPTIONS_SYNC,
+        nullptr,
+        0,
+        node::bob::kMaxCountHint);
+
+    CHECK(!waitingForPull);
+    CHECK_EQ(status, node::bob::STATUS_EOS);
+  };
+
+  // Both reader1 and reader2 should pass identical tests.
+  testRead(reader1);
+  testRead(reader2);
+
+  // We can slice the data queue.
+  std::shared_ptr<DataQueue> slice1 = data_queue->slice(2);
+
+  CHECK_NOT_NULL(slice1);
+
+  // The slice is idempotent.
+  CHECK(slice1->is_idempotent());
+
+  // And capped.
+  CHECK(slice1->is_capped());
+
+  // The size is two-bytes less than the original.
+  CHECK_EQ(slice1->size().value(), data_queue->size().value() - 2);
+
+  const auto testSlice = [&](auto& reader) {
+    // We can read the expected data from reader. Because the entries are
+    // InMemoryEntry instances, reads will be fully synchronous here.
+    bool waitingForPull = true;
+
+    // The first read produces a slice of buffer1
+    int status = reader->Pull(
+        [&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {
+          waitingForPull = false;
+          CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+          CHECK_EQ(count, 1);
+          CHECK_EQ(vecs[0].len, len1 - 2);
+          CHECK_EQ(memcmp(vecs[0].base, buffer1 + 2, len1 - 2), 0);
+          std::move(done)(0);
+        },
+        node::bob::OPTIONS_SYNC,
+        nullptr,
+        0,
+        node::bob::kMaxCountHint);
+
+    CHECK(!waitingForPull);
+    CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+
+    // We can read the expected data from reader1. Because the entries are
+    // InMemoryEntry instances, reads will be fully synchronous here.
+    waitingForPull = true;
+
+    // The second read should have status CONTINUE but no buffer.
+    status = reader->Pull(
+        [&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {
+          waitingForPull = false;
+          CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+          CHECK_EQ(count, 0);
+        },
+        node::bob::OPTIONS_SYNC,
+        nullptr,
+        0,
+        node::bob::kMaxCountHint);
+
+    CHECK(!waitingForPull);
+    CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+
+    // The second read produces buffer2, and should be the end.
+    status = reader->Pull(
+        [&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {
+          waitingForPull = false;
+          CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+          CHECK_EQ(count, 1);
+          CHECK_EQ(vecs[0].len, len2);
+          CHECK_EQ(memcmp(vecs[0].base, buffer2, len2), 0);
+          std::move(done)(0);
+        },
+        node::bob::OPTIONS_SYNC,
+        nullptr,
+        0,
+        node::bob::kMaxCountHint);
+
+    CHECK(!waitingForPull);
+    CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+
+    // The third read produces EOS
+    status = reader->Pull(
+        [&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {
+          waitingForPull = false;
+          CHECK_EQ(status, node::bob::STATUS_EOS);
+          CHECK_EQ(count, 0);
+          CHECK_NULL(vecs);
+          std::move(done)(0);
+        },
+        node::bob::OPTIONS_SYNC,
+        nullptr,
+        0,
+        node::bob::kMaxCountHint);
+
+    CHECK(!waitingForPull);
+    CHECK_EQ(status, node::bob::STATUS_EOS);
+  };
+
+  // We can read the expected slice data.
+  std::shared_ptr<DataQueue::Reader> reader3 = slice1->get_reader();
+  testSlice(reader3);
+
+  // We can slice correctly across boundaries.
+  uint64_t end = 20;
+  std::shared_ptr<DataQueue> slice2 = data_queue->slice(5, end);
+
+  // The size is known.
+  CHECK_EQ(slice2->size().value(), 15);
+
+  const auto testSlice2 = [&](auto& reader) {
+    // We can read the expected data from reader. Because the entries are
+    // InMemoryEntry instances, reads will be fully synchronous here.
+    bool waitingForPull = true;
+
+    // The first read produces a slice of buffer1
+    int status = reader->Pull(
+        [&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {
+          waitingForPull = false;
+
+          CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+          CHECK_EQ(count, 1);
+          CHECK_EQ(vecs[0].len, len1 - 5);
+          CHECK_EQ(memcmp(vecs[0].base, buffer1 + 5, len1 - 5), 0);
+          std::move(done)(0);
+        },
+        node::bob::OPTIONS_SYNC,
+        nullptr,
+        0,
+        node::bob::kMaxCountHint);
+
+    CHECK(!waitingForPull);
+    CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+
+    // We can read the expected data from reader1. Because the entries are
+    // InMemoryEntry instances, reads will be fully synchronous here.
+    waitingForPull = true;
+
+    status = reader->Pull(
+        [&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {
+          waitingForPull = false;
+          CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+          CHECK_EQ(count, 0);
+        },
+        node::bob::OPTIONS_SYNC,
+        nullptr,
+        0,
+        node::bob::kMaxCountHint);
+
+    CHECK(!waitingForPull);
+    CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+
+    // The next read produces buffer2, and should be the end.
+    status = reader->Pull(
+        [&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {
+          waitingForPull = false;
+          CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+          CHECK_EQ(count, 1);
+          CHECK_EQ(vecs[0].len, len2 - 7);
+          CHECK_EQ(memcmp(vecs[0].base, buffer2, len2 - 7), 0);
+          std::move(done)(0);
+        },
+        node::bob::OPTIONS_SYNC,
+        nullptr,
+        0,
+        node::bob::kMaxCountHint);
+
+    CHECK(!waitingForPull);
+    CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+
+    // The next read produces EOS
+    status = reader->Pull(
+        [&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {
+          waitingForPull = false;
+          CHECK_EQ(status, node::bob::STATUS_EOS);
+          CHECK_EQ(count, 0);
+          CHECK_NULL(vecs);
+          std::move(done)(0);
+        },
+        node::bob::OPTIONS_SYNC,
+        nullptr,
+        0,
+        node::bob::kMaxCountHint);
+
+    CHECK(!waitingForPull);
+    CHECK_EQ(status, node::bob::STATUS_EOS);
+  };
+
+  // We can read the expected slice data.
+  std::shared_ptr<DataQueue::Reader> reader4 = slice2->get_reader();
+  testSlice2(reader4);
+}
+
+TEST(DataQueue, NonIdempotentDataQueue) {
+  char buffer1[] = "hello world";
+  char buffer2[] = "what fun this is";
+  char buffer3[] = "not added";
+  size_t len1 = strlen(buffer1);
+  size_t len2 = strlen(buffer2);
+  size_t len3 = strlen(buffer3);
+
+  std::shared_ptr<BackingStore> store1 = ArrayBuffer::NewBackingStore(
+      &buffer1, len1, [](void*, size_t, void*) {}, nullptr);
+
+  std::shared_ptr<BackingStore> store2 = ArrayBuffer::NewBackingStore(
+      &buffer2, len2, [](void*, size_t, void*) {}, nullptr);
+
+  std::shared_ptr<BackingStore> store3 = ArrayBuffer::NewBackingStore(
+      &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();
+
+  CHECK(!data_queue->is_idempotent());
+  CHECK_EQ(data_queue->size().value(), 0);
+
+  data_queue->append(
+      DataQueue::CreateInMemoryEntryFromBackingStore(store1, 0, len1));
+  CHECK_EQ(data_queue->size().value(), len1);
+
+  data_queue->append(
+      DataQueue::CreateInMemoryEntryFromBackingStore(store2, 0, len2));
+  CHECK_EQ(data_queue->size().value(), len1 + len2);
+
+  CHECK(!data_queue->is_capped());
+  CHECK(!data_queue->maybeCapRemaining().has_value());
+
+  data_queue->cap(100);
+  CHECK(data_queue->is_capped());
+  CHECK_EQ(data_queue->maybeCapRemaining().value(), 100 - (len1 + len2));
+
+  data_queue->cap(101);
+  CHECK(data_queue->is_capped());
+  CHECK_EQ(data_queue->maybeCapRemaining().value(), 100 - (len1 + len2));
+
+  data_queue->cap();
+  CHECK(data_queue->is_capped());
+  CHECK_EQ(data_queue->maybeCapRemaining().value(), 0);
+
+  // We can't add any more because the data queue is capped.
+  CHECK_EQ(data_queue
+               ->append(DataQueue::CreateInMemoryEntryFromBackingStore(
+                   store3, 0, len3))
+               .value(),
+           false);
+
+  // We cannot slice a non-idempotent data queue
+  std::shared_ptr<DataQueue> slice1 = data_queue->slice(2);
+  CHECK_NULL(slice1);
+
+  // We can acquire only a single reader for a non-idempotent data queue
+  std::shared_ptr<DataQueue::Reader> reader1 = data_queue->get_reader();
+  std::shared_ptr<DataQueue::Reader> reader2 = data_queue->get_reader();
+
+  CHECK_NOT_NULL(reader1);
+  CHECK_NULL(reader2);
+
+  const auto testRead = [&](auto& reader) {
+    // We can read the expected data from reader. Because the entries are
+    // InMemoryEntry instances, reads will be fully synchronous here.
+    bool waitingForPull = true;
+
+    // The first read produces buffer1
+    int status = reader->Pull(
+        [&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {
+          waitingForPull = false;
+          CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+          CHECK_EQ(count, 1);
+          CHECK_EQ(vecs[0].len, len1);
+          CHECK_EQ(memcmp(vecs[0].base, buffer1, len1), 0);
+          std::move(done)(0);
+        },
+        node::bob::OPTIONS_SYNC,
+        nullptr,
+        0,
+        node::bob::kMaxCountHint);
+
+    CHECK(!waitingForPull);
+    CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+
+    // We can read the expected data from reader1. Because the entries are
+    // InMemoryEntry instances, reads will be fully synchronous here.
+    waitingForPull = true;
+
+    status = reader->Pull(
+        [&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {
+          waitingForPull = false;
+          CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+          CHECK_EQ(count, 0);
+        },
+        node::bob::OPTIONS_SYNC,
+        nullptr,
+        0,
+        node::bob::kMaxCountHint);
+
+    CHECK(!waitingForPull);
+    CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+
+    // The next read produces buffer2, and should be the end.
+    status = reader->Pull(
+        [&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {
+          waitingForPull = false;
+          CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+          CHECK_EQ(count, 1);
+          CHECK_EQ(vecs[0].len, len2);
+          CHECK_EQ(memcmp(vecs[0].base, buffer2, len2), 0);
+          std::move(done)(0);
+        },
+        node::bob::OPTIONS_SYNC,
+        nullptr,
+        0,
+        node::bob::kMaxCountHint);
+
+    CHECK(!waitingForPull);
+    CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+
+    // The next read produces EOS
+    status = reader->Pull(
+        [&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {
+          waitingForPull = false;
+          CHECK_EQ(status, node::bob::STATUS_EOS);
+          CHECK_EQ(count, 0);
+          CHECK_NULL(vecs);
+          std::move(done)(0);
+        },
+        node::bob::OPTIONS_SYNC,
+        nullptr,
+        0,
+        node::bob::kMaxCountHint);
+
+    CHECK(!waitingForPull);
+    CHECK_EQ(status, node::bob::STATUS_EOS);
+  };
+
+  // Reading produces the expected results.
+  testRead(reader1);
+
+  // We still cannot acquire another reader.
+  std::shared_ptr<DataQueue::Reader> reader3 = data_queue->get_reader();
+  CHECK_NULL(reader3);
+
+  CHECK_NOT_NULL(data_queue);
+}
+
+TEST(DataQueue, DataQueueEntry) {
+  char buffer1[] = "hello world";
+  char buffer2[] = "what fun this is";
+  size_t len1 = strlen(buffer1);
+  size_t len2 = strlen(buffer2);
+
+  std::shared_ptr<BackingStore> store1 = ArrayBuffer::NewBackingStore(
+      &buffer1, len1, [](void*, size_t, void*) {}, nullptr);
+
+  std::shared_ptr<BackingStore> store2 = ArrayBuffer::NewBackingStore(
+      &buffer2, len2, [](void*, size_t, void*) {}, nullptr);
+
+  std::vector<std::unique_ptr<DataQueue::Entry>> list;
+  list.push_back(
+      DataQueue::CreateInMemoryEntryFromBackingStore(store1, 0, len1));
+  list.push_back(
+      DataQueue::CreateInMemoryEntryFromBackingStore(store2, 0, len2));
+
+  // We can create an idempotent DataQueue from a list of entries.
+  std::shared_ptr<DataQueue> data_queue =
+      DataQueue::CreateIdempotent(std::move(list));
+
+  CHECK_NOT_NULL(data_queue);
+
+  // We can create an Entry from a data queue.
+  std::unique_ptr<DataQueue::Entry> entry =
+      DataQueue::CreateDataQueueEntry(data_queue);
+
+  // The entry should be idempotent since the data queue is idempotent.
+  CHECK(entry->is_idempotent());
+
+  // The entry size should match the data queue size.
+  CHECK_EQ(entry->size().value(), data_queue->size().value());
+
+  // We can slice it since it is idempotent.
+  uint64_t end = 20;
+  std::unique_ptr<DataQueue::Entry> slice = entry->slice(5, end);
+
+  // The slice has the expected length.
+  CHECK_EQ(slice->size().value(), 15);
+
+  // We can add it to another data queue, even if the new one is not
+  // idempotent.
+
+  std::shared_ptr<DataQueue> data_queue2 = DataQueue::Create();
+  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);
+
+  std::shared_ptr<DataQueue::Reader> reader = data_queue2->get_reader();
+
+  bool pullIsPending = true;
+
+  int status = reader->Pull(
+      [&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {
+        pullIsPending = false;
+        CHECK_EQ(count, 1);
+        CHECK_EQ(memcmp(vecs[0].base, buffer1 + 5, len1 - 5), 0);
+        CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+      },
+      node::bob::OPTIONS_SYNC,
+      nullptr,
+      0);
+
+  // All of the actual entries are in-memory entries so reads should be sync.
+  CHECK(!pullIsPending);
+  CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+
+  // Read to completion...
+  while (status != node::bob::STATUS_EOS) {
+    status = reader->Pull(
+        [&](auto, auto, auto, auto) {}, node::bob::OPTIONS_SYNC, nullptr, 0);
+  }
+
+  // Because the original data queue is idempotent, we can still read from it,
+  // even though we have already consumed the non-idempotent data queue that
+  // contained it.
+
+  std::shared_ptr<DataQueue::Reader> reader2 = data_queue->get_reader();
+  CHECK_NOT_NULL(reader2);
+
+  pullIsPending = true;
+
+  status = reader2->Pull(
+      [&](int status, const DataQueue::Vec* vecs, size_t count, auto done) {
+        pullIsPending = false;
+        CHECK_EQ(count, 1);
+        CHECK_EQ(memcmp(vecs[0].base, buffer1, len1), 0);
+        CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+      },
+      node::bob::OPTIONS_SYNC,
+      nullptr,
+      0);
+
+  // All of the actual entries are in-memory entries so reads should be sync.
+  CHECK(!pullIsPending);
+  CHECK_EQ(status, node::bob::STATUS_CONTINUE);
+}
diff --git a/test/parallel/test-blob-file-backed.js b/test/parallel/test-blob-file-backed.js
new file mode 100644
index 00000000000000..7271fef2f3ce46
--- /dev/null
+++ b/test/parallel/test-blob-file-backed.js
@@ -0,0 +1,81 @@
+'use strict';
+
+const common = require('../common');
+const {
+  strictEqual,
+  rejects,
+} = require('assert');
+const { TextDecoder } = require('util');
+const {
+  writeFileSync,
+  openAsBlob,
+} = require('fs');
+
+const {
+  unlink
+} = require('fs/promises');
+const path = require('path');
+const { Blob } = require('buffer');
+
+const tmpdir = require('../common/tmpdir');
+const testfile = path.join(tmpdir.path, 'test-file-backed-blob.txt');
+const testfile2 = path.join(tmpdir.path, 'test-file-backed-blob2.txt');
+tmpdir.refresh();
+
+const data = `${'a'.repeat(1000)}${'b'.repeat(2000)}`;
+
+writeFileSync(testfile, data);
+writeFileSync(testfile2, data.repeat(100));
+
+(async () => {
+  const blob = await openAsBlob(testfile);
+
+  const ab = await blob.arrayBuffer();
+  const dec = new TextDecoder();
+
+  strictEqual(dec.decode(new Uint8Array(ab)), data);
+  strictEqual(await blob.text(), data);
+
+  // Can be read multiple times
+  let stream = blob.stream();
+  let check = '';
+  for await (const chunk of stream)
+    check = dec.decode(chunk);
+  strictEqual(check, data);
+
+  // Can be combined with other Blob's and read
+  const combined = new Blob(['hello', blob, 'world']);
+  const ab2 = await combined.arrayBuffer();
+  strictEqual(dec.decode(ab2.slice(0, 5)), 'hello');
+  strictEqual(dec.decode(ab2.slice(5, -5)), data);
+  strictEqual(dec.decode(ab2.slice(-5)), 'world');
+
+  // If the file is modified tho, the stream errors.
+  writeFileSync(testfile, data + 'abc');
+
+  stream = blob.stream();
+
+  const read = async () => {
+    // eslint-disable-next-line no-unused-vars, no-empty
+    for await (const _ of stream) {}
+  };
+
+  await rejects(read(), { name: 'NotReadableError' });
+
+  await unlink(testfile);
+})().then(common.mustCall());
+
+(async () => {
+  const blob = await openAsBlob(testfile2);
+  const stream = blob.stream();
+  const read = async () => {
+    // eslint-disable-next-line no-unused-vars
+    for await (const _ of stream) {
+      writeFileSync(testfile2, data + 'abc');
+    }
+  };
+
+  await rejects(read(), { name: 'NotReadableError' });
+
+  await unlink(testfile2);
+})().then(common.mustCall());
diff --git a/test/parallel/test-blob.js b/test/parallel/test-blob.js
index 9ffa31ccdd56e3..68fd2d547a0e80 100644
--- a/test/parallel/test-blob.js
+++ b/test/parallel/test-blob.js
@@ -224,7 +224,6 @@ assert.throws(() => new Blob({}), {
   // The Blob has to be over a specific size for the data to
   // be copied asynchronously..
   const b = new Blob(['hello', 'there'.repeat(820)]);
-  assert.strictEqual(b.arrayBuffer(), b.arrayBuffer());
   b.arrayBuffer().then(common.mustCall());
 }
 
diff --git a/test/parallel/test-bootstrap-modules.js b/test/parallel/test-bootstrap-modules.js
index 693fa9efb4111b..c0fd99055de445 100644
--- a/test/parallel/test-bootstrap-modules.js
+++ b/test/parallel/test-bootstrap-modules.js
@@ -9,83 +9,86 @@ const common = require('../common');
 const assert = require('assert');
 
 const expectedModules = new Set([
-  'Internal Binding async_wrap',
-  'Internal Binding blob',
-  'Internal Binding buffer',
   'Internal Binding builtins',
-  'Internal Binding config',
-  'Internal Binding constants',
-  'Internal Binding contextify',
-  'Internal Binding credentials',
   'Internal Binding errors',
-  'Internal Binding fs',
-  'Internal Binding messaging',
-  'Internal Binding mksnapshot',
-  'Internal Binding module_wrap',
-  'Internal Binding options',
-  'Internal Binding performance',
-  'Internal Binding process_methods',
-  'Internal Binding string_decoder',
-  'Internal Binding symbols',
-  'Internal Binding task_queue',
+  'Internal Binding util',
+  'NativeModule internal/errors',
+  'Internal Binding config',
   'Internal Binding timers',
-  'Internal Binding trace_events',
+  'Internal Binding async_wrap',
+  'Internal Binding task_queue',
+  'Internal Binding symbols',
+  'NativeModule internal/async_hooks',
+  'Internal Binding constants',
   'Internal Binding types',
-  'Internal Binding url',
-  'Internal Binding util',
-  'Internal Binding wasm_web_api',
-  'Internal Binding worker',
-  'NativeModule async_hooks',
-  'NativeModule buffer',
-  'NativeModule events',
-  'NativeModule fs',
+  'NativeModule internal/util',
+  'NativeModule internal/util/types',
+  'NativeModule internal/validators',
+  'NativeModule internal/linkedlist',
+  'NativeModule internal/priority_queue',
   'NativeModule internal/assert',
-  'NativeModule internal/async_hooks',
+  'NativeModule internal/util/inspect',
+  'NativeModule internal/util/debuglog',
+  'NativeModule internal/timers',
+  'NativeModule events',
+  'Internal Binding buffer',
+  'Internal Binding string_decoder',
   'NativeModule internal/buffer',
+  'NativeModule buffer',
+  'Internal Binding messaging',
+  'NativeModule internal/worker/js_transferable',
+  'Internal Binding process_methods',
+  'NativeModule internal/process/per_thread',
+  'Internal Binding credentials',
+  'NativeModule internal/process/promises',
+  'NativeModule internal/fixed_queue',
+  'NativeModule async_hooks',
+  'NativeModule internal/process/task_queues',
+  'NativeModule timers',
+  'Internal Binding trace_events',
+  'NativeModule internal/constants',
+  'NativeModule path',
+  'NativeModule internal/process/execution',
+  'NativeModule internal/process/warning',
   'NativeModule internal/console/constructor',
   'NativeModule internal/console/global',
-  'NativeModule internal/constants',
-  'NativeModule internal/dns/utils',
-  'NativeModule internal/errors',
+  'NativeModule internal/querystring',
+  'NativeModule querystring',
+  'Internal Binding url',
+  'Internal Binding blob',
+  'NativeModule internal/url',
+  'NativeModule util',
+  'Internal Binding performance',
+  'NativeModule internal/perf/utils',
   'NativeModule internal/event_target',
-  'NativeModule internal/fixed_queue',
+  'Internal Binding mksnapshot',
+  'NativeModule internal/v8/startup_snapshot',
+  'NativeModule internal/process/signal',
+  'Internal Binding fs',
+  'NativeModule internal/encoding',
+  'NativeModule internal/webstreams/util',
+  'NativeModule internal/webstreams/queuingstrategies',
+  'NativeModule internal/blob',
   'NativeModule internal/fs/utils',
+  'NativeModule fs',
   'NativeModule internal/idna',
-  'NativeModule internal/linkedlist',
-  'NativeModule internal/modules/cjs/loader',
-  'NativeModule internal/modules/esm/utils',
+  'Internal Binding options',
+  'NativeModule internal/options',
+  'NativeModule internal/source_map/source_map_cache',
+  'Internal Binding contextify',
+  'NativeModule internal/vm',
   'NativeModule internal/modules/helpers',
   'NativeModule internal/modules/package_json_reader',
+  'Internal Binding module_wrap',
+  'NativeModule internal/modules/cjs/loader',
+  'NativeModule internal/vm/module',
+  'NativeModule internal/modules/esm/utils',
+  'Internal Binding wasm_web_api',
+  'Internal Binding worker',
   'NativeModule internal/modules/run_main',
   'NativeModule internal/net',
-  'NativeModule internal/options',
-  'NativeModule internal/perf/utils',
-  'NativeModule internal/priority_queue',
-  'NativeModule internal/process/execution',
-  'NativeModule internal/process/per_thread',
+  'NativeModule internal/dns/utils',
   'NativeModule internal/process/pre_execution',
-  'NativeModule internal/process/promises',
-  'NativeModule internal/process/signal',
-  'NativeModule internal/process/task_queues',
-  'NativeModule internal/process/warning',
-  'NativeModule internal/querystring',
-  'NativeModule internal/source_map/source_map_cache',
-  'NativeModule internal/timers',
-  'NativeModule internal/url',
-  'NativeModule internal/util',
-  'NativeModule internal/util/debuglog',
-  'NativeModule internal/util/inspect',
-  'NativeModule internal/util/types',
-  'NativeModule internal/v8/startup_snapshot',
-  'NativeModule internal/validators',
-  'NativeModule internal/vm',
-  'NativeModule internal/vm/module',
-  'NativeModule internal/worker/js_transferable',
-  'NativeModule path',
-  'NativeModule querystring',
-  'NativeModule timers',
-  'NativeModule url',
-  'NativeModule util',
 ]);
 
 if (!common.isMainThread) {
@@ -127,6 +130,7 @@ if (common.isWindows) {
 
 if (common.hasIntl) {
   expectedModules.add('Internal Binding icu');
+  expectedModules.add('NativeModule url');
 } else {
   expectedModules.add('NativeModule url');
 }
diff --git a/test/parallel/test-trace-events-threadpool.js b/test/parallel/test-trace-events-threadpool.js
index dd1cc02e53d793..a5af5e78661dab 100644
--- a/test/parallel/test-trace-events-threadpool.js
+++ b/test/parallel/test-trace-events-threadpool.js
@@ -5,20 +5,18 @@ const cp = require('child_process');
 const fs = require('fs');
 const path = require('path');
 const tmpdir = require('../common/tmpdir');
+const { scheduler } = require('timers/promises');
 
 if (!common.hasCrypto)
   common.skip('missing crypto');
 
 const { hkdf } = require('crypto');
 const { deflate } = require('zlib');
-const { Blob } = require('buffer');
 
 if (process.env.isChild === '1') {
   hkdf('sha512', 'key', 'salt', 'info', 64, () => {});
   deflate('hello', () => {});
-  // Make async call
-  const blob = new Blob(['h'.repeat(4096 * 2)]);
-  blob.arrayBuffer();
+  scheduler.wait(10);
   return;
 }
 
@@ -46,7 +44,6 @@ const traces = JSON.parse(data.toString()).traceEvents;
 
 assert(traces.length > 0);
 
-let blobCount = 0;
 let zlibCount = 0;
 let cryptoCount = 0;
 
@@ -55,9 +52,7 @@ traces.forEach((item) => {
     'node,node.threadpoolwork,node.threadpoolwork.sync',
     'node,node.threadpoolwork,node.threadpoolwork.async',
   ].includes(item.cat)) {
-    if (item.name === 'blob') {
-      blobCount++;
-    } else if (item.name === 'zlib') {
+    if (item.name === 'zlib') {
       zlibCount++;
     } else if (item.name === 'crypto') {
       cryptoCount++;
@@ -65,7 +60,6 @@ traces.forEach((item) => {
   }
 });
 
-// There are three types, each type has two async events and sync events at least
-assert.ok(blobCount >= 4);
+// There are two types, each type has two async events and sync events at least
 assert.ok(zlibCount >= 4);
 assert.ok(cryptoCount >= 4);
diff --git a/test/sequential/test-async-wrap-getasyncid.js b/test/sequential/test-async-wrap-getasyncid.js
index e6464466aa43e5..dcd33a9f2785db 100644
--- a/test/sequential/test-async-wrap-getasyncid.js
+++ b/test/sequential/test-async-wrap-getasyncid.js
@@ -63,7 +63,7 @@ const { getSystemErrorName } = require('util');
     delete providers.ELDHISTOGRAM;
     delete providers.SIGINTWATCHDOG;
     delete providers.WORKERHEAPSNAPSHOT;
-    delete providers.FIXEDSIZEBLOBCOPY;
+    delete providers.BLOBREADER;
     delete providers.RANDOMPRIMEREQUEST;
     delete providers.CHECKPRIMEREQUEST;