Handling Edge Cases for RPC inconsistencies

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

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

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

@@ -4,12 +4,14 @@ 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 {
+  contractSubscriptionResponseSchema,
+  standardResponseSchema,
+} from "../../../schemas/sharedApiSchemas";
 
 const BodySchema = Type.Object({
   maxBlocksToIndex: Type.Optional(Type.Number({ minimum: 1, maximum: 25 })),
-  contractSubscriptionsRetryDelaySeconds: Type.Optional(Type.String()),
+  contractSubscriptionsRequeryDelaySeconds: Type.Optional(Type.String()),
 });
 
 export async function updateContractSubscriptionsConfiguration(
@@ -28,31 +30,31 @@ export async function updateContractSubscriptionsConfiguration(
       body: BodySchema,
       response: {
         ...standardResponseSchema,
-        [StatusCodes.OK]: ReplySchema,
+        [StatusCodes.OK]: contractSubscriptionResponseSchema,
       },
     },
     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,14 +63,15 @@ export async function updateContractSubscriptionsConfiguration(
 
       await updateConfiguration({
         maxBlocksToIndex,
-        contractSubscriptionsRetryDelaySeconds,
+        contractSubscriptionsRetryDelaySeconds:
+          contractSubscriptionsRequeryDelaySeconds,
       });
       const config = await getConfig(false);
 
       res.status(200).send({
         result: {
           maxBlocksToIndex: config.maxBlocksToIndex,
-          contractSubscriptionsRetryDelaySeconds:
+          contractSubscriptionsRequeryDelaySeconds:
             config.contractSubscriptionsRetryDelaySeconds,
         },
       });

src/server/schemas/sharedApiSchemas.ts

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

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

@@ -33,17 +33,17 @@ export const enqueueProcessEventLogs = async (
   // e.g. 8453:14423125-14423685
   const jobName = `${data.chainId}:${data.fromBlock}-${data.toBlock}`;
   const { contractSubscriptionsRetryDelaySeconds } = await getConfig();
-  const retryDelays = contractSubscriptionsRetryDelaySeconds.split(",");
+  const requeryDelays = contractSubscriptionsRetryDelaySeconds.split(",");
 
   // Enqueue one job immediately and any delayed jobs.
   await _queue.add(jobName, serialized);
 
   // 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 < retryDelays.length; i++) {
-    const delay = parseInt(retryDelays[i]) * 1000;
-    const attempts = i === retryDelays.length - 1 ? 10 : 0;
+  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,
       attempts,

src/worker/queues/processTransactionReceiptsQueue.ts

@@ -34,17 +34,17 @@ export const enqueueProcessTransactionReceipts = async (
   // e.g. 8453:14423125-14423685
   const jobName = `${data.chainId}:${data.fromBlock}-${data.toBlock}`;
   const { contractSubscriptionsRetryDelaySeconds } = await getConfig();
-  const retryDelays = contractSubscriptionsRetryDelaySeconds.split(",");
+  const requeryDelays = contractSubscriptionsRetryDelaySeconds.split(",");
 
   // Enqueue one job immediately and any delayed jobs.
   await _queue.add(jobName, serialized);
 
   // 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 < retryDelays.length; i++) {
-    const delay = parseInt(retryDelays[i]) * 1000;
-    const attempts = i === retryDelays.length - 1 ? 10 : 0;
+  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,
       attempts,