Don't query reorgable blocks before all chains reach reorg threshold (#680)

* Add is in reorg threshold metric

* Set block lag when outside of reorg threshold

* Refactor reorg threshold entry

* Enter reorg threshold when there are no events

* Fix tests

Author: DZakh

codegenerator/cli/npm/envio/src/FetchState.res

@@ -68,8 +68,8 @@ type t = {
   // Fields computed by updateInternal
   latestFullyFetchedBlock: blockNumberAndTimestamp,
   // How much blocks behind the head we should query
-  // Added for the purpose of avoiding reorg handling
-  blockLag: option<int>,
+  // Needed to query before entering reorg threshold
+  blockLag: int,
   //Items ordered from latest to earliest
   queue: array<Internal.eventItem>,
 }
@@ -229,6 +229,7 @@ let updateInternal = (
   ~dcsToStore=fetchState.dcsToStore,
   ~currentBlockHeight=?,
   ~queue=fetchState.queue,
+  ~blockLag=fetchState.blockLag,
 ): t => {
   let firstPartition = partitions->Js.Array2.unsafe_get(0)
   let latestFullyFetchedBlock = ref(firstPartition.latestFetchedBlock)
@@ -284,7 +285,7 @@ let updateInternal = (
     latestFullyFetchedBlock,
     indexingContracts,
     dcsToStore,
-    blockLag: fetchState.blockLag,
+    blockLag,
     queue,
   }
 }
@@ -717,13 +718,12 @@ let getNextQuery = (
   ~currentBlockHeight,
   ~stateId,
 ) => {
-  if currentBlockHeight === 0 {
+  let headBlock = currentBlockHeight - blockLag
+  if headBlock <= 0 {
     WaitingForNewBlock
   } else if concurrencyLimit === 0 {
     ReachedMaxConcurrency
   } else {
-    let headBlock = currentBlockHeight - blockLag->Option.getWithDefault(0)
-
     let fullPartitions = []
     let mergingPartitions = []
     let areMergingPartitionsFetching = ref(false)
@@ -823,14 +823,14 @@ let getNextQuery = (
         switch p->makePartitionQuery(
           ~indexingContracts,
           ~endBlock=switch blockLag {
-          | Some(_) =>
+          | 0 => endBlock
+          | _ =>
             switch endBlock {
             | Some(endBlock) => Some(Pervasives.min(headBlock, endBlock))
             // Force head block as an endBlock when blockLag is set
             // because otherwise HyperSync might return bigger range
             | None => Some(headBlock)
             }
-          | None => endBlock
           },
           ~mergeTarget,
         ) {
@@ -893,18 +893,6 @@ let queueItemBlockNumber = (queueItem: queueItem) => {
   }
 }
 
-let queueItemIsInReorgThreshold = (
-  queueItem: queueItem,
-  ~currentBlockHeight,
-  ~highestBlockBelowThreshold,
-) => {
-  if currentBlockHeight === 0 {
-    false
-  } else {
-    queueItem->queueItemBlockNumber > highestBlockBelowThreshold
-  }
-}
-
 /**
 Simple constructor for no item from partition
 */
@@ -969,7 +957,7 @@ let make = (
   ~contracts: array<indexingContract>,
   ~maxAddrInPartition,
   ~chainId,
-  ~blockLag=?,
+  ~blockLag=0,
 ): t => {
   let latestFetchedBlock = {
     blockTimestamp: 0,
@@ -1226,3 +1214,15 @@ let isActivelyIndexing = ({latestFullyFetchedBlock, endBlock} as fetchState: t)
   | None => true
   }
 }
+
+let isReadyToEnterReorgThreshold = (
+  {latestFullyFetchedBlock, endBlock, blockLag, queue}: t,
+  ~currentBlockHeight,
+) => {
+  currentBlockHeight !== 0 &&
+  switch endBlock {
+  | Some(endBlock) if latestFullyFetchedBlock.blockNumber >= endBlock => true
+  | _ => latestFullyFetchedBlock.blockNumber >= currentBlockHeight - blockLag
+  } &&
+  queue->Utils.Array.isEmpty
+}

codegenerator/cli/npm/envio/src/Prometheus.res

@@ -472,6 +472,17 @@ module ReorgDetectionBlockNumber = {
   }
 }
 
+module ReorgThreshold = {
+  let gauge = PromClient.Gauge.makeGauge({
+    "name": "envio_reorg_threshold",
+    "help": "Whether indexing is currently within the reorg threshold",
+  })
+
+  let set = (~isInReorgThreshold) => {
+    gauge->PromClient.Gauge.set(isInReorgThreshold ? 1 : 0)
+  }
+}
+
 module RollbackEnabled = {
   let gauge = PromClient.Gauge.makeGauge({
     "name": "envio_rollback_enabled",

codegenerator/cli/templates/static/codegen/src/eventFetching/ChainFetcher.res

@@ -36,13 +36,14 @@ let make = (
   ~endBlock,
   ~dbFirstEventBlockNumber,
   ~latestProcessedBlock,
+  ~config: Config.t,
   ~logger,
   ~timestampCaughtUpToHeadOrEndblock,
   ~numEventsProcessed,
   ~numBatchesFetched,
   ~processingFilters,
   ~maxAddrInPartition,
-  ~enableRawEvents,
+  ~isInReorgThreshold,
 ): t => {
   // We don't need the router itself, but only validation logic,
   // since now event router is created for selection of events
@@ -73,7 +74,7 @@ let make = (
 
       // Filter out non-preRegistration events on preRegistration phase
       // so we don't care about it in fetch state and workers anymore
-      let shouldBeIncluded = if enableRawEvents {
+      let shouldBeIncluded = if config.enableRawEvents {
         true
       } else {
         let isRegistered = hasContractRegister || eventConfig.handler->Option.isSome
@@ -133,7 +134,12 @@ let make = (
     ~endBlock,
     ~eventConfigs,
     ~chainId=chainConfig.chain->ChainMap.Chain.toChainId,
-    ~blockLag=?Env.indexingBlockLag,
+    ~blockLag=Pervasives.max(
+      !(config->Config.shouldRollbackOnReorg) || isInReorgThreshold
+        ? 0
+        : chainConfig.confirmedBlockThreshold,
+      Env.indexingBlockLag->Option.getWithDefault(0),
+    ),
   )
 
   {
@@ -156,14 +162,15 @@ let make = (
   }
 }
 
-let makeFromConfig = (chainConfig: Config.chainConfig, ~maxAddrInPartition, ~enableRawEvents) => {
+let makeFromConfig = (chainConfig: Config.chainConfig, ~config, ~maxAddrInPartition) => {
   let logger = Logging.createChild(~params={"chainId": chainConfig.chain->ChainMap.Chain.toChainId})
   let lastBlockScannedHashes = ReorgDetection.LastBlockScannedHashes.empty(
     ~confirmedBlockThreshold=chainConfig.confirmedBlockThreshold,
   )
 
   make(
     ~chainConfig,
+    ~config,
     ~startBlock=chainConfig.startBlock,
     ~endBlock=chainConfig.endBlock,
     ~lastBlockScannedHashes,
@@ -176,7 +183,7 @@ let makeFromConfig = (chainConfig: Config.chainConfig, ~maxAddrInPartition, ~ena
     ~processingFilters=None,
     ~dynamicContracts=[],
     ~maxAddrInPartition,
-    ~enableRawEvents,
+    ~isInReorgThreshold=false,
   )
 }
 
@@ -186,7 +193,8 @@ let makeFromConfig = (chainConfig: Config.chainConfig, ~maxAddrInPartition, ~ena
 let makeFromDbState = async (
   chainConfig: Config.chainConfig,
   ~maxAddrInPartition,
-  ~enableRawEvents,
+  ~isInReorgThreshold,
+  ~config,
   ~sql=Db.sql,
 ) => {
   let logger = Logging.createChild(~params={"chainId": chainConfig.chain->ChainMap.Chain.toChainId})
@@ -288,6 +296,7 @@ let makeFromDbState = async (
     ~chainConfig,
     ~startBlock=restartBlockNumber,
     ~endBlock=chainConfig.endBlock,
+    ~config,
     ~lastBlockScannedHashes,
     ~dbFirstEventBlockNumber=firstEventBlockNumber,
     ~latestProcessedBlock=latestProcessedBlockChainMetadata,
@@ -297,7 +306,7 @@ let makeFromDbState = async (
     ~logger,
     ~processingFilters,
     ~maxAddrInPartition,
-    ~enableRawEvents,
+    ~isInReorgThreshold,
   )
 }