Prevent invalid rollback with the depth 0 (#544)

* Always update hashes state

* Prevent rollback with depth 0

* Better reorg guard naming and fix tests

* Fix merge conflict error

Author: DZakh

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

@@ -940,14 +940,14 @@ let queueItemBlockNumber = (queueItem: queueItem) => {
 let queueItemIsInReorgThreshold = (
   queueItem: queueItem,
   ~currentBlockHeight,
-  ~heighestBlockBelowThreshold,
+  ~highestBlockBelowThreshold,
 ) => {
   if currentBlockHeight === 0 {
     false
   } else {
     switch queueItem {
-    | Item(_) => queueItem->queueItemBlockNumber > heighestBlockBelowThreshold
-    | NoItem(_) => queueItem->queueItemBlockNumber > heighestBlockBelowThreshold
+    | Item(_) => queueItem->queueItemBlockNumber > highestBlockBelowThreshold
+    | NoItem(_) => queueItem->queueItemBlockNumber > highestBlockBelowThreshold
     }
   }
 }

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

@@ -15,8 +15,8 @@ type blockData = {
 external generalizeBlockDataWithTimestamp: blockDataWithTimestamp => blockData = "%identity"
 
 type reorgGuard = {
-  lastBlockScannedData: blockData,
-  firstBlockParentNumberAndHash: option<blockData>,
+  rangeLastBlock: blockData,
+  prevRangeLastBlock: option<blockData>,
 }
 
 type reorgDetected = {
@@ -36,22 +36,31 @@ let reorgDetectedToLogParams = (reorgDetected: reorgDetected, ~shouldRollbackOnR
   }
 }
 
+type reorgResult = NoReorg | ReorgDetected(reorgDetected)
+type validBlockError = NotFound | AlreadyReorgedHashes
+type validBlockResult = result<blockDataWithTimestamp, validBlockError>
+
 module LastBlockScannedHashes: {
   type t
   /**Instantiat t with existing data*/
-  let makeWithData: (array<blockData>, ~confirmedBlockThreshold: int) => t
+  let makeWithData: (
+    array<blockData>,
+    ~confirmedBlockThreshold: int,
+    ~detectedReorgBlock: blockData=?,
+  ) => t
 
   /**Instantiat empty t with no block data*/
   let empty: (~confirmedBlockThreshold: int) => t
 
-  /** Registers a new reorg guard, prunes unnened data and returns the updated data 
-   or an error if a reorg has occured 
-  */
+  /** Registers a new reorg guard, prunes unneeded data, and returns the updated state.
+   * Resets internal state if shouldRollbackOnReorg is false (detect-only mode)
+   */
   let registerReorgGuard: (
     t,
     ~reorgGuard: reorgGuard,
     ~currentBlockHeight: int,
-  ) => result<t, reorgDetected>
+    ~shouldRollbackOnReorg: bool,
+  ) => (t, reorgResult)
 
   /**
   Returns the latest block data which matches block number and hashes in the provided array
@@ -61,7 +70,7 @@ module LastBlockScannedHashes: {
     t,
     ~blockNumbersAndHashes: array<blockDataWithTimestamp>,
     ~currentBlockHeight: int,
-  ) => option<blockDataWithTimestamp>
+  ) => validBlockResult
 
   let getThresholdBlockNumbers: (t, ~currentBlockHeight: int) => array<int>
 
@@ -76,9 +85,14 @@ module LastBlockScannedHashes: {
     // A hash map of recent blockdata by block number to make comparison checks
     // for reorgs.
     dataByBlockNumber: dict<blockData>,
+    // The latest block which detected a reorg
+    // and should never be valid.
+    // We keep track of this to avoid responses
+    // with the stale data from other data-source instances.
+    detectedReorgBlock: option<blockData>,
   }
 
-  let makeWithData = (blocks, ~confirmedBlockThreshold) => {
+  let makeWithData = (blocks, ~confirmedBlockThreshold, ~detectedReorgBlock=?) => {
     let dataByBlockNumber = Js.Dict.empty()
 
     blocks->Belt.Array.forEach(block => {
@@ -88,12 +102,14 @@ module LastBlockScannedHashes: {
     {
       confirmedBlockThreshold,
       dataByBlockNumber,
+      detectedReorgBlock,
     }
   }
   //Instantiates empty LastBlockHashes
   let empty = (~confirmedBlockThreshold) => {
     confirmedBlockThreshold,
     dataByBlockNumber: Js.Dict.empty(),
+    detectedReorgBlock: None,
   }
 
   let getDataByBlockNumberCopyInThreshold = (
@@ -122,33 +138,33 @@ module LastBlockScannedHashes: {
     {confirmedBlockThreshold} as self: t,
     ~reorgGuard: reorgGuard,
     ~currentBlockHeight,
+    ~shouldRollbackOnReorg,
   ) => {
     let dataByBlockNumberCopyInThreshold =
       self->getDataByBlockNumberCopyInThreshold(~currentBlockHeight)
 
-    let {lastBlockScannedData, firstBlockParentNumberAndHash} = reorgGuard
+    let {rangeLastBlock, prevRangeLastBlock} = reorgGuard
 
     let maybeReorgDetected = switch dataByBlockNumberCopyInThreshold->Utils.Dict.dangerouslyGetNonOption(
-      lastBlockScannedData.blockNumber->Int.toString,
+      rangeLastBlock.blockNumber->Int.toString,
     ) {
-    | Some(scannedBlock) if scannedBlock.blockHash !== lastBlockScannedData.blockHash =>
+    | Some(scannedBlock) if scannedBlock.blockHash !== rangeLastBlock.blockHash =>
       Some({
-        receivedBlock: lastBlockScannedData,
+        receivedBlock: rangeLastBlock,
         scannedBlock,
       })
     | _ =>
-      switch firstBlockParentNumberAndHash {
+      switch prevRangeLastBlock {
       //If parentHash is None, then it's the genesis block (no reorg)
       //Need to check that parentHash matches because of the dynamic contracts
       | None => None
-      | Some(firstBlockParentNumberAndHash) =>
+      | Some(prevRangeLastBlock) =>
         switch dataByBlockNumberCopyInThreshold->Utils.Dict.dangerouslyGetNonOption(
-          firstBlockParentNumberAndHash.blockNumber->Int.toString,
+          prevRangeLastBlock.blockNumber->Int.toString,
         ) {
-        | Some(scannedBlock)
-          if scannedBlock.blockHash !== firstBlockParentNumberAndHash.blockHash =>
+        | Some(scannedBlock) if scannedBlock.blockHash !== prevRangeLastBlock.blockHash =>
           Some({
-            receivedBlock: firstBlockParentNumberAndHash,
+            receivedBlock: prevRangeLastBlock,
             scannedBlock,
           })
         | _ => None
@@ -157,25 +173,37 @@ module LastBlockScannedHashes: {
     }
 
     switch maybeReorgDetected {
-    | Some(reorgDetected) => Error(reorgDetected)
+    | Some(reorgDetected) => (
+        shouldRollbackOnReorg
+          ? {
+              ...self,
+              detectedReorgBlock: Some(reorgDetected.scannedBlock),
+            }
+          : empty(~confirmedBlockThreshold),
+        ReorgDetected(reorgDetected),
+      )
     | None => {
         dataByBlockNumberCopyInThreshold->Js.Dict.set(
-          lastBlockScannedData.blockNumber->Int.toString,
-          lastBlockScannedData,
+          rangeLastBlock.blockNumber->Int.toString,
+          rangeLastBlock,
         )
-        switch firstBlockParentNumberAndHash {
+        switch prevRangeLastBlock {
         | None => ()
-        | Some(firstBlockParentNumberAndHash) =>
+        | Some(prevRangeLastBlock) =>
           dataByBlockNumberCopyInThreshold->Js.Dict.set(
-            firstBlockParentNumberAndHash.blockNumber->Int.toString,
-            firstBlockParentNumberAndHash,
+            prevRangeLastBlock.blockNumber->Int.toString,
+            prevRangeLastBlock,
           )
         }
 
-        Ok({
-          confirmedBlockThreshold,
-          dataByBlockNumber: dataByBlockNumberCopyInThreshold,
-        })
+        (
+          {
+            confirmedBlockThreshold,
+            dataByBlockNumber: dataByBlockNumberCopyInThreshold,
+            detectedReorgBlock: None,
+          },
+          NoReorg,
+        )
       }
     }
   }
@@ -186,39 +214,58 @@ module LastBlockScannedHashes: {
     ~currentBlockHeight,
   ) => {
     let verifiedDataByBlockNumber = Js.Dict.empty()
-    blockNumbersAndHashes->Array.forEach(blockData => {
+    for idx in 0 to blockNumbersAndHashes->Array.length - 1 {
+      let blockData = blockNumbersAndHashes->Array.getUnsafe(idx)
       verifiedDataByBlockNumber->Js.Dict.set(blockData.blockNumber->Int.toString, blockData)
-    })
+    }
 
-    let dataByBlockNumber = self->getDataByBlockNumberCopyInThreshold(~currentBlockHeight)
-    // Js engine automatically orders numeric object keys
-    let ascBlockNumberKeys = dataByBlockNumber->Js.Dict.keys
+    let isAlreadyReorgedResponse = switch self.detectedReorgBlock {
+    | Some(detectedReorgBlock) =>
+      switch verifiedDataByBlockNumber->Utils.Dict.dangerouslyGetNonOption(
+        detectedReorgBlock.blockNumber->Int.toString,
+      ) {
+      | Some(verifiedBlockData) => verifiedBlockData.blockHash === detectedReorgBlock.blockHash
+      | None => false
+      }
+    | None => false
+    }
 
-    let getPrevScannedBlock = idx =>
-      ascBlockNumberKeys
-      ->Belt.Array.get(idx - 1)
-      ->Option.flatMap(key => {
-        // We should already validate that the block number is verified at the point
-        verifiedDataByBlockNumber->Utils.Dict.dangerouslyGetNonOption(key)
-      })
+    if isAlreadyReorgedResponse {
+      Error(AlreadyReorgedHashes)
+    } else {
+      let dataByBlockNumber = self->getDataByBlockNumberCopyInThreshold(~currentBlockHeight)
+      // Js engine automatically orders numeric object keys
+      let ascBlockNumberKeys = dataByBlockNumber->Js.Dict.keys
+
+      let getPrevScannedBlock = idx =>
+        switch ascBlockNumberKeys
+        ->Belt.Array.get(idx - 1)
+        ->Option.flatMap(key => {
+          // We should already validate that the block number is verified at the point
+          verifiedDataByBlockNumber->Utils.Dict.dangerouslyGetNonOption(key)
+        }) {
+        | Some(data) => Ok(data)
+        | None => Error(NotFound)
+        }
 
-    let rec loop = idx => {
-      switch ascBlockNumberKeys->Belt.Array.get(idx) {
-      | Some(blockNumberKey) =>
-        let scannedBlock = dataByBlockNumber->Js.Dict.unsafeGet(blockNumberKey)
-        switch verifiedDataByBlockNumber->Utils.Dict.dangerouslyGetNonOption(blockNumberKey) {
-        | None =>
-          Js.Exn.raiseError(
-            `Unexpected case. Couldn't find verified hash for block number ${blockNumberKey}`,
-          )
-        | Some(verifiedBlockData) if verifiedBlockData.blockHash === scannedBlock.blockHash =>
-          loop(idx + 1)
-        | Some(_) => getPrevScannedBlock(idx)
+      let rec loop = idx => {
+        switch ascBlockNumberKeys->Belt.Array.get(idx) {
+        | Some(blockNumberKey) =>
+          let scannedBlock = dataByBlockNumber->Js.Dict.unsafeGet(blockNumberKey)
+          switch verifiedDataByBlockNumber->Utils.Dict.dangerouslyGetNonOption(blockNumberKey) {
+          | None =>
+            Js.Exn.raiseError(
+              `Unexpected case. Couldn't find verified hash for block number ${blockNumberKey}`,
+            )
+          | Some(verifiedBlockData) if verifiedBlockData.blockHash === scannedBlock.blockHash =>
+            loop(idx + 1)
+          | Some(_) => getPrevScannedBlock(idx)
+          }
+        | None => getPrevScannedBlock(idx)
         }
-      | None => getPrevScannedBlock(idx)
       }
+      loop(0)
     }
-    loop(0)
   }
 
   /**
@@ -254,6 +301,7 @@ module LastBlockScannedHashes: {
     {
       confirmedBlockThreshold,
       dataByBlockNumber: newDataByBlockNumber,
+      detectedReorgBlock: None,
     }
   }
 

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

@@ -443,7 +443,7 @@ let hasNoMoreEventsToProcess = (self: t) => {
   self.fetchState->FetchState.queueSize === 0
 }
 
-let getHeighestBlockBelowThreshold = (cf: t): int => {
+let getHighestBlockBelowThreshold = (cf: t): int => {
   let highestBlockBelowThreshold = cf.currentBlockHeight - cf.chainConfig.confirmedBlockThreshold
   highestBlockBelowThreshold < 0 ? 0 : highestBlockBelowThreshold
 }
@@ -475,7 +475,7 @@ let getLastKnownValidBlock = async (
   }
 
   let fallback = async () => {
-    switch await getBlockHashes([chainFetcher->getHeighestBlockBelowThreshold]) {
+    switch await getBlockHashes([chainFetcher->getHighestBlockBelowThreshold]) {
     | [block] => block
     | _ =>
       Js.Exn.raiseError(
@@ -487,15 +487,27 @@ let getLastKnownValidBlock = async (
   switch scannedBlockNumbers {
   | [] => await fallback()
   | _ => {
-      let blockNumbersAndHashes = await getBlockHashes(scannedBlockNumbers)
-
-      switch chainFetcher.lastBlockScannedHashes->ReorgDetection.LastBlockScannedHashes.getLatestValidScannedBlock(
-        ~blockNumbersAndHashes,
-        ~currentBlockHeight=chainFetcher.currentBlockHeight,
-      ) {
-      | Some(block) => block
-      | None => await fallback()
+      let blockRef = ref(None)
+
+      while blockRef.contents->Option.isNone {
+        let blockNumbersAndHashes = await getBlockHashes(scannedBlockNumbers)
+
+        switch chainFetcher.lastBlockScannedHashes->ReorgDetection.LastBlockScannedHashes.getLatestValidScannedBlock(
+          ~blockNumbersAndHashes,
+          ~currentBlockHeight=chainFetcher.currentBlockHeight,
+        ) {
+        | Ok(block) => blockRef := Some(block)
+        | Error(NotFound) => blockRef := Some(await fallback())
+        | Error(AlreadyReorgedHashes) =>
+          let delayMilliseconds = 100
+          chainFetcher.logger->Logging.childTrace(
+            `Failed to find a valid block to rollback to, since received already reorged hashes from another HyperSync instance. HyperSync has multiple instances and it's possible that they drift independently slightly from the head. Indexing should continue correctly after retrying the query in ${delayMilliseconds->Int.toString}ms.`,
+          )
+          await Utils.delay(delayMilliseconds)
+        }
       }
+
+      blockRef.contents->Option.getUnsafe
     }
   }
 }

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

@@ -67,7 +67,7 @@ type isInReorgThresholdRes<'payload> = {
 
 type fetchStateWithData = {
   fetchState: FetchState.t,
-  heighestBlockBelowThreshold: int,
+  highestBlockBelowThreshold: int,
   currentBlockHeight: int,
 }
 
@@ -78,7 +78,7 @@ let isQueueItemEarlierUnorderedBelowReorgThreshold = (
     let data = fetchStatesMap->ChainMap.get(item.chain)
     item.earliestEvent->FetchState.queueItemIsInReorgThreshold(
       ~currentBlockHeight=data.currentBlockHeight,
-      ~heighestBlockBelowThreshold=data.heighestBlockBelowThreshold,
+      ~highestBlockBelowThreshold=data.highestBlockBelowThreshold,
     )
   }
   // The idea here is if we are in undordered multichain mode, always prioritize queue
@@ -112,7 +112,7 @@ let determineNextEvent = (
     ->ChainMap.entries
     ->Array.reduce({isInReorgThreshold: false, val: None}, (
       accum,
-      (chain, {fetchState, currentBlockHeight, heighestBlockBelowThreshold}),
+      (chain, {fetchState, currentBlockHeight, highestBlockBelowThreshold}),
     ) => {
       // If the fetch state has reached the end block we don't need to consider it
       if fetchState->FetchState.isActivelyIndexing {
@@ -124,7 +124,7 @@ let determineNextEvent = (
           let isInReorgThreshold =
             earliestEvent->FetchState.queueItemIsInReorgThreshold(
               ~currentBlockHeight,
-              ~heighestBlockBelowThreshold,
+              ~highestBlockBelowThreshold,
             )
 
           {
@@ -225,7 +225,7 @@ let getFetchStateWithData = (self: t, ~shouldDeepCopy=false): ChainMap.t<fetchSt
   self.chainFetchers->ChainMap.map(cf => {
     {
       fetchState: shouldDeepCopy ? cf.fetchState->FetchState.copy : cf.fetchState,
-      heighestBlockBelowThreshold: cf->ChainFetcher.getHeighestBlockBelowThreshold,
+      highestBlockBelowThreshold: cf->ChainFetcher.getHighestBlockBelowThreshold,
       currentBlockHeight: cf.currentBlockHeight,
     }
   })
@@ -363,7 +363,7 @@ let getSafeChainIdAndBlockNumberArray = (self: t): array<
   ->ChainMap.values
   ->Array.map((cf): DbFunctions.EntityHistory.chainIdAndBlockNumber => {
     chainId: cf.chainConfig.chain->ChainMap.Chain.toChainId,
-    blockNumber: cf->ChainFetcher.getHeighestBlockBelowThreshold,
+    blockNumber: cf->ChainFetcher.getHighestBlockBelowThreshold,
   })
 }