From 5a5ad40ccb18fedf66f63ce291af710dde3f7f4b Mon Sep 17 00:00:00 2001 From: Florian <45694132+flo-bit@users.noreply.github.com> Date: Tue, 4 Aug 2026 17:30:13 +0200 Subject: [PATCH] Clarify backfill states and cap retries --- .changeset/backfill-status.md | 2 +- README.md | 2 +- apps/cloudflare-workers/README.md | 6 +- docs/01-indexing.md | 2 +- docs/02-querying.md | 11 +- packages/contrail/src/cli/commands/dev.ts | 2 +- packages/contrail/src/contrail.ts | 5 +- packages/contrail/src/core/backfill.ts | 81 +++++++++--- packages/contrail/src/core/db/schema.ts | 8 +- packages/contrail/src/core/status.ts | 71 +++++----- .../contrail/tests/backfill-status.test.ts | 123 ++++++++++++++++-- packages/contrail/tests/schema.test.ts | 12 +- 12 files changed, 241 insertions(+), 84 deletions(-) diff --git a/.changeset/backfill-status.md b/.changeset/backfill-status.md index b37ebd3..d7d9e36 100644 --- a/.changeset/backfill-status.md +++ b/.changeset/backfill-status.md @@ -2,4 +2,4 @@ "@atmo-dev/contrail": patch --- -Keep failed PDS and relay work pending with bounded retries instead of marking account rows complete. Retry due PDS accounts automatically in small scheduled slices with persisted exponential backoff and an overlap lease. Add durable backfill and discovery state to the JSON overview and `/status` endpoint, including running/complete state, pending tasks, retry timing, and unreachable-account counts. +Keep failed PDS and relay work pending with bounded retries instead of marking account rows complete. Retry due PDS accounts automatically in small scheduled slices with persisted exponential backoff up to 48 hours, a ten-attempt limit, and an overlap lease. Add durable backfill and discovery state to the JSON overview and `/status` endpoint, including running/complete state, retry timing, per-collection progress, and mutually exclusive complete/pending/retrying/failed account counts. diff --git a/README.md b/README.md index 793a4b8..94f211b 100644 --- a/README.md +++ b/README.md @@ -79,7 +79,7 @@ GET /xrpc/com.example.event.listRecords?startsAtMin=2026-01-01&limit=10 GET /status ``` -The JSON status response reports live cursor lag, indexed records, known backfill progress, and currently unreachable accounts. Failed PDS work remains pending and is retried automatically in small, backed-off slices after scheduled live ingestion. +The JSON status response reports live cursor lag, indexed records, known backfill progress, and mutually exclusive pending/retrying/failed account counts. Failed PDS work is retried automatically in small scheduled slices with backoff up to 48 hours. For ordinary Lexicon parsing, validation, pulling, and TypeScript generation, use [Atcute](https://github.com/mary-ext/atcute) directly. Contrail no longer ships a separate Lexicon toolchain. diff --git a/apps/cloudflare-workers/README.md b/apps/cloudflare-workers/README.md index f229da9..a35d057 100644 --- a/apps/cloudflare-workers/README.md +++ b/apps/cloudflare-workers/README.md @@ -28,7 +28,7 @@ GET https://.workers.dev/xrpc/com.example.event.listRecords?startsA GET https://.workers.dev/status ``` -`/status` returns JSON with the live-ingest cursor, indexed record totals, discovery progress, known backfill work, and counts of pending or currently unreachable accounts. +`/status` returns JSON with the live-ingest cursor, indexed record totals, discovery progress, per-collection progress, and mutually exclusive complete/pending/retrying/failed account counts. ## local dev @@ -39,13 +39,13 @@ pnpm contrail backfill # backfill against the local D1 created by wrangler `pnpm dev` runs `contrail dev` under the hood. On start it inspects the local D1 and offers to start or resume backfill whenever known work remains, then runs `wrangler dev --test-scheduled` with a 60s timer hitting `/__scheduled` so the cron actually fires locally. -A failed PDS account stays pending rather than being marked complete. The initial command uses a bounded attempt budget, then records an exponential retry time. The normal one-minute Worker cron retries a small due slice after live ingestion, so dead PDSes cannot monopolize the tick. Rerunning the command forces an immediate bounded pass. +A failed PDS account stays incomplete rather than being marked complete. The initial command uses a bounded attempt budget, then records an exponential retry time. The normal one-minute Worker cron retries a small due slice after live ingestion, with delays from 15 minutes up to 48 hours and a ten-attempt limit, so dead PDSes cannot monopolize the tick. Rerunning the command resets exhausted rows and forces an immediate bounded pass. ```bash curl http://localhost:8787/status ``` -This reports the durable state without exposing account DIDs or raw upstream errors. `state: "running"` means a manual or scheduled backfill slice currently owns the database lease. `state: "complete"` means discovery and the initial pass finished; `accounts.unreachable` and `retries` may still show deferred account work that cron will continue retrying. +This reports the durable state without exposing account DIDs or raw upstream errors. `state: "running"` means a manual or scheduled backfill slice currently owns the database lease. `state: "complete"` means discovery and the initial pass finished; account counts separately show work that is `pending`, `retrying`, or permanently `failed` until an explicit reset. ## extending diff --git a/docs/01-indexing.md b/docs/01-indexing.md index b4de6d6..c7dfb01 100644 --- a/docs/01-indexing.md +++ b/docs/01-indexing.md @@ -58,7 +58,7 @@ await contrail.backfill({ concurrency: 100 }); // fetch history for registered D `backfill()` picks up each account/collection at its saved PDS cursor. A row is marked complete only after the PDS listing reaches its end. Timeouts, failed identity resolution, `429`, and `5xx` responses leave the row pending with its last error. -Each initial invocation has a bounded failure budget (five attempts by default), so one dead PDS cannot hang the whole command forever. Failed account rows retain their cursors, receive an exponential `next_retry_at`, and remain incomplete. Cloudflare's scheduled Worker retries a small due slice after each live-ingest cycle; an explicit later invocation forces another bounded pass. `backfillAll()` returns a durable `status` summary alongside the number of discovered accounts and accepted records. +Each initial invocation has a bounded failure budget (five attempts by default), so one dead PDS cannot hang the whole command forever. Failed account rows retain their cursors and receive an exponential `next_retry_at`. Scheduled retries start at 15 minutes, double to a maximum of 48 hours, and stop after ten consecutive failures. Cloudflare's scheduled Worker retries a small due slice after each live-ingest cycle; an explicit later invocation resets exhausted rows and forces another bounded pass. `backfillAll()` returns a durable `status` summary alongside the number of discovered accounts and accepted records. ### Workers CLI diff --git a/docs/02-querying.md b/docs/02-querying.md index e54643a..6997372 100644 --- a/docs/02-querying.md +++ b/docs/02-querying.md @@ -32,14 +32,13 @@ Dotted field names become camelCase params — `queryable: { "subject.uri": {} } `GET /status` and `GET /xrpc/{namespace}.getOverview` return the current JSON overview. It includes indexed record totals, the live-ingest cursor and lag, and durable backfill state: -- discovery sources complete, pending, or failed; -- known account totals, including pending and currently unreachable accounts; -- account/collection tasks complete, pending, or failed; -- known-task completion percentage; -- per-collection backfill progress; and +- discovery source progress; +- mutually exclusive account totals for `complete`, `pending`, `retrying`, and `failed`; +- known-account completion percentage; +- the same mutually exclusive totals per collection; and - scheduled/due account retries plus the next retry time. -`state` is `running` while a manual or scheduled slice holds the backfill lease. It becomes `complete` after discovery and the initial pass finish, even when `accounts.unreachable` and `tasks.failed` are non-zero; those account rows are still incomplete and remain scheduled for background retry. `incomplete` means discovery or an initial account attempt has not finished. +`state` is `running` while a manual or scheduled slice holds the backfill lease. It becomes `complete` after discovery and the initial pass finish, even when account-level `retrying` or `failed` counts are non-zero. `pending` means no failure has occurred yet, `retrying` means at least one attempt failed but automatic attempts remain, and `failed` means the ten-attempt automatic budget is exhausted. An explicit backfill resets that budget. `incomplete` means discovery or an initial account attempt has not finished. "Known" is deliberate: while relay discovery is incomplete, Contrail cannot honestly claim how many accounts remain undiscovered. `/health` remains a lightweight liveness response and does not claim that historical backfill is complete. diff --git a/packages/contrail/src/cli/commands/dev.ts b/packages/contrail/src/cli/commands/dev.ts index be997ea..73fcbee 100644 --- a/packages/contrail/src/cli/commands/dev.ts +++ b/packages/contrail/src/cli/commands/dev.ts @@ -61,7 +61,7 @@ export function registerDev(cli: CAC): void { console.log("no backfilled users in the local DB yet."); } else { console.log( - `backfill incomplete: ${backfillStatus.accounts.pending} accounts remain; ${backfillStatus.accounts.unreachable} currently unreachable.` + `backfill incomplete: ${backfillStatus.accounts.pending} pending, ${backfillStatus.accounts.retrying} retrying, ${backfillStatus.accounts.failed} failed.` ); } if ( diff --git a/packages/contrail/src/contrail.ts b/packages/contrail/src/contrail.ts index 184f066..c75c1f3 100644 --- a/packages/contrail/src/contrail.ts +++ b/packages/contrail/src/contrail.ts @@ -230,9 +230,10 @@ export class Contrail { const status = await getBackfillStatus(d, this.config); const elapsedS = ((Date.now() - startedAt) / 1000).toFixed(1); - if (status.state === "complete" && status.accounts.unreachable > 0) { + const unresolved = status.accounts.retrying + status.accounts.failed; + if (status.state === "complete" && unresolved > 0) { logger?.warn?.( - ` done with deferred failures: ${status.accounts.complete}/${status.accounts.total} known accounts complete; ${status.accounts.unreachable} unavailable accounts scheduled for retry (${elapsedS}s)` + ` done with deferred failures: ${status.accounts.complete}/${status.accounts.total} known accounts complete; ${status.accounts.retrying} retrying, ${status.accounts.failed} exhausted (${elapsedS}s)` ); } else if (status.state === "complete") { logger?.log?.( diff --git a/packages/contrail/src/core/backfill.ts b/packages/contrail/src/core/backfill.ts index 368ae89..fda0767 100644 --- a/packages/contrail/src/core/backfill.ts +++ b/packages/contrail/src/core/backfill.ts @@ -49,8 +49,9 @@ const PAGE_SIZE = 100; const DEFAULT_MAX_ATTEMPTS = 5; const REQUEST_TIMEOUT_MS = 10_000; -const BACKFILL_RETRY_BASE_MS = 60_000; -const BACKFILL_RETRY_MAX_MS = 60 * 60_000; +const BACKFILL_RETRY_BASE_MS = 15 * 60_000; +const BACKFILL_RETRY_MAX_MS = 48 * 60 * 60_000; +const DEFAULT_SCHEDULED_MAX_ATTEMPTS = 10; function positiveInteger(value: number | undefined, fallback: number): number { return typeof value === "number" && Number.isFinite(value) && value > 0 @@ -92,7 +93,8 @@ async function markFailed( db: Database, did: string, collection: string, - error: unknown + error: unknown, + exhaustAfterAttempts?: number ): Promise { const row = await db .prepare( @@ -102,19 +104,22 @@ async function markFailed( .first<{ retries: number }>(); const retries = (row?.retries ?? 0) + 1; const now = Date.now(); + const exhausted = + exhaustAfterAttempts !== undefined && retries >= exhaustAfterAttempts; const retryDelay = Math.min( - BACKFILL_RETRY_BASE_MS * 2 ** Math.min(retries - 1, 6), + BACKFILL_RETRY_BASE_MS * 2 ** Math.min(retries - 1, 16), BACKFILL_RETRY_MAX_MS ); await db .prepare( - "UPDATE backfills SET retries = ?, last_error = ?, last_attempt_at = ?, next_retry_at = ? WHERE did = ? AND collection = ?" + "UPDATE backfills SET retries = ?, last_error = ?, last_attempt_at = ?, next_retry_at = ?, retry_exhausted = ? WHERE did = ? AND collection = ?" ) .bind( retries, errorMessage(error), now, - now + retryDelay, + exhausted ? null : now + retryDelay, + exhausted ? 1 : 0, did, collection ) @@ -130,6 +135,8 @@ export interface BackfillOptions { maxRetries?: number; /** Per-request timeout in ms (default: 10000). */ requestTimeout?: number; + /** Mark the row terminal when this consecutive-failure count is reached. */ + exhaustAfterAttempts?: number; } interface BackfillUserAttempt { @@ -170,12 +177,24 @@ async function backfillUserAttempt( const timeout = options?.requestTimeout ?? REQUEST_TIMEOUT_MS; if (!isDid(did)) { - await markFailed(db, did, collection, `Invalid DID: ${did}`); + await markFailed( + db, + did, + collection, + `Invalid DID: ${did}`, + options?.exhaustAfterAttempts + ); return { records: 0, completed: false }; } if (!isNsid(collection)) { - await markFailed(db, did, collection, `Invalid NSID: ${collection}`); + await markFailed( + db, + did, + collection, + `Invalid NSID: ${collection}`, + options?.exhaustAfterAttempts + ); return { records: 0, completed: false }; } @@ -189,7 +208,13 @@ async function backfillUserAttempt( timeout ); } catch (err) { - await markFailed(db, did, collection, err); + await markFailed( + db, + did, + collection, + err, + options?.exhaustAfterAttempts + ); return { records: 0, completed: false }; } } @@ -221,7 +246,8 @@ async function backfillUserAttempt( db, did, collection, - `listRecords status ${response.status} (${detail})` + `listRecords status ${response.status} (${detail})`, + options?.exhaustAfterAttempts ); return { records: totalInserted, completed: false }; } @@ -261,7 +287,7 @@ async function backfillUserAttempt( await db .prepare( - "UPDATE backfills SET pds_cursor = ?, retries = 0, last_error = NULL, last_attempt_at = ?, next_retry_at = NULL WHERE did = ? AND collection = ?" + "UPDATE backfills SET pds_cursor = ?, retries = 0, last_error = NULL, last_attempt_at = ?, next_retry_at = NULL, retry_exhausted = 0 WHERE did = ? AND collection = ?" ) .bind(currentCursor ?? null, now, did, collection) .run(); @@ -272,14 +298,20 @@ async function backfillUserAttempt( } } } catch (err) { - await markFailed(db, did, collection, err); + await markFailed( + db, + did, + collection, + err, + options?.exhaustAfterAttempts + ); return { records: totalInserted, completed: false }; } if (done) { await db .prepare( - "UPDATE backfills SET completed = 1, retries = 0, last_error = NULL, last_attempt_at = ?, next_retry_at = NULL WHERE did = ? AND collection = ?" + "UPDATE backfills SET completed = 1, retries = 0, last_error = NULL, last_attempt_at = ?, next_retry_at = NULL, retry_exhausted = 0 WHERE did = ? AND collection = ?" ) .bind(Date.now(), did, collection) .run(); @@ -346,7 +378,9 @@ async function backfillPendingWork( // The attempt cap is per invocation. Rows remain incomplete and a later run // gets a fresh bounded retry budget. await db - .prepare("UPDATE backfills SET retries = 0 WHERE completed = 0") + .prepare( + "UPDATE backfills SET retries = 0, next_retry_at = NULL, retry_exhausted = 0 WHERE completed = 0" + ) .run(); while (true) { @@ -525,6 +559,8 @@ export async function backfillPending( export interface BackfillRetryOptions { /** Maximum accounts to attempt in one scheduled slice. Default: 5. */ maxAccounts?: number; + /** Consecutive failures before automatic retries stop. Default: 10. */ + maxAttempts?: number; /** Total wall-clock budget for the slice. Default: 10000ms. */ timeoutMs?: number; /** Deadline for each PDS request within the slice. Default: 3000ms. */ @@ -552,6 +588,10 @@ export async function retryPendingBackfills( } const maxAccounts = positiveInteger(options?.maxAccounts, 5); + const maxAttempts = positiveInteger( + options?.maxAttempts, + DEFAULT_SCHEDULED_MAX_ATTEMPTS + ); const timeoutMs = positiveInteger(options?.timeoutMs, 10_000); const requestTimeoutMs = positiveInteger(options?.requestTimeoutMs, 3_000); const deadline = Date.now() + timeoutMs; @@ -563,7 +603,7 @@ export async function retryPendingBackfills( try { const due = await db .prepare( - "SELECT did FROM backfills WHERE completed = 0 AND (next_retry_at IS NULL OR next_retry_at <= ?) GROUP BY did ORDER BY MIN(COALESCE(next_retry_at, 0)), did LIMIT ?" + "SELECT did FROM backfills WHERE completed = 0 AND retry_exhausted = 0 AND (next_retry_at IS NULL OR next_retry_at <= ?) GROUP BY did ORDER BY MIN(COALESCE(next_retry_at, 0)), did LIMIT ?" ) .bind(Date.now(), maxAccounts) .all<{ did: string }>(); @@ -573,7 +613,7 @@ export async function retryPendingBackfills( attempted++; const pending = await db .prepare( - "SELECT collection FROM backfills WHERE did = ? AND completed = 0 AND (next_retry_at IS NULL OR next_retry_at <= ?) ORDER BY collection" + "SELECT collection FROM backfills WHERE did = ? AND completed = 0 AND retry_exhausted = 0 AND (next_retry_at IS NULL OR next_retry_at <= ?) ORDER BY collection" ) .bind(did, Date.now()) .all<{ collection: string }>(); @@ -590,7 +630,13 @@ export async function retryPendingBackfills( ); } catch (error) { for (const collection of collections) { - await markFailed(db, did, collection, error); + await markFailed( + db, + did, + collection, + error, + maxAttempts + ); } failed++; await heartbeatBackfillRun(db, runId); @@ -609,6 +655,7 @@ export async function retryPendingBackfills( requestTimeoutMs, Math.max(1, deadline - Date.now()) ), + exhaustAfterAttempts: maxAttempts, }) ); } diff --git a/packages/contrail/src/core/db/schema.ts b/packages/contrail/src/core/db/schema.ts index c8bdbf3..377b145 100644 --- a/packages/contrail/src/core/db/schema.ts +++ b/packages/contrail/src/core/db/schema.ts @@ -17,7 +17,7 @@ import { getSearchableFields } from "../search"; import { buildLabelsSchema } from "../labels/schema"; import { getMeta, setMeta } from "./meta"; -export const CONTRAIL_SCHEMA_VERSION = 4; +export const CONTRAIL_SCHEMA_VERSION = 5; const SCHEMA_FINGERPRINT_KEY = "schema_fingerprint"; function getResolved(config: ContrailConfig): ResolvedMaps { @@ -41,6 +41,7 @@ CREATE TABLE IF NOT EXISTS backfills ( last_error TEXT, last_attempt_at ${dialect.bigintType}, next_retry_at ${dialect.bigintType}, + retry_exhausted INTEGER NOT NULL DEFAULT 0, PRIMARY KEY (did, collection) ); CREATE TABLE IF NOT EXISTS discovery ( @@ -318,6 +319,11 @@ const MIGRATIONS: MigrationOp[] = [ { table: "backfills", column: "last_error", columnDef: "TEXT" }, { table: "backfills", column: "last_attempt_at", columnDef: "BIGINT" }, { table: "backfills", column: "next_retry_at", columnDef: "BIGINT" }, + { + table: "backfills", + column: "retry_exhausted", + columnDef: "INTEGER NOT NULL DEFAULT 0", + }, { table: "discovery", column: "retries", diff --git a/packages/contrail/src/core/status.ts b/packages/contrail/src/core/status.ts index ef19064..a21104d 100644 --- a/packages/contrail/src/core/status.ts +++ b/packages/contrail/src/core/status.ts @@ -5,24 +5,19 @@ interface AggregateRow { total: number | string | null; complete: number | string | null; pending: number | string | null; + retrying: number | string | null; failed: number | string | null; } -export interface BackfillCounts { - total: number; - complete: number; - pending: number; - failed: number; -} - export interface BackfillAccountCounts { total: number; complete: number; pending: number; - unreachable: number; + retrying: number; + failed: number; } -export interface BackfillCollectionStatus extends BackfillCounts { +export interface BackfillCollectionStatus extends BackfillAccountCounts { collection: string; } @@ -37,8 +32,7 @@ export interface BackfillStatus { state: "not_started" | "running" | "incomplete" | "complete"; known_progress_percent: number; accounts: BackfillAccountCounts; - tasks: BackfillCounts; - discovery: BackfillCounts; + discovery: BackfillAccountCounts; retries: BackfillRetryStatus; collections: BackfillCollectionStatus[]; } @@ -47,11 +41,12 @@ function number(value: number | string | null | undefined): number { return Number(value ?? 0); } -function counts(row: AggregateRow | null): BackfillCounts { +function counts(row: AggregateRow | null): BackfillAccountCounts { return { total: number(row?.total), complete: number(row?.complete), pending: number(row?.pending), + retrying: number(row?.retrying), failed: number(row?.failed) }; } @@ -122,8 +117,9 @@ export async function getBackfillStatus( `SELECT COUNT(*) AS total, COALESCE(SUM(CASE WHEN completed = 1 THEN 1 ELSE 0 END), 0) AS complete, - COALESCE(SUM(CASE WHEN completed = 0 THEN 1 ELSE 0 END), 0) AS pending, - COALESCE(SUM(CASE WHEN completed = 0 AND last_error IS NOT NULL THEN 1 ELSE 0 END), 0) AS failed + COALESCE(SUM(CASE WHEN completed = 0 AND retry_exhausted = 0 AND last_error IS NULL THEN 1 ELSE 0 END), 0) AS pending, + COALESCE(SUM(CASE WHEN completed = 0 AND retry_exhausted = 0 AND last_error IS NOT NULL THEN 1 ELSE 0 END), 0) AS retrying, + COALESCE(SUM(CASE WHEN completed = 0 AND retry_exhausted = 1 THEN 1 ELSE 0 END), 0) AS failed FROM backfills` ) .first(), @@ -132,13 +128,16 @@ export async function getBackfillStatus( `SELECT COUNT(*) AS total, COALESCE(SUM(CASE WHEN completed = 1 THEN 1 ELSE 0 END), 0) AS complete, - COALESCE(SUM(CASE WHEN completed = 0 THEN 1 ELSE 0 END), 0) AS pending, - COALESCE(SUM(CASE WHEN completed = 0 AND unreachable = 1 THEN 1 ELSE 0 END), 0) AS failed + COALESCE(SUM(CASE WHEN completed = 0 AND has_retrying = 0 AND has_failed = 0 THEN 1 ELSE 0 END), 0) AS pending, + COALESCE(SUM(CASE WHEN completed = 0 AND (has_retrying = 1 OR (has_failed = 1 AND has_pending = 1)) THEN 1 ELSE 0 END), 0) AS retrying, + COALESCE(SUM(CASE WHEN completed = 0 AND has_failed = 1 AND has_pending = 0 AND has_retrying = 0 THEN 1 ELSE 0 END), 0) AS failed FROM ( SELECT did, MIN(completed) AS completed, - MAX(CASE WHEN completed = 0 AND last_error IS NOT NULL THEN 1 ELSE 0 END) AS unreachable + MAX(CASE WHEN completed = 0 AND retry_exhausted = 0 AND last_error IS NULL THEN 1 ELSE 0 END) AS has_pending, + MAX(CASE WHEN completed = 0 AND retry_exhausted = 0 AND last_error IS NOT NULL THEN 1 ELSE 0 END) AS has_retrying, + MAX(CASE WHEN completed = 0 AND retry_exhausted = 1 THEN 1 ELSE 0 END) AS has_failed FROM backfills GROUP BY did ) AS accounts` @@ -149,8 +148,9 @@ export async function getBackfillStatus( `SELECT COUNT(*) AS total, COALESCE(SUM(CASE WHEN completed = 1 THEN 1 ELSE 0 END), 0) AS complete, - COALESCE(SUM(CASE WHEN completed = 0 THEN 1 ELSE 0 END), 0) AS pending, - COALESCE(SUM(CASE WHEN completed = 0 AND last_error IS NOT NULL THEN 1 ELSE 0 END), 0) AS failed + COALESCE(SUM(CASE WHEN completed = 0 AND last_error IS NULL THEN 1 ELSE 0 END), 0) AS pending, + COALESCE(SUM(CASE WHEN completed = 0 AND last_error IS NOT NULL THEN 1 ELSE 0 END), 0) AS retrying, + 0 AS failed FROM discovery` ) .first(), @@ -160,8 +160,9 @@ export async function getBackfillStatus( collection, COUNT(*) AS total, COALESCE(SUM(CASE WHEN completed = 1 THEN 1 ELSE 0 END), 0) AS complete, - COALESCE(SUM(CASE WHEN completed = 0 THEN 1 ELSE 0 END), 0) AS pending, - COALESCE(SUM(CASE WHEN completed = 0 AND last_error IS NOT NULL THEN 1 ELSE 0 END), 0) AS failed + COALESCE(SUM(CASE WHEN completed = 0 AND retry_exhausted = 0 AND last_error IS NULL THEN 1 ELSE 0 END), 0) AS pending, + COALESCE(SUM(CASE WHEN completed = 0 AND retry_exhausted = 0 AND last_error IS NOT NULL THEN 1 ELSE 0 END), 0) AS retrying, + COALESCE(SUM(CASE WHEN completed = 0 AND retry_exhausted = 1 THEN 1 ELSE 0 END), 0) AS failed FROM backfills GROUP BY collection ORDER BY collection` @@ -176,7 +177,7 @@ export async function getBackfillStatus( FROM ( SELECT did, MIN(next_retry_at) AS next_retry_at FROM backfills - WHERE completed = 0 AND last_error IS NOT NULL + WHERE completed = 0 AND retry_exhausted = 0 AND last_error IS NOT NULL GROUP BY did ) AS retry_accounts` ) @@ -197,15 +198,9 @@ export async function getBackfillStatus( }>() ]); - const tasks = counts(taskRow); + const taskCounts = counts(taskRow); + const accounts = counts(accountRow); const discovery = counts(discoveryRow); - const accountCounts = counts(accountRow); - const accounts: BackfillAccountCounts = { - total: accountCounts.total, - complete: accountCounts.complete, - pending: accountCounts.pending, - unreachable: accountCounts.failed - }; const collections = (collectionRows.results ?? []).map((row) => ({ collection: row.collection, ...counts(row) @@ -226,6 +221,7 @@ export async function getBackfillStatus( next_retry_at: nextRetryAt, next_retry_date: nextRetryAt ? new Date(nextRetryAt).toISOString() : null }; + const expectsDiscovery = config ? getDiscoverableNsids(config).length > 0 && (config.relays ?? DEFAULT_RELAYS).length > 0 @@ -236,23 +232,24 @@ export async function getBackfillStatus( number(runRow.heartbeat_at) >= now - BACKFILL_RUN_STALE_MS; const state = running ? "running" - : tasks.total === 0 && discovery.total === 0 + : taskCounts.total === 0 && discovery.total === 0 ? expectsDiscovery ? "not_started" : "complete" - : discovery.pending > 0 + : discovery.complete < discovery.total || taskCounts.pending > 0 ? "incomplete" - : tasks.pending === 0 || tasks.failed === tasks.pending - ? "complete" - : "incomplete"; + : "complete"; const knownProgressPercent = - tasks.total === 0 ? (state === "complete" ? 100 : 0) : Math.floor((tasks.complete / tasks.total) * 10_000) / 100; + accounts.total === 0 + ? state === "complete" + ? 100 + : 0 + : Math.floor((accounts.complete / accounts.total) * 10_000) / 100; return { state, known_progress_percent: knownProgressPercent, accounts, - tasks, discovery, retries, collections diff --git a/packages/contrail/tests/backfill-status.test.ts b/packages/contrail/tests/backfill-status.test.ts index 7657d07..7e29669 100644 --- a/packages/contrail/tests/backfill-status.test.ts +++ b/packages/contrail/tests/backfill-status.test.ts @@ -99,8 +99,9 @@ describe("backfill failure state", () => { expect((await getBackfillStatus(db)).accounts).toEqual({ total: 1, complete: 0, - pending: 1, - unreachable: 1 + pending: 0, + retrying: 1, + failed: 0 }); fetchSpy.mockResolvedValue( @@ -120,7 +121,8 @@ describe("backfill failure state", () => { total: 1, complete: 1, pending: 0, - unreachable: 0 + retrying: 0, + failed: 0 }); }); }); @@ -186,6 +188,89 @@ describe("scheduled backfill retries", () => { }); }); + it("caps backoff at 48 hours and stops after ten failures", async () => { + const db = await createTestDbWithSchema(); + await db + .prepare("INSERT INTO identities (did, handle, pds, resolved_at) VALUES (?, ?, ?, ?)") + .bind(DID, "backfill.test", "https://pds.test", Date.now()) + .run(); + await db + .prepare( + "INSERT INTO backfills (did, collection, completed, retries, last_error, next_retry_at) VALUES (?, ?, 0, 8, 'unavailable', ?)" + ) + .bind(DID, EVENT, Date.now() - 1) + .run(); + fetchSpy.mockResolvedValue( + new Response( + JSON.stringify({ error: "Unavailable", message: "try later" }), + { + status: 503, + headers: { "content-type": "application/json" } + } + ) + ); + + const before = Date.now(); + await retryPendingBackfills(db, TEST_CONFIG, { + maxAccounts: 1, + maxAttempts: 10 + }); + let row = await db + .prepare("SELECT retries, next_retry_at, retry_exhausted FROM backfills WHERE did = ?") + .bind(DID) + .first<{ + retries: number; + next_retry_at: number | null; + retry_exhausted: number; + }>(); + expect(row?.retries).toBe(9); + expect(row?.retry_exhausted).toBe(0); + expect(row?.next_retry_at).toBeGreaterThanOrEqual( + before + 48 * 60 * 60_000 + ); + expect(row?.next_retry_at).toBeLessThanOrEqual( + Date.now() + 48 * 60 * 60_000 + ); + + await db + .prepare("UPDATE backfills SET next_retry_at = ? WHERE did = ?") + .bind(Date.now() - 1, DID) + .run(); + await retryPendingBackfills(db, TEST_CONFIG, { + maxAccounts: 1, + maxAttempts: 10 + }); + row = await db + .prepare("SELECT retries, next_retry_at, retry_exhausted FROM backfills WHERE did = ?") + .bind(DID) + .first<{ + retries: number; + next_retry_at: number | null; + retry_exhausted: number; + }>(); + expect(row).toEqual({ + retries: 10, + next_retry_at: null, + retry_exhausted: 1 + }); + expect((await getBackfillStatus(db, TEST_CONFIG)).accounts).toEqual({ + total: 1, + complete: 0, + pending: 0, + retrying: 0, + failed: 1 + }); + + fetchSpy.mockResolvedValue( + new Response(JSON.stringify({ records: [] }), { + status: 200, + headers: { "content-type": "application/json" } + }) + ); + await backfillPending(db, TEST_CONFIG, { maxAttempts: 1 }); + expect((await getBackfillStatus(db, TEST_CONFIG)).accounts.complete).toBe(1); + }); + it("reports an active run and prevents overlapping retry work", async () => { const db = await createTestDbWithSchema(); const runId = await tryStartBackfillRun(db); @@ -282,7 +367,13 @@ describe("backfill status JSON", () => { const status = await getBackfillStatus(db, TEST_CONFIG); expect(status.state).toBe("complete"); - expect(status.accounts.unreachable).toBe(1); + expect(status.accounts).toEqual({ + total: 1, + complete: 0, + pending: 0, + retrying: 1, + failed: 0 + }); expect(status.retries).toEqual({ scheduled_accounts: 1, due_accounts: 0, @@ -291,7 +382,7 @@ describe("backfill status JSON", () => { }); }); - it("reports known work, unreachable accounts, discovery, records, and cursor", async () => { + it("reports mutually exclusive account and collection states", async () => { const db = await createTestDbWithSchema(); await ingestRecords(db, [makeEvent()]); await db @@ -328,15 +419,21 @@ describe("backfill status JSON", () => { expect(overview.ingestion.cursor).toBeTypeOf("number"); expect(overview.backfill).toMatchObject({ state: "incomplete", - known_progress_percent: 40, + known_progress_percent: 33.33, accounts: { total: 3, complete: 1, - pending: 2, - unreachable: 1 + pending: 1, + retrying: 1, + failed: 0 + }, + discovery: { + total: 2, + complete: 1, + pending: 0, + retrying: 1, + failed: 0 }, - tasks: { total: 5, complete: 2, pending: 3, failed: 1 }, - discovery: { total: 2, complete: 1, pending: 1, failed: 1 }, retries: { scheduled_accounts: 1, due_accounts: 1, @@ -351,14 +448,16 @@ describe("backfill status JSON", () => { collection: EVENT, total: 3, complete: 1, - pending: 2, - failed: 1 + pending: 1, + retrying: 1, + failed: 0 }, { collection: RSVP, total: 2, complete: 1, pending: 1, + retrying: 0, failed: 0 } ]); diff --git a/packages/contrail/tests/schema.test.ts b/packages/contrail/tests/schema.test.ts index 48a03ea..b513e02 100644 --- a/packages/contrail/tests/schema.test.ts +++ b/packages/contrail/tests/schema.test.ts @@ -24,7 +24,11 @@ describe("initSchema", () => { .prepare("PRAGMA table_info(backfills)") .all<{ name: string }>(); expect(backfillColumns.results.map((column) => column.name)).toEqual( - expect.arrayContaining(["last_attempt_at", "next_retry_at"]) + expect.arrayContaining([ + "last_attempt_at", + "next_retry_at", + "retry_exhausted", + ]) ); const discoveryColumns = await db @@ -87,7 +91,11 @@ describe("initSchema", () => { .prepare("PRAGMA table_info(backfills)") .all<{ name: string }>(); expect(backfillColumns.results.map((column) => column.name)).toEqual( - expect.arrayContaining(["last_attempt_at", "next_retry_at"]) + expect.arrayContaining([ + "last_attempt_at", + "next_retry_at", + "retry_exhausted", + ]) ); const discoveryColumns = await db -- 2.51.2