Add keep-alive connection tracking and reaping (#3619)

build/build-validation.js

@@ -15,6 +15,7 @@ const ajv = new Ajv({
 const defaultInitOptions = {
   connectionTimeout: 0, // 0 sec
   keepAliveTimeout: 5000, // 5 sec
+  forceCloseConnections: false, // keep-alive connections
   maxRequestsPerSocket: 0, // no limit
   requestTimeout: 0, // no limit
   bodyLimit: 1024 * 1024, // 1 MiB
@@ -49,6 +50,7 @@ const schema = {
   properties: {
     connectionTimeout: { type: 'integer', default: defaultInitOptions.connectionTimeout },
     keepAliveTimeout: { type: 'integer', default: defaultInitOptions.keepAliveTimeout },
+    forceCloseConnections: { type: 'boolean', default: defaultInitOptions.forceCloseConnections },
     maxRequestsPerSocket: { type: 'integer', default: defaultInitOptions.maxRequestsPerSocket, nullable: true },
     requestTimeout: { type: 'integer', default: defaultInitOptions.requestTimeout },
     bodyLimit: { type: 'integer', default: defaultInitOptions.bodyLimit },

docs/Reference/Server.md

@@ -13,6 +13,7 @@ describes the properties available in that options object.
   - [`https`](#https)
   - [`connectionTimeout`](#connectiontimeout)
   - [`keepAliveTimeout`](#keepalivetimeout)
+  - [`forceCloseConnections](#forcecloseconnections)
   - [`maxRequestsPerSocket`](#maxrequestspersocket)
   - [`requestTimeout`](#requesttimeout)
   - [`ignoreTrailingSlash`](#ignoretrailingslash)
@@ -124,6 +125,19 @@ use. Also, when `serverFactory` option is specified, this option is ignored.
 
 + Default: `5000` (5 seconds)
 
+### `forceCloseConnections`
+<a id="forcecloseconnections"></a>
+
+When set to `true` requests with the header `connection: keep-alive` will be
+tracked by the server. Upon [`close`](#close), the server will iterate the
+current persistent connections and [destroy their
+sockets](https://nodejs.org/dist/latest-v16.x/docs/api/net.html#socketdestroyerror).
+This means the server will shutdown immediately instead of waiting for existing
+persistent connections to timeout first. Important: connections are not
+inspected to determine if requests have been completed.
+
++ Default: `false`
+
 ### `maxRequestsPerSocket`
 <a id="factory-max-requests-per-socket"></a>
 

fastify.js

@@ -26,7 +26,8 @@ const {
   kOptions,
   kPluginNameChain,
   kSchemaErrorFormatter,
-  kErrorHandler
+  kErrorHandler,
+  kKeepAliveConnections
 } = require('./lib/symbols.js')
 
 const { createServer } = require('./lib/server')
@@ -45,6 +46,7 @@ const build404 = require('./lib/fourOhFour')
 const getSecuredInitialConfig = require('./lib/initialConfigValidation')
 const override = require('./lib/pluginOverride')
 const warning = require('./lib/warnings')
+const noopSet = require('./lib/noop-set')
 const { defaultInitOptions } = getSecuredInitialConfig
 
 const {
@@ -133,6 +135,7 @@ function fastify (options) {
   // Update the options with the fixed values
   options.connectionTimeout = options.connectionTimeout || defaultInitOptions.connectionTimeout
   options.keepAliveTimeout = options.keepAliveTimeout || defaultInitOptions.keepAliveTimeout
+  options.forceCloseConnections = typeof options.forceCloseConnections === 'boolean' ? options.forceCloseConnections : defaultInitOptions.forceCloseConnections
   options.maxRequestsPerSocket = options.maxRequestsPerSocket || defaultInitOptions.maxRequestsPerSocket
   options.requestTimeout = options.requestTimeout || defaultInitOptions.requestTimeout
   options.logger = logger
@@ -146,6 +149,7 @@ function fastify (options) {
   options.exposeHeadRoutes = exposeHeadRoutes
 
   const initialConfig = getSecuredInitialConfig(options)
+  const keepAliveConnections = options.forceCloseConnections === true ? new Set() : noopSet()
 
   let constraints = options.constraints
   if (options.versioning) {
@@ -176,7 +180,8 @@ function fastify (options) {
       maxParamLength: options.maxParamLength || defaultInitOptions.maxParamLength,
       caseSensitive: options.caseSensitive,
       buildPrettyMeta: defaultBuildPrettyMeta
-    }
+    },
+    keepAliveConnections
   })
 
   // 404 router, used for handling encapsulated 404 handlers
@@ -200,6 +205,7 @@ function fastify (options) {
       closing: false,
       started: false
     },
+    [kKeepAliveConnections]: keepAliveConnections,
     [kOptions]: options,
     [kChildren]: [],
     [kBodyLimit]: bodyLimit,
@@ -375,6 +381,15 @@ function fastify (options) {
       if (fastify[kState].listening) {
         // No new TCP connections are accepted
         instance.server.close(done)
+
+        for (const conn of fastify[kKeepAliveConnections]) {
+          // We must invoke the destroy method instead of merely unreffing
+          // the sockets. If we only unref, then the callback passed to
+          // `fastify.close` will never be invoked; nor will any of the
+          // registered `onClose` hooks.
+          conn.destroy()
+          fastify[kKeepAliveConnections].delete(conn)
+        }
       } else {
         done(null)
       }

lib/noop-set.js

@@ -0,0 +1,10 @@
+'use strict'
+
+module.exports = function noopSet () {
+  return {
+    [Symbol.iterator]: function * () {},
+    add () {},
+    delete () {},
+    has () { return true }
+  }
+}

lib/route.js

@@ -41,6 +41,7 @@ const {
 } = require('./symbols.js')
 
 function buildRouting (options) {
+  const { keepAliveConnections } = options
   const router = FindMyWay(options.config)
 
   let avvio
@@ -345,6 +346,17 @@ function buildRouting (options) {
       }
     }
 
+    // When server.forceCloseConnections is true, we will collect any requests
+    // that have indicated they want persistence so that they can be reaped
+    // on server close. Otherwise, the container is a noop container.
+    const connHeader = String.prototype.toLowerCase.call(req.headers.connection || '')
+    if (connHeader === 'keep-alive') {
+      if (keepAliveConnections.has(req.socket) === false) {
+        keepAliveConnections.add(req.socket)
+        req.socket.on('close', removeTrackedSocket.bind({ keepAliveConnections, socket: req.socket }))
+      }
+    }
+
     // we revert the changes in defaultRoute
     if (req.headers[kRequestAcceptVersion] !== undefined) {
       req.headers['accept-version'] = req.headers[kRequestAcceptVersion]
@@ -485,6 +497,15 @@ function preParsingHookRunner (functions, request, reply, cb) {
   next(null, request[kRequestPayloadStream])
 }
 
+/**
+ * Used within the route handler as a `net.Socket.close` event handler.
+ * The purpose is to remove a socket from the tracked sockets collection when
+ * the socket has naturally timed out.
+ */
+function removeTrackedSocket () {
+  this.keepAliveConnections.delete(this.socket)
+}
+
 function noop () { }
 
 module.exports = { buildRouting, validateBodyLimitOption }

lib/symbols.js

@@ -44,7 +44,8 @@ const keys = {
   kPluginNameChain: Symbol('fastify.pluginNameChain'),
   // This symbol is only meant to be used for fastify tests and should not be used for any other purpose
   kTestInternals: Symbol('fastify.testInternals'),
-  kErrorHandler: Symbol('fastify.errorHandler')
+  kErrorHandler: Symbol('fastify.errorHandler'),
+  kKeepAliveConnections: Symbol('fastify.keepAliveConnections')
 }
 
 module.exports = keys

package.json

@@ -16,7 +16,7 @@
     "lint:typescript": "eslint -c types/.eslintrc.json types/**/*.d.ts test/types/**/*.test-d.ts",
     "prepublishOnly": "tap --no-check-coverage test/internals/version.test.js",
     "test": "npm run lint && npm run unit && npm run test:typescript",
-    "test:ci": "npm run unit -- --cov --coverage-report=lcovonly && npm run test:typescript",
+    "test:ci": "npm run unit -- -R terse --cov --coverage-report=lcovonly && npm run test:typescript",
     "test:report": "npm run lint && npm run unit:report && npm run test:typescript",
     "test:typescript": "tsd",
     "unit": "tap -J test/*.test.js test/*/*.test.js",
@@ -129,6 +129,7 @@
     "@types/pino": "^6.0.1",
     "@typescript-eslint/eslint-plugin": "^4.5.0",
     "@typescript-eslint/parser": "^4.5.0",
+    "JSONStream": "^1.3.5",
     "ajv": "^6.0.0",
     "ajv-errors": "^1.0.1",
     "ajv-formats": "^2.1.1",
@@ -157,7 +158,6 @@
     "hsts": "^2.2.0",
     "http-errors": "^2.0.0",
     "ienoopen": "^1.1.0",
-    "JSONStream": "^1.3.5",
     "license-checker": "^25.0.1",
     "pem": "^1.14.4",
     "proxyquire": "^2.1.3",
@@ -173,7 +173,7 @@
     "then-sleep": "^1.0.1",
     "tsd": "^0.19.0",
     "typescript": "^4.0.2",
-    "undici": "^3.3.5",
+    "undici": "^3.3.6",
     "x-xss-protection": "^2.0.0",
     "yup": "^0.32.0"
   },
@@ -206,4 +206,4 @@
   "tsd": {
     "directory": "test/types"
   }
-}
+}

test/close.test.js

@@ -291,3 +291,41 @@ test('Cannot be reopened the closed server has listen callback', async t => {
     t.ok(err)
   })
 })
+
+test('shutsdown while keep-alive connections are active (non-async)', t => {
+  t.plan(5)
+
+  const timeoutTime = 2 * 60 * 1000
+  const fastify = Fastify({ forceCloseConnections: true })
+
+  fastify.server.setTimeout(timeoutTime)
+  fastify.server.keepAliveTimeout = timeoutTime
+
+  fastify.get('/', (req, reply) => {
+    reply.send({ hello: 'world' })
+  })
+
+  fastify.listen(0, (err, address) => {
+    t.error(err)
+
+    const client = new Client(
+      'http://localhost:' + fastify.server.address().port,
+      { keepAliveTimeout: 1 * 60 * 1000 }
+    )
+    client.request({ path: '/', method: 'GET' }, (err, response) => {
+      t.error(err)
+      t.equal(client.closed, false)
+
+      fastify.close((err) => {
+        t.error(err)
+
+        // Due to the nature of the way we reap these keep-alive connections,
+        // there hasn't been enough time before the server fully closed in order
+        // for the client to have seen the socket get destroyed. The mere fact
+        // that we have reached this callback is enough indication that the
+        // feature being tested works as designed.
+        t.equal(client.closed, false)
+      })
+    })
+  })
+})

test/noop-set.test.js

@@ -0,0 +1,19 @@
+'use strict'
+
+const tap = require('tap')
+const noopSet = require('../lib/noop-set')
+
+tap.test('does a lot of nothing', async t => {
+  const aSet = noopSet()
+  t.type(aSet, 'object')
+
+  const item = {}
+  aSet.add(item)
+  aSet.add({ another: 'item' })
+  aSet.delete(item)
+  t.equal(aSet.has(item), true)
+
+  for (const i of aSet) {
+    t.fail('should not have any items', i)
+  }
+})