Add reorg threshold entry tolerance for multichain indexers

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()
   }

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<int>,
 }
 
-
 let configAddresses = (chainConfig: Config.chain): array<Internal.indexingAddress> => {
   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,

packages/envio/src/ChainState.resi

@@ -20,6 +20,7 @@ let make: (
   ~transactionStore: TransactionStore.t=?,
   ~chainDensity: option<float>=?,
   ~blockStore: BlockStore.t=?,
+  ~reorgThresholdReadyTolerance: int=?,
   ~logger: Pino.t,
 ) => t
 
@@ -74,7 +75,6 @@ let getProgressPercentage: t => float
 let chainDensity: t => option<float>
 let effectiveDensity: t => option<float>
 let hasReadyItem: t => bool
-let isReadyToEnterReorgThreshold: t => bool
 
 // Fetch control.
 let targetBlock: (t, ~chainTargetItems: float) => int

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<Internal.entityConfig>,
@@ -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,

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

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<MockIndexer.Source.itemMock> = 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()}])
+    },
+  )
+})

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,
         )