diff --git a/lang/js/lib/files.js b/lang/js/lib/files.js index 6d63cb45087..983e411ad57 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,8 @@ BlockDecoder.prototype._write = function (chunk, encoding, cb) { tap.pos = 0; if (!this._decodeHeader()) { + // 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; } @@ -231,6 +253,12 @@ BlockDecoder.prototype._writeChunk = function (chunk, encoding, cb) { var block; while ((block = tryReadBlock(tap))) { + if (this._errored) { + // 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')); return; @@ -247,11 +275,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 +585,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..78e35af0dc8 100644 --- a/lang/js/test/test_files.js +++ b/lang/js/test/test_files.js @@ -518,6 +518,100 @@ 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; + // 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(); + }); + + 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; + // 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(); + }); + }); + }); + + 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'));