feat: nonce stats endpoint + queue/nonce log cron (#628)

* feat: nonce stats endpoint + queue/nonce log cron

* chore: Add LogQueue to bullboard and add queue initialisation

* fix: add missed log line

* fix: Update querystring schema for getNonceDetailsRoute

* fix: use last used onchain nonce

* ramped up nonce heuristics

* feat: use redis pipelines

* parse and sort

* chore: rename LogQueue to NonceHealthCheckQueue

* Simplify nonce health heuristic

* feat: Improve nonce health check algorithm

* chore: Update getOnchainNonce to getLastUsedOnchainNonce for consistency

* chore: Refactor nonce health check worker for improved consistency and performance

* fix logger: override default winston log levels to match env

* fixes: consistent address key, run every 20s

* feat: Filter used backend wallets by chainId and walletAddress

This commit modifies the `getUsedBackendWallets` function in the `walletNonce.ts` file to add optional parameters `chainId` and `walletAddress`. These parameters allow filtering the used backend wallets based on the provided chainId and walletAddress. The function now uses these parameters to construct the Redis key pattern for fetching the used backend wallets.

* refactor: Update Queue name in NonceHealthCheckQueue class

* chore: Refactor getCurrentNonceState function to use BigInt for onchainNonce

* chore: Update nonce health check worker frequency to run every minute

Author: d4mr

src/db/wallets/walletNonce.ts

@@ -1,11 +1,48 @@
-import { Address, eth_getTransactionCount, getRpcClient } from "thirdweb";
+import {
+  Address,
+  eth_getTransactionCount,
+  getAddress,
+  getRpcClient,
+} from "thirdweb";
 import { getChain } from "../../utils/chain";
 import { logger } from "../../utils/logger";
 import { normalizeAddress } from "../../utils/primitiveTypes";
 import { redis } from "../../utils/redis/redis";
 import { thirdwebClient } from "../../utils/sdk";
 import { updateNonceMap } from "./nonceMap";
 
+/**
+ * Get all used backend wallets.
+ * Filters by chainId and walletAddress if provided.
+ * Reads all the keys in the format `nonce:${chainId}:${walletAddress}` in the Redis DB.
+ *
+ * @example
+ * getUsedBackendWallets()
+ * // [ { chainId: 80001, walletAddress: "0x1234...5678" } ]
+ */
+export const getUsedBackendWallets = async (
+  chainId?: number,
+  walletAddress?: Address,
+): Promise<
+  {
+    chainId: number;
+    walletAddress: Address;
+  }[]
+> => {
+  const keys = await redis.keys(
+    `nonce:${chainId ?? "*"}:${
+      walletAddress ? normalizeAddress(walletAddress) : "*"
+    }`,
+  );
+  return keys.map((key) => {
+    const tokens = key.split(":");
+    return {
+      chainId: parseInt(tokens[1]),
+      walletAddress: getAddress(tokens[2]),
+    };
+  });
+};
+
 /**
  * The "last used nonce" stores the last nonce submitted onchain.
  * Example: "25"
@@ -14,8 +51,23 @@ export const lastUsedNonceKey = (chainId: number, walletAddress: Address) =>
   `nonce:${chainId}:${normalizeAddress(walletAddress)}`;
 
 /**
- * The "recycled nonces" sorted set stores nonces to be reused or cancelled, sorted by nonce value.
- * Example: [ "23", "24", "25" ]
+ * Split the last used nonce key into chainId and walletAddress.
+ * @param key
+ * @returns { chainId: number, walletAddress: Address }
+ * @example
+ * splitLastUsedNonceKey("nonce:80001:0x1234...5678")
+ * // { chainId: 80001, walletAddress: "0x1234...5678" }
+ */
+export const splitLastUsedNonceKey = (key: string) => {
+  const _splittedKeys = key.split(":");
+  const walletAddress = normalizeAddress(_splittedKeys[2]);
+  const chainId = parseInt(_splittedKeys[1]);
+  return { walletAddress, chainId };
+};
+
+/**
+ * The "recycled nonces" set stores unsorted nonces to be reused or cancelled.
+ * Example: [ "25", "23", "24" ]
  */
 export const recycledNoncesKey = (chainId: number, walletAddress: Address) =>
   `nonce-recycled:${chainId}:${normalizeAddress(walletAddress)}`;

src/server/middleware/adminRoutes.ts

@@ -10,6 +10,7 @@ import { env } from "../../utils/env";
 import { CancelRecycledNoncesQueue } from "../../worker/queues/cancelRecycledNoncesQueue";
 import { MigratePostgresTransactionsQueue } from "../../worker/queues/migratePostgresTransactionsQueue";
 import { MineTransactionQueue } from "../../worker/queues/mineTransactionQueue";
+import { NonceHealthCheckQueue } from "../../worker/queues/nonceHealthCheckQueue";
 import { NonceResyncQueue } from "../../worker/queues/nonceResyncQueue";
 import { ProcessEventsLogQueue } from "../../worker/queues/processEventLogsQueue";
 import { ProcessTransactionReceiptsQueue } from "../../worker/queues/processTransactionReceiptsQueue";
@@ -30,6 +31,7 @@ const QUEUES: Queue[] = [
   PruneTransactionsQueue.q,
   MigratePostgresTransactionsQueue.q,
   NonceResyncQueue.q,
+  NonceHealthCheckQueue.q,
 ];
 
 export const withAdminRoutes = async (fastify: FastifyInstance) => {

src/server/routes/admin/nonces.ts

@@ -0,0 +1,180 @@
+import { Static, Type } from "@sinclair/typebox";
+import { FastifyInstance } from "fastify";
+import { StatusCodes } from "http-status-codes";
+import {
+  Address,
+  eth_getTransactionCount,
+  getAddress,
+  getRpcClient,
+} from "thirdweb";
+import {
+  getUsedBackendWallets,
+  lastUsedNonceKey,
+  recycledNoncesKey,
+  sentNoncesKey,
+} from "../../../db/wallets/walletNonce";
+import { getChain } from "../../../utils/chain";
+import { redis } from "../../../utils/redis/redis";
+import { thirdwebClient } from "../../../utils/sdk";
+import { standardResponseSchema } from "../../schemas/sharedApiSchemas";
+import { walletWithAddressParamSchema } from "../../schemas/wallet";
+
+export const responseBodySchema = Type.Object({
+  result: Type.Array(
+    Type.Object({
+      walletAddress: Type.String({
+        description: "Backend Wallet Address",
+        examples: ["0xcedf3b4d8f7f1f7e0f7f0f7f0f7f0f7f0f7f0f7f"],
+      }),
+      chainId: Type.Number({
+        description: "Chain ID",
+        examples: [80002],
+      }),
+      onchainNonce: Type.Integer({
+        description: "Last mined nonce",
+        examples: [0],
+      }),
+      lastUsedNonce: Type.Integer({
+        description: "Last incremented nonce sent to the RPC",
+        examples: [0],
+      }),
+      sentNonces: Type.Array(Type.Integer(), {
+        description:
+          "Nonces that were successfully sent to the RPC but not mined yet (in descending order)",
+        examples: [[2, 1, 0]],
+      }),
+      recycledNonces: Type.Array(Type.Integer(), {
+        examples: [[3, 2, 1]],
+        description:
+          "Nonces that were acquired but failed to be sent to the blockchain, waiting to be recycled or cancelled (in descending order)",
+      }),
+    }),
+  ),
+});
+
+responseBodySchema.example = {
+  result: [
+    {
+      walletAddress: "0xcedf3b4d8f7f1f7e0f7f0f7f0f7f0f7f0f7f0f7f",
+      onchainNonce: 2,
+      lastUsedNonce: 8,
+      recycledNonces: [6, 7],
+      chainId: 80002,
+    },
+  ],
+};
+
+const walletWithAddressQuerySchema = Type.Partial(walletWithAddressParamSchema);
+
+export async function getNonceDetailsRoute(fastify: FastifyInstance) {
+  fastify.route<{
+    Querystring: Static<typeof walletWithAddressQuerySchema>;
+    Reply: Static<typeof responseBodySchema>;
+  }>({
+    method: "GET",
+    url: "/admin/nonces",
+    schema: {
+      summary: "Get nonce status details for wallets",
+      description:
+        "Admin route to get nonce status details for all wallets filtered by address and chain ",
+      tags: ["Admin"],
+      operationId: "nonceDetails",
+      querystring: walletWithAddressQuerySchema,
+      response: {
+        ...standardResponseSchema,
+        [StatusCodes.OK]: responseBodySchema,
+      },
+      hide: true,
+    },
+    handler: async (request, reply) => {
+      const { walletAddress, chain } = request.query;
+      const result = await getNonceDetails({
+        walletAddress: walletAddress ? getAddress(walletAddress) : undefined,
+        chainId: chain ? parseInt(chain) : undefined,
+      });
+
+      reply.status(StatusCodes.OK).send({
+        result,
+      });
+    },
+  });
+}
+
+export const getNonceDetails = async ({
+  walletAddress,
+  chainId,
+}: {
+  walletAddress?: Address;
+  chainId?: number;
+} = {}) => {
+  const usedBackendWallets = await getUsedBackendWallets(
+    chainId,
+    walletAddress,
+  );
+
+  const pipeline = redis.pipeline();
+  const onchainNoncePromises: Promise<number>[] = [];
+
+  const keyMap = usedBackendWallets.map(({ chainId, walletAddress }) => {
+    pipeline.get(lastUsedNonceKey(chainId, walletAddress));
+    pipeline.smembers(sentNoncesKey(chainId, walletAddress));
+    pipeline.smembers(recycledNoncesKey(chainId, walletAddress));
+
+    onchainNoncePromises.push(getLastUsedOnchainNonce(chainId, walletAddress));
+
+    return { chainId, walletAddress };
+  });
+
+  const [pipelineResults, onchainNonces] = await Promise.all([
+    pipeline.exec(),
+    Promise.all(onchainNoncePromises),
+  ]);
+
+  if (!pipelineResults) {
+    throw new Error("Failed to execute Redis pipeline");
+  }
+
+  return keyMap.map((key, index) => {
+    const pipelineOffset = index * 3;
+    const [lastUsedNonceResult, sentNoncesResult, recycledNoncesResult] =
+      pipelineResults.slice(pipelineOffset, pipelineOffset + 3);
+
+    return {
+      walletAddress: key.walletAddress,
+      chainId: key.chainId,
+      onchainNonce: onchainNonces[index],
+      lastUsedNonce: parseInt(lastUsedNonceResult[1] as string) ?? 0,
+      sentNonces: (sentNoncesResult[1] as string[])
+        .map((nonce) => parseInt(nonce))
+        .sort((a, b) => b - a),
+      recycledNonces: (recycledNoncesResult[1] as string[])
+        .map((nonce) => parseInt(nonce))
+        .sort((a, b) => b - a),
+    };
+  });
+};
+
+/*
+ * Get the last used nonce onchain
+ * @param chainId Chain ID
+ * @param walletAddress Wallet address
+ * @returns Next unused nonce
+ */
+export const getLastUsedOnchainNonce = async (
+  chainId: number,
+  walletAddress: string,
+) => {
+  const rpcRequest = getRpcClient({
+    client: thirdwebClient,
+    chain: await getChain(chainId),
+  });
+
+  // The next unused nonce = transactionCount.
+  const transactionCount = await eth_getTransactionCount(rpcRequest, {
+    address: walletAddress,
+    blockTag: "latest",
+  });
+
+  const onchainNonce = transactionCount - 1;
+  return onchainNonce;
+};

src/server/routes/index.ts

@@ -1,4 +1,5 @@
 import { FastifyInstance } from "fastify";
+import { getNonceDetailsRoute } from "./admin/nonces";
 import { getTransactionDetails } from "./admin/transaction";
 import { createAccessToken } from "./auth/access-tokens/create";
 import { getAllAccessTokens } from "./auth/access-tokens/getAll";
@@ -261,4 +262,5 @@ export const withRoutes = async (fastify: FastifyInstance) => {
 
   // Admin
   await fastify.register(getTransactionDetails);
+  await fastify.register(getNonceDetailsRoute);
 };

src/utils/logger.ts

@@ -1,31 +1,57 @@
 import { createLogger, format, transports } from "winston";
 import { env } from "./env";
 
+type LogLevels = typeof env.LOG_LEVEL;
+
+// Define custom log levels that strictly match the log levels in the env file
+const customLevels: {
+  levels: { [K in LogLevels]: number };
+  colors: { [K in LogLevels]: string };
+} = {
+  levels: {
+    fatal: 0,
+    error: 1,
+    warn: 2,
+    info: 3,
+    debug: 4,
+    trace: 5,
+  },
+  colors: {
+    fatal: "red",
+    error: "red",
+    warn: "yellow",
+    info: "green",
+    debug: "blue",
+    trace: "gray",
+  },
+};
+
 // Custom filter for stdout transport
-const filterOnlyInfoAndWarn = format((info) => {
-  if (info.level === "error") {
-    return false; // Exclude 'error' level logs
+const filterNonErrors = format((info) => {
+  if (info.level === "error" || info.level === "fatal") {
+    return false; // only include non-error logs
   }
   return info;
 });
 
 // Custom filter for stderr transport
-const filterOnlyErrors = format((info) => {
-  if (info.level !== "error") {
-    return false; // Exclude non-error level logs
+const filterErrorsAndFatal = format((info) => {
+  if (info.level === "error" || info.level === "fatal") {
+    return info; // only include error and fatal logs
   }
-  return info;
+  return false;
 });
 
 const colorizeFormat = () => {
   if (env.NODE_ENV === "development") {
-    return format.colorize();
+    return format.colorize({ colors: customLevels.colors });
   } else {
     return format.uncolorize();
   }
 };
 
 const winstonLogger = createLogger({
+  levels: customLevels.levels,
   level: env.LOG_LEVEL,
   format: format.combine(
     format.timestamp(),
@@ -40,15 +66,14 @@ const winstonLogger = createLogger({
     }),
   ),
   transports: [
-    // Transport for stdout
+    // Transport for stdout (non-error logs)
     new transports.Console({
-      format: format.combine(filterOnlyInfoAndWarn()),
-      stderrLevels: [], // Don't log "error" to stdout
+      format: format.combine(filterNonErrors()),
     }),
-    // Transport for stderr
+    // Transport for stderr (error and fatal logs)
     new transports.Console({
-      format: format.combine(filterOnlyErrors()),
-      stderrLevels: ["error"], // Ensure errors go to stderr
+      format: format.combine(filterErrorsAndFatal()),
+      stderrLevels: ["error", "fatal"],
     }),
   ],
 });
@@ -85,7 +110,7 @@ export const logger = ({
   }
 
   if (error) {
-    winstonLogger.error(`${prefix}${message}${suffix}`, { error });
+    winstonLogger.error(level, `${prefix}${message}${suffix}`, { error });
   } else {
     winstonLogger.log(level, `${prefix}${message}${suffix}`);
   }

src/worker/index.ts

@@ -10,6 +10,7 @@ import {
 import { initCancelRecycledNoncesWorker } from "./tasks/cancelRecycledNoncesWorker";
 import { initMigratePostgresTransactionsWorker } from "./tasks/migratePostgresTransactionsWorker";
 import { initMineTransactionWorker } from "./tasks/mineTransactionWorker";
+import { initNonceHealthCheckWorker } from "./tasks/nonceHealthCheckWorker";
 import { initNonceResyncWorker } from "./tasks/nonceResyncWorker";
 import { initProcessEventLogsWorker } from "./tasks/processEventLogsWorker";
 import { initProcessTransactionReceiptsWorker } from "./tasks/processTransactionReceiptsWorker";
@@ -25,6 +26,9 @@ export const initWorker = async () => {
   initSendTransactionWorker();
   initMineTransactionWorker();
   initSendWebhookWorker();
+
+  initNonceHealthCheckWorker();
+
   await initMigratePostgresTransactionsWorker();
 
   await initNonceResyncWorker();