René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"Zlib Stream API corrupts data.","articleBody":"### Version\n\nv24.5.0 and even previous LTS version 22\n\n### Platform\n\n```text\nTried both on MAC OS(M4 arm64) and Windows 11(Intel x64)\n```\n\n### Subsystem\n\nnode:Zlib\n\n### What steps will reproduce the bug?\n\nI am trying to use zlib inside a transform stream which compresses multiple streams with separate zlib instances. like given below\n\n```typescript\n\ntype TSplitStreamContext = {\n        redisTempKey: string,\n        csvObjectStringifier?: ReturnType\u003ctypeof createObjectCsvStringifier\u003e,\n        zipper?: Gzip,\n        downStream?: Stream[]\n    };\n\nconst tagsStreamContext = new Map\u003cstring, TSplitStreamContext\u003e();\n\nconst zipStream = new Transform({\n        objectMode: true,\n        transform(tagWiseCSVChunks: Record\u003cstring, string\u003e, encoding, callback) {\n            const tagIdVSChunkRecords = Object.entries(tagWiseCSVChunks);\n            let upstreamPressure = 0;\n            let downstreamBackPressure = 0;\n\n            for (const [tagId, csvChunk] of tagIdVSChunkRecords) {\n                const existingContext = tagsStreamContext.get(tagId) ?? {} as TSplitStreamContext;\n\n                if (existingContext.zipper == null || existingContext.zipper === undefined) {\n                    existingContext.zipper = createGzip();\n\n                    existingContext.zipper.on('data', (chunk: Buffer) =\u003e {\n                        if (this.push({ [tagId]: chunk }) === false) {\n                            downstreamBackPressure++;\n                            existingContext.zipper!.pause();\n                            this.once('drain', () =\u003e {\n                                downstreamBackPressure--;\n                                if (downstreamBackPressure === 0) {\n                                    existingContext.zipper!.resume();\n                                }\n                            });\n                        }\n                    });\n\n                    tagsStreamContext.set(tagId, existingContext);\n                }\n\n                if (existingContext.zipper.write(csvChunk) === false) {\n                    upstreamPressure++;\n                    existingContext.zipper.once('drain', () =\u003e {\n                        upstreamPressure--;\n                        if (upstreamPressure === 0) {\n                            callback();//This controls upstream flow\n                        }\n                    });\n                }\n            }\n            if (upstreamPressure === 0) {\n                callback();//This controls upstream flow\n            }\n\n        },\n\n        final(callback) {\n            const promiseHandles = [];\n\n            for (const [tagId, context] of tagsStreamContext) {\n                if (context.zipper !== null \u0026\u0026 context.zipper !== undefined) {\n                    promiseHandles.push(new Promise\u003cvoid\u003e((resolve, reject) =\u003e {\n                        context.zipper!.once('end', resolve);\n                        context.zipper!.once('error', reject);\n                        context.zipper!.end();\n                    }));\n                }\n            }\n\n            Promise.all(promiseHandles)\n                .then(() =\u003e { callback(); console.log('All zippers ended.'); })\n                .catch(err =\u003e callback(err));\n        }\n    });\n\n```\n\nI invoke this code with more stream which pass data to this transform stream like file reader, csv parser etc and downstream i have filewriter, Everything this only produces half of the data from the actual file and half is not written.\n\n### How often does it reproduce? Is there a required condition?\n\nThis happens every time.\n\n### What is the expected behavior? Why is that the expected behavior?\n\nIt should normally compress all of the file data and just act like a normal pipe operation.\n\n### What do you see instead?\n\nIt only transforms half of the file.\n\nImportant part is it works for small files \u003c100MB but fails for any bigger files.. i assume this is something to do with its internal buffers i guess (speculation)\n\nrename the attached code file from `ts` to `mts` for some reason github is not allowing `mts` extension\n\n### Additional information\n\nThe Entire code:\n\n```typescript\n\nimport { appendFileSync, createReadStream, createWriteStream, mkdirSync } from 'node:fs';\nimport Stream, { PassThrough, Transform } from 'node:stream';\nimport { pipeline } from 'node:stream/promises';\nimport { createGzip, Gzip } from 'node:zlib';\nimport csvParser from 'csv-parser';\nimport redis from 'redis';\nimport { randomInt } from 'node:crypto';\nimport { createObjectCsvStringifier } from 'csv-writer';\n\nasync function streamTodo(csvFilePath: string, expiryTimeInMilliseconds: number, redisClient: redis.RedisClientType, redisTempKeyPartA = `temp-stream-${randomInt(1e6)}-`, redisTempStreamingTimeout = 10000): Promise\u003cvoid\u003e {\n    type TTagStreamObject = { timestamp: number, value: number, status: number, tag: string };\n    type TSplitStreamContext = {\n        redisTempKey: string,\n        csvObjectStringifier?: ReturnType\u003ctypeof createObjectCsvStringifier\u003e,\n        zipper?: Gzip,\n        downStream?: Stream[]\n    };\n\n    const tagsStreamContext = new Map\u003cstring, TSplitStreamContext\u003e();\n    const rowAccumulatorMap = new WeakMap\u003cTransform, Array\u003cRecord\u003cstring, string\u003e\u003e\u003e();\n\n    const inputFileStream = createReadStream(csvFilePath, { encoding: 'utf8' });\n\n    const csvStream = csvParser();\n\n    const rowAccumulator = new Transform({\n        objectMode: true,\n        highWaterMark: 100, // Adjust this value as needed for batch size.(sweet spot between memory and speed)\n        transform(row: Record\u003cstring, string\u003e, encoding, callback) {\n            const acc = rowAccumulatorMap.get(this) ?? new Array\u003cRecord\u003cstring, string\u003e\u003e();\n            acc.push(row);\n            if (acc.length \u003e= this.readableHighWaterMark) {\n                callback(null, acc);\n                acc.length = 0;\n            } else {\n                rowAccumulatorMap.set(this, acc);\n                callback();\n            }\n        },\n        flush(callback) {\n            const acc = rowAccumulatorMap.get(this) ?? new Array\u003cRecord\u003cstring, string\u003e\u003e();\n            callback(null, acc);\n            acc.length = 0;\n        }\n    });\n\n    const tagSegregationStream = new Transform({\n        objectMode: true,\n        transform(rows: Record\u003cstring, string\u003e[], encoding, callback) {\n            const tagWiseData: Record\u003cstring, TTagStreamObject[]\u003e = {};\n            for (const row of rows) {\n                for (const [key, value] of Object.entries(row)) {\n                    if (key !== 'timestamp' \u0026\u0026 !key.endsWith('_status')) {\n                        const existingArray = tagWiseData[key] ?? [];\n                        existingArray.push({\n                            timestamp: parseFloat(row['timestamp']),\n                            value: parseFloat(value),\n                            status: parseFloat(row[`${key}_status`]),\n                            tag: key\n                        });\n                        tagWiseData[key] = existingArray;\n                    }\n                }\n            }\n            callback(null, tagWiseData);\n        }\n    });\n\n    const outputCSVStream = new Transform({\n        objectMode: true,\n        transform(tagWiseData: Record\u003cstring, TTagStreamObject[]\u003e, encoding, callback) {\n\n            const tagWiseCSVChunks: Record\u003cstring, string\u003e = {};\n            for (const [tagId, data] of Object.entries(tagWiseData)) {\n                tagWiseCSVChunks[tagId] = \"\";\n                const existingContext = tagsStreamContext.get(tagId) ?? {} as TSplitStreamContext;\n                if (existingContext?.csvObjectStringifier == undefined || existingContext?.csvObjectStringifier == null) {\n\n                    existingContext.csvObjectStringifier = createObjectCsvStringifier({\n                        header: [\n                            { id: 'timestamp', title: 'timestamp' },\n                            { id: 'value', title: 'value' },\n                            { id: 'status', title: 'status' }\n                        ]\n                    });\n                    tagsStreamContext.set(tagId, existingContext);\n                    tagWiseCSVChunks[tagId] = existingContext.csvObjectStringifier.getHeaderString() ?? \"\";\n                }\n\n                tagWiseCSVChunks[tagId] += existingContext.csvObjectStringifier.stringifyRecords(data);\n            }\n\n            callback(null, tagWiseCSVChunks);\n        }\n    });\n\n    const zipStream = new Transform({\n        objectMode: true,\n        transform(tagWiseCSVChunks: Record\u003cstring, string\u003e, encoding, callback) {\n            const tagIdVSChunkRecords = Object.entries(tagWiseCSVChunks);\n            let upstreamPressure = 0;\n            let downstreamBackPressure = 0;\n\n            for (const [tagId, csvChunk] of tagIdVSChunkRecords) {\n                const existingContext = tagsStreamContext.get(tagId) ?? {} as TSplitStreamContext;\n\n                if (existingContext.zipper == null || existingContext.zipper === undefined) {\n                    existingContext.zipper = createGzip();\n\n                    existingContext.zipper.on('data', (chunk: Buffer) =\u003e {\n                        if (this.push({ [tagId]: chunk }) === false) {\n                            downstreamBackPressure++;\n                            existingContext.zipper!.pause();\n                            this.once('drain', () =\u003e {\n                                downstreamBackPressure--;\n                                if (downstreamBackPressure === 0) {\n                                    existingContext.zipper!.resume();\n                                }\n                            });\n                        }\n                    });\n\n                    tagsStreamContext.set(tagId, existingContext);\n                }\n\n                if (existingContext.zipper.write(csvChunk) === false) {\n                    upstreamPressure++;\n                    existingContext.zipper.once('drain', () =\u003e {\n                        upstreamPressure--;\n                        if (upstreamPressure === 0) {\n                            callback();//This controls upstream flow\n                        }\n                    });\n                }\n            }\n            if (upstreamPressure === 0) {\n                callback();//This controls upstream flow\n            }\n\n        },\n\n        final(callback) {\n            const promiseHandles = [];\n\n            for (const [tagId, context] of tagsStreamContext) {\n                if (context.zipper !== null \u0026\u0026 context.zipper !== undefined) {\n                    promiseHandles.push(new Promise\u003cvoid\u003e((resolve, reject) =\u003e {\n                        context.zipper!.once('end', resolve);\n                        context.zipper!.once('error', reject);\n                        context.zipper!.end();\n                    }));\n                }\n            }\n\n            Promise.all(promiseHandles)\n                .then(() =\u003e { callback(); console.log('All zippers ended.'); })\n                .catch(err =\u003e callback(err));\n        }\n    });\n\n    const redisWriter = new Transform({\n        objectMode: true,\n        transform(chunk: Record\u003cstring, Buffer\u003e, encoding, callback) {\n            const promiseHandles = [];\n            for (const [tagId, zippedBufferChunk] of Object.entries(chunk)) {\n                const redisTempTagKey = `${redisTempKeyPartA}${tagId}`;\n\n                const existingContext = tagsStreamContext.get(tagId) ?? {} as TSplitStreamContext;\n                if (existingContext.redisTempKey == null || existingContext.redisTempKey == undefined) {\n                    existingContext.redisTempKey = redisTempTagKey;\n                    tagsStreamContext.set(tagId, existingContext);\n                }\n\n                promiseHandles.push(\n                    redisClient.multi()\n                        .append(redisTempTagKey, zippedBufferChunk)\n                        .pExpire(redisTempTagKey, redisTempStreamingTimeout)\n                        .exec()\n                )\n            }\n\n            Promise.allSettled(promiseHandles)\n                .then(() =\u003e callback())\n                .catch(err =\u003e callback(err));\n        },\n\n        final(callback) {\n            const promiseHandles = [];\n            for (const [tagId, info] of tagsStreamContext) {\n                promiseHandles.push(\n                    redisClient.multi()\n                        .rename(info.redisTempKey, tagId)\n                        .pExpire(tagId, expiryTimeInMilliseconds)\n                        .exec()\n                )\n            }\n\n            Promise.allSettled(promiseHandles)\n                .then(() =\u003e callback())\n                .catch(err =\u003e callback(err));\n        },\n    });\n\n    const fileWriter = new Transform({\n        objectMode: true,\n        transform(chunk: Record\u003cstring, Buffer\u003e, encoding, callback) {\n            for (const [tagId, zippedBufferChunk] of Object.entries(chunk)) {\n                appendFileSync(`./zipped/${tagId}.gz`, zippedBufferChunk, { flag: 'a' });\n            }\n            callback();\n        }\n    });\n\n    // const splitStreams = new Transform({\n    //     objectMode: true,\n    //     construct(callback) {\n    //         callback();\n    //     },\n    //     transform(tagWiseCSVChunks: Record\u003cstring, Buffer\u003e, encoding, callback) {\n\n    //         const tagIdVSChunkRecords = Object.entries(tagWiseCSVChunks);\n    //         let upstreamPressure = 0;\n    //         let downstreamBackPressure = 0;\n\n    //         for (const [tagId, csvChunk] of tagIdVSChunkRecords) {\n    //             const existingContext = tagsStreamContext.get(tagId) ?? {} as TSplitStreamContext;\n\n    //             if (Array.isArray(existingContext.downStream) === false || existingContext.downStream.length === 0) {\n    //                 existingContext.downStream = [new PassThrough({ objectMode: true }), createGzip(), createWriteStream(`./zipped/${tagId}.gz`, { flags: 'a' })];\n\n    //                 existingContext.zipper.on('data', (chunk: Buffer) =\u003e {\n    //                     if (this.push({ [tagId]: chunk }) === false) {\n    //                         downstreamBackPressure++;\n    //                         existingContext.zipper!.pause();\n    //                         this.once('drain', () =\u003e {\n    //                             downstreamBackPressure--;\n    //                             if (downstreamBackPressure === 0) {\n    //                                 existingContext.zipper!.resume();\n    //                             }\n    //                         });\n    //                     }\n    //                 });\n\n    //                 tagsStreamContext.set(tagId, existingContext);\n    //             }\n\n    //             if (existingContext.zipper.write(csvChunk) === false) {\n    //                 upstreamPressure++;\n    //                 existingContext.zipper.once('drain', () =\u003e {\n    //                     upstreamPressure--;\n    //                     if (upstreamPressure === 0) {\n    //                         callback();//This controls upstream flow\n    //                     }\n    //                 });\n    //             }\n    //         }\n    //         if (upstreamPressure === 0) {\n    //             callback();//This controls upstream flow\n    //         }\n    //     },\n    //     final(callback) {\n    //         redisWriter.end();\n    //         fileWriter.end();\n    //         callback();\n    //     }\n    // });\n\n    await pipeline(\n        inputFileStream,\n        csvStream,\n        rowAccumulator,\n        tagSegregationStream,\n        outputCSVStream,\n        zipStream,\n        fileWriter\n    );\n\n    tagsStreamContext.clear();\n}\n\nasync function mainModule() {\n    mkdirSync('./zipped', { recursive: true })\n    console.log('Computing...');\n    await streamTodo('./1.csv', 3600000, null as any);\n    // console.log('Unzipping...');\n    // await unzipGzFiles('./zipped');\n    // console.log('Comparing 03...');\n    // await compareCsvFiles('./1.csv', './zipped/0c6ad90b-e90d-40ac-a277-169c8024a003.gz', './zipped/diff_03.csv');\n    // console.log('Comparing 10...');\n    // await compareCsvFiles('./1.csv', './zipped/0c6ad90b-e90d-40ac-a277-169c8024a010.gz', './zipped/diff_10.csv');\n}\n\nmainModule()\n    .then(() =\u003e console.log('All operations completed successfully.'))\n    .catch(err =\u003e console.error('Error during operations:', err));\n\n```","author":{"url":"https://github.com/LRagji","@type":"Person","name":"LRagji"},"datePublished":"2025-12-29T08:02:50.000Z","interactionStatistic":{"@type":"InteractionCounter","interactionType":"https://schema.org/CommentAction","userInteractionCount":3},"url":"https://github.com/61202/node/issues/61202"}