diff --git a/packages/envio/src/ChainFetching.res b/packages/envio/src/ChainFetching.res index 2c159cf13..40e0c49c5 100644 --- a/packages/envio/src/ChainFetching.res +++ b/packages/envio/src/ChainFetching.res @@ -319,22 +319,8 @@ let finishWaitingForNewBlock = ( let chainState = state->IndexerState.getChainState(~chain) chainState->ChainState.updateKnownHeight(~knownHeight) - let isBelowReorgThreshold = - !(state->IndexerState.isInReorgThreshold) && - (state->IndexerState.config).shouldRollbackOnReorg - let shouldEnterReorgThreshold = - isBelowReorgThreshold && - state - ->IndexerState.chainStates - ->Dict.valuesToArray - ->Array.every(cs => { - cs->ChainState.isReadyToEnterReorgThreshold - }) - - // Kick processing in case there are block handlers to run. - if shouldEnterReorgThreshold { - IndexerState.enterReorgThreshold(state) - } + // No reorg-threshold check here: scheduleProcessing always runs at least one + // processNextBatch (even with no items), which owns the entry decision. scheduleFetch() scheduleProcessing() } diff --git a/packages/envio/src/ChainState.res b/packages/envio/src/ChainState.res index 23d7f86fe..cb97dc047 100644 --- a/packages/envio/src/ChainState.res +++ b/packages/envio/src/ChainState.res @@ -39,6 +39,7 @@ type t = { // Holds this chain's blocks (kept in Rust) keyed by block number. Same merge / // prune / rollback lifecycle as the transaction store. blockStore: BlockStore.t, + reorgThresholdReadyTolerance: int, // --- Per-chain metric counters, rendered by Metrics at scrape time. // Floats: cumulative counters outgrow int32. --- mutable blockRangeFetchSeconds: float, @@ -52,7 +53,6 @@ type t = { mutable progressLatencyMs: option, } - let configAddresses = (chainConfig: Config.chain): array => { let addresses = [] chainConfig.contracts->Array.forEach(contract => { @@ -94,6 +94,7 @@ let make = ( ~transactionStore=TransactionStore.make(~ecosystem=Ecosystem.Evm, ~shouldChecksum=false), ~chainDensity=None, ~blockStore=BlockStore.make(~ecosystem=Ecosystem.Evm, ~shouldChecksum=false), + ~reorgThresholdReadyTolerance=100, ~logger: Pino.t, ): t => { validateOnEventRegistrations(~chainId=chainConfig.id, onEventRegistrations) @@ -115,6 +116,7 @@ let make = ( safeCheckpointTracking, transactionStore, blockStore, + reorgThresholdReadyTolerance, blockRangeFetchSeconds: 0., blockRangeParseSeconds: 0., blockRangeFetchCount: 0., @@ -285,6 +287,7 @@ let makeInternal = ( ~ecosystem=config.ecosystem.name, ~shouldChecksum=!lowercaseAddresses, ), + ~reorgThresholdReadyTolerance=config.reorgThresholdReadyTolerance, ~logger, ) } @@ -397,7 +400,6 @@ let recordReorgDetected = (cs: t, ~blockNumber) => { let setRollbackTargetBlock = (cs: t, ~blockNumber) => cs.rollbackTargetBlock = Some(blockNumber) - // Fetch-frontier reads. The FetchState is owned here; callers go through these // rather than reaching into it. let knownHeight = (cs: t) => cs.fetchState.knownHeight @@ -409,7 +411,6 @@ let getProgressPercentage = (cs: t) => cs.fetchState->FetchState.getProgressPerc let chainDensity = (cs: t) => cs.chainDensity let hasReadyItem = (cs: t) => cs.fetchState->FetchState.isActivelyIndexing && cs.fetchState->FetchState.hasReadyItem -let isReadyToEnterReorgThreshold = (cs: t) => cs.fetchState->FetchState.isReadyToEnterReorgThreshold // Mark queries as in flight and reserve their estimated size against the shared // buffer budget in one step, so the counter stays in sync with the pending @@ -950,7 +951,7 @@ let isReadyToEnterReorgThresholdAfterBatch = (cs: t, ~batch: Batch.t) => { | Some(chainAfterBatch) => chainAfterBatch.fetchState | None => cs.fetchState } - fetchState->FetchState.isReadyToEnterReorgThreshold + fetchState->FetchState.isReadyToEnterReorgThreshold(~tolerance=cs.reorgThresholdReadyTolerance) } // Commit the post-batch fetch frontier for a chain that progressed in the batch, diff --git a/packages/envio/src/ChainState.resi b/packages/envio/src/ChainState.resi index c239db9df..b603943ee 100644 --- a/packages/envio/src/ChainState.resi +++ b/packages/envio/src/ChainState.resi @@ -20,6 +20,7 @@ let make: ( ~transactionStore: TransactionStore.t=?, ~chainDensity: option=?, ~blockStore: BlockStore.t=?, + ~reorgThresholdReadyTolerance: int=?, ~logger: Pino.t, ) => t @@ -74,7 +75,6 @@ let getProgressPercentage: t => float let chainDensity: t => option let effectiveDensity: t => option let hasReadyItem: t => bool -let isReadyToEnterReorgThreshold: t => bool // Fetch control. let targetBlock: (t, ~chainTargetItems: float) => int diff --git a/packages/envio/src/Config.res b/packages/envio/src/Config.res index 64478ba09..165df6d87 100644 --- a/packages/envio/src/Config.res +++ b/packages/envio/src/Config.res @@ -79,6 +79,10 @@ type t = { enableRawEvents: bool, maxAddrInPartition: int, batchSize: int, + // Slack (in blocks) below the lagged head within which a chain still counts as + // ready to enter the reorg threshold, absorbing head advances between catch-up + // and the entry check. Overridable in tests. + reorgThresholdReadyTolerance: int, lowercaseAddresses: bool, isDev: bool, userEntitiesByName: dict, @@ -997,6 +1001,7 @@ let fromPublic = (publicConfigJson: JSON.t) => { ecosystem, maxAddrInPartition, batchSize: publicConfig["fullBatchSize"]->Option.getOr(5000), + reorgThresholdReadyTolerance: 100, lowercaseAddresses, isDev: publicConfig["isDev"]->Option.getOr(false), userEntitiesByName, diff --git a/packages/envio/src/FetchState.res b/packages/envio/src/FetchState.res index 433ef26c6..9b033e8db 100644 --- a/packages/envio/src/FetchState.res +++ b/packages/envio/src/FetchState.res @@ -2534,14 +2534,26 @@ let isFetchingAtHead = ({endBlock, blockLag, knownHeight} as fetchState: t) => { } } -let isReadyToEnterReorgThreshold = ({endBlock, blockLag, buffer, knownHeight} as fetchState: t) => { +// Frontier at (or within `tolerance` of) the lagged head with no processable +// items left — the moment the chain can cross into the reorg threshold. +// `tolerance` absorbs the head advancing between a chain catching up and this +// check, so it applies only when the (moving) head is the target: a finite +// endBlock at or below the lagged head is an exact target and is reached without +// tolerance. Uses bufferReadyCount, not an empty buffer: items stuck above the +// frontier behind a lagging partition's gap are reorg-safe (all <= head - +// blockLag) and must not defer entry, or a many-partition chain never enters. +let isReadyToEnterReorgThreshold = ( + ~tolerance, + {endBlock, blockLag, knownHeight} as fetchState: t, +) => { let bufferBlockNumber = fetchState->bufferBlockNumber + let laggedHead = knownHeight - blockLag knownHeight !== 0 && switch endBlock { - | Some(endBlock) if bufferBlockNumber >= endBlock => true - | _ => bufferBlockNumber >= knownHeight - blockLag + | Some(endBlock) if endBlock <= laggedHead => bufferBlockNumber >= endBlock + | _ => bufferBlockNumber >= laggedHead - tolerance } && - buffer->Utils.Array.isEmpty + fetchState->bufferReadyCount == 0 } // Lower progress percentage = further behind = higher priority. Progress is diff --git a/scenarios/test_codegen/test/EnterReorgThreshold_test.res b/scenarios/test_codegen/test/EnterReorgThreshold_test.res new file mode 100644 index 000000000..8650854e8 --- /dev/null +++ b/scenarios/test_codegen/test/EnterReorgThreshold_test.res @@ -0,0 +1,151 @@ +open Vitest + +// Repro: a multichain indexer never enters the reorg threshold. +// +// Entry is a one-time whole-indexer transition that requires EVERY chain to +// satisfy `isReadyToEnterReorgThreshold` at the same batch-completion check +// (BatchProcessing.res `Array.every`). Without a tolerance, a chain that reached +// its lagged head is un-readied the moment its head advances by a block. With +// more than one live chain the head of some chain is always advancing, so the +// conjunction is never observed and the indexer sits below the threshold forever. +// The reorg-threshold ready tolerance closes this: a chain stays ready while +// within `tolerance` blocks of the lagged head, so a small head advance during +// the cross-chain handoff no longer defers entry. +describe("PIN: multichain indexer enters the reorg threshold", () => { + let waitNewHeightPoll = async (sourceMock: MockIndexer.Source.t, ~after) => { + let attempts = ref(0) + while sourceMock.getHeightOrThrowCalls->Array.length <= after && attempts.contents < 1000 { + attempts := attempts.contents + 1 + await Utils.delay(0) + } + if sourceMock.getHeightOrThrowCalls->Array.length <= after { + JsError.throwWithMessage("Timed out waiting for a new getHeightOrThrow poll") + } + } + + Async.it( + "a chain whose head advances after reaching its lagged head still lets the indexer enter the threshold", + async t => { + // Two chains, each lagging maxReorgDepth (200) below head before the + // threshold. Head starts at 1000, so the pre-threshold head is 800. + let chainA = MockIndexer.Source.make( + [#getHeightOrThrow, #getItemsOrThrow, #getBlockHashes], + ~chain=#100, + ) + let chainB = MockIndexer.Source.make( + [#getHeightOrThrow, #getItemsOrThrow, #getBlockHashes], + ~chain=#1337, + ) + + let indexerMock = await MockIndexer.Indexer.make( + ~chains=[ + {chain: #100, sourceConfig: Config.CustomSources([chainA.source]), maxReorgDepth: 200, blockLag: 0}, + {chain: #1337, sourceConfig: Config.CustomSources([chainB.source]), maxReorgDepth: 200, blockLag: 0}, + ], + ~reorgThresholdReadyTolerance=100, + ~reducedPollingInterval=1, + ~targetBufferSize=100, + ) + await Utils.delay(0) + + let initialHeightPolls = chainA.getHeightOrThrowCalls->Array.length + chainA.resolveGetHeightOrThrow(1000) + chainB.resolveGetHeightOrThrow(1000) + await Utils.delay(0) + await Utils.delay(0) + + // Chain A wins the initial priority tie and fetches to its pre-threshold + // head (block 800), seeding a density signal from its events. + await MockIndexer.Helper.waitItemsQuery(chainA) + t.expect( + chainA.getItemsOrThrowCalls->Array.map(call => call.payload["fromBlock"]), + ~message="chain A first fetches from its start block", + ).toEqual([1]) + let densitySeed: array = Array.fromInitializer(~length=100, i => { + MockIndexer.Source.blockNumber: 1 + i * 3, + logIndex: 0, + }) + chainA.resolveGetItemsOrThrow(densitySeed, ~latestFetchedBlockNumber=800, ~knownHeight=1000) + await indexerMock.getBatchWritePromise() + + // Chain A is now at its lagged head with an empty buffer — momentarily + // ready. Chain B has not responded, so the entry check fails here. + t.expect( + await indexerMock.metric("envio_reorg_threshold"), + ~message="cannot enter while chain B is still backfilling", + ).toEqual([{value: "0", labels: Dict.make()}]) + + // Chain A's head advances while it idles at its lagged head, so its frontier + // (800) now trails the lagged head (801). Without a tolerance this would + // un-ready chain A and defer entry; the 100-block tolerance keeps it ready. + // (Chain A cannot re-query 801 yet — chain B holds the shared fetch budget.) + await waitNewHeightPoll(chainA, ~after=initialHeightPolls) + chainA.resolveGetHeightOrThrow(1001) + await Utils.delay(0) + await Utils.delay(0) + + // Chain B now reaches its own pre-threshold head and produces a batch, + // triggering the whole-indexer entry check. + await MockIndexer.Helper.waitItemsQuery(chainB) + t.expect( + chainB.getItemsOrThrowCalls->Array.map(call => call.payload["fromBlock"]), + ~message="chain B first fetches from its start block", + ).toEqual([1]) + chainB.resolveGetItemsOrThrow( + [{MockIndexer.Source.blockNumber: 800, logIndex: 0}], + ~latestFetchedBlockNumber=800, + ~knownHeight=1000, + ) + await indexerMock.getBatchWritePromise() + + // Both chains are within the tolerance of their lagged heads, so the indexer + // enters — even though chain A's head advanced past its frontier before + // chain B caught up. + t.expect( + await indexerMock.metric("envio_reorg_threshold"), + ~message="the indexer enters the threshold with both chains within the tolerance of head", + ).toEqual([{value: "1", labels: Dict.make()}]) + }, + ) + + Async.it( + "enters while still within the configured tolerance of the lagged head", + async t => { + let source = MockIndexer.Source.make( + [#getHeightOrThrow, #getItemsOrThrow, #getBlockHashes], + ~chain=#1337, + ) + let indexerMock = await MockIndexer.Indexer.make( + ~chains=[ + {chain: #1337, sourceConfig: Config.CustomSources([source.source]), maxReorgDepth: 200, blockLag: 0}, + ], + ~reorgThresholdReadyTolerance=100, + ~reducedPollingInterval=1, + ~targetBufferSize=100, + ) + await Utils.delay(0) + + source.resolveGetHeightOrThrow(1000) + await MockIndexer.Helper.waitItemsQuery(source) + // Pre-threshold blockLag is 200, so the chain queries up to block 800. + t.expect( + source.getItemsOrThrowCalls->Array.map(call => call.payload["toBlock"]), + ~message="pre-threshold query stops at the lagged head", + ).toEqual([Some(800)]) + + // Respond 50 blocks short of the lagged head (750 < 800) — within the + // 100-block tolerance, so the chain enters despite not reaching 800 exactly. + source.resolveGetItemsOrThrow( + [{MockIndexer.Source.blockNumber: 750, logIndex: 0}], + ~latestFetchedBlockNumber=750, + ~knownHeight=1000, + ) + await indexerMock.getBatchWritePromise() + + t.expect( + await indexerMock.metric("envio_reorg_threshold"), + ~message="enters within the tolerance below the lagged head", + ).toEqual([{value: "1", labels: Dict.make()}]) + }, + ) +}) diff --git a/scenarios/test_codegen/test/helpers/MockIndexer.res b/scenarios/test_codegen/test/helpers/MockIndexer.res index 11072bd82..5fd903523 100644 --- a/scenarios/test_codegen/test/helpers/MockIndexer.res +++ b/scenarios/test_codegen/test/helpers/MockIndexer.res @@ -420,6 +420,10 @@ module Indexer = { ~shouldRollbackOnReorg=true, ~reducedPollingInterval=?, ~targetBufferSize=?, + // Defaults to 0 (not the production 100) so the small-scale fixtures here + // don't enter the reorg threshold before fetching. Tests exercising the + // tolerance pass an explicit value. + ~reorgThresholdReadyTolerance=0, // Lets regression tests surface fatal errors without terminating the Vitest worker. ~onError=?, // Lets a test intercept storage methods, e.g. to stall writeBatch and @@ -470,6 +474,7 @@ module Indexer = { enableRawEvents, chainMap, batchSize: batchSize->Option.getOr(config.batchSize), + reorgThresholdReadyTolerance, } } @@ -732,6 +737,7 @@ module Indexer = { ~shouldRollbackOnReorg, ~reducedPollingInterval?, ~targetBufferSize?, + ~reorgThresholdReadyTolerance, ~onError, ~mapStorage, ) diff --git a/scenarios/test_codegen/test/lib_tests/FetchState_test.res b/scenarios/test_codegen/test/lib_tests/FetchState_test.res index df321c54b..0c12f5f9b 100644 --- a/scenarios/test_codegen/test/lib_tests/FetchState_test.res +++ b/scenarios/test_codegen/test/lib_tests/FetchState_test.res @@ -3386,14 +3386,14 @@ describe("FetchState.sortForBatch", () => { describe("FetchState.isReadyToEnterReorgThreshold", () => { it("Returns false when we just started the indexer and it has knownHeight=0", t => { let (fetchState, _indexingAddresses) = makeInitial() - t.expect({...fetchState, knownHeight: 0}->FetchState.isReadyToEnterReorgThreshold).toBe(false) + t.expect({...fetchState, knownHeight: 0}->FetchState.isReadyToEnterReorgThreshold(~tolerance=0)).toBe(false) }) it( "Returns false when we just started the indexer and it has knownHeight=0, while start block is more than 0 + reorg threshold", t => { let (fetchState, _indexingAddresses) = makeInitial(~startBlock=6000) - t.expect({...fetchState, knownHeight: 0}->FetchState.isReadyToEnterReorgThreshold).toBe(false) + t.expect({...fetchState, knownHeight: 0}->FetchState.isReadyToEnterReorgThreshold(~tolerance=0)).toBe(false) }, ) @@ -3416,7 +3416,7 @@ describe("FetchState.isReadyToEnterReorgThreshold", () => { ~blockLag=0, ~knownHeight=10, ) - t.expect(fs->FetchState.isReadyToEnterReorgThreshold).toBe(true) + t.expect(fs->FetchState.isReadyToEnterReorgThreshold(~tolerance=0)).toBe(true) }) it("Returns false when endBlock not reached and below head - blockLag", t => { @@ -3438,7 +3438,7 @@ describe("FetchState.isReadyToEnterReorgThreshold", () => { ~blockLag=10, ~knownHeight=60, ) - t.expect(fs->FetchState.isReadyToEnterReorgThreshold).toBe(false) + t.expect(fs->FetchState.isReadyToEnterReorgThreshold(~tolerance=0)).toBe(false) }) it("Returns true when endBlock not reached but latest >= head - blockLag", t => { @@ -3460,7 +3460,7 @@ describe("FetchState.isReadyToEnterReorgThreshold", () => { ~blockLag=10, ~knownHeight=59, ) - t.expect(fs->FetchState.isReadyToEnterReorgThreshold).toBe(true) + t.expect(fs->FetchState.isReadyToEnterReorgThreshold(~tolerance=0)).toBe(true) }) it("Returns true when no endBlock and latest >= head - blockLag (boundary)", t => { @@ -3482,7 +3482,7 @@ describe("FetchState.isReadyToEnterReorgThreshold", () => { ~blockLag=10, ~knownHeight=60, ) - t.expect(fs->FetchState.isReadyToEnterReorgThreshold).toBe(true) + t.expect(fs->FetchState.isReadyToEnterReorgThreshold(~tolerance=0)).toBe(true) }) it("Returns false when no endBlock and latest < head - blockLag", t => { @@ -3504,11 +3504,39 @@ describe("FetchState.isReadyToEnterReorgThreshold", () => { ~blockLag=10, ~knownHeight=60, ) - t.expect(fs->FetchState.isReadyToEnterReorgThreshold).toBe(false) + t.expect(fs->FetchState.isReadyToEnterReorgThreshold(~tolerance=0)).toBe(false) }) - it("Returns false when queue is not empty even if thresholds are met", t => { - // EndBlock reached but queue has items + it("With a tolerance, is ready within it below head - blockLag, false just beyond it", t => { + let isReady = (~knownHeight) => { + let (fs, _indexingAddresses) = makeFs( + ~onEventRegistrations=[baseEventConfig, baseEventConfig2], + ~addresses=[ + { + Internal.address: mockAddress0, + contractName: "Gravatar", + registrationBlock: -1, + }, + ], + // latestFullyFetchedBlock = startBlock - 1 = 99 + ~startBlock=100, + ~endBlock=None, + ~maxAddrInPartition=3, + ~maxOnBlockBufferSize=targetBufferSize, + ~chainId, + ~blockLag=10, + ~knownHeight, + ) + fs->FetchState.isReadyToEnterReorgThreshold(~tolerance=100) + } + // frontier 99, ready cutoff = knownHeight - blockLag - tolerance: 209 -> 99, 210 -> 100 + t.expect((isReady(~knownHeight=209), isReady(~knownHeight=210))).toEqual((true, false)) + }) + + it("Does not apply the tolerance to a finite endBlock at or below the lagged head", t => { + // endBlock 100 sits below the lagged head (150), so it is an exact target. + // frontier 59 is within the tolerance of the lagged head (cutoff 50) but below + // the endBlock, so entry must wait for the endBlock rather than enter early. let (fs, _indexingAddresses) = makeFs( ~onEventRegistrations=[baseEventConfig, baseEventConfig2], ~addresses=[ @@ -3518,16 +3546,46 @@ describe("FetchState.isReadyToEnterReorgThreshold", () => { registrationBlock: -1, }, ], - ~startBlock=6, - ~endBlock=Some(5), + // latestFullyFetchedBlock = startBlock - 1 = 59 + ~startBlock=60, + ~endBlock=Some(100), ~maxAddrInPartition=3, ~maxOnBlockBufferSize=targetBufferSize, ~chainId, ~blockLag=0, - ~knownHeight=10, + ~knownHeight=150, ) - let fsWithQueue = fs->FetchState.updateInternal(~mutItems=[mockEvent(~blockNumber=6)]) - t.expect(fsWithQueue->FetchState.isReadyToEnterReorgThreshold).toBe(false) + t.expect(fs->FetchState.isReadyToEnterReorgThreshold(~tolerance=100)).toBe(false) + }) + + it("Blocks on processable items, but not on items stuck above the frontier", t => { + // frontier (bufferBlockNumber) = 5, endBlock 5 reached. + let readyWithItemAt = itemBlockNumber => { + let (fs, _indexingAddresses) = makeFs( + ~onEventRegistrations=[baseEventConfig, baseEventConfig2], + ~addresses=[ + { + Internal.address: mockAddress0, + contractName: "Gravatar", + registrationBlock: -1, + }, + ], + ~startBlock=6, + ~endBlock=Some(5), + ~maxAddrInPartition=3, + ~maxOnBlockBufferSize=targetBufferSize, + ~chainId, + ~blockLag=0, + ~knownHeight=10, + ) + fs + ->FetchState.updateInternal(~mutItems=[mockEvent(~blockNumber=itemBlockNumber)]) + ->FetchState.isReadyToEnterReorgThreshold(~tolerance=0) + } + // A processable item (<= frontier 5) still needs draining; an item stuck + // above the frontier (as behind a lagging partition's gap) is reorg-safe and + // must not defer entry. + t.expect((readyWithItemAt(5), readyWithItemAt(6))).toEqual((false, true)) }) it("Returns true when the queue is empty and threshold is more than current block height", t => { @@ -3548,7 +3606,7 @@ describe("FetchState.isReadyToEnterReorgThreshold", () => { ~blockLag=200, ~knownHeight=10, ) - t.expect(fs->FetchState.isReadyToEnterReorgThreshold).toBe(true) + t.expect(fs->FetchState.isReadyToEnterReorgThreshold(~tolerance=0)).toBe(true) }) })