fix: destroy IncomingMessage and invalid assertion (#834)

Change so that we destroy IncomingMessage without closing the
socket. Also fixes an invalid assertion that was triggered by
the test.

Author: ronag

lib/client.js

@@ -1520,8 +1520,7 @@ function write (client, request) {
         socket.write('\r\n0\r\n\r\n', 'ascii')
       }
 
-      assert.strictEqual(socket[kParser].timeoutType, TIMEOUT_HEADERS)
-      if (socket[kParser].timeout) {
+      if (socket[kParser].timeout && socket[kParser].timeoutType === TIMEOUT_HEADERS) {
         // istanbul ignore else: only for jest
         if (socket[kParser].timeout.refresh) {
           socket[kParser].timeout.refresh()

lib/core/util.js

@@ -130,9 +130,11 @@ function destroy (stream, err) {
   }
 
   if (typeof stream.destroy === 'function') {
-    if (err || Object.getPrototypeOf(stream).constructor !== IncomingMessage) {
-      stream.destroy(err)
+    if (Object.getPrototypeOf(stream).constructor === IncomingMessage) {
+      // See: https://github.com/nodejs/node/pull/38505/files
+      stream.socket = null
     }
+    stream.destroy(err)
   } else if (err) {
     process.nextTick((stream, err) => {
       stream.emit('error', err)

test/http-req-destroy.js

@@ -0,0 +1,56 @@
+'use strict'
+
+const { test } = require('tap')
+const undici = require('..')
+const { createServer } = require('http')
+const { Readable } = require('stream')
+
+test('do not kill req socket', (t) => {
+  t.plan(3)
+
+  const server1 = createServer((req, res) => {
+    undici.request(`http://localhost:${server2.address().port}`, {
+      method: 'POST',
+      body: req
+    }, (err, response) => {
+      t.error(err)
+      setTimeout(() => {
+        response.body.on('data', buf => {
+          res.write(buf)
+          setTimeout(() => {
+            res.end()
+          }, 100)
+        })
+      }, 100)
+    })
+  })
+  t.teardown(server1.close.bind(server1))
+
+  const server2 = createServer((req, res) => {
+    setTimeout(() => {
+      req.pipe(res)
+    }, 100)
+  })
+  t.teardown(server2.close.bind(server2))
+
+  server1.listen(0, () => {
+    const r = new Readable({ read () {} })
+    r.push('hello')
+    undici.request(`http://localhost:${server1.address().port}`, {
+      method: 'POST',
+      body: r
+    }, (err, response) => {
+      t.error(err)
+      const bufs = []
+      response.body.on('data', (buf) => {
+        bufs.push(buf)
+        r.push(null)
+      })
+      response.body.on('end', () => {
+        t.equal('hello', Buffer.concat(bufs).toString('utf8'))
+      })
+    })
+  })
+
+  server2.listen(0)
+})