diff --git a/apps/chain-indexer/README.md b/apps/chain-indexer/README.md index a98f36f8f3..14750241c0 100644 --- a/apps/chain-indexer/README.md +++ b/apps/chain-indexer/README.md @@ -9,7 +9,9 @@ INDEXER_ROLE = sync | backfill | api | jobs NETWORK = mainnet | sandbox | testnet ``` -Currently implemented: `sync` (live tail with per-block atomic commits, advisory-lock leader election, parent-hash continuity check) and a minimal `api` (healthz + status). `backfill` and `jobs` exit with `ROLE_NOT_IMPLEMENTED`. +Currently implemented: `sync` (live tail with per-block atomic commits and a parent-hash continuity check), `backfill` (historical catch-up over an explicit height range), and a minimal `api` (healthz + status). `jobs` exits with `ROLE_NOT_IMPLEMENTED`. + +Writers do not use leader election. Inserts are natural-keyed and conflict-ignoring and the `indexer_state` checkpoint only moves forward (`GREATEST` upsert), so overlapping writers on the same stream (e.g. two pods during a rolling deploy) duplicate work but cannot corrupt data or regress the checkpoint. Run one replica per writer role (`replicas: 1` for sync, `parallelism: 1` for backfill Jobs) to avoid the wasted work. ## Running locally @@ -34,6 +36,18 @@ curl localhost:3092/v1/status The checkpoint height should advance as blocks land in `cosmos.blocks`, `cosmos.transactions`, and `cosmos.messages`. +## Backfill + +The backfill role fills the database over an explicit, inclusive height range and exits when done, so it fits a one-off K8s Job: + +``` +INDEXER_ROLE=backfill +BACKFILL_FROM_HEIGHT=100000 +BACKFILL_TO_HEIGHT=200000 +``` + +Blocks are fetched from RPC in parallel (`BACKFILL_CONCURRENCY`, default 10) and committed strictly in order in batches of `BACKFILL_BATCH_SIZE` blocks (default 200), each batch in one Postgres transaction together with the checkpoint advance. Progress is checkpointed per range under the `indexer_state` stream `backfill:{from}-{to}`, so killing and restarting the job resumes at the checkpoint without gaps or duplicates, and re-running a completed range exits 0 immediately. Changing the range creates a fresh checkpoint row. All inserts are natural-keyed and conflict-ignoring, so a backfill can run against the same database as live sync, and a duplicate backfill pod on the same range is harmless. + ## Tests ```bash diff --git a/apps/chain-indexer/env/.env.sample b/apps/chain-indexer/env/.env.sample index ec4a54572a..78363d5847 100644 --- a/apps/chain-indexer/env/.env.sample +++ b/apps/chain-indexer/env/.env.sample @@ -4,6 +4,10 @@ POSTGRES_DB_URI=postgres://user:password@localhost:5432/chain-indexer RPC_NODE_ENDPOINTS= SYNC_START_HEIGHT= SYNC_POLL_INTERVAL_MS=3000 +BACKFILL_FROM_HEIGHT= +BACKFILL_TO_HEIGHT= +BACKFILL_CONCURRENCY=10 +BACKFILL_BATCH_SIZE=200 RPC_TIMEOUT_MS=15000 MESSAGE_BODY_MAX_BYTES=65536 PORT=3092 diff --git a/apps/chain-indexer/src/config/env.config.spec.ts b/apps/chain-indexer/src/config/env.config.spec.ts index 5bcf5efab2..cbafcbfc98 100644 --- a/apps/chain-indexer/src/config/env.config.spec.ts +++ b/apps/chain-indexer/src/config/env.config.spec.ts @@ -31,6 +31,34 @@ describe("envSchema", () => { expect(() => setup({ PORT: "3092.5" })).toThrow(); }); + describe("when INDEXER_ROLE is backfill", () => { + it("requires both backfill heights", () => { + expect(() => setup({ INDEXER_ROLE: "backfill" })).toThrow(/BACKFILL_FROM_HEIGHT[\s\S]*BACKFILL_TO_HEIGHT/); + }); + + it("rejects a range where from is above to", () => { + expect(() => setup({ INDEXER_ROLE: "backfill", BACKFILL_FROM_HEIGHT: "100", BACKFILL_TO_HEIGHT: "50" })).toThrow( + "BACKFILL_FROM_HEIGHT must be <= BACKFILL_TO_HEIGHT" + ); + }); + + it("parses a valid range with concurrency and batch size defaults", () => { + const config = setup({ INDEXER_ROLE: "backfill", BACKFILL_FROM_HEIGHT: "100", BACKFILL_TO_HEIGHT: "200" }); + + expect(config.BACKFILL_FROM_HEIGHT).toBe(100); + expect(config.BACKFILL_TO_HEIGHT).toBe(200); + expect(config.BACKFILL_CONCURRENCY).toBe(10); + expect(config.BACKFILL_BATCH_SIZE).toBe(200); + }); + }); + + it("does not require backfill heights for other roles", () => { + const config = setup({ INDEXER_ROLE: "api", BACKFILL_FROM_HEIGHT: "", BACKFILL_TO_HEIGHT: "" }); + + expect(config.BACKFILL_FROM_HEIGHT).toBeUndefined(); + expect(config.BACKFILL_TO_HEIGHT).toBeUndefined(); + }); + function setup(overrides?: Record) { return envSchema.parse({ POSTGRES_DB_URI: "postgres://unit:unit@localhost:5432/unit", ...overrides }); } diff --git a/apps/chain-indexer/src/config/env.config.ts b/apps/chain-indexer/src/config/env.config.ts index eb2ec6e76d..1c00840be0 100644 --- a/apps/chain-indexer/src/config/env.config.ts +++ b/apps/chain-indexer/src/config/env.config.ts @@ -3,7 +3,7 @@ import { z } from "zod"; /** Treats an empty string as absent so `VAR=` lines in env files don't fail coerced-number validation. */ const emptyStringAsUndefined = (value: unknown) => (value === "" ? undefined : value); -export const envSchema = z.object({ +const rawEnvSchema = z.object({ INDEXER_ROLE: z.enum(["sync", "backfill", "api", "jobs"]).default("sync"), NETWORK: z.enum(["mainnet", "sandbox", "testnet"]).default("sandbox"), POSTGRES_DB_URI: z.string(), @@ -14,6 +14,14 @@ export const envSchema = z.object({ /** First height to sync when the database has no checkpoint yet. Defaults to the current chain tip. */ SYNC_START_HEIGHT: z.preprocess(emptyStringAsUndefined, z.number({ coerce: true }).int().positive().optional()), SYNC_POLL_INTERVAL_MS: z.number({ coerce: true }).int().positive().default(3_000), + /** First height of the backfill range (inclusive). Required when INDEXER_ROLE is "backfill". */ + BACKFILL_FROM_HEIGHT: z.preprocess(emptyStringAsUndefined, z.number({ coerce: true }).int().positive().optional()), + /** Last height of the backfill range (inclusive). Required when INDEXER_ROLE is "backfill". */ + BACKFILL_TO_HEIGHT: z.preprocess(emptyStringAsUndefined, z.number({ coerce: true }).int().positive().optional()), + /** How many blocks the backfill fetches from RPC in parallel. */ + BACKFILL_CONCURRENCY: z.number({ coerce: true }).int().min(1).max(64).default(10), + /** How many blocks the backfill commits per Postgres transaction. */ + BACKFILL_BATCH_SIZE: z.number({ coerce: true }).int().min(1).max(1_000).default(200), /** Decoded message bodies above this serialized size are stored as null to keep pathological messages out of Postgres. */ MESSAGE_BODY_MAX_BYTES: z.number({ coerce: true }).int().positive().default(65_536), DRIZZLE_MIGRATIONS_FOLDER: z.string().default("./drizzle"), @@ -23,4 +31,20 @@ export const envSchema = z.object({ PORT: z.number({ coerce: true }).int().min(1).max(65_535).optional().default(3092) }); +export const envSchema = rawEnvSchema.superRefine((env, ctx) => { + if (env.INDEXER_ROLE !== "backfill") { + return; + } + + (["BACKFILL_FROM_HEIGHT", "BACKFILL_TO_HEIGHT"] as const).forEach(key => { + if (env[key] === undefined) { + ctx.addIssue({ code: z.ZodIssueCode.custom, path: [key], message: 'Required when INDEXER_ROLE is "backfill"' }); + } + }); + + if (env.BACKFILL_FROM_HEIGHT !== undefined && env.BACKFILL_TO_HEIGHT !== undefined && env.BACKFILL_FROM_HEIGHT > env.BACKFILL_TO_HEIGHT) { + ctx.addIssue({ code: z.ZodIssueCode.custom, path: ["BACKFILL_FROM_HEIGHT"], message: "BACKFILL_FROM_HEIGHT must be <= BACKFILL_TO_HEIGHT" }); + } +}); + export type EnvConfig = z.infer; diff --git a/apps/chain-indexer/src/index.ts b/apps/chain-indexer/src/index.ts index 3944fc6b74..db288b48b7 100644 --- a/apps/chain-indexer/src/index.ts +++ b/apps/chain-indexer/src/index.ts @@ -1,10 +1,14 @@ import "reflect-metadata"; import "@src/providers"; +import type { LoggerService } from "@akashnetwork/logging"; import { createOtelLogger } from "@akashnetwork/logging/otel"; import { container } from "tsyringe"; import { createApp } from "@src/app"; +import { envSchema } from "@src/config/env.config"; +import { BackfillRunnerService } from "@src/pipeline/backfill-runner.service"; +import { RunnerInterruptedError } from "@src/pipeline/runner-interrupted-error"; import { SyncRunnerService } from "@src/pipeline/sync-runner.service"; import { migrateDb } from "@src/providers/db.provider"; import { AppConfigService } from "@src/services/app-config/app-config.service"; @@ -12,28 +16,29 @@ import { shutdownServer } from "@src/services/shutdown-server/shutdown-server"; import { startServer } from "@src/services/start-server/start-server"; export async function bootstrap(): Promise { + const logger = createOtelLogger({ context: "APP" }); + + if (!validateConfig(logger)) { + process.exitCode = 1; + return; + } + const config = container.resolve(AppConfigService); const role = config.get("INDEXER_ROLE"); - const logger = createOtelLogger({ context: "APP" }); + const port = config.get("PORT"); switch (role) { case "sync": { - await migrateDb(); - const server = await startServer(createApp(), logger, process, { port: config.get("PORT") }); - - try { - await container.resolve(SyncRunnerService).start(); - } catch (error) { - logger.error({ event: "SYNC_FATAL", error }); - process.exitCode = 1; - } - - await shutdownServer(server, logger); + await runRunnerBehindServer(() => container.resolve(SyncRunnerService), "SYNC_FATAL", logger, port); + return; + } + case "backfill": { + await runRunnerBehindServer(() => container.resolve(BackfillRunnerService), "BACKFILL_FATAL", logger, port); return; } case "api": { await migrateDb(); - await startServer(createApp(), logger, process, { port: config.get("PORT") }); + await startServer(createApp(), logger, process, { port }); return; } default: { @@ -42,3 +47,42 @@ export async function bootstrap(): Promise { } } } + +/** Validates env eagerly so a misconfigured role (e.g. a backfill Job missing BACKFILL_FROM/TO_HEIGHT) fails with the actual field errors instead of a tsyringe dependency-injection wrapper around the ZodError. */ +function validateConfig(logger: LoggerService): boolean { + const result = envSchema.safeParse(process.env); + + if (result.success) { + return true; + } + + logger.error({ + event: "CONFIG_INVALID", + issues: result.error.issues.map(issue => ({ path: issue.path.join(".") || "(root)", message: issue.message })) + }); + return false; +} + +/** + * Shared runner-role lifecycle: migrate, serve healthz, run to completion, then shut the server + * down so the process can exit. A fatal error exits non-zero; a run stopped before finishing + * (`RunnerInterruptedError`, e.g. SIGTERM mid-backfill) also exits non-zero so a K8s Job is retried + * and resumes from its checkpoint rather than being marked Complete with the range unfinished. + */ +async function runRunnerBehindServer(resolveRunner: () => { start(): Promise }, fatalEvent: string, logger: LoggerService, port: number): Promise { + await migrateDb(); + const server = await startServer(createApp(), logger, process, { port }); + + try { + await resolveRunner().start(); + } catch (error) { + if (error instanceof RunnerInterruptedError) { + logger.warn({ event: "RUNNER_INTERRUPTED", error }); + } else { + logger.error({ event: fatalEvent, error }); + } + process.exitCode = 1; + } + + await shutdownServer(server, logger); +} diff --git a/apps/chain-indexer/src/lib/retry-with-backoff/retry-with-backoff.spec.ts b/apps/chain-indexer/src/lib/retry-with-backoff/retry-with-backoff.spec.ts new file mode 100644 index 0000000000..6b4589d319 --- /dev/null +++ b/apps/chain-indexer/src/lib/retry-with-backoff/retry-with-backoff.spec.ts @@ -0,0 +1,38 @@ +import { describe, expect, it, vi } from "vitest"; + +import { retryWithBackoff } from "@src/lib/retry-with-backoff/retry-with-backoff"; + +describe(retryWithBackoff.name, () => { + it("returns the result after a transient failure", async () => { + const operation = vi.fn().mockRejectedValueOnce(new Error("transient")).mockResolvedValue("ok"); + const onRetry = vi.fn(); + + const result = await retryWithBackoff(operation, { maxAttempts: 3, baseDelayMs: 1, onRetry }); + + expect(result).toBe("ok"); + expect(onRetry).toHaveBeenCalledWith(expect.any(Error), 1, 1); + }); + + it("rethrows once the attempts are exhausted", async () => { + const operation = vi.fn().mockRejectedValue(new Error("persistent")); + + await expect(retryWithBackoff(operation, { maxAttempts: 3, baseDelayMs: 1, onRetry: vi.fn() })).rejects.toThrow("persistent"); + expect(operation).toHaveBeenCalledTimes(3); + }); + + it("rethrows immediately when shouldRethrow matches", async () => { + const operation = vi.fn().mockRejectedValue(new Error("fatal")); + + await expect(retryWithBackoff(operation, { maxAttempts: 3, baseDelayMs: 1, shouldRethrow: () => true, onRetry: vi.fn() })).rejects.toThrow("fatal"); + expect(operation).toHaveBeenCalledTimes(1); + }); + + it("caps the backoff delay at maxDelayMs", async () => { + const operation = vi.fn().mockRejectedValueOnce(new Error("a")).mockRejectedValueOnce(new Error("b")).mockResolvedValue("ok"); + const onRetry = vi.fn(); + + await retryWithBackoff(operation, { maxAttempts: 5, baseDelayMs: 2, maxDelayMs: 3, onRetry }); + + expect(onRetry.mock.calls.map(call => call[2])).toEqual([2, 3]); + }); +}); diff --git a/apps/chain-indexer/src/lib/retry-with-backoff/retry-with-backoff.ts b/apps/chain-indexer/src/lib/retry-with-backoff/retry-with-backoff.ts new file mode 100644 index 0000000000..3c209d9432 --- /dev/null +++ b/apps/chain-indexer/src/lib/retry-with-backoff/retry-with-backoff.ts @@ -0,0 +1,35 @@ +export interface RetryWithBackoffOptions { + maxAttempts: number; + baseDelayMs: number; + maxDelayMs?: number; + /** Errors for which a retry cannot help (fatal conditions, shutdown in progress); they propagate immediately. */ + shouldRethrow?: (error: unknown) => boolean; + onRetry: (error: unknown, attempt: number, delayMs: number) => void; +} + +export async function retryWithBackoff(operation: () => Promise, options: RetryWithBackoffOptions): Promise { + let attempt = 0; + + while (true) { + attempt++; + try { + return await operation(); + } catch (error) { + if (options.shouldRethrow?.(error) || attempt >= options.maxAttempts) { + throw error; + } + + const uncappedDelayMs = options.baseDelayMs * 2 ** (attempt - 1); + const delayMs = options.maxDelayMs === undefined ? uncappedDelayMs : Math.min(uncappedDelayMs, options.maxDelayMs); + options.onRetry(error, attempt, delayMs); + await delay(delayMs); + } + } +} + +/** Global setTimeout rather than node:timers/promises so tests can advance the backoff with vitest fake timers, which do not intercept node:timers/promises. */ +function delay(ms: number): Promise { + return new Promise(resolve => { + setTimeout(resolve, ms); + }); +} diff --git a/apps/chain-indexer/src/pipeline/backfill-planner.spec.ts b/apps/chain-indexer/src/pipeline/backfill-planner.spec.ts new file mode 100644 index 0000000000..68670e4045 --- /dev/null +++ b/apps/chain-indexer/src/pipeline/backfill-planner.spec.ts @@ -0,0 +1,41 @@ +import { describe, expect, it } from "vitest"; + +import { planBackfill } from "@src/pipeline/backfill-planner"; + +describe(planBackfill.name, () => { + it("runs the full range when there is no checkpoint", () => { + const plan = planBackfill({ fromHeight: 100, toHeight: 200, checkpointHeight: null, tipHeight: 1_000 }); + + expect(plan).toEqual({ kind: "run", startHeight: 100, endHeight: 200 }); + }); + + it("resumes after the checkpoint when one exists mid-range", () => { + const plan = planBackfill({ fromHeight: 100, toHeight: 200, checkpointHeight: 150, tipHeight: 1_000 }); + + expect(plan).toEqual({ kind: "run", startHeight: 151, endHeight: 200 }); + }); + + it("reports already-complete when the checkpoint reached the end of the range", () => { + const plan = planBackfill({ fromHeight: 100, toHeight: 200, checkpointHeight: 200, tipHeight: 1_000 }); + + expect(plan).toEqual({ kind: "already-complete" }); + }); + + it("reports already-complete even when a lagging node returns a stale tip below the range end", () => { + const plan = planBackfill({ fromHeight: 100, toHeight: 200, checkpointHeight: 200, tipHeight: 150 }); + + expect(plan).toEqual({ kind: "already-complete" }); + }); + + it("rejects a range ending above the chain tip", () => { + const plan = planBackfill({ fromHeight: 100, toHeight: 2_000, checkpointHeight: null, tipHeight: 1_000 }); + + expect(plan).toEqual({ kind: "invalid", reason: "BACKFILL_TO_HEIGHT 2000 is above the chain tip 1000" }); + }); + + it("runs a single-block range", () => { + const plan = planBackfill({ fromHeight: 100, toHeight: 100, checkpointHeight: null, tipHeight: 1_000 }); + + expect(plan).toEqual({ kind: "run", startHeight: 100, endHeight: 100 }); + }); +}); diff --git a/apps/chain-indexer/src/pipeline/backfill-planner.ts b/apps/chain-indexer/src/pipeline/backfill-planner.ts new file mode 100644 index 0000000000..739265f1a0 --- /dev/null +++ b/apps/chain-indexer/src/pipeline/backfill-planner.ts @@ -0,0 +1,29 @@ +export interface BackfillPlanInput { + fromHeight: number; + toHeight: number; + checkpointHeight: number | null; + tipHeight: number; +} + +export type BackfillPlan = { kind: "run"; startHeight: number; endHeight: number } | { kind: "already-complete" } | { kind: "invalid"; reason: string }; + +/** + * A range above the chain tip is rejected rather than clamped: clamping would mark the range's + * checkpoint complete for heights that were never indexed. Completion is checked first so a + * re-run of a finished range stays a no-op even when a lagging RPC node reports a stale tip. + */ +export function planBackfill(input: BackfillPlanInput): BackfillPlan { + if (input.checkpointHeight !== null && input.checkpointHeight >= input.toHeight) { + return { kind: "already-complete" }; + } + + if (input.toHeight > input.tipHeight) { + return { kind: "invalid", reason: `BACKFILL_TO_HEIGHT ${input.toHeight} is above the chain tip ${input.tipHeight}` }; + } + + return { + kind: "run", + startHeight: input.checkpointHeight !== null ? input.checkpointHeight + 1 : input.fromHeight, + endHeight: input.toHeight + }; +} diff --git a/apps/chain-indexer/src/pipeline/backfill-runner.service.spec.ts b/apps/chain-indexer/src/pipeline/backfill-runner.service.spec.ts new file mode 100644 index 0000000000..9d81537c25 --- /dev/null +++ b/apps/chain-indexer/src/pipeline/backfill-runner.service.spec.ts @@ -0,0 +1,254 @@ +import { setTimeout as delay } from "node:timers/promises"; +import { describe, expect, it, vi } from "vitest"; +import { mock } from "vitest-mock-extended"; + +import { envSchema } from "@src/config/env.config"; +import { Blocks, IndexerState } from "@src/db/schema"; +import { BackfillRunnerService } from "@src/pipeline/backfill-runner.service"; +import type { BlockCommitterService } from "@src/pipeline/block-committer.service"; +import type { BlockDecoderService } from "@src/pipeline/block-decoder.service"; +import type { DecodedBlock } from "@src/pipeline/decoded-block"; +import { RunnerInterruptedError } from "@src/pipeline/runner-interrupted-error"; +import type { ChainDatabase } from "@src/providers/db.provider"; +import type { LoggerService } from "@src/providers/logging.provider"; +import type { RpcClientPool } from "@src/rpc/rpc-client-pool.service"; +import type { RpcBlockResult } from "@src/rpc/rpc-types"; + +describe(BackfillRunnerService.name, () => { + it("commits blocks in ascending order even when fetches resolve out of order", async () => { + const { runner, committer } = setup({ + fromHeight: 1, + toHeight: 5, + batchSize: 5, + concurrency: 5, + fetchDelayMs: height => (6 - height) * 5 + }); + + await runner.start(); + + expect(committer.commitBatch).toHaveBeenCalledTimes(1); + expect(committedHeights(committer)).toEqual([[1, 2, 3, 4, 5]]); + }); + + it("never fetches more blocks in parallel than the configured concurrency", async () => { + const { runner, maxObservedConcurrency } = setup({ fromHeight: 1, toHeight: 10, batchSize: 10, concurrency: 3, fetchDelayMs: () => 2 }); + + await runner.start(); + + expect(maxObservedConcurrency()).toBeLessThanOrEqual(3); + }); + + it("commits in batches of the configured size under the range-scoped stream", async () => { + const { runner, committer } = setup({ fromHeight: 1, toHeight: 5, batchSize: 2 }); + + await runner.start(); + + expect(committedHeights(committer)).toEqual([[1, 2], [3, 4], [5]]); + expect(committer.commitBatch.mock.calls.map(call => call[1])).toEqual([{ stream: "backfill:1-5" }, { stream: "backfill:1-5" }, { stream: "backfill:1-5" }]); + }); + + it("resumes after the checkpoint and verifies continuity against the checkpoint block", async () => { + const { runner, committer, pool } = setup({ + fromHeight: 1, + toHeight: 5, + checkpointHeight: 3, + seedBlock: { height: 3, hash: heightHash(3) } + }); + + await runner.start(); + + expect(pool.getBlock).not.toHaveBeenCalledWith(3); + expect(committedHeights(committer)).toEqual([[4, 5]]); + }); + + it("throws when the checkpoint block is missing on resume", async () => { + const { runner, committer } = setup({ fromHeight: 1, toHeight: 5, checkpointHeight: 3 }); + + await expect(runner.start()).rejects.toThrow("Checkpoint block 3 is missing"); + expect(committer.commitBatch).not.toHaveBeenCalled(); + }); + + it("exits without fetching anything when the checkpoint already covers the range", async () => { + const { runner, committer, pool, logger } = setup({ fromHeight: 1, toHeight: 5, checkpointHeight: 5 }); + + await runner.start(); + + expect(pool.getBlock).not.toHaveBeenCalled(); + expect(committer.commitBatch).not.toHaveBeenCalled(); + expect(logger.info).toHaveBeenCalledWith(expect.objectContaining({ event: "BACKFILL_ALREADY_COMPLETE" })); + }); + + it("fails when the range ends above the chain tip", async () => { + const { runner, committer } = setup({ fromHeight: 1, toHeight: 5, tipHeight: 3 }); + + await expect(runner.start()).rejects.toThrow("BACKFILL_TO_HEIGHT 5 is above the chain tip 3"); + expect(committer.commitBatch).not.toHaveBeenCalled(); + }); + + it("halts without committing when the parent-hash chain breaks", async () => { + const { runner, committer } = setup({ fromHeight: 1, toHeight: 5, brokenParentAtHeight: 3 }); + + await expect(runner.start()).rejects.toThrow("Parent hash mismatch at height 3; halting backfill"); + expect(committer.commitBatch).not.toHaveBeenCalled(); + }); + + it("retries a failed fetch and still commits the block", async () => { + vi.useFakeTimers(); + + try { + const { runner, committer, logger } = setup({ fromHeight: 1, toHeight: 2, failFetchOnceAtHeight: 2 }); + + const started = runner.start(); + await vi.runAllTimersAsync(); + await started; + + expect(committedHeights(committer)).toEqual([[1, 2]]); + expect(logger.warn).toHaveBeenCalledWith(expect.objectContaining({ event: "BACKFILL_FETCH_RETRY", height: 2, attempt: 1 })); + } finally { + vi.useRealTimers(); + } + }); + + it("logs a completion summary with throughput counters", async () => { + const { runner, logger } = setup({ fromHeight: 1, toHeight: 5, txCountPerBlock: 2 }); + + await runner.start(); + + expect(logger.info).toHaveBeenCalledWith( + expect.objectContaining({ + event: "BACKFILL_COMPLETED", + stream: "backfill:1-5", + startHeight: 1, + endHeight: 5, + blocksCommitted: 5, + transactionsCommitted: 10, + durationMs: expect.any(Number), + blocksPerSecond: expect.any(Number) + }) + ); + }); + + it("rejects with RunnerInterruptedError when stopped before the range completes", async () => { + const { runner, committer } = setup({ fromHeight: 1, toHeight: 10, batchSize: 2, concurrency: 2 }); + committer.commitBatch.mockImplementationOnce(async () => { + await runner.dispose(); + }); + + await expect(runner.start()).rejects.toThrow(RunnerInterruptedError); + expect(committedHeights(committer)).toEqual([[1, 2]]); + }); + + it("reports completion when stopped during the final commit that covers the range", async () => { + const { runner, committer, logger } = setup({ fromHeight: 1, toHeight: 5, batchSize: 5, concurrency: 5 }); + committer.commitBatch.mockImplementationOnce(async () => { + await runner.dispose(); + }); + + await expect(runner.start()).resolves.toBeUndefined(); + expect(committedHeights(committer)).toEqual([[1, 2, 3, 4, 5]]); + expect(logger.info).toHaveBeenCalledWith(expect.objectContaining({ event: "BACKFILL_COMPLETED" })); + }); + + function setup(input: { + fromHeight: number; + toHeight: number; + batchSize?: number; + concurrency?: number; + tipHeight?: number; + checkpointHeight?: number; + seedBlock?: { height: number; hash: Buffer }; + fetchDelayMs?: (height: number) => number; + failFetchOnceAtHeight?: number; + brokenParentAtHeight?: number; + txCountPerBlock?: number; + }) { + const config = envSchema.parse({ + POSTGRES_DB_URI: "postgres://unit:unit@localhost:5432/unit", + INDEXER_ROLE: "backfill", + BACKFILL_FROM_HEIGHT: String(input.fromHeight), + BACKFILL_TO_HEIGHT: String(input.toHeight), + BACKFILL_BATCH_SIZE: String(input.batchSize ?? 200), + BACKFILL_CONCURRENCY: String(input.concurrency ?? 10) + }); + + const dbFake = { + select: () => ({ + from: (table: unknown) => ({ + where: () => { + if (table === IndexerState && input.checkpointHeight !== undefined) { + return Promise.resolve([{ stream: `backfill:${input.fromHeight}-${input.toHeight}`, lastHeight: input.checkpointHeight }]); + } + if (table === Blocks && input.seedBlock) { + return Promise.resolve([input.seedBlock]); + } + return Promise.resolve([]); + } + }) + }) + }; + + let activeFetches = 0; + let maxActiveFetches = 0; + let failedOnce = false; + const pool = mock(); + pool.getTipHeight.mockResolvedValue(input.tipHeight ?? 1_000); + pool.getBlock.mockImplementation(async height => { + if (input.failFetchOnceAtHeight === height && !failedOnce) { + failedOnce = true; + throw new AggregateError([new Error("all nodes failed")], `Failed to fetch block ${height}`); + } + + activeFetches++; + maxActiveFetches = Math.max(maxActiveFetches, activeFetches); + await delay(input.fetchDelayMs?.(height) ?? 0); + activeFetches--; + return { block: { header: { height: String(height) } } } as RpcBlockResult; + }); + pool.getBlockResults.mockResolvedValue({ height: "0", txs_results: null }); + + const decoder = mock(); + decoder.decode.mockImplementation(block => { + const height = parseInt(block.block.header.height); + return buildDecodedBlock(height, { + parentHash: input.brokenParentAtHeight === height ? Buffer.from("bogus") : heightHash(height - 1), + txCount: input.txCountPerBlock ?? 0 + }); + }); + + const committer = mock(); + committer.commitBatch.mockResolvedValue(undefined); + + const logger = mock(); + + const runner = new BackfillRunnerService(dbFake as unknown as ChainDatabase, pool, decoder, committer, config, logger); + + return { runner, committer, pool, logger, maxObservedConcurrency: () => maxActiveFetches }; + } + + function committedHeights(committer: { commitBatch: { mock: { calls: unknown[][] } } }) { + return committer.commitBatch.mock.calls.map(call => (call[0] as DecodedBlock[]).map(block => block.height)); + } + + function buildDecodedBlock(height: number, options: { parentHash: Buffer; txCount: number }): DecodedBlock { + return { + height, + datetime: new Date("2026-08-11T00:00:00Z"), + hash: heightHash(height), + parentHash: options.parentHash, + proposerAddress: "PROPOSER", + transactions: Array.from({ length: options.txCount }, (_, index) => ({ + index, + hash: Buffer.from(`tx-${height}-${index}`), + code: 0, + gasUsed: 0, + gasWanted: 0, + fee: [], + messages: [] + })) + }; + } + + function heightHash(height: number): Buffer { + return Buffer.from(`hash-${height}`); + } +}); diff --git a/apps/chain-indexer/src/pipeline/backfill-runner.service.ts b/apps/chain-indexer/src/pipeline/backfill-runner.service.ts new file mode 100644 index 0000000000..1961726277 --- /dev/null +++ b/apps/chain-indexer/src/pipeline/backfill-runner.service.ts @@ -0,0 +1,241 @@ +import { eq } from "drizzle-orm"; +import { inject, singleton } from "tsyringe"; + +import type { EnvConfig } from "@src/config/env.config"; +import { Blocks, IndexerState } from "@src/db/schema"; +import { retryWithBackoff } from "@src/lib/retry-with-backoff/retry-with-backoff"; +import { planBackfill } from "@src/pipeline/backfill-planner"; +import { BlockCommitterService } from "@src/pipeline/block-committer.service"; +import { BlockDecoderService } from "@src/pipeline/block-decoder.service"; +import { ChainContinuityError } from "@src/pipeline/chain-continuity-error"; +import type { DecodedBlock } from "@src/pipeline/decoded-block"; +import { RunnerInterruptedError } from "@src/pipeline/runner-interrupted-error"; +import { retryTransient } from "@src/pipeline/transient-retry"; +import { APP_CONFIG } from "@src/providers/app-config.provider"; +import type { ChainDatabase } from "@src/providers/db.provider"; +import { CHAIN_DB } from "@src/providers/db.provider"; +import { LoggerService } from "@src/providers/logging.provider"; +import { RpcClientPool } from "@src/rpc/rpc-client-pool.service"; + +const FETCH_RETRY_MAX_ATTEMPTS = 5; +const FETCH_RETRY_BASE_MS = 1_000; + +@singleton() +export class BackfillRunnerService { + readonly #db: ChainDatabase; + readonly #pool: RpcClientPool; + readonly #decoder: BlockDecoderService; + readonly #committer: BlockCommitterService; + readonly #config: EnvConfig; + readonly #logger: LoggerService; + + #stopped = false; + #lastHash: Buffer | null = null; + + constructor( + @inject(CHAIN_DB) db: ChainDatabase, + @inject(RpcClientPool) pool: RpcClientPool, + @inject(BlockDecoderService) decoder: BlockDecoderService, + @inject(BlockCommitterService) committer: BlockCommitterService, + @inject(APP_CONFIG) config: EnvConfig, + @inject(LoggerService) logger: LoggerService + ) { + this.#db = db; + this.#pool = pool; + this.#decoder = decoder; + this.#committer = committer; + this.#config = config; + this.#logger = logger; + this.#logger.setContext("BACKFILL"); + } + + async start(): Promise { + let completed: boolean; + + try { + completed = await this.#run(); + } catch (error) { + if (this.#stopped) { + throw new RunnerInterruptedError("Backfill stopped before completing the range", { cause: error }); + } + throw error; + } + + if (!completed) { + throw new RunnerInterruptedError("Backfill stopped before completing the range"); + } + } + + async dispose(): Promise { + this.#stopped = true; + } + + async #run(): Promise { + const { BACKFILL_FROM_HEIGHT: fromHeight, BACKFILL_TO_HEIGHT: toHeight } = this.#config; + + if (fromHeight === undefined || toHeight === undefined) { + throw new Error("BACKFILL_FROM_HEIGHT and BACKFILL_TO_HEIGHT are required for the backfill role"); + } + + const stream = `backfill:${fromHeight}-${toHeight}`; + const [checkpointHeight, tipHeight] = await Promise.all([ + this.#retryTransient(() => this.#getCheckpointHeight(stream), { event: "BACKFILL_CHECKPOINT_READ_RETRY" }), + this.#retryTransient(() => this.#pool.getTipHeight(), { event: "BACKFILL_TIP_FETCH_RETRY" }) + ]); + const plan = planBackfill({ fromHeight, toHeight, checkpointHeight, tipHeight }); + + if (plan.kind === "invalid") { + this.#logger.error({ event: "BACKFILL_INVALID_RANGE", reason: plan.reason }); + throw new Error(plan.reason); + } + + if (plan.kind === "already-complete") { + this.#logger.info({ event: "BACKFILL_ALREADY_COMPLETE", stream, checkpointHeight }); + return true; + } + + await this.#seedContinuityHash(plan.startHeight, checkpointHeight !== null); + this.#logger.info({ event: "BACKFILL_STARTED", network: this.#config.NETWORK, stream, startHeight: plan.startHeight, endHeight: plan.endHeight }); + + return await this.#backfillRange(plan.startHeight, plan.endHeight, stream); + } + + /** + * Fetches up to BACKFILL_CONCURRENCY blocks in parallel while consuming heights strictly in + * order, so batches handed to the committer are contiguous and ordered by construction. + * Prefetched promises get a no-op catch at insertion: a rejection settling before the loop + * reaches its height would otherwise crash the process as an unhandled rejection; the real + * rejection still surfaces when the loop awaits that height. + * + * Returns whether the whole range committed. Completion is tracked by the last committed height + * rather than the stopped flag, so a shutdown landing during the final commit still reports the + * range as done instead of failing the Job for a spurious retry. + */ + async #backfillRange(startHeight: number, endHeight: number, stream: string): Promise { + const startedAt = Date.now(); + const inflight = new Map>(); + let fetchHead = startHeight; + let blocksCommitted = 0; + let transactionsCommitted = 0; + let lastCommittedHeight = startHeight - 1; + let batch: DecodedBlock[] = []; + + const fillFetchWindow = () => { + while (fetchHead <= endHeight && inflight.size < this.#config.BACKFILL_CONCURRENCY) { + const height = fetchHead; + const prefetched = this.#fetchAndDecode(height); + prefetched.catch(() => undefined); + inflight.set(height, prefetched); + fetchHead++; + } + }; + + try { + for (let height = startHeight; height <= endHeight && !this.#stopped; height++) { + fillFetchWindow(); + const decoded = await inflight.get(height)!; + inflight.delete(height); + + this.#verifyContinuity(decoded); + this.#lastHash = decoded.hash; + batch.push(decoded); + fillFetchWindow(); + + if (batch.length >= this.#config.BACKFILL_BATCH_SIZE || height === endHeight) { + const currentBatch = batch; + await this.#retryTransient(() => this.#committer.commitBatch(currentBatch, { stream }), { event: "BACKFILL_COMMIT_RETRY", height }); + blocksCommitted += batch.length; + transactionsCommitted += batch.reduce((sum, block) => sum + block.transactions.length, 0); + lastCommittedHeight = height; + batch = []; + this.#logger.info({ event: "BACKFILL_PROGRESS", height, endHeight, blocksCommitted }); + } + } + } finally { + await Promise.allSettled([...inflight.values()]); + } + + if (lastCommittedHeight < endHeight) { + return false; + } + + const durationMs = Date.now() - startedAt; + this.#logger.info({ + event: "BACKFILL_COMPLETED", + stream, + startHeight, + endHeight, + blocksCommitted, + transactionsCommitted, + durationMs, + blocksPerSecond: durationMs > 0 ? Math.round((blocksCommitted / durationMs) * 1_000 * 100) / 100 : blocksCommitted + }); + + return true; + } + + /** Retriable steps (checkpoint reads, tip fetches, idempotent batch commits) survive transient blips instead of failing the whole multi-hour Job; fatal errors propagate. */ + async #retryTransient(operation: () => Promise, logContext: { event: string; height?: number }): Promise { + return await retryTransient(operation, { isStopped: () => this.#stopped, logger: this.#logger, logContext }); + } + + /** A pool AggregateError means every RPC endpoint already failed once, so retries back off before another full sweep. */ + async #fetchAndDecode(height: number): Promise { + return await retryWithBackoff( + async () => { + const [block, blockResults] = await Promise.all([this.#pool.getBlock(height), this.#pool.getBlockResults(height)]); + return this.#decoder.decode(block, blockResults); + }, + { + maxAttempts: FETCH_RETRY_MAX_ATTEMPTS, + baseDelayMs: FETCH_RETRY_BASE_MS, + shouldRethrow: () => this.#stopped, + onRetry: (error, attempt, delayMs) => this.#logger.warn({ event: "BACKFILL_FETCH_RETRY", height, attempt, delayMs, error }) + } + ); + } + + #verifyContinuity(block: DecodedBlock): void { + if (this.#lastHash && block.parentHash && !block.parentHash.equals(this.#lastHash)) { + this.#logger.error({ + event: "BACKFILL_CONTINUITY_BROKEN", + height: block.height, + expectedParentHash: this.#lastHash.toString("hex"), + actualParentHash: block.parentHash.toString("hex") + }); + throw new ChainContinuityError(`Parent hash mismatch at height ${block.height}; halting backfill`); + } + } + + /** + * The parent-hash chain is seeded from the block before the start height. On resume that block + * was committed by this stream's checkpoint and must exist; on a fresh start it may have been + * committed by sync or another backfill, and its absence just leaves the first block unverified. + */ + async #seedContinuityHash(startHeight: number, isResume: boolean): Promise { + const [previousBlock] = await this.#retryTransient( + () => + this.#db + .select() + .from(Blocks) + .where(eq(Blocks.height, startHeight - 1)), + { event: "BACKFILL_SEED_READ_RETRY", height: startHeight - 1 } + ); + + if (previousBlock) { + this.#lastHash = previousBlock.hash; + return; + } + + if (isResume) { + throw new Error(`Checkpoint block ${startHeight - 1} is missing; cannot verify continuity on resume`); + } + + this.#lastHash = null; + } + + async #getCheckpointHeight(stream: string): Promise { + const [state] = await this.#db.select().from(IndexerState).where(eq(IndexerState.stream, stream)); + return state?.lastHeight ?? null; + } +} diff --git a/apps/chain-indexer/src/pipeline/block-committer.service.spec.ts b/apps/chain-indexer/src/pipeline/block-committer.service.spec.ts index 543ca4ab4d..3da89ecba0 100644 --- a/apps/chain-indexer/src/pipeline/block-committer.service.spec.ts +++ b/apps/chain-indexer/src/pipeline/block-committer.service.spec.ts @@ -1,6 +1,8 @@ +import type { SQL } from "drizzle-orm"; +import { PgDialect } from "drizzle-orm/pg-core"; import { describe, expect, it } from "vitest"; -import { Messages, MessageTypes } from "@src/db/schema"; +import { Blocks, IndexerState, Messages, MessageTypes } from "@src/db/schema"; import { BlockCommitterService } from "@src/pipeline/block-committer.service"; import type { DecodedBlock } from "@src/pipeline/decoded-block"; import type { ChainDatabase } from "@src/providers/db.provider"; @@ -41,9 +43,66 @@ describe(BlockCommitterService.name, () => { expect(insertedRows.find(call => call.table === Messages)?.rows).toEqual([expect.objectContaining({ typeId: 9 })]); }); + it("advances the sync stream checkpoint when committing a single block", async () => { + const { committer, insertedRows } = setup({ selectResults: [[{ id: 7, type: MSG_SEND }]] }); + + await committer.commit(buildBlock([MSG_SEND])); + + expect(insertedRows.find(call => call.table === IndexerState)?.rows).toEqual(expect.objectContaining({ stream: "sync", lastHeight: 10 })); + }); + + describe("commitBatch", () => { + it("commits all blocks and advances the given stream checkpoint to the batch's last height", async () => { + const { committer, insertedRows } = setup({ selectResults: [[{ id: 7, type: MSG_SEND }]] }); + + await committer.commitBatch([buildBlock([MSG_SEND], 10), buildBlock([MSG_SEND], 11), buildBlock([MSG_SEND], 12)], { stream: "backfill:10-12" }); + + expect(insertedRows.find(call => call.table === Blocks)?.rows).toHaveLength(3); + expect(insertedRows.find(call => call.table === IndexerState)?.rows).toEqual(expect.objectContaining({ stream: "backfill:10-12", lastHeight: 12 })); + }); + + it("throws on a non-contiguous batch before writing anything", async () => { + const { committer, insertedRows } = setup(); + + await expect(committer.commitBatch([buildBlock([MSG_SEND], 10), buildBlock([MSG_SEND], 12)], { stream: "backfill:10-12" })).rejects.toThrow( + "Non-contiguous batch: expected height 11 at position 1, got 12" + ); + expect(insertedRows).toEqual([]); + }); + + it("does nothing for an empty batch", async () => { + const { committer, insertedRows } = setup(); + + await committer.commitBatch([], { stream: "backfill:10-12" }); + + expect(insertedRows).toEqual([]); + }); + + it("only ever moves the checkpoint forward on conflict, so concurrent writers cannot regress it", async () => { + const { committer, conflictUpdates } = setup({ selectResults: [[{ id: 7, type: MSG_SEND }]] }); + + await committer.commitBatch([buildBlock([MSG_SEND], 10)], { stream: "backfill:10-10" }); + + const checkpointSet = conflictUpdates.find(call => call.table === IndexerState)?.config.set as { lastHeight: SQL }; + expect(new PgDialect().sqlToQuery(checkpointSet.lastHeight).sql).toBe('GREATEST("indexer_state"."last_height", EXCLUDED.last_height)'); + }); + + it("splits large row sets into multiple inserts within the same transaction", async () => { + const { committer, insertedRows } = setup({ selectResults: [[{ id: 7, type: MSG_SEND }]] }); + const manyMessages = Array.from({ length: 2_001 }, () => MSG_SEND); + + await committer.commitBatch([buildBlock(manyMessages, 10)], { stream: "backfill:10-10" }); + + const messageInserts = insertedRows.filter(call => call.table === Messages); + expect(messageInserts).toHaveLength(2); + expect(messageInserts.map(call => (call.rows as unknown[]).length)).toEqual([2_000, 1]); + }); + }); + function setup(input?: { selectResults?: Array>; insertReturning?: Array<{ id: number; type: string }> }) { const selectResults = [...(input?.selectResults ?? [[]])]; const insertedRows: Array<{ table: unknown; rows: unknown }> = []; + const conflictUpdates: Array<{ table: unknown; config: { set: unknown } }> = []; const dbFake = { select: () => ({ from: () => ({ where: () => Promise.resolve(selectResults.shift() ?? []) }) }), @@ -55,7 +114,10 @@ describe(BlockCommitterService.name, () => { Object.assign(Promise.resolve(), { returning: () => Promise.resolve(input?.insertReturning ?? []) }), - onConflictDoUpdate: () => Promise.resolve() + onConflictDoUpdate: (config: { set: unknown }) => { + conflictUpdates.push({ table, config }); + return Promise.resolve(); + } }; } }), @@ -63,12 +125,12 @@ describe(BlockCommitterService.name, () => { }; const committer = new BlockCommitterService(dbFake as unknown as ChainDatabase); - return { committer, insertedRows }; + return { committer, insertedRows, conflictUpdates }; } - function buildBlock(typeUrls: string[]): DecodedBlock { + function buildBlock(typeUrls: string[], height = 10): DecodedBlock { return { - height: 10, + height, datetime: new Date("2026-08-11T00:00:00Z"), hash: Buffer.from("aa".repeat(32), "hex"), parentHash: null, diff --git a/apps/chain-indexer/src/pipeline/block-committer.service.ts b/apps/chain-indexer/src/pipeline/block-committer.service.ts index 056c467cf1..72c01f8f42 100644 --- a/apps/chain-indexer/src/pipeline/block-committer.service.ts +++ b/apps/chain-indexer/src/pipeline/block-committer.service.ts @@ -1,4 +1,5 @@ -import { inArray } from "drizzle-orm"; +import { inArray, sql } from "drizzle-orm"; +import chunk from "lodash/chunk"; import { inject, singleton } from "tsyringe"; import { Blocks, IndexerState, Messages, MessageTypes, Transactions } from "@src/db/schema"; @@ -8,6 +9,9 @@ import { CHAIN_DB } from "@src/providers/db.provider"; export const SYNC_STREAM = "sync"; +/** Keeps multi-row inserts well under postgres.js's ~65k bind-parameter limit when batches span hundreds of blocks. */ +const INSERT_CHUNK_SIZE = 2_000; + @singleton() export class BlockCommitterService { readonly #db: ChainDatabase; @@ -18,61 +22,96 @@ export class BlockCommitterService { } async commit(block: DecodedBlock): Promise { - const typeIds = await this.#internMessageTypes(block); + await this.commitBatch([block], { stream: SYNC_STREAM }); + } + + /** + * Commits contiguous blocks and the checkpoint advance in one transaction, so the checkpoint + * never points past uncommitted data. Inserts are conflict-ignoring and the checkpoint only + * moves forward, so concurrent writers on the same stream (e.g. two pods overlapping during a + * rolling deploy) duplicate work but cannot corrupt data or regress the checkpoint. + */ + async commitBatch(blocks: DecodedBlock[], options: { stream: string }): Promise { + if (blocks.length === 0) { + return; + } - const transactionRows = block.transactions.map(tx => ({ + this.#verifyContiguous(blocks); + const typeIds = await this.#internMessageTypes(blocks); + + const blockRows = blocks.map(block => ({ height: block.height, - index: tx.index, - hash: tx.hash, - code: tx.code, - gasUsed: tx.gasUsed, - gasWanted: tx.gasWanted, - fee: tx.fee + datetime: block.datetime, + hash: block.hash, + parentHash: block.parentHash, + proposerAddress: block.proposerAddress, + txCount: block.transactions.length })); - const messageRows = block.transactions.flatMap(tx => - tx.messages.map(message => ({ + const transactionRows = blocks.flatMap(block => + block.transactions.map(tx => ({ height: block.height, - txIndex: tx.index, - index: message.index, - typeId: typeIds.get(message.typeUrl) as number, - body: message.body + index: tx.index, + hash: tx.hash, + code: tx.code, + gasUsed: tx.gasUsed, + gasWanted: tx.gasWanted, + fee: tx.fee })) ); - await this.#db.transaction(async tx => { - await tx - .insert(Blocks) - .values({ + const messageRows = blocks.flatMap(block => + block.transactions.flatMap(tx => + tx.messages.map(message => ({ height: block.height, - datetime: block.datetime, - hash: block.hash, - parentHash: block.parentHash, - proposerAddress: block.proposerAddress, - txCount: block.transactions.length - }) - .onConflictDoNothing(); - - if (transactionRows.length > 0) { - await tx.insert(Transactions).values(transactionRows).onConflictDoNothing(); + txIndex: tx.index, + index: message.index, + typeId: typeIds.get(message.typeUrl) as number, + body: message.body + })) + ) + ); + + const lastHeight = blocks[blocks.length - 1].height; + + await this.#db.transaction(async tx => { + for (const blockChunk of chunk(blockRows, INSERT_CHUNK_SIZE)) { + await tx.insert(Blocks).values(blockChunk).onConflictDoNothing(); } - if (messageRows.length > 0) { - await tx.insert(Messages).values(messageRows).onConflictDoNothing(); + for (const transactionChunk of chunk(transactionRows, INSERT_CHUNK_SIZE)) { + await tx.insert(Transactions).values(transactionChunk).onConflictDoNothing(); + } + + for (const messageChunk of chunk(messageRows, INSERT_CHUNK_SIZE)) { + await tx.insert(Messages).values(messageChunk).onConflictDoNothing(); } await tx .insert(IndexerState) - .values({ stream: SYNC_STREAM, lastHeight: block.height, updatedAt: new Date() }) + .values({ stream: options.stream, lastHeight, updatedAt: new Date() }) .onConflictDoUpdate({ target: IndexerState.stream, - set: { lastHeight: block.height, updatedAt: new Date() } + set: { lastHeight: sql`GREATEST(${IndexerState.lastHeight}, EXCLUDED.last_height)`, updatedAt: new Date() } }); }); } - async #internMessageTypes(block: DecodedBlock): Promise> { - const typeUrls = new Set(block.transactions.flatMap(tx => tx.messages.map(message => message.typeUrl))); + /** The checkpoint advances to the batch's last height, which is only correct when the batch has no gaps or reordering. */ + #verifyContiguous(blocks: DecodedBlock[]): void { + const baseHeight = blocks[0].height; + + blocks.forEach((block, index) => { + const expectedHeight = baseHeight + index; + + if (block.height !== expectedHeight) { + throw new Error(`Non-contiguous batch: expected height ${expectedHeight} at position ${index}, got ${block.height}`); + } + }); + } + + async #internMessageTypes(blocks: DecodedBlock[]): Promise> { + const typeUrls = new Set(blocks.flatMap(block => block.transactions.flatMap(tx => tx.messages.map(message => message.typeUrl)))); const uncached = [...typeUrls].filter(typeUrl => !this.#typeIds.has(typeUrl)); if (uncached.length > 0) { diff --git a/apps/chain-indexer/src/pipeline/chain-continuity-error.ts b/apps/chain-indexer/src/pipeline/chain-continuity-error.ts new file mode 100644 index 0000000000..8b2efb703b --- /dev/null +++ b/apps/chain-indexer/src/pipeline/chain-continuity-error.ts @@ -0,0 +1,2 @@ +/** Parent-hash continuity break; fatal by design so the process halts instead of committing a forked history. */ +export class ChainContinuityError extends Error {} diff --git a/apps/chain-indexer/src/pipeline/runner-interrupted-error.ts b/apps/chain-indexer/src/pipeline/runner-interrupted-error.ts new file mode 100644 index 0000000000..b14fb0cd16 --- /dev/null +++ b/apps/chain-indexer/src/pipeline/runner-interrupted-error.ts @@ -0,0 +1,6 @@ +/** + * A runner was asked to stop (SIGTERM / container disposal) before finishing its work. The process + * must exit non-zero so a K8s Job resumes from its checkpoint on the next attempt instead of being + * marked Complete with the range still unfinished. + */ +export class RunnerInterruptedError extends Error {} diff --git a/apps/chain-indexer/src/pipeline/sync-runner.service.ts b/apps/chain-indexer/src/pipeline/sync-runner.service.ts index 16350f8113..294eea08a5 100644 --- a/apps/chain-indexer/src/pipeline/sync-runner.service.ts +++ b/apps/chain-indexer/src/pipeline/sync-runner.service.ts @@ -1,38 +1,24 @@ import { eq } from "drizzle-orm"; import { setTimeout as delay } from "node:timers/promises"; -import type postgres from "postgres"; import { inject, singleton } from "tsyringe"; import type { EnvConfig } from "@src/config/env.config"; -import { PgClientService } from "@src/db/pg-client.service"; import { Blocks, IndexerState } from "@src/db/schema"; import { BlockCommitterService, SYNC_STREAM } from "@src/pipeline/block-committer.service"; import { BlockDecoderService } from "@src/pipeline/block-decoder.service"; +import { ChainContinuityError } from "@src/pipeline/chain-continuity-error"; import type { DecodedBlock } from "@src/pipeline/decoded-block"; +import { retryTransient } from "@src/pipeline/transient-retry"; import { APP_CONFIG } from "@src/providers/app-config.provider"; import type { ChainDatabase } from "@src/providers/db.provider"; import { CHAIN_DB } from "@src/providers/db.provider"; import { LoggerService } from "@src/providers/logging.provider"; import { RpcClientPool } from "@src/rpc/rpc-client-pool.service"; -/** Arbitrary but fixed application-wide key for the sync leader pg advisory lock. */ -const SYNC_LEADER_LOCK_KEY = 7_431_001; -const LEADERSHIP_RETRY_MS = 5_000; const PROGRESS_LOG_EVERY_BLOCKS = 100; -const LEADERSHIP_CHECK_EVERY_BLOCKS = 100; -const TRANSIENT_RETRY_MAX_ATTEMPTS = 10; -const TRANSIENT_RETRY_BASE_MS = 1_000; -const TRANSIENT_RETRY_MAX_MS = 30_000; - -/** Parent-hash continuity break; fatal by design so the process halts instead of committing a forked history. */ -export class ChainContinuityError extends Error {} - -/** The advisory-lock session was replaced (e.g. a transparent driver reconnect), so another process may hold leadership. */ -export class SyncLeadershipLostError extends Error {} @singleton() export class SyncRunnerService { - readonly #pgClient: PgClientService; readonly #db: ChainDatabase; readonly #pool: RpcClientPool; readonly #decoder: BlockDecoderService; @@ -41,12 +27,9 @@ export class SyncRunnerService { readonly #logger: LoggerService; #stopped = false; - #reserved: postgres.ReservedSql | undefined; - #leaderBackendPid: number | undefined; #lastHash: Buffer | null = null; constructor( - @inject(PgClientService) pgClient: PgClientService, @inject(CHAIN_DB) db: ChainDatabase, @inject(RpcClientPool) pool: RpcClientPool, @inject(BlockDecoderService) decoder: BlockDecoderService, @@ -54,7 +37,6 @@ export class SyncRunnerService { @inject(APP_CONFIG) config: EnvConfig, @inject(LoggerService) logger: LoggerService ) { - this.#pgClient = pgClient; this.#db = db; this.#pool = pool; this.#decoder = decoder; @@ -78,26 +60,14 @@ export class SyncRunnerService { async dispose(): Promise { this.#stopped = true; - this.#releaseReservedLockConnection(); - } - - /** release() may throw when the pg pool was ended first; the connection is already gone then, which is the goal. */ - #releaseReservedLockConnection(): void { - try { - this.#reserved?.release(); - } catch (error) { - this.#logger.debug({ event: "SYNC_LOCK_RELEASE_SKIPPED", error }); - } } async #run(): Promise { - await this.#acquireLeadership(); let nextHeight = await this.#resolveStartHeight(); this.#logger.info({ event: "SYNC_STARTED", network: this.#config.NETWORK, nextHeight }); while (!this.#stopped) { - await this.#assertLeadership(); - const tipHeight = await this.#retryTransient(() => this.#getTipHeight(), { event: "SYNC_TIP_FETCH_RETRY" }); + const tipHeight = await this.#retryTransient(() => this.#pool.getTipHeight(), { event: "SYNC_TIP_FETCH_RETRY" }); if (nextHeight > tipHeight) { await delay(this.#config.SYNC_POLL_INTERVAL_MS); @@ -108,34 +78,12 @@ export class SyncRunnerService { const height = nextHeight; await this.#retryTransient(() => this.#syncBlock(height), { event: "SYNC_BLOCK_RETRY", height }); nextHeight++; - - if (nextHeight % LEADERSHIP_CHECK_EVERY_BLOCKS === 0) { - await this.#assertLeadership(); - } } } } - /** Transient failures (RPC timeouts, connection drops) are retried with capped exponential backoff; persistent ones and fatal errors propagate and halt the process. */ async #retryTransient(operation: () => Promise, logContext: { event: string; height?: number }): Promise { - let attempt = 0; - - while (true) { - attempt++; - try { - return await operation(); - } catch (error) { - const isFatal = error instanceof ChainContinuityError || error instanceof SyncLeadershipLostError; - - if (this.#stopped || isFatal || attempt >= TRANSIENT_RETRY_MAX_ATTEMPTS) { - throw error; - } - - const delayMs = Math.min(TRANSIENT_RETRY_BASE_MS * 2 ** (attempt - 1), TRANSIENT_RETRY_MAX_MS); - this.#logger.warn({ ...logContext, attempt, delayMs, error }); - await delay(delayMs); - } - } + return await retryTransient(operation, { isStopped: () => this.#stopped, logger: this.#logger, logContext }); } async #syncBlock(height: number): Promise { @@ -165,38 +113,6 @@ export class SyncRunnerService { } } - async #acquireLeadership(): Promise { - this.#reserved = await this.#pgClient.client.reserve(); - - while (!this.#stopped) { - const [{ acquired }] = await this.#reserved`SELECT pg_try_advisory_lock(${SYNC_LEADER_LOCK_KEY}) AS acquired`; - - if (acquired) { - const [{ pid }] = await this.#reserved`SELECT pg_backend_pid() AS pid`; - this.#leaderBackendPid = pid; - this.#logger.info({ event: "SYNC_LEADERSHIP_ACQUIRED", backendPid: pid }); - return; - } - - this.#logger.info({ event: "SYNC_LEADERSHIP_WAITING" }); - await delay(LEADERSHIP_RETRY_MS); - } - } - - /** Advisory locks are session-scoped: a transparent driver reconnect creates a fresh session WITHOUT the lock, so the backend pid is re-checked to detect silent leadership loss before more blocks are committed. */ - async #assertLeadership(): Promise { - if (!this.#reserved) { - throw new SyncLeadershipLostError("Reserved advisory-lock connection is gone; halting sync"); - } - - const [{ pid }] = await this.#reserved`SELECT pg_backend_pid() AS pid`; - - if (pid !== this.#leaderBackendPid) { - this.#logger.error({ event: "SYNC_LEADERSHIP_LOST", expectedBackendPid: this.#leaderBackendPid, actualBackendPid: pid }); - throw new SyncLeadershipLostError(`Advisory-lock session changed (backend pid ${this.#leaderBackendPid} -> ${pid}); halting sync`); - } - } - async #resolveStartHeight(): Promise { const [state] = await this.#db.select().from(IndexerState).where(eq(IndexerState.stream, SYNC_STREAM)); @@ -210,11 +126,6 @@ export class SyncRunnerService { return this.#config.SYNC_START_HEIGHT; } - return await this.#getTipHeight(); - } - - async #getTipHeight(): Promise { - const status = await this.#pool.getStatus(); - return parseInt(status.sync_info.latest_block_height); + return await this.#pool.getTipHeight(); } } diff --git a/apps/chain-indexer/src/pipeline/transient-retry.ts b/apps/chain-indexer/src/pipeline/transient-retry.ts new file mode 100644 index 0000000000..16776f1d0f --- /dev/null +++ b/apps/chain-indexer/src/pipeline/transient-retry.ts @@ -0,0 +1,21 @@ +import { retryWithBackoff } from "@src/lib/retry-with-backoff/retry-with-backoff"; +import { ChainContinuityError } from "@src/pipeline/chain-continuity-error"; +import type { LoggerService } from "@src/providers/logging.provider"; + +const MAX_ATTEMPTS = 10; +const BASE_DELAY_MS = 1_000; +const MAX_DELAY_MS = 30_000; + +/** Transient failures (RPC timeouts, connection drops) are retried with capped exponential backoff; fatal pipeline errors and an in-progress shutdown propagate immediately. */ +export function retryTransient( + operation: () => Promise, + options: { isStopped: () => boolean; logger: LoggerService; logContext: { event: string; height?: number } } +): Promise { + return retryWithBackoff(operation, { + maxAttempts: MAX_ATTEMPTS, + baseDelayMs: BASE_DELAY_MS, + maxDelayMs: MAX_DELAY_MS, + shouldRethrow: error => options.isStopped() || error instanceof ChainContinuityError, + onRetry: (error, attempt, delayMs) => options.logger.warn({ ...options.logContext, attempt, delayMs, error }) + }); +} diff --git a/apps/chain-indexer/src/rpc/rpc-client-pool.service.spec.ts b/apps/chain-indexer/src/rpc/rpc-client-pool.service.spec.ts index fe43228414..e506dd097d 100644 --- a/apps/chain-indexer/src/rpc/rpc-client-pool.service.spec.ts +++ b/apps/chain-indexer/src/rpc/rpc-client-pool.service.spec.ts @@ -71,6 +71,13 @@ describe(RpcClientPool.name, () => { expect(fetchMock).toHaveBeenCalledTimes(2); }); + it("parses the tip height from the status payload", async () => { + const { pool, fetchMock } = setup(); + fetchMock.mockResolvedValue(jsonResponse({ result: { sync_info: { latest_block_height: "1234" } } })); + + await expect(pool.getTipHeight()).resolves.toBe(1234); + }); + it("requests block and block results with the height as a query parameter", async () => { const { pool, fetchMock } = setup(); fetchMock.mockResolvedValue(jsonResponse({ result: { height: "7" } })); diff --git a/apps/chain-indexer/src/rpc/rpc-client-pool.service.ts b/apps/chain-indexer/src/rpc/rpc-client-pool.service.ts index 60587e5b35..27f76063e4 100644 --- a/apps/chain-indexer/src/rpc/rpc-client-pool.service.ts +++ b/apps/chain-indexer/src/rpc/rpc-client-pool.service.ts @@ -46,6 +46,11 @@ export class RpcClientPool { return await this.#get("/status"); } + async getTipHeight(): Promise { + const status = await this.getStatus(); + return parseInt(status.sync_info.latest_block_height); + } + async getBlock(height: number): Promise { return await this.#get(`/block?height=${height}`); }