lib: fix blob.stream() causing hanging promises

Refs: nodejs#47993 (comment)
debadree25 · May 28, 2023 · 24ef7d6 · 24ef7d6
1 parent d81151e
commit 24ef7d6
Show file tree

Hide file tree

Showing 2 changed files with 63 additions and 28 deletions.
diff --git a/lib/internal/blob.js b/lib/internal/blob.js
@@ -329,34 +329,38 @@ 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;
-          }
-          const pending = this.pendingPulls.shift();
-          if (status === 0) {
-            // EOS
-            c.close();
-            pending.resolve();
-            return;
-          } else if (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.
-            const error = lazyDOMException('The blob could not be read', 'NotReadableError');
-
-            c.error(error);
-            pending.reject(error);
-            return;
-          }
-          if (buffer !== undefined) {
-            c.enqueue(new Uint8Array(buffer));
-          }
-          pending.resolve();
-        });
+        const readNext = () => {
+          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();
+              const pending = this.pendingPulls.shift();
+              pending.resolve();
+              return;
+            } else if (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.
+              const error = lazyDOMException('The blob could not be read', 'NotReadableError');
+              const pending = this.pendingPulls.shift();
+              c.error(error);
+              pending.reject(error);
+              return;
+            }
+            if (buffer !== undefined) {
+              c.enqueue(new Uint8Array(buffer));
+            }
+            // We keep reading until we either reach EOS or some error
+            queueMicrotask(() => readNext());
+          });
+        };
+        readNext();
         return promise;
       },
       cancel(reason) {

diff --git a/test/parallel/test-blob.js b/test/parallel/test-blob.js
@@ -237,6 +237,37 @@ assert.throws(() => new Blob({}), {
   assert(res.done);
 })().then(common.mustCall());
 
+(async () => {
+  const b = new Blob(Array(10).fill('hello'));
+  const reader = b.stream().getReader();
+  const chunks = [];
+  while (true) {
+    const res = await reader.read();
+    if (res.done) break;
+    assert.strictEqual(res.value.byteLength, 5);
+    chunks.push(res.value);
+  }
+  assert.strictEqual(chunks.length, 10);
+})().then(common.mustCall());
+
+(async () => {
+  const b = new Blob(Array(10).fill('hello'));
+  const reader = b.stream().getReader();
+  const chunks = [];
+  while (true) {
+    const res = await reader.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());
+
 {
   const b = new Blob(['hello\n'], { endings: 'native' });
   assert.strictEqual(b.size, EOL.length + 5);