diff --git a/README.md b/README.md index 6a7ee5a..e7e3e7e 100644 --- a/README.md +++ b/README.md @@ -56,6 +56,12 @@ $ npm install agentkeepalive --save * `socketActiveTTL` {Number} Sets the socket active time to live, even if it's in use. If not set, the behaviour keeps the same (the socket will be released only when free) Default = `null`. + * `testOnBorrow` {Boolean} When set to `true`, validates free sockets using wall-clock + time before reuse. Sockets idle longer than their effective `freeSocketTimeout` are + destroyed and replaced with fresh connections. This is essential for environments where + the process is suspended between invocations (e.g., AWS Lambda, Azure Functions) and + `socket.setTimeout()` callbacks do not fire during the freeze. + Default = `false`. ## Usage @@ -178,6 +184,35 @@ const req = http This behavior is consistent with Node.js core. But through `agentkeepalive`, you can use this feature in older Node.js version. +### Usage with Serverless (AWS Lambda, Azure Functions) + +In serverless environments, the process is frozen between invocations and `socket.setTimeout()` callbacks do not fire. Enable `testOnBorrow` to validate sockets using wall-clock time before reuse: + +```js +const https = require('https'); +const HttpsAgent = require('agentkeepalive').HttpsAgent; + +const agent = new HttpsAgent({ + keepAlive: true, + freeSocketTimeout: 30000, + testOnBorrow: true, // validate sockets against real elapsed time before reuse +}); + +// Use in your Lambda handler +exports.handler = async (event) => { + const res = await new Promise((resolve, reject) => { + https.get('https://api.example.com/data', { agent }, (res) => { + res.resume(); // drain the response to free the socket back to the pool + res.on('end', () => resolve(res)); + res.on('error', reject); + }).on('error', reject); + }); + // ... +}; +``` + +Without `testOnBorrow`, sockets that were idle during a process freeze may appear valid (their timeout timer was frozen too) but the server has already closed the connection — resulting in `EPIPE` or `ECONNRESET` errors. + ## [Benchmark](https://github.com/node-modules/agentkeepalive/tree/master/benchmark) run the benchmark: diff --git a/index.d.ts b/index.d.ts index f4dbcec..fb9ac0d 100644 --- a/index.d.ts +++ b/index.d.ts @@ -23,6 +23,8 @@ interface Constants { SOCKET_NAME: Symbol; SOCKET_REQUEST_COUNT: Symbol; SOCKET_REQUEST_FINISHED_COUNT: Symbol; + SOCKET_ENTER_FREE_POOL_TIME: Symbol; + SOCKET_ENTER_FREE_POOL_TIMEOUT: Symbol; } /** @@ -49,6 +51,7 @@ declare namespace AgentKeepAlive { freeSocketKeepAliveTimeout?: number | undefined; timeout?: number | undefined; socketActiveTTL?: number | undefined; + testOnBorrow?: boolean | undefined; } export interface HttpOptions extends http.AgentOptions, CommonHttpOption { } diff --git a/lib/agent.js b/lib/agent.js index 8bd354e..b1d894a 100644 --- a/lib/agent.js +++ b/lib/agent.js @@ -11,6 +11,8 @@ const { SOCKET_NAME, SOCKET_REQUEST_COUNT, SOCKET_REQUEST_FINISHED_COUNT, + SOCKET_ENTER_FREE_POOL_TIME, + SOCKET_ENTER_FREE_POOL_TIMEOUT, } = require('./constants'); // OriginalAgent come from @@ -159,12 +161,67 @@ class Agent extends OriginalAgent { socket[SOCKET_NAME], socket[SOCKET_REQUEST_COUNT], socket[SOCKET_REQUEST_FINISHED_COUNT], customTimeout); return false; } + + // When testOnBorrow is enabled, stamp the wall-clock time and effective timeout + // so addRequest() can validate sockets using real elapsed time instead of timers. + // This is needed in environments where the process is frozen (e.g., AWS Lambda) + // and socket.setTimeout() callbacks do not fire. + if (this.options.testOnBorrow) { + socket[SOCKET_ENTER_FREE_POOL_TIME] = Date.now(); + socket[SOCKET_ENTER_FREE_POOL_TIMEOUT] = customTimeout; + } + if (socket.timeout !== customTimeout) { socket.setTimeout(customTimeout); } return true; } + /** + * Remove sockets from the free pool that have been idle longer than + * their effective timeout using wall-clock time. This handles environments + * where the process is frozen between invocations (e.g., AWS Lambda) and + * socket.setTimeout() callbacks do not fire. + * @param {string} name - The socket pool key (from getName()). + */ + _purgeStaleFreeSockets(name) { + const freeSockets = this.freeSockets[name]; + if (!freeSockets) return; + + const now = Date.now(); + for (let i = freeSockets.length - 1; i >= 0; i--) { + const socket = freeSockets[i]; + const enterTime = socket[SOCKET_ENTER_FREE_POOL_TIME]; + const timeout = socket[SOCKET_ENTER_FREE_POOL_TIMEOUT]; + if (enterTime && timeout && (now - enterTime) > timeout) { + debug('%s(requests: %s, finished: %s) %dms idle, %dms threshold, destroying stale socket', + socket[SOCKET_NAME], socket[SOCKET_REQUEST_COUNT], socket[SOCKET_REQUEST_FINISHED_COUNT], + now - enterTime, timeout); + freeSockets.splice(i, 1); + this.timeoutSocketCount++; + socket.destroy(); + } + } + if (freeSockets.length === 0) { + delete this.freeSockets[name]; + } + } + + addRequest(req, options, ...args) { + if (this.options.testOnBorrow) { + // Node's super.addRequest() merges agent options over request options before + // calling getName(). We must replicate that merge here so our pool-key matches + // the one Node uses to store sockets (important for HTTPS options like ca/cert/key). + const opts = typeof options === 'string' + ? { host: options, port: args[0], localAddress: args[1] } + : options; + const effectiveOpts = { ...opts, ...this.options }; + this._purgeStaleFreeSockets(this.getName(effectiveOpts)); + } + // Pass original arguments untouched — let Node handle them as it always did. + super.addRequest(req, options, ...args); + } + // only call on addRequest reuseSocket(...args) { // reuseSocket(socket, req) diff --git a/lib/constants.js b/lib/constants.js index ca7ab97..8fe04e1 100644 --- a/lib/constants.js +++ b/lib/constants.js @@ -11,4 +11,6 @@ module.exports = { SOCKET_NAME: Symbol('agentkeepalive#socketName'), SOCKET_REQUEST_COUNT: Symbol('agentkeepalive#socketRequestCount'), SOCKET_REQUEST_FINISHED_COUNT: Symbol('agentkeepalive#socketRequestFinishedCount'), + SOCKET_ENTER_FREE_POOL_TIME: Symbol('agentkeepalive#socketEnterFreePoolTime'), + SOCKET_ENTER_FREE_POOL_TIMEOUT: Symbol('agentkeepalive#socketEnterFreePoolTimeout'), }; diff --git a/test/wall-clock-free-socket-timeout.test.js b/test/wall-clock-free-socket-timeout.test.js new file mode 100644 index 0000000..7fc91ee --- /dev/null +++ b/test/wall-clock-free-socket-timeout.test.js @@ -0,0 +1,268 @@ +'use strict'; + +const http = require('http'); +const https = require('https'); +const fs = require('fs'); +const assert = require('assert'); +const Agent = require('..'); +const HttpsAgent = require('..').HttpsAgent; +const { + SOCKET_ENTER_FREE_POOL_TIME, +} = require('../lib/constants'); + +describe('test/wall-clock-free-socket-timeout.test.js', () => { + let app; + let port; + + before(done => { + app = http.createServer((req, res) => { + res.end('ok'); + }); + app.listen(0, () => { + port = app.address().port; + done(); + }); + }); + + after(done => { + app.close(done); + }); + + it('should stamp SOCKET_ENTER_FREE_POOL_TIME when socket enters free pool', done => { + const agent = new Agent({ + keepAlive: true, + testOnBorrow: true, + freeSocketTimeout: 5000, + }); + + http.get({ agent, port, path: '/' }, res => { + const socket = res.socket; + res.resume(); + res.on('end', () => { + setImmediate(() => { + assert(socket[SOCKET_ENTER_FREE_POOL_TIME] > 0, + 'should have a positive timestamp'); + assert(Date.now() - socket[SOCKET_ENTER_FREE_POOL_TIME] < 5000, + 'timestamp should be recent (within 5s to tolerate slow CI)'); + agent.destroy(); + done(); + }); + }); + }); + }); + + it('should reuse socket when idle time is within threshold', done => { + const agent = new Agent({ + keepAlive: true, + testOnBorrow: true, + freeSocketTimeout: 5000, + }); + + http.get({ agent, port, path: '/' }, res => { + const socket1 = res.socket; + res.resume(); + res.on('end', () => { + setImmediate(() => { + http.get({ agent, port, path: '/' }, res2 => { + assert.strictEqual(res2.socket, socket1, + 'should reuse the socket when idle time is within threshold'); + res2.resume(); + res2.on('end', () => { + agent.destroy(); + done(); + }); + }); + }); + }); + }); + }); + + it('should destroy socket when idle time exceeds freeSocketTimeout (simulated freeze)', done => { + const agent = new Agent({ + keepAlive: true, + testOnBorrow: true, + freeSocketTimeout: 1000, + }); + + http.get({ agent, port, path: '/' }, res => { + const socket1 = res.socket; + res.resume(); + res.on('end', () => { + setImmediate(() => { + // Simulate a Lambda freeze by setting the timestamp far in the past + socket1[SOCKET_ENTER_FREE_POOL_TIME] = Date.now() - 10000; + + http.get({ agent, port, path: '/' }, res2 => { + // addRequest purges stale sockets before selecting, so a new socket is created + assert.notStrictEqual(res2.socket, socket1, + 'should not reuse stale socket'); + assert(socket1.destroyed, 'stale socket should be destroyed'); + res2.resume(); + res2.on('end', () => { + agent.destroy(); + done(); + }); + }); + }); + }); + }); + }); + + it('should handle multiple stale sockets in the pool', done => { + const agent = new Agent({ + keepAlive: true, + testOnBorrow: true, + freeSocketTimeout: 1000, + maxSockets: 5, + maxFreeSockets: 5, + }); + + let completed = 0; + const sockets = []; + + function onComplete() { + completed++; + if (completed < 2) return; + + setImmediate(() => { + assert.strictEqual(sockets.length, 2); + assert(sockets[0][SOCKET_ENTER_FREE_POOL_TIME] > 0); + assert(sockets[1][SOCKET_ENTER_FREE_POOL_TIME] > 0); + + // Simulate freeze + sockets.forEach(s => { + s[SOCKET_ENTER_FREE_POOL_TIME] = Date.now() - 5000; + }); + + // addRequest will purge both stale sockets and create a new connection + http.get({ agent, port, path: '/' }, res => { + assert.notStrictEqual(res.socket, sockets[0], + 'should not reuse first stale socket'); + assert.notStrictEqual(res.socket, sockets[1], + 'should not reuse second stale socket'); + assert(sockets[0].destroyed, 'first stale socket should be destroyed'); + assert(sockets[1].destroyed, 'second stale socket should be destroyed'); + res.resume(); + res.on('end', () => { + agent.destroy(); + done(); + }); + }); + }); + } + + http.get({ agent, port, path: '/' }, res => { + sockets.push(res.socket); + res.resume(); + res.on('end', onComplete); + }); + + http.get({ agent, port, path: '/' }, res => { + sockets.push(res.socket); + res.resume(); + res.on('end', onComplete); + }); + }); + + it('should not destroy socket when freeSocketTimeout is 0 (disabled)', done => { + const agent = new Agent({ + keepAlive: true, + testOnBorrow: true, + freeSocketTimeout: 0, + }); + + http.get({ agent, port, path: '/' }, res => { + const socket1 = res.socket; + res.resume(); + res.on('end', () => { + setImmediate(() => { + socket1[SOCKET_ENTER_FREE_POOL_TIME] = Date.now() - 999999; + + http.get({ agent, port, path: '/' }, res2 => { + assert.strictEqual(res2.socket, socket1, + 'should reuse when freeSocketTimeout is disabled'); + res2.resume(); + res2.on('end', () => { + agent.destroy(); + done(); + }); + }); + }); + }); + }); + }); + + it('should work correctly with HTTPS agent', done => { + const httpsApp = https.createServer({ + key: fs.readFileSync(__dirname + '/fixtures/agenttest-key.pem'), + cert: fs.readFileSync(__dirname + '/fixtures/agenttest-cert.pem'), + }, (req, res) => { + res.end('ok'); + }); + + httpsApp.listen(0, () => { + const httpsPort = httpsApp.address().port; + const agent = new HttpsAgent({ + keepAlive: true, + testOnBorrow: true, + freeSocketTimeout: 1000, + rejectUnauthorized: false, + }); + + https.get({ agent, port: httpsPort, path: '/', rejectUnauthorized: false }, res => { + const socket1 = res.socket; + res.resume(); + res.on('end', () => { + setImmediate(() => { + assert(socket1[SOCKET_ENTER_FREE_POOL_TIME] > 0, + 'HTTPS socket should have timestamp'); + + // Simulate freeze + socket1[SOCKET_ENTER_FREE_POOL_TIME] = Date.now() - 5000; + + https.get({ agent, port: httpsPort, path: '/', rejectUnauthorized: false }, res2 => { + assert.notStrictEqual(res2.socket, socket1, + 'should not reuse stale HTTPS socket'); + assert(socket1.destroyed, 'stale HTTPS socket should be destroyed'); + res2.resume(); + res2.on('end', () => { + agent.destroy(); + httpsApp.close(done); + }); + }); + }); + }); + }); + }); + }); + + it('should not purge stale sockets when testOnBorrow is false (default)', done => { + const agent = new Agent({ + keepAlive: true, + freeSocketTimeout: 1000, + // testOnBorrow not set — defaults to false + }); + + http.get({ agent, port, path: '/' }, res => { + const socket1 = res.socket; + res.resume(); + res.on('end', () => { + setImmediate(() => { + // Manually stamp as if testOnBorrow were enabled (simulating stale socket) + socket1[SOCKET_ENTER_FREE_POOL_TIME] = Date.now() - 10000; + + // Without testOnBorrow, addRequest does not purge — socket is reused + http.get({ agent, port, path: '/' }, res2 => { + assert.strictEqual(res2.socket, socket1, + 'should reuse socket when testOnBorrow is disabled (no purge)'); + res2.resume(); + res2.on('end', () => { + agent.destroy(); + done(); + }); + }); + }); + }); + }); + }); +});