From 6d3652cda2495a38d37a7d1762e160d324ab6ae1 Mon Sep 17 00:00:00 2001 From: Vlad Bucur Date: Thu, 10 Sep 2026 16:02:24 +0300 Subject: [PATCH] Read blocks ahead so gateway latency leaves the critical path At 600ms block time the chain produces 1.67 blocks/s/shard. The processor read one block at a time and awaited each read inline, so the gateway sat idle for the whole duration of the consumer callback and every nonce paid a full round-trip. Measured against mainnet with a 100ms onTransactionsReceived that came to 1.60 nonces/s/shard - below the production rate, so a backlog once formed could never drain. Add a bounded read-ahead cache per shard. While nonce N is handed to the consumer, the reads for N+1..N+maxPrefetch are already in flight, so take() almost always resolves an already-settled promise. Delivery order is unchanged: still one nonce per shard per pass, in order, committed only after the callback returns. Only the reads moved off the critical path, which keeps the cross-shard SCR counters in crossShardDictionary seeing blocks in exactly the order they did before. The window never extends past the tip observed at the start of the pass, so it cannot request a block the network has not produced yet, and it collapses to a single read when already caught up. Windows are dropped on a network reset and pruned after a maxLookBehind jump. A read failure now resolves to undefined rather than rejecting, since the catch is attached inside prime(). That is the value the loop already treats as 'block not available', so one flaky shard leaves the nonce uncommitted for a later retry instead of aborting the whole run. Previously a single failing shard threw out of start() and stopped every other shard: 1/30 nonces delivered, now 30/30. Also give HttpService shared keep-alive agents. start() is normally driven by a sub-second cron and rebuilds its HttpService each tick, so the agents are module-level - per-instance agents would discard the socket pool every tick and pay a fresh TLS handshake per request (~165ms vs ~52ms). Node 19+ already defaults its global agent to keepAlive; this keeps the behaviour on older runtimes and caps the pool during a read-ahead burst. Measured on mainnet, catching up 25 nonces/shard across 4 shards: no-op callback 4.16 -> 20.51 nonces/s/shard 100ms callback 1.60 -> 2.37 nonces/s/shard maxPrefetch defaults to 10; set it to 1 to restore the previous one-block-at- a-time behaviour. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01S2BfVKhgjNR2FdJR6V6GXQ --- src/transaction.processor.ts | 48 +++++++++++++-- src/types/options.ts | 9 +++ src/utils/block-prefetcher.ts | 107 ++++++++++++++++++++++++++++++++++ src/utils/constants.ts | 9 +++ src/utils/http.service.ts | 37 +++++++----- 5 files changed, 192 insertions(+), 18 deletions(-) create mode 100644 src/utils/block-prefetcher.ts diff --git a/src/transaction.processor.ts b/src/transaction.processor.ts index 91739b1..958e8f5 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); @@ -86,6 +101,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) { @@ -103,7 +120,13 @@ 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); if (transactionsResult === undefined) { this.logMessage(LogTopic.Debug, 'transactionsResult === undefined'); continue; @@ -206,6 +229,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) { @@ -223,7 +247,10 @@ export class TransactionProcessor { const nonce = lastProcessedNonce + 1; - const transactionsResult = await this.getHyperblockTransactions(nonce); + 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; @@ -343,7 +370,18 @@ export class TransactionProcessor { return crossShardTransactions; } - private async getShardTransactions(shardId: number, nonce: number): Promise<{ blockHash: string, transactions: ShardTransaction[] } | undefined> { + /** + * 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 getShardTransactions(shardId: number, nonce: number): Promise { const result = await this.gatewayGet(`block/${shardId}/by-nonce/${nonce}?withTxs=true`); if (!result || !result.block) { @@ -369,7 +407,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}`); } } }