Merge branch 'main' into dz/storage-load-metrics

Author: DZakh

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

@@ -93,25 +93,6 @@ let copy = (fetchState: t) => {
   }
 }
 
-/*
-Comapritor for two events from the same chain. No need for chain id or timestamp
-*/
-let eventItemGt = (a: Internal.eventItem, b: Internal.eventItem) =>
-  if a.blockNumber > b.blockNumber {
-    true
-  } else if a.blockNumber === b.blockNumber {
-    a.logIndex > b.logIndex
-  } else {
-    false
-  }
-
-/*
-Merges two event queues on a single event fetcher
-
-Pass the shorter list into A for better performance
-*/
-let mergeSortedEventList = (a, b) => Utils.Array.mergeSorted(eventItemGt, a, b)
-
 let mergeIntoPartition = (p: partition, ~target: partition, ~maxAddrInPartition) => {
   switch (p, target) {
   | ({selection: {dependsOnAddresses: true}}, {selection: {dependsOnAddresses: true}}) => {
@@ -565,6 +546,18 @@ type query = {
 exception UnexpectedPartitionNotFound({partitionId: string})
 exception UnexpectedMergeQueryResponse({message: string})
 
+/*
+Comparitor for two events from the same chain. No need for chain id or timestamp
+*/
+let compareBufferItem = (a: Internal.eventItem, b: Internal.eventItem) => {
+  let blockDiff = b.blockNumber - a.blockNumber
+  if blockDiff === 0 {
+    b.logIndex - a.logIndex
+  } else {
+    blockDiff
+  }
+}
+
 /*
 Updates fetchState with a response for a given query.
 Returns Error if the partition with given query cannot be found (unexpected)
@@ -576,7 +569,7 @@ let handleQueryResult = (
   {partitions} as fetchState: t,
   ~query: query,
   ~latestFetchedBlock: blockNumberAndTimestamp,
-  ~reversedNewItems,
+  ~newItems,
   ~currentBlockHeight,
 ): result<t, exn> =>
   {
@@ -633,7 +626,12 @@ let handleQueryResult = (
     fetchState->updateInternal(
       ~partitions,
       ~currentBlockHeight,
-      ~queue=mergeSortedEventList(reversedNewItems, fetchState.queue),
+      ~queue=fetchState.queue
+      ->Array.concat(newItems)
+      // Theoretically it could be faster to asume that
+      // the items are sorted, but there are cases
+      // when the data source returns them unsorted
+      ->Js.Array2.sortInPlaceWith(compareBufferItem),
     )
   })
 

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

@@ -240,6 +240,8 @@ let toSqlParams = (table: table, ~schema, ~pgSchema) => {
           switch field {
           | Field(f) =>
             switch f.fieldType {
+            // The case for `BigDecimal! @config(precision: 10, scale: 8)`
+            | Custom(fieldType) if fieldType->Js.String2.startsWith("NUMERIC(") => fieldType
             | Custom(fieldType) => `${(Text :> string)}[]::"${pgSchema}".${(fieldType :> string)}`
             | Boolean => `${(Integer :> string)}[]::${(f.fieldType :> string)}`
             | fieldType => (fieldType :> string)

codegenerator/cli/templates/dynamic/codegen/src/TestHelpers_MockDb.res.hbs

@@ -318,7 +318,7 @@ and makeProcessEvents = (mockDb: t, ~chainId=?) => async (
   events: array<Types.eventLog<unknown>>,
 ) => {
   let config = RegisterHandlers.getConfig()
-  let reversedWithContractRegister = []
+  let itemsWithContractRegister = []
 
   let processingChainId = ref(chainId)
   let eventItems = events->Array.map(event => {
@@ -355,7 +355,7 @@ and makeProcessEvents = (mockDb: t, ~chainId=?) => async (
       blockNumber: event.block->Types.Block.getNumber,
     }
     if eventConfig.contractRegister->Option.isSome {
-      reversedWithContractRegister->Js.Array2.unshift(eventItem)->ignore
+      itemsWithContractRegister->Js.Array2.push(eventItem)->ignore
     }
     eventItem
   })
@@ -387,8 +387,8 @@ and makeProcessEvents = (mockDb: t, ~chainId=?) => async (
   //No need to check contract is registered or return anything.
   //The only purpose is to test the registerContract function and to
   //add the entity to the in memory store for asserting registrations
-  if reversedWithContractRegister->Utils.Array.notEmpty {
-    let dcs = await ChainFetcher.runContractRegistersOrThrow(~reversedWithContractRegister, ~config)
+  if itemsWithContractRegister->Utils.Array.notEmpty {
+    let dcs = await ChainFetcher.runContractRegistersOrThrow(~itemsWithContractRegister, ~config)
 
     // TODO: Reuse FetchState logic to clean up duplicate dcs
     if dcs->Utils.Array.notEmpty {

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

@@ -353,7 +353,7 @@ let getContractStartBlock = (
 }
 
 let runContractRegistersOrThrow = async (
-  ~reversedWithContractRegister: array<Internal.eventItem>,
+  ~itemsWithContractRegister: array<Internal.eventItem>,
   ~config: Config.t,
 ) => {
   let dynamicContracts = []
@@ -396,8 +396,8 @@ let runContractRegistersOrThrow = async (
   }
 
   let promises = []
-  for idx in reversedWithContractRegister->Array.length - 1 downto 0 {
-    let eventItem = reversedWithContractRegister->Array.getUnsafe(idx)
+  for idx in 0 to itemsWithContractRegister->Array.length - 1 {
+    let eventItem = itemsWithContractRegister->Array.getUnsafe(idx)
     let contractRegister = switch eventItem.eventConfig.contractRegister {
     | Some(contractRegister) => contractRegister
     | None =>
@@ -452,7 +452,7 @@ Returns Error if the node with given id cannot be found (unexpected)
 let handleQueryResult = (
   chainFetcher: t,
   ~query: FetchState.query,
-  ~reversedNewItems,
+  ~newItems,
   ~dynamicContracts,
   ~latestFetchedBlock,
   ~currentBlockHeight,
@@ -470,7 +470,7 @@ let handleQueryResult = (
   ->FetchState.handleQueryResult(
     ~query,
     ~latestFetchedBlock,
-    ~reversedNewItems,
+    ~newItems,
     ~currentBlockHeight,
   )
   ->Result.map(fetchState => {

codegenerator/cli/templates/static/codegen/src/globalState/GlobalState.res

@@ -123,7 +123,7 @@ type action =
   // because when processing the response, there might be an async contract registration.
   // So after it's finished we dispatch the  submit action to get the latest fetch state.
   | SubmitPartitionQueryResponse({
-      reversedNewItems: array<Internal.eventItem>,
+      newItems: array<Internal.eventItem>,
       dynamicContracts: array<FetchState.indexingContract>,
       currentBlockHeight: int,
       latestFetchedBlock: FetchState.blockNumberAndTimestamp,
@@ -469,7 +469,7 @@ let validatePartitionQueryResponse = (
 
 let submitPartitionQueryResponse = (
   state,
-  ~reversedNewItems,
+  ~newItems,
   ~dynamicContracts,
   ~currentBlockHeight,
   ~latestFetchedBlock,
@@ -484,7 +484,7 @@ let submitPartitionQueryResponse = (
       ~query,
       ~currentBlockHeight,
       ~latestFetchedBlock,
-      ~reversedNewItems,
+      ~newItems,
       ~dynamicContracts,
     )
     ->Utils.unwrapResultExn
@@ -538,12 +538,10 @@ let processPartitionQueryResponse = async (
     latestFetchedBlockTimestamp,
   } = response
 
-  let reversedWithContractRegister = []
-  let reversedNewItems = []
+  let itemsWithContractRegister = []
+  let newItems = []
 
-  // It's cheaper to reverse only items with contract register
-  // Then all items. That's why we use downto loop
-  for idx in parsedQueueItems->Array.length - 1 downto 0 {
+  for idx in 0 to parsedQueueItems->Array.length - 1 {
     let item = parsedQueueItems->Array.getUnsafe(idx)
     if (
       switch chainFetcher.processingFilters {
@@ -552,29 +550,29 @@ let processPartitionQueryResponse = async (
       }
     ) {
       if item.eventConfig.contractRegister !== None {
-        reversedWithContractRegister->Array.push(item)
+        itemsWithContractRegister->Array.push(item)
       }
 
       // TODO: Don't really need to keep it in the queue
       // when there's no handler (besides raw_events, processed counter, and dcsToStore consuming)
-      reversedNewItems->Array.push(item)
+      newItems->Array.push(item)
     }
   }
 
-  let dynamicContracts = switch reversedWithContractRegister {
+  let dynamicContracts = switch itemsWithContractRegister {
   | [] as empty =>
     // A small optimisation to not recreate an empty array
     empty->(Utils.magic: array<Internal.eventItem> => array<FetchState.indexingContract>)
   | _ =>
     await ChainFetcher.runContractRegistersOrThrow(
-      ~reversedWithContractRegister,
+      ~itemsWithContractRegister,
       ~config=state.config,
     )
   }
 
   dispatchAction(
     SubmitPartitionQueryResponse({
-      reversedNewItems,
+      newItems,
       dynamicContracts,
       currentBlockHeight,
       latestFetchedBlock: {
@@ -643,15 +641,15 @@ let actionReducer = (state: t, action: action) => {
   | ValidatePartitionQueryResponse(partitionQueryResponse) =>
     state->validatePartitionQueryResponse(partitionQueryResponse)
   | SubmitPartitionQueryResponse({
-      reversedNewItems,
+      newItems,
       dynamicContracts,
       currentBlockHeight,
       latestFetchedBlock,
       query,
       chain,
     }) =>
     state->submitPartitionQueryResponse(
-      ~reversedNewItems,
+      ~newItems,
       ~dynamicContracts,
       ~currentBlockHeight,
       ~latestFetchedBlock,
@@ -878,24 +876,23 @@ let injectedTaskReducer = (
           state.writeThrottlers.deepCleanCount = state.writeThrottlers.deepCleanCount + 1
           false
         }
-        let timeRef = Hrtime.makeTimer()
-        let _ = await Promise.all(
-          Entities.allEntities->Belt.Array.map(entityConfig => {
-            Db.sql->DbFunctions.EntityHistory.pruneStaleEntityHistory(
-              ~entityName=entityConfig.name,
-              ~safeChainIdAndBlockNumberArray,
-              ~shouldDeepClean,
-            )
-          }),
-        )
-
-        if Env.Benchmark.shouldSaveData {
-          let elapsedTimeMillis = Hrtime.timeSince(timeRef)->Hrtime.toMillis->Hrtime.floatFromMillis
 
-          Benchmark.addSummaryData(
-            ~group="Other",
-            ~label="Prune Stale History Time (ms)",
-            ~value=elapsedTimeMillis,
+        for idx in 0 to Entities.allEntities->Array.length - 1 {
+          if idx !== 0 {
+            // Add some delay between entities
+            // To unblock the pg client if it's needed for something else
+            await Utils.delay(400)
+          }
+          let entityConfig = Entities.allEntities->Array.getUnsafe(idx)
+          let timeRef = Hrtime.makeTimer()
+          await Db.sql->DbFunctions.EntityHistory.pruneStaleEntityHistory(
+            ~entityName=entityConfig.name,
+            ~safeChainIdAndBlockNumberArray,
+            ~shouldDeepClean,
+          )
+          Prometheus.RollbackHistoryPrune.increment(
+            ~timeMillis=Hrtime.timeSince(timeRef)->Hrtime.toMillis,
+            ~entityName=entityConfig.name,
           )
         }
       }

scenarios/test_codegen/schema.graphql

@@ -114,6 +114,7 @@ type EntityWithAllTypes {
   arrayOfBigInts: [BigInt!]!
   bigDecimal: BigDecimal!
   optBigDecimal: BigDecimal
+  bigDecimalWithConfig: BigDecimal! @config(precision: 10, scale: 8)
   arrayOfBigDecimals: [BigDecimal!]!
   # NOTE: Timestamp serialization is currently just a type cast and so testing is non deterministic
   # timestamp: Timestamp!
@@ -141,6 +142,7 @@ type EntityWithAllNonArrayTypes {
   optBigInt: BigInt
   bigDecimal: BigDecimal!
   optBigDecimal: BigDecimal
+  bigDecimalWithConfig: BigDecimal! @config(precision: 10, scale: 8)
   enumField: AccountType!
   optEnumField: AccountType
 }

scenarios/test_codegen/src/EventHandlers.ts

@@ -540,6 +540,7 @@ Gravatar.FactoryEvent.handlerWithLoader({
           optBigInt: 2n,
           arrayOfBigInts: [3n, 4n],
           bigDecimal: new BigDecimal("1.1"),
+          bigDecimalWithConfig: new BigDecimal("1.1"),
           optBigDecimal: new BigDecimal("2.2"),
           arrayOfBigDecimals: [new BigDecimal("3.3"), new BigDecimal("4.4")],
           json: { foo: ["bar"] },
@@ -563,6 +564,7 @@ Gravatar.FactoryEvent.handlerWithLoader({
           optBigInt: 2n,
           arrayOfBigInts: [3n, 4n],
           bigDecimal: new BigDecimal("1.1"),
+          bigDecimalWithConfig: new BigDecimal("1.1"),
           optBigDecimal: new BigDecimal("2.2"),
           arrayOfBigDecimals: [new BigDecimal("3.3"), new BigDecimal("4.4")],
           json: { foo: ["bar"] },

scenarios/test_codegen/test/ChainManager_test.res

@@ -80,7 +80,7 @@ let populateChainQueuesWithRandomEvents = (~runTime=1000, ~maxBlockTime=15, ())
               blockNumber: batchItem.blockNumber,
               blockTimestamp: batchItem.timestamp,
             },
-            ~reversedNewItems=[batchItem],
+            ~newItems=[batchItem],
             ~currentBlockHeight=currentBlockNumber.contents,
           )
           ->Result.getExn

scenarios/test_codegen/test/SerDe_Test.res

@@ -34,6 +34,7 @@ describe("SerDe Test", () => {
       optBigInt: Some(BigInt.fromInt(2)),
       arrayOfBigInts: [BigInt.fromInt(3), BigInt.fromInt(4)],
       bigDecimal: BigDecimal.fromStringUnsafe("1.1"),
+      bigDecimalWithConfig: BigDecimal.fromStringUnsafe("1.1"),
       optBigDecimal: Some(BigDecimal.fromStringUnsafe("2.2")),
       arrayOfBigDecimals: [BigDecimal.fromStringUnsafe("3.3"), BigDecimal.fromStringUnsafe("4.4")],
       //TODO: get timestamp working
@@ -123,7 +124,7 @@ describe("SerDe Test", () => {
       Entities.EntityWithAllNonArrayTypes.table->PgStorage.makeCreateTableQuery(~pgSchema="public")
     Assert.equal(
       createQuery,
-      `CREATE TABLE IF NOT EXISTS "public"."EntityWithAllNonArrayTypes"("bigDecimal" NUMERIC NOT NULL, "bigInt" NUMERIC NOT NULL, "bool" BOOLEAN NOT NULL, "enumField" "public".AccountType NOT NULL, "float_" DOUBLE PRECISION NOT NULL, "id" TEXT NOT NULL, "int_" INTEGER NOT NULL, "optBigDecimal" NUMERIC, "optBigInt" NUMERIC, "optBool" BOOLEAN, "optEnumField" "public".AccountType, "optFloat" DOUBLE PRECISION, "optInt" INTEGER, "optString" TEXT, "string" TEXT NOT NULL, "db_write_timestamp" TIMESTAMP DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY("id"));`,
+      `CREATE TABLE IF NOT EXISTS "public"."EntityWithAllNonArrayTypes"("bigDecimal" NUMERIC NOT NULL, "bigDecimalWithConfig" NUMERIC(10, 8) NOT NULL, "bigInt" NUMERIC NOT NULL, "bool" BOOLEAN NOT NULL, "enumField" "public".AccountType NOT NULL, "float_" DOUBLE PRECISION NOT NULL, "id" TEXT NOT NULL, "int_" INTEGER NOT NULL, "optBigDecimal" NUMERIC, "optBigInt" NUMERIC, "optBool" BOOLEAN, "optEnumField" "public".AccountType, "optFloat" DOUBLE PRECISION, "optInt" INTEGER, "optString" TEXT, "string" TEXT NOT NULL, "db_write_timestamp" TIMESTAMP DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY("id"));`,
     )
     let query = PgStorage.makeInsertUnnestSetQuery(
       ~table=Entities.EntityWithAllNonArrayTypes.table,
@@ -134,8 +135,8 @@ describe("SerDe Test", () => {
 
     Assert.equal(
       query,
-      `INSERT INTO "public"."EntityWithAllNonArrayTypes" ("bigDecimal", "bigInt", "bool", "enumField", "float_", "id", "int_", "optBigDecimal", "optBigInt", "optBool", "optEnumField", "optFloat", "optInt", "optString", "string")
-SELECT * FROM unnest($1::NUMERIC[],$2::NUMERIC[],$3::INTEGER[]::BOOLEAN[],$4::TEXT[]::"public".AccountType[],$5::DOUBLE PRECISION[],$6::TEXT[],$7::INTEGER[],$8::NUMERIC[],$9::NUMERIC[],$10::INTEGER[]::BOOLEAN[],$11::TEXT[]::"public".AccountType[],$12::DOUBLE PRECISION[],$13::INTEGER[],$14::TEXT[],$15::TEXT[])ON CONFLICT("id") DO UPDATE SET "bigDecimal" = EXCLUDED."bigDecimal","bigInt" = EXCLUDED."bigInt","bool" = EXCLUDED."bool","enumField" = EXCLUDED."enumField","float_" = EXCLUDED."float_","int_" = EXCLUDED."int_","optBigDecimal" = EXCLUDED."optBigDecimal","optBigInt" = EXCLUDED."optBigInt","optBool" = EXCLUDED."optBool","optEnumField" = EXCLUDED."optEnumField","optFloat" = EXCLUDED."optFloat","optInt" = EXCLUDED."optInt","optString" = EXCLUDED."optString","string" = EXCLUDED."string";`,
+      `INSERT INTO "public"."EntityWithAllNonArrayTypes" ("bigDecimal", "bigDecimalWithConfig", "bigInt", "bool", "enumField", "float_", "id", "int_", "optBigDecimal", "optBigInt", "optBool", "optEnumField", "optFloat", "optInt", "optString", "string")
+SELECT * FROM unnest($1::NUMERIC[],$2::NUMERIC(10, 8)[],$3::NUMERIC[],$4::INTEGER[]::BOOLEAN[],$5::TEXT[]::"public".AccountType[],$6::DOUBLE PRECISION[],$7::TEXT[],$8::INTEGER[],$9::NUMERIC[],$10::NUMERIC[],$11::INTEGER[]::BOOLEAN[],$12::TEXT[]::"public".AccountType[],$13::DOUBLE PRECISION[],$14::INTEGER[],$15::TEXT[],$16::TEXT[])ON CONFLICT("id") DO UPDATE SET "bigDecimal" = EXCLUDED."bigDecimal","bigDecimalWithConfig" = EXCLUDED."bigDecimalWithConfig","bigInt" = EXCLUDED."bigInt","bool" = EXCLUDED."bool","enumField" = EXCLUDED."enumField","float_" = EXCLUDED."float_","int_" = EXCLUDED."int_","optBigDecimal" = EXCLUDED."optBigDecimal","optBigInt" = EXCLUDED."optBigInt","optBool" = EXCLUDED."optBool","optEnumField" = EXCLUDED."optEnumField","optFloat" = EXCLUDED."optFloat","optInt" = EXCLUDED."optInt","optString" = EXCLUDED."optString","string" = EXCLUDED."string";`,
     )
   })
 
@@ -156,6 +157,7 @@ SELECT * FROM unnest($1::NUMERIC[],$2::NUMERIC[],$3::INTEGER[]::BOOLEAN[],$4::TE
       optBigInt: Some(BigInt.fromInt(2)),
       bigDecimal: BigDecimal.fromStringUnsafe("1.1"),
       optBigDecimal: Some(BigDecimal.fromStringUnsafe("2.2")),
+      bigDecimalWithConfig: BigDecimal.fromStringUnsafe("1.1"),
       enumField: ADMIN,
       optEnumField: Some(ADMIN),
     }