destroy request on error to avoid hanging sockets

package.json

@@ -3,7 +3,7 @@
   "engines": {
     "node": ">=20"
   },
-  "version": "1.3.1",
+  "version": "1.3.2",
   "description": "Hyperdrive appliance client library",
   "main": "lib/index.js",
   "types": "lib/index.d.ts",

src/hdcontroller.ts

@@ -47,11 +47,13 @@ function _createRequest(req: http.RequestOptions, log: werelogs.RequestLogger,
     }).on('error', (err: HDProxydError) => {
         if (!callbackCalled) {
             callbackCalled = true;
+            request.destroy(err);
             return callback(err);
         }
         if (err.code !== 'ERR_SOCKET_TIMEOUT') {
             log.error('got socket error after response', { err });
         }
+        request.destroy(err);
         return null;
     });
 
@@ -280,7 +282,7 @@ export class HDProxydClient {
                     method: '_handleRequest',
                     component: 'sproxydclient',
                 });
-                request.end();
+                request.destroy(err);
             });
         } else {
             headers['content-length'] = isBatchDelete ? size : 0;

tests/unit/hdSocketCleanup.spec.ts

@@ -0,0 +1,311 @@
+'use strict';
+
+/**
+ * Regression tests: after the HTTP response callback has run, Node may still emit
+ * ClientRequest 'error' (e.g. ERR_SOCKET_TIMEOUT on keep-alive / agent sockets).
+ * HDProxydClient must call request.destroy() so FDs do not accumulate under retries.
+ *
+ */
+
+import * as assert from 'assert';
+import * as sinon from 'sinon';
+import * as stream from 'stream';
+import * as http from 'http';
+import { AddressInfo } from 'net';
+import { HDProxydClient, HDProxydOptions } from '../../src/hdcontroller';
+
+describe('HDProxydClient — outbound socket cleanup after response', () => {
+    let sandbox: sinon.SinonSandbox;
+    let server: http.Server;
+
+    beforeEach(() => {
+        sandbox = sinon.createSandbox();
+        server = http.createServer((req, res) => {
+            if (req.url?.startsWith('/store/')) {
+                res.writeHead(200, { 'scal-key': 'test-key' });
+                res.end('ok');
+                return;
+            }
+            // eslint-disable-next-line no-param-reassign
+            res.statusCode = 404;
+            res.end();
+        });
+    });
+
+    afterEach(done => {
+        sandbox.restore();
+        const s = server as http.Server & { closeAllConnections?: () => void };
+        if (typeof s.closeAllConnections === 'function') {
+            s.closeAllConnections();
+        }
+        server.close(() => done());
+    });
+
+    function listenServer(cb: (port: number) => void): void {
+        server.listen(0, '127.0.0.1', () => {
+            const addr = server.address() as AddressInfo;
+            cb(addr.port);
+        });
+    }
+
+    type DestroyCapture = {
+        destroyCalled: boolean;
+        destroyCallCount: number;
+        lastDestroyArg?: unknown;
+        firstDestroyArgFromStream?: unknown;
+    };
+
+
+    function stubRequestAndCaptureDestroy(options: {
+        assignCapture: (cap: DestroyCapture) => void;
+        postResponseError?: {
+            code: 'ERR_SOCKET_TIMEOUT' | 'ECONNRESET';
+            message: string;
+        };
+    }): void {
+        const origRequest = http.request;
+        sandbox.stub(http, 'request').callsFake(((opts, cb) => {
+            const req = origRequest(
+                opts as string | http.RequestOptions,
+                cb as (res: http.IncomingMessage) => void,
+            );
+            const destroyCapture: DestroyCapture = { destroyCalled: false, destroyCallCount: 0 };
+            options.assignCapture(destroyCapture);
+            const innerDestroy = req.destroy.bind(req);
+            req.destroy = (error?: Error) => {
+                destroyCapture.destroyCalled = true;
+                destroyCapture.destroyCallCount += 1;
+                destroyCapture.lastDestroyArg = error;
+                if (destroyCapture.firstDestroyArgFromStream === undefined) {
+                    destroyCapture.firstDestroyArgFromStream = error;
+                }
+                return innerDestroy();
+            };
+            if (options.postResponseError) {
+                req.once('response', () => {
+                    process.nextTick(() => {
+                        const err = new Error(options.postResponseError!.message) as NodeJS.ErrnoException;
+                        err.code = options.postResponseError!.code;
+                        req.emit('error', err);
+                    });
+                });
+            }
+            return req;
+        }) as typeof http.request);
+    }
+
+    function countAgentSockets(agent: unknown): number {
+        const a = agent as {
+            sockets?: Record<string, unknown>;
+            freeSockets?: Record<string, unknown>;
+        };
+        const countPool = (pool?: Record<string, unknown>) => {
+            if (!pool) {
+                return 0;
+            }
+            return (Object.values(pool) as unknown[]).reduce<number>((sum, entry) => {
+                if (Array.isArray(entry)) {
+                    return sum + entry.length;
+                }
+                return sum + 1;
+            }, 0);
+        };
+        return countPool(a.sockets) + countPool(a.freeSockets);
+    }
+
+    it('calls ClientRequest.destroy when ERR_SOCKET_TIMEOUT fires after response', done => {
+        let destroyCapture: DestroyCapture | undefined;
+        let finished = false;
+        const finish = (e?: Error) => {
+            if (finished) {return;}
+            finished = true;
+            done(e);
+        };
+        stubRequestAndCaptureDestroy({
+            postResponseError: {
+                code: 'ERR_SOCKET_TIMEOUT',
+                message: 'simulated idle timeout',
+            },
+            assignCapture: (capture: DestroyCapture) => { destroyCapture = capture; },
+        });
+
+        listenServer(port => {
+            const h = new HDProxydClient({
+                bootstrap: [`127.0.0.1:${port}`],
+            } as HDProxydOptions);
+            h.get('anykey', [], '', (err, st) => {
+                assert.ifError(err);
+                assert.ok(st);
+                const body = st as stream.Readable;
+                body.on('error', () => {});
+                body.resume();
+                setImmediate(() => {
+                    try {
+                        assert.ok(destroyCapture!, 'capture should be set');
+                        assert.ok(destroyCapture!.destroyCalled, 'request.destroy must run for post-response socket error');
+                        assert.strictEqual(destroyCapture!.destroyCallCount, 1, 'request.destroy should run once');
+                        const arg0 = destroyCapture!.lastDestroyArg as NodeJS.ErrnoException;
+                        assert.strictEqual(arg0?.code, 'ERR_SOCKET_TIMEOUT');
+                        h.destroy();
+                        finish();
+                    } catch (e) {
+                        h.destroy();
+                        finish(e as Error);
+                    }
+                });
+            });
+        });
+    });
+
+    it('calls ClientRequest.destroy when a non-timeout error fires after response', done => {
+        let destroyCapture: DestroyCapture | undefined;
+        let finished = false;
+        const finish = (e?: Error) => {
+            if (finished) { return; }
+            finished = true;
+            done(e);
+        };
+        stubRequestAndCaptureDestroy({
+            postResponseError: {
+                code: 'ECONNRESET',
+                message: 'simulated reset',
+            },
+            assignCapture: (capture: DestroyCapture) => { destroyCapture = capture; },
+        });
+
+        listenServer(port => {
+            const h = new HDProxydClient({
+                bootstrap: [`127.0.0.1:${port}`],
+            } as HDProxydOptions);
+            h.get('anykey', [], '', (err, st) => {
+                assert.ifError(err);
+                assert.ok(st);
+                const body = st as stream.Readable;
+                body.on('error', () => {});
+                body.resume();
+                setImmediate(() => {
+                    try {
+                        assert.ok(destroyCapture, 'capture should be set');
+                        assert.ok(destroyCapture!.destroyCalled, 'request.destroy must run');
+                        assert.strictEqual(destroyCapture!.destroyCallCount, 1, 'request.destroy should run once');
+                        const arg0 = destroyCapture!.lastDestroyArg as NodeJS.ErrnoException;
+                        assert.strictEqual(arg0?.code, 'ECONNRESET');
+                        h.destroy();
+                        finish();
+                    } catch (e) {
+                        h.destroy();
+                        finish(e as Error);
+                    }
+                });
+            });
+        });
+    });
+
+    it('calls ClientRequest.destroy when upload stream errors mid-transfer (PUT stream path)', done => {
+        let destroyCapture: DestroyCapture | undefined;
+        let finished = false;
+        const finish = (e?: Error) => {
+            if (finished) {return;}
+            finished = true;
+            done(e);
+        };
+        const injectedError = new Error('simulated upload stream failure');
+        stubRequestAndCaptureDestroy({
+            assignCapture: (capture: DestroyCapture) => { destroyCapture = capture; },
+        });
+
+        listenServer(port => {
+            const h = new HDProxydClient({
+                bootstrap: [`127.0.0.1:${port}`],
+            } as HDProxydOptions);
+            const upload = new stream.Readable({
+                read() {
+                    this.push(Buffer.from('hello'));
+                    this.destroy(injectedError);
+                },
+            });
+            h.put(upload, 5, {
+                bucketName: 'test',
+                owner: 'test',
+                namespace: 'zenko',
+            }, '', () => {
+                setImmediate(() => {
+                    try {
+                        assert.ok(destroyCapture, 'capture should be set');
+                        assert.ok(destroyCapture!.destroyCalled, 'request.destroy must run for upload stream error');
+                        // Node may issue a second destroy from its internal pipe error handling; 
+                        // we are allowing both while requiring our stream error to trigger first.
+                        assert.ok(
+                            destroyCapture!.destroyCallCount === 1 || destroyCapture!.destroyCallCount === 2,
+                            `request.destroy should run once (or twice at most if http request also errors); 
+                            callCount=${destroyCapture!.destroyCallCount}`,
+                        );
+                        assert.strictEqual(destroyCapture!.firstDestroyArgFromStream, injectedError);
+                        h.destroy();
+                        finish();
+                    } catch (e) {
+                        h.destroy();
+                        finish(e as Error);
+                    }
+                });
+            });
+        });
+    });
+
+    it('does not accumulate agent sockets across repeated post-response ECONNRESET errors', done => {
+        let finished = false;
+        const finish = (e?: Error) => {
+            if (finished) {return;}
+            finished = true;
+            done(e);
+        };
+        stubRequestAndCaptureDestroy({
+            postResponseError: {
+                code: 'ECONNRESET',
+                message: 'simulated reset for leak regression',
+            },
+            assignCapture: () => {},
+        });
+
+        listenServer(port => {
+            const h = new HDProxydClient({
+                bootstrap: [`127.0.0.1:${port}`],
+            } as HDProxydOptions);
+            const baselineSockets = countAgentSockets((h as unknown as { httpAgent: unknown }).httpAgent);
+            let remaining = 20;
+
+            const runOnce = () => {
+                h.get('anykey', [], '', (err, st) => {
+                    if (err) {
+                        h.destroy();
+                        return finish(err);
+                    }
+                    assert.ok(st);
+                    const body = st as stream.Readable;
+                    body.on('error', () => {});
+                    body.resume();
+                    return setImmediate(() => {
+                        remaining -= 1;
+                        if (remaining === 0) {
+                            const afterSockets = countAgentSockets((h as unknown as { httpAgent: unknown }).httpAgent);
+                            try {
+                                assert.ok(
+                                    afterSockets <= baselineSockets + 2,
+                                    `socket count should stay bounded (baseline=${baselineSockets}, 
+                                    after=${afterSockets})`,
+                                );
+                                h.destroy();
+                                return finish();
+                            } catch (e) {
+                                h.destroy();
+                                return finish(e as Error);
+                            }
+                        }
+                        return runOnce();
+                    });
+                });
+            };
+            runOnce();
+        });
+    });
+});