stream: readable.push returns true at internal buffer size equal highWaterMark

[zbinlin]

• Version: v7.2.1
• Platform: Arch linux 4.8.12-2 x86_64
• Subsystem: stream

const { Readable } = require("stream");

const readable = new Readable({
    objectMode: true,
    highWaterMark: 1,
    read() {
        push("inside call:");
    },
});

const lines = [];

function push(w) {
    while (lines.length) {
        const state = readable._readableState;
        console.log(w, "before push,", state.length, state.buffer.length, state.highWaterMark);
        const ret = readable.push(lines.shift());
        console.log(w, "after push,", "return:", ret, state.length, state.buffer.length, state.highWaterMark);
        if (!ret) {
            break;
        }
    }
}

setTimeout(() => {
    lines.push("foo", "bar");
    push("outside call:");
});

readable.on("data", function (data) {
    readable.pause();
});

On above testcase code, it will output:

outside call: before push, 0 0 1
inside call: before push, 0 0 1
inside call: after push, return: false 1 1 1
outside call: after push, return: true 1 1 1

expect output:

outside call: before push, 0 0 1
inside call: before push, 0 0 1
inside call: after push, return: false 1 1 1
outside call: after push, return: false 1 1 1

[addaleax]

/cc @nodejs/streams

[mcollina]

I think this is not a bug. If you see this code, https://github.com/nodejs/node/blob/master/lib/_stream_readable.js#L419-L424, another event is scheduled to happen in the future, and because of this true  is returned.

Some questions:

a. is this causing issues in real-world code?
b. Apart from increasing buffering of 1 unit, does it cause any other issue?

[zbinlin]

@mcollina
In real-world code, like this:

function push(w) {
    while (lines.length) {
        const ret = readable.push(lines.shift());
        if (!ret) {
            input.pause(); 
            return;
        }
    }
    // if first readable.push returns false, the seconds call returns true,
    // it will re-resume the paused input stream
    if (lines.length === 0) {
        input.resume();
    }
}
const input = readline(); // readline returns a stream that reads line by line
input.on("data", (data) => {
    lines.push(data);
});

[mcollina]

@zbinlin I'm not 100% understanding the exact implication for this, and assessing the impact for the community and the ecosystem.

Can you please upload the full example? It seems an endless loop from here, also that interaction between pause() and resume()  seems something I would not do in those contexts.

[Trott]

This issue has been inactive for sufficiently long that it seems like perhaps it should be closed. Feel free to re-open (or leave a comment requesting that it be re-opened) if you disagree. I'm just tidying up and not acting on a super-strong opinion or anything like that.