0f708e7
@@ -100,6 +100,7 @@
export const GATE: readonly string[] = [
  "test/git-trust.test.ts",
  "test/helpers.test.ts",
  "test/shards.test.ts",
  "test/snapshot-store.test.ts",
  "test/migrate.test.ts",
  "test/schema-additive.test.ts",
  "test/schema-migrations.test.ts",
@@ -49,7 +49,54 @@
export function s3FromEnv(
  };
}
/**
 * A store that tries again. Object stores answer a burst, a flake or a
 * moment of maintenance with a 5xx — Backblaze says so in its own docs —
 * and one such answer in a thousand puts used to fail the whole snapshot
 * with "S3Error: an unexpected error has occurred" and no key named. Every
 * call is tried `attempts` times with doubling waits, and what finally
 * fails says which key it was.
 */
export function withRetries(
  inner: Store,
  options: {
    attempts?: number;
    baseMs?: number;
    sleep?: (ms: number) => Promise<void>;
  } = {},
): Store {
  const attempts = Math.max(1, options.attempts ?? 5);
  const baseMs = options.baseMs ?? 500;
  const sleep = options.sleep ?? ((ms) => Bun.sleep(ms));
  async function tried<T>(what: string, fn: () => Promise<T>): Promise<T> {
    let last: unknown;
    for (let n = 0; n < attempts; n++) {
      try {
        return await fn();
      } catch (error) {
        last = error;
        if (n + 1 < attempts) await sleep(baseMs * 2 ** n);
      }
    }
    throw new Error(
      `${what} failed after ${attempts} attempt${attempts === 1 ? "" : "s"}: ${String(last instanceof Error ? last.message : last)}`,
    );
  }
  return {
    where: inner.where,
    put: (key, body) => tried(`put ${key}`, () => inner.put(key, body)),
    get: (key) => tried(`get ${key}`, () => inner.get(key)),
    list: (prefix) => tried(`list ${prefix}`, () => inner.list(prefix)),
    names: (prefix) => tried(`names ${prefix}`, () => inner.names(prefix)),
    delete: (key) => tried(`delete ${key}`, () => inner.delete(key)),
  };
}
export function s3Store(target: S3Target): Store {
  return withRetries(rawS3Store(target));
}
function rawS3Store(target: S3Target): Store {
  const client = new Bun.S3Client({
    endpoint: target.endpoint,
    bucket: target.bucket,
@@ -0,0 +1,101 @@
/**
 * The snapshot store tries again. An object store answers a burst or a
 * moment of maintenance with a 5xx; one such answer used to fail a whole
 * nightly snapshot and name no key. `withRetries` tries each call several
 * times with doubling waits and, when it still fails, says which key.
 */
import { describe, expect, test } from "bun:test";
import { withRetries, type Store } from "../src/snapshot/store.ts";
/** A store that fails the first `failures` calls of each kind, then works. */
function flaky(failures: number): { store: Store; calls: string[] } {
  const calls: string[] = [];
  const left = new Map<string, number>();
  const gate = async (what: string): Promise<void> => {
    calls.push(what);
    const n = left.get(what) ?? failures;
    if (n > 0) {
      left.set(what, n - 1);
      throw new Error("S3Error: an unexpected error has occurred");
    }
  };
  const kept = new Map<string, Uint8Array | string>();
  return {
    calls,
    store: {
      where: "flaky",
      async put(key, body) {
        await gate(`put ${key}`);
        kept.set(key, body);
      },
      async get(key) {
        await gate(`get ${key}`);
        const v = kept.get(key);
        return v === undefined
          ? undefined
          : typeof v === "string"
            ? new TextEncoder().encode(v)
            : v;
      },
      async list(prefix) {
        await gate(`list ${prefix}`);
        return [...kept.keys()].filter((k) => k.startsWith(prefix));
      },
      async names(prefix) {
        await gate(`names ${prefix}`);
        return [];
      },
      async delete(key) {
        await gate(`delete ${key}`);
        kept.delete(key);
      },
    },
  };
}
describe("the snapshot store, with retries", () => {
  test("a call that fails twice and then works is one call to the caller, after two waits that double", async () => {
    const waits: number[] = [];
    const f = flaky(2);
    const store = withRetries(f.store, {
      attempts: 5,
      baseMs: 100,
      sleep: async (ms) => {
        waits.push(ms);
      },
    });
    await store.put("snapshots/a/forge.db", "db");
    expect(
      f.calls.filter((c) => c === "put snapshots/a/forge.db"),
    ).toHaveLength(3);
    expect(waits).toEqual([100, 200]);
    expect(await store.list("snapshots/")).toEqual(["snapshots/a/forge.db"]);
  });
  test("a call that never works fails after the attempts, naming the key and the last reason", async () => {
    const f = flaky(99);
    const store = withRetries(f.store, {
      attempts: 3,
      baseMs: 1,
      sleep: async () => {},
    });
    await expect(store.put("snapshots/b/repos/x.bundle", "x")).rejects.toThrow(
      /put snapshots\/b\/repos\/x\.bundle failed after 3 attempts: S3Error: an unexpected error has occurred/,
    );
    expect(f.calls).toHaveLength(3);
  });
  test("a store that works is called once, with no wait", async () => {
    const waits: number[] = [];
    const f = flaky(0);
    const store = withRetries(f.store, {
      sleep: async (ms) => {
        waits.push(ms);
      },
    });
    await store.put("k", "v");
    expect(new TextDecoder().decode(await store.get("k"))).toBe("v");
    expect(waits).toEqual([]);
    expect(f.calls).toEqual(["put k", "get k"]);
  });
});
		@@ -100,6 +100,7 @@ export const GATE: readonly string[] = [
100	100	"test/git-trust.test.ts",
101	101	"test/helpers.test.ts",
102	102	"test/shards.test.ts",
	103	"test/snapshot-store.test.ts",
103	104	"test/migrate.test.ts",
104	105	"test/schema-additive.test.ts",
105	106	"test/schema-migrations.test.ts",
		@@ -49,7 +49,54 @@ export function s3FromEnv(
49	49	};
50	50	}
51	51
	52	/**
	53	* A store that tries again. Object stores answer a burst, a flake or a
	54	* moment of maintenance with a 5xx — Backblaze says so in its own docs —
	55	* and one such answer in a thousand puts used to fail the whole snapshot
	56	* with "S3Error: an unexpected error has occurred" and no key named. Every
	57	* call is tried `attempts` times with doubling waits, and what finally
	58	* fails says which key it was.
	59	*/
	60	export function withRetries(
	61	inner: Store,
	62	options: {
	63	attempts?: number;
	64	baseMs?: number;
	65	sleep?: (ms: number) => Promise<void>;
	66	} = {},
	67	): Store {
	68	const attempts = Math.max(1, options.attempts ?? 5);
	69	const baseMs = options.baseMs ?? 500;
	70	const sleep = options.sleep ?? ((ms) => Bun.sleep(ms));
	71	async function tried<T>(what: string, fn: () => Promise<T>): Promise<T> {
	72	let last: unknown;
	73	for (let n = 0; n < attempts; n++) {
	74	try {
	75	return await fn();
	76	} catch (error) {
	77	last = error;
	78	if (n + 1 < attempts) await sleep(baseMs * 2 ** n);
	79	}
	80	}
	81	throw new Error(
	82	`${what} failed after ${attempts} attempt${attempts === 1 ? "" : "s"}: ${String(last instanceof Error ? last.message : last)}`,
	83	);
	84	}
	85	return {
	86	where: inner.where,
	87	put: (key, body) => tried(`put ${key}`, () => inner.put(key, body)),
	88	get: (key) => tried(`get ${key}`, () => inner.get(key)),
	89	list: (prefix) => tried(`list ${prefix}`, () => inner.list(prefix)),
	90	names: (prefix) => tried(`names ${prefix}`, () => inner.names(prefix)),
	91	delete: (key) => tried(`delete ${key}`, () => inner.delete(key)),
	92	};
	93	}
	94
52	95	export function s3Store(target: S3Target): Store {
	96	return withRetries(rawS3Store(target));
	97	}
	98
	99	function rawS3Store(target: S3Target): Store {
53	100	const client = new Bun.S3Client({
54	101	endpoint: target.endpoint,
55	102	bucket: target.bucket,
		@@ -0,0 +1,101 @@
	1	/**
	2	* The snapshot store tries again. An object store answers a burst or a
	3	* moment of maintenance with a 5xx; one such answer used to fail a whole
	4	* nightly snapshot and name no key. `withRetries` tries each call several
	5	* times with doubling waits and, when it still fails, says which key.
	6	*/
	7	import { describe, expect, test } from "bun:test";
	8	import { withRetries, type Store } from "../src/snapshot/store.ts";
	9
	10	/** A store that fails the first `failures` calls of each kind, then works. */
	11	function flaky(failures: number): { store: Store; calls: string[] } {
	12	const calls: string[] = [];
	13	const left = new Map<string, number>();
	14	const gate = async (what: string): Promise<void> => {
	15	calls.push(what);
	16	const n = left.get(what) ?? failures;
	17	if (n > 0) {
	18	left.set(what, n - 1);
	19	throw new Error("S3Error: an unexpected error has occurred");
	20	}
	21	};
	22	const kept = new Map<string, Uint8Array \| string>();
	23	return {
	24	calls,
	25	store: {
	26	where: "flaky",
	27	async put(key, body) {
	28	await gate(`put ${key}`);
	29	kept.set(key, body);
	30	},
	31	async get(key) {
	32	await gate(`get ${key}`);
	33	const v = kept.get(key);
	34	return v === undefined
	35	? undefined
	36	: typeof v === "string"
	37	? new TextEncoder().encode(v)
	38	: v;
	39	},
	40	async list(prefix) {
	41	await gate(`list ${prefix}`);
	42	return [...kept.keys()].filter((k) => k.startsWith(prefix));
	43	},
	44	async names(prefix) {
	45	await gate(`names ${prefix}`);
	46	return [];
	47	},
	48	async delete(key) {
	49	await gate(`delete ${key}`);
	50	kept.delete(key);
	51	},
	52	},
	53	};
	54	}
	55
	56	describe("the snapshot store, with retries", () => {
	57	test("a call that fails twice and then works is one call to the caller, after two waits that double", async () => {
	58	const waits: number[] = [];
	59	const f = flaky(2);
	60	const store = withRetries(f.store, {
	61	attempts: 5,
	62	baseMs: 100,
	63	sleep: async (ms) => {
	64	waits.push(ms);
	65	},
	66	});
	67	await store.put("snapshots/a/forge.db", "db");
	68	expect(
	69	f.calls.filter((c) => c === "put snapshots/a/forge.db"),
	70	).toHaveLength(3);
	71	expect(waits).toEqual([100, 200]);
	72	expect(await store.list("snapshots/")).toEqual(["snapshots/a/forge.db"]);
	73	});
	74
	75	test("a call that never works fails after the attempts, naming the key and the last reason", async () => {
	76	const f = flaky(99);
	77	const store = withRetries(f.store, {
	78	attempts: 3,
	79	baseMs: 1,
	80	sleep: async () => {},
	81	});
	82	await expect(store.put("snapshots/b/repos/x.bundle", "x")).rejects.toThrow(
	83	/put snapshots\/b\/repos\/x\.bundle failed after 3 attempts: S3Error: an unexpected error has occurred/,
	84	);
	85	expect(f.calls).toHaveLength(3);
	86	});
	87
	88	test("a store that works is called once, with no wait", async () => {
	89	const waits: number[] = [];
	90	const f = flaky(0);
	91	const store = withRetries(f.store, {
	92	sleep: async (ms) => {
	93	waits.push(ms);
	94	},
	95	});
	96	await store.put("k", "v");
	97	expect(new TextDecoder().decode(await store.get("k"))).toBe("v");
	98	expect(waits).toEqual([]);
	99	expect(f.calls).toEqual(["put k", "get k"]);
	100	});
	101	});