stream.Transform changing order of items

[zuozp8]

Version

v18.14.1

Platform

Linux 703748a51615 5.18.8-051808-generic #202206290850 SMP PREEMPT_DYNAMIC Wed Jun 29 08:59:08 UTC 2022 x86_64 Linux

Subsystem

stream

What steps will reproduce the bug?

const stream = require("node:stream");
const consumers = require("node:stream/consumers");
const createInnerTransfrom = () => new stream.Transform({
    objectMode: true,
    construct(callback) {
        this.push('header from constructor\n');
        callback();
    },
    transform: (row, encoding, callback) => {
        callback(null, JSON.stringify(row) + '\n');
    },
});
const createOuterTransfrom = () => {
    let innerTranfrorm;
    return new stream.Transform({
        objectMode: true,
        transform(row, encoding, callback) {
            if (!innerTranfrorm) {
                innerTranfrorm = createInnerTransfrom();
                innerTranfrorm.on('data', (data) => this.push(data));
                callback();
            }
            else if (innerTranfrorm.write(row)) {
                process.nextTick(callback);
            }
            else {
                innerTranfrorm.once('drain', callback);
            }
        },
    });
};
consumers.text(stream.Readable.from([
    'create InnerTransform',
    'firstLine',
    'secondLine',
]).pipe(createOuterTransfrom())).then((text) => console.log('output:\n', text));

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

always

What is the expected behavior?

output:
 header from constructor
"firstLine"
"secondLine"

What do you see instead?

output:
 header from constructor
"secondLine"
"firstLine"

Additional information

I expect Transform to always process incoming items in order, even with reckless code like in the example.

[rluvaton]

You accidentally wrote the expected in the current behavior

[zuozp8]

You accidentally wrote the expected in the current behavior

I corrected it, thank you

[ronag]

Do you think you can make an even simpler example?

[rluvaton]

What I saw is when you remove the construct it works well

[ronag]

That's a good hint.

[zuozp8]

sometimes it starts/stops occuring when you change some process.nextTick(callback) into callack() or the other way around, so it's some kind of race-condition

I wasn't able to replicate the problem without nested Transforms

[rluvaton]

When you wrap the:

else if (innerTranfrorm.write(row)) {
                process.nextTick(callback);
            } else

like this:

} else if (innerTransform.write('outer | ' + row)) {
          process.nextTick(() => {
            process.nextTick(callback);
          });
        } else {

it solves the problem

[rluvaton]

Found another solution and updated my PR

[mcollina]

This is indeed a bug. Here is a simplified version:

const stream = require("node:stream");
const s = new stream.Transform({
  objectMode: true,
  construct(callback) {
    this.push('header from constructor\n');
    callback();
  },
  transform: (row, encoding, callback) => {
    callback(null, JSON.stringify(row) + '\n');
  },
});

s.pipe(process.stdout);
s.write('firstLine');
process.nextTick(() => s.write('secondLine'));

This fixes the example:

const stream = require("node:stream");
const s = new stream.Transform({
  objectMode: true,
  construct(callback) {
    this.push('header from constructor\n');
    process.nextTick(callback);
  },
  transform: (row, encoding, callback) => {
    callback(null, JSON.stringify(row) + '\n');
  },
});

s.pipe(process.stdout);
s.write('firstLine');
process.nextTick(() => s.write('secondLine'));

So I think we are missing a nextTick in the constructor logic.

[mcollina]

I think https://github.com/nodejs/node/blob/1f75a9513fc829fbd5326a5c9bfa7678b7431eb8/lib/internal/streams/readable.js#LL222C7-L222C7 should be delayed with a nextTick.

[rluvaton]

I think 1f75a95/lib/internal/streams/readable.js#LL222C7-L222C7 should be delayed with a nextTick.

delaying it with nextTick did not worked...

[mcollina]

Given the following script:

const stream = require("node:stream");
const consumers = require("node:stream/consumers");
const createInnerTransfrom = () => new stream.Transform({
    objectMode: true,
    construct(callback) {
        this.push('header from constructor\n');
        process.nextTick(callback);
    },
    transform: (row, encoding, callback) => {
        callback(null, JSON.stringify(row) + '\n');
    },
});
const createOuterTransfrom = () => {
    let innerTranfrorm;
    return new stream.Transform({
        objectMode: true,
        transform(row, encoding, callback) {
            if (!innerTranfrorm) {
                innerTranfrorm = createInnerTransfrom();
                innerTranfrorm.on('data', (data) => this.push(data));
                callback();
            }
            else if (innerTranfrorm.write(row)) {
                process.nextTick(callback);
            }
            else {
                innerTranfrorm.once('drain', callback);
            }
        },
    });
};
consumers.text(stream.Readable.from([
    'create InnerTransform',
    'firstLine',
    'secondLine',
]).pipe(createOuterTransfrom())).then((text) => console.log('output:\n', text));

It will error:

node:internal/process/promises:288
            triggerUncaughtException(err, true /* fromPromise */);
            ^

Error [ERR_STREAM_PUSH_AFTER_EOF]: stream.push() after EOF
    at new NodeError (node:internal/errors:399:5)
    at readableAddChunk (node:internal/streams/readable:285:30)
    at Readable.push (node:internal/streams/readable:234:10)
    at Transform.<anonymous> (/Users/matteo/tmp/aaa.js:20:58)
    at Transform.emit (node:events:513:28)
    at addChunk (node:internal/streams/readable:324:12)
    at readableAddChunk (node:internal/streams/readable:297:9)
    at Readable.push (node:internal/streams/readable:234:10)
    at node:internal/streams/transform:182:12
    at Transform.transform [as _transform] (/Users/matteo/tmp/aaa.js:10:9) {
  code: 'ERR_STREAM_PUSH_AFTER_EOF'
}

This is correct, as you are pushing things after the stream ended.

[rluvaton]

I tested with your simplified version

[mcollina]

Here is the fix: #46818