Turn reorg checkpoints into general checkpoints

Author: DZakh

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

@@ -1,3 +1,28 @@
+open Belt
+
+type batchCheckpoint = {
+  checkpointId: bigint,
+  chainId: int,
+  blockNumber: int,
+  // Might be empty if we are not in reorg threshold
+  // or rollback on reorg is disabled
+  blockHash?: string,
+  // Might be empty if we it's a reorg guard block
+  items?: array<Internal.item>,
+}
+
+type chainBeforeBatch = {
+  chainId: int,
+  fetchState: FetchState.t,
+  blocks: ChainBlocks.t,
+  totalEventsProcessed: int,
+}
+
+type mutChainAcc = {
+  mutable batchSize: int,
+  mutable lastCheckpoint: batchCheckpoint,
+}
+
 type progressedChain = {
   chainId: int,
   batchSize: int,
@@ -6,11 +31,12 @@ type progressedChain = {
 }
 
 type t = {
-  items: array<Internal.item>,
+  checkpoints: array<batchCheckpoint>,
   progressedChains: array<progressedChain>,
-  updatedFetchStates: ChainMap.t<FetchState.t>,
-  dcsToStoreByChainId: dict<array<FetchState.indexingContract>>,
-  creationTimeMs: int,
+  totalBatchSize: int,
+  // updatedFetchStates: ChainMap.t<FetchState.t>,
+  // dcsToStoreByChainId: dict<array<FetchState.indexingContract>>,
+  // creationTimeMs: int,
 }
 
 /**
@@ -21,7 +47,7 @@ let getOrderedNextChain = (fetchStates: ChainMap.t<FetchState.t>, ~batchSizePerC
   let earliestChainTimestamp = ref(0)
   let chainKeys = fetchStates->ChainMap.keys
   for idx in 0 to chainKeys->Array.length - 1 {
-    let chain = chainKeys->Array.get(idx)
+    let chain = chainKeys->Array.getUnsafe(idx)
     let fetchState = fetchStates->ChainMap.get(chain)
     if fetchState->FetchState.isActivelyIndexing {
       let timestamp = fetchState->FetchState.getTimestampAt(
@@ -120,41 +146,59 @@ let prepareOrderedBatch = (
 }
 
 let prepareUnorderedBatch = (
+  ~chains: array<chainBeforeBatch>,
   ~batchSizeTarget,
-  ~fetchStates: ChainMap.t<FetchState.t>,
-  ~mutBatchSizePerChain: dict<int>,
-) => {
+  ~progressCheckpointId,
+): t => {
+  let progressCheckpointId = ref(progressCheckpointId)
+  let accPerChain = Js.Dict.empty()
+
   let preparedFetchStates =
-    fetchStates
-    ->ChainMap.values
+    chains
+    ->Array.map(chain => chain.fetchState)
     ->FetchState.filterAndSortForUnorderedBatch(~batchSizeTarget)
 
   let chainIdx = ref(0)
   let preparedNumber = preparedFetchStates->Array.length
-  let batchSize = ref(0)
+  let totalBatchSize = ref(0)
 
   let items = []
 
   // Accumulate items for all actively indexing chains
   // the way to group as many items from a single chain as possible
   // This way the loaders optimisations will hit more often
-  while batchSize.contents < batchSizeTarget && chainIdx.contents < preparedNumber {
+  while totalBatchSize.contents < batchSizeTarget && chainIdx.contents < preparedNumber {
     let fetchState = preparedFetchStates->Js.Array2.unsafe_get(chainIdx.contents)
+    let chainAcc = switch accPerChain->Utils.Dict.dangerouslyGetByIntNonOption(fetchState.chainId) {
+    | Some(chainAcc) => chainAcc
+    | None =>
+      let acc = {
+        batchSize: 0,
+        lastCheckpoint: %raw(`null`),
+      }
+      accPerChain->Utils.Dict.setByInt(fetchState.chainId, acc)
+      acc
+    }
+
     let chainBatchSize =
       fetchState->FetchState.getReadyItemsCount(
-        ~targetSize=batchSizeTarget - batchSize.contents,
+        ~targetSize=batchSizeTarget - totalBatchSize.contents,
         ~fromItem=0,
       )
     if chainBatchSize > 0 {
       for idx in 0 to chainBatchSize - 1 {
         items->Js.Array2.push(fetchState.buffer->Belt.Array.getUnsafe(idx))->ignore
       }
-      batchSize := batchSize.contents + chainBatchSize
-      mutBatchSizePerChain->Utils.Dict.setByInt(fetchState.chainId, chainBatchSize)
+      totalBatchSize := totalBatchSize.contents + chainBatchSize
+      chainAcc.batchSize = chainBatchSize
     }
 
     chainIdx := chainIdx.contents + 1
   }
 
-  items
+  {
+    totalBatchSize: totalBatchSize.contents,
+    progressedChains: [],
+    checkpoints: [],
+  }
 }

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

@@ -0,0 +1,49 @@
+open Belt
+
+type t = {
+  shouldRollbackOnReorg: bool,
+  lastBlockScannedHashes: ReorgDetection.LastBlockScannedHashes.t,
+}
+
+let make = (
+  ~chainId,
+  ~maxReorgDepth,
+  ~shouldRollbackOnReorg,
+  ~reorgCheckpoints: array<InternalTable.Checkpoints.t>,
+) => {
+  {
+    shouldRollbackOnReorg,
+    lastBlockScannedHashes: reorgCheckpoints
+    ->Array.keepMapU(reorgCheckpoint => {
+      if reorgCheckpoint.chainId === chainId {
+        Some({
+          ReorgDetection.blockNumber: reorgCheckpoint.blockNumber,
+          blockHash: reorgCheckpoint.blockHash,
+        })
+      } else {
+        None
+      }
+    })
+    ->ReorgDetection.LastBlockScannedHashes.makeWithData(~maxReorgDepth),
+  }
+}
+
+let registerReorgGuard = (
+  chainBlocks: t,
+  ~reorgGuard: ReorgDetection.reorgGuard,
+  ~currentBlockHeight: int,
+) => {
+  let (updatedLastBlockScannedHashes, reorgResult: ReorgDetection.reorgResult) =
+    chainBlocks.lastBlockScannedHashes->ReorgDetection.LastBlockScannedHashes.registerReorgGuard(
+      ~reorgGuard,
+      ~currentBlockHeight,
+      ~shouldRollbackOnReorg=chainBlocks.shouldRollbackOnReorg,
+    )
+  (
+    {
+      ...chainBlocks,
+      lastBlockScannedHashes: updatedLastBlockScannedHashes,
+    },
+    reorgResult,
+  )
+}

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

@@ -17,6 +17,7 @@ type initialState = {
   cleanRun: bool,
   cache: dict<effectCacheRecord>,
   chains: array<InternalTable.Chains.t>,
+  reorgCheckpoints: array<InternalTable.Checkpoints.t>,
 }
 
 type operator = [#">" | #"="]

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

@@ -65,7 +65,7 @@ let makeInitializeTransaction = (
   let generalTables = [
     InternalTable.Chains.table,
     InternalTable.PersistedState.table,
-    InternalTable.ReorgCheckpoints.table,
+    InternalTable.Checkpoints.table,
     InternalTable.RawEvents.table,
   ]
 
@@ -707,6 +707,7 @@ let make = (
     {
       cleanRun: true,
       cache,
+      reorgCheckpoints: [],
       chains: chainConfigs->Js.Array2.map(InternalTable.Chains.initialFromConfig),
     }
   }
@@ -895,17 +896,23 @@ let make = (
   }
 
   let resumeInitialState = async (): Persistence.initialState => {
-    let (cache, chains) = await Promise.all2((
+    let (cache, chains, reorgCheckpoints) = await Promise.all3((
       restoreEffectCache(~withUpload=false),
       sql
       ->Postgres.unsafe(
         makeLoadAllQuery(~pgSchema, ~tableName=InternalTable.Chains.table.tableName),
       )
       ->(Utils.magic: promise<array<unknown>> => promise<array<InternalTable.Chains.t>>),
+      sql
+      ->Postgres.unsafe(
+        makeLoadAllQuery(~pgSchema, ~tableName=InternalTable.Checkpoints.table.tableName),
+      )
+      ->(Utils.magic: promise<array<unknown>> => promise<array<InternalTable.Checkpoints.t>>),
     ))
 
     {
       cleanRun: false,
+      reorgCheckpoints,
       cache,
       chains,
     }

codegenerator/cli/npm/envio/src/db/InternalTable.res

@@ -206,7 +206,13 @@ WHERE "id" = $1;`
     Promise.all(promises)
   }
 
-  let setProgressedChains = (sql, ~pgSchema, ~progressedChains: array<Batch.progressedChain>) => {
+  type progressedChain = {
+    chainId: int,
+    progressBlockNumber: int,
+    totalEventsProcessed: int,
+  }
+
+  let setProgressedChains = (sql, ~pgSchema, ~progressedChains: array<progressedChain>) => {
     let query = makeProgressFieldsUpdateQuery(~pgSchema)
 
     let promises = []
@@ -259,24 +265,27 @@ module PersistedState = {
   )
 }
 
-module ReorgCheckpoints = {
+module Checkpoints = {
   type t = {
     id: bigint,
     @as("chain_id")
     chainId: int,
     @as("block_number")
     blockNumber: int,
     @as("block_hash")
-    blockHash: string,
+    blockHash: Js.null<string>,
+    @as("events_processed")
+    eventsProcessed: int,
   }
 
   let table = mkTable(
-    "envio_reorg_checkpoints",
+    "envio_checkpoints",
     ~fields=[
       mkField("id", Numeric, ~fieldSchema=S.bigint, ~isPrimaryKey),
       mkField("chain_id", Integer, ~fieldSchema=S.int),
       mkField("block_number", Integer, ~fieldSchema=S.int),
-      mkField("block_hash", Text, ~fieldSchema=S.string),
+      mkField("block_hash", Text, ~fieldSchema=S.null(S.string), ~isNullable),
+      mkField("events_processed", Integer, ~fieldSchema=S.int),
     ],
   )
 }

codegenerator/cli/npm/envio/src/db/Table.res

@@ -5,6 +5,7 @@ type derived
 @unboxed
 type fieldType =
   | @as("INTEGER") Integer
+  | @as("BIGINT") BigInt
   | @as("BOOLEAN") Boolean
   | @as("NUMERIC") Numeric
   | @as("DOUBLE PRECISION") DoublePrecision

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

@@ -21,6 +21,7 @@ type t = {
   firstEventBlockNumber: option<int>,
   numEventsProcessed: int,
   numBatchesFetched: int,
+  blocks: ChainBlocks.t,
   //An optional list of filters to apply on event queries
   //Used for reorgs and restarts
   processingFilters: option<array<processingFilter>>,
@@ -42,6 +43,7 @@ let make = (
   ~numBatchesFetched,
   ~isInReorgThreshold,
   ~maxReorgDepth,
+  ~reorgCheckpoints,
 ): t => {
   // We don't need the router itself, but only validation logic,
   // since now event router is created for selection of events
@@ -202,6 +204,12 @@ let make = (
     timestampCaughtUpToHeadOrEndblock,
     numEventsProcessed,
     numBatchesFetched,
+    blocks: ChainBlocks.make(
+      ~chainId=chainConfig.id,
+      ~maxReorgDepth,
+      ~reorgCheckpoints,
+      ~shouldRollbackOnReorg=config->Config.shouldRollbackOnReorg,
+    ),
     processingFilters: None,
   }
 }
@@ -224,6 +232,7 @@ let makeFromConfig = (chainConfig: InternalConfig.chain, ~config, ~targetBufferS
     ~dynamicContracts=[],
     ~isInReorgThreshold=false,
     ~maxReorgDepth=chainConfig.maxReorgDepth,
+    ~reorgCheckpoints=[],
   )
 }
 
@@ -233,6 +242,7 @@ let makeFromConfig = (chainConfig: InternalConfig.chain, ~config, ~targetBufferS
 let makeFromDbState = async (
   chainConfig: InternalConfig.chain,
   ~resumedChainState: InternalTable.Chains.t,
+  ~reorgCheckpoints: array<InternalTable.Checkpoints.t>,
   ~isInReorgThreshold,
   ~config,
   ~targetBufferSize,
@@ -271,6 +281,7 @@ let makeFromDbState = async (
     ~logger,
     ~targetBufferSize,
     ~isInReorgThreshold,
+    ~reorgCheckpoints,
   )
 }
 

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

@@ -3,6 +3,11 @@ open Belt
 type t = {
   chainFetchers: ChainMap.t<ChainFetcher.t>,
   multichain: InternalConfig.multichain,
+  // This is an in-memory pointer for checkpoints
+  // Checkpiont references a block in the order it was processed
+  // On restart we should use the last checkpoint id used for entity history
+  // or it's fine to reset it to 0 when there's no history yet
+  progressCheckpointId: bigint,
   isInReorgThreshold: bool,
 }
 
@@ -25,6 +30,7 @@ let makeFromConfig = (~config: Config.t): t => {
   {
     chainFetchers,
     multichain: config.multichain,
+    progressCheckpointId: 0n,
     isInReorgThreshold: false,
   }
 }
@@ -65,6 +71,7 @@ let makeFromDbState = async (~initialState: Persistence.initialState, ~config: C
         await chainConfig->ChainFetcher.makeFromDbState(
           ~resumedChainState,
           ~isInReorgThreshold,
+          ~reorgCheckpoints=initialState.reorgCheckpoints,
           ~targetBufferSize,
           ~config,
         ),
@@ -78,6 +85,7 @@ let makeFromDbState = async (~initialState: Persistence.initialState, ~config: C
     multichain: config.multichain,
     chainFetchers,
     isInReorgThreshold,
+    progressCheckpointId: 0n, // FIXME: get from initialState
   }
 }