stream: do not swallow errors with async iterators and pipeline

[mcollina]

Before this patch, pipeline() could swallow errors by pre-emptively
producing a ERR_STREAM_PREMATURE_CLOSE that was not really helpful
to the user.

Checklist

• make -j4 test (UNIX), or vcbuild test (Windows) passes
• tests and/or benchmarks are included
• commit message follows commit guidelines

[nodejs-github-bot]

CI: https://ci.nodejs.org/job/node-test-pull-request/29494/

[ronag]

Something else is going on here. I don't understand how this can cause premature close. It seems that the catch clause in pump is never reached which I find strange.

The test can be further simplified to:

{
  const rs = new Readable({
    read() {
      setImmediate(() => {
        rs.push('hello');
      });
    }
  });
  pipeline(rs, async function*(stream) {
    /* eslint no-unused-vars: off */
    for await (const chunk of stream) {
      throw new Error('kaboom');
    }
  }, common.mustCall((err) => {
    assert.strictEqual(err.message, 'kaboom');
  }));
}

EDIT:

The throw inside the iterable seems to call iterator.return instead of iterator.throw which I find strange, but maybe that's the way it's supposed to work?

    for await (const chunk of stream) {
      throw new Error('kaboom');
    }

this will call iterator.return() on the stream iterator which in turn will call stream.destroy() and not propagate the error directly. I might need to read up on the iterator protocol. I would have assumed that it would call iterator.throw(err).

[mcollina]

I thought the same as well, and I do not understand why the catch call is never reached. I was not be able to reproduce the problem with just our async iterator code.

The “bug” comes from the interaction between all the streams and the async iterator.

Not that for an async iterator in pipeline we have 3 finished() calls that gets into play (1 in the main async iterator, 1 inside the finish function called from return, and one inside pipeline).

The internal error is swallowed because the iterator ends abruptly and it causes the other streams to close. Those premature closes are forwarded earlier than the error from the internal async iterator. At least, that’s my read of the problem.

Let me know if you can crack this :/.

[mcollina]

The throw inside the iterable seems to call iterator.return instead of iterator.throw which I find strange, but maybe that's the way it's supposed to work?

That’s what I discovered as well. I think it’s how it works.

[ronag]

Sorry, the close was a misclick.

I don't like this solution but I can't really think of anything better...

Let me know if you can crack this :/.

If this is not an urgent issue please give me a few days to try to crack it before landing this.

[devsnek]

The throw inside the iterable seems to call iterator.return instead of iterator.throw which I find strange, but maybe that's the way it's supposed to work?

I can confirm this is the correct behaviour (I have absolutely no idea why though, I'd have to dig into old tc39 notes)

[ronag]

@devsnek Do you have any advice here? We had a short mail conversation in regards to a related topic some time ago.

In particular, I'm curious why the iterator protocol would not call iterator.throw(err) when an error occurs inside the for block. Is there anyway to make it so, e.g. with some kind of hack using yield*? See the later part of this comment.

EDIT: wow, nice timing @devsnek

[devsnek]

Ok this is basically a difference in the node stream model and the async iterator model. In node.js, the streams own errors that happen while they are being consumed (emit('data') will throw if one of the handlers throws), but in async iterators, they do not.

[bakkot]

To expand a bit:

My mental model of GeneratorPrototype.throw is that it is for cases where the generator needs a value to continue execution which the consumer is unable to supply because it has hit an error while attempting to produce that value. It is not for cases where the consumer hits an error while just passively consuming the generator's output; that's the consumer's problem.

[mcollina]

@ronag this should be a better fix. The problem we were facing was a race between the ERR_STREAM_PREMATURE_CLOSE of the last stream and the error produced by the async iteration. I've solved the bug by having the callback called when all streams have been successfully destroyed and just not the last.

[ronag]

Nice! A few questions though.

Why not just set finishCount to the number of calls to wrap? That way you only need to decrement it in the destroyer callback.

[mcollina]

CI: https://ci.nodejs.org/job/node-test-pull-request/2​96​79/

it changed too much since last review, PTAL.

[mcollina]

@ronag @jasnell @addaleax @benjamingr PTAL.

[nodejs-github-bot]

CI: https://ci.nodejs.org/job/node-test-pull-request/29679/

[ronag]

Would prefer to keep final in destroyer

[nodejs-github-bot]

CI: https://ci.nodejs.org/job/node-test-pull-request/29711/

[nodejs-github-bot]

CI: https://ci.nodejs.org/job/node-test-pull-request/29729/

[mcollina]

Landed as d7fe554