Handle connectivity issues while reading the body (#1343)

test/unit/client.test.js

@@ -21,7 +21,7 @@
 
 const { test } = require('tap')
 const { URL } = require('url')
-const { Client, ConnectionPool, Transport } = require('../../index')
+const { Client, ConnectionPool, Transport, errors } = require('../../index')
 const { CloudConnectionPool } = require('../../lib/pool')
 const { buildServer } = require('../utils')
 
@@ -1191,3 +1191,55 @@ test('name property as symbol', t => {
 
   t.strictEqual(client.name, symbol)
 })
+
+// The nodejs http agent will try to wait for the whole
+// body to arrive before closing the request, so this
+// test might take some time.
+test('Bad content length', t => {
+  t.plan(3)
+
+  let count = 0
+  function handler (req, res) {
+    count += 1
+    const body = JSON.stringify({ hello: 'world' })
+    res.setHeader('Content-Type', 'application/json;utf=8')
+    res.setHeader('Content-Length', body.length + '')
+    res.end(body.slice(0, -5))
+  }
+
+  buildServer(handler, ({ port }, server) => {
+    const client = new Client({ node: `http://localhost:${port}`, maxRetries: 1 })
+    client.info((err, { body }) => {
+      t.ok(err instanceof errors.ConnectionError)
+      t.is(err.message, 'Response aborted while reading the body')
+      t.strictEqual(count, 2)
+      server.stop()
+    })
+  })
+})
+
+test('Socket destryed while reading the body', t => {
+  t.plan(3)
+
+  let count = 0
+  function handler (req, res) {
+    count += 1
+    const body = JSON.stringify({ hello: 'world' })
+    res.setHeader('Content-Type', 'application/json;utf=8')
+    res.setHeader('Content-Length', body.length + '')
+    res.write(body.slice(0, -5))
+    setTimeout(() => {
+      res.socket.destroy()
+    }, 500)
+  }
+
+  buildServer(handler, ({ port }, server) => {
+    const client = new Client({ node: `http://localhost:${port}`, maxRetries: 1 })
+    client.info((err, { body }) => {
+      t.ok(err instanceof errors.ConnectionError)
+      t.is(err.message, 'Response aborted while reading the body')
+      t.strictEqual(count, 2)
+      server.stop()
+    })
+  })
+})

test/unit/connection.test.js

@@ -21,7 +21,6 @@
 
 const { test } = require('tap')
 const { inspect } = require('util')
-const { createGzip, createDeflate } = require('zlib')
 const { URL } = require('url')
 const { Agent } = require('http')
 const hpagent = require('hpagent')
@@ -400,90 +399,6 @@ test('Send body as stream', t => {
   })
 })
 
-test('Should handle compression', t => {
-  t.test('gzip', t => {
-    t.plan(3)
-
-    function handler (req, res) {
-      res.writeHead(200, {
-        'Content-Type': 'application/json;utf=8',
-        'Content-Encoding': 'gzip'
-      })
-      intoStream(JSON.stringify({ hello: 'world' }))
-        .pipe(createGzip())
-        .pipe(res)
-    }
-
-    buildServer(handler, ({ port }, server) => {
-      const connection = new Connection({
-        url: new URL(`http://localhost:${port}`)
-      })
-      connection.request({
-        path: '/hello',
-        method: 'GET'
-      }, (err, res) => {
-        t.error(err)
-
-        t.match(res.headers, {
-          'content-type': 'application/json;utf=8',
-          'content-encoding': 'gzip'
-        })
-
-        var payload = ''
-        res.setEncoding('utf8')
-        res.on('data', chunk => { payload += chunk })
-        res.on('error', err => t.fail(err))
-        res.on('end', () => {
-          t.deepEqual(JSON.parse(payload), { hello: 'world' })
-          server.stop()
-        })
-      })
-    })
-  })
-
-  t.test('deflate', t => {
-    t.plan(3)
-
-    function handler (req, res) {
-      res.writeHead(200, {
-        'Content-Type': 'application/json;utf=8',
-        'Content-Encoding': 'deflate'
-      })
-      intoStream(JSON.stringify({ hello: 'world' }))
-        .pipe(createDeflate())
-        .pipe(res)
-    }
-
-    buildServer(handler, ({ port }, server) => {
-      const connection = new Connection({
-        url: new URL(`http://localhost:${port}`)
-      })
-      connection.request({
-        path: '/hello',
-        method: 'GET'
-      }, (err, res) => {
-        t.error(err)
-
-        t.match(res.headers, {
-          'content-type': 'application/json;utf=8',
-          'content-encoding': 'deflate'
-        })
-
-        var payload = ''
-        res.setEncoding('utf8')
-        res.on('data', chunk => { payload += chunk })
-        res.on('error', err => t.fail(err))
-        res.on('end', () => {
-          t.deepEqual(JSON.parse(payload), { hello: 'world' })
-          server.stop()
-        })
-      })
-    })
-  })
-
-  t.end()
-})
-
 test('Should not close a connection if there are open requests', t => {
   t.plan(4)
 

test/unit/transport.test.js

@@ -22,7 +22,7 @@
 const { test } = require('tap')
 const { URL } = require('url')
 const FakeTimers = require('@sinonjs/fake-timers')
-const { createGunzip } = require('zlib')
+const { createGunzip, gzipSync } = require('zlib')
 const os = require('os')
 const intoStream = require('into-stream')
 const {
@@ -1665,13 +1665,17 @@ test('Should cast to boolean HEAD request', t => {
 })
 
 test('Suggest compression', t => {
-  t.plan(2)
+  t.plan(3)
   function handler (req, res) {
     t.match(req.headers, {
       'accept-encoding': 'gzip,deflate'
     })
+
+    const body = gzipSync(JSON.stringify({ hello: 'world' }))
     res.setHeader('Content-Type', 'application/json;utf=8')
-    res.end(JSON.stringify({ hello: 'world' }))
+    res.setHeader('Content-Encoding', 'gzip')
+    res.setHeader('Content-Length', Buffer.byteLength(body))
+    res.end(body)
   }
 
   buildServer(handler, ({ port }, server) => {
@@ -1694,6 +1698,46 @@ test('Suggest compression', t => {
       path: '/hello'
     }, (err, { body }) => {
       t.error(err)
+      t.deepEqual(body, { hello: 'world' })
+      server.stop()
+    })
+  })
+})
+
+test('Broken compression', t => {
+  t.plan(2)
+  function handler (req, res) {
+    t.match(req.headers, {
+      'accept-encoding': 'gzip,deflate'
+    })
+
+    const body = gzipSync(JSON.stringify({ hello: 'world' }))
+    res.setHeader('Content-Type', 'application/json;utf=8')
+    res.setHeader('Content-Encoding', 'gzip')
+    // we are not setting the content length on purpose
+    res.end(body.slice(0, -5))
+  }
+
+  buildServer(handler, ({ port }, server) => {
+    const pool = new ConnectionPool({ Connection })
+    pool.addConnection(`http://localhost:${port}`)
+
+    const transport = new Transport({
+      emit: () => {},
+      connectionPool: pool,
+      serializer: new Serializer(),
+      maxRetries: 3,
+      requestTimeout: 30000,
+      sniffInterval: false,
+      sniffOnStart: false,
+      suggestCompression: true
+    })
+
+    transport.request({
+      method: 'GET',
+      path: '/hello'
+    }, (err, { body }) => {
+      t.ok(err)
       server.stop()
     })
   })