[WIP] The Great Streams Consolidation of 2026

src/workerd/api/container.c++

@@ -154,8 +154,8 @@ jsg::Promise<jsg::Ref<ExecOutput>> ExecProcess::output(jsg::Lock& js) {
     stdoutPromise =
         stream->getController()
             .readAllBytes(js, IoContext::current().getLimitEnforcer().getBufferingLimit())
-            .then(js, [](jsg::Lock&, jsg::BufferSource bytes) {
-      return kj::heapArray(bytes.asArrayPtr());
+            .then(js, [](jsg::Lock& js, jsg::JsRef<jsg::JsArrayBuffer> bytes) {
+      return bytes.getHandle(js).copy();
     });
   }
 
@@ -165,8 +165,8 @@ jsg::Promise<jsg::Ref<ExecOutput>> ExecProcess::output(jsg::Lock& js) {
         "Cannot call output() after stderr has started being consumed.");
     stderrPromise = stream->getController()
                         .readAllBytes(js, kj::maxValue)
-                        .then(js, [](jsg::Lock&, jsg::BufferSource bytes) {
-      return kj::heapArray(bytes.asArrayPtr());
+                        .then(js, [](jsg::Lock& js, jsg::JsRef<jsg::JsArrayBuffer> bytes) {
+      return bytes.getHandle(js).copy();
     });
   }
 
@@ -525,7 +525,7 @@ jsg::Promise<jsg::Ref<ExecProcess>> Container::exec(
         KJ_CASE_ONEOF(readable, jsg::Ref<ReadableStream>) {
           auto sink = newSystemStream(kj::mv(stdinWriter), StreamEncoding::IDENTITY, ioContext);
           auto pipePromise =
-              (ioContext.waitForDeferredProxy(readable->pumpTo(js, kj::mv(sink), true)));
+              (ioContext.waitForDeferredProxy(readable->pumpTo(js, kj::mv(sink), End::YES)));
           ioContext.addTask(pipePromise.attach(readable.addRef()));
         }
         // user sets "pipe"... they want to consume the API with the stdin WritableStream

src/workerd/api/crypto/crypto.c++

@@ -800,7 +800,7 @@ void DigestStream::dispose(jsg::Lock& js) {
     KJ_IF_SOME(ready, state.tryGet<Ready>()) {
       auto reason = js.typeError("The DigestStream was disposed.");
       ready.resolver.reject(js, reason);
-      state.init<StreamStates::Errored>(js.v8Ref<v8::Value>(reason));
+      state.init<StreamStates::Errored>(reason.addRef(js));
     }
   }
   JSG_CATCH(exception) {
@@ -859,7 +859,7 @@ void DigestStream::abort(jsg::Lock& js, jsg::JsValue reason) {
   // If the state is already closed or errored, then this is a non-op
   KJ_IF_SOME(ready, state.tryGet<Ready>()) {
     ready.resolver.reject(js, reason);
-    state.init<StreamStates::Errored>(js.v8Ref<v8::Value>(reason));
+    state.init<StreamStates::Errored>(reason.addRef(js));
   }
 }
 
@@ -870,7 +870,7 @@ jsg::Ref<DigestStream> DigestStream::constructor(jsg::Lock& js, Algorithm algori
       interpretAlgorithmParam(kj::mv(algorithm)), kj::mv(paf.resolver), kj::mv(paf.promise));
 
   // clang-format off
-  stream->getController().setup(js, UnderlyingSink{
+  auto sink = kj::heap<UnderlyingSinkImpl>(js, UnderlyingSink{
     .write = [&stream = *stream](jsg::Lock& js, v8::Local<v8::Value> chunk, auto c) mutable {
       return js.tryCatch([&] {
         // Make sure what we got can be interpreted as bytes...
@@ -916,9 +916,11 @@ jsg::Ref<DigestStream> DigestStream::constructor(jsg::Lock& js, Algorithm algori
         return js.resolvedPromise();
       }, [&](jsg::Value exception) { return js.rejectedPromise<void>(kj::mv(exception)); });
     }
-  }, kj::none);
+  }, StreamQueuingStrategy {});
   // clang-format on
 
+  stream->getController().setup(js, kj::mv(sink));
+
   return kj::mv(stream);
 }
 

src/workerd/api/eventsource.c++

@@ -474,8 +474,8 @@ void EventSource::run(jsg::Lock& js,
   // pumping the body into an EventSourceSink until the body is closed, canceled,
   // or errored.
   context
-      .awaitIo(
-          js, processBody(context, readable->pumpTo(js, kj::heap<EventSourceSink>(*this), true)))
+      .awaitIo(js,
+          processBody(context, readable->pumpTo(js, kj::heap<EventSourceSink>(*this), End::YES)))
       .then(js, kj::mv(onSuccess), kj::mv(onFailed));
 }
 

src/workerd/api/filesystem.c++

@@ -490,7 +490,7 @@ void FileSystemModule::close(jsg::Lock& js, int fd) {
 }
 
 uint32_t FileSystemModule::write(
-    jsg::Lock& js, int fd, kj::Array<jsg::BufferSource> data, WriteOptions options) {
+    jsg::Lock& js, int fd, kj::Array<jsg::JsRef<jsg::JsBufferSource>> data, WriteOptions options) {
   auto& vfs = workerd::VirtualFileSystem::current(js);
 
   KJ_IF_SOME(opened, vfs.tryGetFd(js, fd)) {
@@ -513,7 +513,7 @@ uint32_t FileSystemModule::write(
         auto pos = getPosition(js, opened.addRef(), file.addRef(), options);
         uint32_t total = 0;
         for (auto& buffer: data) {
-          KJ_SWITCH_ONEOF(file->write(js, pos, buffer)) {
+          KJ_SWITCH_ONEOF(file->write(js, pos, buffer.getHandle(js).asArrayPtr())) {
             KJ_CASE_ONEOF(written, uint32_t) {
               pos += written;
               total += written;
@@ -546,7 +546,7 @@ uint32_t FileSystemModule::write(
 }
 
 uint32_t FileSystemModule::read(
-    jsg::Lock& js, int fd, kj::Array<jsg::BufferSource> data, WriteOptions options) {
+    jsg::Lock& js, int fd, kj::Array<jsg::JsRef<jsg::JsUint8Array>> data, WriteOptions options) {
   auto& vfs = workerd::VirtualFileSystem::current(js);
   KJ_IF_SOME(opened, vfs.tryGetFd(js, fd)) {
     if (!opened->read) {
@@ -561,11 +561,12 @@ uint32_t FileSystemModule::read(
         }
         uint32_t total = 0;
         for (auto& buffer: data) {
-          auto read = file->read(js, pos, buffer);
+          auto handle = buffer.getHandle(js);
+          auto read = file->read(js, pos, handle.asArrayPtr());
           // if read is less than the size of the buffer, we are at EOF.
           pos += read;
           total += read;
-          if (read < buffer.size()) break;
+          if (read < handle.size()) break;
         }
         // We only update the position if the options.position is not set.
         if (options.position == kj::none) {
@@ -588,7 +589,7 @@ uint32_t FileSystemModule::read(
   }
 }
 
-jsg::BufferSource FileSystemModule::readAll(jsg::Lock& js, kj::OneOf<int, FilePath> pathOrFd) {
+jsg::JsUint8Array FileSystemModule::readAll(jsg::Lock& js, kj::OneOf<int, FilePath> pathOrFd) {
   auto& vfs = workerd::VirtualFileSystem::current(js);
   KJ_SWITCH_ONEOF(pathOrFd) {
     KJ_CASE_ONEOF(path, FilePath) {
@@ -597,8 +598,8 @@ jsg::BufferSource FileSystemModule::readAll(jsg::Lock& js, kj::OneOf<int, FilePa
         KJ_SWITCH_ONEOF(node) {
           KJ_CASE_ONEOF(file, kj::Rc<workerd::File>) {
             KJ_SWITCH_ONEOF(file->readAllBytes(js)) {
-              KJ_CASE_ONEOF(data, jsg::BufferSource) {
-                return kj::mv(data);
+              KJ_CASE_ONEOF(data, jsg::JsUint8Array) {
+                return data;
               }
               KJ_CASE_ONEOF(err, workerd::FsError) {
                 throwFsError(js, err, "readAll"_kj);
@@ -635,8 +636,8 @@ jsg::BufferSource FileSystemModule::readAll(jsg::Lock& js, kj::OneOf<int, FilePa
           });
 
           KJ_SWITCH_ONEOF(file->readAllBytes(js)) {
-            KJ_CASE_ONEOF(data, jsg::BufferSource) {
-              return kj::mv(data);
+            KJ_CASE_ONEOF(data, jsg::JsUint8Array) {
+              return data;
             }
             KJ_CASE_ONEOF(err, workerd::FsError) {
               throwFsError(js, err, "freadAll"_kj);
@@ -656,7 +657,7 @@ jsg::BufferSource FileSystemModule::readAll(jsg::Lock& js, kj::OneOf<int, FilePa
 
 uint32_t FileSystemModule::writeAll(jsg::Lock& js,
     kj::OneOf<int, FilePath> pathOrFd,
-    jsg::BufferSource data,
+    jsg::JsBufferSource data,
     WriteAllOptions options) {
   auto& vfs = workerd::VirtualFileSystem::current(js);
 
@@ -684,7 +685,7 @@ uint32_t FileSystemModule::writeAll(jsg::Lock& js,
             // If the append option is set, we will write to the end of the file
             // instead of overwriting it.
             if (options.append) {
-              KJ_SWITCH_ONEOF(file->write(js, stat.size, data)) {
+              KJ_SWITCH_ONEOF(file->write(js, stat.size, data.asArrayPtr())) {
                 KJ_CASE_ONEOF(written, uint32_t) {
                   return written;
                 }
@@ -696,7 +697,7 @@ uint32_t FileSystemModule::writeAll(jsg::Lock& js,
             }
 
             // Otherwise, we overwrite the entire file.
-            KJ_SWITCH_ONEOF(file->writeAll(js, data)) {
+            KJ_SWITCH_ONEOF(file->writeAll(js, data.asArrayPtr())) {
               KJ_CASE_ONEOF(written, uint32_t) {
                 return written;
               }
@@ -737,7 +738,7 @@ uint32_t FileSystemModule::writeAll(jsg::Lock& js,
               node::THROW_ERR_UV_EPERM(js, "writeAll"_kj);
             }
             auto file = workerd::File::newWritable(js, static_cast<uint32_t>(data.size()));
-            KJ_SWITCH_ONEOF(file->writeAll(js, data)) {
+            KJ_SWITCH_ONEOF(file->writeAll(js, data.asArrayPtr())) {
               KJ_CASE_ONEOF(written, uint32_t) {
                 KJ_IF_SOME(err, dir->add(js, relative.name, kj::mv(file))) {
                   throwFsError(js, err, "writeAll"_kj);
@@ -788,14 +789,14 @@ uint32_t FileSystemModule::writeAll(jsg::Lock& js,
           // If the file descriptor was opened in append mode, or if the append option
           // is set, then we'll use write instead to append to the end of the file.
           if (opened->append || options.append) {
-            return write(js, fd, kj::arr(kj::mv(data)),
+            return write(js, fd, kj::arr(data.addRef(js)),
                 {
                   .position = stat.size,
                 });
           }
 
           // Otherwise, we overwrite the entire file.
-          KJ_SWITCH_ONEOF(file->writeAll(js, data)) {
+          KJ_SWITCH_ONEOF(file->writeAll(js, data.asArrayPtr())) {
             KJ_CASE_ONEOF(written, uint32_t) {
               return written;
             }
@@ -1890,9 +1891,8 @@ jsg::Ref<Blob> FileSystemModule::openAsBlob(
       }
       KJ_CASE_ONEOF(file, kj::Rc<workerd::File>) {
         KJ_SWITCH_ONEOF(file->readAllBytes(js)) {
-          KJ_CASE_ONEOF(bytes, jsg::BufferSource) {
-            return js.alloc<Blob>(
-                js, bytes.getJsHandle(js), kj::mv(options.type).orDefault(kj::String()));
+          KJ_CASE_ONEOF(bytes, jsg::JsUint8Array) {
+            return js.alloc<Blob>(js, bytes, kj::mv(options.type).orDefault(kj::String()));
           }
           KJ_CASE_ONEOF(err, workerd::FsError) {
             throwFsError(js, err, "open"_kj);
@@ -2557,10 +2557,10 @@ jsg::Promise<jsg::Ref<File>> FileSystemFileHandle::getFile(
       KJ_CASE_ONEOF(file, kj::Rc<workerd::File>) {
         auto stat = file->stat(js);
         KJ_SWITCH_ONEOF(file->readAllBytes(js)) {
-          KJ_CASE_ONEOF(bytes, jsg::BufferSource) {
+          KJ_CASE_ONEOF(bytes, jsg::JsUint8Array) {
             return js.resolvedPromise(
-                js.alloc<File>(js, bytes.getJsHandle(js), jsg::USVString(kj::str(getName(js))),
-                    kj::String(), (stat.lastModified - kj::UNIX_EPOCH) / kj::MILLISECONDS));
+                js.alloc<File>(js, bytes, jsg::USVString(kj::str(getName(js))), kj::String(),
+                    (stat.lastModified - kj::UNIX_EPOCH) / kj::MILLISECONDS));
           }
           KJ_CASE_ONEOF(err, workerd::FsError) {
             return js.rejectedPromise<jsg::Ref<File>>(
@@ -2713,7 +2713,9 @@ jsg::Promise<jsg::Ref<FileSystemWritableFileStream>> FileSystemFileHandle::creat
       return js.resolvedPromise();
     }, [&](jsg::Value exception) { return js.rejectedPromise<void>(kj::mv(exception)); });
   };
-  stream->getController().setup(js, kj::mv(sink), kj::none);
+
+  stream->getController().setup(
+      js, kj::heap<UnderlyingSinkImpl>(js, kj::mv(sink), StreamQueuingStrategy{}));
 
   return js.resolvedPromise(kj::mv(stream));
 }
@@ -2724,7 +2726,7 @@ FileSystemWritableFileStream::FileSystemWritableFileStream(
       sharedState(kj::mv(sharedState)) {}
 
 jsg::Promise<void> FileSystemWritableFileStream::write(jsg::Lock& js,
-    kj::OneOf<jsg::Ref<Blob>, jsg::BufferSource, kj::String, WriteParams> data,
+    kj::OneOf<jsg::Ref<Blob>, jsg::JsBufferSource, kj::String, WriteParams> data,
     const jsg::TypeHandler<jsg::Ref<jsg::DOMException>>& deHandler) {
   JSG_REQUIRE(!getController().isLockedToWriter(), TypeError,
       "Cannot write to a stream that is locked to a reader");
@@ -2750,8 +2752,8 @@ jsg::Promise<void> FileSystemWritableFileStream::writeImpl(jsg::Lock& js,
             }
           }
         }
-        KJ_CASE_ONEOF(buffer, jsg::BufferSource) {
-          KJ_SWITCH_ONEOF(inner->write(js, state.position, buffer)) {
+        KJ_CASE_ONEOF(buffer, jsg::JsBufferSource) {
+          KJ_SWITCH_ONEOF(inner->write(js, state.position, buffer.asArrayPtr())) {
             KJ_CASE_ONEOF(written, uint32_t) {
               state.position += written;
             }
@@ -2799,8 +2801,8 @@ jsg::Promise<void> FileSystemWritableFileStream::writeImpl(jsg::Lock& js,
                     }
                     KJ_UNREACHABLE;
                   }
-                  KJ_CASE_ONEOF(buffer, jsg::BufferSource) {
-                    KJ_SWITCH_ONEOF(inner->write(js, offset, buffer)) {
+                  KJ_CASE_ONEOF(buffer, jsg::JsRef<jsg::JsBufferSource>) {
+                    KJ_SWITCH_ONEOF(inner->write(js, offset, buffer.getHandle(js).asArrayPtr())) {
                       KJ_CASE_ONEOF(written, uint32_t) {
                         state.position = offset + written;
                         return js.resolvedPromise();

src/workerd/api/filesystem.h

@@ -103,10 +103,10 @@ class FileSystemModule final: public jsg::Object {
     JSG_STRUCT(position);
   };
 
-  uint32_t write(jsg::Lock& js, int fd, kj::Array<jsg::BufferSource> data, WriteOptions options);
-  uint32_t read(jsg::Lock& js, int fd, kj::Array<jsg::BufferSource> data, WriteOptions options);
+  uint32_t write(jsg::Lock& js, int fd, kj::Array<jsg::JsRef<jsg::JsBufferSource>> data, WriteOptions options);
+  uint32_t read(jsg::Lock& js, int fd, kj::Array<jsg::JsRef<jsg::JsUint8Array>> data, WriteOptions options);
 
-  jsg::BufferSource readAll(jsg::Lock& js, kj::OneOf<int, FilePath> pathOrFd);
+  jsg::JsUint8Array readAll(jsg::Lock& js, kj::OneOf<int, FilePath> pathOrFd);
 
   struct WriteAllOptions {
     bool exclusive;
@@ -116,7 +116,7 @@ class FileSystemModule final: public jsg::Object {
 
   uint32_t writeAll(jsg::Lock& js,
       kj::OneOf<int, FilePath> pathOrFd,
-      jsg::BufferSource data,
+      jsg::JsBufferSource data,
       WriteAllOptions options);
 
   struct RenameOrCopyOptions {
@@ -298,12 +298,12 @@ struct FileSystemFileWriteParams {
   jsg::Optional<uint32_t> position;
   // Yes, wrapping the kj::Maybe with a jsg::Optional is intentional here. We need to
   // be able to accept null or undefined values and handle them per the spec.
-  jsg::Optional<kj::Maybe<kj::OneOf<jsg::Ref<Blob>, jsg::BufferSource, kj::String>>> data;
+  jsg::Optional<kj::Maybe<kj::OneOf<jsg::Ref<Blob>, jsg::JsRef<jsg::JsBufferSource>, kj::String>>> data;
   JSG_STRUCT(type, size, position, data);
 };
 
 using FileSystemWritableData =
-    kj::OneOf<jsg::Ref<Blob>, jsg::BufferSource, kj::String, FileSystemFileWriteParams>;
+    kj::OneOf<jsg::Ref<Blob>, jsg::JsBufferSource, kj::String, FileSystemFileWriteParams>;
 
 class FileSystemFileHandle final: public FileSystemHandle {
  public:

src/workerd/api/html-rewriter.c++

@@ -727,7 +727,7 @@ kj::Promise<void> Rewriter::replacerThunkPromise(
 
     auto streamSink = kj::heap<ReplacerStreamSink>(sink, registration.isHtml);
     return ioContext.waitForDeferredProxy(
-        registration.stream->pumpTo(lock, kj::mv(streamSink), true));
+        registration.stream->pumpTo(lock, kj::mv(streamSink), End::YES));
   });
 }
 
@@ -1294,10 +1294,10 @@ jsg::Ref<Response> HTMLRewriter::transform(jsg::Lock& js, jsg::Ref<Response> res
   //   after we know that nothing else (like invalid encoding) could cause an exception.
 
   // Drive and flush the parser asynchronously.
-  ioContext.addTask(
-      ioContext
-          .waitForDeferredProxy(KJ_ASSERT_NONNULL(maybeInput)->pumpTo(js, kj::mv(rewriter), true))
-          .catch_([](kj::Exception&& e) {
+  ioContext.addTask(ioContext
+                        .waitForDeferredProxy(
+                            KJ_ASSERT_NONNULL(maybeInput)->pumpTo(js, kj::mv(rewriter), End::YES))
+                        .catch_([](kj::Exception&& e) {
     // Errors in pumpTo() are already propagated to the destination stream. We don't want to
     // throw them from here since it'll cause an uncaught exception to be reported via taskFailed(),
     // which would poison the IoContext even though the application may have handled the error.