stream: fix handling generators in Readable.from

[vadzim]

Checklist

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

closes #32842

[ronag]

Good spot. I'm not sure needToClose is required, just always call return() in _destroy(). Also, please add a test.

[mcollina]

Would you mind to add a unit test for this? Thanks

[vadzim]

@ronag @mcollina sure, I'll add a test as soon as it builds on my machine.

[vadzim]

I'm not sure needToClose is required, just always call return() in _destroy()

for-of loop does this check. Why we shouldn't?

[ronag]

Here is a "desugared" pseudo version of for-of I got from Gus Caplan:

const iterator = B[Symbol.asyncIterator]();
while (true) {
  const next = await iterator.next();
  if (next.done) {
    break;
  }
  const A = next.value;

  // pseudocode follows, because dealing with `break` and `throw` is difficult to desugar.
  const result = BODY;
  if (result is a break or result is a throw) {
    if (iterator.return !== undefined) {
      let innerResult = iterator.return();
      if (innerResult is not a throw) {
        // if iterator.return() threw an exception, we don't want to await it.
        innerResult = await innerResult;
      }
      if (result is a throw) {
        // if BODY threw an exception, we want to throw that one instead of the possible
        // exception from iterator.return()
        throw result;
      }
      if (innerResult is a throw) {
        // if iterator.return() threw, either directly or because it returned a rejected promise,
        // now we re-throw that exception.
        throw innerResult;
      }
      break;
    }
  }
}

I think that's probably what we should aim for.

As you've noticed the throw + return() part is tricky.

[ronag]

for-of loop does this check.

Where do you see this? Maybe?

[vadzim]

I'm sorry, it's quite difficult for me to understand that pseudocode.
The next is what I got from the node cli:

// next returns ordinary value, done = false:
for(const x of {
  [Symbol.iterator](){ return this },
  next(){ console.log('next'); return { value: 42, done: false } },
  return(){ console.log('return'); return { done: true } }
}) {
  console.log(x)
  break
}
// output:
//    next
//    42
//    return
//    undefined

// next returns done = true:
for(const x of {
  [Symbol.iterator](){ return this },
  next(){ console.log('next'); return { value: 42, done: true } },
  return(){ console.log('return'); return { done: true } }
}) {
  console.log(x)
  break
}
// output:
//    next
//    undefined

// next throws
for(const x of {
  [Symbol.iterator](){ return this },
  next(){ console.log('next'); throw 42 },
  return(){ console.log('return'); return { done: true } }
}) {
  console.log(x)
  break
}
// output:
//    next
//    Uncaught 42

It seems that return isn't called if next throws.

[ronag]

The next is what I got from the node cli:

What do you want to illustrate with this?

It seems that return isn't called if next throws.

Yes, that's what we are fixing in this PR?

[vadzim]

What do you want to illustrate with this?

I mean

It seems that return isn't called if next throws.

[ronag]

Ah, I think I understand what you mean.

[ronag]

What happens if you do the same thing with async iterators?

[ronag]

It seems that return isn't called if next throws.

But then the problem with #32842 remains?

I think looking at the for await..of would be relevant.

[vadzim]

What happens if you do the same thing with async iterators?

Same thing happens if we use for-await-of with async iterator.
.return isn't called if .next throws or returns {done: true}.

[vadzim]

Giving this more thought. I think the absolute simplest and most correct implementation is to just simply use a for..of.

We still need to deal with buffer in readable stream.
After readable.push returns false we should wait for the next read call.
We just need _destroy implementation.

[ronag]

LGTM once tests are added, good job!

[vadzim]

I'll write the tests, though it can take some time cause I'm gonna cover all these mentioned edge cases.

[ronag]

Also, please take a look at the commit guidelines.

You should use stream: as subsystem, and Fixes: before the url for the issue it fixed.

[vadzim]

Sure, I'll squash my draft commits into one when the job is done.

[vadzim]

I've added tests and it's allowed me to catch one more case.
But I'm afraid those tests are too complex to support and now I like your proposal to use for-of cycle. I'm gonna to prepare another PR to compare.

[vadzim]

Actually the code does not look simpler) and the tests should remain in any case.
But you can choose either #32855

[ronag]

LGTM

[nodejs-github-bot]

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

[mcollina]

lgtm

[nodejs-github-bot]

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

[vadzim]

@ronag @mcollina
Thank you for the great review. It's really valueble for my experience.

Btw, actually the code didn't contain any iterator.return() call, so it didn't close the iterator when not of its values were consumed. E.g. when the stream is destroyed.

I guess this bugfix could be applied to node@12. Does it make sense?

[mcollina]

Absolutely, in due time. It will be included in the closest release of 12.x after 2 weeks of being in 13.x.

[ronag]

Landed in 8a3fa32

[benjamingr]

Hi, I don't understand why needToClose is here? If someone destroys the stream we need to always close the generator don't we?

[benjamingr]

That is - if someone creates a stream, calls .read but .next is never called on the generator - needToClose is false and blocks don't run (since await iterator.next()` is async and waits a microtick)