fix: don't reset socket until pipeline is empty

lib/client.js

@@ -412,6 +412,7 @@ class Parser extends HTTPParser {
     this.statusCode = null
     this.upgrade = false
     this.headers = null
+    this.shouldKeepAlive = false
     this.read = 0
   }
 
@@ -513,13 +514,15 @@ class Parser extends HTTPParser {
 
     this.headers = util.parseHeaders(rawHeaders, this.headers)
     this.statusCode = statusCode
+    this.shouldKeepAlive = shouldKeepAlive
 
     if (upgrade || request.method === 'CONNECT') {
       this.upgrade = true
       return 2
     }
 
     if (!shouldKeepAlive) {
+      // Stop more requests from being dispatched.
       client[kReset] = true
     }
 
@@ -606,15 +609,14 @@ class Parser extends HTTPParser {
       // Response completed before request.
       util.destroy(socket, new InformationalError('request reset'))
     } else if (client[kReset]) {
-      // https://tools.ietf.org/html/rfc7231#section-4.3.1
-      // https://tools.ietf.org/html/rfc7231#section-4.3.2
-      // https://tools.ietf.org/html/rfc7231#section-4.3.5
-
-      // Sending a payload body on a request that does not
-      // expect it can cause undefined behavior on some
-      // servers and corrupt connection state. Do not
-      // re-use the connection for further requests.
-
+      // 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'))
+      }
+    } else if (!this.shouldKeepAlive) {
       util.destroy(socket, new InformationalError('request reset'))
     } else {
       socket.resume()
@@ -686,6 +688,7 @@ function onSocketClose () {
     // Retry all idempotent requests except for the one
     // at the head of the pipeline.
 
+    // TODO: Allow retry if err.code === UND_ERR_INFO
     client[kQueue][client[kRunningIdx]++].onError(err)
 
     const retryRequests = []

lib/request.js

@@ -213,7 +213,14 @@ class Request {
     }
 
     if (body && !this.expectsPayload && util.bodyLength(body) !== 0) {
-      // Reset if method does not expect payload.
+      // https://tools.ietf.org/html/rfc7231#section-4.3.1
+      // https://tools.ietf.org/html/rfc7231#section-4.3.2
+      // https://tools.ietf.org/html/rfc7231#section-4.3.5
+
+      // Sending a payload body on a request that does not
+      // expect it can cause undefined behavior on some
+      // servers and corrupt connection state. Do not
+      // re-use the connection for further requests.
       return true
     }
 

test/client-pipelining.js

@@ -386,7 +386,7 @@ test('pipelining non-idempotent w body', (t) => {
 })
 
 test('pipelining HEAD busy', (t) => {
-  t.plan(6)
+  t.plan(7)
 
   const server = createServer()
   server.on('request', (req, res) => {
@@ -401,6 +401,11 @@ test('pipelining HEAD busy', (t) => {
     t.tearDown(client.close.bind(client))
 
     client[kConnect](() => {
+      let ended = false
+      client.once('disconnect', () => {
+        t.strictEqual(ended, true)
+      })
+
       {
         const body = new Readable({
           read () { }
@@ -434,6 +439,7 @@ test('pipelining HEAD busy', (t) => {
           data.body
             .resume()
             .on('end', () => {
+              ended = true
               t.pass()
             })
         })
@@ -444,6 +450,71 @@ test('pipelining HEAD busy', (t) => {
   })
 })
 
+test('pipelining empty pipeline before reset', (t) => {
+  t.plan(7)
+
+  let c = 0
+  const server = createServer()
+  server.on('request', (req, res) => {
+    if (c++ === 0) {
+      res.end('asd')
+    } else {
+      setTimeout(() => {
+        res.end('asd')
+      }, 100)
+    }
+  })
+  t.tearDown(server.close.bind(server))
+
+  server.listen(0, () => {
+    const client = new Client(`http://localhost:${server.address().port}`, {
+      pipelining: 10
+    })
+    t.tearDown(client.close.bind(client))
+
+    client[kConnect](() => {
+      let ended = false
+      client.once('disconnect', () => {
+        t.strictEqual(ended, true)
+      })
+
+      const body = new Readable({
+        read () { }
+      })
+
+      client.request({
+        path: '/',
+        method: 'GET'
+      }, (err, data) => {
+        t.error(err)
+        data.body
+          .resume()
+          .on('end', () => {
+            t.pass()
+            body.push(null)
+          })
+      })
+      t.strictEqual(client.busy, false)
+
+      client.request({
+        path: '/',
+        method: 'HEAD',
+        body: 'asd'
+      }, (err, data) => {
+        t.error(err)
+        data.body
+          .resume()
+          .on('end', () => {
+            ended = true
+            t.pass()
+          })
+      })
+      t.strictEqual(client.busy, true)
+      t.strictEqual(client.running, 2)
+    })
+  })
+})
+
 test('pipelining idempotent busy', (t) => {
   t.plan(12)