aborting stream using the signal does not close the stream when the from generator never finish

[rluvaton]

Version

v20.0.0-pre (master)

Platform

Darwin MBP 22.3.0 Darwin Kernel Version 22.3.0: Mon Jan 30 20:38:37 PST 2023; root:xnu-8792.81.3~2/RELEASE_ARM64_T6000 arm64

Subsystem

No response

What steps will reproduce the bug?

This fails as the catch is not called

{
  let reached = false;

  const ac = new AbortController();
  let resolve;
  const promise = new Promise((res) => resolve = res);

  const stream = from((async function *() {
    yield 1;
    await promise;
    reached = true;
  })());

  setTimeout(() => ac.abort(), 1000);
  addAbortSignal(ac.signal, stream);
  stream
    .toArray({ signal: ac.signal })
    .then(common.mustNotCall())
    .catch(common.mustCall((e) => {
      strictEqual(e.name, 'AbortError')
      strictEqual(reached, false);
    }));
}

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

always

What is the expected behavior? Why is that the expected behavior?

to close the stream

What do you see instead?

error:

Mismatched <anonymous> function calls. Expected exactly 1, actual 0.
    at Proxy.mustCall (/open-source/node/node-fork/test/common/index.js:407:10)
    at Object.<anonymous> (/open-source/node/node-fork/test/parallel/test-stream-drop-take.js:30:19)
    at Module._compile (node:internal/modules/cjs/loader:1287:14)
    at Module._extensions..js (node:internal/modules/cjs/loader:1341:10)
    at Module.load (node:internal/modules/cjs/loader:1145:32)
    at Module._load (node:internal/modules/cjs/loader:984:12)
    at Function.executeUserEntryPoint [as runMain] (node:internal/modules/run_main:83:12)
    at node:internal/main/run_main_module:23:47

Additional information

If I omit the ``addAbortSignal(ac.signal, stream);, the stream.destroy` function is not even called

But when I add the abort signal (addAbortSignal(ac.signal, stream);) to the original stream it does get called but still, the test fails.

this is because we get to this line:

node/lib/internal/streams/from.js

Line 69 in 3a648af

const { value, done } = await iterator.throw(error);

but not to this:

node/lib/internal/streams/from.js

Line 70 in 3a648af

await value;

[rluvaton]

I see that this happens with async generator/iterator when next but not awaited and then iterator.throw (which is basically what happens in the stream core)

// Keep process open
setInterval(() => {
}, 10000);

let resolve;
const promise = new Promise(res => resolve = res);
async function* neverEnding() {
  yield 1;
  await promise;

  yield 2;
  yield 3;
}

async function run() {
  const iterator = neverEnding();

  console.log(await iterator.next());
  const res = iterator.next();
  console.log(await iterator.throw(new Error('throw error')));

  console.log('nothing here')
}

run()
  .then(result => console.log('success', result))
  .catch(error => console.log('failure', error))

what get logged is:

{ value: 1, done: false }

is there something we can do to fix this? or this is JavaScript engine (V8) internal bug?

[rluvaton]

Looking at the Generator GeneratorResumeAbrupt specs (that was called from Generator.prototype.throw spec) it's not counting for executing [[GeneratorState]] which is the current generator state when we call iterator.throw

In the debugger the [[GeneratorState]] is actually suspended which is weird because it's not one of the possible values which are undefined, suspendedStart, suspendedYield, executing, or completed

[ljharb]

https://tc39.es/ecma262/#sec-properties-of-asyncgenerator-intances are the possible values, and GeneratorState is a slot on sync generators, not async generators.

[rluvaton]

Thanks @ljharb, so the example of the async generator not throwing is the expected behavior?

[ljharb]

I'm not sure what the expected behavior is here tbh.

However, by aborting after 10ms, you're basically making a race condition - I think there's no guarantee that stream.toArray({ signal: ac.signal }) will settle before the 10ms have elapsed.

[rluvaton]

However, by aborting after 10ms, you're basically making a race condition - I think there's no guarantee that stream.toArray({ signal: ac.signal }) will settle before the 10ms have elapsed.

the 10ms is just an example it can be 1s as well and the same result, toArray will never finish

I'm not sure what the expected behavior is here tbh.

Lets say I have this use case:
Taking the first 10 chunks the user wrote, if not written 10 chunks in 10 seconds aborting

code:

const ac = new AbortController();

setTimeout(() => {
  console.log('10s passed')
  ac.abort();
}, 10000);

process.stdin
  .map(item => {
    console.log('item', item)
    return item.toString();
  }, {signal: ac.signal})
  .take(10, {signal: ac.signal})
  .toArray({signal: ac.signal})
  .then(arr => {
    console.log('success', arr);
  })
  .catch((e) => {
    console.error('Aborted', e);
  });

In the current behavior, if you don't type anything and 10s pass, the Aborted log won't be logged until you type something

I don't wanna wait for user input when I close the stream, I expect that when I call ac.abort() it will abort immediately (can be in the next tick of course)

[ljharb]

Your then and catch, though, don't take a signal because Promises aren't cancellable.

[rluvaton]

This is why I don't pass them a signal

[jagadeeshmeesala]

can I work on this issue?

[rluvaton]

The problem is if this is even possible, After talking to @giltayar it seems like this can't be fixed as this is how async generator work

WDYT @benjamingr

[benjamingr]

I was OOO for a while and I'll be abroad until after Passover so my capacity is limited (hence why I didn't weigh in on the transform issue even though I still think the fix is unintuitive).

This just looks like a bug to me where the error is not being propagated correctly, when the stream is destroyed catch should be called with an AbortError even if the generator is being thrown into.

[benjamingr]

If you remove addAbortSignal the behavior is correct because the stream in fact never finishes