feat: add new debug endpoints to debug nonces (#635)

* feat: add new debug endpoints to debug nonces

* add files

* fixes while testing

* reset-nonces will sync nonces

* update resync logic

Author: arcoraven

src/db/transactions/db.ts

@@ -158,6 +158,7 @@ export class TransactionDB {
 
   /**
    * Lists all transaction details by status.
+   * Returns results paginated in descending order.
    * @param status "queued" | "mined" | "cancelled" | "errored"
    * @param page
    * @param limit

src/db/wallets/nonceMap.ts

@@ -0,0 +1,80 @@
+import { Address } from "thirdweb";
+import { env } from "../../utils/env";
+import { normalizeAddress } from "../../utils/primitiveTypes";
+import { redis } from "../../utils/redis/redis";
+
+/**
+ * The "nonce map" sorted set stores the queue ID that acquired each nonce.
+ * It is pruned to the latest 10k per wallet.
+ *
+ * Example:
+ *   {
+ *     "10": "e0fa731e-a947-4587-a48a-c56c02f8e7a8"
+ *     "11": "d111435a-1c0c-4308-ba40-59bad0868ee6"
+ *   }
+ */
+const nonceMapKey = (chainId: number, walletAddress: Address) =>
+  `nonce-map:${chainId}:${normalizeAddress(walletAddress)}`;
+
+export const updateNonceMap = async (args: {
+  chainId: number;
+  walletAddress: Address;
+  nonce: number;
+  queueId: string;
+}) => {
+  const { chainId, walletAddress, nonce, queueId } = args;
+  const key = nonceMapKey(chainId, walletAddress);
+  await redis.zadd(key, nonce, queueId);
+};
+
+/**
+ * Returns (nonce, queueId) pairs sorted by ascending nonce for the
+ * given wallet between the specified range.
+ */
+export const getNonceMap = async (args: {
+  chainId: number;
+  walletAddress: Address;
+  fromNonce: number;
+  toNonce?: number;
+}): Promise<{ nonce: number; queueId: string }[]> => {
+  const { chainId, walletAddress, fromNonce, toNonce } = args;
+  const key = nonceMapKey(chainId, walletAddress);
+
+  // Returns [ queueId1, nonce1, queueId2, nonce2, ... ]
+  const elementsWithScores = await redis.zrangebyscore(
+    key,
+    fromNonce,
+    // If toNonce is not provided, do not set an upper bound on the score.
+    toNonce ?? "+inf",
+    "WITHSCORES",
+  );
+
+  const result: { nonce: number; queueId: string }[] = [];
+  for (let i = 0; i < elementsWithScores.length; i += 2) {
+    result.push({
+      queueId: elementsWithScores[i],
+      nonce: parseInt(elementsWithScores[i + 1]),
+    });
+  }
+  return result;
+};
+
+export const pruneNonceMaps = async () => {
+  const pipeline = redis.pipeline();
+  const keys = await redis.keys("nonce-map:*");
+  for (const key of keys) {
+    pipeline.zremrangebyrank(key, 0, -env.NONCE_MAP_COUNT);
+  }
+  const results = await pipeline.exec();
+  if (!results) {
+    return 0;
+  }
+
+  let numDeleted = 0;
+  for (const [error, result] of results) {
+    if (!error) {
+      numDeleted += parseInt(result as string);
+    }
+  }
+  return numDeleted;
+};

src/db/wallets/walletNonce.ts

@@ -4,6 +4,7 @@ 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";
 
 /**
  * The "last used nonce" stores the last nonce submitted onchain.
@@ -76,30 +77,43 @@ export const isSentNonce = async (
 };
 
 /**
- * Acquire an unused nonce.
- * This should be used to send an EOA transaction with this nonce.
+ * Acquire an unused nonce to send an EOA transaction for the given backend wallet.
  * @param chainId
  * @param walletAddress
  * @returns number
  */
-export const acquireNonce = async (
-  chainId: number,
-  walletAddress: Address,
-): Promise<{ nonce: number; isRecycledNonce: boolean }> => {
+export const acquireNonce = async (args: {
+  queueId: string;
+  chainId: number;
+  walletAddress: Address;
+}): Promise<{ nonce: number; isRecycledNonce: boolean }> => {
+  const { queueId, chainId, walletAddress } = args;
+
+  let isRecycledNonce = false;
+
   // Try to acquire the lowest recycled nonce first
-  const recycledNonce = await _acquireRecycledNonce(chainId, walletAddress);
-  if (recycledNonce !== null) {
-    return { nonce: recycledNonce, isRecycledNonce: true };
+  let nonce = await _acquireRecycledNonce(chainId, walletAddress);
+  if (nonce !== null) {
+    isRecycledNonce = true;
+  } else {
+    // Else increment the last used nonce.
+    const key = lastUsedNonceKey(chainId, walletAddress);
+    nonce = await redis.incr(key);
+    if (nonce === 1) {
+      // If INCR returned 1, the nonce was not set.
+      // This may be a newly imported wallet.
+      // Sync the onchain value and increment again.
+      await syncLatestNonceFromOnchain(chainId, walletAddress);
+      nonce = await redis.incr(key);
+    }
   }
 
-  // Else increment the last used nonce.
-  const key = lastUsedNonceKey(chainId, walletAddress);
-  let nonce = await redis.incr(key);
-  if (nonce === 1) {
-    // If INCR returned 1, the nonce was not set.
-    // This may be a newly imported wallet.
-    nonce = await _syncNonce(chainId, walletAddress);
-  }
+  await updateNonceMap({
+    chainId,
+    walletAddress,
+    nonce,
+    queueId,
+  });
   return { nonce, isRecycledNonce: false };
 };
 
@@ -146,10 +160,14 @@ const _acquireRecycledNonce = async (
   return parseInt(result[0]);
 };
 
-const _syncNonce = async (
+/**
+ * Resync the nonce to the onchain nonce.
+ * @TODO: Redis lock this to make this method safe to call concurrently.
+ */
+export const syncLatestNonceFromOnchain = async (
   chainId: number,
   walletAddress: Address,
-): Promise<number> => {
+) => {
   const rpcRequest = getRpcClient({
     client: thirdwebClient,
     chain: await getChain(chainId),
@@ -162,8 +180,7 @@ const _syncNonce = async (
   });
 
   const key = lastUsedNonceKey(chainId, walletAddress);
-  await redis.set(key, transactionCount);
-  return transactionCount;
+  await redis.set(key, transactionCount - 1);
 };
 
 /**
@@ -195,11 +212,12 @@ export const deleteAllNonces = async () => {
 };
 
 /**
- * Resync the nonce i.e., max of transactionCount +1 and lastUsedNonce.
- * @param chainId
- * @param walletAddress
+ * Resync the nonce to the higher of (db nonce, onchain nonce).
  */
-export const resyncNonce = async (chainId: number, walletAddress: Address) => {
+export const syncLatestNonceFromOnchainIfHigher = async (
+  chainId: number,
+  walletAddress: Address,
+) => {
   const rpcRequest = getRpcClient({
     client: thirdwebClient,
     chain: await getChain(chainId),

src/server/routes/backend-wallet/getTransactions.ts

@@ -1,9 +1,9 @@
 import { Static, Type } from "@sinclair/typebox";
 import { FastifyInstance } from "fastify";
 import { StatusCodes } from "http-status-codes";
+import { getAddress } from "thirdweb";
 import { TransactionDB } from "../../../db/transactions/db";
-import { env } from "../../../utils/env";
-import { normalizeAddress } from "../../../utils/primitiveTypes";
+import { PaginationSchema } from "../../schemas/pagination";
 import { standardResponseSchema } from "../../schemas/sharedApiSchemas";
 import {
   TransactionSchema,
@@ -12,41 +12,64 @@ import {
 import { walletWithAddressParamSchema } from "../../schemas/wallet";
 import { getChainIdFromChain } from "../../utils/chain";
 
-const ParamsSchema = walletWithAddressParamSchema;
+const requestParamsSchema = walletWithAddressParamSchema;
+
+const requestQuerySchema = Type.Object({
+  ...PaginationSchema.properties,
+  status: Type.Union(
+    [
+      // Note: 'queued' returns all transcations, not just transactions currently queued.
+      Type.Literal("queued"),
+      Type.Literal("mined"),
+      Type.Literal("cancelled"),
+      Type.Literal("errored"),
+    ],
+    {
+      description:
+        "The status to query: 'queued', 'mined', 'errored', or 'cancelled'. Default: 'queued'",
+      default: "queued",
+    },
+  ),
+});
 
 const responseBodySchema = Type.Object({
   result: Type.Object({
     transactions: Type.Array(TransactionSchema),
   }),
 });
 
-export async function getAllTransactions(fastify: FastifyInstance) {
+export async function getTransactionsForBackendWallet(
+  fastify: FastifyInstance,
+) {
   fastify.route<{
-    Params: Static<typeof ParamsSchema>;
+    Querystring: Static<typeof requestQuerySchema>;
+    Params: Static<typeof requestParamsSchema>;
     Reply: Static<typeof responseBodySchema>;
   }>({
     method: "GET",
     url: "/backend-wallet/:chain/:walletAddress/get-all-transactions",
     schema: {
-      summary: "Get all transactions",
-      description: "Get all transactions for a backend wallet.",
+      summary: "Get recent transactions",
+      description: "Get recent transactions for this backend wallet.",
       tags: ["Backend Wallet"],
-      operationId: "getAllTransactions",
-      params: ParamsSchema,
+      operationId: "getTransactionsForBackendWallet",
+      querystring: requestQuerySchema,
+      params: requestParamsSchema,
       response: {
         ...standardResponseSchema,
         [StatusCodes.OK]: responseBodySchema,
       },
     },
     handler: async (req, res) => {
       const { chain, walletAddress: _walletAddress } = req.params;
+      const { page, limit, status } = req.query;
       const chainId = await getChainIdFromChain(chain);
-      const walletAddress = normalizeAddress(_walletAddress);
+      const walletAddress = getAddress(_walletAddress);
 
       const { transactions } = await TransactionDB.getTransactionListByStatus({
-        status: "queued",
-        page: 1,
-        limit: env.TRANSACTION_HISTORY_COUNT,
+        status,
+        page,
+        limit,
       });
       const filtered = transactions.filter(
         (t) => t.chainId === chainId && t.from === walletAddress,

src/server/routes/backend-wallet/getTransactionsByNonce.ts

@@ -0,0 +1,104 @@
+import { Static, Type } from "@sinclair/typebox";
+import { FastifyInstance } from "fastify";
+import { StatusCodes } from "http-status-codes";
+import { TransactionDB } from "../../../db/transactions/db";
+import { getNonceMap } from "../../../db/wallets/nonceMap";
+import { normalizeAddress } from "../../../utils/primitiveTypes";
+import { AnyTransaction } from "../../../utils/transaction/types";
+import { standardResponseSchema } from "../../schemas/sharedApiSchemas";
+import {
+  TransactionSchema,
+  toTransactionSchema,
+} from "../../schemas/transaction";
+import { walletWithAddressParamSchema } from "../../schemas/wallet";
+import { getChainIdFromChain } from "../../utils/chain";
+
+const requestParamsSchema = walletWithAddressParamSchema;
+
+const requestQuerySchema = Type.Object({
+  fromNonce: Type.Integer({
+    description: "The earliest nonce, inclusive.",
+    examples: [100],
+    minimum: 0,
+  }),
+  toNonce: Type.Optional(
+    Type.Integer({
+      description:
+        "The latest nonce, inclusive. If omitted, queries up to the latest sent nonce.",
+      examples: [100],
+      minimum: 0,
+    }),
+  ),
+});
+
+const responseBodySchema = Type.Object({
+  result: Type.Array(
+    Type.Object({
+      nonce: Type.Number(),
+      // Returns the transaction details by default.
+      // Falls back to the queueId if the transaction details have been pruned.
+      transaction: Type.Union([TransactionSchema, Type.String()]),
+    }),
+  ),
+});
+
+export async function getTransactionsForBackendWalletByNonce(
+  fastify: FastifyInstance,
+) {
+  fastify.route<{
+    Querystring: Static<typeof requestQuerySchema>;
+    Params: Static<typeof requestParamsSchema>;
+    Reply: Static<typeof responseBodySchema>;
+  }>({
+    method: "GET",
+    url: "/backend-wallet/:chain/:walletAddress/get-transactions-by-nonce",
+    schema: {
+      summary: "Get recent transactions by nonce",
+      description:
+        "Get recent transactions for this backend wallet, sorted by descending nonce.",
+      tags: ["Backend Wallet"],
+      operationId: "getTransactionsForBackendWalletByNonce",
+      querystring: requestQuerySchema,
+      params: requestParamsSchema,
+      response: {
+        ...standardResponseSchema,
+        [StatusCodes.OK]: responseBodySchema,
+      },
+    },
+    handler: async (req, res) => {
+      const { chain, walletAddress: _walletAddress } = req.params;
+      const { fromNonce, toNonce } = req.query;
+      const chainId = await getChainIdFromChain(chain);
+      const walletAddress = normalizeAddress(_walletAddress);
+
+      // Get queueIds.
+      const nonceMap = await getNonceMap({
+        chainId,
+        walletAddress,
+        fromNonce,
+        toNonce,
+      });
+
+      // Build map of { queueId => transaction }.
+      const queueIds = nonceMap.map(({ queueId }) => queueId);
+      const transactions = await TransactionDB.bulkGet(queueIds);
+      const transactionsMap = new Map<string, AnyTransaction>();
+      for (const transaction of transactions) {
+        transactionsMap.set(transaction.queueId, transaction);
+      }
+
+      // Hydrate the transaction, if found.
+      const result = nonceMap.map(({ nonce, queueId }) => {
+        const transaction = transactionsMap.get(queueId);
+        return {
+          nonce,
+          transaction: transaction ? toTransactionSchema(transaction) : queueId,
+        };
+      });
+
+      res.status(StatusCodes.OK).send({
+        result,
+      });
+    },
+  });
+}