stream: handle a pending pull request from a released reader

In order to meet the specification, this includes mainly the followings:
- Adding the 'release steps' to ReadableStreamController
- Responding to a pull request from a released reader in
ReadableByteStreamController
Signed-off-by: Daeyeon Jeong daeyeon.dev@gmail.com
PR-URL: #44702
Refs: https://streams.spec.whatwg.org/#abstract-opdef-readablebytestreamcontroller-releasesteps
Refs: https://streams.spec.whatwg.org/#readable-byte-stream-controller-respond-in-readable-state
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Minwoo Jung <nodecorelab@gmail.com>

Author: daeyeon

doc/api/webstreams.md

@@ -651,6 +651,10 @@ Signals an error that causes the {ReadableStream} to error and close.
 
 <!-- YAML
 added: v16.5.0
+changes:
+ - version: REPLACEME
+ pr-url: https://github.com/nodejs/node/pull/44702
+ description: Support handling a BYOB pull request from a released reader.
 -->
 
 Every {ReadableStream} has a controller that is responsible for

lib/internal/webstreams/readablestream.js

@@ -139,6 +139,7 @@ const kClose = Symbol('kClose');
 constkChunk=Symbol('kChunk');
 constkError=Symbol('kError');
 constkPull=Symbol('kPull');
+constkRelease=Symbol('kRelease');
 
 /**
  * @typedef {import('../abort_controller').AbortSignal} AbortSignal
@@ -1019,6 +1020,8 @@ class ReadableStreamDefaultController {
 readableStreamDefaultControllerPullSteps(this,readRequest);
 }
 
+[kRelease](){}
+
 [kInspect](depth,options){
 returncustomInspect(depth,options,this[kType],{});
 }
@@ -1143,6 +1146,17 @@ class ReadableByteStreamController {
 readableByteStreamControllerPullSteps(this,readRequest);
 }
 
+[kRelease](){
+const{
+ pendingPullIntos,
+}=this[kState];
+if(pendingPullIntos.length>0){
+constfirstPendingPullInto=pendingPullIntos[0];
+firstPendingPullInto.type='none';
+this[kState].pendingPullIntos=[firstPendingPullInto];
+}
+}
+
 [kInspect](depth,options){
 returncustomInspect(depth,options,this[kType],{});
 }
@@ -2060,6 +2074,9 @@ function readableStreamReaderGenericRelease(reader) {
 };
 }
 setPromiseHandled(reader[kState].close.promise);
+
+stream[kState].controller[kRelease]();
+
 stream[kState].reader=undefined;
 reader[kState].stream=undefined;
 }
@@ -2365,6 +2382,8 @@ function readableByteStreamControllerClose(controller) {
 
 functionreadableByteStreamControllerCommitPullIntoDescriptor(stream,desc){
 assert(stream[kState].state!=='errored');
+assert(desc.type!=='none');
+
 letdone=false;
 if(stream[kState].state==='closed'){
 desc.bytesFilled=0;
@@ -2574,6 +2593,9 @@ function readableByteStreamControllerRespond(controller, bytesWritten) {
 
 functionreadableByteStreamControllerRespondInClosedState(controller,desc){
 assert(!desc.bytesFilled);
+if(desc.type==='none'){
+readableByteStreamControllerShiftPendingPullInto(controller);
+}
 const{
  stream,
 }=controller[kState];
@@ -2663,6 +2685,31 @@ function readableByteStreamControllerEnqueue(controller, chunk) {
 readableByteStreamControllerCallPullIfNeeded(controller);
 }
 
+functionreadableByteStreamControllerEnqueueClonedChunkToQueue(
+controller,
+buffer,
+byteOffset,
+byteLength
+){
+letcloneResult;
+try{
+cloneResult=ArrayBufferPrototypeSlice(
+buffer,
+byteOffset,
+byteOffset+byteLength
+);
+}catch(error){
+readableByteStreamControllerError(controller,error);
+throwerror;
+}
+readableByteStreamControllerEnqueueChunkToQueue(
+controller,
+cloneResult,
+0,
+byteLength
+);
+}
+
 functionreadableByteStreamControllerEnqueueChunkToQueue(
 controller,
 buffer,
@@ -2678,6 +2725,29 @@ function readableByteStreamControllerEnqueueChunkToQueue(
 controller[kState].queueTotalSize+=byteLength;
 }
 
+functionreadableByteStreamControllerEnqueueDetachedPullIntoToQueue(
+controller,
+desc
+){
+const{
+ buffer,
+ byteOffset,
+ bytesFilled,
+ type,
+}=desc;
+assert(type==='none');
+
+if(bytesFilled>0){
+readableByteStreamControllerEnqueueClonedChunkToQueue(
+controller,
+buffer,
+byteOffset,
+bytesFilled
+);
+}
+readableByteStreamControllerShiftPendingPullInto(controller);
+}
+
 functionreadableByteStreamControllerFillPullIntoDescriptorFromQueue(
 controller,
 desc){
@@ -2773,6 +2843,7 @@ function readableByteStreamControllerRespondInReadableState(
  buffer,
  bytesFilled,
  byteLength,
+ type,
 }=desc;
 
 if(bytesFilled+bytesWritten>byteLength)
@@ -2783,6 +2854,17 @@ function readableByteStreamControllerRespondInReadableState(
 bytesWritten,
 desc);
 
+if(type==='none'){
+readableByteStreamControllerEnqueueDetachedPullIntoToQueue(
+controller,
+desc
+);
+readableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(
+controller
+);
+return;
+}
+
 if(desc.bytesFilled<desc.elementSize)
 return;
 

test/wpt/status/streams.json

@@ -16,17 +16,9 @@
 "fail": {
 "expected": [
 "ReadableStream with byte source: enqueue() discards auto-allocated BYOB request",
-"ReadableStream with byte source: releaseLock() with pending read(view), read(view) on second reader, respond()",
-"ReadableStream with byte source: releaseLock() with pending read(view), read(view) on second reader with 1 element Uint16Array, respond(1)",
-"ReadableStream with byte source: releaseLock() with pending read(view), read(view) on second reader with 2 element Uint8Array, respond(3)",
-"ReadableStream with byte source: releaseLock() with pending read(view), read(view) on second reader, respondWithNewView()",
 "ReadableStream with byte source: releaseLock() with pending read(view), read(view) on second reader, enqueue()",
-"ReadableStream with byte source: releaseLock() with pending read(view), read(view) on second reader, close(), respond(0)",
-"ReadableStream with byte source: autoAllocateChunkSize, releaseLock() with pending read(), read() on second reader, respond()",
 "ReadableStream with byte source: autoAllocateChunkSize, releaseLock() with pending read(), read() on second reader, enqueue()",
-"ReadableStream with byte source: autoAllocateChunkSize, releaseLock() with pending read(), read(view) on second reader, respond()",
 "ReadableStream with byte source: autoAllocateChunkSize, releaseLock() with pending read(), read(view) on second reader, enqueue()",
-"ReadableStream with byte source: read(view) with 1 element Uint16Array, respond(1), releaseLock(), read(view) on second reader with 1 element Uint16Array, respond(1)",
 "ReadableStream with byte source: read(view) with 1 element Uint16Array, respond(1), releaseLock(), read() on second reader, enqueue()"
  ]
  }