Optimize transaction throughput using batch requests (#278)

adam-maj · web-flow · commit 66b152c543b6 · 2023-10-31T13:42:36.000-07:00
* Send transactions in one batch

* Update yarn test:load
diff --git a/src/worker/tasks/processTx.ts b/src/worker/tasks/processTx.ts
@@ -1,8 +1,12 @@
 import { Static } from "@sinclair/typebox";
-import { getDefaultGasOverrides } from "@thirdweb-dev/sdk";
+import {
+  StaticJsonRpcBatchProvider,
+  getDefaultGasOverrides,
+} from "@thirdweb-dev/sdk";
 import { ERC4337EthersSigner } from "@thirdweb-dev/wallets/dist/declarations/src/evm/connectors/smart-wallet/lib/erc4337-signer";
 import { ethers } from "ethers";
 import { BigNumber } from "ethers/lib/ethers";
+import { RpcResponse } from "viem/_types/utils/rpc";
 import { prisma } from "../../db/client";
 import { getConfiguration } from "../../db/configuration/getConfiguration";
 import { getQueuedTxs } from "../../db/transactions/getQueuedTxs";
@@ -100,6 +104,15 @@ export const processTx = async () => {
             walletAddress,
           });
 
+          const [signer, provider] = await Promise.all([
+            sdk.getSigner(),
+            sdk.getProvider() as StaticJsonRpcBatchProvider,
+          ]);
+
+          if (!signer || !provider) {
+            return;
+          }
+
           // - For each wallet address, check the nonce in database and the mempool
           const [walletBalance, mempoolNonceData, dbNonceData, gasOverrides] =
             await Promise.all([
@@ -110,7 +123,7 @@ export const processTx = async () => {
                 chainId,
                 address: walletAddress,
               }),
-              getDefaultGasOverrides(sdk.getProvider()),
+              getDefaultGasOverrides(provider),
             ]);
 
           // Wallet Balance Webhook
@@ -159,93 +172,115 @@ export const processTx = async () => {
             startNonce = dbNonce;
           }
 
-          let incrementNonce = 0;
-
-          // - Wait for transactions to be sent successfully
-          const txStatuses: SentTxStatus[] = [];
+          // Group all transactions into a single batch rpc request
+          const rpcRequests = [];
           for (const i in txsToSend) {
             const tx = txsToSend[i];
             const nonce = startNonce.add(i);
 
-            try {
-              logger.worker.info(
-                `[Transaction] [${tx.queueId}] Sending with nonce '${nonce}'`,
-              );
-              const res = await sdk.getSigner()!.sendTransaction({
-                to: tx.toAddress!,
-                from: tx.fromAddress!,
-                data: tx.data!,
-                value: tx.value!,
-                nonce,
-                ...gasOverrides,
-              });
-
-              logger.worker.info(
-                `[Transaction] [${tx.queueId}] Submitted with nonce '${nonce}' & hash '${res.hash}'`,
-              );
-
-              // - Keep track of the number of transactions that went through successfully
-              incrementNonce++;
-              txStatuses.push({
-                status: TransactionStatusEnum.Submitted,
-                queueId: tx.queueId!,
-                res,
-                sentAtBlockNumber: await sdk.getProvider().getBlockNumber(),
-              });
-            } catch (err: any) {
-              logger.worker.warn(
-                `[Transaction] [${tx.queueId}] [Nonce: ${nonce}] Failed to send with error - ${err}`,
-              );
-
-              txStatuses.push({
-                status: TransactionStatusEnum.Errored,
-                queueId: tx.queueId!,
-                errorMessage:
-                  err?.message ||
-                  err?.toString() ||
-                  `Failed to handle transaction`,
-              });
-            }
-
-            // - After sending transactions, update database for each transaction
-            await Promise.all(
-              txStatuses.map(async (tx) => {
-                switch (tx.status) {
-                  case TransactionStatusEnum.Submitted:
-                    await updateTx({
-                      pgtx,
-                      queueId: tx.queueId,
-                      data: {
-                        status: TransactionStatusEnum.Submitted,
-                        res: tx.res,
-                        sentAtBlockNumber: await sdk
-                          .getProvider()
-                          .getBlockNumber(),
-                      },
-                    });
-                    break;
-                  case TransactionStatusEnum.Errored:
-                    await updateTx({
-                      pgtx,
-                      queueId: tx.queueId,
-                      data: {
-                        status: TransactionStatusEnum.Errored,
-                        errorMessage: tx.errorMessage,
-                      },
-                    });
-                    break;
-                }
-              }),
-            );
+            const txRequest = await signer.populateTransaction({
+              to: tx.toAddress!,
+              from: tx.fromAddress!,
+              data: tx.data!,
+              value: tx.value!,
+              nonce,
+              ...gasOverrides,
+            });
+            const signature = await signer.signTransaction(txRequest);
 
-            // - And finally update the nonce with the number of successful transactions
-            await updateWalletNonce({
-              pgtx,
-              address: walletAddress,
-              chainId,
-              nonce: startNonce.toNumber() + incrementNonce,
+            rpcRequests.push({
+              id: i,
+              jsonrpc: "2.0",
+              method: "eth_sendRawTransaction",
+              params: [signature],
             });
           }
+
+          // Send all the transactions as one batch request
+          const res = await fetch(provider.connection.url, {
+            method: "POST",
+            headers: {
+              "Content-Type": "application/json",
+            },
+            body: JSON.stringify(rpcRequests),
+          });
+          const rpcResponses: RpcResponse[] = await res.json();
+
+          // Check how many transactions succeeded and increment nonce
+          const incrementNonce = rpcResponses.reduce((acc, curr) => {
+            return curr.result && !curr.error ? acc + 1 : acc;
+          }, 0);
+
+          await updateWalletNonce({
+            pgtx,
+            address: walletAddress,
+            chainId,
+            nonce: startNonce.toNumber() + incrementNonce,
+          });
+
+          // Update transaction records with updated data
+          const txStatuses: SentTxStatus[] = await Promise.all(
+            rpcResponses.map(async (rpcRes, i) => {
+              const tx = txsToSend[i];
+              if (rpcRes.result) {
+                const txHash = rpcRes.result;
+                const txRes = await provider.getTransaction(txHash);
+
+                return {
+                  status: TransactionStatusEnum.Submitted,
+                  queueId: tx.queueId!,
+                  res: txRes,
+                  sentAtBlockNumber: await provider.getBlockNumber(),
+                };
+              } else {
+                logger.worker.warn(
+                  `[Transaction] [${
+                    tx.queueId
+                  }] Failed to send with error - ${JSON.stringify(
+                    rpcRes.error,
+                  )}`,
+                );
+
+                return {
+                  status: TransactionStatusEnum.Errored,
+                  queueId: tx.queueId!,
+                  errorMessage:
+                    rpcRes.error?.message ||
+                    rpcRes.error?.toString() ||
+                    `Failed to handle transaction`,
+                };
+              }
+            }),
+          );
+
+          // - After sending transactions, update database for each transaction
+          await Promise.all(
+            txStatuses.map(async (tx) => {
+              switch (tx.status) {
+                case TransactionStatusEnum.Submitted:
+                  await updateTx({
+                    pgtx,
+                    queueId: tx.queueId,
+                    data: {
+                      status: TransactionStatusEnum.Submitted,
+                      res: tx.res,
+                      sentAtBlockNumber: await provider.getBlockNumber(),
+                    },
+                  });
+                  break;
+                case TransactionStatusEnum.Errored:
+                  await updateTx({
+                    pgtx,
+                    queueId: tx.queueId,
+                    data: {
+                      status: TransactionStatusEnum.Errored,
+                      errorMessage: tx.errorMessage,
+                    },
+                  });
+                  break;
+              }
+            }),
+          );
         });
 
         // 5. Send all user operations in parallel with multi-dimensional nonce
diff --git a/test/load/index.ts b/test/load/index.ts
@@ -29,6 +29,7 @@ type OptionsInput = z.input<typeof OptionsSchema>;
 type Result = {
   requestTime: number;
   status?: "mined" | "errored" | "pending";
+  sendTime?: number;
   mineTime?: number;
 };
 
@@ -162,7 +163,7 @@ program
 
               const {
                 res: {
-                  result: { status },
+                  result: { status, sentAt, minedAt },
                 },
               } = await fetchEngine({
                 host: options.host as string,
@@ -173,8 +174,10 @@ program
 
               if (status === "mined") {
                 result.status = "mined";
-                result.mineTime = timer.ellapsed();
+                result.sendTime = timer.ellapsed(new Date(sentAt));
+                result.mineTime = timer.ellapsed(new Date(minedAt));
                 break;
+              } else if (res.result?.status === "sent") {
               } else if (res.result?.status === "errored") {
                 result.status = "errored";
                 break;
@@ -214,6 +217,16 @@ program
       return;
     }
 
+    console.log("\n\nsent times (seconds)");
+    console.table({
+      average:
+        results.reduce((acc, curr) => acc + curr.sendTime!, 0) / results.length,
+      minimum: results.sort((a, b) => a.sendTime! - b.sendTime!)[0].sendTime,
+      maximum: results.sort((a, b) => a.sendTime! - b.sendTime!)[
+        results.length - 1
+      ].sendTime,
+    });
+
     console.log("\n\nmine times (seconds)");
     console.table({
       average:
diff --git a/test/load/utils/time.ts b/test/load/utils/time.ts
@@ -14,7 +14,13 @@ export const time = async <TReturnType extends { [key: string]: any }>(
 export const createTimer = () => {
   const start = Date.now();
   return {
-    ellapsed: () => (Date.now() - start) / 1000,
+    ellapsed: (end?: Date) => {
+      if (end) {
+        return (end.getTime() - start) / 1000;
+      }
+
+      return (Date.now() - start) / 1000;
+    },
   };
 };
 
