Refactor indexing primitives and optimize bulk creation

.github/workflows/ci.yml

@@ -35,7 +35,11 @@ jobs:
       # use a matrix here to speed up the process of build both docker images.
       matrix:
         environment: [ "production", "development" ]
-        platform: [ "linux/amd64", "linux/arm64" ]
+        platform:
+          - "linux/amd64"
+          # Disabled this until we research why this takes more than 1 hour to build something that
+          # take less than 10 minutes on linux/amd64
+          # - "linux/arm64"
     with:
       env: ${{ matrix.environment }}
       platform: ${{ matrix.platform }}

project.ts

@@ -435,7 +435,7 @@ const project: CosmosProject = {
             // and parallel operations.
             handler: "indexingHandler",
             kind: CosmosHandlerKind.Block,
-          }
+          },
         ],
       },
     },

scripts/node-entrypoint.sh

@@ -10,7 +10,10 @@ set -e
 # all this is to been able to join WATCH and Normal execution in a single dockerfile
 cmd="env NODE_OPTIONS=$NODE_OPTIONS \
 NODE_ENV=$NODE_ENV \
-CHAIN_ID=$CHAIN_ID"
+CHAIN_ID=$CHAIN_ID \
+BATCH_SIZE=$BATCH_SIZE \
+START_BLOCK=$START_BLOCK \
+DB_SCHEMA=$DB_SCHEMA"
 
 if [ "$NODE_ENV" = "test" ]
 then
@@ -37,7 +40,10 @@ EOF
     DB_PASS=$DB_PASS \
     DB_DATABASE=$DB_DATABASE \
     DB_HOST=$DB_HOST \
-    DB_PORT=$DB_PORT"
+    DB_PORT=$DB_PORT \
+    POCKETDEX_DB_PAGE_LIMIT=$POCKETDEX_DB_PAGE_LIMIT \
+    POCKETDEX_DB_BATCH_SIZE=$POCKETDEX_DB_BATCH_SIZE \
+    POCKETDEX_DB_BULK_WRITE_CONCURRENCY=$POCKETDEX_DB_BULK_WRITE_CONCURRENCY"
 
   if [ "$NODE_ENV" = "development" ]
   then
@@ -58,9 +64,15 @@ EOF
     # move the dist folder to the mounted folder in run time
     update_project
     params=$(get_params)
+
+
     # run the main node
     cmd="$cmd node ./node_modules/@subql/node-cosmos/bin/run $params $@"
   fi
 fi
 
+# Sanitize and log the command before execution
+sanitized_cmd=$(sanitize_cmd "$cmd")
+info_log "Executing command: $sanitized_cmd"
+
 eval $cmd

scripts/shared.sh

@@ -77,3 +77,8 @@ get_env_file_name(){
   fi
   echo "$file"
 }
+
+# Function to sanitize sensitive values in the command string for logging
+sanitize_cmd() {
+  echo "$1" | sed "s/DB_PASS=[^ ]*/DB_PASS=****/g"
+}

src/mappings/bank/moduleAccounts.ts

@@ -8,7 +8,6 @@ import {
   getBalanceId,
   getBlockId,
 } from "../utils/ids";
-import { stringify } from "../utils/json";
 import getQueryClient from "../utils/query_client";
 import { enforceAccountsExists } from "./balanceChange";
 
@@ -53,7 +52,7 @@ export async function queryModuleAccounts(block: CosmosBlock): Promise<Array<Ext
     ),
   );
 
-  logger.debug(`[queryModuleAccounts] extendedAccounts=${stringify(extendedAccounts, undefined, 2)}`);
+  logger.debug(`[queryModuleAccounts] extendedAccounts=${extendedAccounts.length}`);
 
   return extendedAccounts;
 }

src/mappings/bank/supply.ts

@@ -40,7 +40,6 @@ export async function queryTotalSupply(block: CosmosBlock): Promise<Coin[]> {
 
   // Initial call to get the first set of results
   const initialResponse: QueryTotalSupplyResponse = await queryClient.bank.totalSupply() as unknown as QueryTotalSupplyResponse;
-  logger.debug(`[handleTotalSupply]: initialResponse=${stringify(initialResponse, undefined, 2)}`);
   finalSupply.push(...initialResponse.supply);
   paginationKey = initialResponse.pagination?.nextKey;
 

src/mappings/indexer.manager.ts

@@ -36,7 +36,10 @@ import {
   EventByType,
   MessageByType,
 } from "./types/common";
-import { GetIdFromEventAttribute, RecordGetId } from "./types/stake";
+import {
+  GetIdFromEventAttribute,
+  RecordGetId,
+} from "./types/stake";
 import { optimizedBulkCreate } from "./utils/db";
 import { getBlockId } from "./utils/ids";
 import { stringify } from "./utils/json";
@@ -84,15 +87,12 @@ function handleByType(typeUrl: string | Array<string>, byTypeMap: MessageByType
 }
 
 // anything primitive types
-async function indexPrimitives(block: CosmosBlock, events: Array<CosmosEvent>) {
+async function indexPrimitives(block: CosmosBlock) {
   await profilerWrap(handleGenesis, "indexPrimitives", "handleGenesis")(block);
 
   await Promise.all([
     profilerWrap(handleBlock, "indexPrimitives", "handleGenesis")(block),
     profilerWrap(handleSupply, "indexPrimitives", "handleSupply")(block),
-    profilerWrap(handleTransactions, "indexPrimitives", "handleTransactions")(block.transactions),
-    profilerWrap(handleMessages, "indexPrimitives", "handleMessages")(block.messages),
-    profilerWrap(handleEvents, "indexPrimitives", "handleEvents")(events),
   ]);
 }
 
@@ -690,19 +690,33 @@ async function _indexingHandler(block: CosmosBlock): Promise<void> {
     logger.warn(`[indexer.manager] unhandledEventTypes=${stringify(Array.from(unhandledEventTypes))} eventsByType=${stringify(eventsByType, jsonArrayCounter, 0)}`);
   }
 
-  await profilerWrap(indexPrimitives, "indexingHandler", "indexPrimitives")(block, filteredEvents);
+  // let's serialize to maybe avoid issues do to even loop saturation?
+  await profilerWrap(indexPrimitives, "indexingHandler", "indexPrimitives")(block);
 
-  await Promise.all([
-    profilerWrap(indexBalances, "indexingHandler", "indexBalances")(block, msgsByType as MessageByType, eventsByType),
-    profilerWrap(indexParams, "indexingHandler", "indexParams")(msgsByType as MessageByType),
-    profilerWrap(indexGrants, "indexingHandler", "indexGrants")(msgsByType as MessageByType, eventsByType),
-    profilerWrap(indexService, "indexingHandler", "indexService")(msgsByType as MessageByType, eventsByType),
-    profilerWrap(indexValidators, "indexingHandler", "indexValidators")(msgsByType as MessageByType, eventsByType),
-    profilerWrap(indexStake, "indexingHandler", "indexStake")(msgsByType as MessageByType, eventsByType),
-    profilerWrap(indexRelays, "indexingHandler", "indexRelays")(msgsByType as MessageByType, eventsByType),
-    profilerWrap(indexMigrationAccounts, "indexingHandler", "indexMigrationAccounts")(msgsByType as MessageByType),
-    profilerWrap(generateReports, "indexingHandler", "generateReports")(block),
-  ])
+  // lets this happens first because is massive
+  await profilerWrap(handleEvents, "indexPrimitives", "handleEvents")(filteredEvents);
+
+  await profilerWrap(handleTransactions, "indexPrimitives", "handleTransactions")(block.transactions);
+
+  await profilerWrap(handleMessages, "indexPrimitives", "handleMessages")(block.messages);
+
+  await profilerWrap(indexBalances, "indexingHandler", "indexBalances")(block, msgsByType as MessageByType, eventsByType);
+
+  await profilerWrap(indexParams, "indexingHandler", "indexParams")(msgsByType as MessageByType);
+
+  await profilerWrap(indexGrants, "indexingHandler", "indexGrants")(msgsByType as MessageByType, eventsByType);
+
+  await profilerWrap(indexService, "indexingHandler", "indexService")(msgsByType as MessageByType, eventsByType);
+
+  await profilerWrap(indexValidators, "indexingHandler", "indexValidators")(msgsByType as MessageByType, eventsByType);
+
+  await profilerWrap(indexStake, "indexingHandler", "indexStake")(msgsByType as MessageByType, eventsByType);
+
+  await profilerWrap(indexRelays, "indexingHandler", "indexRelays")(msgsByType as MessageByType, eventsByType);
+
+  await profilerWrap(indexMigrationAccounts, "indexingHandler", "indexMigrationAccounts")(msgsByType as MessageByType);
+
+  await profilerWrap(generateReports, "indexingHandler", "generateReports")(block);
 }
 
 export async function indexingHandler(block: CosmosBlock): Promise<void> {

src/mappings/pocket/migration.ts

@@ -127,7 +127,7 @@ export async function updateMorseClaimableAccounts(
       {
         hooks: false,
         where: {
-          id: morseAddress ? morseAddress : pubKeyToAddress(
+          id: morseAddress ? morseAddress.toLowerCase() : pubKeyToAddress(
             Ed25519,
             publicKey!,
             undefined,

src/mappings/pocket/suppliers.ts

@@ -5,26 +5,27 @@ import {
 } from "@subql/types-cosmos";
 import { Coin } from "../../client/cosmos/base/v1beta1/coin";
 import {
+  EventSupplierServiceConfigActivated,
   EventSupplierUnbondingBegin as EventSupplierUnbondingBeginEntity,
   EventSupplierUnbondingEnd as EventSupplierUnbondingEndEntity,
-  MsgStakeSupplier as MsgStakeSupplierEntity,
+  MorseSupplierClaimSignerType,
   MsgClaimMorseSupplier as MsgClaimMorseSupplierEntity,
+  MsgStakeSupplier as MsgStakeSupplierEntity,
   MsgUnstakeSupplier as MsgUnstakeSupplierEntity,
   StakeStatus,
   Supplier,
   SupplierEndpoint,
   SupplierRevShare,
   SupplierServiceConfig,
   SupplierUnbondingReason,
-  MorseSupplierClaimSignerType,
-  EventSupplierServiceConfigActivated,
 } from "../../types";
 import { MsgClaimMorseSupplierProps } from "../../types/models/MsgClaimMorseSupplier";
 import { MsgStakeSupplierServiceProps } from "../../types/models/MsgStakeSupplierService";
 import { SupplierServiceConfigProps } from "../../types/models/SupplierServiceConfig";
 import { CoinSDKType } from "../../types/proto-interfaces/cosmos/base/v1beta1/coin";
+import { MorseSupplierClaimSignerTypeSDKType } from "../../types/proto-interfaces/pocket/migration/morse_onchain";
 import { MsgClaimMorseSupplier } from "../../types/proto-interfaces/pocket/migration/tx";
-import {SupplierServiceConfig as SupplierServiceConfigType} from '../../types/proto-interfaces/pocket/shared/service'
+import { SupplierServiceConfig as SupplierServiceConfigType } from "../../types/proto-interfaces/pocket/shared/service";
 import { SupplierSDKType } from "../../types/proto-interfaces/pocket/shared/supplier";
 import {
   supplierUnbondingReasonFromJSON,
@@ -42,10 +43,12 @@ import {
   getStakeServiceId,
   messageId,
 } from "../utils/ids";
-import { Ed25519, pubKeyToAddress } from "../utils/pub_key";
+import {
+  Ed25519,
+  pubKeyToAddress,
+} from "../utils/pub_key";
 import { updateMorseClaimableAccounts } from "./migration";
 import { fetchAllSupplierServiceConfigBySupplier } from "./pagination";
-import { MorseSupplierClaimSignerTypeSDKType } from "../../types/proto-interfaces/pocket/migration/morse_onchain";
 
 function getMorseSupplierClaimSignerType(item: typeof MorseSupplierClaimSignerTypeSDKType | string | number): MorseSupplierClaimSignerType {
   switch (item) {