From 34ed9d6de7653a50b143b164e3095c18107a9d69 Mon Sep 17 00:00:00 2001 From: elina-chertova Date: Sat, 20 Jun 2026 02:28:54 +0000 Subject: [PATCH] dump-cli: recover from transient ForkException on finalized dump instead of crashing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The finalized archive dump crashes the whole process on a ForkException. Because finalized blocks are immutable, such an exception is never a real reorg — it is transient upstream/provider inconsistency (e.g. a load-balanced RPC backend that jumped to a recent snapshot and serves a gapped/forked view of an already-finalized slot range). The process exits, k8s restarts it, it resumes from the last persisted chunk and hits the same inconsistency, and the dumper crash-loops indefinitely, stalling the writer (no new chunks). appendRawBlocks only persists complete chunks, so the in-flight (unflushed) buffer is discarded on throw and the resume point is re-derived from the last persisted chunk on the next call. Wrap the finalized append loop so a ForkException triggers an in-process retry from that durable boundary — identical to a clean process restart, but without crash-looping. Recovery is bounded by progress: if retries stop advancing the last written block, the exception is re-thrown so a genuine, persistent divergence still surfaces as a hard failure rather than being retried forever. Detection is structural (isSqdForkException marker) to avoid adding a dependency on the data-source package. Co-Authored-By: Claude Opus 4.8 --- util/util-internal-dump-cli/src/dumper.ts | 113 +++++++++++++++++++--- 1 file changed, 97 insertions(+), 16 deletions(-) diff --git a/util/util-internal-dump-cli/src/dumper.ts b/util/util-internal-dump-cli/src/dumper.ts index 945a3e47e..004fa007d 100644 --- a/util/util-internal-dump-cli/src/dumper.ts +++ b/util/util-internal-dump-cli/src/dumper.ts @@ -1,6 +1,6 @@ import {createLogger, Logger} from '@subsquid/logger' import {RpcClient} from '@subsquid/rpc-client' -import {assertNotNull, def, last, runProgram, Throttler, waitDrain} from '@subsquid/util-internal' +import {assertNotNull, def, last, runProgram, Throttler, wait, waitDrain} from '@subsquid/util-internal' import { ArchiveLayout, checkShorHashMatch, @@ -263,23 +263,52 @@ export abstract class Dumper this.ingest(nextBlock, prevHash), - range: this.range(), - chunkSize: chunkSize * 1024 * 1024, - onSuccessWrite: ctx => { - const blockHeight = ctx.blockRange.to.number; - prometheus.setLastWrittenBlock(blockHeight); - - const cachedTimestamp = this.timestampCache.get(blockHeight); - if (cachedTimestamp) { - prometheus.setProcessedBlockMetrics(cachedTimestamp); - this.log().debug(`Processed block ${blockHeight} at ${cachedTimestamp}`); - } else { - this.log().warn(`No cached timestamp available for height ${blockHeight}`); + + // The finalized dump stream can throw a ForkException when the + // upstream RPC serves inconsistent finalized data (e.g. a + // load-balanced provider whose backend jumped to a recent + // snapshot and now reports a gapped/forked view of an already + // finalized slot range). Finalized blocks are immutable, so + // this is never a real reorg — it's transient provider noise. + // appendRawBlocks only persists complete chunks, so the + // in-flight (unflushed) buffer is discarded on throw and the + // resume point is re-derived from the last persisted chunk on + // the next call: identical to a process restart, but without + // crash-looping the dumper. We only give up (re-throw) when + // retries stop making progress, so a genuine, persistent + // archive/chain divergence still surfaces as a hard failure. + let lastWrittenBlock = -1 + await appendWithForkRecovery( + () => archive.appendRawBlocks({ + blocks: (nextBlock, prevHash) => this.ingest(nextBlock, prevHash), + range: this.range(), + chunkSize: chunkSize * 1024 * 1024, + onSuccessWrite: ctx => { + const blockHeight = ctx.blockRange.to.number; + lastWrittenBlock = blockHeight; + prometheus.setLastWrittenBlock(blockHeight); + + const cachedTimestamp = this.timestampCache.get(blockHeight); + if (cachedTimestamp) { + prometheus.setProcessedBlockMetrics(cachedTimestamp); + this.log().debug(`Processed block ${blockHeight} at ${cachedTimestamp}`); + } else { + this.log().warn(`No cached timestamp available for height ${blockHeight}`); + } } + }), + () => lastWrittenBlock, + async ({message, lastWrittenBlock, stuckRetries}) => { + this.log().warn( + {reason: message, lastWrittenBlock, attempt: stuckRetries}, + 'finalized dump stream hit a chain-continuity error. Finalized data ' + + 'is immutable, so this is transient upstream/provider inconsistency ' + + 'rather than a real reorg. Discarding the in-flight buffer and ' + + 'retrying from the last persisted chunk.' + ) + await wait(Math.min(30_000, 1000 * 2 ** stuckRetries)) } - }) + ) } }, err => { if (err instanceof ErrorMessage) { @@ -297,3 +326,55 @@ export class ErrorMessage extends Error { super(msg) } } + + +/** + * A ForkException raised by a data source (it carries the `isSqdForkException` + * marker). Detected structurally so this package doesn't need to depend on the + * data-source package that defines the class. + */ +export function isForkException(err: unknown): boolean { + return err instanceof Error && (err as {isSqdForkException?: boolean}).isSqdForkException === true +} + + +/** + * Run `append` (a finalized archive append loop), recovering from transient + * ForkExceptions instead of crashing. + * + * On a ForkException the in-flight, not-yet-persisted buffer is dropped and + * `append` is invoked again — it re-derives its resume point from the last + * persisted chunk, so retrying is equivalent to a clean process restart. + * + * Recovery is bounded by progress, not a fixed attempt count: as long as a + * retry manages to persist a higher block than before, the counter resets. + * Only when `maxStuckRetries` consecutive retries fail to advance the last + * written block do we re-throw — that signals a genuine, persistent divergence + * (not transient provider noise) which must surface as a hard failure rather + * than be silently retried forever. + */ +export async function appendWithForkRecovery( + append: () => Promise, + getLastWrittenBlock: () => number, + onForkRetry: (info: {message: string; lastWrittenBlock: number; stuckRetries: number}) => Promise, + maxStuckRetries = 10, +): Promise { + let lastProgress = -1 + let stuckRetries = 0 + while (true) { + try { + return await append() + } catch (err: unknown) { + if (!isForkException(err)) throw err + let lastWrittenBlock = getLastWrittenBlock() + if (lastWrittenBlock > lastProgress) { + stuckRetries = 0 + } else { + stuckRetries += 1 + } + lastProgress = lastWrittenBlock + if (stuckRetries > maxStuckRetries) throw err + await onForkRetry({message: (err as Error).message, lastWrittenBlock, stuckRetries}) + } + } +}