http: IncomingMessage emits 'end' after 'close'

[ronag]

'end' can be emitted after 'close'. This can cause e.g. pipeline to error with a
ERR_STREAM_PREMATURE_CLOSE.

See, nxtedition#1 for repo test.

[ronag]

Possibly related #27916 @lpinca

[ronag]

If this is resolved, I believe 779a05d can be re-applied inside a 'close' listener.

[ronag]

Slightly related to addaleax@9aedf72 @addaleax

[ronag]

There are PR's for this

[gustavomassa]

@ronag Sorry to bother you, but i think I'm facing this issue.

Node Version: 10.15.3
Linux Mint 19.3 x64

I'm testing the node streams back-pressuring with pipeline and transform stream to stream data from mongodb cursor dynamically instead of allocating the entire mongodb cursor on RAM.
I'm reading from the mongodb cursor, using a transform stream to stringfy the objects and piping the results to express.response(http).
I'm receiving the error "Premature close", but the data was sent correctly.
Maybe just ignore the premature close error?

export class MongoCursorTransform extends Transform {
    private firstChunk: boolean;

    constructor() {
        super({ readableObjectMode: true, writableObjectMode: true });
        this.firstChunk = true;
    }

    _transform(chunk, encoding, callback) {
        if (this.firstChunk) {
            this.firstChunk = false;
            callback(null, '[' + JSON.stringify(chunk) + ',');

        } else callback(null, JSON.stringify(chunk) + ',');
    }

    _flush(callback) {
        callback(null, '{}]');
    }
}

const operation = new Promise((resolve, reject, onCancel) => {
                onCancel(() => {
                    if (cursor) cursor.destroy();
                    if (res) res.destroy();
                    return;
                });

                //NODE STREAM PIPELINE TEST
                const res = reqParams['res'];
                res.once('error', function (err) {
                    Webbuffet.logError(err);
                    //reject(Webbuffet.error(Status.FAILED, err.message));
                    return Webbuffet.error(Status.FAILED, err.message);
                });
                res.writeHead(200, {
                    'Content-Type': 'application/json',
                    'Transfer-Encoding': 'chunked'
                });

                const cursor = this.collection(this.processCollectionName).aggregate(aggregationPipeline, { maxTimeMS: timeout, allowDiskUse: false, cursor: { batchSize: 0 } });
                pipeline(
                    <any>cursor,
                    new MongoCursorTransform(),
                    res,
                    (err) => {
                        if (err) {
                            reject(Webbuffet.error(Status.FAILED, err.message));
                        } else resolve(Webbuffet.success(true));
                    }
                );
            });

            return operation.timeout(Webbuffet.getRemainingTimeout(reqParams.start, reqParams.timeout));

As workaround I just ignored the error code 'ERR_STREAM_PREMATURE_CLOSE'

const cursor = this.collection(this.processCollectionName).aggregate(aggregationPipeline, { maxTimeMS: timeout, allowDiskUse: false, cursor: { batchSize: 0 } });
                pipeline(
                    <any>cursor,
                    new MongoCursorTransform(),
                    res,
                    (err) => {
                        if (err && err.code !== 'ERR_STREAM_PREMATURE_CLOSE') {
                            reject(Webbuffet.error(Status.FAILED, err.message));
                        } else resolve(Webbuffet.success(true));
                    }
                );

Another question not related to the issues, is there a way to know when I'm receiving the last chunk inside the transform stream? Before the end/finish events.