zlib: do not coalesce multiple .flush() calls

This is an approach to address the issue linked below. Previously,
when `.write()` and `.flush()` calls to a zlib stream were interleaved
synchronously (i.e. without waiting for these operations to finish),
multiple flush calls would have been coalesced into a single flushing
operation.
This patch changes behaviour so that each `.flush()` all corresponds
to one flushing operation on the underlying zlib resource, and the
order of operations is as if the `.flush()` call were a `.write()`
call.
One test had to be removed because it specifically tested the previous
behaviour.
As a drive-by fix, this also makes sure that all flush callbacks are
called. Previously, that was not the case.
Fixes: #28478
PR-URL: #28520
Reviewed-By: Rich Trott <rtrott@gmail.com>
Reviewed-By: Ruben Bridgewater <ruben@bridgewater.de>
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>

Author: addaleax

lib/zlib.js

@@ -49,6 +49,8 @@ const {
 }=require('buffer');
 const{ owner_symbol }=require('internal/async_hooks').symbols;
 
+constkFlushFlag=Symbol('kFlushFlag');
+
 constconstants=internalBinding('constants').zlib;
 const{
 // Zlib flush levels
@@ -261,7 +263,6 @@ function ZlibBase(opts, mode, handle, { flush, finishFlush, fullFlush }) {
 this._chunkSize=chunkSize;
 this._defaultFlushFlag=flush;
 this._finishFlushFlag=finishFlush;
-this._nextFlush=-1;
 this._defaultFullFlushFlag=fullFlush;
 this.once('end',this.close);
 this._info=opts&&opts.info;
@@ -308,21 +309,35 @@ ZlibBase.prototype._flush = function(callback) {
 
 // If a flush is scheduled while another flush is still pending, a way to figure
 // out which one is the "stronger" flush is needed.
+// This is currently only used to figure out which flush flag to use for the
+// last chunk.
 // Roughly, the following holds:
 // Z_NO_FLUSH (< Z_TREES) < Z_BLOCK < Z_PARTIAL_FLUSH <
 // Z_SYNC_FLUSH < Z_FULL_FLUSH < Z_FINISH
 constflushiness=[];
 leti=0;
-for(constflushFlagof[Z_NO_FLUSH,Z_BLOCK,Z_PARTIAL_FLUSH,
-Z_SYNC_FLUSH,Z_FULL_FLUSH,Z_FINISH]){
+constkFlushFlagList=[Z_NO_FLUSH,Z_BLOCK,Z_PARTIAL_FLUSH,
+Z_SYNC_FLUSH,Z_FULL_FLUSH,Z_FINISH];
+for(constflushFlagofkFlushFlagList){
 flushiness[flushFlag]=i++;
 }
 
 functionmaxFlush(a,b){
 returnflushiness[a]>flushiness[b] ? a : b;
 }
 
-constflushBuffer=Buffer.alloc(0);
+// Set up a list of 'special' buffers that can be written using .write()
+// from the .flush() code as a way of introducing flushing operations into the
+// write sequence.
+constkFlushBuffers=[];
+{
+constdummyArrayBuffer=newArrayBuffer();
+for(constflushFlagofkFlushFlagList){
+kFlushBuffers[flushFlag]=Buffer.from(dummyArrayBuffer);
+kFlushBuffers[flushFlag][kFlushFlag]=flushFlag;
+}
+}
+
 ZlibBase.prototype.flush=function(kind,callback){
 constws=this._writableState;
 
@@ -337,13 +352,8 @@ ZlibBase.prototype.flush = function(kind, callback) {
 }elseif(ws.ending){
 if(callback)
 this.once('end',callback);
-}elseif(this._nextFlush!==-1){
-// This means that there is a flush currently in the write queue.
-// We currently coalesce this flush into the pending one.
-this._nextFlush=maxFlush(this._nextFlush,kind);
 }else{
-this._nextFlush=kind;
-this.write(flushBuffer,'',callback);
+this.write(kFlushBuffers[kind],'',callback);
 }
 };
 
@@ -361,9 +371,8 @@ ZlibBase.prototype._transform = function(chunk, encoding, cb) {
 varflushFlag=this._defaultFlushFlag;
 // We use a 'fake' zero-length chunk to carry information about flushes from
 // the public API to the actual stream implementation.
-if(chunk===flushBuffer){
-flushFlag=this._nextFlush;
-this._nextFlush=-1;
+if(typeofchunk[kFlushFlag]==='number'){
+flushFlag=chunk[kFlushFlag];
 }
 
 // For the last chunk, also apply `_finishFlushFlag`.

test/parallel/test-zlib-flush-multiple-scheduled.js

@@ -0,0 +1,57 @@
+'use strict';
+constcommon=require('../common');
+constassert=require('assert');
+const{ createGzip, createGunzip,Z_PARTIAL_FLUSH}=require('zlib');
+
+// Verify that .flush() behaves like .write() in terms of ordering, e.g. in
+// a sequence like .write() + .flush() + .write() + .flush() each .flush() call
+// only affects the data written before it.
+// Refs: https://github.com/nodejs/node/issues/28478
+
+constcompress=createGzip();
+constdecompress=createGunzip();
+decompress.setEncoding('utf8');
+
+constevents=[];
+constcompressedChunks=[];
+
+for(constchunkof['abc','def','ghi']){
+compress.write(chunk,common.mustCall(()=>events.push({written: chunk})));
+compress.flush(Z_PARTIAL_FLUSH,common.mustCall(()=>{
+events.push('flushed');
+constchunk=compress.read();
+if(chunk!==null)
+compressedChunks.push(chunk);
+}));
+}
+
+compress.end(common.mustCall(()=>{
+events.push('compress end');
+writeToDecompress();
+}));
+
+functionwriteToDecompress(){
+// Write the compressed chunks to a decompressor, one by one, in order to
+// verify that the flushes actually worked.
+constchunk=compressedChunks.shift();
+if(chunk===undefined)returndecompress.end();
+decompress.write(chunk,common.mustCall(()=>{
+events.push({read: decompress.read()});
+writeToDecompress();
+}));
+}
+
+process.on('exit',()=>{
+assert.deepStrictEqual(events,[
+{written: 'abc'},
+'flushed',
+{written: 'def'},
+'flushed',
+{written: 'ghi'},
+'flushed',
+'compress end',
+{read: 'abc'},
+{read: 'def'},
+{read: 'ghi'}
+]);
+});

test/parallel/test-zlib-flush-write-sync-interleaved.js

@@ -39,7 +39,6 @@ for (const [ createCompress, createDecompress ] of [
 gunz.on('data',(c)=>output+=c);
 gunz.on('end',common.mustCall(()=>{
 assert.strictEqual(output,input);
-assert.strictEqual(gzip._nextFlush,-1);
 }));
 
 // Make sure that flush/write doesn't trigger an assert failure