fix(catalog) : retry transient storage quorum failures
This commit is contained in:
@@ -0,0 +1,56 @@
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
import { retryTransient } from "./sync-retry";
|
||||
|
||||
const label = "Asset catalog/materials/UI_ItemIcon_121767.png";
|
||||
const quorumError = new Error('Internal error: Could not reach quorum of 2 (sets=Some(1)). 1 of 2 request succeeded, others returned errors: ["5f70d8ec32eade44: Network error: Not connected: 5f70d8ec32eade44"]');
|
||||
|
||||
afterEach(() => vi.useRealTimers());
|
||||
|
||||
describe("catalog sync retries", () => {
|
||||
it.each([quorumError, new Error("Network error: Not connected: 5f70d8ec32eade44"), new Error("Request timed out")])("recovers from %s", async (cause) => {
|
||||
vi.useFakeTimers();
|
||||
const task = vi.fn().mockRejectedValueOnce(cause).mockResolvedValue("uploaded");
|
||||
const result = retryTransient(label, task);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(999);
|
||||
expect(task).toHaveBeenCalledTimes(1);
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await expect(result).resolves.toBe("uploaded");
|
||||
expect(task).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("stops after three failed attempts and preserves the asset and cause", async () => {
|
||||
vi.useFakeTimers();
|
||||
const task = vi.fn().mockRejectedValue(quorumError);
|
||||
const result = expect(retryTransient(label, task)).rejects.toMatchObject({
|
||||
message: `${label}: ${quorumError.message}`,
|
||||
cause: quorumError,
|
||||
});
|
||||
|
||||
await vi.advanceTimersByTimeAsync(1_000);
|
||||
expect(task).toHaveBeenCalledTimes(2);
|
||||
await vi.advanceTimersByTimeAsync(2_000);
|
||||
await result;
|
||||
expect(task).toHaveBeenCalledTimes(3);
|
||||
});
|
||||
|
||||
it("fails immediately on a permanent storage error", async () => {
|
||||
const task = vi.fn().mockRejectedValue(new Error("AccessDenied"));
|
||||
await expect(retryTransient(label, task)).rejects.toThrow(`${label}: AccessDenied`);
|
||||
expect(task).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("does not retry after cancellation during backoff", async () => {
|
||||
vi.useFakeTimers();
|
||||
const controller = new AbortController();
|
||||
const task = vi.fn().mockRejectedValue(quorumError);
|
||||
const result = expect(retryTransient(label, task, controller.signal)).rejects.toMatchObject({ name: "AbortError" });
|
||||
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
controller.abort();
|
||||
await vi.advanceTimersByTimeAsync(1_000);
|
||||
await result;
|
||||
expect(task).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,16 @@
|
||||
export async function retryTransient<T>(label: string, task: () => Promise<T>, signal?: AbortSignal): Promise<T> {
|
||||
for (let attempt = 1; ; attempt += 1) {
|
||||
signal?.throwIfAborted();
|
||||
try {
|
||||
return await task();
|
||||
} catch (cause) {
|
||||
signal?.throwIfAborted();
|
||||
const message = cause instanceof Error ? cause.message : String(cause);
|
||||
// Storage quorum failures can recover when a disconnected replica rejoins.
|
||||
if (attempt >= 3 || !/timed out|timeout|unexpected error has occurred|could not reach quorum|network error: not connected/i.test(message)) {
|
||||
throw new Error(`${label}: ${message}`, { cause });
|
||||
}
|
||||
await new Promise((resolve) => setTimeout(resolve, attempt * 1_000));
|
||||
}
|
||||
}
|
||||
}
|
||||
+1
-16
@@ -35,6 +35,7 @@ import type {
|
||||
} from "@/lib/catalog/lunaris";
|
||||
import { customTravelers, isUpstreamTraveler } from "@/lib/catalog/travelers";
|
||||
import { withCatalogSyncLock } from "@/lib/catalog/sync-lock";
|
||||
import { retryTransient } from "@/lib/catalog/sync-retry";
|
||||
import { getRedisClient, redisEventPrefix } from "@/lib/redis/client";
|
||||
import {
|
||||
catalogFormatVersion,
|
||||
@@ -128,22 +129,6 @@ interface SyncedCharacter {
|
||||
|
||||
type SyncControl = { signal?: AbortSignal; checkCancelled: () => Promise<void>; report: (phase: string, completed: number, total: number, message: string) => Promise<void> };
|
||||
|
||||
async function retryTransient<T>(label: string, task: () => Promise<T>, signal?: AbortSignal): Promise<T> {
|
||||
for (let attempt = 1; ; attempt += 1) {
|
||||
signal?.throwIfAborted();
|
||||
try {
|
||||
return await task();
|
||||
} catch (cause) {
|
||||
signal?.throwIfAborted();
|
||||
const message = cause instanceof Error ? cause.message : String(cause);
|
||||
if (attempt >= 3 || !/timed out|timeout|unexpected error has occurred/i.test(message)) {
|
||||
throw new Error(`${label}: ${message}`, { cause });
|
||||
}
|
||||
await new Promise((resolve) => setTimeout(resolve, attempt * 1_000));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function getJson(url: string, signal?: AbortSignal): Promise<unknown> {
|
||||
return retryTransient(`Lunaris request ${url}`, async () => {
|
||||
const response = await fetch(url, {
|
||||
|
||||
Reference in New Issue
Block a user