Add tests for destroy() after post-response ClientRequest errors

Issue: HD-4352

Author: benzekrimaha

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();
+        });
+    });
+});