From 7852dcab991a051d960c541caea326e32974e579 Mon Sep 17 00:00:00 2001 From: David de Boer Date: Tue, 8 Sep 2026 14:12:57 +0200 Subject: [PATCH 1/2] fix(distribution-downloader): time out on a stalled transfer instead of a slow one MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit – Replace the fixed 5-minute whole-transfer budget with an idle timeout that is reset on every chunk received, so a large file that keeps flowing is never cut off while a stalled server still is – Add ‘timeout’ (idle milliseconds, default 300 000) and ‘signal’ to DownloadOptions – Cover the trickling, stalled and cancelled cases with a local HTTP server --- docs/reference/distribution-downloader.md | 10 ++- .../distribution-downloader/src/download.ts | 82 +++++++++++++---- .../test/idleTimeout.test.ts | 90 +++++++++++++++++++ .../distribution-downloader/vite.config.ts | 4 +- 4 files changed, 164 insertions(+), 22 deletions(-) create mode 100644 packages/distribution-downloader/test/idleTimeout.test.ts diff --git a/docs/reference/distribution-downloader.md b/docs/reference/distribution-downloader.md index fb3d65f9..a01ab916 100644 --- a/docs/reference/distribution-downloader.md +++ b/docs/reference/distribution-downloader.md @@ -29,10 +29,18 @@ const { path, headers } = await downloader.download(distribution); - **Target path.** `download()` takes an optional second parameter, `target` – the file path to save to, defaulting to `/`. The resolved target must stay inside the base directory; a target that escapes it (e.g. via `../`) throws `Download target escapes the base directory`. - **Up-to-date skip.** When the local file is already up to date – judged by the distribution’s `lastModified` (and, when available, `byteSize`) against the file’s mtime and size – no HTTP request is made at all, and the result carries an **empty** `Headers` object. Only an actual download returns the response headers. -- **Timeout.** The fetch is bounded by a fixed 300 000 ms (5 minute) timeout. +- **Timeout.** The download is aborted when no bytes have arrived for `timeout` milliseconds (default 300 000 ms, 5 minutes), whether while waiting for the response headers or midway through the body. This is an idle timeout, not a budget for the whole transfer: a large file that keeps flowing is never cut off, however long it takes, while a stalled server is. Pass `signal` in `DownloadOptions` to cancel a download yourself. Either way the partial file is removed before the error propagates. - **Empty downloads are rejected.** A downloaded file of 1 byte or less throws `Distribution download is empty` – a body that small is a faulty distribution, not data. A download that fails midway is cleaned up before the error propagates. - **Logging.** Pass a `logger` via the third parameter, `DownloadOptions`; it defaults to a no-op logger, so the downloader is silent unless you provide one. +## `DownloadOptions` + +| Option | Type | Default | Description | +| --------- | ------------- | ------------ | ------------------------------------------------------------------------ | +| `logger` | `Logger` | no-op logger | Receives debug messages. | +| `timeout` | `number` | `300_000` | Idle timeout in milliseconds – abort when no bytes arrive for this long. | +| `signal` | `AbortSignal` | – | Cancels the download when aborted. | + ## The `Downloader` interface The package also exports the `Downloader` interface that `LastModifiedDownloader` implements: diff --git a/packages/distribution-downloader/src/download.ts b/packages/distribution-downloader/src/download.ts index b4263084..d2955543 100644 --- a/packages/distribution-downloader/src/download.ts +++ b/packages/distribution-downloader/src/download.ts @@ -1,6 +1,7 @@ import { Distribution } from '@lde/dataset'; import filenamifyUrl from 'filenamify-url'; import { dirname, join, resolve, sep } from 'node:path'; +import { Transform } from 'node:stream'; import { pipeline } from 'node:stream/promises'; import { createWriteStream } from 'node:fs'; import { access, mkdir, rm, stat } from 'node:fs/promises'; @@ -26,6 +27,17 @@ const noopLogger: Logger = { export interface DownloadOptions { logger?: Logger; + /** + * Idle timeout in milliseconds: the download is aborted when no bytes have + * arrived for this long, whether while waiting for the response headers or + * midway through the body. A large file that keeps flowing is never cut off, + * however long it takes. Defaults to 300 000 ms (5 minutes). + */ + timeout?: number; + /** + * Cancels the download when aborted; the partial file is removed. + */ + signal?: AbortSignal; } export interface DownloadResult { @@ -64,30 +76,62 @@ export class LastModifiedDownloader implements Downloader { return { path: filePath, headers: new Headers() }; } - const downloadResponse = await fetch(downloadUrl, { - signal: AbortSignal.timeout(300_000), - }); - if (!downloadResponse.ok || !downloadResponse.body) { - throw new Error( - `Failed to download ${downloadUrl}: ${downloadResponse.statusText}`, + const idleTimeout = options?.timeout ?? 300_000; + const idleAbortController = new AbortController(); + let idleTimer: NodeJS.Timeout | undefined; + const restartIdleTimer = () => { + clearTimeout(idleTimer); + idleTimer = setTimeout( + () => + idleAbortController.abort( + new Error(`No data received for ${idleTimeout} ms`), + ), + idleTimeout, ); - } + }; + const signal = + options?.signal === undefined + ? idleAbortController.signal + : AbortSignal.any([options.signal, idleAbortController.signal]); + restartIdleTimer(); try { - await mkdir(dirname(filePath), { recursive: true }); - await pipeline(downloadResponse.body, createWriteStream(filePath)); - } catch (error) { - await rm(filePath, { force: true }); - throw new Error(`Failed to save ${downloadUrl} to ${filePath}: ${error}`); - } + const downloadResponse = await fetch(downloadUrl, { signal }); + if (!downloadResponse.ok || !downloadResponse.body) { + throw new Error( + `Failed to download ${downloadUrl}: ${downloadResponse.statusText}`, + ); + } - const stats = await stat(filePath); - if (stats.size <= 1) { - logger.debug(`Distribution download ${downloadUrl} is empty`); - throw new Error('Distribution download is empty'); - } + try { + await mkdir(dirname(filePath), { recursive: true }); + await pipeline( + downloadResponse.body, + new Transform({ + transform(chunk, _encoding, callback) { + restartIdleTimer(); + callback(null, chunk); + }, + }), + createWriteStream(filePath), + ); + } catch (error) { + await rm(filePath, { force: true }); + throw new Error( + `Failed to save ${downloadUrl} to ${filePath}: ${error}`, + ); + } + + const stats = await stat(filePath); + if (stats.size <= 1) { + logger.debug(`Distribution download ${downloadUrl} is empty`); + throw new Error('Distribution download is empty'); + } - return { path: filePath, headers: downloadResponse.headers }; + return { path: filePath, headers: downloadResponse.headers }; + } finally { + clearTimeout(idleTimer); + } } private async localFileIsUpToDate( diff --git a/packages/distribution-downloader/test/idleTimeout.test.ts b/packages/distribution-downloader/test/idleTimeout.test.ts new file mode 100644 index 00000000..ab89a77e --- /dev/null +++ b/packages/distribution-downloader/test/idleTimeout.test.ts @@ -0,0 +1,90 @@ +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import { Distribution } from '@lde/dataset'; +import { LastModifiedDownloader } from '../src/download.js'; +import { createServer, Server, ServerResponse } from 'node:http'; +import { AddressInfo } from 'node:net'; +import { join } from 'node:path'; +import os from 'node:os'; +import fs from 'node:fs/promises'; + +// Runs against a real local HTTP server rather than nock, whose fetch +// interceptor buffers the whole response and so hides the chunk timing. +const downloader = new LastModifiedDownloader(os.tmpdir()); + +describe('LastModifiedDownloader idle timeout', () => { + let server: Server; + let pendingResponses: ServerResponse[]; + let serverFile: string; + const serverDistribution = (path: string) => + new Distribution( + new URL( + path, + `http://localhost:${(server.address() as AddressInfo).port}`, + ), + 'application/n-triples', + ); + + beforeEach(async () => { + pendingResponses = []; + server = createServer((request, response) => { + pendingResponses.push(response); + response.writeHead(200); + if (request.url === '/trickle') { + // Ten chunks 20 ms apart: 200 ms in total, never idle for long. + let chunksSent = 0; + const interval = setInterval(() => { + response.write('chunk '); + if (++chunksSent === 10) { + clearInterval(interval); + response.end(); + } + }, 20); + } else { + // One chunk, then silence. + response.write('partial'); + } + }); + await new Promise((resolve) => server.listen(0, resolve)); + const { port } = server.address() as AddressInfo; + serverFile = join(os.tmpdir(), `lde-idle-timeout-${port}`); + }); + + afterEach(async () => { + for (const response of pendingResponses) { + response.destroy(); + } + await new Promise((resolve) => server.close(() => resolve())); + await fs.rm(serverFile, { force: true }); + }); + + it('completes a slow download whose total time exceeds the timeout', async () => { + await downloader.download(serverDistribution('/trickle'), serverFile, { + timeout: 100, + }); + + expect(await fs.readFile(serverFile, 'utf8')).toBe('chunk '.repeat(10)); + }); + + it('aborts a stalled download and removes the partial file', async () => { + await expect( + downloader.download(serverDistribution('/stall'), serverFile, { + timeout: 100, + }), + ).rejects.toThrow('No data received for 100 ms'); + + await expect(fs.access(serverFile)).rejects.toThrow(); + }); + + it('cancels the download when the caller aborts the signal', async () => { + const abortController = new AbortController(); + setTimeout(() => abortController.abort(new Error('Cancelled')), 20); + + await expect( + downloader.download(serverDistribution('/stall'), serverFile, { + signal: abortController.signal, + }), + ).rejects.toThrow('Cancelled'); + + await expect(fs.access(serverFile)).rejects.toThrow(); + }); +}); diff --git a/packages/distribution-downloader/vite.config.ts b/packages/distribution-downloader/vite.config.ts index 66268575..4da82094 100644 --- a/packages/distribution-downloader/vite.config.ts +++ b/packages/distribution-downloader/vite.config.ts @@ -11,10 +11,10 @@ export default mergeConfig( coverage: { thresholds: { autoUpdate: true, - lines: 91.17, + lines: 97.82, functions: 100, branches: 100, - statements: 91.17, + statements: 97.82, }, }, }, From f30822971321607d6f46c8c90a83a07f803549ba Mon Sep 17 00:00:00 2001 From: David de Boer Date: Tue, 8 Sep 2026 14:18:40 +0200 Subject: [PATCH 2/2] =?UTF-8?q?test(distribution-downloader):=20widen=20th?= =?UTF-8?q?e=20trickle=20test=E2=80=99s=20timing=20margins?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit – A 20 ms chunk interval against a 100 ms idle timeout slipped on a loaded CI runner; use 50 ms chunks against a 500 ms timeout instead --- .../distribution-downloader/test/idleTimeout.test.ts | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/packages/distribution-downloader/test/idleTimeout.test.ts b/packages/distribution-downloader/test/idleTimeout.test.ts index ab89a77e..a70cdeaa 100644 --- a/packages/distribution-downloader/test/idleTimeout.test.ts +++ b/packages/distribution-downloader/test/idleTimeout.test.ts @@ -30,15 +30,16 @@ describe('LastModifiedDownloader idle timeout', () => { pendingResponses.push(response); response.writeHead(200); if (request.url === '/trickle') { - // Ten chunks 20 ms apart: 200 ms in total, never idle for long. + // Twenty chunks 50 ms apart: 1 s in total, never idle for long. The + // margins are wide because CI runners run many test suites at once. let chunksSent = 0; const interval = setInterval(() => { response.write('chunk '); - if (++chunksSent === 10) { + if (++chunksSent === 20) { clearInterval(interval); response.end(); } - }, 20); + }, 50); } else { // One chunk, then silence. response.write('partial'); @@ -59,10 +60,10 @@ describe('LastModifiedDownloader idle timeout', () => { it('completes a slow download whose total time exceeds the timeout', async () => { await downloader.download(serverDistribution('/trickle'), serverFile, { - timeout: 100, + timeout: 500, }); - expect(await fs.readFile(serverFile, 'utf8')).toBe('chunk '.repeat(10)); + expect(await fs.readFile(serverFile, 'utf8')).toBe('chunk '.repeat(20)); }); it('aborts a stalled download and removes the partial file', async () => {