From d70916136fd6d3d8ba52b1bf478000a38f7e3ebd Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 21 Jul 2026 09:59:00 +0000
Subject: [PATCH 1/8] Fix multichain indexer never entering the reorg threshold

Reorg-threshold entry is a one-time whole-indexer transition that requires
every chain to satisfy isReadyToEnterReorgThreshold at the same batch check
(BatchProcessing / ChainFetching Array.every). The per-chain predicate was
instantaneous and non-monotonic: a chain that reached its lagged head with a
drained buffer was retracted the moment its head advanced by a block. On a live
multichain indexer some chain's head is always advancing, so the conjunction is
never observed for all chains at once and the indexer stays below the threshold
forever. The same race hits a single busy chain with many partitions, where the
slowest partition holds the frontier back past each batch boundary.

Latch the per-chain readiness on ChainState (reachedReorgThresholdEdge): once a
chain's frontier momentarily reaches its lagged head with an empty buffer, it
stays ready even after its head moves on. Entry then converges monotonically as
each chain latches. Safe because nothing above head - maxReorgDepth is processed
before the threshold, so a latched chain that later trails the head has still
not touched an un-reorg-safe block.

Add EnterReorgThreshold_test.res reproducing the multichain race through the
real indexer loop: it fails (envio_reorg_threshold stuck at 0) before the fix
and passes after.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TTuugjVrBRsj5Kb7EFRyVv
---
 packages/envio/src/ChainState.res             |  23 +++-
 packages/envio/src/FetchState.res             |   3 +
 .../test/EnterReorgThreshold_test.res         | 108 ++++++++++++++++++
 3 files changed, 132 insertions(+), 2 deletions(-)
 create mode 100644 scenarios/test_codegen/test/EnterReorgThreshold_test.res

diff --git a/packages/envio/src/ChainState.res b/packages/envio/src/ChainState.res
index 21b53cb69..dac78822a 100644
--- a/packages/envio/src/ChainState.res
+++ b/packages/envio/src/ChainState.res
@@ -39,6 +39,15 @@ 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,
+  // Latches true the first time the fetch frontier reaches this chain's lagged
+  // head. Reorg-threshold entry requires every chain to have reached its head at
+  // once, but a live chain's head keeps advancing, so an instantaneous "at head"
+  // check is almost never true for all chains at the same batch. Latching makes
+  // the per-chain readiness monotonic: a chain that momentarily reached its head
+  // stays ready even after its head moves on. Safe because nothing above
+  // head - maxReorgDepth is processed before the threshold, so a latched chain
+  // that later trails the head has still not touched an un-reorg-safe block.
+  mutable reachedReorgThresholdEdge: bool,
 }
 
 // Per-chain shape returned by the status API.
@@ -121,6 +130,7 @@ let make = (
     safeCheckpointTracking,
     transactionStore,
     blockStore,
+    reachedReorgThresholdEdge: false,
   }
 }
 
@@ -384,7 +394,16 @@ 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
+// Latches once the frontier reaches the lagged head with a drained buffer. See
+// reachedReorgThresholdEdge.
+let latchReorgThresholdEdge = (cs: t, fetchState: FetchState.t) => {
+  if !cs.reachedReorgThresholdEdge && fetchState->FetchState.isReadyToEnterReorgThreshold {
+    cs.reachedReorgThresholdEdge = true
+  }
+  cs.reachedReorgThresholdEdge
+}
+
+let isReadyToEnterReorgThreshold = (cs: t) => cs->latchReorgThresholdEdge(cs.fetchState)
 
 // 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
@@ -907,7 +926,7 @@ let isReadyToEnterReorgThresholdAfterBatch = (cs: t, ~batch: Batch.t) => {
   | Some(chainAfterBatch) => chainAfterBatch.fetchState
   | None => cs.fetchState
   }
-  fetchState->FetchState.isReadyToEnterReorgThreshold
+  cs->latchReorgThresholdEdge(fetchState)
 }
 
 // Commit the post-batch fetch frontier for a chain that progressed in the batch,
diff --git a/packages/envio/src/FetchState.res b/packages/envio/src/FetchState.res
index e1e4fd77c..6d14ea01c 100644
--- a/packages/envio/src/FetchState.res
+++ b/packages/envio/src/FetchState.res
@@ -2556,6 +2556,9 @@ let isFetchingAtHead = ({endBlock, blockLag, knownHeight} as fetchState: t) => {
     }
 }
 
+// Whether the chain's fetch frontier is at the lagged head with a drained
+// buffer — the moment it can cross into the reorg threshold. Latched by
+// ChainState so a later head advance doesn't retract it.
 let isReadyToEnterReorgThreshold = ({endBlock, blockLag, buffer, knownHeight} as fetchState: t) => {
   let bufferBlockNumber = fetchState->bufferBlockNumber
   knownHeight !== 0 &&
diff --git a/scenarios/test_codegen/test/EnterReorgThreshold_test.res b/scenarios/test_codegen/test/EnterReorgThreshold_test.res
new file mode 100644
index 000000000..e22f48f34
--- /dev/null
+++ b/scenarios/test_codegen/test/EnterReorgThreshold_test.res
@@ -0,0 +1,108 @@
+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`). Readiness is instantaneous and
+// non-monotonic: 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.
+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},
+        ],
+        ~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 by one block while it idles at its lagged head.
+      // Under the instantaneous predicate this un-readies chain A: its lagged
+      // head moves to 801, ahead of its frontier at 800. (Chain A cannot even
+      // re-query 801 yet — chain B holds the shared fetch budget — so it stays
+      // below its new lagged head.)
+      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 have reached their lagged heads at some point, so the
+      // indexer must be inside the reorg threshold — 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 once every chain has reached its lagged head",
+      ).toEqual([{value: "1", labels: Dict.make()}])
+    },
+  )
+})

From 96e6f6775ce652d554bd45931cb71ce89cc8dc75 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 21 Jul 2026 11:23:06 +0000
Subject: [PATCH 2/8] Add single-chain reorg-threshold latch test

Covers the single-chain manifestation of the entry bug at the unit level: a
chain that reaches its lagged head with a drained buffer must stay ready after
its head advances by a block. Without the latch the readiness predicate is
retracted on the head bump (true -> false); with it, readiness is monotonic.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TTuugjVrBRsj5Kb7EFRyVv
---
 .../test/lib_tests/ChainState_test.res        | 69 +++++++++++++++++++
 1 file changed, 69 insertions(+)

diff --git a/scenarios/test_codegen/test/lib_tests/ChainState_test.res b/scenarios/test_codegen/test/lib_tests/ChainState_test.res
index 855183bcf..f38e2894f 100644
--- a/scenarios/test_codegen/test/lib_tests/ChainState_test.res
+++ b/scenarios/test_codegen/test/lib_tests/ChainState_test.res
@@ -197,3 +197,72 @@ describe("ChainState chain density EMA (per batch)", () => {
     t.expect(cs->ChainState.chainDensity, ~message="full-window batch replaces").toEqual(Some(25.))
   })
 })
+
+describe("ChainState reorg-threshold readiness latch", () => {
+  // A chain with no address partitions, so bufferBlockNumber follows
+  // latestOnBlockBlockNumber (the fetch frontier) and readiness can be set
+  // directly. blockLag defaults to 0, so the lagged head is knownHeight.
+  let makeAtFrontier = (~knownHeight, ~frontier) => {
+    let base = FetchState.make(
+      ~onEventRegistrations=[],
+      ~contractConfigs=Dict.make(),
+      ~addresses=[],
+      ~startBlock=0,
+      ~endBlock=None,
+      ~maxAddrInPartition=3,
+      ~maxOnBlockBufferSize=10000,
+      ~chainId,
+      ~knownHeight=0,
+      ~onBlockRegistrations=[
+        {
+          Internal.index: 0,
+          name: "latch-test",
+          chainId,
+          startBlock: None,
+          endBlock: None,
+          interval: 1,
+          handler: "mock onBlock handler"->(
+            Utils.magic: string => Internal.onBlockArgs => promise<unit>
+          ),
+        },
+      ],
+    )
+    let fetchState = {...base, knownHeight, latestOnBlockBlockNumber: frontier, buffer: []}
+    ChainState.make(
+      ~chainConfig={...baseChainConfig, id: chainId},
+      ~fetchState,
+      ~indexingAddresses=IndexingAddresses.make(~contractConfigs=Dict.make(), ~addresses=[]),
+      ~sourceManager=SourceManager.make(
+        ~sources=[MockIndexer.Source.make([], ~chain=#1).source],
+        ~isRealtime=false,
+      ),
+      ~reorgDetection=ReorgDetection.make(
+        ~chainReorgCheckpoints=[],
+        ~maxReorgDepth=200,
+        ~shouldRollbackOnReorg=true,
+      ),
+      ~committedProgressBlockNumber=-1,
+      ~logger=Logging.getLogger(),
+    )
+  }
+
+  it("is not ready below the head, and stays ready after the head advances", t => {
+    let belowHead = makeAtFrontier(~knownHeight=1000, ~frontier=990)
+
+    let atHead = makeAtFrontier(~knownHeight=1000, ~frontier=1000)
+    let readyAtHead = atHead->ChainState.isReadyToEnterReorgThreshold
+    // A new block arrives after the chain reached its lagged head. The
+    // instantaneous predicate would retract readiness (frontier 1000 < head
+    // 1001); the latch keeps it ready so multichain entry can still converge.
+    atHead->ChainState.updateKnownHeight(~knownHeight=1001)
+
+    t.expect(
+      (
+        belowHead->ChainState.isReadyToEnterReorgThreshold,
+        readyAtHead,
+        atHead->ChainState.isReadyToEnterReorgThreshold,
+      ),
+      ~message="readiness latches on reaching the head and survives a later head advance",
+    ).toEqual((false, true, true))
+  })
+})

From 1104d93548baeb6186301f34ca87cf53d7b25bc2 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 21 Jul 2026 11:28:16 +0000
Subject: [PATCH 3/8] Trim reorg-threshold latch comments

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TTuugjVrBRsj5Kb7EFRyVv
---
 packages/envio/src/ChainState.res                 | 15 +++++----------
 packages/envio/src/FetchState.res                 |  5 ++---
 .../test/lib_tests/ChainState_test.res            |  5 ++---
 3 files changed, 9 insertions(+), 16 deletions(-)

diff --git a/packages/envio/src/ChainState.res b/packages/envio/src/ChainState.res
index dac78822a..2f7dfa2e5 100644
--- a/packages/envio/src/ChainState.res
+++ b/packages/envio/src/ChainState.res
@@ -39,14 +39,10 @@ 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,
-  // Latches true the first time the fetch frontier reaches this chain's lagged
-  // head. Reorg-threshold entry requires every chain to have reached its head at
-  // once, but a live chain's head keeps advancing, so an instantaneous "at head"
-  // check is almost never true for all chains at the same batch. Latching makes
-  // the per-chain readiness monotonic: a chain that momentarily reached its head
-  // stays ready even after its head moves on. Safe because nothing above
-  // head - maxReorgDepth is processed before the threshold, so a latched chain
-  // that later trails the head has still not touched an un-reorg-safe block.
+  // Latches true once the frontier first reaches the lagged head, keeping
+  // readiness monotonic so a later head advance can't retract it — otherwise the
+  // all-chains-ready check almost never lines up on a live indexer. Safe because
+  // nothing above head - maxReorgDepth is processed before the threshold.
   mutable reachedReorgThresholdEdge: bool,
 }
 
@@ -394,8 +390,7 @@ 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
-// Latches once the frontier reaches the lagged head with a drained buffer. See
-// reachedReorgThresholdEdge.
+// See reachedReorgThresholdEdge.
 let latchReorgThresholdEdge = (cs: t, fetchState: FetchState.t) => {
   if !cs.reachedReorgThresholdEdge && fetchState->FetchState.isReadyToEnterReorgThreshold {
     cs.reachedReorgThresholdEdge = true
diff --git a/packages/envio/src/FetchState.res b/packages/envio/src/FetchState.res
index 6d14ea01c..12b1e3768 100644
--- a/packages/envio/src/FetchState.res
+++ b/packages/envio/src/FetchState.res
@@ -2556,9 +2556,8 @@ let isFetchingAtHead = ({endBlock, blockLag, knownHeight} as fetchState: t) => {
     }
 }
 
-// Whether the chain's fetch frontier is at the lagged head with a drained
-// buffer — the moment it can cross into the reorg threshold. Latched by
-// ChainState so a later head advance doesn't retract it.
+// Frontier at the lagged head with a drained buffer — the moment the chain can
+// cross into the reorg threshold. Latched by ChainState.
 let isReadyToEnterReorgThreshold = ({endBlock, blockLag, buffer, knownHeight} as fetchState: t) => {
   let bufferBlockNumber = fetchState->bufferBlockNumber
   knownHeight !== 0 &&
diff --git a/scenarios/test_codegen/test/lib_tests/ChainState_test.res b/scenarios/test_codegen/test/lib_tests/ChainState_test.res
index f38e2894f..be55ffd8d 100644
--- a/scenarios/test_codegen/test/lib_tests/ChainState_test.res
+++ b/scenarios/test_codegen/test/lib_tests/ChainState_test.res
@@ -251,9 +251,8 @@ describe("ChainState reorg-threshold readiness latch", () => {
 
     let atHead = makeAtFrontier(~knownHeight=1000, ~frontier=1000)
     let readyAtHead = atHead->ChainState.isReadyToEnterReorgThreshold
-    // A new block arrives after the chain reached its lagged head. The
-    // instantaneous predicate would retract readiness (frontier 1000 < head
-    // 1001); the latch keeps it ready so multichain entry can still converge.
+    // New block after the chain reached its head: without the latch this would
+    // retract readiness (frontier 1000 < head 1001).
     atHead->ChainState.updateKnownHeight(~knownHeight=1001)
 
     t.expect(

From ff96497865b97082ce7f905715744b6ce8f8276c Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 21 Jul 2026 13:06:04 +0000
Subject: [PATCH 4/8] Add reorg-threshold ready tolerance (default 100,
 configurable)

Even with the latch, a chain must reach its lagged head exactly once for the
readiness predicate to fire; a head that advances between a chain's catch-up
query being sized and its response landing can keep deferring that instant.
Add a tolerance (in blocks) below the lagged head within which a chain still
counts as ready, absorbing that gap. Applied to the head comparison only, not
to endBlock (an exact target).

Threaded as a Config.t field defaulting to 100 in production and stored on
ChainState, so the latch reads it per chain. MockIndexer defaults it to 0 so
the small-scale test fixtures (head 300, maxReorgDepth 200) don't enter the
threshold before fetching; tests exercising the tolerance pass an explicit
value. Adds unit coverage (FetchState predicate, ChainState latch beyond the
tolerance) and an e2e test entering within the tolerance below the lagged head.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TTuugjVrBRsj5Kb7EFRyVv
---
 packages/envio/src/ChainState.res             |  9 +++-
 packages/envio/src/ChainState.resi            |  1 +
 packages/envio/src/Config.res                 |  5 +++
 packages/envio/src/FetchState.res             | 13 ++++--
 .../test/EnterReorgThreshold_test.res         | 41 +++++++++++++++++++
 .../test_codegen/test/helpers/MockIndexer.res |  6 +++
 .../test/lib_tests/ChainState_test.res        | 14 ++++---
 .../test/lib_tests/FetchState_test.res        | 26 ++++++++++++
 8 files changed, 104 insertions(+), 11 deletions(-)

diff --git a/packages/envio/src/ChainState.res b/packages/envio/src/ChainState.res
index 2f7dfa2e5..035a5dd5b 100644
--- a/packages/envio/src/ChainState.res
+++ b/packages/envio/src/ChainState.res
@@ -44,6 +44,7 @@ type t = {
   // all-chains-ready check almost never lines up on a live indexer. Safe because
   // nothing above head - maxReorgDepth is processed before the threshold.
   mutable reachedReorgThresholdEdge: bool,
+  reorgThresholdReadyTolerance: int,
 }
 
 // Per-chain shape returned by the status API.
@@ -105,6 +106,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)
@@ -127,6 +129,7 @@ let make = (
     transactionStore,
     blockStore,
     reachedReorgThresholdEdge: false,
+    reorgThresholdReadyTolerance,
   }
 }
 
@@ -288,6 +291,7 @@ let makeInternal = (
       ~ecosystem=config.ecosystem.name,
       ~shouldChecksum=!lowercaseAddresses,
     ),
+    ~reorgThresholdReadyTolerance=config.reorgThresholdReadyTolerance,
     ~logger,
   )
 }
@@ -392,7 +396,10 @@ let hasReadyItem = (cs: t) =>
   cs.fetchState->FetchState.isActivelyIndexing && cs.fetchState->FetchState.hasReadyItem
 // See reachedReorgThresholdEdge.
 let latchReorgThresholdEdge = (cs: t, fetchState: FetchState.t) => {
-  if !cs.reachedReorgThresholdEdge && fetchState->FetchState.isReadyToEnterReorgThreshold {
+  if (
+    !cs.reachedReorgThresholdEdge &&
+    fetchState->FetchState.isReadyToEnterReorgThreshold(~tolerance=cs.reorgThresholdReadyTolerance)
+  ) {
     cs.reachedReorgThresholdEdge = true
   }
   cs.reachedReorgThresholdEdge
diff --git a/packages/envio/src/ChainState.resi b/packages/envio/src/ChainState.resi
index 91b98a3d8..bc69ebc03 100644
--- a/packages/envio/src/ChainState.resi
+++ b/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
 
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<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,
diff --git a/packages/envio/src/FetchState.res b/packages/envio/src/FetchState.res
index 12b1e3768..bd1dccf0e 100644
--- a/packages/envio/src/FetchState.res
+++ b/packages/envio/src/FetchState.res
@@ -2556,14 +2556,19 @@ let isFetchingAtHead = ({endBlock, blockLag, knownHeight} as fetchState: t) => {
     }
 }
 
-// Frontier at the lagged head with a drained buffer — the moment the chain can
-// cross into the reorg threshold. Latched by ChainState.
-let isReadyToEnterReorgThreshold = ({endBlock, blockLag, buffer, knownHeight} as fetchState: t) => {
+// Frontier at (or within `tolerance` of) the lagged head with a drained buffer —
+// the moment the chain can cross into the reorg threshold. Latched by ChainState.
+// `tolerance` absorbs the head advancing between a chain catching up and this
+// check; it is not applied to endBlock, where reaching the end is an exact target.
+let isReadyToEnterReorgThreshold = (
+  ~tolerance=0,
+  {endBlock, blockLag, buffer, knownHeight} as fetchState: t,
+) => {
   let bufferBlockNumber = fetchState->bufferBlockNumber
   knownHeight !== 0 &&
   switch endBlock {
   | Some(endBlock) if bufferBlockNumber >= endBlock => true
-  | _ => bufferBlockNumber >= knownHeight - blockLag
+  | _ => bufferBlockNumber >= knownHeight - blockLag - tolerance
   } &&
   buffer->Utils.Array.isEmpty
 }
diff --git a/scenarios/test_codegen/test/EnterReorgThreshold_test.res b/scenarios/test_codegen/test/EnterReorgThreshold_test.res
index e22f48f34..d9fada43d 100644
--- a/scenarios/test_codegen/test/EnterReorgThreshold_test.res
+++ b/scenarios/test_codegen/test/EnterReorgThreshold_test.res
@@ -105,4 +105,45 @@ describe("PIN: multichain indexer enters the reorg threshold", () => {
       ).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 56feaf2e7..3e32418b8 100644
--- a/scenarios/test_codegen/test/helpers/MockIndexer.res
+++ b/scenarios/test_codegen/test/helpers/MockIndexer.res
@@ -393,6 +393,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
@@ -443,6 +447,7 @@ module Indexer = {
         enableRawEvents,
         chainMap,
         batchSize: batchSize->Option.getOr(config.batchSize),
+        reorgThresholdReadyTolerance,
       }
     }
 
@@ -662,6 +667,7 @@ module Indexer = {
           ~shouldRollbackOnReorg,
           ~reducedPollingInterval?,
           ~targetBufferSize?,
+          ~reorgThresholdReadyTolerance,
           ~onError,
           ~mapStorage,
         )
diff --git a/scenarios/test_codegen/test/lib_tests/ChainState_test.res b/scenarios/test_codegen/test/lib_tests/ChainState_test.res
index be55ffd8d..9e94482ff 100644
--- a/scenarios/test_codegen/test/lib_tests/ChainState_test.res
+++ b/scenarios/test_codegen/test/lib_tests/ChainState_test.res
@@ -246,14 +246,16 @@ describe("ChainState reorg-threshold readiness latch", () => {
     )
   }
 
-  it("is not ready below the head, and stays ready after the head advances", t => {
-    let belowHead = makeAtFrontier(~knownHeight=1000, ~frontier=990)
+  it("is not ready far below the head, and stays ready after the head jumps past the tolerance", t => {
+    // blockLag 0, tolerance 100: the ready cutoff is knownHeight - 100.
+    let belowHead = makeAtFrontier(~knownHeight=1000, ~frontier=850)
 
     let atHead = makeAtFrontier(~knownHeight=1000, ~frontier=1000)
     let readyAtHead = atHead->ChainState.isReadyToEnterReorgThreshold
-    // New block after the chain reached its head: without the latch this would
-    // retract readiness (frontier 1000 < head 1001).
-    atHead->ChainState.updateKnownHeight(~knownHeight=1001)
+    // Head jumps beyond the tolerance after the chain reached it. Without the
+    // latch this retracts readiness (frontier 1000 < cutoff 1100); the latch
+    // keeps it ready.
+    atHead->ChainState.updateKnownHeight(~knownHeight=1200)
 
     t.expect(
       (
@@ -261,7 +263,7 @@ describe("ChainState reorg-threshold readiness latch", () => {
         readyAtHead,
         atHead->ChainState.isReadyToEnterReorgThreshold,
       ),
-      ~message="readiness latches on reaching the head and survives a later head advance",
+      ~message="readiness latches on reaching the head and survives a later head jump",
     ).toEqual((false, true, true))
   })
 })
diff --git a/scenarios/test_codegen/test/lib_tests/FetchState_test.res b/scenarios/test_codegen/test/lib_tests/FetchState_test.res
index df321c54b..5d22d63ca 100644
--- a/scenarios/test_codegen/test/lib_tests/FetchState_test.res
+++ b/scenarios/test_codegen/test/lib_tests/FetchState_test.res
@@ -3507,6 +3507,32 @@ describe("FetchState.isReadyToEnterReorgThreshold", () => {
     t.expect(fs->FetchState.isReadyToEnterReorgThreshold).toBe(false)
   })
 
+  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("Returns false when queue is not empty even if thresholds are met", t => {
     // EndBlock reached but queue has items
     let (fs, _indexingAddresses) = makeFs(

From 0a9d9a3b2009c0337e2fe701b2e10cf90df27348 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 21 Jul 2026 13:19:24 +0000
Subject: [PATCH 5/8] Drop the reorg-threshold latch and the redundant entry
 check

The ready tolerance already keeps a chain ready across the head advancing
between catch-up and the entry check, so the monotonic latch is unnecessary;
readiness goes back to the plain tolerant predicate.

Also remove the reorg-threshold entry check from finishWaitingForNewBlock: it
already calls scheduleProcessing, and startProcessing always runs at least one
processNextBatch (even with no items or progress), which owns the entry
decision via the same predicate. The duplicated check added nothing.

Tests updated: the ChainState readiness test now asserts the tolerance boundary
instead of the latch; the multichain repro sets an explicit tolerance (it no
longer has the latch to absorb the head advance during the handoff).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TTuugjVrBRsj5Kb7EFRyVv
---
 packages/envio/src/ChainFetching.res          | 18 ++---------
 packages/envio/src/ChainState.res             | 22 ++------------
 .../test/EnterReorgThreshold_test.res         | 30 ++++++++++---------
 .../test/lib_tests/ChainState_test.res        | 27 +++++++----------
 4 files changed, 31 insertions(+), 66 deletions(-)

diff --git a/packages/envio/src/ChainFetching.res b/packages/envio/src/ChainFetching.res
index da43e3c3b..e0a39440c 100644
--- a/packages/envio/src/ChainFetching.res
+++ b/packages/envio/src/ChainFetching.res
@@ -333,22 +333,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 035a5dd5b..b299c9880 100644
--- a/packages/envio/src/ChainState.res
+++ b/packages/envio/src/ChainState.res
@@ -39,11 +39,6 @@ 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,
-  // Latches true once the frontier first reaches the lagged head, keeping
-  // readiness monotonic so a later head advance can't retract it — otherwise the
-  // all-chains-ready check almost never lines up on a live indexer. Safe because
-  // nothing above head - maxReorgDepth is processed before the threshold.
-  mutable reachedReorgThresholdEdge: bool,
   reorgThresholdReadyTolerance: int,
 }
 
@@ -128,7 +123,6 @@ let make = (
     safeCheckpointTracking,
     transactionStore,
     blockStore,
-    reachedReorgThresholdEdge: false,
     reorgThresholdReadyTolerance,
   }
 }
@@ -394,18 +388,8 @@ 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
-// See reachedReorgThresholdEdge.
-let latchReorgThresholdEdge = (cs: t, fetchState: FetchState.t) => {
-  if (
-    !cs.reachedReorgThresholdEdge &&
-    fetchState->FetchState.isReadyToEnterReorgThreshold(~tolerance=cs.reorgThresholdReadyTolerance)
-  ) {
-    cs.reachedReorgThresholdEdge = true
-  }
-  cs.reachedReorgThresholdEdge
-}
-
-let isReadyToEnterReorgThreshold = (cs: t) => cs->latchReorgThresholdEdge(cs.fetchState)
+let isReadyToEnterReorgThreshold = (cs: t) =>
+  cs.fetchState->FetchState.isReadyToEnterReorgThreshold(~tolerance=cs.reorgThresholdReadyTolerance)
 
 // 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
@@ -928,7 +912,7 @@ let isReadyToEnterReorgThresholdAfterBatch = (cs: t, ~batch: Batch.t) => {
   | Some(chainAfterBatch) => chainAfterBatch.fetchState
   | None => cs.fetchState
   }
-  cs->latchReorgThresholdEdge(fetchState)
+  fetchState->FetchState.isReadyToEnterReorgThreshold(~tolerance=cs.reorgThresholdReadyTolerance)
 }
 
 // Commit the post-batch fetch frontier for a chain that progressed in the batch,
diff --git a/scenarios/test_codegen/test/EnterReorgThreshold_test.res b/scenarios/test_codegen/test/EnterReorgThreshold_test.res
index d9fada43d..8650854e8 100644
--- a/scenarios/test_codegen/test/EnterReorgThreshold_test.res
+++ b/scenarios/test_codegen/test/EnterReorgThreshold_test.res
@@ -4,11 +4,13 @@ open Vitest
 //
 // 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`). Readiness is instantaneous and
-// non-monotonic: 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.
+// (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)
@@ -40,6 +42,7 @@ describe("PIN: multichain indexer enters the reorg threshold", () => {
           {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,
       )
@@ -72,11 +75,10 @@ describe("PIN: multichain indexer enters the reorg threshold", () => {
         ~message="cannot enter while chain B is still backfilling",
       ).toEqual([{value: "0", labels: Dict.make()}])
 
-      // Chain A's head advances by one block while it idles at its lagged head.
-      // Under the instantaneous predicate this un-readies chain A: its lagged
-      // head moves to 801, ahead of its frontier at 800. (Chain A cannot even
-      // re-query 801 yet — chain B holds the shared fetch budget — so it stays
-      // below its new lagged head.)
+      // 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)
@@ -96,12 +98,12 @@ describe("PIN: multichain indexer enters the reorg threshold", () => {
       )
       await indexerMock.getBatchWritePromise()
 
-      // Both chains have reached their lagged heads at some point, so the
-      // indexer must be inside the reorg threshold — even though chain A's head
-      // advanced past its frontier before chain B caught up.
+      // 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 once every chain has reached its lagged head",
+        ~message="the indexer enters the threshold with both chains within the tolerance of head",
       ).toEqual([{value: "1", labels: Dict.make()}])
     },
   )
diff --git a/scenarios/test_codegen/test/lib_tests/ChainState_test.res b/scenarios/test_codegen/test/lib_tests/ChainState_test.res
index 9e94482ff..df463a500 100644
--- a/scenarios/test_codegen/test/lib_tests/ChainState_test.res
+++ b/scenarios/test_codegen/test/lib_tests/ChainState_test.res
@@ -198,7 +198,7 @@ describe("ChainState chain density EMA (per batch)", () => {
   })
 })
 
-describe("ChainState reorg-threshold readiness latch", () => {
+describe("ChainState reorg-threshold readiness", () => {
   // A chain with no address partitions, so bufferBlockNumber follows
   // latestOnBlockBlockNumber (the fetch frontier) and readiness can be set
   // directly. blockLag defaults to 0, so the lagged head is knownHeight.
@@ -216,7 +216,7 @@ describe("ChainState reorg-threshold readiness latch", () => {
       ~onBlockRegistrations=[
         {
           Internal.index: 0,
-          name: "latch-test",
+          name: "tolerance-test",
           chainId,
           startBlock: None,
           endBlock: None,
@@ -246,24 +246,17 @@ describe("ChainState reorg-threshold readiness latch", () => {
     )
   }
 
-  it("is not ready far below the head, and stays ready after the head jumps past the tolerance", t => {
-    // blockLag 0, tolerance 100: the ready cutoff is knownHeight - 100.
-    let belowHead = makeAtFrontier(~knownHeight=1000, ~frontier=850)
-
-    let atHead = makeAtFrontier(~knownHeight=1000, ~frontier=1000)
-    let readyAtHead = atHead->ChainState.isReadyToEnterReorgThreshold
-    // Head jumps beyond the tolerance after the chain reached it. Without the
-    // latch this retracts readiness (frontier 1000 < cutoff 1100); the latch
-    // keeps it ready.
-    atHead->ChainState.updateKnownHeight(~knownHeight=1200)
+  it("applies the configured tolerance below the lagged head", t => {
+    // blockLag 0, default tolerance 100: the ready cutoff is knownHeight - 100.
+    let beyondTolerance = makeAtFrontier(~knownHeight=1000, ~frontier=850)
+    let withinTolerance = makeAtFrontier(~knownHeight=1000, ~frontier=950)
 
     t.expect(
       (
-        belowHead->ChainState.isReadyToEnterReorgThreshold,
-        readyAtHead,
-        atHead->ChainState.isReadyToEnterReorgThreshold,
+        beyondTolerance->ChainState.isReadyToEnterReorgThreshold,
+        withinTolerance->ChainState.isReadyToEnterReorgThreshold,
       ),
-      ~message="readiness latches on reaching the head and survives a later head jump",
-    ).toEqual((false, true, true))
+      ~message="ready within the tolerance below the lagged head, not beyond it",
+    ).toEqual((false, true))
   })
 })

From df575c7803f8623bcc4de49fecbddf2a39ab3147 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 21 Jul 2026 13:28:56 +0000
Subject: [PATCH 6/8] Require explicit reorg-threshold tolerance; drop
 test-only helper

Make FetchState.isReadyToEnterReorgThreshold's ~tolerance a required argument
instead of defaulting to 0, so callers state the tolerance explicitly.

Remove ChainState.isReadyToEnterReorgThreshold: after entry moved entirely into
processNextBatch (via isReadyToEnterReorgThresholdAfterBatch), the plain
per-chain accessor had no production caller and existed only for a unit test.
The tolerance is covered by FetchState predicate tests and the end-to-end
EnterReorgThreshold test, so the ChainState-level test and its helper are removed.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TTuugjVrBRsj5Kb7EFRyVv
---
 packages/envio/src/ChainState.res             |  2 -
 packages/envio/src/ChainState.resi            |  1 -
 packages/envio/src/FetchState.res             |  8 +--
 .../test/lib_tests/ChainState_test.res        | 63 -------------------
 .../test/lib_tests/FetchState_test.res        | 18 +++---
 5 files changed, 13 insertions(+), 79 deletions(-)

diff --git a/packages/envio/src/ChainState.res b/packages/envio/src/ChainState.res
index b299c9880..166c19d34 100644
--- a/packages/envio/src/ChainState.res
+++ b/packages/envio/src/ChainState.res
@@ -388,8 +388,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(~tolerance=cs.reorgThresholdReadyTolerance)
 
 // 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
diff --git a/packages/envio/src/ChainState.resi b/packages/envio/src/ChainState.resi
index bc69ebc03..38063aa25 100644
--- a/packages/envio/src/ChainState.resi
+++ b/packages/envio/src/ChainState.resi
@@ -64,7 +64,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
diff --git a/packages/envio/src/FetchState.res b/packages/envio/src/FetchState.res
index bd1dccf0e..7c252e976 100644
--- a/packages/envio/src/FetchState.res
+++ b/packages/envio/src/FetchState.res
@@ -2557,11 +2557,11 @@ let isFetchingAtHead = ({endBlock, blockLag, knownHeight} as fetchState: t) => {
 }
 
 // Frontier at (or within `tolerance` of) the lagged head with a drained buffer —
-// the moment the chain can cross into the reorg threshold. Latched by ChainState.
-// `tolerance` absorbs the head advancing between a chain catching up and this
-// check; it is not applied to endBlock, where reaching the end is an exact target.
+// the moment the chain can cross into the reorg threshold. `tolerance` absorbs
+// the head advancing between a chain catching up and this check; it is not
+// applied to endBlock, where reaching the end is an exact target.
 let isReadyToEnterReorgThreshold = (
-  ~tolerance=0,
+  ~tolerance,
   {endBlock, blockLag, buffer, knownHeight} as fetchState: t,
 ) => {
   let bufferBlockNumber = fetchState->bufferBlockNumber
diff --git a/scenarios/test_codegen/test/lib_tests/ChainState_test.res b/scenarios/test_codegen/test/lib_tests/ChainState_test.res
index df463a500..855183bcf 100644
--- a/scenarios/test_codegen/test/lib_tests/ChainState_test.res
+++ b/scenarios/test_codegen/test/lib_tests/ChainState_test.res
@@ -197,66 +197,3 @@ describe("ChainState chain density EMA (per batch)", () => {
     t.expect(cs->ChainState.chainDensity, ~message="full-window batch replaces").toEqual(Some(25.))
   })
 })
-
-describe("ChainState reorg-threshold readiness", () => {
-  // A chain with no address partitions, so bufferBlockNumber follows
-  // latestOnBlockBlockNumber (the fetch frontier) and readiness can be set
-  // directly. blockLag defaults to 0, so the lagged head is knownHeight.
-  let makeAtFrontier = (~knownHeight, ~frontier) => {
-    let base = FetchState.make(
-      ~onEventRegistrations=[],
-      ~contractConfigs=Dict.make(),
-      ~addresses=[],
-      ~startBlock=0,
-      ~endBlock=None,
-      ~maxAddrInPartition=3,
-      ~maxOnBlockBufferSize=10000,
-      ~chainId,
-      ~knownHeight=0,
-      ~onBlockRegistrations=[
-        {
-          Internal.index: 0,
-          name: "tolerance-test",
-          chainId,
-          startBlock: None,
-          endBlock: None,
-          interval: 1,
-          handler: "mock onBlock handler"->(
-            Utils.magic: string => Internal.onBlockArgs => promise<unit>
-          ),
-        },
-      ],
-    )
-    let fetchState = {...base, knownHeight, latestOnBlockBlockNumber: frontier, buffer: []}
-    ChainState.make(
-      ~chainConfig={...baseChainConfig, id: chainId},
-      ~fetchState,
-      ~indexingAddresses=IndexingAddresses.make(~contractConfigs=Dict.make(), ~addresses=[]),
-      ~sourceManager=SourceManager.make(
-        ~sources=[MockIndexer.Source.make([], ~chain=#1).source],
-        ~isRealtime=false,
-      ),
-      ~reorgDetection=ReorgDetection.make(
-        ~chainReorgCheckpoints=[],
-        ~maxReorgDepth=200,
-        ~shouldRollbackOnReorg=true,
-      ),
-      ~committedProgressBlockNumber=-1,
-      ~logger=Logging.getLogger(),
-    )
-  }
-
-  it("applies the configured tolerance below the lagged head", t => {
-    // blockLag 0, default tolerance 100: the ready cutoff is knownHeight - 100.
-    let beyondTolerance = makeAtFrontier(~knownHeight=1000, ~frontier=850)
-    let withinTolerance = makeAtFrontier(~knownHeight=1000, ~frontier=950)
-
-    t.expect(
-      (
-        beyondTolerance->ChainState.isReadyToEnterReorgThreshold,
-        withinTolerance->ChainState.isReadyToEnterReorgThreshold,
-      ),
-      ~message="ready within the tolerance below the lagged head, not beyond it",
-    ).toEqual((false, true))
-  })
-})
diff --git a/scenarios/test_codegen/test/lib_tests/FetchState_test.res b/scenarios/test_codegen/test/lib_tests/FetchState_test.res
index 5d22d63ca..670166d8d 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,7 +3504,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("With a tolerance, is ready within it below head - blockLag, false just beyond it", t => {
@@ -3553,7 +3553,7 @@ describe("FetchState.isReadyToEnterReorgThreshold", () => {
       ~knownHeight=10,
     )
     let fsWithQueue = fs->FetchState.updateInternal(~mutItems=[mockEvent(~blockNumber=6)])
-    t.expect(fsWithQueue->FetchState.isReadyToEnterReorgThreshold).toBe(false)
+    t.expect(fsWithQueue->FetchState.isReadyToEnterReorgThreshold(~tolerance=0)).toBe(false)
   })
 
   it("Returns true when the queue is empty and threshold is more than current block height", t => {
@@ -3574,7 +3574,7 @@ describe("FetchState.isReadyToEnterReorgThreshold", () => {
       ~blockLag=200,
       ~knownHeight=10,
     )
-    t.expect(fs->FetchState.isReadyToEnterReorgThreshold).toBe(true)
+    t.expect(fs->FetchState.isReadyToEnterReorgThreshold(~tolerance=0)).toBe(true)
   })
 })
 

From fe15c1308a4471be90eb8a3563d38ad7662ecc02 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 21 Jul 2026 13:56:55 +0000
Subject: [PATCH 7/8] Enter reorg threshold on no processable items, not an
 empty buffer

The readiness check required the whole merged buffer to be empty. With many
partitions tracking a moving head, faster partitions leave items stuck above
bufferBlockNumber (the slowest partition's frontier) behind the gap, so the
buffer is never empty and the chain never enters the threshold.

Require bufferReadyCount == 0 instead: only processable items (at or below the
frontier) defer entry. Items stuck above the frontier are reorg-safe (all
<= head - blockLag before the threshold) and are processed in-threshold once
their partition catches up. Single-partition behaviour is unchanged, since with
no gap every buffered item is processable.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TTuugjVrBRsj5Kb7EFRyVv
---
 packages/envio/src/FetchState.res             | 15 +++---
 .../test/lib_tests/FetchState_test.res        | 49 +++++++++++--------
 2 files changed, 37 insertions(+), 27 deletions(-)

diff --git a/packages/envio/src/FetchState.res b/packages/envio/src/FetchState.res
index 7c252e976..0151c7650 100644
--- a/packages/envio/src/FetchState.res
+++ b/packages/envio/src/FetchState.res
@@ -2556,13 +2556,16 @@ let isFetchingAtHead = ({endBlock, blockLag, knownHeight} as fetchState: t) => {
     }
 }
 
-// Frontier at (or within `tolerance` of) the lagged head with a drained buffer —
-// the moment the chain can cross into the reorg threshold. `tolerance` absorbs
-// the head advancing between a chain catching up and this check; it is not
-// applied to endBlock, where reaching the end is an exact target.
+// 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; it is not applied to endBlock, where reaching the end is an exact
+// target. 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, buffer, knownHeight} as fetchState: t,
+  {endBlock, blockLag, knownHeight} as fetchState: t,
 ) => {
   let bufferBlockNumber = fetchState->bufferBlockNumber
   knownHeight !== 0 &&
@@ -2570,7 +2573,7 @@ let isReadyToEnterReorgThreshold = (
   | Some(endBlock) if bufferBlockNumber >= endBlock => true
   | _ => bufferBlockNumber >= knownHeight - blockLag - tolerance
   } &&
-  buffer->Utils.Array.isEmpty
+  fetchState->bufferReadyCount == 0
 }
 
 // Lower progress percentage = further behind = higher priority. Progress is
diff --git a/scenarios/test_codegen/test/lib_tests/FetchState_test.res b/scenarios/test_codegen/test/lib_tests/FetchState_test.res
index 670166d8d..00db377cf 100644
--- a/scenarios/test_codegen/test/lib_tests/FetchState_test.res
+++ b/scenarios/test_codegen/test/lib_tests/FetchState_test.res
@@ -3533,27 +3533,34 @@ describe("FetchState.isReadyToEnterReorgThreshold", () => {
     t.expect((isReady(~knownHeight=209), isReady(~knownHeight=210))).toEqual((true, false))
   })
 
-  it("Returns false when queue is not empty even if thresholds are met", t => {
-    // EndBlock reached but queue has items
-    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,
-    )
-    let fsWithQueue = fs->FetchState.updateInternal(~mutItems=[mockEvent(~blockNumber=6)])
-    t.expect(fsWithQueue->FetchState.isReadyToEnterReorgThreshold(~tolerance=0)).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 => {

From 09434c82fba60ac20b0536865299c09b6702704e Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 21 Jul 2026 14:16:44 +0000
Subject: [PATCH 8/8] Don't apply reorg-threshold tolerance to a finite
 endBlock

The tolerance leaked into the endBlock branch: a chain whose endBlock sat within
`tolerance` of the lagged head entered the reorg threshold before reaching the
endBlock, processing the last blocks with rollback history unnecessarily.

Apply the tolerance only when the moving head is the target. A finite endBlock at
or below the lagged head is an exact target (no moving head to absorb), so it is
reached without tolerance; an endBlock beyond the lagged head still tracks the
head like a chain with no endBlock, so it isn't regressed.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TTuugjVrBRsj5Kb7EFRyVv
---
 packages/envio/src/FetchState.res             | 10 +++++---
 .../test/lib_tests/FetchState_test.res        | 25 +++++++++++++++++++
 2 files changed, 31 insertions(+), 4 deletions(-)

diff --git a/packages/envio/src/FetchState.res b/packages/envio/src/FetchState.res
index cfa854a17..9b033e8db 100644
--- a/packages/envio/src/FetchState.res
+++ b/packages/envio/src/FetchState.res
@@ -2537,8 +2537,9 @@ let isFetchingAtHead = ({endBlock, blockLag, 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; it is not applied to endBlock, where reaching the end is an exact
-// target. Uses bufferReadyCount, not an empty buffer: items stuck above the
+// 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 = (
@@ -2546,10 +2547,11 @@ let isReadyToEnterReorgThreshold = (
   {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 - tolerance
+  | Some(endBlock) if endBlock <= laggedHead => bufferBlockNumber >= endBlock
+  | _ => bufferBlockNumber >= laggedHead - tolerance
   } &&
   fetchState->bufferReadyCount == 0
 }
diff --git a/scenarios/test_codegen/test/lib_tests/FetchState_test.res b/scenarios/test_codegen/test/lib_tests/FetchState_test.res
index 00db377cf..0c12f5f9b 100644
--- a/scenarios/test_codegen/test/lib_tests/FetchState_test.res
+++ b/scenarios/test_codegen/test/lib_tests/FetchState_test.res
@@ -3533,6 +3533,31 @@ describe("FetchState.isReadyToEnterReorgThreshold", () => {
     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=[
+        {
+          Internal.address: mockAddress0,
+          contractName: "Gravatar",
+          registrationBlock: -1,
+        },
+      ],
+      // latestFullyFetchedBlock = startBlock - 1 = 59
+      ~startBlock=60,
+      ~endBlock=Some(100),
+      ~maxAddrInPartition=3,
+      ~maxOnBlockBufferSize=targetBufferSize,
+      ~chainId,
+      ~blockLag=0,
+      ~knownHeight=150,
+    )
+    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 => {