feat(indexer): backfill role for historical catch-up

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

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

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<string, string>) {
     return envSchema.parse({ POSTGRES_DB_URI: "postgres://unit:unit@localhost:5432/unit", ...overrides });
   }

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<typeof envSchema>;

apps/chain-indexer/src/index.ts

@@ -1,39 +1,44 @@
 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";
 import { shutdownServer } from "@src/services/shutdown-server/shutdown-server";
 import { startServer } from "@src/services/start-server/start-server";
 
 export async function bootstrap(): Promise<void> {
+  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<void> {
     }
   }
 }
+
+/** 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<void> }, fatalEvent: string, logger: LoggerService, port: number): Promise<void> {
+  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);
+}

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]);
+  });
+});

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<T>(operation: () => Promise<T>, options: RetryWithBackoffOptions): Promise<T> {
+  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<void> {
+  return new Promise(resolve => {
+    setTimeout(resolve, ms);
+  });
+}

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 });
+  });
+});

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
+  };
+}