Merge branch 'main' of github.com:enviodev/hyperindex into dz/track-blocks

Author: DZakh

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

@@ -3,6 +3,7 @@ open Belt
 type chainAfterBatch = {
   batchSize: int,
   progressBlockNumber: int,
+  isProgressAtHead: bool,
   totalEventsProcessed: int,
   fetchState: FetchState.t,
   dcsToStore: option<array<FetchState.indexingContract>>,

codegenerator/cli/npm/envio/src/bindings/PromClient.res

@@ -11,6 +11,14 @@ type registry
 @send external metrics: registry => Promise.t<string> = "metrics"
 @get external getContentType: registry => string = "contentType"
 
+type metricValue = {
+  value: int,
+  labels: dict<string>,
+}
+type metricInstance = {get: unit => promise<{"values": array<metricValue>}>}
+@send external getSingleMetric: (registry, string) => option<metricInstance> = "getSingleMetric"
+@send external resetMetrics: registry => unit = "resetMetrics"
+
 module Counter = {
   type counter
   @new @module("prom-client") external makeCounter: customMetric<'a> => counter = "Counter"
@@ -36,6 +44,8 @@ module Gauge = {
   @send external setFloat: (gauge, float) => unit = "set"
 
   @send external labels: (gauge, 'labelsObject) => gauge = "labels"
+
+  @send external get: gauge => promise<{"values": array<dict<string>>}> = "get"
 }
 
 module Histogram = {

scenarios/test_codegen/test/E2E_test.res

@@ -37,4 +37,49 @@ describe("E2E tests", () => {
       ~message="Should request items from start block to reorg threshold",
     )
   })
+
+  Async.it("Correctly sets Prom metrics", async () => {
+    let sourceMock = Mock.Source.make(
+      [#getHeightOrThrow, #getItemsOrThrow, #getBlockHashes],
+      ~chain=#1337,
+    )
+    let indexerMock = await Mock.Indexer.make(
+      ~chains=[
+        {
+          chain: #1337,
+          sources: [sourceMock.source],
+        },
+      ],
+    )
+    await Utils.delay(0)
+
+    Assert.deepEqual(
+      await indexerMock.metric("envio_reorg_threshold"),
+      [{value: "0", labels: Js.Dict.empty()}],
+    )
+    Assert.deepEqual(
+      await indexerMock.metric("hyperindex_synced_to_head"),
+      [{value: "0", labels: Js.Dict.empty()}],
+    )
+
+    await Mock.Helper.initialEnterReorgThreshold(~sourceMock)
+
+    Assert.deepEqual(
+      await indexerMock.metric("envio_reorg_threshold"),
+      [{value: "1", labels: Js.Dict.empty()}],
+    )
+    Assert.deepEqual(
+      await indexerMock.metric("hyperindex_synced_to_head"),
+      [{value: "0", labels: Js.Dict.empty()}],
+    )
+
+    sourceMock.resolveGetItemsOrThrow([], ~latestFetchedBlockNumber=300)
+    await indexerMock.getBatchWritePromise()
+
+    Assert.deepEqual(
+      await indexerMock.metric("hyperindex_synced_to_head"),
+      [{value: "1", labels: Js.Dict.empty()}],
+      ~message="should have set hyperindex_synced_to_head metric to 1",
+    )
+  })
 })

scenarios/test_codegen/test/helpers/Mock.res

@@ -207,19 +207,26 @@ module Storage = {
 }
 
 module Indexer = {
+  type metric = {
+    value: string,
+    labels: dict<string>,
+  }
   type t = {
     getBatchWritePromise: unit => promise<unit>,
     getRollbackReadyPromise: unit => promise<unit>,
     query: 'entity. module(Entities.Entity with type t = 'entity) => promise<array<'entity>>,
     queryHistory: 'entity. module(Entities.Entity with type t = 'entity) => promise<
       array<EntityHistory.historyRow<'entity>>,
     >,
+    metric: string => promise<array<metric>>,
   }
 
   type chainConfig = {chain: Types.chain, sources: array<Source.t>, startBlock?: int}
 
   let make = async (~chains: array<chainConfig>, ~multichain=InternalConfig.Unordered) => {
     DbHelpers.resetPostgresClient()
+    // TODO: Should stop using global client
+    PromClient.defaultRegister->PromClient.resetMetrics
 
     let config = RegisterHandlers.registerAllHandlers()
 
@@ -344,6 +351,16 @@ module Indexer = {
           >
         )
       },
+      metric: async name => {
+        switch PromClient.defaultRegister->PromClient.getSingleMetric(name) {
+        | Some(m) =>
+          (await m.get())["values"]->Js.Array2.map(v => {
+            value: v.value->Belt.Int.toString,
+            labels: v.labels,
+          })
+        | None => []
+        }
+      },
     }
   }
 }
@@ -573,6 +590,33 @@ module Source = {
   }
 }
 
+module Helper = {
+  let initialEnterReorgThreshold = async (~sourceMock: Source.t) => {
+    open RescriptMocha
+
+    Assert.deepEqual(
+      sourceMock.getHeightOrThrowCalls->Array.length,
+      1,
+      ~message="should have called getHeightOrThrow to get initial height",
+    )
+    sourceMock.resolveGetHeightOrThrow(300)
+    await Utils.delay(0)
+    await Utils.delay(0)
+
+    let expectedGetItemsCall1 = {"fromBlock": 0, "toBlock": Some(100), "retry": 0}
+
+    Assert.deepEqual(
+      sourceMock.getItemsOrThrowCalls,
+      [expectedGetItemsCall1],
+      ~message="Should request items until reorg threshold",
+    )
+    sourceMock.resolveGetItemsOrThrow([])
+    await Utils.delay(0)
+    await Utils.delay(0)
+    await Utils.delay(0)
+  }
+}
+
 @genType
 let mockRawEventRow: InternalTable.RawEvents.t = {
   chainId: 1,

scenarios/test_codegen/test/rollback/Rollback_test.res

@@ -417,29 +417,6 @@ describe("Single Chain Simple Rollback", () => {
 let undefined = (%raw(`undefined`): option<'a>)
 
 describe("E2E rollback tests", () => {
-  let initialEnterReorgThreshold = async (~sourceMock: M.Source.t) => {
-    Assert.deepEqual(
-      sourceMock.getHeightOrThrowCalls->Array.length,
-      1,
-      ~message="should have called getHeightOrThrow to get initial height",
-    )
-    sourceMock.resolveGetHeightOrThrow(300)
-    await Utils.delay(0)
-    await Utils.delay(0)
-
-    let expectedGetItemsCall1 = {"fromBlock": 0, "toBlock": Some(100), "retry": 0}
-
-    Assert.deepEqual(
-      sourceMock.getItemsOrThrowCalls,
-      [expectedGetItemsCall1],
-      ~message="Should request items until reorg threshold",
-    )
-    sourceMock.resolveGetItemsOrThrow([])
-    await Utils.delay(0)
-    await Utils.delay(0)
-    await Utils.delay(0)
-  }
-
   let testSingleChainRollback = async (~sourceMock: M.Source.t, ~indexerMock: M.Indexer.t) => {
     Assert.deepEqual(
       sourceMock.getItemsOrThrowCalls->Utils.Array.last,
@@ -736,7 +713,7 @@ describe("E2E rollback tests", () => {
     )
     await Utils.delay(0)
 
-    await initialEnterReorgThreshold(~sourceMock)
+    await M.Helper.initialEnterReorgThreshold(~sourceMock)
     await testSingleChainRollback(~sourceMock, ~indexerMock)
   })
 
@@ -766,8 +743,8 @@ describe("E2E rollback tests", () => {
       await Utils.delay(0)
 
       let _ = await Promise.all2((
-        initialEnterReorgThreshold(~sourceMock=sourceMock1),
-        initialEnterReorgThreshold(~sourceMock=sourceMock2),
+        M.Helper.initialEnterReorgThreshold(~sourceMock=sourceMock1),
+        M.Helper.initialEnterReorgThreshold(~sourceMock=sourceMock2),
       ))
 
       await testSingleChainRollback(~sourceMock=sourceMock1, ~indexerMock)
@@ -789,7 +766,7 @@ describe("E2E rollback tests", () => {
     )
     await Utils.delay(0)
 
-    await initialEnterReorgThreshold(~sourceMock)
+    await M.Helper.initialEnterReorgThreshold(~sourceMock)
 
     let calls = []
     let handler: Types.HandlerTypes.loader<unit, unit> = async ({event}) => {
@@ -1050,8 +1027,8 @@ This might be wrong after we start exposing a block hash for progress block.`,
     await Utils.delay(0)
 
     let _ = await Promise.all2((
-      initialEnterReorgThreshold(~sourceMock=sourceMock1337),
-      initialEnterReorgThreshold(~sourceMock=sourceMock100),
+      M.Helper.initialEnterReorgThreshold(~sourceMock=sourceMock1337),
+      M.Helper.initialEnterReorgThreshold(~sourceMock=sourceMock100),
     ))
 
     let callCount = ref(0)