Implement drainingRead mechanism for JS-backed streams (#5838)

Author: jasnell

src/workerd/api/streams/common.h

@@ -71,6 +71,14 @@ struct ReadResult {
   }
 };
 
+// Result type for draining read operations. Always returns bytes, even for value streams.
+// Used by DrainingReader for optimized pipe-to operations with vectored writes.
+// This is a C++ only type - not exposed to JavaScript.
+struct DrainingReadResult {
+  kj::Array<kj::Array<kj::byte>> chunks;  // Multiple byte arrays for vectored writes
+  bool done = false;                      // True if stream is closed/closing
+};
+
 struct StreamQueuingStrategy {
   using SizeAlgorithm = uint64_t(v8::Local<v8::Value>);
 
@@ -498,6 +506,25 @@ class ReadableStreamController {
   virtual kj::Maybe<jsg::Promise<ReadResult>> read(
       jsg::Lock& js, kj::Maybe<ByobOptions> byobOptions) = 0;
 
+  // Performs a draining read operation that:
+  // 1. Drains all currently buffered data from the queue
+  // 2. Pumps the controller for synchronously available data (respecting pull promise state)
+  // 3. Returns bytes even for value streams (converting ArrayBuffer/ArrayBufferView/string)
+  // 4. Has mutual exclusion with regular reads - returns rejected promise if pending regular reads
+  // 5. Returns done: true with final data when stream is closing
+  //
+  // This is a C++ only API (not exposed to JavaScript) intended for optimized pipe operations.
+  // Returns kj::none if the stream is locked in a way that prevents the read.
+  //
+  // The maxRead parameter provides a soft limit on how much data to read. It only applies to
+  // subsequent synchronous pump attempts after draining the currently buffered data. That is,
+  // drainingRead will first drain all currently buffered data (potentially exceeding maxRead),
+  // then will only proceed with additional synchronous reads if the total bytes read so far
+  // is less than maxRead. This prevents runaway reads from neverending or slow streams while
+  // still allowing efficient batch reads for normal streams.
+  virtual kj::Maybe<jsg::Promise<DrainingReadResult>> drainingRead(
+      jsg::Lock& js, size_t maxRead = kj::maxValue) = 0;
+
   // The pipeTo implementation fully consumes the stream by directing all of its data at the
   // destination. Controllers should try to be as efficient as possible here. For instance, if
   // a ReadableStreamInternalController is piping to a WritableStreamInternalController, then

src/workerd/api/streams/internal.c++

@@ -677,6 +677,14 @@ kj::Maybe<jsg::Promise<ReadResult>> ReadableStreamInternalController::read(
   KJ_UNREACHABLE;
 }
 
+kj::Maybe<jsg::Promise<DrainingReadResult>> ReadableStreamInternalController::drainingRead(
+    jsg::Lock& js, size_t maxRead) {
+  // TODO(later): Implement proper drainingRead for internal controller.
+  // For now, return a rejected promise as a placeholder.
+  return js.rejectedPromise<DrainingReadResult>(
+      js.v8TypeError("drainingRead is not yet implemented for internal streams"_kj));
+}
+
 jsg::Promise<void> ReadableStreamInternalController::pipeTo(
     jsg::Lock& js, WritableStreamController& destination, PipeToOptions options) {
 

src/workerd/api/streams/internal.h

@@ -62,6 +62,9 @@ class ReadableStreamInternalController: public ReadableStreamController {
   kj::Maybe<jsg::Promise<ReadResult>> read(
       jsg::Lock& js, kj::Maybe<ByobOptions> byobOptions) override;
 
+  kj::Maybe<jsg::Promise<DrainingReadResult>> drainingRead(
+      jsg::Lock& js, size_t maxRead = kj::maxValue) override;
+
   jsg::Promise<void> pipeTo(
       jsg::Lock& js, WritableStreamController& destination, PipeToOptions options) override;