AVRO-4307: [javascript] Destroy BlockDecoder on error to stop the source

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,22 +202,22 @@ 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;
   }
 
   try {
     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;
 }
 
 

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'));