lib: fix blob.stream() causing hanging promises

Refs: #47993 (comment) PR-URL: #48232 Reviewed-By: James M Snell <[email protected]> Reviewed-By: Matteo Collina <[email protected]> Reviewed-By: Benjamin Gruenbaum <[email protected]>

Author: debadree25

lib/internal/blob.js

@@ -329,34 +329,45 @@ class Blob{
 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;
-}
-constpending=this.pendingPulls.shift();
-if(status===0){
-// EOS
-c.close();
-pending.resolve();
-return;
-}elseif(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.
-consterror=lazyDOMException('The blob could not be read','NotReadableError');
-
-c.error(error);
-pending.reject(error);
-return;
-}
-if(buffer!==undefined){
-c.enqueue(newUint8Array(buffer));
-}
-pending.resolve();
-});
+constreadNext=()=>{
+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;
+}
+if(status===0){
+// EOS
+c.close();
+constpending=this.pendingPulls.shift();
+pending.resolve();
+return;
+}elseif(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.
+consterror=lazyDOMException('The blob could not be read','NotReadableError');
+constpending=this.pendingPulls.shift();
+c.error(error);
+pending.reject(error);
+return;
+}
+if(buffer!==undefined){
+c.enqueue(newUint8Array(buffer));
+}
+// We keep reading until we either reach EOS, some error, or we
+// hit the flow rate of the stream (c.desiredSize).
+queueMicrotask(()=>{
+if(c.desiredSize<=0){
+// A manual backpressure check.
+return;
+}
+readNext();
+});
+});
+};
+readNext();
 returnpromise;
 },
 cancel(reason){

test/parallel/test-blob.js

@@ -1,11 +1,12 @@
-// Flags: --no-warnings
+// Flags: --no-warnings --expose-internals
 'use strict';
 
 constcommon=require('../common');
 constassert=require('assert');
 const{ Blob }=require('buffer');
 const{ inspect }=require('util');
 const{EOL}=require('os');
+const{ kState }=require('internal/webstreams/util');
 
 {
 constb=newBlob();
@@ -237,6 +238,50 @@ assert.throws(() => new Blob({}),{
 assert(res.done);
 })().then(common.mustCall());
 
+(async()=>{
+constb=newBlob(Array(10).fill('hello'));
+constreader=b.stream().getReader();
+constchunks=[];
+while(true){
+constres=awaitreader.read();
+if(res.done)break;
+assert.strictEqual(res.value.byteLength,5);
+chunks.push(res.value);
+}
+assert.strictEqual(chunks.length,10);
+})().then(common.mustCall());
+
+(async()=>{
+constb=newBlob(Array(10).fill('hello'));
+constreader=b.stream().getReader();
+constchunks=[];
+while(true){
+constres=awaitreader.read();
+if(chunks.length===5){
+reader.cancel('boom');
+break;
+}
+if(res.done)break;
+assert.strictEqual(res.value.byteLength,5);
+chunks.push(res.value);
+}
+assert.strictEqual(chunks.length,5);
+reader.closed.then(common.mustCall());
+})().then(common.mustCall());
+
+(async()=>{
+constb=newBlob(Array(10).fill('hello'));
+conststream=b.stream();
+constreader=stream.getReader();
+assert.strictEqual(stream[kState].controller.desiredSize,1);
+const{ value, done }=awaitreader.read();
+assert.strictEqual(value.byteLength,5);
+assert(!done);
+setTimeout(()=>{
+assert.strictEqual(stream[kState].controller.desiredSize,0);
+},0);
+})().then(common.mustCall());
+
 {
 constb=newBlob(['hello\n'],{endings: 'native'});
 assert.strictEqual(b.size,EOL.length+5);