stream.pipeline does not wait for the last stream to flush before calling the final callback

[aravindanve]

• Version: v12.8.0
• Platform: macOS 10.15.4 (Catalina)
• Subsystem: stream

What steps will reproduce the bug?

// reproduce.js

const util = require('util');
const stream = require('stream');

const call = async (fn, ...args) => fn(...args);

const map = (fn) => {
  const tx = new stream.Transform({ objectMode: true });

  tx._transform = (chunk, enc, cb) =>
    call(fn, chunk).then(
      (modified) => cb(null, modified),
      (error) => cb(error),
    );

  return tx;
};

const tap = (fn) => {
  const tx = new stream.Transform({ objectMode: true });

  tx._transform = (chunk, enc, cb) =>
    call(fn, chunk).then(
      () => cb(null, chunk),
      (error) => cb(error),
    );

  return tx;
};

const fork = (...t) => {
  let done;
  let doneError;
  let flush;

  const tx = new stream.Transform({ objectMode: true });
  const pt = new stream.PassThrough({ objectMode: true });

  stream.pipeline(pt, ...t, (error) => {
    done = true;
    doneError = error;
    flush && flush(doneError);
  });

  tx._flush = (cb) => {
    pt.push(null);
    flush = cb;
    done && flush(doneError);
  };

  tx._transform = (chunk, enc, cb) => {
    pt.push(chunk, enc);
    cb(null, chunk);
  };

  return tx;
};

const readableStream = new stream.PassThrough({ objectMode: true });
const pipeline = util.promisify(stream.pipeline);

async function run() {
  await pipeline(
    readableStream,
    fork(
      tap(() => console.log('fork 1: do something with obj for 2s')),
      map((obj) => new Promise((done) => setTimeout(() => done(obj), 2000))),
      tap(() => console.log('fork 1 done!')),
    ),
    fork(
      tap(() => console.log('fork 2: do something with obj for 4s')),
      map((obj) => new Promise((done) => setTimeout(() => done(obj), 4000))),
      tap(() => console.log('fork 2 done!')),
    ),
    fork(
      tap(() => console.log('fork 3: do something with obj for 6s')),
      map((obj) => new Promise((done) => setTimeout(() => done(obj), 6000))),
      tap(() => console.log('fork 3 done!')),
    ),
    // new stream.PassThrough({ objectMode: true }),
    // ^___ adding an extra stream in the pipeline seems to fix the problem
  );
  console.log('done!');
}

run().catch(console.error);

readableStream.push({ name: 'test' });
readableStream.push(null);

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

always

What is the expected behavior?

Console output should look like:

fork 1: do something with obj for 2s
fork 2: do something with obj for 4s
fork 3: do something with obj for 6s
fork 1 done!
fork 2 done!
fork 3 done!
done!

What do you see instead?

Console output actually looks like:

fork 1: do something with obj for 2s
fork 2: do something with obj for 4s
fork 3: do something with obj for 6s
fork 1 done!
fork 2 done!
done!
fork 3 done!

Additional information

As noted in the code above, adding an extra stream at the end seems to mitigate the problem for now.

await pipeline(
    ...
    new stream.PassThrough({ objectMode: true }),
);

console.log('done!);

[rickyes]

Hi, Aravindan. Thanks for your feedback, I am trying to locate this problem, and I reproduced this problem in v14.5.0.

[aravindanve]

@rickyes Just checked, I too get this in both 12.8.0 and 14.5.0. Let me know if you need anything else from me.

[ronag]

are you able to simplify the repro case?

[aravindanve]

@ronag here you go:

const stream = require('stream');

const makeStream = (i) =>
  new stream.Transform({
    transform: (chunk, enc, cb) => cb(null, chunk),
    flush: (cb) =>
      setTimeout(() => {
        console.log('done flushing', i);
        cb(null);
      }),
  });

const input = new stream.Readable();

stream.pipeline(
  input,
  makeStream(1),
  makeStream(2),
  makeStream(3),
  () => console.log('done!'),
);

input.push('test');
input.push(null);

Prints:

done flushing 1
done flushing 2
done!
done flushing 3

[rickyes]

I think I'm getting close to the problem and check again tonight.

[ronag]

Simplest repro:

const stream = require('stream');

const makeStream = (i) =>
  new stream.Transform({
    transform: (chunk, enc, cb) => cb(null, chunk),
    flush: (cb) =>
      setTimeout(() => {
        console.log('done flushing', i);
        cb(null);
      }),
  });

const input = new stream.Readable();
input.push(null);

stream.pipeline(
  input,
  makeStream(1),
  () => console.log('done!'),
);

[ronag]

This is actually an unfortunate case of Transform using 'prefinish' instead of _final to invoke _flush.

I don't think this is fixable without breaking something else. See notes here.

We would have to change Transform to properly use _final to fix this.

@mcollina WDYT?

[ronag]

I think I found a solution.

[oprogramador]

I just reproduced this issue in Node.js 14.15.4 but I see it's fixed in 15.5.1

[oprogramador]

The simplest workaround seems to declare a promise somewhere outside, assign a value in flush to the promise and await the promise in the final callback.

[boutell]

Thanks for reading this, I know it's a little long but I believe it is relevant. Bear with me 🙏

I implemented _final in a PassThrough stream in order to check for a special condition before invoking the callback. As it happens, I'm using the AWS SDK API, which allows me to pass in a readable stream to an upload method, but doesn't simply implement the writable stream interface.

So like many before me I wrapped it in a PassThrough stream. Now I can call await pipeline(someReadableStream, myPassThroughStream).

Except: once the AWS SDK reads the last byte from that passthrough stream, we're "done" as far as pipeline knows. So Node.js prematurely thinks the upload is complete. Boo.

"Aha!" I said. "I'll implement _final in my PassThrough stream and defer invoking the callback until both the superclass implementation of _final and the done method of the AWS Upload class are complete. That way I know that (1) all the input came through and (2) the actual upload also completed."

But... it turns out that tons of operations are still going on even after the pipelines are all 100% convinced the are finished. I couldn't figure it out until I found this thread.

The fix here in Node.js specifically decides not to wait for my _final method before finalizing the stream, and I wind up with lots of work still going on after pipeline:

a65218f

Augh!

Is there just no way to impose an additional requirement upon finalization in a PassThrough stream and have that be respected by pipeline()? Will I have to do something like returning both a stream and a promise, and awaiting the pipeline and then the promise? Or perhaps implementing a writable stream from scratch, whatever that looks like in 2023?

I did try implementing _flush() instead, but trying to pause there just interferes with the uploader's ability to know that it has received all of its input. Catch-22.

Here is my current implementation. Note that settle waits until the outcome of both the superclass _final method and the uploader's done() method are known before invoking the callback of _final. This is the technique I need to replace, somehow, in a way that is compatible with pipeline() and the change that was made to not wait for _final when it has been extended in this way.

Thanks so much for reading!

      const body = new PassThrough();
      const superFinal = body._final;
      let streamError = undefined;
      let awaitError = undefined;
      let streamFinalCallback;

      // We have to wait for both the stream to finish and the upload to finish
      body._final = function(callback) {
        superFinal.call(body, function(err) {
          streamError = err || null;
          streamFinalCallback = callback;
          return settle();
        });
      };
          
      const uploader = new Upload({
        client: self.getClient(uri),
        params: {
          // etc
          Body: body
        }
      });
      (async () => {
        try {
          // Triggers the flow of data
          await uploader.done();
          awaitError = null;
          return settle();
        } catch (err) {
          awaitError = err;
          return settle();
        }
      })();
      // return the passthrough stream for use in pipeline()
      return body;

      function settle() {
        if ((streamError !== undefined) && (awaitError !== undefined)) {
          return streamFinalCallback(streamError || awaitError);
        }
        // Not yet
      }