feat: ws specific rpc urls (#574)

Author: NikolasHaimerl

packages/indexer/src/data-indexing/service/filtering.ts

@@ -75,7 +75,7 @@ export const createCctpBurnFilter = async (
   logger: Logger,
 ): Promise<boolean> => {
   // Check if receipt is present in payload
-  const receipt = payload.transactionReceipt;
+  const receipt = await payload.transactionReceipt;
   if (!receipt) {
     logger.debug({
       at: "createSwapApiFilter",
@@ -145,7 +145,7 @@ export const createCctpMintFilter = async (
   logger: Logger,
 ): Promise<boolean> => {
   // Check if receipt is present in payload
-  const receipt = payload.transactionReceipt;
+  const receipt = await payload.transactionReceipt;
   if (!receipt) {
     logger.debug({
       at: "createCctpMintFilter",

packages/indexer/src/data-indexing/service/genericEventListening.ts

@@ -1,3 +1,4 @@
+import pRetry from "p-retry";
 import Bottleneck from "bottleneck";
 import {
   parseAbi,
@@ -7,8 +8,10 @@ import {
   type Chain,
   type Transaction,
   TransactionReceipt,
+  GetBlockReturnType,
 } from "viem";
 import { Logger } from "winston";
+import PLazy from "p-lazy";
 import { DataDogMetricsService } from "../../services/MetricsService";
 
 /**
@@ -22,6 +25,10 @@ import { DataDogMetricsService } from "../../services/MetricsService";
  * - Pushing the clean payload into the next stage of the pipeline via a callback.
  */
 
+const MAX_RETRY_TIMEOUT = 60000;
+const RETRY_ATTEMPTS = 10;
+const RETRY_BACKOFF_EXPONENT = 2;
+
 /**
  * Defines the configuration for a single event subscription.
  * This allows the listener service to be generic and data-driven.
@@ -56,7 +63,7 @@ export interface IndexerEventPayload {
   /** The transaction that generated the event. */
   transaction?: Transaction;
   /** The receipt of the transaction that generated the event. */
-  transactionReceipt?: TransactionReceipt;
+  transactionReceipt?: Promise<TransactionReceipt>;
 }
 
 /**
@@ -93,7 +100,6 @@ export interface ProcessLogBatchArgs<TPayload> {
   config: EventConfig;
   chainId: number;
   client: PublicClient<Transport, Chain>;
-
   onEvent: (payload: TPayload) => void;
   logger: Logger;
   metrics?: DataDogMetricsService;
@@ -183,11 +189,8 @@ async function processLogBatch<TPayload>(
   ];
 
   // Create caches for this batch to avoid duplicate requests/parallel fetching
-  const blockCache = new Map<
-    bigint,
-    { timestamp: bigint; transactions: Transaction[] }
-  >();
-  const receiptCache = new Map<string, TransactionReceipt>();
+  const blockCache = new Map<bigint, GetBlockReturnType<Chain, true>>();
+  const receiptCache = new Map<string, Promise<TransactionReceipt>>();
 
   // Process all logs in parallel
   await Promise.all(
@@ -200,17 +203,34 @@ async function processLogBatch<TPayload>(
         }
 
         // --- Fetch Block & Transactions (Deduplicated) ---
-        let blockInformation = blockCache.get(logItem.blockNumber);
-        if (!blockInformation) {
-          blockInformation = await client.getBlock({
-            blockNumber: logItem.blockNumber,
-            includeTransactions: true,
-          });
-          metrics?.addCountMetric("rpcCallGetBlock", tags);
-
-          blockCache.set(logItem.blockNumber, blockInformation);
+        let block = blockCache.get(logItem.blockNumber);
+        if (!block) {
+          block = await pRetry(
+            () => {
+              metrics?.addCountMetric("rpcCallGetBlock", tags);
+              return client.getBlock({
+                blockNumber: logItem.blockNumber!,
+                includeTransactions: true,
+              });
+            },
+            {
+              retries: RETRY_ATTEMPTS,
+              factor: RETRY_BACKOFF_EXPONENT,
+              maxTimeout: MAX_RETRY_TIMEOUT,
+              onFailedAttempt: (error) => {
+                logger.warn({
+                  at: "genericEventListener#processLogBatch",
+                  message: `Failed to fetch block ${logItem.blockNumber}, retrying...`,
+                  attempt: error.attemptNumber,
+                  retriesLeft: error.retriesLeft,
+                  error,
+                });
+              },
+            },
+          );
+          blockCache.set(logItem.blockNumber, block);
         }
-        const { timestamp: blockTimestamp, transactions } = blockInformation;
+        const { timestamp: blockTimestamp, transactions } = await block;
 
         // --- Find Transaction ---
         let transaction: Transaction | undefined;
@@ -221,15 +241,40 @@ async function processLogBatch<TPayload>(
         }
 
         // --- Fetch Transaction Receipt (Deduplicated) ---
-        let transactionReceipt: TransactionReceipt | undefined;
+        let transactionReceiptPromise: Promise<TransactionReceipt> | undefined;
         if (logItem.transactionHash) {
-          transactionReceipt = receiptCache.get(logItem.transactionHash);
-          if (!transactionReceipt) {
-            transactionReceipt = await client.getTransactionReceipt({
-              hash: logItem.transactionHash,
+          transactionReceiptPromise = receiptCache.get(logItem.transactionHash);
+          if (!transactionReceiptPromise) {
+            transactionReceiptPromise = new PLazy((resolve, reject) => {
+              pRetry(
+                () => {
+                  metrics?.addCountMetric("rpcCallGetTransactionReceipt", tags);
+                  return client.getTransactionReceipt({
+                    hash: logItem.transactionHash!,
+                  });
+                },
+                {
+                  retries: RETRY_ATTEMPTS,
+                  factor: RETRY_BACKOFF_EXPONENT,
+                  maxTimeout: MAX_RETRY_TIMEOUT,
+                  onFailedAttempt: (error) => {
+                    logger.warn({
+                      at: "genericEventListener#processLogBatch",
+                      message: `Failed to fetch receipt for ${logItem.transactionHash}, retrying...`,
+                      attempt: error.attemptNumber,
+                      retriesLeft: error.retriesLeft,
+                      error,
+                    });
+                  },
+                },
+              )
+                .then(resolve)
+                .catch(reject);
             });
-            metrics?.addCountMetric("rpcCallGetTransactionReceipt", tags);
-            receiptCache.set(logItem.transactionHash, transactionReceipt);
+            receiptCache.set(
+              logItem.transactionHash,
+              transactionReceiptPromise,
+            );
           }
         }
 
@@ -241,7 +286,7 @@ async function processLogBatch<TPayload>(
           // The events are emitted at head so we can simply take the block number of the log as the currentBlockHeight
           currentBlockHeight: logItem.blockNumber,
           transaction,
-          transactionReceipt,
+          transactionReceipt: transactionReceiptPromise,
         } as TPayload;
 
         // Trigger the side effect (Forward it to an event processor or message queue)
@@ -274,6 +319,7 @@ async function processLogBatch<TPayload>(
           error,
           logIndex: logItem.logIndex,
           txHash: logItem.transactionHash,
+          chainId,
         });
         metrics?.addCountMetric("processLogError", tags);
       }