Ensure late errors on requests are passed through properly

lib/fetch.js

@@ -18,16 +18,19 @@ function fetch (uri, params, cb) {
     makeRequest.call(client, uri, params, function (er, req) {
       if (er) return cb(er)
 
-      req.on('error', function (er) {
+      req.once('error', retryOnError)
+
+      function retryOnError (er) {
         if (operation.retry(er)) {
           client.log.info('retry', 'will retry, error on last attempt: ' + er)
         } else {
           cb(er)
         }
-      })
+      }
 
       req.on('response', function (res) {
         client.log.http('fetch', '' + res.statusCode, uri)
+        req.removeListener('error', retryOnError)
 
         var er
         var statusCode = res && res.statusCode
@@ -37,6 +40,10 @@ function fetch (uri, params, cb) {
           res.resume()
           if (process.version === 'v0.10.0') unstick(res)
 
+          req.once('error', function (er) {
+            res.emit('error', er)
+          })
+
           return cb(null, res)
           // Only retry on 408, 5xx or no `response`.
         } else if (statusCode === 408) {

package.json

@@ -25,6 +25,7 @@
     "negotiator": "^0.6.1",
     "nock": "^8.0.0",
     "readable-stream": "^2.1.5",
+    "require-inject": "^1.4.0",
     "rimraf": "^2.5.4",
     "standard": "~8.5.0",
     "tap": "^7.0.0"

test/econnreset.js

@@ -0,0 +1,95 @@
+'use strict'
+var requireInject = require('require-inject')
+var test = require('tap').test
+var EventEmitter = require('events').EventEmitter
+var PassThrough = require('readable-stream').PassThrough
+
+var content = [
+  'first chunk',
+  'second chunk'
+]
+var requests = 0
+var fetch = requireInject('../lib/fetch.js', {
+  request: function (opts) {
+    var req = new EventEmitter()
+    ++requests
+    setTimeout(function () {
+      var res = new PassThrough()
+      res.statusCode = 200
+
+      setTimeout(function () {
+        res.write(content[0])
+      }, 50)
+
+      if (requests === 1) {
+        setTimeout(function () {
+          var err = new Error('read ECONNRESET')
+          err.code = 'ECONNRESET'
+          req.emit('error', err)
+        }, 100)
+      } else {
+        // we allow success on retries, though in practice we won't be
+        // retrying.
+        setTimeout(function () {
+          res.end(content[1])
+        }, 100)
+      }
+      req.emit('response', res)
+    }, 50)
+    return req
+  }
+})
+
+function clientMock (t) {
+  return {
+    log: {
+      info: function (section, message) {
+        t.comment('[info] ' + section + ': ' + [].slice.call(arguments, 1).join(' '))
+      },
+      http: function (section, message) {
+        t.comment('[http] ' + section + ': ' + [].slice.call(arguments, 1).join(' '))
+      }
+    },
+    authify: function (alwaysAuth, parsed, headers, auth) {
+      return
+    },
+    initialize: function (parsed, method, accept, headers) {
+      return {}
+    },
+    attempt: require('../lib/attempt.js'),
+    config: {
+      retry: {
+        retries: 2,
+        factor: 10,
+        minTimeout: 10000,
+        maxTimeout: 60000
+      }
+    }
+  }
+}
+
+/*
+This tests that errors that occur in the REQUEST object AFTER a `response`
+event has been emitted will be passed through to the `response` stream.
+This means that we won't try to retry these sorts of errors ourselves.
+*/
+
+test('econnreset', function (t) {
+  var client = clientMock(t)
+  fetch.call(client, 'http://example.com/', {}, function (err, res) {
+    t.ifError(err, 'initial fetch ok')
+    var data = ''
+    res.on('data', function (chunk) {
+      data += chunk
+    })
+    res.on('error', function (err) {
+      t.comment('ERROR:', err.stack)
+      t.pass("errored and that's ok")
+      t.done()
+    })
+    res.on('end', function () {
+      t.is(data, content.join(''), 'succeeded and got the right data')
+      t.done()
+    })
+  })
+})