diff --git a/src/transaction.processor.ts b/src/transaction.processor.ts index 9be7cd9..2362740 100644 --- a/src/transaction.processor.ts +++ b/src/transaction.processor.ts @@ -1,6 +1,6 @@ import { GatewayBlockResponse } from './types/gateway/block-response'; import { GatewayMiniblockProcessingType } from './types/gateway/miniblock-processing-type.enum'; -import { METACHAIN, NETWORK_RESET_NONCE_THRESHOLD } from './utils/constants'; +import { DEFAULT_MAX_PREFETCH, METACHAIN, NETWORK_RESET_NONCE_THRESHOLD } from './utils/constants'; import { TransactionProcessorMode } from './types/transaction-processor-mode.enum'; import { LogTopic } from './types/log-topic'; import { TransactionStatistics } from './types/transaction-statistics'; @@ -13,6 +13,9 @@ import { GatewayMiniblock } from './types/gateway/miniblock'; import { GatewayTransaction } from './types/gateway/transaction'; import { ShardsMaintainerService } from './shards-maintainer.service'; import { HttpService } from './utils/http.service'; +import { BlockPrefetcher } from './utils/block-prefetcher'; + +type BlockTransactions = { blockHash: string, transactions: ShardTransaction[] }; export class TransactionProcessor { private startDate: Date = new Date(); @@ -24,6 +27,18 @@ export class TransactionProcessor { private httpService: HttpService | undefined; private readonly shardsMaintainerService: ShardsMaintainerService = new ShardsMaintainerService(); + // Kept on the instance rather than per pass: start() is typically re-entered by a sub-second + // cron, so blocks read ahead near the end of one pass are still warm at the start of the next. + private readonly shardBlockPrefetcher = new BlockPrefetcher( + (shardId, nonce) => this.getShardTransactions(shardId, nonce), + (shardId, nonce, error) => this.logMessage(LogTopic.Debug, `Could not read block for shardId ${shardId} and nonce ${nonce}: ${error}`), + ); + + private readonly hyperblockPrefetcher = new BlockPrefetcher( + (_shardId, nonce) => this.getHyperblockTransactions(nonce), + (_shardId, nonce, error) => this.logMessage(LogTopic.Debug, `Could not read hyperblock for nonce ${nonce}: ${error}`), + ); + async start(options: TransactionProcessorOptions): Promise { this.options = options; this.httpService = new HttpService(this.options.gatewayUrl, this.options.timeout); @@ -94,8 +109,8 @@ export class TransactionProcessor { startLastProcessedNonces[shardId] = lastProcessedNonce; } - if (transactionsResult == null) { - this.logMessage(LogTopic.Debug, 'transactionsResult is null'); + if (transactionsResult === undefined) { + this.logMessage(LogTopic.Debug, 'transactionsResult === undefined'); continue; } @@ -196,6 +211,7 @@ export class TransactionProcessor { if (lastProcessedNonce > currentNonce + NETWORK_RESET_NONCE_THRESHOLD) { this.logMessage(LogTopic.Debug, `Detected network reset. Setting last processed nonce to ${currentNonce}`); lastProcessedNonce = currentNonce; + this.hyperblockPrefetcher.clearShard(METACHAIN); } if (lastProcessedNonce > currentNonce) { @@ -213,9 +229,12 @@ export class TransactionProcessor { const nonce = lastProcessedNonce + 1; - const transactionsResult = await this.getHyperblockTransactions(nonce); - if (transactionsResult == null) { - this.logMessage(LogTopic.Debug, 'transactionsResult is null'); + this.hyperblockPrefetcher.prune(METACHAIN, nonce); + this.hyperblockPrefetcher.prime(METACHAIN, nonce, nonce + this.getPrefetchSize(currentNonce - lastProcessedNonce) - 1); + + const transactionsResult = await this.hyperblockPrefetcher.take(METACHAIN, nonce); + if (transactionsResult === undefined) { + this.logMessage(LogTopic.Debug, 'transactionsResult === undefined'); continue; } @@ -333,6 +352,17 @@ export class TransactionProcessor { return crossShardTransactions; } + /** + * Size of the read-ahead window for a shard that is `noncesBehind` blocks behind the tip. + * Never exceeds the distance to the tip: at the tip there is nothing to speculate on, and + * reading past it would only ask the gateway for blocks that do not exist yet. + */ + private getPrefetchSize(noncesBehind: number): number { + const maxPrefetch = this.options.maxPrefetch ?? DEFAULT_MAX_PREFETCH; + + return Math.max(1, Math.min(maxPrefetch, noncesBehind)); + } + private async fetchNextShardBlock( shardId: number, currentNonce: number, @@ -342,7 +372,7 @@ export class TransactionProcessor { currentNonce: number; lastProcessedNonce: number; nonce: number; - transactionsResult: { blockHash: string, transactions: ShardTransaction[] } | undefined; + transactionsResult: BlockTransactions | undefined; } | undefined> { let lastProcessedNonce = await this.getLastProcessedNonceOrCurrent(shardId, currentNonce); @@ -358,6 +388,8 @@ export class TransactionProcessor { 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; + // Everything read ahead belongs to the pre-reset chain and can never be delivered. + this.shardBlockPrefetcher.clearShard(shardId); } if (lastProcessedNonce > currentNonce) { @@ -370,12 +402,19 @@ export class TransactionProcessor { } const nonce = lastProcessedNonce + 1; - const transactionsResult = await this.getShardTransactions(shardId, nonce); + + // Top the read-ahead window up before waiting on this nonce, so that while this block is + // being handed to the consumer the reads for the following nonces are already in flight. + // Only nonces up to the tip observed at the start of the pass are ever requested. + this.shardBlockPrefetcher.prune(shardId, nonce); + this.shardBlockPrefetcher.prime(shardId, nonce, nonce + this.getPrefetchSize(currentNonce - lastProcessedNonce) - 1); + + const transactionsResult = await this.shardBlockPrefetcher.take(shardId, nonce); return { shardId, currentNonce, lastProcessedNonce, nonce, transactionsResult }; } - private async getShardTransactions(shardId: number, nonce: number): Promise<{ blockHash: string, transactions: ShardTransaction[] } | undefined> { + private async getShardTransactions(shardId: number, nonce: number): Promise { const result = await this.gatewayGet(`block/${shardId}/by-nonce/${nonce}?withTxs=true`); if (!result || !result.block) { @@ -401,7 +440,7 @@ export class TransactionProcessor { .map(ShardTransaction.build); } - private async getHyperblockTransactions(nonce: number): Promise<{ blockHash: string, transactions: ShardTransaction[] } | undefined> { + private async getHyperblockTransactions(nonce: number): Promise { const result = await this.gatewayGet(`hyperblock/by-nonce/${nonce}`); if (!result) { return undefined; diff --git a/src/types/options.ts b/src/types/options.ts index 3af15f1..eeb3daa 100644 --- a/src/types/options.ts +++ b/src/types/options.ts @@ -6,6 +6,15 @@ import { TransactionStatistics } from "./transaction-statistics"; export class TransactionProcessorOptions { gatewayUrl?: string; maxLookBehind?: number; + /** + * How many blocks per shard to read ahead of the block currently being processed, so that + * gateway latency is paid concurrently instead of once per nonce. Read-ahead never targets a + * nonce above the tip observed when the pass started, so it cannot request a block the network + * has not produced yet. Delivery stays strictly one nonce at a time, in order. + * + * Defaults to 10. Set to 1 to restore reading exactly one block at a time. + */ + maxPrefetch?: number; waitForFinalizedCrossShardSmartContractResults?: boolean; notifyEmptyBlocks?: boolean; includeCrossShardStartedTransactions?: boolean; diff --git a/src/utils/block-prefetcher.ts b/src/utils/block-prefetcher.ts new file mode 100644 index 0000000..54a4614 --- /dev/null +++ b/src/utils/block-prefetcher.ts @@ -0,0 +1,107 @@ +/** + * Read-ahead cache for block reads. + * + * The processor hands blocks to the consumer strictly one nonce at a time, in order, which means + * the gateway sits idle for the whole duration of the consumer callback and every nonce pays a + * full round-trip. This keeps a bounded number of reads per shard in flight ahead of the nonce + * being processed, so `take` almost always resolves from an already-settled promise. + * + * Delivery order is unchanged: the caller still asks for one explicit nonce at a time. Only the + * reads move off the critical path. + */ +export class BlockPrefetcher { + private readonly inFlight: Map>> = new Map(); + + constructor( + private readonly fetchBlock: (shardId: number, nonce: number) => Promise, + private readonly onFetchError: (shardId: number, nonce: number, error: unknown) => void, + ) { } + + /** + * Starts reads for [fromNonce, toNonce] without waiting for them. Nonces already in flight are + * left alone, so repeated calls across passes top the pipeline up instead of duplicating reads. + */ + prime(shardId: number, fromNonce: number, toNonce: number): void { + const entries = this.getShardEntries(shardId); + + for (let nonce = fromNonce; nonce <= toNonce; nonce++) { + if (entries.has(nonce)) { + continue; + } + + // The catch is attached synchronously, so a read-ahead failure can never surface as an + // unhandled rejection and can never reject the caller's Promise.all. It resolves to + // undefined instead - the value the processor already treats as 'block not available' - + // which leaves the nonce uncommitted to be retried on a later pass. + entries.set(nonce, this.fetchBlock(shardId, nonce).catch(error => { + this.onFetchError(shardId, nonce, error); + return undefined; + })); + } + } + + /** + * Returns the block for a single nonce, waiting for its read only if it has not settled yet. + */ + async take(shardId: number, nonce: number): Promise { + const entries = this.getShardEntries(shardId); + + let pending = entries.get(nonce); + if (!pending) { + this.prime(shardId, nonce, nonce); + pending = entries.get(nonce); + } + + try { + return await pending; + } finally { + // Consumed either way: a block that was missing or failed to read must be re-read on the + // next attempt rather than served again from the cache. + entries.delete(nonce); + } + } + + /** + * Drops reads for nonces below `belowNonce`, i.e. blocks that were read ahead but then skipped + * (a maxLookBehind jump). + */ + prune(shardId: number, belowNonce: number): void { + const entries = this.inFlight.get(shardId); + if (!entries) { + return; + } + + for (const nonce of entries.keys()) { + if (nonce < belowNonce) { + entries.delete(nonce); + } + } + } + + /** + * Drops every read for a shard, for cases where the whole read-ahead window became meaningless + * (a network reset restarting nonces from zero). + */ + clearShard(shardId: number): void { + this.inFlight.delete(shardId); + } + + get pendingCount(): number { + let total = 0; + for (const entries of this.inFlight.values()) { + total += entries.size; + } + + return total; + } + + private getShardEntries(shardId: number): Map> { + let entries = this.inFlight.get(shardId); + if (!entries) { + entries = new Map(); + this.inFlight.set(shardId, entries); + } + + return entries; + } +} diff --git a/src/utils/constants.ts b/src/utils/constants.ts index e1b4b5c..2d45e62 100644 --- a/src/utils/constants.ts +++ b/src/utils/constants.ts @@ -1,2 +1,11 @@ export const NETWORK_RESET_NONCE_THRESHOLD = 10000; export const METACHAIN = 4294967295; +export const DEFAULT_GATEWAY_URL = 'https://gateway.multiversx.com'; +export const DEFAULT_TIMEOUT = 5000; + +// Upper bound on concurrent sockets kept open against the gateway. Sized above the largest burst +// read-ahead can produce (shard count x maxPrefetch) so the pipeline is never socket-starved. +export const MAX_SOCKETS_PER_HOST = 64; + +// Blocks read ahead per shard, on top of the block currently being processed. +export const DEFAULT_MAX_PREFETCH = 10; diff --git a/src/utils/http.service.ts b/src/utils/http.service.ts index c482038..a1f315d 100644 --- a/src/utils/http.service.ts +++ b/src/utils/http.service.ts @@ -1,31 +1,42 @@ -import axios from "axios"; +import axios, { AxiosInstance } from "axios"; +import * as http from "http"; +import * as https from "https"; +import { DEFAULT_GATEWAY_URL, DEFAULT_TIMEOUT, MAX_SOCKETS_PER_HOST } from "./constants"; + +// Deliberately shared by every HttpService instance rather than owned per instance. +// TransactionProcessor.start() is normally driven by a sub-second cron and rebuilds its +// HttpService on each tick; per-instance agents would discard the socket pool every tick and +// every request would pay a fresh TLS handshake again (~165ms vs ~52ms against the public +// gateway). Node 19+ defaults its global agent to keepAlive, this keeps the behaviour on +// older runtimes too. +const keepAliveHttpAgent = new http.Agent({ keepAlive: true, maxSockets: MAX_SOCKETS_PER_HOST }); +const keepAliveHttpsAgent = new https.Agent({ keepAlive: true, maxSockets: MAX_SOCKETS_PER_HOST }); export class HttpService { - private readonly DEFAULT_TIMEOUT = 5000; - private readonly baseUrl: string | undefined; - private readonly timeout: number; + private readonly baseUrl: string; + private readonly client: AxiosInstance; constructor( baseUrl: string | undefined, timeout: number | undefined = undefined, ) { - this.baseUrl = baseUrl; - this.timeout = timeout ?? this.DEFAULT_TIMEOUT; + this.baseUrl = baseUrl ?? DEFAULT_GATEWAY_URL; + this.client = axios.create({ + baseURL: this.baseUrl, + timeout: timeout ?? DEFAULT_TIMEOUT, + httpAgent: keepAliveHttpAgent, + httpsAgent: keepAliveHttpsAgent, + }); } async get( path: string, ): Promise { - const gatewayUrl = this.baseUrl ?? 'https://gateway.multiversx.com'; - const fullUrl = `${gatewayUrl}/${path}`; - try { - const result = await axios.get(fullUrl, { - timeout: this.timeout, - }); + const result = await this.client.get(path); return result.data.data; } catch (error) { - throw new Error(`Error when getting from url ${fullUrl}: ${error}`); + throw new Error(`Error when getting from url ${this.baseUrl}/${path}: ${error}`); } } }