fix(distribution-downloader): time out on a stalled transfer instead of a slow one

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 `<baseDir>/<filename derived from the access URL>`. 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:

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(

packages/distribution-downloader/test/idleTimeout.test.ts

@@ -0,0 +1,91 @@
+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') {
+        // 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 === 20) {
+            clearInterval(interval);
+            response.end();
+          }
+        }, 50);
+      } else {
+        // One chunk, then silence.
+        response.write('partial');
+      }
+    });
+    await new Promise<void>((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<void>((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: 500,
+    });
+
+    expect(await fs.readFile(serverFile, 'utf8')).toBe('chunk '.repeat(20));
+  });
+
+  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();
+  });
+});

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,
         },
       },
     },