From 02c9b8706e407d86c9dddbd0a4fa7fe38fd6d0de Mon Sep 17 00:00:00 2001 From: sneak Date: Wed, 23 Sep 2026 00:48:13 +0000 Subject: [PATCH] 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. streamDecrypt cancels the response body when decryption or the write fails, so a failed file no longer holds its connection. Model: opus-5-5 --- README.md | 25 ++++--- TODO.md | 6 ++ src/api/client.ts | 86 ++++++++++++++++-------- src/download/index.ts | 84 +++++++++++++---------- test/api/client.test.ts | 118 ++++++++++++++++++++++++++------- test/download/download.test.ts | 35 ++++++++++ 6 files changed, 255 insertions(+), 99 deletions(-) diff --git a/README.md b/README.md index 2388dd1..2fd00d2 100644 --- a/README.md +++ b/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 diff --git a/TODO.md b/TODO.md index 70fbc6e..6d7bef0 100644 --- a/TODO.md +++ b/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. `streamDecrypt` cancels the response body when decryption or the + write fails, so a failed file no longer holds its connection. - 2026-09-23: Made the CLI testable and tested it (issue 12). The command bodies moved from `bin/quak.ts` into `src/cli-commands.ts` as functions that take their options and a context (output streams, session directory, cache diff --git a/src/api/client.ts b/src/api/client.ts index b7bc281..3d3697e 100644 --- a/src/api/client.ts +++ b/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 | 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, - signal: AbortSignal, + deadline: ReturnType, ): ReadableStream => { + const { signal } = deadline; const reader = body.getReader(); let rejectOnAbort: (reason: unknown) => void = () => undefined; const aborted = new Promise((_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({ async pull(controller) { @@ -84,6 +110,7 @@ const deadlineStream = ( controller.close(); return; } + deadline.restart(); controller.enqueue(next.value); } catch (err) { release(); @@ -350,22 +377,27 @@ export class ApiClient { opts?: StreamOptions, ): Promise> { const once = async (): Promise> => { - // 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); } diff --git a/src/download/index.ts b/src/download/index.ts index d3a82f5..3691602 100644 --- a/src/download/index.ts +++ b/src/download/index.ts @@ -98,47 +98,59 @@ 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; } + } 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. + await reader.cancel(err).catch(() => undefined); + throw err; + } finally { + reader.releaseLock(); } // Only the last chunk of a secretstream carries TAG_FINAL. Everything a diff --git a/test/api/client.test.ts b/test/api/client.test.ts index 9ed6094..dea1306 100644 --- a/test/api/client.test.ts +++ b/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 @@ -661,13 +665,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 () => { @@ -781,25 +783,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({ - pull: () => new Promise(() => {}), - }), - { 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({ + pull: () => new Promise(() => {}), + }), + { 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({ + 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 diff --git a/test/download/download.test.ts b/test/download/download.test.ts index ae25b0e..cba7e4b 100644 --- a/test/download/download.test.ts +++ b/test/download/download.test.ts @@ -1337,6 +1337,41 @@ 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({ + 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); + }); }); // ---------------------------------------------------------------------------