Refactor the indexing process and improve configuration management.

We were trigger a lot of promises in parallel, and for some reason when we do this, Sequelize in the background hangs the transactions.

Lowering the number of parallel promises looks like giving more free space into the event loop, which allows To Sequelize to do his job.

- Reordered asynchronous operations in `indexer.manager.ts` for better handling of event processing.
- Introduced a new function `sanitize_cmd` in `shared.sh` to redact sensitive command-line values for logging.
- Updated default values for `BATCH_SIZE` and `CONCURRENCY` in `db.ts`, and added a log entry for global config parameters.
- Enhanced `node-entrypoint.sh` to include additional environment variables and sanitized command logging for execution safety.
- Fixed minor formatting issues in `project.ts` handler configuration.

Author: jorgecuesta

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/indexer.manager.ts

@@ -690,21 +690,31 @@ async function _indexingHandler(block: CosmosBlock): Promise<void> {
     logger.warn(`[indexer.manager] unhandledEventTypes=${stringify(Array.from(unhandledEventTypes))} eventsByType=${stringify(eventsByType, jsonArrayCounter, 0)}`);
   }
 
+  // let's serialize to maybe avoid issues do to even loop saturation?
   await profilerWrap(indexPrimitives, "indexingHandler", "indexPrimitives")(block);
 
-  await Promise.all([
-    profilerWrap(handleTransactions, "indexPrimitives", "handleTransactions")(block.transactions),
-    profilerWrap(handleMessages, "indexPrimitives", "handleMessages")(block.messages),
-    profilerWrap(handleEvents, "indexPrimitives", "handleEvents")(filteredEvents),
-    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),
-  ]);
+  // 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);
 }

src/mappings/utils/db.ts

@@ -27,9 +27,10 @@ type Transformer<T> = (doc: T) => any;
 
 // PAGE_LIMIT should be less or equal to --query-limit=<value> otherwise Subql will throw an error.
 const PAGE_LIMIT = isNumber(process.env.POCKETDEX_DB_PAGE_LIMIT) ? Number(process.env.POCKETDEX_DB_PAGE_LIMIT) : 1000;
-const BATCH_SIZE = isNumber(process.env.POCKETDEX_DB_BATCH_SIZE) ? Number(process.env.POCKETDEX_DB_BATCH_SIZE) : 15000;
-const CONCURRENCY = isNumber(process.env.POCKETDEX_DB_BULK_WRITE_CONCURRENCY) ? Number(process.env.POCKETDEX_DB_BULK_WRITE_CONCURRENCY) : 10;
+const BATCH_SIZE = isNumber(process.env.POCKETDEX_DB_BATCH_SIZE) ? Number(process.env.POCKETDEX_DB_BATCH_SIZE) : 5000;
+const CONCURRENCY = isNumber(process.env.POCKETDEX_DB_BULK_WRITE_CONCURRENCY) ? Number(process.env.POCKETDEX_DB_BULK_WRITE_CONCURRENCY) : 5;
 
+logger.info(`[Global] PAGE_LIMIT=${PAGE_LIMIT} BATCH_SIZE=${BATCH_SIZE} CONCURRENCY=${CONCURRENCY}`);
 
 /**
  * Retrieves the store model associated with the given name.