From a2e4f3e2d33bdb4fc33a6686f1f0c79e9130af23 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Mon, 13 Jul 2026 08:04:19 +0200 Subject: [PATCH 1/3] AVRO-4307: [javascript] Destroy BlockDecoder on error; stop the source BlockDecoder decompresses blocks asynchronously, so many block callbacks can fail for the same corrupt/oversized input. The failure path emitted 'error' once per failing block with a bare emit(), which: - produced an error storm (one 'error' event per block; ~1000 for a modest file rather than a single error), - never destroyed the decoder, so its readable side was left dangling, and - via createFileDecoder's plain .pipe(), left the upstream file ReadStream reading a file already known to be bad and its descriptor open (only unpiped, never destroyed). Route every fatal error through a new _onError() that guards on an _errored flag and calls destroy(err), so exactly one 'error' is surfaced and the stream is torn down. Decrement _pending before the error check so accounting stays correct (previously it was skipped on error). Guard _write/_writeChunk so no further blocks are processed once errored. Switch createFileDecoder from .pipe() to stream.pipeline() so the source file stream is destroyed on error and its descriptor released. Verified on a 394 KB multi-block file decoded with an always-failing codec: error events 947 -> 1, bytes read after first error 65536 -> 0, source and decoder both destroyed. Adds regression tests for single-error emission and for source teardown via createFileDecoder. Assisted-by: GitHub Copilot:claude-opus-4.8 --- lang/js/lib/files.js | 44 +++++++++++++++++++++++++++++++----- lang/js/test/test_files.js | 46 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 84 insertions(+), 6 deletions(-) diff --git a/lang/js/lib/files.js b/lang/js/lib/files.js index 6d63cb45087..7bc054fadaa 100644 --- a/lang/js/lib/files.js +++ b/lang/js/lib/files.js @@ -156,6 +156,7 @@ function BlockDecoder(opts) { this._pending = 0; // Number of blocks undergoing decompression. this._needPush = false; this._finished = false; + this._errored = false; // Whether a fatal error was already surfaced. this.on('finish', function () { this._finished = true; @@ -166,6 +167,25 @@ function BlockDecoder(opts) { } util.inherits(BlockDecoder, stream.Duplex); +/** + * Surface a fatal decoding error exactly once and tear the stream down. + * + * Block decompression is asynchronous, so several block callbacks can fail for + * the same corrupt or oversized input. Without a guard each one would emit its + * own 'error' event (an error storm), and a bare `emit('error')` would neither + * stop a piped source (e.g. the file ReadStream in `createFileDecoder`) from + * continuing to read nor release its descriptor. Guard on a flag and `destroy` + * so only the first error is reported and the pipe unwinds (which unpipes and + * lets the upstream stream be cleaned up). + */ +BlockDecoder.prototype._onError = function (err) { + if (this._errored) { + return; + } + this._errored = true; + this.destroy(err); // Emits 'error' with `err`, then tears the stream down. +}; + BlockDecoder.getDefaultCodecs = function () { return { 'null': function (buf, cb) { cb(null, buf); }, @@ -182,14 +202,14 @@ BlockDecoder.prototype._decodeHeader = function () { } if (!MAGIC_BYTES.equals(header.magic)) { - this.emit('error', new Error('invalid magic bytes')); + this._onError(new Error('invalid magic bytes')); return; } var codec = (header.meta['avro.codec'] || 'null').toString(); this._decompress = (this._codecs || BlockDecoder.getDefaultCodecs())[codec]; if (!this._decompress) { - this.emit('error', new Error(f('unknown codec: %s', codec))); + this._onError(new Error(f('unknown codec: %s', codec))); return; } @@ -197,7 +217,7 @@ BlockDecoder.prototype._decodeHeader = function () { var schema = JSON.parse(header.meta['avro.schema'].toString()); this._type = parse(schema, this._parseOpts); } catch (err) { - this.emit('error', err); + this._onError(err); return; } @@ -213,6 +233,9 @@ BlockDecoder.prototype._write = function (chunk, encoding, cb) { tap.pos = 0; if (!this._decodeHeader()) { + if (this._errored) { + return; // Destroyed while decoding the header; leave the write callback. + } process.nextTick(cb); return; } @@ -231,6 +254,9 @@ BlockDecoder.prototype._writeChunk = function (chunk, encoding, cb) { var block; while ((block = tryReadBlock(tap))) { + if (this._errored) { + return; // A prior block already failed and destroyed the stream. + } if (!this._syncMarker.equals(block.sync)) { cb(new Error('invalid sync marker')); return; @@ -247,11 +273,11 @@ BlockDecoder.prototype._createBlockCallback = function () { this._pending++; return function (err, data) { + self._pending--; if (err) { - self.emit('error', err); + self._onError(err); return; } - self._pending--; self._queue.push(new BlockData(index, data)); if (self._needPush) { self._needPush = false; @@ -557,7 +583,13 @@ function extractFileHeader(path, opts) { * */ function createFileDecoder(path, opts) { - return fs.createReadStream(path).pipe(new BlockDecoder(opts)); + var decoder = new BlockDecoder(opts); + // Use pipeline (not pipe) so that if decoding fails, the source file stream is + // destroyed and its descriptor released, rather than left open reading a file + // already known to be bad. Errors still surface on the returned decoder via + // its own 'error' event; the callback just keeps pipeline from throwing. + stream.pipeline(fs.createReadStream(path), decoder, function () {}); + return decoder; } diff --git a/lang/js/test/test_files.js b/lang/js/test/test_files.js index 43d5df60162..54dc62f97c1 100644 --- a/lang/js/test/test_files.js +++ b/lang/js/test/test_files.js @@ -518,6 +518,52 @@ describe('files', function () { encoder.end(1); }); + it('surfaces a single error for many failing blocks', function (cb) { + // A codec that fails on every block must not emit one 'error' per block + // (an error storm); only the first should surface and the stream should be + // destroyed. + var t = createType('int'); + var codecs = { + 'null': function (data, cb) { cb(new Error('ouch')); } + }; + var encoder = new streams.BlockEncoder(t, {codec: 'null', blockSize: 1}); + var decoder = new streams.BlockDecoder({codecs: codecs}); + var errorCount = 0; + decoder.on('data', function () {}).on('error', function () { errorCount++; }); + encoder.pipe(decoder); + // Write many records; blockSize 1 forces many separate blocks. + for (var i = 0; i < 100; i++) { encoder.write(i); } + encoder.end(); + setTimeout(function () { + assert.equal(errorCount, 1); + assert(decoder.destroyed); + cb(); + }, 100); + }); + + it('createFileDecoder tears down the source on error', function (cb) { + // Write a valid file, then decode it with a codec that always fails; the + // underlying file stream must be destroyed (descriptor released) rather + // than left reading a file already known to be bad. + var t = createType('int'); + var filePath = tmp.fileSync().name; + var encoder = files.createFileEncoder(filePath, t); + for (var i = 0; i < 100; i++) { encoder.write(i); } + encoder.end(); + encoder.getDownstream().on('finish', function () { + var errorCount = 0; + var decoder = files.createFileDecoder(filePath, { + codecs: { 'null': function (data, cb) { cb(new Error('ouch')); } } + }); + decoder.on('data', function () {}).on('error', function () { errorCount++; }); + setTimeout(function () { + assert.equal(errorCount, 1); + assert(decoder.destroyed); + cb(); + }, 100); + }); + }); + it('decompression late read', function (cb) { var chunks = []; var encoder = new streams.BlockEncoder(createType('int')); From 35425f0728002ef2ee41b1b4310ff5990d9ddc76 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Mon, 13 Jul 2026 12:03:15 +0200 Subject: [PATCH 2/3] AVRO-4307: [javascript] Make BlockDecoder teardown tests deterministic --- lang/js/test/test_files.js | 48 ++++++++++++++++++++++++++++---------- 1 file changed, 36 insertions(+), 12 deletions(-) diff --git a/lang/js/test/test_files.js b/lang/js/test/test_files.js index 54dc62f97c1..fe5bbcc9600 100644 --- a/lang/js/test/test_files.js +++ b/lang/js/test/test_files.js @@ -529,16 +529,21 @@ describe('files', function () { var encoder = new streams.BlockEncoder(t, {codec: 'null', blockSize: 1}); var decoder = new streams.BlockDecoder({codecs: codecs}); var errorCount = 0; - decoder.on('data', function () {}).on('error', function () { errorCount++; }); + // Finish deterministically when the decoder is actually destroyed ('close') + // rather than after a fixed delay: the failing codec can call back + // synchronously, so a timeout is both slower and race-prone. + decoder + .on('data', function () {}) + .on('error', function () { errorCount++; }) + .on('close', function () { + assert.equal(errorCount, 1); + assert(decoder.destroyed); + cb(); + }); encoder.pipe(decoder); // Write many records; blockSize 1 forces many separate blocks. for (var i = 0; i < 100; i++) { encoder.write(i); } encoder.end(); - setTimeout(function () { - assert.equal(errorCount, 1); - assert(decoder.destroyed); - cb(); - }, 100); }); it('createFileDecoder tears down the source on error', function (cb) { @@ -552,15 +557,34 @@ describe('files', function () { encoder.end(); encoder.getDownstream().on('finish', function () { var errorCount = 0; - var decoder = files.createFileDecoder(filePath, { - codecs: { 'null': function (data, cb) { cb(new Error('ouch')); } } - }); - decoder.on('data', function () {}).on('error', function () { errorCount++; }); - setTimeout(function () { + // Capture the source ReadStream that createFileDecoder opens so we can + // assert it is torn down (its descriptor released), not just the decoder. + var origCreateReadStream = fs.createReadStream; + var src; + fs.createReadStream = function () { + src = origCreateReadStream.apply(fs, arguments); + return src; + }; + var decoder; + try { + decoder = files.createFileDecoder(filePath, { + codecs: { 'null': function (data, cb) { cb(new Error('ouch')); } } + }); + } finally { + fs.createReadStream = origCreateReadStream; + } + decoder + .on('data', function () {}) + .on('error', function () { errorCount++; }); + // Finish when the source stream is actually closed (descriptor released), + // which is the teardown this test verifies; by then the decoder has been + // destroyed and the single error surfaced. + src.on('close', function () { assert.equal(errorCount, 1); assert(decoder.destroyed); + assert(src.destroyed); cb(); - }, 100); + }); }); }); From 8dde40a987ac3afb57f7f85eaafadc426b180649 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Mon, 13 Jul 2026 13:46:29 +0200 Subject: [PATCH 3/3] AVRO-4307: [javascript] Release the write callback when a fatal error destroys the decoder --- lang/js/lib/files.js | 10 ++++++---- lang/js/test/test_files.js | 24 ++++++++++++++++++++++++ 2 files changed, 30 insertions(+), 4 deletions(-) diff --git a/lang/js/lib/files.js b/lang/js/lib/files.js index 7bc054fadaa..983e411ad57 100644 --- a/lang/js/lib/files.js +++ b/lang/js/lib/files.js @@ -233,9 +233,8 @@ BlockDecoder.prototype._write = function (chunk, encoding, cb) { tap.pos = 0; if (!this._decodeHeader()) { - if (this._errored) { - return; // Destroyed while decoding the header; leave the write callback. - } + // Release the write callback even when a fatal header error destroyed the + // stream, so upstream writers/pipelines are not left stalled mid-write. process.nextTick(cb); return; } @@ -255,7 +254,10 @@ BlockDecoder.prototype._writeChunk = function (chunk, encoding, cb) { var block; while ((block = tryReadBlock(tap))) { if (this._errored) { - return; // A prior block already failed and destroyed the stream. + // A prior block already failed and destroyed the stream; release the + // write callback so upstream writers/pipelines are not left stalled. + process.nextTick(cb); + return; } if (!this._syncMarker.equals(block.sync)) { cb(new Error('invalid sync marker')); diff --git a/lang/js/test/test_files.js b/lang/js/test/test_files.js index fe5bbcc9600..78e35af0dc8 100644 --- a/lang/js/test/test_files.js +++ b/lang/js/test/test_files.js @@ -588,6 +588,30 @@ describe('files', function () { }); }); + it('releases the write callback on a fatal header error', function (cb) { + // A fatal header error destroys the stream; _write must still invoke its + // write callback, otherwise upstream writers/pipelines stall mid-write. + // _onError is stubbed to only flag the error (not destroy), so this + // isolates the _write callback from Node's own destroy machinery: if the + // callback were left pending the test would hang and time out. + var t = createType('int'); + var chunks = []; + var encoder = new streams.BlockEncoder(t); + encoder.on('data', function (chunk) { chunks.push(chunk); }); + encoder.on('end', function () { + var buf = Buffer.concat(chunks); + buf[0] = buf[0] ^ 0xff; // corrupt the magic bytes -> fatal header error + var decoder = new streams.BlockDecoder(); + var errored = false; + decoder._onError = function () { this._errored = errored = true; }; + decoder._write(buf, undefined, function () { + assert(errored); + cb(); + }); + }); + encoder.end(1); + }); + it('decompression late read', function (cb) { var chunks = []; var encoder = new streams.BlockEncoder(createType('int'));