handle resend rpc errors

Author: arcoraven

src/db/wallets/walletNonce.ts

@@ -53,7 +53,7 @@ export const acquireNonce = async (
  * @param walletAddress
  * @param nonce
  */
-export const releaseNonce = async (
+export const recycleNonce = async (
   chainId: number,
   walletAddress: Address,
   nonce: number,
@@ -93,14 +93,7 @@ const _syncNonce = async (
 
   const key = lastUsedNonceKey(chainId, walletAddress);
   await redis.set(key, transactionCount);
-
-  // Set last used nonce to transactionCount - 1.
-  // But don't lower the Redis nonce.
-
-  // @TODO:
-  // Set to max(value, transactionCount-1) -> 6
-
-  return await redis.incr(key);
+  return transactionCount;
 };
 
 /**

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

@@ -42,11 +42,11 @@ export async function getAllTransactions(fastify: FastifyInstance) {
       const chainId = await getChainIdFromChain(chain);
       const walletAddress = normalizeAddress(_walletAddress);
 
-      // @TODO: This query is not optimized. Cap the results to the most recent 1000 total transactions for performance reasons.
+      // @TODO: This query is not optimized. Cap the results to the most recent 10k total transactions for performance reasons.
       const { transactions } = await TransactionDB.listByStatus({
         status: "queued",
         page: 1,
-        limit: 1000,
+        limit: 10_000,
       });
       const filtered = transactions.filter(
         (t) => t.chainId === chainId && t.from === walletAddress,

src/server/schemas/transaction/index.ts

@@ -1,5 +1,5 @@
 import { Static, Type } from "@sinclair/typebox";
-import { Address, Hex } from "thirdweb";
+import { Hex } from "thirdweb";
 import { stringify } from "thirdweb/utils";
 import { AnyTransaction } from "../../../utils/transaction/types";
 
@@ -300,12 +300,9 @@ export const toTransactionSchema = (
 
     // User Operation
     signerAddress: transaction.from,
-    accountAddress:
-      "accountAddress" in transaction
-        ? (transaction.accountAddress as Address)
-        : null,
-    target: "target" in transaction ? (transaction.target as Address) : null,
-    sender: "sender" in transaction ? (transaction.sender as Address) : null,
+    accountAddress: transaction.accountAddress ?? null,
+    target: transaction.target ?? null,
+    sender: transaction.sender ?? null,
     initCode: null,
     callData: null,
     callGasLimit: null,

src/worker/queues/mineTransactionQueue.ts

@@ -16,10 +16,10 @@ export class MineTransactionQueue {
       ...defaultJobOptions,
       // Delay confirming the tx by 500ms.
       delay: 500,
-      // Check in 5s, 10s, 20s, 40s, 80s, 160s, 320s, 640s, 1280s, 2560s (~45 minutes)
-      // This needs to be long enough to handle transactions stuck in mempool for a while.
+      // Retry after 2s, 4s, 8s, 16s, 32s, 64s, 128s, 256s, 512s, 1024s (17 minutes)
+      // This needs to be long enough to handle transactions stuck in mempool.
       attempts: 10,
-      backoff: { type: "exponential", delay: 5_000 },
+      backoff: { type: "exponential", delay: 2_000 },
     },
   });
 

src/worker/tasks/cancelUnusedNoncesWorker.ts

@@ -1,7 +1,7 @@
 import { Job, Processor, Worker } from "bullmq";
 import { ethers } from "ethers";
 import { Address } from "thirdweb";
-import { releaseNonce } from "../../db/wallets/walletNonce";
+import { recycleNonce } from "../../db/wallets/walletNonce";
 import { isEthersErrorCode } from "../../utils/ethers";
 import { logger } from "../../utils/logger";
 import { redis } from "../../utils/redis/redis";
@@ -38,8 +38,8 @@ const handler: Processor<any, void, string> = async (job: Job<string>) => {
           if (isEthersErrorCode(e, ethers.errors.NONCE_EXPIRED)) {
             ignore.push(nonce);
           } else {
-            job.log(`Releasing nonce: ${nonce}`);
-            await releaseNonce(chainId, walletAddress, nonce);
+            job.log(`Recycling nonce: ${nonce}`);
+            await recycleNonce(chainId, walletAddress, nonce);
             fail.push(nonce);
           }
         }

src/worker/tasks/mineTransactionWorker.ts

@@ -10,7 +10,7 @@ import {
 import { stringify } from "thirdweb/utils";
 import { getUserOpReceiptRaw } from "thirdweb/wallets/smart";
 import { TransactionDB } from "../../db/transactions/db";
-import { releaseNonce } from "../../db/wallets/walletNonce";
+import { recycleNonce } from "../../db/wallets/walletNonce";
 import { getBlockNumberish } from "../../utils/block";
 import { getConfig } from "../../utils/cache/getConfig";
 import { getChain } from "../../utils/chain";
@@ -108,13 +108,15 @@ const _mineTransaction = async (
 ): Promise<MinedTransaction | null> => {
   assert(!sentTransaction.isUserOp);
 
-  const { queueId, chainId, sentTransactionHashes } = sentTransaction;
+  const { queueId, chainId, sentTransactionHashes, sentAtBlock, resendCount } =
+    sentTransaction;
 
   // Check all sent transaction hashes since any of them might succeed.
   const rpcRequest = getRpcClient({
     client: thirdwebClient,
     chain: await getChain(chainId),
   });
+  job.log(`Mining transactionHashes: ${sentTransactionHashes}`);
   const receiptResults = await Promise.allSettled(
     sentTransactionHashes.map((hash) =>
       eth_getTransactionReceipt(rpcRequest, { hash }),
@@ -125,7 +127,7 @@ const _mineTransaction = async (
   for (const result of receiptResults) {
     if (result.status === "fulfilled") {
       const receipt = result.value;
-      job.log(`Receipt found: ${receipt.transactionHash} `);
+      job.log(`Found receipt on block ${receipt.blockNumber}.`);
       return {
         ...sentTransaction,
         status: "mined",
@@ -145,20 +147,17 @@ const _mineTransaction = async (
 
   // Retry the transaction (after some initial delay).
   const config = await getConfig();
-  if (sentTransaction.resendCount < config.maxRetriesPerTx) {
-    const ellapsedBlocks =
-      (await getBlockNumberish(chainId)) - sentTransaction.sentAtBlock;
+  if (resendCount < config.maxRetriesPerTx) {
+    const blockNumber = await getBlockNumberish(chainId);
+    const ellapsedBlocks = blockNumber - sentAtBlock;
     if (ellapsedBlocks >= config.minEllapsedBlocksBeforeRetry) {
-      job.log(`Retrying transaction after ${ellapsedBlocks} blocks`);
-      logger({
-        service: "worker",
-        level: "warn",
-        queueId,
-        message: `Retrying transaction after ${ellapsedBlocks} blocks`,
-      });
+      const message = `Retrying transaction after ${ellapsedBlocks} blocks. blockNumber=${blockNumber} sentAtBlock=${sentAtBlock}`;
+      job.log(message);
+      logger({ service: "worker", level: "info", queueId, message });
+
       await SendTransactionQueue.add({
         queueId,
-        resendCount: sentTransaction.resendCount + 1,
+        resendCount: resendCount + 1,
       });
     }
   }
@@ -175,6 +174,7 @@ const _mineUserOp = async (
   const { chainId, userOpHash } = sentTransaction;
   const chain = await getChain(chainId);
 
+  job.log(`Mining userOpHash: ${userOpHash}`);
   const userOpReceiptRaw = await getUserOpReceiptRaw({
     client: thirdwebClient,
     chain,
@@ -185,30 +185,28 @@ const _mineUserOp = async (
   }
 
   const { transactionHash } = userOpReceiptRaw.receipt;
-  job.log(
-    `Found transactionHash ${transactionHash} from userOpHash ${userOpHash}.`,
-  );
+  job.log(`Found transactionHash: ${transactionHash}`);
 
   const rpcRequest = getRpcClient({ client: thirdwebClient, chain });
   const transaction = await eth_getTransactionByHash(rpcRequest, {
     hash: transactionHash,
   });
-  const transactionReceipt = await eth_getTransactionReceipt(rpcRequest, {
+  const receipt = await eth_getTransactionReceipt(rpcRequest, {
     hash: transaction.hash,
   });
 
   return {
     ...sentTransaction,
     status: "mined",
-    transactionHash: transactionReceipt.transactionHash,
+    transactionHash: receipt.transactionHash,
     minedAt: new Date(),
-    minedAtBlock: transactionReceipt.blockNumber,
-    transactionType: transactionReceipt.type,
-    onchainStatus: transactionReceipt.status,
-    gasUsed: transactionReceipt.gasUsed,
-    effectiveGasPrice: transactionReceipt.effectiveGasPrice,
-    gas: transactionReceipt.gasUsed,
-    cumulativeGasUsed: transactionReceipt.cumulativeGasUsed,
+    minedAtBlock: receipt.blockNumber,
+    transactionType: receipt.type,
+    onchainStatus: receipt.status,
+    gasUsed: receipt.gasUsed,
+    effectiveGasPrice: receipt.effectiveGasPrice,
+    gas: receipt.gasUsed,
+    cumulativeGasUsed: receipt.cumulativeGasUsed,
     sender: userOpReceiptRaw.sender as Address,
   };
 };
@@ -243,8 +241,8 @@ _worker.on("failed", async (job: Job<string> | undefined) => {
 
     if (!sentTransaction.isUserOp) {
       // Release the nonce to allow it to be reused or cancelled.
-      job.log(`Releasing nonce: ${sentTransaction.nonce}`);
-      await releaseNonce(
+      job.log(`Recycling nonce: ${sentTransaction.nonce}`);
+      await recycleNonce(
         sentTransaction.chainId,
         sentTransaction.from,
         sentTransaction.nonce,

src/worker/tasks/sendTransactionWorker.ts

@@ -8,7 +8,7 @@ import { bundleUserOp } from "thirdweb/wallets/smart";
 import type { TransactionSerializable } from "viem";
 import { getContractAddress } from "viem";
 import { TransactionDB } from "../../db/transactions/db";
-import { acquireNonce, releaseNonce } from "../../db/wallets/walletNonce";
+import { acquireNonce, recycleNonce } from "../../db/wallets/walletNonce";
 import { getAccount } from "../../utils/account";
 import { getBlockNumberish } from "../../utils/block";
 import { getChain } from "../../utils/chain";
@@ -51,8 +51,9 @@ const handler: Processor<any, void, string> = async (job: Job<string>) => {
 
   // SentTransaction = the transaction or userOp was submitted successfully.
   // ErroredTransaction = the transaction failed and should not be re-attempted.
+  // null = the transaction attemped to resend but was not needed. Ignore.
   // A thrown exception indicates a retry-able error occurred (e.g. RPC outage).
-  let resultTransaction: SentTransaction | ErroredTransaction;
+  let resultTransaction: SentTransaction | ErroredTransaction | null;
   if (transaction.status === "queued") {
     if (transaction.isUserOp) {
       resultTransaction = await _sendUserOp(job, transaction);
@@ -66,16 +67,18 @@ const handler: Processor<any, void, string> = async (job: Job<string>) => {
     return;
   }
 
-  // Handle the resulting "sent" or "errored" transaction.
-  await TransactionDB.set(resultTransaction);
-  await enqueueTransactionWebhook(resultTransaction);
+  if (resultTransaction) {
+    // Handle the resulting "sent" or "errored" transaction.
+    await TransactionDB.set(resultTransaction);
+    await enqueueTransactionWebhook(resultTransaction);
 
-  if (resultTransaction.status === "sent") {
-    job.log(`Transaction sent: ${stringify(resultTransaction)}.`);
-    await MineTransactionQueue.add({ queueId: resultTransaction.queueId });
-    await _reportUsageSuccess(resultTransaction);
-  } else if (resultTransaction.status === "errored") {
-    _reportUsageError(resultTransaction);
+    if (resultTransaction.status === "sent") {
+      job.log(`Transaction sent: ${stringify(resultTransaction)}.`);
+      await MineTransactionQueue.add({ queueId: resultTransaction.queueId });
+      await _reportUsageSuccess(resultTransaction);
+    } else if (resultTransaction.status === "errored") {
+      _reportUsageError(resultTransaction);
+    }
   }
 };
 
@@ -158,10 +161,10 @@ const _sendTransaction = async (
     transactionHash = sendTransactionResult.transactionHash;
     job.log(`Sent transaction: ${transactionHash}`);
   } catch (error: unknown) {
-    // Release the nonce if it has not expired.
+    // NONCE_EXPIRED indicates the nonce was already used onchain. Do not recycle it.
     if (!isEthersErrorCode(error, ethers.errors.NONCE_EXPIRED)) {
-      job.log(`Releasing nonce: ${nonce}`);
-      await releaseNonce(chainId, from, nonce);
+      job.log(`Recycling nonce: ${nonce}`);
+      await recycleNonce(chainId, from, nonce);
     }
     throw error;
   }
@@ -190,7 +193,7 @@ const _resendTransaction = async (
   job: Job,
   sentTransaction: SentTransaction,
   resendCount: number,
-): Promise<SentTransaction | ErroredTransaction> => {
+): Promise<SentTransaction | null> => {
   assert(!sentTransaction.isUserOp);
 
   // Populate the transaction with double gas.
@@ -223,10 +226,32 @@ const _resendTransaction = async (
 
   // Send transaction to RPC.
   // This call throws if the RPC rejects the transaction.
-  const account = await getAccount({ chainId, from });
-  const { transactionHash } = await account.sendTransaction(
-    populatedTransaction,
-  );
+  let transactionHash: Hex;
+  try {
+    const account = await getAccount({ chainId, from });
+    const sendTransactionResult = await account.sendTransaction(
+      populatedTransaction,
+    );
+    transactionHash = sendTransactionResult.transactionHash;
+    job.log(`Sent transaction: ${transactionHash}`);
+  } catch (error: unknown) {
+    // NONCE_EXPIRED indicates that this transaction was already mined.
+    // This is not an error.
+    if (!isEthersErrorCode(error, ethers.errors.NONCE_EXPIRED)) {
+      job.log(
+        `Nonce used. This transaction was likely already mined. Do not resend.`,
+      );
+      return null;
+    }
+    // REPLACEMENT_UNDERPRICED indicates the mempool is already aware of this transaction
+    // with >= gas fees. Wait for that one to be mined instead.
+    // This is not an error.
+    if (!isEthersErrorCode(error, ethers.errors.REPLACEMENT_UNDERPRICED)) {
+      job.log("A pending transaction exists with >= gas fees. Do not resend.");
+      return null;
+    }
+    throw error;
+  }
 
   return {
     ...sentTransaction,