🐛 Fix indexTokenBalances

Author: sameersubudhi

services/blockchain-indexer/config.js

@@ -110,6 +110,13 @@ config.queue = {
 	},
 };
 
+config.set = {
+	accountBalanceUpdate: {
+		name: 'AccountBalanceUpdate',
+		batchSize: 1000,
+	},
+};
+
 config.operations = {
 	isDataRetrievalModeEnabled: Boolean(
 		String(process.env.ENABLE_DATA_RETRIEVAL_MODE).toLowerCase() !== 'false',

services/blockchain-indexer/shared/indexer/accountBalanceIndex.js

@@ -32,9 +32,6 @@ const redis = new Redis(config.endpoints.cache);
 
 const MYSQL_ENDPOINT = config.endpoints.mysql;
 
-const ACCOUNTS_BALANCE_UPDATE_SET_NAME = 'AccountsBalanceUpdate';
-const MAX_ACCOUNT_COUNT_IN_ONE_EXECUTION = 1000; // 1e3
-
 const getAccountBalancesTable = () => getTableInstance(accountBalancesTableSchema, MYSQL_ENDPOINT);
 
 const updateAccountBalances = async address => {
@@ -53,7 +50,7 @@ const updateAccountBalances = async address => {
 
 const scheduleAddressesBalanceUpdate = async addresses => {
 	if (addresses.length) {
-		redis.sadd(ACCOUNTS_BALANCE_UPDATE_SET_NAME, addresses);
+		redis.sadd(config.set.accountBalanceUpdate.name, addresses);
 	}
 };
 
@@ -79,8 +76,8 @@ const getAddressesFromTokenEvents = events => {
 
 const triggerAccountsBalanceUpdate = async () => {
 	const addresses = await redis.spop(
-		ACCOUNTS_BALANCE_UPDATE_SET_NAME,
-		MAX_ACCOUNT_COUNT_IN_ONE_EXECUTION,
+		config.set.accountBalanceUpdate.name,
+		config.set.accountBalanceUpdate.batchSize,
 	);
 
 	try {

services/blockchain-indexer/shared/indexer/genesisBlock.js

@@ -18,19 +18,22 @@ const {
 		MySQL: { getTableInstance },
 	},
 	Signals,
+	Logger,
 } = require('lisk-service-framework');
 
 const { MODULE, MODULE_SUB_STORE, getGenesisHeight } = require('../constants');
 const { updateTotalStake, updateTotalSelfStake } = require('./transactionProcessor/pos/stake');
 const { requestConnector } = require('../utils/request');
-const { accountBalanceIndexQueue } = require('./accountBalanceIndex');
+const { updateAccountBalances } = require('./accountBalanceIndex');
 const { updateTotalLockedAmounts } = require('./utils/blockchainIndex');
 
 const requestAll = require('../utils/requestAll');
 const config = require('../../config');
 const commissionsTableSchema = require('../database/schema/commissions');
 const { getIndexStats } = require('./indexStatus');
 
+const logger = Logger();
+
 const MYSQL_ENDPOINT = config.endpoints.mysql;
 
 const getCommissionsTable = () => getTableInstance(commissionsTableSchema, MYSQL_ENDPOINT);
@@ -144,7 +147,16 @@ const indexGenesisBlockAssets = async dbTrx => {
 };
 
 const indexTokenBalances = async () => {
-	allAccountsAddresses.forEach(async address => accountBalanceIndexQueue.add({ address }));
+	// eslint-disable-next-line no-restricted-syntax
+	for (const address of allAccountsAddresses) {
+		await updateAccountBalances(address).catch(err => {
+			const errorMessage = `Updating account balance for ${address} failed. Retrying.\nError: ${err.message}.`;
+			logger.warn(errorMessage);
+			logger.debug(err.stack);
+
+			allAccountsAddresses.push(address);
+		});
+	}
 	isTokensBalanceIndexed = true;
 };