Handling Edge Cases for RPC inconsistencies (#528)

* Added checks to make sure RPC is good

* updated throw to log & return

* added comments delayed requerying

* udptes

* updated user facing intefaces for contractSubscriptionsRetryDelaySeconds

* updates

* updates to solve build issue

Author: farhanW3

src/db/configuration/getConfiguration.ts

@@ -1,6 +1,7 @@
 import { Configuration } from "@prisma/client";
 import { LocalWallet } from "@thirdweb-dev/wallets";
 import { ethers } from "ethers";
+import { Config } from "../../schema/config";
 import { WalletType } from "../../schema/wallet";
 import { mandatoryAllowedCorsUrls } from "../../server/utils/cors-urls";
 import { decrypt } from "../../utils/crypto";
@@ -9,38 +10,6 @@ import { logger } from "../../utils/logger";
 import { prisma } from "../client";
 import { updateConfiguration } from "./updateConfiguration";
 
-interface Config
-  extends Omit<
-    Configuration,
-    | "awsAccessKeyId"
-    | "awsSecretAccessKey"
-    | "awsRegion"
-    | "gcpApplicationProjectId"
-    | "gcpKmsLocationId"
-    | "gcpKmsKeyRingId"
-    | "gcpApplicationCredentialEmail"
-    | "gcpApplicationCredentialPrivateKey"
-  > {
-  walletConfiguration:
-    | {
-        type: WalletType.local;
-      }
-    | {
-        type: WalletType.awsKms;
-        awsAccessKeyId: string;
-        awsSecretAccessKey: string;
-        awsRegion: string;
-      }
-    | {
-        type: WalletType.gcpKms;
-        gcpApplicationProjectId: string;
-        gcpKmsLocationId: string;
-        gcpKmsKeyRingId: string;
-        gcpApplicationCredentialEmail: string;
-        gcpApplicationCredentialPrivateKey: string;
-      };
-}
-
 const withWalletConfig = async (config: Configuration): Promise<Config> => {
   // We destructure the config to omit wallet related fields to prevent direct access
   const {
@@ -52,6 +21,7 @@ const withWalletConfig = async (config: Configuration): Promise<Config> => {
     gcpKmsKeyRingId,
     gcpApplicationCredentialEmail,
     gcpApplicationCredentialPrivateKey,
+    contractSubscriptionsRetryDelaySeconds,
     ...restConfig
   } = config;
 
@@ -84,8 +54,13 @@ const withWalletConfig = async (config: Configuration): Promise<Config> => {
       }
     }
 
+    // Renaming contractSubscriptionsRetryDelaySeconds
+    // to contractSubscriptionsRequeryDelaySeconds to reflect its purpose
+    // as we are requerying (& not retrying) with different delays
     return {
       ...restConfig,
+      contractSubscriptionsRequeryDelaySeconds:
+        contractSubscriptionsRetryDelaySeconds,
       walletConfiguration: {
         type: WalletType.awsKms,
         awsRegion,
@@ -132,6 +107,8 @@ const withWalletConfig = async (config: Configuration): Promise<Config> => {
 
     return {
       ...restConfig,
+      contractSubscriptionsRequeryDelaySeconds:
+        contractSubscriptionsRetryDelaySeconds,
       walletConfiguration: {
         type: WalletType.gcpKms,
         gcpApplicationProjectId,
@@ -145,6 +122,8 @@ const withWalletConfig = async (config: Configuration): Promise<Config> => {
 
   return {
     ...restConfig,
+    contractSubscriptionsRequeryDelaySeconds:
+      contractSubscriptionsRetryDelaySeconds,
     walletConfiguration: {
       type: WalletType.local,
     },

src/schema/config.ts

@@ -0,0 +1,36 @@
+import { Configuration } from "@prisma/client";
+import { WalletType } from "./wallet";
+
+export interface Config
+  extends Omit<
+    Configuration,
+    | "awsAccessKeyId"
+    | "awsSecretAccessKey"
+    | "awsRegion"
+    | "gcpApplicationProjectId"
+    | "gcpKmsLocationId"
+    | "gcpKmsKeyRingId"
+    | "gcpApplicationCredentialEmail"
+    | "gcpApplicationCredentialPrivateKey"
+    | "contractSubscriptionsRetryDelaySeconds"
+  > {
+  walletConfiguration:
+    | {
+        type: WalletType.local;
+      }
+    | {
+        type: WalletType.awsKms;
+        awsAccessKeyId: string;
+        awsSecretAccessKey: string;
+        awsRegion: string;
+      }
+    | {
+        type: WalletType.gcpKms;
+        gcpApplicationProjectId: string;
+        gcpKmsLocationId: string;
+        gcpKmsKeyRingId: string;
+        gcpApplicationCredentialEmail: string;
+        gcpApplicationCredentialPrivateKey: string;
+      };
+  contractSubscriptionsRequeryDelaySeconds: string;
+}

src/server/routes/configuration/contract-subscriptions/get.ts

@@ -2,20 +2,20 @@ import { Static, Type } from "@sinclair/typebox";
 import { FastifyInstance } from "fastify";
 import { StatusCodes } from "http-status-codes";
 import { getConfig } from "../../../../utils/cache/getConfig";
-import { standardResponseSchema } from "../../../schemas/sharedApiSchemas";
+import {
+  contractSubscriptionConfigurationSchema,
+  standardResponseSchema,
+} from "../../../schemas/sharedApiSchemas";
 
-export const ReplySchema = Type.Object({
-  result: Type.Object({
-    maxBlocksToIndex: Type.Number(),
-    contractSubscriptionsRetryDelaySeconds: Type.String(),
-  }),
+const responseSchema = Type.Object({
+  result: contractSubscriptionConfigurationSchema,
 });
 
 export async function getContractSubscriptionsConfiguration(
   fastify: FastifyInstance,
 ) {
   fastify.route<{
-    Reply: Static<typeof ReplySchema>;
+    Reply: Static<typeof responseSchema>;
   }>({
     method: "GET",
     url: "/configuration/contract-subscriptions",
@@ -26,16 +26,16 @@ export async function getContractSubscriptionsConfiguration(
       operationId: "getContractSubscriptionsConfiguration",
       response: {
         ...standardResponseSchema,
-        [StatusCodes.OK]: ReplySchema,
+        [StatusCodes.OK]: responseSchema,
       },
     },
     handler: async (req, res) => {
       const config = await getConfig();
       res.status(200).send({
         result: {
           maxBlocksToIndex: config.maxBlocksToIndex,
-          contractSubscriptionsRetryDelaySeconds:
-            config.contractSubscriptionsRetryDelaySeconds,
+          contractSubscriptionsRequeryDelaySeconds:
+            config.contractSubscriptionsRequeryDelaySeconds,
         },
       });
     },

src/server/routes/configuration/contract-subscriptions/update.ts

@@ -4,19 +4,25 @@ import { StatusCodes } from "http-status-codes";
 import { updateConfiguration } from "../../../../db/configuration/updateConfiguration";
 import { getConfig } from "../../../../utils/cache/getConfig";
 import { createCustomError } from "../../../middleware/error";
-import { standardResponseSchema } from "../../../schemas/sharedApiSchemas";
-import { ReplySchema } from "./get";
+import {
+  contractSubscriptionConfigurationSchema,
+  standardResponseSchema,
+} from "../../../schemas/sharedApiSchemas";
 
-const BodySchema = Type.Object({
+const requestBodySchema = Type.Object({
   maxBlocksToIndex: Type.Optional(Type.Number({ minimum: 1, maximum: 25 })),
-  contractSubscriptionsRetryDelaySeconds: Type.Optional(Type.String()),
+  contractSubscriptionsRequeryDelaySeconds: Type.Optional(Type.String()),
+});
+
+const responseSchema = Type.Object({
+  result: contractSubscriptionConfigurationSchema,
 });
 
 export async function updateContractSubscriptionsConfiguration(
   fastify: FastifyInstance,
 ) {
   fastify.route<{
-    Body: Static<typeof BodySchema>;
+    Body: Static<typeof requestBodySchema>;
   }>({
     method: "POST",
     url: "/configuration/contract-subscriptions",
@@ -25,34 +31,34 @@ export async function updateContractSubscriptionsConfiguration(
       description: "Update the configuration for Contract Subscriptions",
       tags: ["Configuration"],
       operationId: "updateContractSubscriptionsConfiguration",
-      body: BodySchema,
+      body: requestBodySchema,
       response: {
         ...standardResponseSchema,
-        [StatusCodes.OK]: ReplySchema,
+        [StatusCodes.OK]: responseSchema,
       },
     },
     handler: async (req, res) => {
-      const { maxBlocksToIndex, contractSubscriptionsRetryDelaySeconds } =
+      const { maxBlocksToIndex, contractSubscriptionsRequeryDelaySeconds } =
         req.body;
 
-      if (!maxBlocksToIndex && !contractSubscriptionsRetryDelaySeconds) {
+      if (!maxBlocksToIndex && !contractSubscriptionsRequeryDelaySeconds) {
         throw createCustomError(
           "At least one parameter is required",
           StatusCodes.BAD_REQUEST,
           "BAD_REQUEST",
         );
       }
 
-      if (contractSubscriptionsRetryDelaySeconds) {
+      if (contractSubscriptionsRequeryDelaySeconds) {
         try {
-          contractSubscriptionsRetryDelaySeconds.split(",").forEach((d) => {
+          contractSubscriptionsRequeryDelaySeconds.split(",").forEach((d) => {
             if (Number.isNaN(parseInt(d))) {
               throw "Invalid number";
             }
           });
         } catch {
           throw createCustomError(
-            'At least one integer "contractSubscriptionsRetryDelaySeconds" is required',
+            'At least one integer "contractSubscriptionsRequeryDelaySeconds" is required',
             StatusCodes.BAD_REQUEST,
             "BAD_REQUEST",
           );
@@ -61,15 +67,16 @@ export async function updateContractSubscriptionsConfiguration(
 
       await updateConfiguration({
         maxBlocksToIndex,
-        contractSubscriptionsRetryDelaySeconds,
+        contractSubscriptionsRetryDelaySeconds:
+          contractSubscriptionsRequeryDelaySeconds,
       });
       const config = await getConfig(false);
 
       res.status(200).send({
         result: {
           maxBlocksToIndex: config.maxBlocksToIndex,
-          contractSubscriptionsRetryDelaySeconds:
-            config.contractSubscriptionsRetryDelaySeconds,
+          contractSubscriptionsRequeryDelaySeconds:
+            config.contractSubscriptionsRequeryDelaySeconds,
         },
       });
     },

src/server/schemas/sharedApiSchemas.ts

@@ -332,3 +332,8 @@ export const walletDetailsSchema = Type.Object({
     Type.Null(),
   ]),
 });
+
+export const contractSubscriptionConfigurationSchema = Type.Object({
+  maxBlocksToIndex: Type.Number(),
+  contractSubscriptionsRequeryDelaySeconds: Type.String(),
+});

src/utils/cache/getConfig.ts

@@ -1,40 +1,7 @@
-import { Configuration } from "@prisma/client";
 import { getConfiguration } from "../../db/configuration/getConfiguration";
-import { WalletType } from "../../schema/wallet";
+import { Config } from "../../schema/config";
 
 const cacheKey = "config";
-interface Config
-  extends Omit<
-    Configuration,
-    | "awsAccessKeyId"
-    | "awsSecretAccessKey"
-    | "awsRegion"
-    | "gcpApplicationProjectId"
-    | "gcpKmsLocationId"
-    | "gcpKmsKeyRingId"
-    | "gcpApplicationCredentialEmail"
-    | "gcpApplicationCredentialPrivateKey"
-  > {
-  walletConfiguration:
-    | {
-        type: WalletType.local;
-      }
-    | {
-        type: WalletType.awsKms;
-        awsAccessKeyId: string;
-        awsSecretAccessKey: string;
-        awsRegion: string;
-      }
-    | {
-        type: WalletType.gcpKms;
-        gcpApplicationProjectId: string;
-        gcpKmsLocationId: string;
-        gcpKmsKeyRingId: string;
-        gcpApplicationCredentialEmail: string;
-        gcpApplicationCredentialPrivateKey: string;
-      };
-}
-
 export const configCache = new Map<string, Config>();
 
 export const getConfig = async (retrieveFromCache = true): Promise<Config> => {
@@ -51,7 +18,6 @@ export const getConfig = async (retrieveFromCache = true): Promise<Config> => {
   }
 
   const configData = await getConfiguration();
-
   configCache.set(cacheKey, configData);
   return configData;
 };

src/worker/index.ts

@@ -1,3 +1,5 @@
+import { logger } from "../utils/logger";
+import { redis } from "../utils/redis/redis";
 import { chainIndexerListener } from "./listeners/chainIndexerListener";
 import {
   newConfigurationListener,
@@ -36,5 +38,13 @@ export const initWorker = async () => {
   await newWebhooksListener();
   await updatedWebhooksListener();
 
-  await chainIndexerListener();
+  if (redis) {
+    await chainIndexerListener();
+  } else {
+    logger({
+      service: "worker",
+      level: "warn",
+      message: `Chain Indexer Listener not started, Redis not available`,
+    });
+  }
 };

src/worker/queues/processEventLogsQueue.ts

@@ -32,14 +32,22 @@ export const enqueueProcessEventLogs = async (
   const serialized = SuperJSON.stringify(data);
   // e.g. 8453:14423125-14423685
   const jobName = `${data.chainId}:${data.fromBlock}-${data.toBlock}`;
-  const { contractSubscriptionsRetryDelaySeconds } = await getConfig();
-  const retryDelays = contractSubscriptionsRetryDelaySeconds.split(",");
+  const { contractSubscriptionsRequeryDelaySeconds } = await getConfig();
+  const requeryDelays = contractSubscriptionsRequeryDelaySeconds.split(",");
 
   // Enqueue one job immediately and any delayed jobs.
   await _queue.add(jobName, serialized);
-  for (const retryDelay of retryDelays) {
+
+  // The last attempt should attempt repeatedly to handle extended RPC issues.
+  // This backoff attempts at intervals:
+  // 30s, 1m, 2m, 4m, 8m, 16m, 32m, ~1h, ~2h, ~4h
+  for (let i = 0; i < requeryDelays.length; i++) {
+    const delay = parseInt(requeryDelays[i]) * 1000;
+    const attempts = i === requeryDelays.length - 1 ? 10 : 0;
     await _queue.add(jobName, serialized, {
-      delay: parseInt(retryDelay) * 1000,
+      delay,
+      attempts,
+      backoff: { type: "exponential", delay: 30_000 },
     });
   }
 };

src/worker/queues/processTransactionReceiptsQueue.ts

@@ -33,14 +33,22 @@ export const enqueueProcessTransactionReceipts = async (
   const serialized = SuperJSON.stringify(data);
   // e.g. 8453:14423125-14423685
   const jobName = `${data.chainId}:${data.fromBlock}-${data.toBlock}`;
-  const { contractSubscriptionsRetryDelaySeconds } = await getConfig();
-  const retryDelays = contractSubscriptionsRetryDelaySeconds.split(",");
+  const { contractSubscriptionsRequeryDelaySeconds } = await getConfig();
+  const requeryDelays = contractSubscriptionsRequeryDelaySeconds.split(",");
 
   // Enqueue one job immediately and any delayed jobs.
   await _queue.add(jobName, serialized);
-  for (const retryDelay of retryDelays) {
+
+  // The last attempt should attempt repeatedly to handle extended RPC issues.
+  // This backoff attempts at intervals:
+  // 30s, 1m, 2m, 4m, 8m, 16m, 32m, ~1h, ~2h, ~4h
+  for (let i = 0; i < requeryDelays.length; i++) {
+    const delay = parseInt(requeryDelays[i]) * 1000;
+    const attempts = i === requeryDelays.length - 1 ? 10 : 0;
     await _queue.add(jobName, serialized, {
-      delay: parseInt(retryDelay) * 1000,
+      delay,
+      attempts,
+      backoff: { type: "exponential", delay: 30_000 },
     });
   }
 };