fix: body error on premature close (#336)

lib/client.js

@@ -492,6 +492,10 @@ class Parser extends HTTPParser {
     const { client, socket } = this
     const request = client[kQueue][client[kRunningIdx]]
 
+    if (socket.destroyed) {
+      return
+    }
+
     // TODO: Check for content-length mismatch from server?
 
     assert(!this.upgrade)
@@ -551,7 +555,7 @@ class Parser extends HTTPParser {
   [HTTPParser.kOnBody] (chunk, offset, length) {
     const { client, socket, statusCode } = this
 
-    if (!statusCode) {
+    if (socket.destroyed) {
       return
     }
 
@@ -570,7 +574,6 @@ class Parser extends HTTPParser {
       // until requests has either completed or socket is destroyed.
       if (this.read > client[kMaxAbortedPayload]) {
         client[kQueue][client[kRunningIdx]++] = null
-        this.statusCode = null
         util.destroy(socket, new InformationalError('max aborted payload'))
       }
     } else if (ret === false) {
@@ -582,7 +585,7 @@ class Parser extends HTTPParser {
     const { client, socket, statusCode, headers, upgrade } = this
     const request = client[kQueue][client[kRunningIdx]]
 
-    if (!statusCode) {
+    if (socket.destroyed) {
       return
     }
 
@@ -611,17 +614,17 @@ class Parser extends HTTPParser {
       // TOOD: keep writing until maxAbortedPayload?
 
       // Response completed before request.
-      util.destroy(socket, new InformationalError('request reset'))
+      util.destroy(socket, new InformationalError('reset'))
     } else if (client[kReset]) {
       // Destroy socket once all requests have completed.
       // The request at the tail of the pipeline is the one
       // that requested reset and no further requests should
       // have been queued since then.
       if (!client.running) {
-        util.destroy(socket, new InformationalError('request reset'))
+        util.destroy(socket, new InformationalError('reset'))
       }
     } else if (!this.shouldKeepAlive) {
-      util.destroy(socket, new InformationalError('request reset'))
+      util.destroy(socket, new InformationalError('reset'))
     } else {
       socket.resume()
       resume(client)

test/client-errors.js

@@ -687,7 +687,7 @@ test('socket fail while ending request body', (t) => {
 })
 
 test('queued request should not fail on socket destroy', (t) => {
-  t.plan(2)
+  t.plan(4)
 
   const server = createServer()
   server.on('request', (req, res) => {
@@ -706,21 +706,25 @@ test('queued request should not fail on socket destroy', (t) => {
       method: 'GET'
     }, (err, data) => {
       t.error(err)
-      data.body.resume()
+      data.body.resume().on('error', () => {
+        t.pass()
+      })
       client[kSocket].destroy()
       client.request({
         path: '/',
         method: 'GET'
       }, (err, data) => {
         t.error(err)
-        data.body.resume()
+        data.body.resume().on('end', () => {
+          t.pass()
+        })
       })
     })
   })
 })
 
 test('queued request should fail on client destroy', (t) => {
-  t.plan(5)
+  t.plan(6)
 
   const server = createServer()
   server.on('request', (req, res) => {
@@ -741,6 +745,9 @@ test('queued request should fail on client destroy', (t) => {
     }, (err, data) => {
       t.error(err)
       data.body.resume()
+        .on('error', () => {
+          t.pass()
+        })
       client.destroy((err) => {
         t.error(err)
         t.strictEqual(requestErrored, true)
@@ -771,7 +778,7 @@ test('retry idempotent inflight', (t) => {
     const client = new Client(`http://localhost:${server.address().port}`, {
       pipelining: 3
     })
-    t.tearDown(client.destroy.bind(client))
+    t.tearDown(client.close.bind(client))
 
     client.request({
       path: '/',

test/client-reconnect.js

@@ -6,7 +6,7 @@ const { createServer } = require('http')
 const FakeTimers = require('@sinonjs/fake-timers')
 
 test('multiple reconnect', (t) => {
-  t.plan(2)
+  t.plan(3)
 
   const clock = FakeTimers.install()
   t.teardown(clock.uninstall.bind(clock))
@@ -21,7 +21,11 @@ test('multiple reconnect', (t) => {
 
   client.request({ path: '/', method: 'GET' }, (err, data) => {
     t.error(err)
-    data.body.resume()
+    data.body
+      .resume()
+      .on('end', () => {
+        t.pass()
+      })
   })
 
   let n = 0

test/client.js

@@ -620,7 +620,7 @@ test('url-like url', (t) => {
 })
 
 test('multiple destroy callback', (t) => {
-  t.plan(3)
+  t.plan(4)
 
   const server = createServer((req, res) => {
     res.end()
@@ -637,7 +637,11 @@ test('multiple destroy callback', (t) => {
 
     client.request({ path: '/', method: 'GET' }, (err, data) => {
       t.error(err)
-      data.body.resume()
+      data.body
+        .resume()
+        .on('error', () => {
+          t.pass()
+        })
       client.destroy(new Error(), (err) => {
         t.error(err)
       })
@@ -649,7 +653,7 @@ test('multiple destroy callback', (t) => {
 })
 
 test('only one streaming req at a time', (t) => {
-  t.plan(6)
+  t.plan(7)
 
   const server = createServer((req, res) => {
     req.pipe(res)
@@ -693,15 +697,19 @@ test('only one streaming req at a time', (t) => {
         })
       }, (err, data) => {
         t.error(err)
-        data.body.resume()
+        data.body
+          .resume()
+          .on('end', () => {
+            t.pass()
+          })
       })
       t.strictEqual(client.busy, true)
     })
   })
 })
 
 test('300 requests succeed', (t) => {
-  t.plan(300 * 2)
+  t.plan(300 * 3)
 
   const server = createServer((req, res) => {
     res.end('asd')
@@ -720,6 +728,8 @@ test('300 requests succeed', (t) => {
         t.error(err)
         data.body.on('data', (chunk) => {
           t.strictEqual(chunk.toString(), 'asd')
+        }).on('end', () => {
+          t.pass()
         })
       })
     }

test/pool.js

@@ -440,7 +440,7 @@ test('pool pipeline args validation', (t) => {
 })
 
 test('300 requests succeed', (t) => {
-  t.plan(300 * 2)
+  t.plan(300 * 3)
 
   const server = createServer((req, res) => {
     res.end('asd')
@@ -461,6 +461,8 @@ test('300 requests succeed', (t) => {
         t.error(err)
         data.body.on('data', (chunk) => {
           t.strictEqual(chunk.toString(), 'asd')
+        }).on('end', () => {
+          t.pass()
         })
       })
     }

test/request-timeout.js

@@ -70,7 +70,7 @@ test('request timeout immutable opts', (t) => {
 })
 
 test('Subsequent request starves', (t) => {
-  t.plan(2)
+  t.plan(3)
 
   const clock = FakeTimers.install()
   t.teardown(clock.uninstall.bind(clock))
@@ -89,6 +89,11 @@ test('Subsequent request starves', (t) => {
 
     client.request({ path: '/', method: 'GET' }, (err, response) => {
       t.error(err)
+      response.body
+        .resume()
+        .on('end', () => {
+          t.pass()
+        })
     })
 
     client.request({ path: '/', method: 'GET', requestTimeout: 50 }, (err, response) => {