Add nonce incrementing on nonce too low (#355)

* Block worker on lock acquisition

* Update

* Update

* Add nonce incrementing on nonce too low

* Update

* Update

* Don't add 1

Author: adam-maj

src/worker/tasks/processTx.ts

@@ -41,6 +41,13 @@ type SentTxStatus =
       errorMessage: string;
     };
 
+type RpcResponseData = {
+  queueId: string;
+  tx: ethers.providers.TransactionRequest;
+  res: RpcResponse;
+  sentAt: Date;
+};
+
 export const processTx = async () => {
   try {
     // 0. Initialize queueIds to send webhook
@@ -202,16 +209,14 @@ export const processTx = async () => {
               startNonce = dbNonce;
             }
 
-            // Group all transactions into a single batch rpc request
-            let sentTxCount = 0;
-            const rpcResponses: {
-              queueId: string;
-              tx: ethers.providers.TransactionRequest;
-              res: RpcResponse;
-              sentAt: Date;
-            }[] = [];
-            for (const tx of txsToSend) {
-              const nonce = startNonce.add(sentTxCount);
+            const rpcResponses: RpcResponseData[] = [];
+
+            let txIndex = 0;
+            let nonceIncrement = 0;
+
+            while (txIndex < txsToSend.length) {
+              const nonce = startNonce.add(nonceIncrement);
+              const tx = txsToSend[txIndex];
 
               try {
                 let value: ethers.BigNumberish = tx.value!;
@@ -298,17 +303,43 @@ export const processTx = async () => {
                   data: rpcResponse,
                 });
 
-                rpcResponses.push({
-                  queueId: tx.queueId!,
-                  tx: txRequest,
-                  res: rpcResponse,
-                  sentAt: new Date(),
-                });
-
                 if (!rpcResponse.error && !!rpcResponse.result) {
-                  sentTxCount++;
+                  // Success (continue to next transaction)
+                  nonceIncrement++;
+                  txIndex++;
+
+                  rpcResponses.push({
+                    queueId: tx.queueId!,
+                    tx: txRequest,
+                    res: rpcResponse,
+                    sentAt: new Date(),
+                  });
+                  sendWebhookForQueueIds.push(tx.queueId!);
+                } else if (
+                  typeof rpcResponse.error?.message === "string" &&
+                  (rpcResponse.error.message as string)
+                    .toLowerCase()
+                    .includes("nonce too low")
+                ) {
+                  // Nonce too low (retry same transaction with higher nonce)
+                  nonceIncrement++;
+                } else {
+                  // Error (continue to next transaction)
+                  txIndex++;
+
+                  rpcResponses.push({
+                    queueId: tx.queueId!,
+                    tx: txRequest,
+                    res: rpcResponse,
+                    sentAt: new Date(),
+                  });
+                  sendWebhookForQueueIds.push(tx.queueId!);
                 }
               } catch (err: any) {
+                // Error (continue to next transaction)
+                txIndex++;
+                sendWebhookForQueueIds.push(tx.queueId!);
+
                 logger({
                   service: "worker",
                   level: "warn",
@@ -328,20 +359,14 @@ export const processTx = async () => {
                       `Failed to handle transaction`,
                   },
                 });
-                sendWebhookForQueueIds.push(tx.queueId!);
               }
             }
 
-            // Check how many transactions succeeded and increment nonce
-            const incrementNonce = rpcResponses.reduce((acc, curr) => {
-              return curr.res.result && !curr.res.error ? acc + 1 : acc;
-            }, 0);
-
             await updateWalletNonce({
               pgtx,
               address: walletAddress,
               chainId,
-              nonce: startNonce.toNumber() + incrementNonce,
+              nonce: startNonce.add(nonceIncrement).toNumber(),
             });
 
             // Update transaction records with updated data