Retry read operations up to 10 times on EIO (#70)

Some faulty SDCard, or SDCard readers might throw EIO when reading data.
Retrying a couple of times makes the issue go away.

See: #73
Signed-off-by: Juan Cruz Viotti <jviotti@openmailbox.org>

lib/checksum.js

@@ -18,11 +18,11 @@
 
 const _ = require('lodash');
 const Bluebird = require('bluebird');
-const fs = require('fs');
 const sliceStream = require('slice-stream2');
 const progressStream = require('progress-stream');
 const CRC32Stream = require('crc32-stream');
 const DevNullStream = require('dev-null-stream');
+const ImageReadStream = require('./image-read-stream');
 const utils = require('./utils');
 
 /**
@@ -33,13 +33,15 @@ const utils = require('./utils');
  * @param {Number} deviceFileDescriptor - device file descriptor
  * @param {Object} options - options
  * @param {Number} options.imageSize - image size in bytes
+ * @param {Number} options.chunkSize - chunk size
  * @param {Function} options.progress - progress callback (state)
  * @fulfil {String} - checksum
  * @returns {Promise}
  *
  * const fd = fs.openSync('/dev/rdisk2', 'rs+');
  * checksum.calculateDeviceChecksum(fd, {
  *   imageSize: 65536 * 128,
+ *   chunkSize: 65536 * 16
  *   progress: (state) => {
  *     console.log(state);
  *   }
@@ -53,13 +55,8 @@ exports.calculateDeviceChecksum = (deviceFileDescriptor, options = {}) => {
   return new Bluebird((resolve, reject) => {
     utils.safePipe([
       {
-        stream: fs.createReadStream(null, {
-          fd: deviceFileDescriptor,
-
-          // Make sure we start reading from the beginning,
-          // since the fd might have been modified before.
-          start: 0
-
+        stream: new ImageReadStream(deviceFileDescriptor, {
+          chunkSize: options.chunkSize
         })
       },
       {

lib/filesystem.js

@@ -88,3 +88,56 @@ exports.writeChunk = (fd, chunk, position, callback, retries = 1) => {
     return callback();
   });
 };
+
+/**
+ * @summary Read a chunk from a file descriptor
+ * @function
+ * @public
+ *
+ * @description
+ * This function is simply a convenient wrapper around `fs.read()`.
+ *
+ * @param {Number} fd - file descriptor
+ * @param {Number} size - chunk size to read
+ * @param {Number} position - position to read from
+ * @param {Function} callback - callback (error, buffer)
+ * @param {Number} [retries=1] - number of pending retries
+ *
+ * @example
+ * const fd = fs.openSync('/dev/rdisk2', 'rs+');
+ *
+ * filesystem.readChunk(fd, 512, 0, (error, buffer) => {
+ *   if (error) {
+ *     throw error;
+ *   }
+ *
+ *   console.log(buffer);
+ * });
+ */
+exports.readChunk = (fd, size, position, callback, retries = 1) => {
+  fs.read(fd, Buffer.allocUnsafe(size), 0, size, position, (error, bytesRead, buffer) => {
+
+    if (error) {
+
+      // In some faulty SDCards or SDCard readers you might randomly
+      // get EIO errors, but they usually go away after some retries.
+      if (error.code === 'EIO') {
+        if (retries > MAXIMUM_RETRIES) {
+          return callback(error);
+        }
+
+        return setTimeout(() => {
+          return exports.readChunk(fd, size, position, callback, retries + 1);
+        }, RETRY_BASE_TIMEOUT * retries);
+      }
+
+      return callback(error);
+    }
+
+    if (bytesRead === 0) {
+      return callback(null, null);
+    }
+
+    return callback(null, buffer);
+  });
+};

lib/image-read-stream.js

@@ -0,0 +1,71 @@
+/*
+ * Copyright 2016 Resin.io
+ *
+ * Licensed under the Apache License, Version 2.0 (the 'License');
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an 'AS IS' BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+'use strict';
+
+const stream = require('stream');
+const filesystem = require('./filesystem');
+const ImageStreamPosition = require('./image-stream-position');
+
+module.exports = class ImageWriteStream extends stream.Readable {
+
+  /**
+   * @summary Create an instance of ImageReadStream
+   * @name ImageReadStream
+   * @class
+   * @public
+   *
+   * @param {Number} fileDescriptor - file descriptor
+   * @param {Object} options - options
+   * @param {Number} options.chunkSize - chunk size
+   *
+   * @example
+   * const fd = fs.openSync('/dev/rdisk2', 'rs+');
+   *
+   * const stream = new ImageReadStream(fd, {
+   *   chunkSize: 65536 * 16
+   * });
+   *
+   * stream.pipe(fs.createWriteStream('path/to/output/img'))
+   *   .pipe(new ImageWriteStream(fd));
+   */
+  constructor(fileDescriptor, options) {
+    const position = new ImageStreamPosition();
+
+    super({
+      highWaterMark: options.chunkSize,
+      objectMode: false,
+      read: (size) => {
+        const streamPosition = position.getStreamPosition();
+
+        filesystem.readChunk(this.fileDescriptor, size, streamPosition, (error, buffer) => {
+          if (error) {
+            return this.emit('error', error);
+          }
+
+          if (buffer) {
+            position.incrementStreamPosition(buffer.length);
+          }
+
+          this.push(buffer);
+        });
+      }
+    });
+
+    this.fileDescriptor = fileDescriptor;
+  }
+
+};

lib/index.js

@@ -126,16 +126,17 @@ exports.write = (drive, image, options) => {
     throw new Error('Invalid drive size: ' + drive.size);
   }
 
-  win32.prepare(drive.device).then(() => {
+  // 64K * 16 = 1024K = 1M
+  const CHUNK_SIZE = 65536 * 16;
+
+  win32.prepare(drive.device).then(function() {
     return write.inferFromOptions(drive.fd, {
       imageSize: image.size,
       imageStream: image.stream,
       transformStream: options.transform,
       bmapContents: options.bmap,
       bytesToZeroOutFromTheBeginning: options.bytesToZeroOutFromTheBeginning,
-
-      // 64K * 16 = 1024K = 1M
-      chunkSize: 65536 * 16,
+      chunkSize: CHUNK_SIZE,
 
       // 64K * 8 = 512K
       minimumChunkSize: 65536 * 8,
@@ -152,6 +153,7 @@ exports.write = (drive, image, options) => {
       bmapContents: options.bmap,
       imageSize: results.transferredBytes,
       imageChecksum: results.checksum,
+      chunkSize: CHUNK_SIZE,
       progress: (state) => {
         state.type = 'check';
         emitter.emit('progress', state);

lib/validate.js

@@ -71,6 +71,7 @@ exports.usingBmap = (deviceFileDescriptor, options = {}) => {
  * @param {Object} options - options
  * @param {Number} options.imageSize - image size
  * @param {String} options.imageChecksum - image checksum
+ * @param {Number} options.chunkSize - chunk size
  * @param {Function} options.progress - progress callback (state)
  * @fulfil {Object} - results
  * @returns {Promise}
@@ -80,6 +81,7 @@ exports.usingBmap = (deviceFileDescriptor, options = {}) => {
  * validate.usingChecksum(fd, {
  *   imageSize: 65536 * 128,
  *   imageChecksum: '717a34c1',
+ *   chunkSize: 65536 * 16,
  *   progress: (state) => {
  *     console.log(state);
  *   }
@@ -90,7 +92,8 @@ exports.usingBmap = (deviceFileDescriptor, options = {}) => {
 exports.usingChecksum = (deviceFileDescriptor, options = {}) => {
   return checksum.calculateDeviceChecksum(deviceFileDescriptor, {
     imageSize: options.imageSize,
-    progress: options.progress
+    progress: options.progress,
+    chunkSize: options.chunkSize
   }).then((deviceChecksum) => {
     if (options.imageChecksum !== deviceChecksum) {
       throw errors.ValidationError;
@@ -152,24 +155,22 @@ exports.mock = (deviceFileDescriptor, options = {}) => {
  * });
  */
 exports.inferFromOptions = (deviceFileDescriptor, options = {}) => {
-  return Bluebird.try(() => {
-    if (options.omitValidation) {
-      return exports.mock(deviceFileDescriptor, {
-        imageChecksum: options.imageChecksum
-      });
-    }
-
-    if (options.bmapContents) {
-      return exports.usingBmap(deviceFileDescriptor, {
-        bmapContents: options.bmapContents,
-        progress: options.progress
-      });
-    }
+  if (options.omitValidation) {
+    return exports.mock(deviceFileDescriptor, {
+      imageChecksum: options.imageChecksum
+    });
+  }
 
-    return exports.usingChecksum(deviceFileDescriptor, {
-      imageSize: options.imageSize,
-      imageChecksum: options.imageChecksum,
+  if (options.bmapContents) {
+    return exports.usingBmap(deviceFileDescriptor, {
+      bmapContents: options.bmapContents,
       progress: options.progress
     });
+  }
+
+  return exports.usingChecksum(deviceFileDescriptor, {
+    imageSize: options.imageSize,
+    imageChecksum: options.imageChecksum,
+    progress: options.progress
   });
 };

tests/evalidation/test.js

@@ -21,23 +21,22 @@ const path = require('path');
 const Bluebird = require('bluebird');
 const fs = Bluebird.promisifyAll(require('fs'));
 const imageWrite = require('../../lib');
+const checksum = require('../../lib/checksum');
 
 module.exports = [
 
   {
     name: 'should throw EVALIDATION is check fails',
     data: {
       input: path.join(__dirname, 'input'),
-      mock: path.join(__dirname, 'mock'),
       output: path.join(__dirname, 'output')
     },
     case: (data) => {
       const outputFileDescriptor = fs.openSync(data.output, 'rs+');
       const stream = fs.createReadStream(data.input);
-      const mockStream = fs.createReadStream(data.mock);
 
-      const createReadStreamStub = m.sinon.stub(fs, 'createReadStream');
-      createReadStreamStub.returns(mockStream);
+      const calculateDeviceChecksumStub = m.sinon.stub(checksum, 'calculateDeviceChecksum');
+      calculateDeviceChecksumStub.returns(Bluebird.resolve('xxxxxxx'));
 
       return new Bluebird((resolve, reject) => {
         const imageSize = fs.statSync(data.input).size;
@@ -64,7 +63,7 @@ module.exports = [
       }).catch((error) => {
         m.chai.expect(error.code).to.equal('EVALIDATION');
       }).finally(() => {
-        createReadStreamStub.restore();
+        calculateDeviceChecksumStub.restore();
         return fs.closeAsync(outputFileDescriptor);
       });
     }