diff --git a/src/utils/axiosUtils.ts b/src/utils/axiosUtils.ts index f9f0d46..1dd64e2 100644 --- a/src/utils/axiosUtils.ts +++ b/src/utils/axiosUtils.ts @@ -40,6 +40,11 @@ export async function axiosGet( return await axios.get(url); } } catch (error) { + // Failed streaming requests still own a response socket. Release it before + // returning or throwing so registry fallback cannot keep the process alive. + if (axiosOptions.responseType === 'stream') { + error.response?.data?.destroy?.(); + } if (error.response?.status === 401) { return error.response; } else { diff --git a/test/utils/axiosUtils.test.ts b/test/utils/axiosUtils.test.ts index 3f5f733..8c39876 100644 --- a/test/utils/axiosUtils.test.ts +++ b/test/utils/axiosUtils.test.ts @@ -1,5 +1,7 @@ +import { Readable } from 'stream'; import axios from 'axios'; import { HttpsProxyAgent } from 'https-proxy-agent'; +import { DefaultRegistryClient } from '../../src/registry/DefaultRegistryClient'; import { axiosGet } from '../../src/utils/axiosUtils'; describe('axiosUtils', () => { @@ -12,10 +14,65 @@ describe('axiosUtils', () => { afterEach(() => { delete process.env.HTTPS_PROXY; - getSpy.mockClear(); + getSpy.mockReset(); }); describe('#axiosGet', () => { + it('releases failed response streams before falling back to another registry', async () => { + const failedStream = Readable.from(['Not found']); + const tarballStream = Readable.from(['tarball']); + getSpy + .mockRejectedValueOnce({ response: { status: 404, data: failedStream } }) + .mockResolvedValueOnce({ status: 200, data: tarballStream }); + // Explicitly use the built-in registries regardless of the environment. + const previousRegistry = process.env.FPL_REGISTRY; + delete process.env.FPL_REGISTRY; + try { + const result = await new DefaultRegistryClient({}).download('us.nlm.vsac', '0.24.0'); + expect(result).toBe(tarballStream); + expect(failedStream.destroyed).toBeTrue(); + expect(tarballStream.destroyed).toBeFalse(); + expect(getSpy).toHaveBeenCalledTimes(2); + } finally { + if (previousRegistry !== undefined) process.env.FPL_REGISTRY = previousRegistry; + tarballStream.destroy(); + } + }); + + it('releases a failed response stream and rethrows the original error', async () => { + const data = Readable.from(['Server error']); + const error = { response: { status: 500, data } }; + getSpy.mockRejectedValueOnce(error); + await expect( + axiosGet('https://example.org/package', { responseType: 'stream' }) + ).rejects.toBe(error); + expect(data.destroyed).toBeTrue(); + }); + + it('releases unauthorized response streams while still returning the response', async () => { + const data = Readable.from(['Unauthorized']); + const response = { status: 401, data }; + getSpy.mockRejectedValueOnce({ response }); + await expect( + axiosGet('https://example.org/package', { responseType: 'stream' }) + ).resolves.toBe(response); + expect(data.destroyed).toBeTrue(); + }); + + it('preserves errors without a response when requesting a stream', async () => { + const error = new Error('Connection failed'); + getSpy.mockRejectedValueOnce(error); + await expect( + axiosGet('https://example.org/package', { responseType: 'stream' }) + ).rejects.toBe(error); + }); + + it('preserves non-stream error responses', async () => { + const response = { status: 401, data: { message: 'Unauthorized' } }; + getSpy.mockRejectedValueOnce({ response }); + await expect(axiosGet('https://example.org/package')).resolves.toBe(response); + }); + it('returns get with responseType assigned when it is used as argument', async () => { getSpy.mockImplementationOnce(() => Promise.resolve({})); await axiosGet('packages.fhir.org/de.basisprofil.r4/1.1.0', { responseType: 'arraybuffer' });