Read blocks ahead so gateway latency leaves the critical path

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<BlockTransactions>(
+    (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<BlockTransactions>(
+    (_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<void> {
     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<BlockTransactions | undefined> {
     const result = await this.gatewayGet<GatewayBlockResponse>(`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<BlockTransactions | undefined> {
     const result = await this.gatewayGet(`hyperblock/by-nonce/${nonce}`);
     if (!result) {
       return undefined;

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;

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<T> {
+  private readonly inFlight: Map<number, Map<number, Promise<T | undefined>>> = new Map();
+
+  constructor(
+    private readonly fetchBlock: (shardId: number, nonce: number) => Promise<T | undefined>,
+    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<T | undefined> {
+    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<number, Promise<T | undefined>> {
+    let entries = this.inFlight.get(shardId);
+    if (!entries) {
+      entries = new Map();
+      this.inFlight.set(shardId, entries);
+    }
+
+    return entries;
+  }
+}

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;

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<T = any>(
     path: string,
   ): Promise<T> {
-    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}`);
     }
   }
 }