Different behavior between transform function and async generator using pipelines

[Xstoudi]

• Version: v13.11.0
• Platform: Windows 10 64-bit
• Subsystem: stream

What steps will reproduce the bug?

index.mjs:

import { pipeline, PassThrough } from 'stream'

async function* asyncGen(source) {
  for await (const chunk of source) {
    yield chunk;
  }
}

pipeline(process.stdin, asyncGen, process.stdout, err => {
  console.error('end')
})

then echo "hello" | node index.mjs ends up like this:

"hello"
end

but if you replace asyncGen in the pipeline by new PassThrough(), it ends up like this:

"hello"

As you can see, the callback is never called.

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

What is the expected behavior?

"hello"
end

What do you see instead?

"hello"

Additional information

I don't really know if it's an expected behavior or a bug to be honest. As I can't find anything related in the documentation, it feels like a bug.

[targos]

@nodejs/streams

[ronag]

Minimal repro on master.

'use strict';

const common = require('../common');
const assert = require('assert');
const { pipeline, PassThrough } = require('stream');
const cp = require('child_process');

if (process.argv[2] === 'child') {
  pipeline(
    process.stdin,
    new PassThrough(),
    process.stdout,
    common.mustCall((err) => {
      console.error('end');
    })
  );
  return;
}

cp.exec([
  'echo',
  '"hello"',
  '|',
  `"${process.execPath}"`,
  `"${__filename}"`,
  'child'
].join(' '), common.mustCall((err, stdout, stderr) => {
  console.log(err)
  assert.ifError(err);
  console.log(stdout);
}));

[ronag]

Seems like stdout is never finished, even though the passthrough is writableEnded.

[ronag]

So this fails:

  pipeline(
    process.stdin,
    new PassThrough(),
    process.stdout,
    common.mustCall((err) => {
      console.error('end');
    })
  );

while this doesn't:

  pipeline(
    process.stdin,
    p.on('end', () => {
      process.stdout.end()
    }),
    process.stdout,
    common.mustCall((err) => {
      console.error('end');
    })
  );

[ronag]

@addaleax This does not seem related to pipeline or PassThrough it's seems to be something with process.stdout. This fails as well:

'use strict';

const common = require('../common');
const assert = require('assert');

if (process.argv[2] === 'child') {
  process.stdin
    .pipe(process.stdout)
    .on('finish', common.mustCall(() => {
      console.log('finish');
    }));
} else {
  const cp = require('child_process');

  cp.exec([
    'echo',
    '"hello"',
    '|',
    `"${process.execPath}"`,
    `"${__filename}"`,
    'child'
  ].join(' '), common.mustCall((err, stdout, stderr) => {
    assert.ifError(err);
  }));
}

[ronag]

The reason this probably works with pipeline + generator is because that path does not use pipe.

Maybe pipeline should prefer the 'readable' API when possible over pipe, would that provide better compatibility? @mcollina

[ronag]

I believe this is the culprit https://github.com/nodejs/node/blob/master/lib/_stream_readable.js#L654. Unfortunately I don't understand this part at the moment.

Basically pipe won't end() the dest if it is process.stdout. Not sure why that is the case. If there is a good reason for that, then maybe pipeline + generators shouldn't do it either?

[vweevers]

If there is a good reason for that

#7606 (comment)

[ronag]

@vweevers: Does that mean

then maybe pipeline + generators shouldn't do it either?

Do we need a special case here in pipeline to complete once the stream before stdout has emitted 'end'?

[vweevers]

Not sure we can do that? Before completing, we'd want to make sure the writes to stdout have been flushed, but stdout can be synchronous or asynchronous depending on platform and destination.

Alternative point of view: stdout is documented as being different from regular streams. The fact that stdout cannot be closed implies that a pipeline to stdout cannot complete. That's unfortunate, but logically solid.

[ronag]

If there is a good reason for that

#7606 (comment)

It is a bit random though... since this only applies for the pipe protocol. We have 3 other "read" protocols where this does currently not apply.

The fact that stdout cannot be closed

It can be closed, it's just that pipe doesn't close it.

[ronag]

Alternative point of view: stdout is documented as being different from regular streams.

I can see that, but I can't see the documentation mentioning this particular thing? Should we add it?