Compare commits
2
Commits
02c9b8706e
...
b9bb0c5946
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b9bb0c5946 | ||
|
|
2b410c3ed6 |
@@ -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
|
same option, which is how the test suite exercises the whole policy without
|
||||||
waiting.
|
waiting.
|
||||||
|
|
||||||
Two deadlines, applied with `AbortSignal.timeout()` and renewed for each
|
Two deadlines, renewed for each attempt:
|
||||||
attempt:
|
|
||||||
|
|
||||||
| Option | Default | Applies to |
|
| Option | Default | Applies to | Kind |
|
||||||
| ------------------- | -------- | ------------------------------------------- |
|
| ------------------- | ------- | ------------------------------------------- | ------------------------------------- |
|
||||||
| `requestTimeoutMs` | `30000` | `getJSON`, `postJSON`, `putJSON`, `putFile` |
|
| `requestTimeoutMs` | `30000` | `getJSON`, `postJSON`, `putJSON`, `putFile` | the whole request |
|
||||||
| `downloadTimeoutMs` | `600000` | file and thumbnail body transfers |
|
| `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
|
They are different kinds because a download's length depends on the file and the
|
||||||
keep a hung API call from stalling a backup would cancel a legitimate
|
link: a whole-transfer deadline short enough to catch a hung connection would
|
||||||
multi-gigabyte download. The download deadline covers the body, not just the
|
cancel a large video on a slow link that is still making progress. The download
|
||||||
headers — `getFileStream` returns as soon as headers arrive, so a deadline that
|
deadline restarts every time bytes arrive, so a slow download runs as long as it
|
||||||
only guarded the initial request would leave the same hang one layer down.
|
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`
|
**Non-idempotent requests are not blindly replayed.** `postJSON` and `putJSON`
|
||||||
send every `POST` and `PUT` in the endpoint list above; some of them change
|
send every `POST` and `PUT` in the endpoint list above; some of them change
|
||||||
|
|||||||
@@ -18,6 +18,17 @@ Tag v1.0.0.
|
|||||||
|
|
||||||
# Completed Steps
|
# 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
|
||||||
|
parameters are percent-encoded. A path containing `?` or `#` is rejected with
|
||||||
|
an error instead of being silently cut.
|
||||||
- 2026-09-23: Made the CLI testable and tested it (issue 12). The command bodies
|
- 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
|
moved from `bin/quak.ts` into `src/cli-commands.ts` as functions that take
|
||||||
their options and a context (output streams, session directory, cache
|
their options and a context (output streams, session directory, cache
|
||||||
|
|||||||
+84
-40
@@ -19,14 +19,12 @@ const DEFAULT_FILES_ORIGIN = "https://files.ente.io";
|
|||||||
const DEFAULT_THUMBS_ORIGIN = "https://thumbnails.ente.io";
|
const DEFAULT_THUMBS_ORIGIN = "https://thumbnails.ente.io";
|
||||||
const CLIENT_PACKAGE = "berlin.sneak.quak";
|
const CLIENT_PACKAGE = "berlin.sneak.quak";
|
||||||
|
|
||||||
// Two deadlines rather than one, because a single number cannot serve both
|
// Two deadlines of different kinds. `requestTimeoutMs` bounds a whole JSON
|
||||||
// jobs. Thirty seconds is generous for a JSON call and short enough that a
|
// call. `downloadTimeoutMs` is an idle deadline: a file or thumbnail download
|
||||||
// hung API connection cannot stall a backup for long. A file body is a
|
// is aborted only when no bytes have arrived for that long, so a large video on
|
||||||
// different shape of problem: the deadline has to cover the whole transfer,
|
// a slow link that keeps making progress is never cut off.
|
||||||
// which for a large video on a slow link is minutes, so a value sane for JSON
|
|
||||||
// would cancel legitimate downloads.
|
|
||||||
export const DEFAULT_REQUEST_TIMEOUT_MS = 30_000;
|
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 {
|
export interface ApiClientOptions {
|
||||||
apiOrigin?: string;
|
apiOrigin?: string;
|
||||||
@@ -48,18 +46,43 @@ export interface StreamOptions {
|
|||||||
retry?: boolean;
|
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
|
// `getFileStream` returns as soon as headers arrive; the bytes are pulled
|
||||||
// later, in the download layer. Whether the signal passed to `fetch` also
|
// 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
|
// 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,
|
// 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
|
// each chunk that arrives restarts the deadline, and an abort errors the
|
||||||
// classifier recognises.
|
// stream with the abort reason — which the retry classifier recognises.
|
||||||
const deadlineStream = (
|
const deadlineStream = (
|
||||||
body: ReadableStream<Uint8Array>,
|
body: ReadableStream<Uint8Array>,
|
||||||
signal: AbortSignal,
|
deadline: ReturnType<typeof idleDeadline>,
|
||||||
): ReadableStream<Uint8Array> => {
|
): ReadableStream<Uint8Array> => {
|
||||||
|
const { signal } = deadline;
|
||||||
const reader = body.getReader();
|
const reader = body.getReader();
|
||||||
let rejectOnAbort: (reason: unknown) => void = () => undefined;
|
let rejectOnAbort: (reason: unknown) => void = () => undefined;
|
||||||
const aborted = new Promise<never>((_resolve, reject) => {
|
const aborted = new Promise<never>((_resolve, reject) => {
|
||||||
@@ -73,7 +96,10 @@ const deadlineStream = (
|
|||||||
const onAbort = (): void => rejectOnAbort(signal.reason);
|
const onAbort = (): void => rejectOnAbort(signal.reason);
|
||||||
if (signal.aborted) onAbort();
|
if (signal.aborted) onAbort();
|
||||||
else signal.addEventListener("abort", onAbort, { once: true });
|
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>({
|
return new ReadableStream<Uint8Array>({
|
||||||
async pull(controller) {
|
async pull(controller) {
|
||||||
@@ -84,6 +110,7 @@ const deadlineStream = (
|
|||||||
controller.close();
|
controller.close();
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
deadline.restart();
|
||||||
controller.enqueue(next.value);
|
controller.enqueue(next.value);
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
release();
|
release();
|
||||||
@@ -98,6 +125,30 @@ const deadlineStream = (
|
|||||||
});
|
});
|
||||||
};
|
};
|
||||||
|
|
||||||
|
// The one place a request URL is built. `origin` may carry a base path (a
|
||||||
|
// self-hosted server behind a prefix) and may end in a slash; `path` may or
|
||||||
|
// may not start with one. Query parameters go only through `query`, which
|
||||||
|
// percent-encodes them: a `?` or `#` in `path` is an error, because
|
||||||
|
// `new URL` would otherwise quietly treat what follows as something else.
|
||||||
|
const buildURL = (
|
||||||
|
origin: string,
|
||||||
|
path: string,
|
||||||
|
query?: Record<string, string | number | undefined>,
|
||||||
|
): string => {
|
||||||
|
if (path.includes("?") || path.includes("#")) {
|
||||||
|
throw new Error(
|
||||||
|
`request path must not contain "?" or "#"; pass query parameters separately: ${path}`,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
const url = new URL(
|
||||||
|
`${origin.replace(/\/+$/, "")}/${path.replace(/^\/+/, "")}`,
|
||||||
|
);
|
||||||
|
for (const [k, v] of Object.entries(query ?? {})) {
|
||||||
|
if (v !== undefined) url.searchParams.set(k, String(v));
|
||||||
|
}
|
||||||
|
return url.href;
|
||||||
|
};
|
||||||
|
|
||||||
export class ApiClient {
|
export class ApiClient {
|
||||||
private readonly apiOrigin: string;
|
private readonly apiOrigin: string;
|
||||||
private readonly isCustomOrigin: boolean;
|
private readonly isCustomOrigin: boolean;
|
||||||
@@ -202,22 +253,10 @@ export class ApiClient {
|
|||||||
path: string,
|
path: string,
|
||||||
query?: Record<string, string | number | undefined>,
|
query?: Record<string, string | number | undefined>,
|
||||||
): Promise<T> {
|
): Promise<T> {
|
||||||
const url = new URL(path, this.apiOrigin + "/");
|
const url = buildURL(this.apiOrigin, path, query);
|
||||||
// new URL with a base resolves relative paths; ensure we keep the
|
|
||||||
// origin from apiOrigin even when path starts with /
|
|
||||||
url.protocol = new URL(this.apiOrigin).protocol;
|
|
||||||
url.host = new URL(this.apiOrigin).host;
|
|
||||||
url.pathname = path;
|
|
||||||
if (query) {
|
|
||||||
for (const [k, v] of Object.entries(query)) {
|
|
||||||
if (v !== undefined) {
|
|
||||||
url.searchParams.set(k, String(v));
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
// A GET changes nothing, so it is retried under the full policy.
|
// A GET changes nothing, so it is retried under the full policy.
|
||||||
return withRetry(async () => {
|
return withRetry(async () => {
|
||||||
const resp = await this._fetch(url.href, {
|
const resp = await this._fetch(url, {
|
||||||
method: "GET",
|
method: "GET",
|
||||||
headers: this.headers(),
|
headers: this.headers(),
|
||||||
signal: AbortSignal.timeout(this.requestTimeoutMs),
|
signal: AbortSignal.timeout(this.requestTimeoutMs),
|
||||||
@@ -228,7 +267,7 @@ export class ApiClient {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async postJSON<T>(path: string, body: unknown): Promise<T> {
|
async postJSON<T>(path: string, body: unknown): Promise<T> {
|
||||||
const url = `${this.apiOrigin}${path}`;
|
const url = buildURL(this.apiOrigin, path);
|
||||||
// Not idempotent: a POST is replayed only when `isSafeToReplay`
|
// Not idempotent: a POST is replayed only when `isSafeToReplay`
|
||||||
// says no request byte can have reached the server. The endpoints
|
// says no request byte can have reached the server. The endpoints
|
||||||
// this covers are listed in the README under "Endpoints used".
|
// this covers are listed in the README under "Endpoints used".
|
||||||
@@ -261,8 +300,8 @@ export class ApiClient {
|
|||||||
opts?: StreamOptions,
|
opts?: StreamOptions,
|
||||||
): Promise<ReadableStream<Uint8Array>> {
|
): Promise<ReadableStream<Uint8Array>> {
|
||||||
const url = this.isCustomOrigin
|
const url = this.isCustomOrigin
|
||||||
? `${this.apiOrigin}/files/download/${fileID}`
|
? buildURL(this.apiOrigin, `/files/download/${fileID}`)
|
||||||
: `${this.filesOrigin}/?fileID=${fileID}`;
|
: buildURL(this.filesOrigin, "/", { fileID });
|
||||||
return this.streamRequest(url, opts);
|
return this.streamRequest(url, opts);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -304,7 +343,7 @@ export class ApiClient {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async putJSON<T>(path: string, body: unknown): Promise<T> {
|
async putJSON<T>(path: string, body: unknown): Promise<T> {
|
||||||
const url = `${this.apiOrigin}${path}`;
|
const url = buildURL(this.apiOrigin, path);
|
||||||
// Same replay and redirect rules as `postJSON`, for the same reasons.
|
// Same replay and redirect rules as `postJSON`, for the same reasons.
|
||||||
return withRetry(
|
return withRetry(
|
||||||
async () => {
|
async () => {
|
||||||
@@ -340,8 +379,8 @@ export class ApiClient {
|
|||||||
opts?: StreamOptions,
|
opts?: StreamOptions,
|
||||||
): Promise<ReadableStream<Uint8Array>> {
|
): Promise<ReadableStream<Uint8Array>> {
|
||||||
const url = this.isCustomOrigin
|
const url = this.isCustomOrigin
|
||||||
? `${this.apiOrigin}/files/preview/${fileID}`
|
? buildURL(this.apiOrigin, `/files/preview/${fileID}`)
|
||||||
: `${this.thumbsOrigin}/?fileID=${fileID}`;
|
: buildURL(this.thumbsOrigin, "/", { fileID });
|
||||||
return this.streamRequest(url, opts);
|
return this.streamRequest(url, opts);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -350,22 +389,27 @@ export class ApiClient {
|
|||||||
opts?: StreamOptions,
|
opts?: StreamOptions,
|
||||||
): Promise<ReadableStream<Uint8Array>> {
|
): Promise<ReadableStream<Uint8Array>> {
|
||||||
const once = async (): Promise<ReadableStream<Uint8Array>> => {
|
const once = async (): Promise<ReadableStream<Uint8Array>> => {
|
||||||
// A fresh deadline per attempt, so a retry gets the whole budget
|
// A fresh deadline per attempt. It also covers the wait for the
|
||||||
// rather than the remainder of the one that just expired.
|
// headers, when no bytes have arrived either.
|
||||||
const signal = AbortSignal.timeout(this.downloadTimeoutMs);
|
const deadline = idleDeadline(this.downloadTimeoutMs);
|
||||||
|
try {
|
||||||
const resp = await this._fetch(url, {
|
const resp = await this._fetch(url, {
|
||||||
method: "GET",
|
method: "GET",
|
||||||
headers: this.headers(),
|
headers: this.headers(),
|
||||||
signal,
|
signal: deadline.signal,
|
||||||
});
|
});
|
||||||
await this.throwIfError(resp);
|
await this.throwIfError(resp);
|
||||||
if (!resp.body) {
|
if (!resp.body) {
|
||||||
// Carries the status, and is not retryable: a response that
|
// Carries the status, and is not retryable: a response
|
||||||
// arrived without a body is malformed, and asking again
|
// that arrived without a body is malformed, and asking
|
||||||
// produces the same malformed response.
|
// again produces the same malformed response.
|
||||||
throw new ApiError("response body is null", resp.status);
|
throw new ApiError("response body is null", resp.status);
|
||||||
}
|
}
|
||||||
return deadlineStream(resp.body, signal);
|
return deadlineStream(resp.body, deadline);
|
||||||
|
} catch (err) {
|
||||||
|
deadline.stop();
|
||||||
|
throw err;
|
||||||
|
}
|
||||||
};
|
};
|
||||||
return opts?.retry === false ? once() : withRetry(once, this.retry);
|
return opts?.retry === false ? once() : withRetry(once, this.retry);
|
||||||
}
|
}
|
||||||
|
|||||||
+26
-10
@@ -98,6 +98,7 @@ const streamDecrypt = async (
|
|||||||
onProgress?.(totalPlain);
|
onProgress?.(totalPlain);
|
||||||
};
|
};
|
||||||
|
|
||||||
|
try {
|
||||||
for (;;) {
|
for (;;) {
|
||||||
const { done, value } = await reader.read();
|
const { done, value } = await reader.read();
|
||||||
if (value && value.length > 0) {
|
if (value && value.length > 0) {
|
||||||
@@ -107,8 +108,9 @@ const streamDecrypt = async (
|
|||||||
|
|
||||||
while (pendingBytes >= ENC_CHUNK_SIZE) {
|
while (pendingBytes >= ENC_CHUNK_SIZE) {
|
||||||
const encChunk = takeContiguous(ENC_CHUNK_SIZE);
|
const encChunk = takeContiguous(ENC_CHUNK_SIZE);
|
||||||
// A whole chunk that fails to authenticate while the stream carries
|
// A whole chunk that fails to authenticate while the stream
|
||||||
// on is corruption, not truncation; that error propagates unchanged.
|
// carries on is corruption, not truncation; that error
|
||||||
|
// propagates unchanged.
|
||||||
const { plaintext, tag } = pullStreamChunk(state, encChunk);
|
const { plaintext, tag } = pullStreamChunk(state, encChunk);
|
||||||
await consume(plaintext, tag);
|
await consume(plaintext, tag);
|
||||||
}
|
}
|
||||||
@@ -118,14 +120,15 @@ const streamDecrypt = async (
|
|||||||
const buffer = takeContiguous(pendingBytes);
|
const buffer = takeContiguous(pendingBytes);
|
||||||
// Whatever is left over once every whole chunk has been
|
// Whatever is left over once every whole chunk has been
|
||||||
// consumed must be the stream's final chunk, and a final
|
// consumed must be the stream's final chunk, and a final
|
||||||
// chunk that actually arrived in full authenticates. If it
|
// chunk that actually arrived in full authenticates. If
|
||||||
// does not, the body stopped part-way through a chunk — the
|
// it does not, the body stopped part-way through a chunk
|
||||||
// ordinary shape of a dropped connection. Poly1305 cannot
|
// — the ordinary shape of a dropped connection. Poly1305
|
||||||
// tell a partial chunk from a corrupt one, so this is
|
// cannot tell a partial chunk from a corrupt one, so this
|
||||||
// reported as the truncation it almost always is, with the
|
// is reported as the truncation it almost always is, with
|
||||||
// authentication failure kept as the error's cause. Only the
|
// the authentication failure kept as the error's cause.
|
||||||
// pull is guarded: a sink failure on a chunk that did
|
// Only the pull is guarded: a sink failure on a chunk
|
||||||
// authenticate is a disk error, not a truncation.
|
// that did authenticate is a disk error, not a
|
||||||
|
// truncation.
|
||||||
let pulled;
|
let pulled;
|
||||||
try {
|
try {
|
||||||
pulled = pullStreamChunk(state, buffer);
|
pulled = pullStreamChunk(state, buffer);
|
||||||
@@ -140,6 +143,9 @@ const streamDecrypt = async (
|
|||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
} finally {
|
||||||
|
reader.releaseLock();
|
||||||
|
}
|
||||||
|
|
||||||
// Only the last chunk of a secretstream carries TAG_FINAL. Everything a
|
// Only the last chunk of a secretstream carries TAG_FINAL. Everything a
|
||||||
// dropped connection did deliver still decrypts and authenticates, so the
|
// dropped connection did deliver still decrypts and authenticates, so the
|
||||||
@@ -245,6 +251,7 @@ const decryptToTemp = async (
|
|||||||
onProgress?: ProgressCallback,
|
onProgress?: ProgressCallback,
|
||||||
): Promise<number> => {
|
): Promise<number> => {
|
||||||
let bytesWritten = 0;
|
let bytesWritten = 0;
|
||||||
|
try {
|
||||||
await stageAtomic(destination, async (handle) => {
|
await stageAtomic(destination, async (handle) => {
|
||||||
bytesWritten = await streamDecrypt(
|
bytesWritten = await streamDecrypt(
|
||||||
stream,
|
stream,
|
||||||
@@ -256,6 +263,15 @@ const decryptToTemp = async (
|
|||||||
onProgress,
|
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;
|
return bytesWritten;
|
||||||
};
|
};
|
||||||
|
|
||||||
|
|||||||
+166
-11
@@ -38,14 +38,18 @@
|
|||||||
* the network. The fake records every call for assertion.
|
* the network. The fake records every call for assertion.
|
||||||
*/
|
*/
|
||||||
|
|
||||||
import { describe, expect, it } from "vitest";
|
import { describe, expect, it, vi } from "vitest";
|
||||||
import {
|
import {
|
||||||
ApiClient,
|
ApiClient,
|
||||||
ApiError,
|
ApiError,
|
||||||
DEFAULT_DOWNLOAD_TIMEOUT_MS,
|
DEFAULT_DOWNLOAD_TIMEOUT_MS,
|
||||||
DEFAULT_REQUEST_TIMEOUT_MS,
|
DEFAULT_REQUEST_TIMEOUT_MS,
|
||||||
} from "../../src/api/client.js";
|
} from "../../src/api/client.js";
|
||||||
import type { RetryOptions } from "../../src/retry.js";
|
import {
|
||||||
|
isRetryable,
|
||||||
|
isSafeToReplay,
|
||||||
|
type RetryOptions,
|
||||||
|
} from "../../src/retry.js";
|
||||||
|
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
// Test helpers
|
// Test helpers
|
||||||
@@ -385,6 +389,93 @@ describe("ApiClient custom origins", () => {
|
|||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
describe("ApiClient request URLs", () => {
|
||||||
|
it("accepts a path with or without a leading slash", async () => {
|
||||||
|
const { fetch, calls } = recordingFetch(
|
||||||
|
jsonResponse({}),
|
||||||
|
jsonResponse({}),
|
||||||
|
jsonResponse({}),
|
||||||
|
);
|
||||||
|
const client = new ApiClient({ fetch });
|
||||||
|
await client.getJSON("health");
|
||||||
|
await client.postJSON("users/ott", {});
|
||||||
|
await client.putJSON("/files/thumbnail", {});
|
||||||
|
|
||||||
|
expect(calls.map((c) => c.url)).toEqual([
|
||||||
|
"https://api.ente.io/health",
|
||||||
|
"https://api.ente.io/users/ott",
|
||||||
|
"https://api.ente.io/files/thumbnail",
|
||||||
|
]);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("accepts an apiOrigin with a trailing slash", async () => {
|
||||||
|
const { fetch, calls } = recordingFetch(jsonResponse({}));
|
||||||
|
const client = new ApiClient({
|
||||||
|
fetch,
|
||||||
|
apiOrigin: "https://my-ente.example.com/",
|
||||||
|
});
|
||||||
|
await client.getJSON("/health");
|
||||||
|
|
||||||
|
expect(calls[0]!.url).toBe("https://my-ente.example.com/health");
|
||||||
|
});
|
||||||
|
|
||||||
|
it("keeps a base path in a self-hosted apiOrigin for every request", async () => {
|
||||||
|
const body = new Uint8Array([1]);
|
||||||
|
const { fetch, calls } = recordingFetch(
|
||||||
|
jsonResponse({}),
|
||||||
|
jsonResponse({}),
|
||||||
|
jsonResponse({}),
|
||||||
|
streamResponse(body),
|
||||||
|
streamResponse(body),
|
||||||
|
);
|
||||||
|
const client = new ApiClient({
|
||||||
|
fetch,
|
||||||
|
apiOrigin: "https://example.com/ente/",
|
||||||
|
});
|
||||||
|
await client.getJSON("/collections/v2", { sinceTime: 0 });
|
||||||
|
await client.postJSON("/users/ott", {});
|
||||||
|
await client.putJSON("/files/thumbnail", {});
|
||||||
|
await client.getFileStream(99);
|
||||||
|
await client.getThumbnailStream(77);
|
||||||
|
|
||||||
|
expect(calls.map((c) => c.url)).toEqual([
|
||||||
|
"https://example.com/ente/collections/v2?sinceTime=0",
|
||||||
|
"https://example.com/ente/users/ott",
|
||||||
|
"https://example.com/ente/files/thumbnail",
|
||||||
|
"https://example.com/ente/files/download/99",
|
||||||
|
"https://example.com/ente/files/preview/77",
|
||||||
|
]);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("percent-encodes query parameters and skips undefined ones", async () => {
|
||||||
|
const { fetch, calls } = recordingFetch(jsonResponse({}));
|
||||||
|
const client = new ApiClient({ fetch });
|
||||||
|
await client.getJSON("/search", {
|
||||||
|
q: "a&b=c/d é",
|
||||||
|
limit: 5,
|
||||||
|
cursor: undefined,
|
||||||
|
});
|
||||||
|
|
||||||
|
const url = new URL(calls[0]!.url);
|
||||||
|
expect(url.pathname).toBe("/search");
|
||||||
|
expect(url.search).toBe("?q=a%26b%3Dc%2Fd+%C3%A9&limit=5");
|
||||||
|
expect(url.searchParams.get("q")).toBe("a&b=c/d é");
|
||||||
|
});
|
||||||
|
|
||||||
|
it("rejects a path that carries its own query string", async () => {
|
||||||
|
const { fetch, calls } = recordingFetch();
|
||||||
|
const client = new ApiClient({ fetch });
|
||||||
|
|
||||||
|
await expect(client.getJSON("/diff?sinceTime=0")).rejects.toThrow(
|
||||||
|
/must not contain "\?"/,
|
||||||
|
);
|
||||||
|
await expect(client.postJSON("/users/ott?x=1", {})).rejects.toThrow(
|
||||||
|
/must not contain "\?"/,
|
||||||
|
);
|
||||||
|
expect(calls).toHaveLength(0);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
describe("ApiError", () => {
|
describe("ApiError", () => {
|
||||||
it("throws ApiError on 4xx with status, code, requestID", async () => {
|
it("throws ApiError on 4xx with status, code, requestID", async () => {
|
||||||
const { fetch } = recordingFetch(
|
const { fetch } = recordingFetch(
|
||||||
@@ -661,13 +752,11 @@ describe("ApiClient retries", () => {
|
|||||||
|
|
||||||
describe("ApiClient timeouts", () => {
|
describe("ApiClient timeouts", () => {
|
||||||
it("ships bounded default deadlines", () => {
|
it("ships bounded default deadlines", () => {
|
||||||
// Asserted here so the README and the code cannot drift. Two numbers
|
// Asserted here so the README and the code cannot drift. The request
|
||||||
// rather than one, because a deadline that is sane for a JSON call is
|
// deadline bounds a whole JSON call; the download deadline is an idle
|
||||||
// nowhere near enough for a multi-gigabyte body, and a deadline long
|
// one, measured from the last byte that arrived.
|
||||||
// enough for that body would let a hung API call stall a backup for
|
|
||||||
// ten minutes.
|
|
||||||
expect(DEFAULT_REQUEST_TIMEOUT_MS).toBe(30_000);
|
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 () => {
|
it("attaches an abort signal to every request", async () => {
|
||||||
@@ -781,6 +870,11 @@ describe("ApiClient timeouts", () => {
|
|||||||
// that never produces a chunk and never observes the signal, so the
|
// 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
|
// only thing that can unblock the read is quak's own enforcement of
|
||||||
// the deadline over the stream it hands out.
|
// the deadline over the stream it hands out.
|
||||||
|
//
|
||||||
|
// The clock is faked, so the test runs under the real default
|
||||||
|
// deadline and waits for nothing.
|
||||||
|
vi.useFakeTimers();
|
||||||
|
try {
|
||||||
const stalling = new Response(
|
const stalling = new Response(
|
||||||
new ReadableStream<Uint8Array>({
|
new ReadableStream<Uint8Array>({
|
||||||
pull: () => new Promise<void>(() => {}),
|
pull: () => new Promise<void>(() => {}),
|
||||||
@@ -790,16 +884,77 @@ describe("ApiClient timeouts", () => {
|
|||||||
const { fetch } = scriptedFetch(stalling);
|
const { fetch } = scriptedFetch(stalling);
|
||||||
const client = new ApiClient({
|
const client = new ApiClient({
|
||||||
fetch,
|
fetch,
|
||||||
downloadTimeoutMs: 20,
|
|
||||||
retry: { ...noWait, attempts: 1 },
|
retry: { ...noWait, attempts: 1 },
|
||||||
});
|
});
|
||||||
|
|
||||||
const stream = await client.getFileStream(42);
|
const stream = await client.getFileStream(42);
|
||||||
const err: unknown = await readAll(stream).catch((e: unknown) => e);
|
let settled = false;
|
||||||
|
const result = readAll(stream).then(
|
||||||
|
(n) => n,
|
||||||
|
(e: unknown) => e,
|
||||||
|
);
|
||||||
|
void result.finally(() => {
|
||||||
|
settled = true;
|
||||||
|
});
|
||||||
|
|
||||||
|
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).toBeInstanceOf(Error);
|
||||||
expect((err as Error).name).toBe("TimeoutError");
|
expect((err as Error).name).toBe("TimeoutError");
|
||||||
}, 5000);
|
// 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 () => {
|
it("lets a body that arrives in time through untouched", async () => {
|
||||||
// The counterpart to the previous test: enforcing the deadline over
|
// The counterpart to the previous test: enforcing the deadline over
|
||||||
|
|||||||
@@ -1337,6 +1337,102 @@ describe("download retries: corruption is not retried", () => {
|
|||||||
|
|
||||||
expect(requests()).toBe(1);
|
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);
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
|
|||||||
Reference in New Issue
Block a user