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,164 @@
+'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;
+            }
+            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 = { called: boolean; firstArg?: unknown };
+
+
+    function stubRequestWithPostResponseError(
+        code: 'ERR_SOCKET_TIMEOUT' | 'ECONNRESET',
+        message: string,
+        assignCapture: (cap: DestroyCapture) => void,
+    ): 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 cap: DestroyCapture = { called: false };
+            assignCapture(cap);
+            const innerDestroy = req.destroy.bind(req);
+            req.destroy = (error?: Error) => {
+                cap.called = true;
+                cap.firstArg = error;
+                return innerDestroy();
+            };
+            req.once('response', () => {
+                process.nextTick(() => {
+                    const err = new Error(message) as NodeJS.ErrnoException;
+                    err.code = code;
+                    req.emit('error', err);
+                });
+            });
+            return req;
+        }) as typeof http.request);
+    }
+
+    it('calls ClientRequest.destroy when ERR_SOCKET_TIMEOUT fires after response', done => {
+        let cap: DestroyCapture | undefined;
+        let finished = false;
+        const finish = (e?: Error) => {
+            if (finished) return;
+            finished = true;
+            done(e);
+        };
+        stubRequestWithPostResponseError(
+            'ERR_SOCKET_TIMEOUT',
+            'simulated idle timeout',
+            c => { cap = c; },
+        );
+
+        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(cap, 'capture should be set');
+                        assert.ok(cap!.called, 'request.destroy must run for post-response socket error');
+                        const arg0 = cap!.firstArg 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 cap: DestroyCapture | undefined;
+        let finished = false;
+        const finish = (e?: Error) => {
+            if (finished) return;
+            finished = true;
+            done(e);
+        };
+        stubRequestWithPostResponseError(
+            'ECONNRESET',
+            'simulated reset',
+            c => { cap = c; },
+        );
+
+        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(cap, 'capture should be set');
+                        assert.ok(cap!.called, 'request.destroy must run');
+                        const arg0 = cap!.firstArg as NodeJS.ErrnoException;
+                        assert.strictEqual(arg0?.code, 'ECONNRESET');
+                        h.destroy();
+                        finish();
+                    } catch (e) {
+                        h.destroy();
+                        finish(e as Error);
+                    }
+                });
+            });
+        });
+    });
+});