Duplex.from({ writable, readable }) breaks on backpressure

[pavelhoral]

Version

v18.10.0

Platform

Microsoft Windows NT 10.0.19044.0 x64

Subsystem

No response

What steps will reproduce the bug?

I have created repository with failing test case - https://github.com/pavelhoral/node-duplexify-issue that can be cloned and run:

import { Duplex, PassThrough, Readable, Transform } from 'node:stream';

describe('Duplex.from({ writable, readable })', () => {

  it('flushes stream after filling buffer', async () => {
    // Simple pass-through as a placeholder for more complex setup
    const through = new PassThrough({ objectMode: true });

    // Stream prepared values, pipe through simple duplex and async transformer for backpressure
    const stream = Readable.from(['foo', 'bar'], { objectMode: true })
      .pipe(Duplex.from({
        writable: through,
        readable: through
      }))
      .pipe(new Transform({
        objectMode: true,
        highWaterMark: 1, // Setting 1 to force backpressure after a single item
        transform(chunk, encoding, callback) {
          setTimeout(() => callback(null, chunk), 0);
        }
      }));

    // This never finishes when high water mark is reached
    const result = await stream.toArray();

    expect(result).toEqual(['foo', 'bar']);
  });

});

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

The issue happens every time in my test case when the internal buffers are full and read must be paused (I guess this condition must happen - https://github.com/nodejs/node/blob/v18.10.0/lib/internal/streams/duplexify.js#L350).

What is the expected behavior?

Stream should correctly finish processing after read buffers are free again.

What do you see instead?

Duplex stream never correctly resumes its operation after the read has to pause.

Additional information

No response

[pavelhoral]

Maybe a better issue title would be Duplex.from({ writable, readable }) breaks on backpressure?

[xtx1130]

In objectMode, highWaterMark default set to 16, please see: https://github.com/nodejs/node/blob/v18.10.0/lib/internal/streams/state.js#L15-L17

The stream will temporarily stop reading data from the underlying resource, and toArray method is an async function, so nodejs will consider the promise is not resolved yet, and exit with code 13. You can set:

new Transform({
        objectMode: true,
        emitClose: false,
        highWaterMark:100,
        ...

to avoid this.

[rluvaton]

@xtx1130 this sounds like a bug, no?

[bnoordhuis]

You're calling await stream.toArray() inside an async function - i.e., a promise - but what awaits that promise? Try excluding vitest and using only built-in modules.

[pavelhoral]

I think you are missing the point. It is not about a specific highWaterMark number, it is about resuming the stream after the buffer limit is reached (btw. I have updated the test in the original comment).

And vitest and toArray here are also irrelevant - the stream never correctly resumes after backpressure occurs. That is the reason the promise is never resolved. If you change hightWaterMark so that the backpressure is never an issue the test passes.

[pavelhoral]

Without vitest and promises (I expect to see foo and bar in the console, but get only foo):

const { PassThrough, Duplex, Transform, Readable } = require('node:stream');

// Simple pass-through as a placeholder for more complex setup
const through = new PassThrough({ objectMode: true });

// Stream prepared values, pipe through simple duplex and async transformer for backpressure
Readable.from(['foo', 'bar'], { objectMode: true })
  .pipe(Duplex.from({
    writable: through,
    readable: through
  }))
  .pipe(new Transform({
    objectMode: true,
    highWaterMark: 1, // Setting 1 to force backpressure after a single item
    transform(chunk, encoding, callback) {
      setTimeout(() => callback(null, chunk), 0);
    }
  }))
  .on('data', chunk => console.log(chunk));

[climba03003]

Seems like setTimeout inside the transform break something.
Changing setTimeout to process.nextTick works perfectly without error.

[targos]

@nodejs/streams

[pavelhoral]

Tried to clone NodeJS repository to write a test for this and found out there are other things broken :(. First thing I did was to change this line to make some existing test fail and start from there

node/test/parallel/test-stream-duplex-from.js

Line 143 in 36805e8

assert.strictEqual(ret, 'abcdefghi');

and the test did not start failing :(. I guess this is off-topic for this particular issue but wanted to write a comment here.

[ronag]

I'm a little low on time atm. But if someone can improve tests (i.e. add failing tests), then I could try to have a quick look at this.

[climba03003]

Interesting, if I implement the logic with new Duplex also works fine with setTimeout.
It can be further narrow down to Duplex.from is not compatible to setTimeout?

import { Duplex, PassThrough, Readable, Transform } from 'node:stream';

// Hold node process for 3s to see result
setTimeout(() => {}, 2999)

// Simple pass-through as a placeholder for more complex setup
const through = new PassThrough({
  objectMode: true, 
  highWaterMark: 1,
  transform(chunk, encoding, callback) { 
    console.log('passthrough', chunk)
    callback(null, chunk)
  }
});

// Self implement of duplex
const duplex = new Duplex({
  readableObjectMode: true,
  writableObjectMode: true,
  read(size) {
    return this.push(passthrough.read(size))
  },
  write(chunk, encoding, callback) {
    passthrough.write(chunk, encoding, callback)
  }
})

// Stream prepared values, pipe through simple duplex and async transformer for backpressure
Readable.from(['foo', 'bar', 'baz'])
  // working with self implemented duplex
  .pipe(duplex)
  // not working with Duplex.from
  // .pipe(Duplex.from({
  //   writable: passthrough,
  //   readable: passthrough
  // }))
  .pipe(new Transform({
    objectMode: true,
    highWaterMark: 1, // Setting 1 to force backpressure after a single item
    transform(chunk, encoding, callback) {
      console.log('transform', chunk)
      // setTimeout is not working with Duplex.from
      setTimeout(() => {
        callback(null, chunk)
      }, 100);
    }
  }))
  .on('data', (chunk) => { console.log('onData', chunk) })

The duplexify logic is too complicated and hard to follow. Hope my finding give some insight to the others to troubleshoot deeper.

[pavelhoral]

I have added failing test here - https://github.com/pavelhoral/node/tree/duplex-issue / pavelhoral@69c7752. The interesting thing is that it fails even without forcing the asynchronous processing (i.e. setTimeout), which I thought was significant part of the issue. So I am not that sure about anything anymore 🗡️

cc @ronag

P.S.: there is another issue in tests there #44925 (comment) that should be fixed - pavelhoral@d5c069b