fix: Simplify Contract Subscriptions block range (#744)

* fix: Simplify Contract Subscriptions block range

* Fix build

* logs

* undo try/catch

---------

Co-authored-by: Prithvish Baidya <[email protected]>

Author: arcoraven

src/db/chainIndexers/getChainIndexer.ts

@@ -1,4 +1,4 @@
-import { PrismaTransaction } from "../../schema/prisma";
+import type { PrismaTransaction } from "../../schema/prisma";
 import { getPrismaWithPostgresTx } from "../client";
 
 interface GetLastIndexedBlockParams {
@@ -45,5 +45,5 @@ export const getBlockForIndexing = async ({
       "chainId"=${chainId}
     FOR UPDATE NOWAIT
   `;
-  return lastIndexedBlock[0]["lastIndexedBlock"];
+  return lastIndexedBlock[0].lastIndexedBlock;
 };

src/utils/env.ts

@@ -69,10 +69,6 @@ export const env = createEnv({
     SDK_BATCH_TIME_LIMIT: z.coerce.number().default(0),
     SDK_BATCH_SIZE_LIMIT: z.coerce.number().default(100),
     ENABLE_KEYPAIR_AUTH: boolEnvSchema(false),
-    CONTRACT_SUBSCRIPTIONS_DELAY_SECONDS: z.coerce
-      .number()
-      .nonnegative()
-      .default(0),
     REDIS_URL: z.string(),
     SEND_TRANSACTION_QUEUE_CONCURRENCY: z.coerce.number().default(200),
     CONFIRM_TRANSACTION_QUEUE_CONCURRENCY: z.coerce.number().default(200),
@@ -125,8 +121,6 @@ export const env = createEnv({
     SDK_BATCH_TIME_LIMIT: process.env.SDK_BATCH_TIME_LIMIT,
     SDK_BATCH_SIZE_LIMIT: process.env.SDK_BATCH_SIZE_LIMIT,
     ENABLE_KEYPAIR_AUTH: process.env.ENABLE_KEYPAIR_AUTH,
-    CONTRACT_SUBSCRIPTIONS_DELAY_SECONDS:
-      process.env.CONTRACT_SUBSCRIPTIONS_DELAY_SECONDS,
     REDIS_URL: process.env.REDIS_URL,
     SEND_TRANSACTION_QUEUE_CONCURRENCY:
       process.env.SEND_TRANSACTION_QUEUE_CONCURRENCY,

src/utils/indexer/getBlockTime.ts

@@ -1,49 +1,26 @@
-import { getSdk } from "../cache/getSdk";
-import { logger } from "../logger";
+import { eth_getBlockByNumber, getRpcClient } from "thirdweb";
+import { getChain } from "../chain";
+import { thirdwebClient } from "../sdk";
 
-const KNOWN_BLOCKTIME_SECONDS = {
-  1: 12,
-  137: 2,
-} as Record<number, number>;
+export const getBlockTimeSeconds = async (
+  chainId: number,
+  blocksToEstimate: number,
+) => {
+  const chain = await getChain(chainId);
+  const rpcRequest = getRpcClient({
+    client: thirdwebClient,
+    chain,
+  });
 
-const DEFAULT_BLOCKTIME_SECONDS = 10;
-const BLOCKS_TO_ESTIMATE = 100;
+  const latestBlock = await eth_getBlockByNumber(rpcRequest, {
+    blockTag: "latest",
+    includeTransactions: false,
+  });
+  const referenceBlock = await eth_getBlockByNumber(rpcRequest, {
+    blockNumber: latestBlock.number - BigInt(blocksToEstimate),
+    includeTransactions: false,
+  });
 
-export const getBlockTimeSeconds = async (chainId: number) => {
-  if (KNOWN_BLOCKTIME_SECONDS[chainId]) {
-    return KNOWN_BLOCKTIME_SECONDS[chainId];
-  }
-
-  const sdk = await getSdk({ chainId });
-  const provider = sdk.getProvider();
-  try {
-    const latestBlockNumber = await provider.getBlockNumber();
-    const blockNumbers = Array.from(
-      { length: BLOCKS_TO_ESTIMATE },
-      (_, i) => latestBlockNumber - i - 1,
-    );
-
-    const blocks = await Promise.all(
-      blockNumbers.map(async (blockNumber) => {
-        const block = await provider.getBlock(blockNumber);
-        return block;
-      }),
-    );
-
-    let totalTimeDiff = 0;
-    for (let i = 0; i < blocks.length - 1; i++) {
-      totalTimeDiff += blocks[i].timestamp - blocks[i + 1].timestamp;
-    }
-
-    const averageBlockTime = totalTimeDiff / (blocks.length - 1);
-    return averageBlockTime;
-  } catch (error) {
-    logger({
-      service: "worker",
-      level: "error",
-      message: `Error estimating block time for chainId ${chainId}:`,
-      error,
-    });
-    return DEFAULT_BLOCKTIME_SECONDS;
-  }
+  const diffSeconds = latestBlock.timestamp - referenceBlock.timestamp;
+  return Number(diffSeconds) / (blocksToEstimate + 1);
 };

src/worker/indexers/chainIndexerRegistry.ts

@@ -1,67 +1,57 @@
 import cron from "node-cron";
-import { getConfig } from "../../utils/cache/getConfig";
-import { env } from "../../utils/env";
 import { getBlockTimeSeconds } from "../../utils/indexer/getBlockTime";
 import { logger } from "../../utils/logger";
-import { createChainIndexerTask } from "../tasks/chainIndexer";
+import { handleContractSubscriptions } from "../tasks/chainIndexer";
 
+// @TODO: Move all worker logic to Bullmq to better handle multiple hosts.
 export const INDEXER_REGISTRY = {} as Record<number, cron.ScheduledTask>;
 
 export const addChainIndexer = async (chainId: number) => {
   if (INDEXER_REGISTRY[chainId]) {
+    return;
+  }
+
+  // Estimate the block time in the last 100 blocks. Default to 2 second block times.
+  let blockTimeSeconds: number;
+  try {
+    blockTimeSeconds = await getBlockTimeSeconds(chainId, 100);
+  } catch (error) {
     logger({
       service: "worker",
-      level: "warn",
-      message: `Chain Indexer already exists: ${chainId}`,
+      level: "error",
+      message: `Could not estimate block time for chain ${chainId}`,
+      error,
     });
-    return;
+    blockTimeSeconds = 2;
   }
-
-  let processStarted = false;
-  const config = await getConfig();
-
-  // Estimate block time.
-  const blockTimeSeconds = await getBlockTimeSeconds(chainId);
-
-  const blocksIn5Seconds = Math.round((1 / blockTimeSeconds) * 5);
-  const maxBlocksToIndex = Math.max(
-    config.maxBlocksToIndex,
-    blocksIn5Seconds * 4,
-  );
-
-  // Compute block offset based on delay.
-  // Example: 10s delay with a 3s block time = 4 blocks offset
-  const toBlockOffset = env.CONTRACT_SUBSCRIPTIONS_DELAY_SECONDS
-    ? Math.ceil(env.CONTRACT_SUBSCRIPTIONS_DELAY_SECONDS / blockTimeSeconds)
-    : 0;
-
-  const handler = await createChainIndexerTask({
-    chainId,
-    maxBlocksToIndex,
-    toBlockOffset,
-  });
-
   const cronSchedule = createScheduleSeconds(
     Math.max(Math.round(blockTimeSeconds), 1),
   );
-
   logger({
     service: "worker",
     level: "info",
-    message: `Indexing contracts on chainId: ${chainId} with schedule: ${cronSchedule}, max blocks to index: ${maxBlocksToIndex}`,
+    message: `Indexing contracts on chain ${chainId} with schedule: ${cronSchedule}`,
   });
 
+  let inProgress = false;
+
   const task = cron.schedule(cronSchedule, async () => {
-    if (!processStarted) {
-      processStarted = true;
+    if (inProgress) {
+      return;
+    }
 
-      try {
-        await handler();
-      } catch (error) {
-        // do nothing
-      } finally {
-        processStarted = false;
-      }
+    inProgress = true;
+    try {
+      await handleContractSubscriptions(chainId);
+    } catch (error) {
+      logger({
+        service: "worker",
+        level: "error",
+        message: `Failed to index on chain ${chainId}`,
+        error,
+      });
+    } finally {
+      inProgress = false;
     }
   });
 
@@ -70,13 +60,7 @@ export const addChainIndexer = async (chainId: number) => {
 
 export const removeChainIndexer = async (chainId: number) => {
   const task = INDEXER_REGISTRY[chainId];
-
   if (!task) {
-    logger({
-      service: "worker",
-      level: "warn",
-      message: `Chain Indexer doesn't exist: ${chainId}`,
-    });
     return;
   }
 

src/worker/listeners/chainIndexerListener.ts

@@ -1,30 +1,13 @@
 import cron from "node-cron";
 import { getConfig } from "../../utils/cache/getConfig";
 import { logger } from "../../utils/logger";
-import { redis } from "../../utils/redis/redis";
 import { manageChainIndexers } from "../tasks/manageChainIndexers";
 
 let processChainIndexerStarted = false;
 let task: cron.ScheduledTask;
 
 export const chainIndexerListener = async (): Promise<void> => {
-  if (!redis) {
-    logger({
-      service: "worker",
-      level: "warn",
-      message: `Chain Indexer Listener not started, Redis not available`,
-    });
-    return;
-  }
-
-  logger({
-    service: "worker",
-    level: "info",
-    message: `Listening for indexed contracts`,
-  });
-
   const config = await getConfig();
-
   if (!config.indexerListenerCronSchedule) {
     return;
   }
@@ -41,7 +24,7 @@ export const chainIndexerListener = async (): Promise<void> => {
       logger({
         service: "worker",
         level: "warn",
-        message: `manageChainIndexers already running, skipping`,
+        message: "manageChainIndexers already running, skipping",
       });
     }
   });