'stream.push() after EOF' error when using Readable.fromWeb

[kyrylkov]

Version

v18.0.0-rc.1

Platform

Microsoft Windows NT 10.0.22000.0 x64 / Linux 5.10.96-90.460.amzn2022.aarch64

Subsystem

Streams

What steps will reproduce the bug?

1. Get a large file using fetch
2. Convert body to readable stream with Readable.fromWeb
3. Try to save the stream into a file

const { body } = await fetch(url);
const readable = Readable.fromWeb(body);
const filePath = path.join('file.txt');
await pipeline(
  readable,
  fs.createWriteStream(filePath, { flags: 'w' })
);

How often does it reproduce? Is there a required condition?

For some specific files

What is the expected behavior?

Save file to disk

What do you see instead?

stream.push() after EOF and incomplete file on disk

Additional information

No response

[benjamingr]

@kyrylkov any chance you can tell us what files cause this? (large ones? ones with funky permissions?)

[kyrylkov]

@benjamingr No, nothing funky. It's just one client account with similar API requests to download different files and save them to disk. All files work, but one consistently fails with this error. Unfortunately the file is behind private credentials and also contains private info. I can probably check how different its size is from the ones that work just fine.

[swandir]

// error.mjs

import { Readable } from "node:stream";

const response = new Response("Hi");
const readable = Readable.fromWeb(response.body);
readable.pipe(process.stdout);

$ node error.mjs
node:events:515
      throw er; // Unhandled 'error' event
      ^

Error [ERR_STREAM_PUSH_AFTER_EOF]: stream.push() after EOF
    at new NodeError (node:internal/errors:388:5)
    at readableAddChunk (node:internal/streams/readable:285:30)
    at Readable.push (node:internal/streams/readable:234:10)
    at node:internal/webstreams/adapters:482:22
    at process.processTicksAndRejections (node:internal/process/task_queues:95:5)
Emitted 'error' event on Readable instance at:
    at emitErrorNT (node:internal/streams/destroy:151:8)
    at emitErrorCloseNT (node:internal/streams/destroy:116:3)
    at process.processTicksAndRejections (node:internal/process/task_queues:82:21) {
  code: 'ERR_STREAM_PUSH_AFTER_EOF'
}

Node.js v18.4.0

[swandir]

// error.mjs

Interestingly, the error does not occur in CommonJS (.mjs -> .cjs, import -> require)

[davedoesdev]

It's strange, I'm seeing data being pushed even after the reader's closed resolves.

Same applies to the Duplex adapter by the way.

The reader's closed promise shouldn't resolve until all data has been read.

[davedoesdev]

Cause seems to be this:

https://github.com/nodejs/node/blob/main/lib/internal/webstreams/readablestream.js#L2280

Note even if I move this above the conditional before it which does the close, I still get the error, because the close gets done first, due to async nature I guess.

Removing the line then I don't get the error (but of course neither do I get the final chunk).

[davedoesdev]

This patch fixes the issue. I'll work on a test and then submit a PR:

diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js
index ade543c957..b689e79a66 100644
--- a/lib/internal/webstreams/readablestream.js
+++ b/lib/internal/webstreams/readablestream.js
@@ -804,7 +804,7 @@ class ReadableStreamDefaultReader {
           'The reader is not attached to a stream'));
     }
     const readRequest = new DefaultReadRequest();
-    readableStreamDefaultReaderRead(this, readRequest);
+    process.nextTick(() => readableStreamDefaultReaderRead(this, readRequest));
     return readRequest.promise;
   }
 
@@ -2271,13 +2271,13 @@ function readableStreamDefaultControllerPullSteps(controller, readRequest) {
   } = controller[kState];
   if (queue.length) {
     const chunk = dequeueValue(controller);
+    readRequest[kChunk](chunk);
     if (controller[kState].closeRequested && !queue.length) {
       readableStreamDefaultControllerClearAlgorithms(controller);
       readableStreamClose(stream);
     } else {
       readableStreamDefaultControllerCallPullIfNeeded(controller);
     }
-    readRequest[kChunk](chunk);
     return;
   }
   readableStreamAddReadRequest(stream, readRequest);

[davedoesdev]

Hmm, the wpt tests fail. I got it down to a single failure here: https://github.com/nodejs/node/blob/main/test/fixtures/wpt/streams/readable-streams/tee.any.js#L106

with this patch:

diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js
index ade543c957..d405b1dd56 100644
--- a/lib/internal/webstreams/readablestream.js
+++ b/lib/internal/webstreams/readablestream.js
@@ -1866,7 +1866,7 @@ function readableStreamCancel(stream, reason) {
     () => {});
 }
 
-function readableStreamClose(stream) {
+function readableStreamClose(stream, resolveAsync) {
   assert(stream[kState].state === 'readable');
   stream[kState].state = 'closed';
 
@@ -1877,7 +1877,14 @@ function readableStreamClose(stream) {
   if (reader === undefined)
     return;
 
-  reader[kState].close.resolve();
+  const {
+    resolve,
+  } = reader[kState].close;
+
+  if (resolveAsync)
+    process.nextTick(resolve);
+  else
+    resolve();
 
   if (readableStreamHasDefaultReader(stream)) {
     for (let n = 0; n < reader[kState].readRequests.length; n++)
@@ -2271,13 +2278,13 @@ function readableStreamDefaultControllerPullSteps(controller, readRequest) {
   } = controller[kState];
   if (queue.length) {
     const chunk = dequeueValue(controller);
+    readRequest[kChunk](chunk);
     if (controller[kState].closeRequested && !queue.length) {
       readableStreamDefaultControllerClearAlgorithms(controller);
-      readableStreamClose(stream);
+      readableStreamClose(stream, true);
     } else {
       readableStreamDefaultControllerCallPullIfNeeded(controller);
     }
-    readRequest[kChunk](chunk);
     return;
   }

I added this test for replicating the push after EOF issue:

import { mustCall } from '../common/index.mjs';
import assert from 'assert';

const rs = new ReadableStream({
  start(controller) {
    controller.enqueue(new Uint8Array(3));
    controller.close();
  }
});

let data = null;

const reader = rs.getReader();

reader.closed.then(mustCall(() => {
  assert(data, 'data not read yet');
}));

data = await reader.read();
assert(data);
assert(!data.done);
assert.strictEqual(data.value.length, 3);

data = await reader.read();
assert(data);
assert(data.done);
assert(!data.value);

[davedoesdev]

This is a minimal repro:

import { Readable } from 'stream';

Readable.fromWeb(new ReadableStream({
  start(controller) {
    controller.enqueue(new Uint8Array(1));
    controller.close();
  }
})).resume();

I have a fix but since browsers exhibit a similar behaviour (closed resolves before the final data is read) I'm not sure it's going to be accepted. Indeed, the spec https://streams.spec.whatwg.org/#rs-default-controller-private-pull says resolve the closed promise before the final read promise. That isn't what I'd expect to happen but it was written into the spec.

So I think this needs to solved specifically for the Node adapters by removing the code which waits on closed and instead only push null when the final data has been read.

[davedoesdev]

Related issue whatwg/streams#1244 explains that because closed promise is subscribed to first, it'll get called first.