diff --git a/package.json b/package.json index a1ee899..cea9f88 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@multiversx/sdk-transaction-processor", - "version": "0.1.35", + "version": "0.1.36", "description": "Real-time transaction processor", "main": "lib/transaction.processor.js", "types": "lib/transaction.processor.d.ts", @@ -33,4 +33,4 @@ "dependencies": { "axios": "^1.7.4" } -} \ No newline at end of file +} diff --git a/src/transaction.processor.ts b/src/transaction.processor.ts index 91739b1..9be7cd9 100644 --- a/src/transaction.processor.ts +++ b/src/transaction.processor.ts @@ -70,42 +70,32 @@ export class TransactionProcessor { do { reachedTip = true; - for (const shardId of this.shardIds) { - const currentNonce = currentNonces[shardId]; - let lastProcessedNonce = await this.getLastProcessedNonceOrCurrent(shardId, currentNonce); - - this.logMessage(LogTopic.Debug, `shardId: ${shardId}, currentNonce: ${currentNonce}, lastProcessedNonce: ${lastProcessedNonce}`); - - if (lastProcessedNonce === currentNonce) { - this.logMessage(LogTopic.Debug, 'lastProcessedNonce === currentNonce'); + const shardBlockResults = await Promise.allSettled( + this.shardIds.map(shardId => this.fetchNextShardBlock(shardId, currentNonces[shardId], options)), + ); + + for (const [index, result] of shardBlockResults.entries()) { + if (result.status === 'rejected') { + const shardId = this.shardIds[index]; + const reason = result.reason instanceof Error ? result.reason.message : String(result.reason); + this.logMessage(LogTopic.Error, `Failed to fetch next shard block for shardId ${shardId}: ${reason}`); + reachedTip = false; continue; } - // this is to handle the situation where the current nonce is reset - // (e.g. devnet/testnet reset where the nonces start again from zero) - if (lastProcessedNonce > currentNonce + NETWORK_RESET_NONCE_THRESHOLD) { - this.logMessage(LogTopic.Debug, `Detected network reset. Setting last processed nonce to ${currentNonce} for shard ${shardId}`); - lastProcessedNonce = currentNonce; - } - - if (lastProcessedNonce > currentNonce) { - this.logMessage(LogTopic.Debug, 'lastProcessedNonce > currentNonce'); + const shardBlock = result.value; + if (shardBlock == null) { continue; } - if (options.maxLookBehind && currentNonce - lastProcessedNonce > options.maxLookBehind) { - lastProcessedNonce = currentNonce - options.maxLookBehind; - } + const { shardId, currentNonce, lastProcessedNonce, nonce, transactionsResult } = shardBlock; if (!startLastProcessedNonces[shardId]) { startLastProcessedNonces[shardId] = lastProcessedNonce; } - const nonce = lastProcessedNonce + 1; - - const transactionsResult = await this.getShardTransactions(shardId, nonce); - if (transactionsResult === undefined) { - this.logMessage(LogTopic.Debug, 'transactionsResult === undefined'); + if (transactionsResult == null) { + this.logMessage(LogTopic.Debug, 'transactionsResult is null'); continue; } @@ -224,8 +214,8 @@ export class TransactionProcessor { const nonce = lastProcessedNonce + 1; const transactionsResult = await this.getHyperblockTransactions(nonce); - if (transactionsResult === undefined) { - this.logMessage(LogTopic.Debug, 'transactionsResult === undefined'); + if (transactionsResult == null) { + this.logMessage(LogTopic.Debug, 'transactionsResult is null'); continue; } @@ -343,6 +333,48 @@ export class TransactionProcessor { return crossShardTransactions; } + private async fetchNextShardBlock( + shardId: number, + currentNonce: number, + options: TransactionProcessorOptions, + ): Promise<{ + shardId: number; + currentNonce: number; + lastProcessedNonce: number; + nonce: number; + transactionsResult: { blockHash: string, transactions: ShardTransaction[] } | undefined; + } | undefined> { + let lastProcessedNonce = await this.getLastProcessedNonceOrCurrent(shardId, currentNonce); + + this.logMessage(LogTopic.Debug, `shardId: ${shardId}, currentNonce: ${currentNonce}, lastProcessedNonce: ${lastProcessedNonce}`); + + if (lastProcessedNonce === currentNonce) { + this.logMessage(LogTopic.Debug, 'lastProcessedNonce === currentNonce'); + return undefined; + } + + // this is to handle the situation where the current nonce is reset + // (e.g. devnet/testnet reset where the nonces start again from zero) + if (lastProcessedNonce > currentNonce + NETWORK_RESET_NONCE_THRESHOLD) { + this.logMessage(LogTopic.Debug, `Detected network reset. Setting last processed nonce to ${currentNonce} for shard ${shardId}`); + lastProcessedNonce = currentNonce; + } + + if (lastProcessedNonce > currentNonce) { + this.logMessage(LogTopic.Debug, 'lastProcessedNonce > currentNonce'); + return undefined; + } + + if (options.maxLookBehind && currentNonce - lastProcessedNonce > options.maxLookBehind) { + lastProcessedNonce = currentNonce - options.maxLookBehind; + } + + const nonce = lastProcessedNonce + 1; + const transactionsResult = await this.getShardTransactions(shardId, nonce); + + return { shardId, currentNonce, lastProcessedNonce, nonce, transactionsResult }; + } + private async getShardTransactions(shardId: number, nonce: number): Promise<{ blockHash: string, transactions: ShardTransaction[] } | undefined> { const result = await this.gatewayGet(`block/${shardId}/by-nonce/${nonce}?withTxs=true`); diff --git a/tsconfig.json b/tsconfig.json index 4dc02af..e352038 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -1,6 +1,7 @@ { "compilerOptions": { "target": "es2015", + "lib": ["es2020"], "module": "commonjs", "declaration": true, "outDir": "./lib",