stream: trim webstreams sink wrappers and BYOB transfers

lib/internal/webstreams/readablestream.js

@@ -111,6 +111,7 @@ const {
   extractSizeAlgorithm,
   getNonWritablePropertyDescriptor,
   isBrandCheck,
+  kDeferredStart,
   kEmptyQueue,
   kParkedAlgorithmResult,
   kPendingPromise,
@@ -521,10 +522,11 @@ class ReadableStream {
   [kTransfer]() {
     if (!isReadableStream(this))
       throw new ERR_INVALID_THIS('ReadableStream');
+    const transfer = readableStreamTransferState(this);
     if (this.locked) {
-      this[kState].transfer.port1?.close();
-      this[kState].transfer.port1 = undefined;
-      this[kState].transfer.port2 = undefined;
+      transfer.port1?.close();
+      transfer.port1 = undefined;
+      transfer.port2 = undefined;
       throw new DOMException(
         'Cannot transfer a locked ReadableStream',
         'DataCloneError');
@@ -533,24 +535,23 @@ class ReadableStream {
     const {
       writable,
       promise,
-    } = lazyTransfer().newCrossRealmWritableSink(
-      this,
-      this[kState].transfer.port1);
+    } = lazyTransfer().newCrossRealmWritableSink(this, transfer.port1);
 
-    this[kState].transfer.writable = writable;
-    this[kState].transfer.promise = promise;
+    transfer.writable = writable;
+    transfer.promise = promise;
 
     return {
-      data: { port: this[kState].transfer.port2 },
+      data: { port: transfer.port2 },
       deserializeInfo:
         'internal/webstreams/readablestream:TransferredReadableStream',
     };
   }
 
   [kTransferList]() {
     const { port1, port2 } = new MessageChannel();
-    this[kState].transfer.port1 = port1;
-    this[kState].transfer.port2 = port2;
+    const transfer = readableStreamTransferState(this);
+    transfer.port1 = port1;
+    transfer.port2 = port2;
     return [ port2 ];
   }
 
@@ -1415,10 +1416,15 @@ class ReadableStreamState {
   state = 'readable';
   storedError = undefined;
   controller = undefined;
-  transfer = new ReadableStreamTransferState();
+  // Materialized on transfer, see readableStreamTransferState().
+  transfer = undefined;
 }
 ObjectSetPrototypeOf(ReadableStreamState.prototype, null);
 
+function readableStreamTransferState(stream) {
+  return stream[kState].transfer ??= new ReadableStreamTransferState();
+}
+
 function createReadableStreamState() {
   return new ReadableStreamState();
 }
@@ -2382,13 +2388,16 @@ function readableByteStreamControllerConvertPullIntoDescriptor(desc) {
   if (bytesFilled > byteLength)
     throw new ERR_INVALID_STATE.RangeError('The buffer size is invalid');
   assert(!(bytesFilled % elementSize));
-  const transferredBuffer = ArrayBufferPrototypeTransfer(buffer);
+  // The spec transfers the buffer once more here. Nothing else can reach
+  // it by now: it was transferred when the descriptor was created, and
+  // the BYOB request that exposes it has been invalidated, which always
+  // goes together with a transfer, so the view is built on it as-is.
 
   if (ctor === Buffer) {
-    return Buffer.from(transferredBuffer, byteOffset, bytesFilled / elementSize);
+    return Buffer.from(buffer, byteOffset, bytesFilled / elementSize);
   }
 
-  return new ctor(transferredBuffer, byteOffset, bytesFilled / elementSize);
+  return new ctor(buffer, byteOffset, bytesFilled / elementSize);
 }
 
 function isReadableStreamLocked(stream) {
@@ -2961,7 +2970,7 @@ function readableStreamDefaultControllerError(controller, error) {
 
 function readableStreamDefaultControllerCancelSteps(controller, reason) {
   resetQueue(controller);
-  const result = controller[kState].cancelAlgorithm(reason);
+  const result = controller[kState].cancelAlgorithm(reason, controller);
   readableStreamDefaultControllerClearAlgorithms(controller);
   return result;
 }
@@ -3019,10 +3028,16 @@ function setupReadableStreamDefaultController(
     started: false,
     sizeAlgorithm,
     stream,
+    // The TransformStream this controller is the source of, if any; its
+    // shared source algorithms reach the stream through it.
+    transformStream: undefined,
   };
   stream[kState].controller = controller;
 
   const startResult = startAlgorithm();
+  if (startResult === kDeferredStart)
+    return;
+
   // A non-thenable start result guarantees fulfillment, and no .then
   // lookup on it is observable.
   const startFulfilled = startResult === null ||
@@ -3036,12 +3051,7 @@ function setupReadableStreamDefaultController(
     return;
   }
 
-  const started = () => {
-    controller[kState].started = true;
-    assert(!controller[kState].pulling);
-    assert(!controller[kState].pullAgain);
-    readableStreamDefaultControllerCallPullIfNeeded(controller);
-  };
+  const started = () => readableStreamDefaultControllerStarted(controller);
 
   if (startFulfilled) {
     // The post-start step runs at the exact microtask position the
@@ -3058,6 +3068,15 @@ function setupReadableStreamDefaultController(
     (error) => readableStreamDefaultControllerError(controller, error));
 }
 
+// The post-start steps; transform streams deliver them directly (see
+// transformStreamStart).
+function readableStreamDefaultControllerStarted(controller) {
+  controller[kState].started = true;
+  assert(!controller[kState].pulling);
+  assert(!controller[kState].pullAgain);
+  readableStreamDefaultControllerCallPullIfNeeded(controller);
+}
+
 function setupReadableStreamDefaultControllerFromSource(
   stream,
   source,
@@ -3411,11 +3430,15 @@ function readableByteStreamControllerEnqueue(controller, chunk) {
       );
     }
 
-    readableByteStreamControllerInvalidateBYOBRequest(controller);
-
-    firstPendingPullInto.buffer = ArrayBufferPrototypeTransfer(
-      firstPendingPullInto.buffer,
-    );
+    // The transfer detaches the view the BYOB request exposed; without
+    // a request the buffer is unreachable and there is nothing to
+    // detach.
+    if (controller[kState].byobRequest !== null) {
+      readableByteStreamControllerInvalidateBYOBRequest(controller);
+      firstPendingPullInto.buffer = ArrayBufferPrototypeTransfer(
+        firstPendingPullInto.buffer,
+      );
+    }
 
     if (firstPendingPullInto.type === 'none') {
       readableByteStreamControllerEnqueueDetachedPullIntoToQueue(
@@ -4033,6 +4056,7 @@ module.exports = {
   readableStreamDefaultControllerError,
   readableStreamDefaultControllerCancelSteps,
   readableStreamDefaultControllerPullSteps,
+  readableStreamDefaultControllerStarted,
   setupReadableStreamDefaultController,
   setupReadableStreamDefaultControllerFromSource,
   readableByteStreamControllerClose,