Make the download deadline an idle deadline and cancel failed bodies (closes #24)

downloadTimeoutMs now aborts a file or thumbnail download only when no
bytes have arrived for that long (default 60 s, was a 600 s cap on the
whole transfer), so a slow download that keeps making progress completes.
The abort reason is still a TimeoutError, so retry classification is
unchanged. A download that fails before reading the whole response body
cancels it, including when the temp file cannot be opened or the header
is malformed, so a failed file no longer holds its connection.

Model: opus-5-5

README.md

@@ -363,19 +363,22 @@ and a half seconds of waiting. `sleep` and `random` are injectable through the
 same option, which is how the test suite exercises the whole policy without
 waiting.
 
-Two deadlines, applied with `AbortSignal.timeout()` and renewed for each
-attempt:
+Two deadlines, renewed for each attempt:
 
-| Option              | Default  | Applies to                                  |
-| ------------------- | -------- | ------------------------------------------- |
-| `requestTimeoutMs`  | `30000`  | `getJSON`, `postJSON`, `putJSON`, `putFile` |
-| `downloadTimeoutMs` | `600000` | file and thumbnail body transfers           |
+| Option              | Default | Applies to                                  | Kind                                  |
+| ------------------- | ------- | ------------------------------------------- | ------------------------------------- |
+| `requestTimeoutMs`  | `30000` | `getJSON`, `postJSON`, `putJSON`, `putFile` | the whole request                     |
+| `downloadTimeoutMs` | `60000` | file and thumbnail downloads                | idle: no bytes received for this long |
 
-They are separate because one number cannot serve both: a value short enough to
-keep a hung API call from stalling a backup would cancel a legitimate
-multi-gigabyte download. The download deadline covers the body, not just the
-headers — `getFileStream` returns as soon as headers arrive, so a deadline that
-only guarded the initial request would leave the same hang one layer down.
+They are different kinds because a download's length depends on the file and the
+link: a whole-transfer deadline short enough to catch a hung connection would
+cancel a large video on a slow link that is still making progress. The download
+deadline restarts every time bytes arrive, so a slow download runs as long as it
+keeps moving, and one that stalls is aborted after 60 seconds of silence. It
+covers the wait for the headers and the body — `getFileStream` returns as soon
+as headers arrive, so a deadline that only guarded the initial request would
+leave the same hang one layer down. There is no limit on the total length of a
+download.
 
 **Non-idempotent requests are not blindly replayed.** `postJSON` and `putJSON`
 send every `POST` and `PUT` in the endpoint list above; some of them change

TODO.md

@@ -18,6 +18,12 @@ Tag v1.0.0.
 
 # Completed Steps
 
+- 2026-09-23: Made the download deadline an idle deadline (issue 24).
+  `downloadTimeoutMs` now aborts a file or thumbnail download only after no
+  bytes have arrived for that long, default 60 seconds, instead of bounding the
+  whole transfer at 10 minutes, so a slow download that keeps making progress
+  completes. A download that fails before reading the whole body cancels it, so
+  a failed file no longer holds its connection.
 - 2026-09-23: Every `ApiClient` request URL is now built by one function next to
   the class (issue 18), so a self-hosted `apiOrigin` with a base path keeps it
   on every request, a path works with or without a leading slash, and query

src/api/client.ts

@@ -19,14 +19,12 @@ const DEFAULT_FILES_ORIGIN = "https://files.ente.io";
 const DEFAULT_THUMBS_ORIGIN = "https://thumbnails.ente.io";
 const CLIENT_PACKAGE = "berlin.sneak.quak";
 
-// Two deadlines rather than one, because a single number cannot serve both
-// jobs. Thirty seconds is generous for a JSON call and short enough that a
-// hung API connection cannot stall a backup for long. A file body is a
-// different shape of problem: the deadline has to cover the whole transfer,
-// which for a large video on a slow link is minutes, so a value sane for JSON
-// would cancel legitimate downloads.
+// Two deadlines of different kinds. `requestTimeoutMs` bounds a whole JSON
+// call. `downloadTimeoutMs` is an idle deadline: a file or thumbnail download
+// is aborted only when no bytes have arrived for that long, so a large video on
+// a slow link that keeps making progress is never cut off.
 export const DEFAULT_REQUEST_TIMEOUT_MS = 30_000;
-export const DEFAULT_DOWNLOAD_TIMEOUT_MS = 600_000;
+export const DEFAULT_DOWNLOAD_TIMEOUT_MS = 60_000;
 
 export interface ApiClientOptions {
     apiOrigin?: string;
@@ -48,18 +46,43 @@ export interface StreamOptions {
     retry?: boolean;
 }
 
-// Enforce a deadline over a response body, not merely over its headers.
+// An abort signal that fires once `ms` pass without a call to `restart`. It
+// aborts with a `TimeoutError`, the same reason `AbortSignal.timeout()` gives,
+// so the retry classifier treats an idle download exactly as it treats any
+// other deadline. `stop` must be called when the download ends, or the timer
+// keeps the process alive until it fires.
+const idleDeadline = (ms: number) => {
+    const controller = new AbortController();
+    let timer: ReturnType<typeof setTimeout> | undefined;
+    const stop = (): void => clearTimeout(timer);
+    const restart = (): void => {
+        stop();
+        timer = setTimeout(() => {
+            controller.abort(
+                new DOMException(
+                    `download stalled: no bytes received for ${ms} ms`,
+                    "TimeoutError",
+                ),
+            );
+        }, ms);
+    };
+    restart();
+    return { signal: controller.signal, restart, stop };
+};
+
+// Enforce the idle deadline over a response body, not merely over its headers.
 //
 // `getFileStream` returns as soon as headers arrive; the bytes are pulled
 // later, in the download layer. Whether the signal passed to `fetch` also
 // tears down the body afterwards is up to the fetch implementation, so this
 // wrapper makes it a property of quak instead: every read races the signal,
-// and an abort errors the stream with the abort reason — which the retry
-// classifier recognises.
+// each chunk that arrives restarts the deadline, and an abort errors the
+// stream with the abort reason — which the retry classifier recognises.
 const deadlineStream = (
     body: ReadableStream<Uint8Array>,
-    signal: AbortSignal,
+    deadline: ReturnType<typeof idleDeadline>,
 ): ReadableStream<Uint8Array> => {
+    const { signal } = deadline;
     const reader = body.getReader();
     let rejectOnAbort: (reason: unknown) => void = () => undefined;
     const aborted = new Promise<never>((_resolve, reject) => {
@@ -73,7 +96,10 @@ const deadlineStream = (
     const onAbort = (): void => rejectOnAbort(signal.reason);
     if (signal.aborted) onAbort();
     else signal.addEventListener("abort", onAbort, { once: true });
-    const release = (): void => signal.removeEventListener("abort", onAbort);
+    const release = (): void => {
+        deadline.stop();
+        signal.removeEventListener("abort", onAbort);
+    };
 
     return new ReadableStream<Uint8Array>({
         async pull(controller) {
@@ -84,6 +110,7 @@ const deadlineStream = (
                     controller.close();
                     return;
                 }
+                deadline.restart();
                 controller.enqueue(next.value);
             } catch (err) {
                 release();
@@ -362,22 +389,27 @@ export class ApiClient {
         opts?: StreamOptions,
     ): Promise<ReadableStream<Uint8Array>> {
         const once = async (): Promise<ReadableStream<Uint8Array>> => {
-            // A fresh deadline per attempt, so a retry gets the whole budget
-            // rather than the remainder of the one that just expired.
-            const signal = AbortSignal.timeout(this.downloadTimeoutMs);
-            const resp = await this._fetch(url, {
-                method: "GET",
-                headers: this.headers(),
-                signal,
-            });
-            await this.throwIfError(resp);
-            if (!resp.body) {
-                // Carries the status, and is not retryable: a response that
-                // arrived without a body is malformed, and asking again
-                // produces the same malformed response.
-                throw new ApiError("response body is null", resp.status);
+            // A fresh deadline per attempt. It also covers the wait for the
+            // headers, when no bytes have arrived either.
+            const deadline = idleDeadline(this.downloadTimeoutMs);
+            try {
+                const resp = await this._fetch(url, {
+                    method: "GET",
+                    headers: this.headers(),
+                    signal: deadline.signal,
+                });
+                await this.throwIfError(resp);
+                if (!resp.body) {
+                    // Carries the status, and is not retryable: a response
+                    // that arrived without a body is malformed, and asking
+                    // again produces the same malformed response.
+                    throw new ApiError("response body is null", resp.status);
+                }
+                return deadlineStream(resp.body, deadline);
+            } catch (err) {
+                deadline.stop();
+                throw err;
             }
-            return deadlineStream(resp.body, signal);
         };
         return opts?.retry === false ? once() : withRetry(once, this.retry);
     }

src/download/index.ts

@@ -98,47 +98,53 @@ const streamDecrypt = async (
         onProgress?.(totalPlain);
     };
 
-    for (;;) {
-        const { done, value } = await reader.read();
-        if (value && value.length > 0) {
-            pending.push(value);
-            pendingBytes += value.length;
-        }
+    try {
+        for (;;) {
+            const { done, value } = await reader.read();
+            if (value && value.length > 0) {
+                pending.push(value);
+                pendingBytes += value.length;
+            }
 
-        while (pendingBytes >= ENC_CHUNK_SIZE) {
-            const encChunk = takeContiguous(ENC_CHUNK_SIZE);
-            // A whole chunk that fails to authenticate while the stream carries
-            // on is corruption, not truncation; that error propagates unchanged.
-            const { plaintext, tag } = pullStreamChunk(state, encChunk);
-            await consume(plaintext, tag);
-        }
+            while (pendingBytes >= ENC_CHUNK_SIZE) {
+                const encChunk = takeContiguous(ENC_CHUNK_SIZE);
+                // A whole chunk that fails to authenticate while the stream
+                // carries on is corruption, not truncation; that error
+                // propagates unchanged.
+                const { plaintext, tag } = pullStreamChunk(state, encChunk);
+                await consume(plaintext, tag);
+            }
 
-        if (done) {
-            if (pendingBytes > 0) {
-                const buffer = takeContiguous(pendingBytes);
-                // Whatever is left over once every whole chunk has been
-                // consumed must be the stream's final chunk, and a final
-                // chunk that actually arrived in full authenticates. If it
-                // does not, the body stopped part-way through a chunk — the
-                // ordinary shape of a dropped connection. Poly1305 cannot
-                // tell a partial chunk from a corrupt one, so this is
-                // reported as the truncation it almost always is, with the
-                // authentication failure kept as the error's cause. Only the
-                // pull is guarded: a sink failure on a chunk that did
-                // authenticate is a disk error, not a truncation.
-                let pulled;
-                try {
-                    pulled = pullStreamChunk(state, buffer);
-                } catch (err) {
-                    throw new TruncatedStreamError(
-                        `download: stream truncated: response body ended with ${buffer.length} trailing bytes that did not authenticate as a final chunk (transfer stopped mid-chunk, or the data is corrupt)`,
-                        { cause: err },
-                    );
+            if (done) {
+                if (pendingBytes > 0) {
+                    const buffer = takeContiguous(pendingBytes);
+                    // Whatever is left over once every whole chunk has been
+                    // consumed must be the stream's final chunk, and a final
+                    // chunk that actually arrived in full authenticates. If
+                    // it does not, the body stopped part-way through a chunk
+                    // — the ordinary shape of a dropped connection. Poly1305
+                    // cannot tell a partial chunk from a corrupt one, so this
+                    // is reported as the truncation it almost always is, with
+                    // the authentication failure kept as the error's cause.
+                    // Only the pull is guarded: a sink failure on a chunk
+                    // that did authenticate is a disk error, not a
+                    // truncation.
+                    let pulled;
+                    try {
+                        pulled = pullStreamChunk(state, buffer);
+                    } catch (err) {
+                        throw new TruncatedStreamError(
+                            `download: stream truncated: response body ended with ${buffer.length} trailing bytes that did not authenticate as a final chunk (transfer stopped mid-chunk, or the data is corrupt)`,
+                            { cause: err },
+                        );
+                    }
+                    await consume(pulled.plaintext, pulled.tag);
                 }
-                await consume(pulled.plaintext, pulled.tag);
+                break;
             }
-            break;
         }
+    } finally {
+        reader.releaseLock();
     }
 
     // Only the last chunk of a secretstream carries TAG_FINAL. Everything a
@@ -245,17 +251,27 @@ const decryptToTemp = async (
     onProgress?: ProgressCallback,
 ): Promise<number> => {
     let bytesWritten = 0;
-    await stageAtomic(destination, async (handle) => {
-        bytesWritten = await streamDecrypt(
-            stream,
-            header,
-            key,
-            async (plaintext) => {
-                await handle.write(plaintext);
-            },
-            onProgress,
-        );
-    });
+    try {
+        await stageAtomic(destination, async (handle) => {
+            bytesWritten = await streamDecrypt(
+                stream,
+                header,
+                key,
+                async (plaintext) => {
+                    await handle.write(plaintext);
+                },
+                onProgress,
+            );
+        });
+    } catch (err) {
+        // Cancel the body so its connection is closed now rather than held
+        // until the stream is garbage collected. A backup run carries on past
+        // a failed file, so without this every failure would hold a socket.
+        // This covers every failure, including a temp file that cannot be
+        // opened and a header that is rejected before the body is read.
+        await stream.cancel(err).catch(() => undefined);
+        throw err;
+    }
     return bytesWritten;
 };
 

test/api/client.test.ts

@@ -38,14 +38,18 @@
  * the network. The fake records every call for assertion.
  */
 
-import { describe, expect, it } from "vitest";
+import { describe, expect, it, vi } from "vitest";
 import {
     ApiClient,
     ApiError,
     DEFAULT_DOWNLOAD_TIMEOUT_MS,
     DEFAULT_REQUEST_TIMEOUT_MS,
 } from "../../src/api/client.js";
-import type { RetryOptions } from "../../src/retry.js";
+import {
+    isRetryable,
+    isSafeToReplay,
+    type RetryOptions,
+} from "../../src/retry.js";
 
 // ---------------------------------------------------------------------------
 // Test helpers
@@ -748,13 +752,11 @@ describe("ApiClient retries", () => {
 
 describe("ApiClient timeouts", () => {
     it("ships bounded default deadlines", () => {
-        // Asserted here so the README and the code cannot drift. Two numbers
-        // rather than one, because a deadline that is sane for a JSON call is
-        // nowhere near enough for a multi-gigabyte body, and a deadline long
-        // enough for that body would let a hung API call stall a backup for
-        // ten minutes.
+        // Asserted here so the README and the code cannot drift. The request
+        // deadline bounds a whole JSON call; the download deadline is an idle
+        // one, measured from the last byte that arrived.
         expect(DEFAULT_REQUEST_TIMEOUT_MS).toBe(30_000);
-        expect(DEFAULT_DOWNLOAD_TIMEOUT_MS).toBe(600_000);
+        expect(DEFAULT_DOWNLOAD_TIMEOUT_MS).toBe(60_000);
     });
 
     it("attaches an abort signal to every request", async () => {
@@ -868,25 +870,91 @@ describe("ApiClient timeouts", () => {
         // that never produces a chunk and never observes the signal, so the
         // only thing that can unblock the read is quak's own enforcement of
         // the deadline over the stream it hands out.
-        const stalling = new Response(
-            new ReadableStream<Uint8Array>({
-                pull: () => new Promise<void>(() => {}),
-            }),
-            { status: 200 },
-        );
-        const { fetch } = scriptedFetch(stalling);
-        const client = new ApiClient({
-            fetch,
-            downloadTimeoutMs: 20,
-            retry: { ...noWait, attempts: 1 },
-        });
+        //
+        // The clock is faked, so the test runs under the real default
+        // deadline and waits for nothing.
+        vi.useFakeTimers();
+        try {
+            const stalling = new Response(
+                new ReadableStream<Uint8Array>({
+                    pull: () => new Promise<void>(() => {}),
+                }),
+                { status: 200 },
+            );
+            const { fetch } = scriptedFetch(stalling);
+            const client = new ApiClient({
+                fetch,
+                retry: { ...noWait, attempts: 1 },
+            });
 
-        const stream = await client.getFileStream(42);
-        const err: unknown = await readAll(stream).catch((e: unknown) => e);
+            const stream = await client.getFileStream(42);
+            let settled = false;
+            const result = readAll(stream).then(
+                (n) => n,
+                (e: unknown) => e,
+            );
+            void result.finally(() => {
+                settled = true;
+            });
 
-        expect(err).toBeInstanceOf(Error);
-        expect((err as Error).name).toBe("TimeoutError");
-    }, 5000);
+            await vi.advanceTimersByTimeAsync(DEFAULT_DOWNLOAD_TIMEOUT_MS - 1);
+            expect(settled).toBe(false);
+            await vi.advanceTimersByTimeAsync(1);
+
+            const err = await result;
+            expect(err).toBeInstanceOf(Error);
+            expect((err as Error).name).toBe("TimeoutError");
+            // Classified as every deadline is: retried by the idempotent
+            // downloads, never replayed for a POST or PUT.
+            expect(isRetryable(err)).toBe(true);
+            expect(isSafeToReplay(err)).toBe(false);
+        } finally {
+            vi.useRealTimers();
+        }
+    });
+
+    it("does not abort a slow body that keeps making progress", async () => {
+        // A deadline over the whole transfer would cut off a large video on a
+        // slow link however steadily it was arriving. The deadline restarts
+        // with every chunk, so a body that sends one byte every 600 ms for
+        // well over the 1000 ms deadline completes.
+        vi.useFakeTimers();
+        try {
+            let sent = 0;
+            const trickling = new Response(
+                new ReadableStream<Uint8Array>({
+                    async pull(controller) {
+                        await new Promise((resolve) =>
+                            setTimeout(resolve, 600),
+                        );
+                        if (sent === 10) {
+                            controller.close();
+                            return;
+                        }
+                        controller.enqueue(new Uint8Array([sent++]));
+                    },
+                }),
+                { status: 200 },
+            );
+            const { fetch } = scriptedFetch(trickling);
+            const client = new ApiClient({
+                fetch,
+                downloadTimeoutMs: 1000,
+                retry: { ...noWait, attempts: 1 },
+            });
+
+            const stream = await client.getFileStream(42);
+            const result = readAll(stream).then(
+                (n) => n,
+                (e: unknown) => e,
+            );
+            await vi.advanceTimersByTimeAsync(11 * 600);
+
+            expect(await result).toBe(10);
+        } finally {
+            vi.useRealTimers();
+        }
+    });
 
     it("lets a body that arrives in time through untouched", async () => {
         // The counterpart to the previous test: enforcing the deadline over

test/download/download.test.ts

@@ -1337,6 +1337,102 @@ describe("download retries: corruption is not retried", () => {
 
         expect(requests()).toBe(1);
     });
+
+    it("cancels the response body when decryption fails", async () => {
+        // A backup run carries on past a failed file, so a body left open on
+        // failure would hold its connection until garbage collection, once
+        // per failed file. This body delivers a corrupt chunk and then stays
+        // open, so only a cancel from the downloader can close it.
+        const corrupted = Uint8Array.from(multiChunk.body);
+        corrupted[10] ^= 0xff;
+        let cancelled = false;
+        const fetch = (async () =>
+            new Response(
+                new ReadableStream<Uint8Array>({
+                    start(controller) {
+                        controller.enqueue(corrupted);
+                    },
+                    cancel() {
+                        cancelled = true;
+                    },
+                }),
+                { status: 200 },
+            )) as typeof globalThis.fetch;
+        const api = new ApiClient({ fetch, retry: { ...noWait, attempts: 1 } });
+        const file = buildMockEnteFile(
+            multiChunkKey,
+            multiChunk.header,
+            multiChunk.header,
+        );
+        const outPath = join(mkdtempSync(join(testDir, "cancel-")), "c.bin");
+
+        await expect(downloadFile(api, file, outPath)).rejects.toThrow(
+            /authentication failed/i,
+        );
+
+        expect(cancelled).toBe(true);
+    });
+
+    it("cancels the response body when the temp file cannot be opened", async () => {
+        // The download fails before a byte of the body is read, so the body
+        // is still open and only a cancel from the downloader can close it.
+        let cancelled = false;
+        const fetch = (async () =>
+            new Response(
+                new ReadableStream<Uint8Array>({
+                    start(controller) {
+                        controller.enqueue(multiChunk.body);
+                    },
+                    cancel() {
+                        cancelled = true;
+                    },
+                }),
+                { status: 200 },
+            )) as typeof globalThis.fetch;
+        const api = new ApiClient({ fetch, retry: { ...noWait, attempts: 1 } });
+        const file = buildMockEnteFile(
+            multiChunkKey,
+            multiChunk.header,
+            multiChunk.header,
+        );
+        const outPath = join(
+            mkdtempSync(join(testDir, "cancel-")),
+            "missing",
+            "c.bin",
+        );
+
+        await expect(downloadFile(api, file, outPath)).rejects.toMatchObject({
+            code: "ENOENT",
+        });
+
+        expect(cancelled).toBe(true);
+    });
+
+    it("cancels the response body when the header is malformed", async () => {
+        // The header is rejected before the body is read, so the body is
+        // still open and only a cancel from the downloader can close it.
+        let cancelled = false;
+        const fetch = (async () =>
+            new Response(
+                new ReadableStream<Uint8Array>({
+                    start(controller) {
+                        controller.enqueue(multiChunk.body);
+                    },
+                    cancel() {
+                        cancelled = true;
+                    },
+                }),
+                { status: 200 },
+            )) as typeof globalThis.fetch;
+        const api = new ApiClient({ fetch, retry: { ...noWait, attempts: 1 } });
+        const shortHeader = multiChunk.header.subarray(0, 5);
+        const file = buildMockEnteFile(multiChunkKey, shortHeader, shortHeader);
+        const outPath = join(mkdtempSync(join(testDir, "cancel-")), "c.bin");
+
+        await expect(downloadFile(api, file, outPath)).rejects.toThrow();
+
+        expect(cancelled).toBe(true);
+    });
 });
 
 // ---------------------------------------------------------------------------