From 60a7babf1b79b68a4b8fc3dcc5ab8186d68f6cad Mon Sep 17 00:00:00 2001 From: Maxime Beauchamp <15185355+baktun14@users.noreply.github.com> Date: Tue, 11 Aug 2026 18:22:25 +0530 Subject: [PATCH 01/14] feat(indexer): add backfill range and tuning env config --- apps/chain-indexer/env/.env.sample | 4 +++ .../src/config/env.config.spec.ts | 28 +++++++++++++++++++ apps/chain-indexer/src/config/env.config.ts | 26 ++++++++++++++++- 3 files changed, 57 insertions(+), 1 deletion(-) 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; From 50378c53c3714b5b74834e7adeca5766f046ff21 Mon Sep 17 00:00:00 2001 From: Maxime Beauchamp <15185355+baktun14@users.noreply.github.com> Date: Tue, 11 Aug 2026 18:23:57 +0530 Subject: [PATCH 02/14] feat(indexer): support ordered batched commits with a parameterized checkpoint stream --- .../pipeline/block-committer.service.spec.ts | 53 ++++++++- .../src/pipeline/block-committer.service.ts | 109 ++++++++++++------ 2 files changed, 125 insertions(+), 37 deletions(-) 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..aec0e650a4 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,6 @@ 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,6 +41,53 @@ 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("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 }> = []; @@ -66,9 +113,9 @@ describe(BlockCommitterService.name, () => { return { committer, insertedRows }; } - 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..e142a33262 100644 --- a/apps/chain-indexer/src/pipeline/block-committer.service.ts +++ b/apps/chain-indexer/src/pipeline/block-committer.service.ts @@ -8,6 +8,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 +21,89 @@ 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. */ + async commitBatch(blocks: DecodedBlock[], options: { stream: string }): Promise { + if (blocks.length === 0) { + return; + } + + this.#verifyContiguous(blocks); + const typeIds = await this.#internMessageTypes(blocks); - const transactionRows = block.transactions.map(tx => ({ + 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 chunk of chunked(blockRows, INSERT_CHUNK_SIZE)) { + await tx.insert(Blocks).values(chunk).onConflictDoNothing(); } - if (messageRows.length > 0) { - await tx.insert(Messages).values(messageRows).onConflictDoNothing(); + for (const chunk of chunked(transactionRows, INSERT_CHUNK_SIZE)) { + await tx.insert(Transactions).values(chunk).onConflictDoNothing(); + } + + for (const chunk of chunked(messageRows, INSERT_CHUNK_SIZE)) { + await tx.insert(Messages).values(chunk).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, 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 { + blocks.forEach((block, index) => { + const expectedHeight = blocks[0].height + 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) { @@ -108,3 +139,13 @@ export class BlockCommitterService { } } } + +function chunked(rows: T[], size: number): T[][] { + const chunks: T[][] = []; + + for (let start = 0; start < rows.length; start += size) { + chunks.push(rows.slice(start, start + size)); + } + + return chunks; +} From ff12f2135178f59fc213f3d8f4387dc15f1bd946 Mon Sep 17 00:00:00 2001 From: Maxime Beauchamp <15185355+baktun14@users.noreply.github.com> Date: Tue, 11 Aug 2026 18:24:34 +0530 Subject: [PATCH 03/14] feat(indexer): add backfill range planner --- .../src/pipeline/backfill-planner.spec.ts | 35 +++++++++++++++++++ .../src/pipeline/backfill-planner.ts | 25 +++++++++++++ 2 files changed, 60 insertions(+) create mode 100644 apps/chain-indexer/src/pipeline/backfill-planner.spec.ts create mode 100644 apps/chain-indexer/src/pipeline/backfill-planner.ts 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..79d6518570 --- /dev/null +++ b/apps/chain-indexer/src/pipeline/backfill-planner.spec.ts @@ -0,0 +1,35 @@ +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("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..d5a3d1de8c --- /dev/null +++ b/apps/chain-indexer/src/pipeline/backfill-planner.ts @@ -0,0 +1,25 @@ +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. */ +export function planBackfill(input: BackfillPlanInput): BackfillPlan { + if (input.toHeight > input.tipHeight) { + return { kind: "invalid", reason: `BACKFILL_TO_HEIGHT ${input.toHeight} is above the chain tip ${input.tipHeight}` }; + } + + if (input.checkpointHeight !== null && input.checkpointHeight >= input.toHeight) { + return { kind: "already-complete" }; + } + + return { + kind: "run", + startHeight: input.checkpointHeight !== null ? input.checkpointHeight + 1 : input.fromHeight, + endHeight: input.toHeight + }; +} From 00b466bf4b6ec41dc6b1f961d6ce10ad1f31aa1b Mon Sep 17 00:00:00 2001 From: Maxime Beauchamp <15185355+baktun14@users.noreply.github.com> Date: Tue, 11 Aug 2026 18:27:58 +0530 Subject: [PATCH 04/14] feat(indexer): implement the backfill role for historical catch-up --- apps/chain-indexer/README.md | 14 +- apps/chain-indexer/src/index.ts | 15 + .../pipeline/backfill-runner.service.spec.ts | 233 +++++++++++++++ .../src/pipeline/backfill-runner.service.ts | 277 ++++++++++++++++++ 4 files changed, 538 insertions(+), 1 deletion(-) create mode 100644 apps/chain-indexer/src/pipeline/backfill-runner.service.spec.ts create mode 100644 apps/chain-indexer/src/pipeline/backfill-runner.service.ts diff --git a/apps/chain-indexer/README.md b/apps/chain-indexer/README.md index a98f36f8f3..ccfd93e409 100644 --- a/apps/chain-indexer/README.md +++ b/apps/chain-indexer/README.md @@ -9,7 +9,7 @@ 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, advisory-lock leader election, 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`. ## Running locally @@ -34,6 +34,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; a separate advisory lock prevents two concurrent backfills. + ## Tests ```bash diff --git a/apps/chain-indexer/src/index.ts b/apps/chain-indexer/src/index.ts index 3944fc6b74..1768a774ad 100644 --- a/apps/chain-indexer/src/index.ts +++ b/apps/chain-indexer/src/index.ts @@ -5,6 +5,7 @@ import { createOtelLogger } from "@akashnetwork/logging/otel"; import { container } from "tsyringe"; import { createApp } from "@src/app"; +import { BackfillRunnerService } from "@src/pipeline/backfill-runner.service"; 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"; @@ -31,6 +32,20 @@ export async function bootstrap(): Promise { await shutdownServer(server, logger); return; } + case "backfill": { + await migrateDb(); + const server = await startServer(createApp(), logger, process, { port: config.get("PORT") }); + + try { + await container.resolve(BackfillRunnerService).start(); + } catch (error) { + logger.error({ event: "BACKFILL_FATAL", error }); + process.exitCode = 1; + } + + await shutdownServer(server, logger); + return; + } case "api": { await migrateDb(); await startServer(createApp(), logger, process, { port: config.get("PORT") }); 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..b18dc9ce7d --- /dev/null +++ b/apps/chain-indexer/src/pipeline/backfill-runner.service.spec.ts @@ -0,0 +1,233 @@ +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 type { PgClientService } from "@src/db/pg-client.service"; +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 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 () => { + const { runner, committer, logger } = setup({ fromHeight: 1, toHeight: 2, failFetchOnceAtHeight: 2 }); + + await runner.start(); + + expect(committedHeights(committer)).toEqual([[1, 2]]); + expect(logger.warn).toHaveBeenCalledWith(expect.objectContaining({ event: "BACKFILL_FETCH_RETRY", height: 2, attempt: 1 })); + }); + + 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) + }) + ); + }); + + 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 reserved = Object.assign( + vi.fn((strings: TemplateStringsArray) => Promise.resolve(strings.join("?").includes("pg_try_advisory_lock") ? [{ acquired: true }] : [{ pid: 42 }])), + { release: vi.fn() } + ); + const pgClient = mock({ + client: { reserve: vi.fn().mockResolvedValue(reserved) } as unknown as PgClientService["client"] + }); + + 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.getStatus.mockResolvedValue({ sync_info: { latest_block_height: String(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(pgClient, 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..73a083a791 --- /dev/null +++ b/apps/chain-indexer/src/pipeline/backfill-runner.service.ts @@ -0,0 +1,277 @@ +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 { planBackfill } from "@src/pipeline/backfill-planner"; +import { BlockCommitterService } from "@src/pipeline/block-committer.service"; +import { BlockDecoderService } from "@src/pipeline/block-decoder.service"; +import type { DecodedBlock } from "@src/pipeline/decoded-block"; +import { ChainContinuityError } from "@src/pipeline/sync-runner.service"; +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 backfill leader pg advisory lock; distinct from the sync leader key so a backfill never contends with live sync. */ +const BACKFILL_LEADER_LOCK_KEY = 7_431_002; +const LEADERSHIP_RETRY_MS = 5_000; +const FETCH_RETRY_MAX_ATTEMPTS = 5; +const FETCH_RETRY_BASE_MS = 1_000; + +@singleton() +export class BackfillRunnerService { + readonly #pgClient: PgClientService; + readonly #db: ChainDatabase; + readonly #pool: RpcClientPool; + readonly #decoder: BlockDecoderService; + readonly #committer: BlockCommitterService; + readonly #config: EnvConfig; + 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, + @inject(BlockCommitterService) committer: BlockCommitterService, + @inject(APP_CONFIG) config: EnvConfig, + @inject(LoggerService) logger: LoggerService + ) { + this.#pgClient = pgClient; + 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 { + try { + await this.#run(); + } catch (error) { + if (this.#stopped) { + this.#logger.info({ event: "BACKFILL_STOPPED_DURING_SHUTDOWN" }); + return; + } + throw error; + } + } + + 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: "BACKFILL_LOCK_RELEASE_SKIPPED", error }); + } + } + + 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"); + } + + await this.#acquireLeadership(); + + if (this.#stopped) { + return; + } + + const stream = `backfill:${fromHeight}-${toHeight}`; + const checkpointHeight = await this.#getCheckpointHeight(stream); + const tipHeight = await this.#getTipHeight(); + 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; + } + + await this.#seedContinuityHash(plan.startHeight, checkpointHeight !== null); + this.#logger.info({ event: "BACKFILL_STARTED", network: this.#config.NETWORK, stream, startHeight: plan.startHeight, endHeight: plan.endHeight }); + 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. + */ + 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 batch: DecodedBlock[] = []; + + const fillFetchWindow = () => { + while (fetchHead <= endHeight && inflight.size < this.#config.BACKFILL_CONCURRENCY) { + const height = fetchHead; + inflight.set(height, this.#fetchAndDecode(height)); + fetchHead++; + } + }; + + try { + for (let height = startHeight; height <= endHeight && !this.#stopped; height++) { + fillFetchWindow(); + const decoded = await (inflight.get(height) ?? this.#fetchAndDecode(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) { + await this.#assertLeadership(); + await this.#committer.commitBatch(batch, { stream }); + blocksCommitted += batch.length; + transactionsCommitted += batch.reduce((sum, block) => sum + block.transactions.length, 0); + batch = []; + this.#logger.info({ event: "BACKFILL_PROGRESS", height, endHeight, blocksCommitted }); + } + } + } finally { + await Promise.allSettled([...inflight.values()]); + } + + if (this.#stopped) { + return; + } + + 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 + }); + } + + /** A pool AggregateError means every RPC endpoint already failed once, so retries back off before another full sweep. */ + async #fetchAndDecode(height: number): Promise { + let attempt = 0; + + while (true) { + attempt++; + try { + const [block, blockResults] = await Promise.all([this.#pool.getBlock(height), this.#pool.getBlockResults(height)]); + return this.#decoder.decode(block, blockResults); + } catch (error) { + if (this.#stopped || attempt >= FETCH_RETRY_MAX_ATTEMPTS) { + throw error; + } + + const delayMs = FETCH_RETRY_BASE_MS * 2 ** (attempt - 1); + this.#logger.warn({ event: "BACKFILL_FETCH_RETRY", height, attempt, delayMs, error }); + await delay(delayMs); + } + } + } + + #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.#db + .select() + .from(Blocks) + .where(eq(Blocks.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; + } + + async #acquireLeadership(): Promise { + this.#reserved = await this.#pgClient.client.reserve(); + + while (!this.#stopped) { + const [{ acquired }] = await this.#reserved`SELECT pg_try_advisory_lock(${BACKFILL_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: "BACKFILL_LEADERSHIP_ACQUIRED", backendPid: pid }); + return; + } + + this.#logger.info({ event: "BACKFILL_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 before each batch commit. */ + async #assertLeadership(): Promise { + if (!this.#reserved) { + throw new Error("Reserved advisory-lock connection is gone; halting backfill"); + } + + const [{ pid }] = await this.#reserved`SELECT pg_backend_pid() AS pid`; + + if (pid !== this.#leaderBackendPid) { + this.#logger.error({ event: "BACKFILL_LEADERSHIP_LOST", expectedBackendPid: this.#leaderBackendPid, actualBackendPid: pid }); + throw new Error(`Advisory-lock session changed (backend pid ${this.#leaderBackendPid} -> ${pid}); halting backfill`); + } + } + + async #getTipHeight(): Promise { + const status = await this.#pool.getStatus(); + return parseInt(status.sync_info.latest_block_height); + } +} From 9b722c9fce492fabe0b5226474296129229e6132 Mon Sep 17 00:00:00 2001 From: Maxime Beauchamp <15185355+baktun14@users.noreply.github.com> Date: Tue, 11 Aug 2026 18:51:03 +0530 Subject: [PATCH 05/14] fix(indexer): address backfill review findings - attach a no-op catch to prefetched blocks so a rejection settling before its height is consumed cannot crash as an unhandled rejection - check range completion before tip validity so a finished range stays a no-op even when a lagging node reports a stale tip - extract the shared PgAdvisoryLeaderLock used by both runners - use lodash chunk instead of a local helper - build the pg client test double with a nested mock instead of a cast --- .../src/db/pg-advisory-leader-lock.ts | 72 +++++++++++++++++++ .../src/pipeline/backfill-planner.spec.ts | 6 ++ .../src/pipeline/backfill-planner.ts | 14 ++-- .../pipeline/backfill-runner.service.spec.ts | 2 +- .../src/pipeline/backfill-runner.service.ts | 68 +++++------------- .../src/pipeline/block-committer.service.ts | 23 ++---- .../src/pipeline/sync-runner.service.ts | 66 ++++------------- 7 files changed, 124 insertions(+), 127 deletions(-) create mode 100644 apps/chain-indexer/src/db/pg-advisory-leader-lock.ts diff --git a/apps/chain-indexer/src/db/pg-advisory-leader-lock.ts b/apps/chain-indexer/src/db/pg-advisory-leader-lock.ts new file mode 100644 index 0000000000..d6c3d27ca3 --- /dev/null +++ b/apps/chain-indexer/src/db/pg-advisory-leader-lock.ts @@ -0,0 +1,72 @@ +import { setTimeout as delay } from "node:timers/promises"; +import type postgres from "postgres"; + +import type { LoggerService } from "@src/providers/logging.provider"; + +const ACQUIRE_RETRY_MS = 5_000; + +/** The advisory-lock session was replaced (e.g. a transparent driver reconnect), so another process may hold leadership. */ +export class LeadershipLostError extends Error {} + +/** + * Single-leader election on a session-scoped pg advisory lock. The lock dies silently with its + * session, so holders must re-verify leadership with assertHeld() before trusting it. + */ +export class PgAdvisoryLeaderLock { + readonly #client: postgres.Sql; + readonly #lockKey: number; + readonly #logger: LoggerService; + readonly #eventPrefix: string; + + #reserved: postgres.ReservedSql | undefined; + #leaderBackendPid: number | undefined; + + constructor(options: { client: postgres.Sql; lockKey: number; logger: LoggerService; eventPrefix: string }) { + this.#client = options.client; + this.#lockKey = options.lockKey; + this.#logger = options.logger; + this.#eventPrefix = options.eventPrefix; + } + + /** Spins until the lock is acquired or shouldAbort() returns true. */ + async acquire(shouldAbort: () => boolean): Promise { + this.#reserved = await this.#client.reserve(); + + while (!shouldAbort()) { + const [{ acquired }] = await this.#reserved`SELECT pg_try_advisory_lock(${this.#lockKey}) AS acquired`; + + if (acquired) { + const [{ pid }] = await this.#reserved`SELECT pg_backend_pid() AS pid`; + this.#leaderBackendPid = pid; + this.#logger.info({ event: `${this.#eventPrefix}_LEADERSHIP_ACQUIRED`, backendPid: pid }); + return; + } + + this.#logger.info({ event: `${this.#eventPrefix}_LEADERSHIP_WAITING` }); + await delay(ACQUIRE_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. */ + async assertHeld(): Promise { + if (!this.#reserved) { + throw new LeadershipLostError("Reserved advisory-lock connection is gone"); + } + + const [{ pid }] = await this.#reserved`SELECT pg_backend_pid() AS pid`; + + if (pid !== this.#leaderBackendPid) { + this.#logger.error({ event: `${this.#eventPrefix}_LEADERSHIP_LOST`, expectedBackendPid: this.#leaderBackendPid, actualBackendPid: pid }); + throw new LeadershipLostError(`Advisory-lock session changed (backend pid ${this.#leaderBackendPid} -> ${pid})`); + } + } + + /** release() may throw when the pg pool was ended first; the connection is already gone then, which is the goal. */ + release(): void { + try { + this.#reserved?.release(); + } catch (error) { + this.#logger.debug({ event: `${this.#eventPrefix}_LOCK_RELEASE_SKIPPED`, error }); + } + } +} diff --git a/apps/chain-indexer/src/pipeline/backfill-planner.spec.ts b/apps/chain-indexer/src/pipeline/backfill-planner.spec.ts index 79d6518570..68670e4045 100644 --- a/apps/chain-indexer/src/pipeline/backfill-planner.spec.ts +++ b/apps/chain-indexer/src/pipeline/backfill-planner.spec.ts @@ -21,6 +21,12 @@ describe(planBackfill.name, () => { 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 }); diff --git a/apps/chain-indexer/src/pipeline/backfill-planner.ts b/apps/chain-indexer/src/pipeline/backfill-planner.ts index d5a3d1de8c..739265f1a0 100644 --- a/apps/chain-indexer/src/pipeline/backfill-planner.ts +++ b/apps/chain-indexer/src/pipeline/backfill-planner.ts @@ -7,16 +7,20 @@ export interface BackfillPlanInput { 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. */ +/** + * 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.toHeight > input.tipHeight) { - return { kind: "invalid", reason: `BACKFILL_TO_HEIGHT ${input.toHeight} is above the chain tip ${input.tipHeight}` }; - } - 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, diff --git a/apps/chain-indexer/src/pipeline/backfill-runner.service.spec.ts b/apps/chain-indexer/src/pipeline/backfill-runner.service.spec.ts index b18dc9ce7d..ea91930b6f 100644 --- a/apps/chain-indexer/src/pipeline/backfill-runner.service.spec.ts +++ b/apps/chain-indexer/src/pipeline/backfill-runner.service.spec.ts @@ -147,7 +147,7 @@ describe(BackfillRunnerService.name, () => { { release: vi.fn() } ); const pgClient = mock({ - client: { reserve: vi.fn().mockResolvedValue(reserved) } as unknown as PgClientService["client"] + client: mock({ reserve: vi.fn().mockResolvedValue(reserved) }) }); const dbFake = { diff --git a/apps/chain-indexer/src/pipeline/backfill-runner.service.ts b/apps/chain-indexer/src/pipeline/backfill-runner.service.ts index 73a083a791..d3c5769970 100644 --- a/apps/chain-indexer/src/pipeline/backfill-runner.service.ts +++ b/apps/chain-indexer/src/pipeline/backfill-runner.service.ts @@ -1,9 +1,9 @@ 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 { PgAdvisoryLeaderLock } from "@src/db/pg-advisory-leader-lock"; import { PgClientService } from "@src/db/pg-client.service"; import { Blocks, IndexerState } from "@src/db/schema"; import { planBackfill } from "@src/pipeline/backfill-planner"; @@ -19,23 +19,20 @@ import { RpcClientPool } from "@src/rpc/rpc-client-pool.service"; /** Arbitrary but fixed application-wide key for the backfill leader pg advisory lock; distinct from the sync leader key so a backfill never contends with live sync. */ const BACKFILL_LEADER_LOCK_KEY = 7_431_002; -const LEADERSHIP_RETRY_MS = 5_000; const FETCH_RETRY_MAX_ATTEMPTS = 5; const FETCH_RETRY_BASE_MS = 1_000; @singleton() export class BackfillRunnerService { - readonly #pgClient: PgClientService; readonly #db: ChainDatabase; readonly #pool: RpcClientPool; readonly #decoder: BlockDecoderService; readonly #committer: BlockCommitterService; readonly #config: EnvConfig; readonly #logger: LoggerService; + readonly #leaderLock: PgAdvisoryLeaderLock; #stopped = false; - #reserved: postgres.ReservedSql | undefined; - #leaderBackendPid: number | undefined; #lastHash: Buffer | null = null; constructor( @@ -47,7 +44,6 @@ export class BackfillRunnerService { @inject(APP_CONFIG) config: EnvConfig, @inject(LoggerService) logger: LoggerService ) { - this.#pgClient = pgClient; this.#db = db; this.#pool = pool; this.#decoder = decoder; @@ -55,6 +51,12 @@ export class BackfillRunnerService { this.#config = config; this.#logger = logger; this.#logger.setContext("BACKFILL"); + this.#leaderLock = new PgAdvisoryLeaderLock({ + client: pgClient.client, + lockKey: BACKFILL_LEADER_LOCK_KEY, + logger: this.#logger, + eventPrefix: "BACKFILL" + }); } async start(): Promise { @@ -71,16 +73,7 @@ export class BackfillRunnerService { 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: "BACKFILL_LOCK_RELEASE_SKIPPED", error }); - } + this.#leaderLock.release(); } async #run(): Promise { @@ -90,7 +83,7 @@ export class BackfillRunnerService { throw new Error("BACKFILL_FROM_HEIGHT and BACKFILL_TO_HEIGHT are required for the backfill role"); } - await this.#acquireLeadership(); + await this.#leaderLock.acquire(() => this.#stopped); if (this.#stopped) { return; @@ -119,6 +112,9 @@ export class BackfillRunnerService { /** * 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. */ async #backfillRange(startHeight: number, endHeight: number, stream: string): Promise { const startedAt = Date.now(); @@ -131,7 +127,9 @@ export class BackfillRunnerService { const fillFetchWindow = () => { while (fetchHead <= endHeight && inflight.size < this.#config.BACKFILL_CONCURRENCY) { const height = fetchHead; - inflight.set(height, this.#fetchAndDecode(height)); + const prefetched = this.#fetchAndDecode(height); + prefetched.catch(() => undefined); + inflight.set(height, prefetched); fetchHead++; } }; @@ -148,7 +146,7 @@ export class BackfillRunnerService { fillFetchWindow(); if (batch.length >= this.#config.BACKFILL_BATCH_SIZE || height === endHeight) { - await this.#assertLeadership(); + await this.#leaderLock.assertHeld(); await this.#committer.commitBatch(batch, { stream }); blocksCommitted += batch.length; transactionsCommitted += batch.reduce((sum, block) => sum + block.transactions.length, 0); @@ -238,38 +236,6 @@ export class BackfillRunnerService { return state?.lastHeight ?? null; } - async #acquireLeadership(): Promise { - this.#reserved = await this.#pgClient.client.reserve(); - - while (!this.#stopped) { - const [{ acquired }] = await this.#reserved`SELECT pg_try_advisory_lock(${BACKFILL_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: "BACKFILL_LEADERSHIP_ACQUIRED", backendPid: pid }); - return; - } - - this.#logger.info({ event: "BACKFILL_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 before each batch commit. */ - async #assertLeadership(): Promise { - if (!this.#reserved) { - throw new Error("Reserved advisory-lock connection is gone; halting backfill"); - } - - const [{ pid }] = await this.#reserved`SELECT pg_backend_pid() AS pid`; - - if (pid !== this.#leaderBackendPid) { - this.#logger.error({ event: "BACKFILL_LEADERSHIP_LOST", expectedBackendPid: this.#leaderBackendPid, actualBackendPid: pid }); - throw new Error(`Advisory-lock session changed (backend pid ${this.#leaderBackendPid} -> ${pid}); halting backfill`); - } - } - async #getTipHeight(): Promise { const status = await this.#pool.getStatus(); return parseInt(status.sync_info.latest_block_height); diff --git a/apps/chain-indexer/src/pipeline/block-committer.service.ts b/apps/chain-indexer/src/pipeline/block-committer.service.ts index e142a33262..c0d7824978 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 chunk from "lodash/chunk"; import { inject, singleton } from "tsyringe"; import { Blocks, IndexerState, Messages, MessageTypes, Transactions } from "@src/db/schema"; @@ -69,16 +70,16 @@ export class BlockCommitterService { const lastHeight = blocks[blocks.length - 1].height; await this.#db.transaction(async tx => { - for (const chunk of chunked(blockRows, INSERT_CHUNK_SIZE)) { - await tx.insert(Blocks).values(chunk).onConflictDoNothing(); + for (const blockChunk of chunk(blockRows, INSERT_CHUNK_SIZE)) { + await tx.insert(Blocks).values(blockChunk).onConflictDoNothing(); } - for (const chunk of chunked(transactionRows, INSERT_CHUNK_SIZE)) { - await tx.insert(Transactions).values(chunk).onConflictDoNothing(); + for (const transactionChunk of chunk(transactionRows, INSERT_CHUNK_SIZE)) { + await tx.insert(Transactions).values(transactionChunk).onConflictDoNothing(); } - for (const chunk of chunked(messageRows, INSERT_CHUNK_SIZE)) { - await tx.insert(Messages).values(chunk).onConflictDoNothing(); + for (const messageChunk of chunk(messageRows, INSERT_CHUNK_SIZE)) { + await tx.insert(Messages).values(messageChunk).onConflictDoNothing(); } await tx @@ -139,13 +140,3 @@ export class BlockCommitterService { } } } - -function chunked(rows: T[], size: number): T[][] { - const chunks: T[][] = []; - - for (let start = 0; start < rows.length; start += size) { - chunks.push(rows.slice(start, start + size)); - } - - return chunks; -} diff --git a/apps/chain-indexer/src/pipeline/sync-runner.service.ts b/apps/chain-indexer/src/pipeline/sync-runner.service.ts index 16350f8113..d1b71f48f0 100644 --- a/apps/chain-indexer/src/pipeline/sync-runner.service.ts +++ b/apps/chain-indexer/src/pipeline/sync-runner.service.ts @@ -1,9 +1,9 @@ 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 { LeadershipLostError, PgAdvisoryLeaderLock } from "@src/db/pg-advisory-leader-lock"; 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"; @@ -17,7 +17,6 @@ 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; @@ -27,22 +26,17 @@ 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; readonly #committer: BlockCommitterService; readonly #config: EnvConfig; readonly #logger: LoggerService; + readonly #leaderLock: PgAdvisoryLeaderLock; #stopped = false; - #reserved: postgres.ReservedSql | undefined; - #leaderBackendPid: number | undefined; #lastHash: Buffer | null = null; constructor( @@ -54,7 +48,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; @@ -62,6 +55,7 @@ export class SyncRunnerService { this.#config = config; this.#logger = logger; this.#logger.setContext("SYNC"); + this.#leaderLock = new PgAdvisoryLeaderLock({ client: pgClient.client, lockKey: SYNC_LEADER_LOCK_KEY, logger: this.#logger, eventPrefix: "SYNC" }); } async start(): Promise { @@ -78,25 +72,21 @@ export class SyncRunnerService { async dispose(): Promise { this.#stopped = true; - this.#releaseReservedLockConnection(); + this.#leaderLock.release(); } - /** 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.#leaderLock.acquire(() => this.#stopped); + + if (this.#stopped) { + return; } - } - 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(); + await this.#leaderLock.assertHeld(); const tipHeight = await this.#retryTransient(() => this.#getTipHeight(), { event: "SYNC_TIP_FETCH_RETRY" }); if (nextHeight > tipHeight) { @@ -110,7 +100,7 @@ export class SyncRunnerService { nextHeight++; if (nextHeight % LEADERSHIP_CHECK_EVERY_BLOCKS === 0) { - await this.#assertLeadership(); + await this.#leaderLock.assertHeld(); } } } @@ -125,7 +115,7 @@ export class SyncRunnerService { try { return await operation(); } catch (error) { - const isFatal = error instanceof ChainContinuityError || error instanceof SyncLeadershipLostError; + const isFatal = error instanceof ChainContinuityError || error instanceof LeadershipLostError; if (this.#stopped || isFatal || attempt >= TRANSIENT_RETRY_MAX_ATTEMPTS) { throw error; @@ -165,38 +155,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)); From 25a74c953d348eb6e6acea00664f1baca51932e0 Mon Sep 17 00:00:00 2001 From: Maxime Beauchamp <15185355+baktun14@users.noreply.github.com> Date: Tue, 11 Aug 2026 19:09:38 +0530 Subject: [PATCH 06/14] refactor(indexer): share retry backoff and runner-role lifecycle across roles - extract retryWithBackoff and use it in both runners - retry backfill startup reads and idempotent batch commits so a transient blip no longer kills a multi-hour job - collapse the sync and backfill bootstrap cases into one helper --- apps/chain-indexer/src/index.ts | 44 +++++++-------- .../retry-with-backoff.spec.ts | 38 +++++++++++++ .../retry-with-backoff/retry-with-backoff.ts | 30 ++++++++++ .../src/pipeline/backfill-runner.service.ts | 55 ++++++++++++------- .../src/pipeline/sync-runner.service.ts | 26 +++------ 5 files changed, 132 insertions(+), 61 deletions(-) create mode 100644 apps/chain-indexer/src/lib/retry-with-backoff/retry-with-backoff.spec.ts create mode 100644 apps/chain-indexer/src/lib/retry-with-backoff/retry-with-backoff.ts diff --git a/apps/chain-indexer/src/index.ts b/apps/chain-indexer/src/index.ts index 1768a774ad..8ea313ac7f 100644 --- a/apps/chain-indexer/src/index.ts +++ b/apps/chain-indexer/src/index.ts @@ -12,43 +12,26 @@ 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"; +type AppLogger = ReturnType; + export async function bootstrap(): Promise { const config = container.resolve(AppConfigService); const role = config.get("INDEXER_ROLE"); + const port = config.get("PORT"); const logger = createOtelLogger({ context: "APP" }); 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 migrateDb(); - const server = await startServer(createApp(), logger, process, { port: config.get("PORT") }); - - try { - await container.resolve(BackfillRunnerService).start(); - } catch (error) { - logger.error({ event: "BACKFILL_FATAL", error }); - process.exitCode = 1; - } - - await shutdownServer(server, logger); + 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: { @@ -57,3 +40,18 @@ export async function bootstrap(): Promise { } } } + +/** Shared runner-role lifecycle: migrate, serve healthz, run to completion or fatal error (exit code 1), then shut the server down so the process can exit. */ +async function runRunnerBehindServer(resolveRunner: () => { start(): Promise }, fatalEvent: string, logger: AppLogger, port: number): Promise { + await migrateDb(); + const server = await startServer(createApp(), logger, process, { port }); + + try { + await resolveRunner().start(); + } catch (error) { + 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..e6cf78eb9c --- /dev/null +++ b/apps/chain-indexer/src/lib/retry-with-backoff/retry-with-backoff.ts @@ -0,0 +1,30 @@ +import { setTimeout as delay } from "node:timers/promises"; + +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); + } + } +} diff --git a/apps/chain-indexer/src/pipeline/backfill-runner.service.ts b/apps/chain-indexer/src/pipeline/backfill-runner.service.ts index d3c5769970..0a94041fe7 100644 --- a/apps/chain-indexer/src/pipeline/backfill-runner.service.ts +++ b/apps/chain-indexer/src/pipeline/backfill-runner.service.ts @@ -1,11 +1,11 @@ import { eq } from "drizzle-orm"; -import { setTimeout as delay } from "node:timers/promises"; import { inject, singleton } from "tsyringe"; import type { EnvConfig } from "@src/config/env.config"; -import { PgAdvisoryLeaderLock } from "@src/db/pg-advisory-leader-lock"; +import { LeadershipLostError, PgAdvisoryLeaderLock } from "@src/db/pg-advisory-leader-lock"; import { PgClientService } from "@src/db/pg-client.service"; 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"; @@ -21,6 +21,9 @@ import { RpcClientPool } from "@src/rpc/rpc-client-pool.service"; const BACKFILL_LEADER_LOCK_KEY = 7_431_002; const FETCH_RETRY_MAX_ATTEMPTS = 5; const FETCH_RETRY_BASE_MS = 1_000; +const TRANSIENT_RETRY_MAX_ATTEMPTS = 10; +const TRANSIENT_RETRY_BASE_MS = 1_000; +const TRANSIENT_RETRY_MAX_MS = 30_000; @singleton() export class BackfillRunnerService { @@ -90,8 +93,8 @@ export class BackfillRunnerService { } const stream = `backfill:${fromHeight}-${toHeight}`; - const checkpointHeight = await this.#getCheckpointHeight(stream); - const tipHeight = await this.#getTipHeight(); + const checkpointHeight = await this.#retryTransient(() => this.#getCheckpointHeight(stream), { event: "BACKFILL_CHECKPOINT_READ_RETRY" }); + const tipHeight = await this.#retryTransient(() => this.#getTipHeight(), { event: "BACKFILL_TIP_FETCH_RETRY" }); const plan = planBackfill({ fromHeight, toHeight, checkpointHeight, tipHeight }); if (plan.kind === "invalid") { @@ -146,8 +149,14 @@ export class BackfillRunnerService { fillFetchWindow(); if (batch.length >= this.#config.BACKFILL_BATCH_SIZE || height === endHeight) { - await this.#leaderLock.assertHeld(); - await this.#committer.commitBatch(batch, { stream }); + const currentBatch = batch; + await this.#retryTransient( + async () => { + await this.#leaderLock.assertHeld(); + await this.#committer.commitBatch(currentBatch, { stream }); + }, + { event: "BACKFILL_COMMIT_RETRY", height } + ); blocksCommitted += batch.length; transactionsCommitted += batch.reduce((sum, block) => sum + block.transactions.length, 0); batch = []; @@ -175,25 +184,31 @@ export class BackfillRunnerService { }); } + /** 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 retryWithBackoff(operation, { + maxAttempts: TRANSIENT_RETRY_MAX_ATTEMPTS, + baseDelayMs: TRANSIENT_RETRY_BASE_MS, + maxDelayMs: TRANSIENT_RETRY_MAX_MS, + shouldRethrow: error => this.#stopped || error instanceof ChainContinuityError || error instanceof LeadershipLostError, + onRetry: (error, attempt, delayMs) => this.#logger.warn({ ...logContext, attempt, delayMs, error }) + }); + } + /** A pool AggregateError means every RPC endpoint already failed once, so retries back off before another full sweep. */ async #fetchAndDecode(height: number): Promise { - let attempt = 0; - - while (true) { - attempt++; - try { + return await retryWithBackoff( + async () => { const [block, blockResults] = await Promise.all([this.#pool.getBlock(height), this.#pool.getBlockResults(height)]); return this.#decoder.decode(block, blockResults); - } catch (error) { - if (this.#stopped || attempt >= FETCH_RETRY_MAX_ATTEMPTS) { - throw error; - } - - const delayMs = FETCH_RETRY_BASE_MS * 2 ** (attempt - 1); - this.#logger.warn({ event: "BACKFILL_FETCH_RETRY", height, attempt, delayMs, error }); - await delay(delayMs); + }, + { + 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 { diff --git a/apps/chain-indexer/src/pipeline/sync-runner.service.ts b/apps/chain-indexer/src/pipeline/sync-runner.service.ts index d1b71f48f0..a0d80f53b1 100644 --- a/apps/chain-indexer/src/pipeline/sync-runner.service.ts +++ b/apps/chain-indexer/src/pipeline/sync-runner.service.ts @@ -6,6 +6,7 @@ import type { EnvConfig } from "@src/config/env.config"; import { LeadershipLostError, PgAdvisoryLeaderLock } from "@src/db/pg-advisory-leader-lock"; import { PgClientService } from "@src/db/pg-client.service"; import { Blocks, IndexerState } from "@src/db/schema"; +import { retryWithBackoff } from "@src/lib/retry-with-backoff/retry-with-backoff"; import { BlockCommitterService, SYNC_STREAM } from "@src/pipeline/block-committer.service"; import { BlockDecoderService } from "@src/pipeline/block-decoder.service"; import type { DecodedBlock } from "@src/pipeline/decoded-block"; @@ -108,24 +109,13 @@ export class SyncRunnerService { /** 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 LeadershipLostError; - - 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 retryWithBackoff(operation, { + maxAttempts: TRANSIENT_RETRY_MAX_ATTEMPTS, + baseDelayMs: TRANSIENT_RETRY_BASE_MS, + maxDelayMs: TRANSIENT_RETRY_MAX_MS, + shouldRethrow: error => this.#stopped || error instanceof ChainContinuityError || error instanceof LeadershipLostError, + onRetry: (error, attempt, delayMs) => this.#logger.warn({ ...logContext, attempt, delayMs, error }) + }); } async #syncBlock(height: number): Promise { From 760fd5da1d1d4ef51222364f39675ca509740ff0 Mon Sep 17 00:00:00 2001 From: Maxime Beauchamp <15185355+baktun14@users.noreply.github.com> Date: Tue, 11 Aug 2026 19:26:09 +0530 Subject: [PATCH 07/14] refactor(indexer): move continuity error and transient retry policy to shared modules --- .../src/pipeline/backfill-runner.service.ts | 16 ++++---------- .../src/pipeline/chain-continuity-error.ts | 2 ++ .../src/pipeline/sync-runner.service.ts | 20 ++++------------- .../src/pipeline/transient-retry.ts | 22 +++++++++++++++++++ 4 files changed, 32 insertions(+), 28 deletions(-) create mode 100644 apps/chain-indexer/src/pipeline/chain-continuity-error.ts create mode 100644 apps/chain-indexer/src/pipeline/transient-retry.ts diff --git a/apps/chain-indexer/src/pipeline/backfill-runner.service.ts b/apps/chain-indexer/src/pipeline/backfill-runner.service.ts index 0a94041fe7..030bdd88e8 100644 --- a/apps/chain-indexer/src/pipeline/backfill-runner.service.ts +++ b/apps/chain-indexer/src/pipeline/backfill-runner.service.ts @@ -2,15 +2,16 @@ import { eq } from "drizzle-orm"; import { inject, singleton } from "tsyringe"; import type { EnvConfig } from "@src/config/env.config"; -import { LeadershipLostError, PgAdvisoryLeaderLock } from "@src/db/pg-advisory-leader-lock"; +import { PgAdvisoryLeaderLock } from "@src/db/pg-advisory-leader-lock"; import { PgClientService } from "@src/db/pg-client.service"; 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 { ChainContinuityError } from "@src/pipeline/sync-runner.service"; +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"; @@ -21,9 +22,6 @@ import { RpcClientPool } from "@src/rpc/rpc-client-pool.service"; const BACKFILL_LEADER_LOCK_KEY = 7_431_002; const FETCH_RETRY_MAX_ATTEMPTS = 5; const FETCH_RETRY_BASE_MS = 1_000; -const TRANSIENT_RETRY_MAX_ATTEMPTS = 10; -const TRANSIENT_RETRY_BASE_MS = 1_000; -const TRANSIENT_RETRY_MAX_MS = 30_000; @singleton() export class BackfillRunnerService { @@ -186,13 +184,7 @@ export class BackfillRunnerService { /** 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 retryWithBackoff(operation, { - maxAttempts: TRANSIENT_RETRY_MAX_ATTEMPTS, - baseDelayMs: TRANSIENT_RETRY_BASE_MS, - maxDelayMs: TRANSIENT_RETRY_MAX_MS, - shouldRethrow: error => this.#stopped || error instanceof ChainContinuityError || error instanceof LeadershipLostError, - onRetry: (error, attempt, delayMs) => this.#logger.warn({ ...logContext, attempt, delayMs, error }) - }); + 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. */ 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/sync-runner.service.ts b/apps/chain-indexer/src/pipeline/sync-runner.service.ts index a0d80f53b1..67d233eb1f 100644 --- a/apps/chain-indexer/src/pipeline/sync-runner.service.ts +++ b/apps/chain-indexer/src/pipeline/sync-runner.service.ts @@ -3,13 +3,14 @@ import { setTimeout as delay } from "node:timers/promises"; import { inject, singleton } from "tsyringe"; import type { EnvConfig } from "@src/config/env.config"; -import { LeadershipLostError, PgAdvisoryLeaderLock } from "@src/db/pg-advisory-leader-lock"; +import { PgAdvisoryLeaderLock } from "@src/db/pg-advisory-leader-lock"; import { PgClientService } from "@src/db/pg-client.service"; import { Blocks, IndexerState } from "@src/db/schema"; -import { retryWithBackoff } from "@src/lib/retry-with-backoff/retry-with-backoff"; 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"; @@ -20,12 +21,6 @@ import { RpcClientPool } from "@src/rpc/rpc-client-pool.service"; const SYNC_LEADER_LOCK_KEY = 7_431_001; 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 {} @singleton() export class SyncRunnerService { @@ -107,15 +102,8 @@ export class SyncRunnerService { } } - /** 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 { - return await retryWithBackoff(operation, { - maxAttempts: TRANSIENT_RETRY_MAX_ATTEMPTS, - baseDelayMs: TRANSIENT_RETRY_BASE_MS, - maxDelayMs: TRANSIENT_RETRY_MAX_MS, - shouldRethrow: error => this.#stopped || error instanceof ChainContinuityError || error instanceof LeadershipLostError, - onRetry: (error, attempt, delayMs) => this.#logger.warn({ ...logContext, attempt, delayMs, error }) - }); + return await retryTransient(operation, { isStopped: () => this.#stopped, logger: this.#logger, logContext }); } async #syncBlock(height: number): Promise { 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..442fb2dbc5 --- /dev/null +++ b/apps/chain-indexer/src/pipeline/transient-retry.ts @@ -0,0 +1,22 @@ +import { LeadershipLostError } from "@src/db/pg-advisory-leader-lock"; +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 || error instanceof LeadershipLostError, + onRetry: (error, attempt, delayMs) => options.logger.warn({ ...options.logContext, attempt, delayMs, error }) + }); +} From c50ddd69c49b9a5cf87e920245c326387f73a2b1 Mon Sep 17 00:00:00 2001 From: Maxime Beauchamp <15185355+baktun14@users.noreply.github.com> Date: Tue, 11 Aug 2026 19:42:03 +0530 Subject: [PATCH 08/14] refactor(indexer): retry the continuity seed read like other transient reads --- .../src/pipeline/backfill-runner.service.ts | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/apps/chain-indexer/src/pipeline/backfill-runner.service.ts b/apps/chain-indexer/src/pipeline/backfill-runner.service.ts index 030bdd88e8..3c95dabb05 100644 --- a/apps/chain-indexer/src/pipeline/backfill-runner.service.ts +++ b/apps/chain-indexer/src/pipeline/backfill-runner.service.ts @@ -221,10 +221,14 @@ export class BackfillRunnerService { * 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.#db - .select() - .from(Blocks) - .where(eq(Blocks.height, startHeight - 1)); + 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; From 628c0bb84591cf82683f31c971d2ac6b235cabf8 Mon Sep 17 00:00:00 2001 From: Maxime Beauchamp <15185355+baktun14@users.noreply.github.com> Date: Tue, 11 Aug 2026 19:51:43 +0530 Subject: [PATCH 09/14] refactor(indexer): drop advisory-lock leader election for concurrency-safe writes Inserts were already idempotent; making the indexer_state checkpoint upsert monotonic (GREATEST) makes overlapping writers harmless, so the leader lock, its backend-pid liveness checks, and LeadershipLostError can all go. Single writers per role are enforced at the orchestration level instead. --- apps/chain-indexer/README.md | 6 +- .../src/db/pg-advisory-leader-lock.ts | 72 ------------------- .../pipeline/backfill-runner.service.spec.ts | 13 +--- .../src/pipeline/backfill-runner.service.ts | 27 +------ .../pipeline/block-committer.service.spec.ts | 19 ++++- .../src/pipeline/block-committer.service.ts | 11 ++- .../src/pipeline/sync-runner.service.ts | 20 ------ .../src/pipeline/transient-retry.ts | 3 +- 8 files changed, 33 insertions(+), 138 deletions(-) delete mode 100644 apps/chain-indexer/src/db/pg-advisory-leader-lock.ts diff --git a/apps/chain-indexer/README.md b/apps/chain-indexer/README.md index ccfd93e409..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), `backfill` (historical catch-up over an explicit height range), and a minimal `api` (healthz + status). `jobs` exits 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 @@ -44,7 +46,7 @@ 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; a separate advisory lock prevents two concurrent backfills. +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 diff --git a/apps/chain-indexer/src/db/pg-advisory-leader-lock.ts b/apps/chain-indexer/src/db/pg-advisory-leader-lock.ts deleted file mode 100644 index d6c3d27ca3..0000000000 --- a/apps/chain-indexer/src/db/pg-advisory-leader-lock.ts +++ /dev/null @@ -1,72 +0,0 @@ -import { setTimeout as delay } from "node:timers/promises"; -import type postgres from "postgres"; - -import type { LoggerService } from "@src/providers/logging.provider"; - -const ACQUIRE_RETRY_MS = 5_000; - -/** The advisory-lock session was replaced (e.g. a transparent driver reconnect), so another process may hold leadership. */ -export class LeadershipLostError extends Error {} - -/** - * Single-leader election on a session-scoped pg advisory lock. The lock dies silently with its - * session, so holders must re-verify leadership with assertHeld() before trusting it. - */ -export class PgAdvisoryLeaderLock { - readonly #client: postgres.Sql; - readonly #lockKey: number; - readonly #logger: LoggerService; - readonly #eventPrefix: string; - - #reserved: postgres.ReservedSql | undefined; - #leaderBackendPid: number | undefined; - - constructor(options: { client: postgres.Sql; lockKey: number; logger: LoggerService; eventPrefix: string }) { - this.#client = options.client; - this.#lockKey = options.lockKey; - this.#logger = options.logger; - this.#eventPrefix = options.eventPrefix; - } - - /** Spins until the lock is acquired or shouldAbort() returns true. */ - async acquire(shouldAbort: () => boolean): Promise { - this.#reserved = await this.#client.reserve(); - - while (!shouldAbort()) { - const [{ acquired }] = await this.#reserved`SELECT pg_try_advisory_lock(${this.#lockKey}) AS acquired`; - - if (acquired) { - const [{ pid }] = await this.#reserved`SELECT pg_backend_pid() AS pid`; - this.#leaderBackendPid = pid; - this.#logger.info({ event: `${this.#eventPrefix}_LEADERSHIP_ACQUIRED`, backendPid: pid }); - return; - } - - this.#logger.info({ event: `${this.#eventPrefix}_LEADERSHIP_WAITING` }); - await delay(ACQUIRE_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. */ - async assertHeld(): Promise { - if (!this.#reserved) { - throw new LeadershipLostError("Reserved advisory-lock connection is gone"); - } - - const [{ pid }] = await this.#reserved`SELECT pg_backend_pid() AS pid`; - - if (pid !== this.#leaderBackendPid) { - this.#logger.error({ event: `${this.#eventPrefix}_LEADERSHIP_LOST`, expectedBackendPid: this.#leaderBackendPid, actualBackendPid: pid }); - throw new LeadershipLostError(`Advisory-lock session changed (backend pid ${this.#leaderBackendPid} -> ${pid})`); - } - } - - /** release() may throw when the pg pool was ended first; the connection is already gone then, which is the goal. */ - release(): void { - try { - this.#reserved?.release(); - } catch (error) { - this.#logger.debug({ event: `${this.#eventPrefix}_LOCK_RELEASE_SKIPPED`, error }); - } - } -} diff --git a/apps/chain-indexer/src/pipeline/backfill-runner.service.spec.ts b/apps/chain-indexer/src/pipeline/backfill-runner.service.spec.ts index ea91930b6f..1bf37a5027 100644 --- a/apps/chain-indexer/src/pipeline/backfill-runner.service.spec.ts +++ b/apps/chain-indexer/src/pipeline/backfill-runner.service.spec.ts @@ -1,9 +1,8 @@ import { setTimeout as delay } from "node:timers/promises"; -import { describe, expect, it, vi } from "vitest"; +import { describe, expect, it } from "vitest"; import { mock } from "vitest-mock-extended"; import { envSchema } from "@src/config/env.config"; -import type { PgClientService } from "@src/db/pg-client.service"; import { Blocks, IndexerState } from "@src/db/schema"; import { BackfillRunnerService } from "@src/pipeline/backfill-runner.service"; import type { BlockCommitterService } from "@src/pipeline/block-committer.service"; @@ -142,14 +141,6 @@ describe(BackfillRunnerService.name, () => { BACKFILL_CONCURRENCY: String(input.concurrency ?? 10) }); - const reserved = Object.assign( - vi.fn((strings: TemplateStringsArray) => Promise.resolve(strings.join("?").includes("pg_try_advisory_lock") ? [{ acquired: true }] : [{ pid: 42 }])), - { release: vi.fn() } - ); - const pgClient = mock({ - client: mock({ reserve: vi.fn().mockResolvedValue(reserved) }) - }); - const dbFake = { select: () => ({ from: (table: unknown) => ({ @@ -199,7 +190,7 @@ describe(BackfillRunnerService.name, () => { const logger = mock(); - const runner = new BackfillRunnerService(pgClient, dbFake as unknown as ChainDatabase, pool, decoder, committer, config, logger); + const runner = new BackfillRunnerService(dbFake as unknown as ChainDatabase, pool, decoder, committer, config, logger); return { runner, committer, pool, logger, maxObservedConcurrency: () => maxActiveFetches }; } diff --git a/apps/chain-indexer/src/pipeline/backfill-runner.service.ts b/apps/chain-indexer/src/pipeline/backfill-runner.service.ts index 3c95dabb05..cc2fa7348e 100644 --- a/apps/chain-indexer/src/pipeline/backfill-runner.service.ts +++ b/apps/chain-indexer/src/pipeline/backfill-runner.service.ts @@ -2,8 +2,6 @@ import { eq } from "drizzle-orm"; import { inject, singleton } from "tsyringe"; import type { EnvConfig } from "@src/config/env.config"; -import { PgAdvisoryLeaderLock } from "@src/db/pg-advisory-leader-lock"; -import { PgClientService } from "@src/db/pg-client.service"; 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"; @@ -18,8 +16,6 @@ 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 backfill leader pg advisory lock; distinct from the sync leader key so a backfill never contends with live sync. */ -const BACKFILL_LEADER_LOCK_KEY = 7_431_002; const FETCH_RETRY_MAX_ATTEMPTS = 5; const FETCH_RETRY_BASE_MS = 1_000; @@ -31,13 +27,11 @@ export class BackfillRunnerService { readonly #committer: BlockCommitterService; readonly #config: EnvConfig; readonly #logger: LoggerService; - readonly #leaderLock: PgAdvisoryLeaderLock; #stopped = false; #lastHash: Buffer | null = null; constructor( - @inject(PgClientService) pgClient: PgClientService, @inject(CHAIN_DB) db: ChainDatabase, @inject(RpcClientPool) pool: RpcClientPool, @inject(BlockDecoderService) decoder: BlockDecoderService, @@ -52,12 +46,6 @@ export class BackfillRunnerService { this.#config = config; this.#logger = logger; this.#logger.setContext("BACKFILL"); - this.#leaderLock = new PgAdvisoryLeaderLock({ - client: pgClient.client, - lockKey: BACKFILL_LEADER_LOCK_KEY, - logger: this.#logger, - eventPrefix: "BACKFILL" - }); } async start(): Promise { @@ -74,7 +62,6 @@ export class BackfillRunnerService { async dispose(): Promise { this.#stopped = true; - this.#leaderLock.release(); } async #run(): Promise { @@ -84,12 +71,6 @@ export class BackfillRunnerService { throw new Error("BACKFILL_FROM_HEIGHT and BACKFILL_TO_HEIGHT are required for the backfill role"); } - await this.#leaderLock.acquire(() => this.#stopped); - - if (this.#stopped) { - return; - } - const stream = `backfill:${fromHeight}-${toHeight}`; const checkpointHeight = await this.#retryTransient(() => this.#getCheckpointHeight(stream), { event: "BACKFILL_CHECKPOINT_READ_RETRY" }); const tipHeight = await this.#retryTransient(() => this.#getTipHeight(), { event: "BACKFILL_TIP_FETCH_RETRY" }); @@ -148,13 +129,7 @@ export class BackfillRunnerService { if (batch.length >= this.#config.BACKFILL_BATCH_SIZE || height === endHeight) { const currentBatch = batch; - await this.#retryTransient( - async () => { - await this.#leaderLock.assertHeld(); - await this.#committer.commitBatch(currentBatch, { stream }); - }, - { event: "BACKFILL_COMMIT_RETRY", height } - ); + 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); batch = []; 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 aec0e650a4..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,3 +1,5 @@ +import type { SQL } from "drizzle-orm"; +import { PgDialect } from "drizzle-orm/pg-core"; import { describe, expect, it } from "vitest"; import { Blocks, IndexerState, Messages, MessageTypes } from "@src/db/schema"; @@ -76,6 +78,15 @@ describe(BlockCommitterService.name, () => { 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); @@ -91,6 +102,7 @@ describe(BlockCommitterService.name, () => { 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() ?? []) }) }), @@ -102,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(); + } }; } }), @@ -110,7 +125,7 @@ describe(BlockCommitterService.name, () => { }; const committer = new BlockCommitterService(dbFake as unknown as ChainDatabase); - return { committer, insertedRows }; + return { committer, insertedRows, conflictUpdates }; } function buildBlock(typeUrls: string[], height = 10): DecodedBlock { diff --git a/apps/chain-indexer/src/pipeline/block-committer.service.ts b/apps/chain-indexer/src/pipeline/block-committer.service.ts index c0d7824978..c587c6ff3f 100644 --- a/apps/chain-indexer/src/pipeline/block-committer.service.ts +++ b/apps/chain-indexer/src/pipeline/block-committer.service.ts @@ -1,4 +1,4 @@ -import { inArray } from "drizzle-orm"; +import { inArray, sql } from "drizzle-orm"; import chunk from "lodash/chunk"; import { inject, singleton } from "tsyringe"; @@ -25,7 +25,12 @@ export class BlockCommitterService { 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. */ + /** + * 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; @@ -87,7 +92,7 @@ export class BlockCommitterService { .values({ stream: options.stream, lastHeight, updatedAt: new Date() }) .onConflictDoUpdate({ target: IndexerState.stream, - set: { lastHeight, updatedAt: new Date() } + set: { lastHeight: sql`GREATEST(${IndexerState.lastHeight}, EXCLUDED.last_height)`, updatedAt: new Date() } }); }); } diff --git a/apps/chain-indexer/src/pipeline/sync-runner.service.ts b/apps/chain-indexer/src/pipeline/sync-runner.service.ts index 67d233eb1f..9783d31daa 100644 --- a/apps/chain-indexer/src/pipeline/sync-runner.service.ts +++ b/apps/chain-indexer/src/pipeline/sync-runner.service.ts @@ -3,8 +3,6 @@ import { setTimeout as delay } from "node:timers/promises"; import { inject, singleton } from "tsyringe"; import type { EnvConfig } from "@src/config/env.config"; -import { PgAdvisoryLeaderLock } from "@src/db/pg-advisory-leader-lock"; -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"; @@ -17,10 +15,7 @@ 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 PROGRESS_LOG_EVERY_BLOCKS = 100; -const LEADERSHIP_CHECK_EVERY_BLOCKS = 100; @singleton() export class SyncRunnerService { @@ -30,13 +25,11 @@ export class SyncRunnerService { readonly #committer: BlockCommitterService; readonly #config: EnvConfig; readonly #logger: LoggerService; - readonly #leaderLock: PgAdvisoryLeaderLock; #stopped = false; #lastHash: Buffer | null = null; constructor( - @inject(PgClientService) pgClient: PgClientService, @inject(CHAIN_DB) db: ChainDatabase, @inject(RpcClientPool) pool: RpcClientPool, @inject(BlockDecoderService) decoder: BlockDecoderService, @@ -51,7 +44,6 @@ export class SyncRunnerService { this.#config = config; this.#logger = logger; this.#logger.setContext("SYNC"); - this.#leaderLock = new PgAdvisoryLeaderLock({ client: pgClient.client, lockKey: SYNC_LEADER_LOCK_KEY, logger: this.#logger, eventPrefix: "SYNC" }); } async start(): Promise { @@ -68,21 +60,13 @@ export class SyncRunnerService { async dispose(): Promise { this.#stopped = true; - this.#leaderLock.release(); } async #run(): Promise { - await this.#leaderLock.acquire(() => this.#stopped); - - if (this.#stopped) { - return; - } - let nextHeight = await this.#resolveStartHeight(); this.#logger.info({ event: "SYNC_STARTED", network: this.#config.NETWORK, nextHeight }); while (!this.#stopped) { - await this.#leaderLock.assertHeld(); const tipHeight = await this.#retryTransient(() => this.#getTipHeight(), { event: "SYNC_TIP_FETCH_RETRY" }); if (nextHeight > tipHeight) { @@ -94,10 +78,6 @@ 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.#leaderLock.assertHeld(); - } } } } diff --git a/apps/chain-indexer/src/pipeline/transient-retry.ts b/apps/chain-indexer/src/pipeline/transient-retry.ts index 442fb2dbc5..16776f1d0f 100644 --- a/apps/chain-indexer/src/pipeline/transient-retry.ts +++ b/apps/chain-indexer/src/pipeline/transient-retry.ts @@ -1,4 +1,3 @@ -import { LeadershipLostError } from "@src/db/pg-advisory-leader-lock"; 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"; @@ -16,7 +15,7 @@ export function retryTransient( maxAttempts: MAX_ATTEMPTS, baseDelayMs: BASE_DELAY_MS, maxDelayMs: MAX_DELAY_MS, - shouldRethrow: error => options.isStopped() || error instanceof ChainContinuityError || error instanceof LeadershipLostError, + shouldRethrow: error => options.isStopped() || error instanceof ChainContinuityError, onRetry: (error, attempt, delayMs) => options.logger.warn({ ...options.logContext, attempt, delayMs, error }) }); } From a82f39a457b5fdbebb0862d29f83085d0deb4c10 Mon Sep 17 00:00:00 2001 From: Maxime Beauchamp <15185355+baktun14@users.noreply.github.com> Date: Tue, 11 Aug 2026 20:07:53 +0530 Subject: [PATCH 10/14] refactor(indexer): use the exported LoggerService type instead of a local alias --- apps/chain-indexer/src/index.ts | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/apps/chain-indexer/src/index.ts b/apps/chain-indexer/src/index.ts index 8ea313ac7f..cbcdf6b729 100644 --- a/apps/chain-indexer/src/index.ts +++ b/apps/chain-indexer/src/index.ts @@ -1,6 +1,7 @@ import "reflect-metadata"; import "@src/providers"; +import type { LoggerService } from "@akashnetwork/logging"; import { createOtelLogger } from "@akashnetwork/logging/otel"; import { container } from "tsyringe"; @@ -12,8 +13,6 @@ 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"; -type AppLogger = ReturnType; - export async function bootstrap(): Promise { const config = container.resolve(AppConfigService); const role = config.get("INDEXER_ROLE"); @@ -42,7 +41,7 @@ export async function bootstrap(): Promise { } /** Shared runner-role lifecycle: migrate, serve healthz, run to completion or fatal error (exit code 1), then shut the server down so the process can exit. */ -async function runRunnerBehindServer(resolveRunner: () => { start(): Promise }, fatalEvent: string, logger: AppLogger, port: number): Promise { +async function runRunnerBehindServer(resolveRunner: () => { start(): Promise }, fatalEvent: string, logger: LoggerService, port: number): Promise { await migrateDb(); const server = await startServer(createApp(), logger, process, { port }); From 5458736558f22516f3e2cc446310c57068561166 Mon Sep 17 00:00:00 2001 From: Maxime Beauchamp <15185355+baktun14@users.noreply.github.com> Date: Wed, 12 Aug 2026 01:11:05 +0530 Subject: [PATCH 11/14] fix(indexer): retry an interrupted backfill instead of completing the Job Addresses review findings on the backfill role: - A backfill stopped before finishing its range (SIGTERM mid-run) now throws RunnerInterruptedError so the process exits non-zero. Backfill runs as a K8s Job where exit 0 marks it Complete, so the previous clean exit silently left the range unfinished and unretried; it now resumes from its checkpoint on the next attempt. - Drop the unreachable prefetch fallback in the consume loop; the sliding window always has the current height inflight before it is awaited. - Back the retry delay with the global setTimeout so the fetch-retry test can advance it with fake timers instead of sleeping a real second. --- apps/chain-indexer/src/index.ts | 14 +++++++-- .../retry-with-backoff/retry-with-backoff.ts | 9 ++++-- .../pipeline/backfill-runner.service.spec.ts | 29 +++++++++++++++---- .../src/pipeline/backfill-runner.service.ts | 10 +++++-- .../src/pipeline/runner-interrupted-error.ts | 6 ++++ 5 files changed, 56 insertions(+), 12 deletions(-) create mode 100644 apps/chain-indexer/src/pipeline/runner-interrupted-error.ts diff --git a/apps/chain-indexer/src/index.ts b/apps/chain-indexer/src/index.ts index cbcdf6b729..2cd663191b 100644 --- a/apps/chain-indexer/src/index.ts +++ b/apps/chain-indexer/src/index.ts @@ -7,6 +7,7 @@ import { container } from "tsyringe"; import { createApp } from "@src/app"; 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"; @@ -40,7 +41,12 @@ export async function bootstrap(): Promise { } } -/** Shared runner-role lifecycle: migrate, serve healthz, run to completion or fatal error (exit code 1), then shut the server down so the process can exit. */ +/** + * 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 }); @@ -48,7 +54,11 @@ async function runRunnerBehindServer(resolveRunner: () => { start(): Promise(operation: () => Promise, options: } } } + +/** 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-runner.service.spec.ts b/apps/chain-indexer/src/pipeline/backfill-runner.service.spec.ts index 1bf37a5027..67099d78ea 100644 --- a/apps/chain-indexer/src/pipeline/backfill-runner.service.spec.ts +++ b/apps/chain-indexer/src/pipeline/backfill-runner.service.spec.ts @@ -1,5 +1,5 @@ import { setTimeout as delay } from "node:timers/promises"; -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import { mock } from "vitest-mock-extended"; import { envSchema } from "@src/config/env.config"; @@ -8,6 +8,7 @@ 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"; @@ -92,12 +93,20 @@ describe(BackfillRunnerService.name, () => { }); it("retries a failed fetch and still commits the block", async () => { - const { runner, committer, logger } = setup({ fromHeight: 1, toHeight: 2, failFetchOnceAtHeight: 2 }); + vi.useFakeTimers(); - await runner.start(); + try { + const { runner, committer, logger } = setup({ fromHeight: 1, toHeight: 2, failFetchOnceAtHeight: 2 }); - expect(committedHeights(committer)).toEqual([[1, 2]]); - expect(logger.warn).toHaveBeenCalledWith(expect.objectContaining({ event: "BACKFILL_FETCH_RETRY", height: 2, attempt: 1 })); + 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 () => { @@ -119,6 +128,16 @@ describe(BackfillRunnerService.name, () => { ); }); + 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]]); + }); + function setup(input: { fromHeight: number; toHeight: number; diff --git a/apps/chain-indexer/src/pipeline/backfill-runner.service.ts b/apps/chain-indexer/src/pipeline/backfill-runner.service.ts index cc2fa7348e..ee0403d040 100644 --- a/apps/chain-indexer/src/pipeline/backfill-runner.service.ts +++ b/apps/chain-indexer/src/pipeline/backfill-runner.service.ts @@ -9,6 +9,7 @@ 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"; @@ -53,11 +54,14 @@ export class BackfillRunnerService { await this.#run(); } catch (error) { if (this.#stopped) { - this.#logger.info({ event: "BACKFILL_STOPPED_DURING_SHUTDOWN" }); - return; + throw new RunnerInterruptedError("Backfill stopped before completing the range", { cause: error }); } throw error; } + + if (this.#stopped) { + throw new RunnerInterruptedError("Backfill stopped before completing the range"); + } } async dispose(): Promise { @@ -119,7 +123,7 @@ export class BackfillRunnerService { try { for (let height = startHeight; height <= endHeight && !this.#stopped; height++) { fillFetchWindow(); - const decoded = await (inflight.get(height) ?? this.#fetchAndDecode(height)); + const decoded = await inflight.get(height)!; inflight.delete(height); this.#verifyContinuity(decoded); 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 {} From 6b04f635908f1fb43bbaf87a56d1b7301d6a2e5b Mon Sep 17 00:00:00 2001 From: Maxime Beauchamp <15185355+baktun14@users.noreply.github.com> Date: Wed, 12 Aug 2026 01:12:13 +0530 Subject: [PATCH 12/14] fix(indexer): surface env validation errors without the DI wrapper A misconfigured role (e.g. a backfill Job missing BACKFILL_FROM/TO_HEIGHT) failed with a tsyringe "Cannot inject the dependency" wrapper around the ZodError plus an internal stack trace. Validate env eagerly at bootstrap and log a CONFIG_INVALID event listing the offending fields and messages instead. --- apps/chain-indexer/src/index.ts | 24 +++++++++++++++++++++++- 1 file changed, 23 insertions(+), 1 deletion(-) diff --git a/apps/chain-indexer/src/index.ts b/apps/chain-indexer/src/index.ts index 2cd663191b..db288b48b7 100644 --- a/apps/chain-indexer/src/index.ts +++ b/apps/chain-indexer/src/index.ts @@ -6,6 +6,7 @@ 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"; @@ -15,10 +16,16 @@ 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 port = config.get("PORT"); - const logger = createOtelLogger({ context: "APP" }); switch (role) { case "sync": { @@ -41,6 +48,21 @@ 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 From a4ae2477ed4c560439c6943c3c4ffe8f95ae4e30 Mon Sep 17 00:00:00 2001 From: Maxime Beauchamp <15185355+baktun14@users.noreply.github.com> Date: Wed, 12 Aug 2026 01:48:58 +0530 Subject: [PATCH 13/14] refactor(indexer): parallelize backfill startup reads and signal completion via return Fetch the checkpoint height and chain tip concurrently instead of serially, hoist the loop-invariant base height in the contiguity check, and have the backfill run return whether it completed so start() no longer re-reads the stopped flag. --- .../src/pipeline/backfill-runner.service.ts | 18 ++++++++++++------ .../src/pipeline/block-committer.service.ts | 4 +++- 2 files changed, 15 insertions(+), 7 deletions(-) diff --git a/apps/chain-indexer/src/pipeline/backfill-runner.service.ts b/apps/chain-indexer/src/pipeline/backfill-runner.service.ts index ee0403d040..0f5d2bc210 100644 --- a/apps/chain-indexer/src/pipeline/backfill-runner.service.ts +++ b/apps/chain-indexer/src/pipeline/backfill-runner.service.ts @@ -50,8 +50,10 @@ export class BackfillRunnerService { } async start(): Promise { + let completed: boolean; + try { - await this.#run(); + completed = await this.#run(); } catch (error) { if (this.#stopped) { throw new RunnerInterruptedError("Backfill stopped before completing the range", { cause: error }); @@ -59,7 +61,7 @@ export class BackfillRunnerService { throw error; } - if (this.#stopped) { + if (!completed) { throw new RunnerInterruptedError("Backfill stopped before completing the range"); } } @@ -68,7 +70,7 @@ export class BackfillRunnerService { this.#stopped = true; } - async #run(): Promise { + async #run(): Promise { const { BACKFILL_FROM_HEIGHT: fromHeight, BACKFILL_TO_HEIGHT: toHeight } = this.#config; if (fromHeight === undefined || toHeight === undefined) { @@ -76,8 +78,10 @@ export class BackfillRunnerService { } const stream = `backfill:${fromHeight}-${toHeight}`; - const checkpointHeight = await this.#retryTransient(() => this.#getCheckpointHeight(stream), { event: "BACKFILL_CHECKPOINT_READ_RETRY" }); - const tipHeight = await this.#retryTransient(() => this.#getTipHeight(), { event: "BACKFILL_TIP_FETCH_RETRY" }); + const [checkpointHeight, tipHeight] = await Promise.all([ + this.#retryTransient(() => this.#getCheckpointHeight(stream), { event: "BACKFILL_CHECKPOINT_READ_RETRY" }), + this.#retryTransient(() => this.#getTipHeight(), { event: "BACKFILL_TIP_FETCH_RETRY" }) + ]); const plan = planBackfill({ fromHeight, toHeight, checkpointHeight, tipHeight }); if (plan.kind === "invalid") { @@ -87,12 +91,14 @@ export class BackfillRunnerService { if (plan.kind === "already-complete") { this.#logger.info({ event: "BACKFILL_ALREADY_COMPLETE", stream, checkpointHeight }); - return; + 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 }); await this.#backfillRange(plan.startHeight, plan.endHeight, stream); + + return !this.#stopped; } /** diff --git a/apps/chain-indexer/src/pipeline/block-committer.service.ts b/apps/chain-indexer/src/pipeline/block-committer.service.ts index c587c6ff3f..72c01f8f42 100644 --- a/apps/chain-indexer/src/pipeline/block-committer.service.ts +++ b/apps/chain-indexer/src/pipeline/block-committer.service.ts @@ -99,8 +99,10 @@ export class BlockCommitterService { /** 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 = blocks[0].height + index; + const expectedHeight = baseHeight + index; if (block.height !== expectedHeight) { throw new Error(`Non-contiguous batch: expected height ${expectedHeight} at position ${index}, got ${block.height}`); From d6a4dd1c8c3081b4b8835ffa138051faf14cfaf1 Mon Sep 17 00:00:00 2001 From: Maxime Beauchamp <15185355+baktun14@users.noreply.github.com> Date: Wed, 12 Aug 2026 01:53:51 +0530 Subject: [PATCH 14/14] fix(indexer): track backfill completion by committed height, not the stopped flag A shutdown landing during the final commit left the range fully committed but reported it as interrupted, failing the Job and triggering a spurious retry. Completion is now derived from the last committed height reaching the range end. Also dedupes the identical getTipHeight helper into RpcClientPool so both runners share one implementation. --- .../pipeline/backfill-runner.service.spec.ts | 13 +++++++++- .../src/pipeline/backfill-runner.service.ts | 24 ++++++++++--------- .../src/pipeline/sync-runner.service.ts | 9 ++----- .../src/rpc/rpc-client-pool.service.spec.ts | 7 ++++++ .../src/rpc/rpc-client-pool.service.ts | 5 ++++ 5 files changed, 39 insertions(+), 19 deletions(-) diff --git a/apps/chain-indexer/src/pipeline/backfill-runner.service.spec.ts b/apps/chain-indexer/src/pipeline/backfill-runner.service.spec.ts index 67099d78ea..9d81537c25 100644 --- a/apps/chain-indexer/src/pipeline/backfill-runner.service.spec.ts +++ b/apps/chain-indexer/src/pipeline/backfill-runner.service.spec.ts @@ -138,6 +138,17 @@ describe(BackfillRunnerService.name, () => { 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; @@ -180,7 +191,7 @@ describe(BackfillRunnerService.name, () => { let maxActiveFetches = 0; let failedOnce = false; const pool = mock(); - pool.getStatus.mockResolvedValue({ sync_info: { latest_block_height: String(input.tipHeight ?? 1_000) } }); + pool.getTipHeight.mockResolvedValue(input.tipHeight ?? 1_000); pool.getBlock.mockImplementation(async height => { if (input.failFetchOnceAtHeight === height && !failedOnce) { failedOnce = true; diff --git a/apps/chain-indexer/src/pipeline/backfill-runner.service.ts b/apps/chain-indexer/src/pipeline/backfill-runner.service.ts index 0f5d2bc210..1961726277 100644 --- a/apps/chain-indexer/src/pipeline/backfill-runner.service.ts +++ b/apps/chain-indexer/src/pipeline/backfill-runner.service.ts @@ -80,7 +80,7 @@ export class BackfillRunnerService { const stream = `backfill:${fromHeight}-${toHeight}`; const [checkpointHeight, tipHeight] = await Promise.all([ this.#retryTransient(() => this.#getCheckpointHeight(stream), { event: "BACKFILL_CHECKPOINT_READ_RETRY" }), - this.#retryTransient(() => this.#getTipHeight(), { event: "BACKFILL_TIP_FETCH_RETRY" }) + this.#retryTransient(() => this.#pool.getTipHeight(), { event: "BACKFILL_TIP_FETCH_RETRY" }) ]); const plan = planBackfill({ fromHeight, toHeight, checkpointHeight, tipHeight }); @@ -96,9 +96,8 @@ export class BackfillRunnerService { await this.#seedContinuityHash(plan.startHeight, checkpointHeight !== null); this.#logger.info({ event: "BACKFILL_STARTED", network: this.#config.NETWORK, stream, startHeight: plan.startHeight, endHeight: plan.endHeight }); - await this.#backfillRange(plan.startHeight, plan.endHeight, stream); - return !this.#stopped; + return await this.#backfillRange(plan.startHeight, plan.endHeight, stream); } /** @@ -107,13 +106,18 @@ export class BackfillRunnerService { * 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 { + 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 = () => { @@ -142,6 +146,7 @@ export class BackfillRunnerService { 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 }); } @@ -150,8 +155,8 @@ export class BackfillRunnerService { await Promise.allSettled([...inflight.values()]); } - if (this.#stopped) { - return; + if (lastCommittedHeight < endHeight) { + return false; } const durationMs = Date.now() - startedAt; @@ -165,6 +170,8 @@ export class BackfillRunnerService { 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. */ @@ -231,9 +238,4 @@ export class BackfillRunnerService { const [state] = await this.#db.select().from(IndexerState).where(eq(IndexerState.stream, stream)); return state?.lastHeight ?? null; } - - async #getTipHeight(): Promise { - const status = await this.#pool.getStatus(); - return parseInt(status.sync_info.latest_block_height); - } } diff --git a/apps/chain-indexer/src/pipeline/sync-runner.service.ts b/apps/chain-indexer/src/pipeline/sync-runner.service.ts index 9783d31daa..294eea08a5 100644 --- a/apps/chain-indexer/src/pipeline/sync-runner.service.ts +++ b/apps/chain-indexer/src/pipeline/sync-runner.service.ts @@ -67,7 +67,7 @@ export class SyncRunnerService { this.#logger.info({ event: "SYNC_STARTED", network: this.#config.NETWORK, nextHeight }); while (!this.#stopped) { - 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); @@ -126,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/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}`); }