prune transaction lists and details

Author: arcoraven

src/db/transactions/db.ts

@@ -1,5 +1,4 @@
 import superjson from "superjson";
-import { env } from "../../utils/env";
 import { redis } from "../../utils/redis/redis";
 import { AnyTransaction } from "../../utils/transaction/types";
 
@@ -42,29 +41,15 @@ export class TransactionDB {
   /**
    * Inserts or replaces a transaction details.
    * Also adds to the appropriate "status" sorted set.
-   * Sets a TTL for completed statuses (mined, cancelled, errored).
-   *
    * @param transaction
    */
   static set = async (transaction: AnyTransaction) => {
-    const pipeline = redis.pipeline();
-
-    const shouldExpire = ["mined", "cancelled", "errored"].includes(
-      transaction.status,
-    );
-    if (shouldExpire) {
-      const ttlSeconds = env.PRUNE_TRANSACTIONS * 24 * 60 * 60;
-      pipeline.setex(
-        this.transactionDetailsKey(transaction.queueId),
-        ttlSeconds,
-        superjson.stringify(transaction),
-      );
-    } else {
-      pipeline.set(
+    const pipeline = redis
+      .pipeline()
+      .set(
         this.transactionDetailsKey(transaction.queueId),
         superjson.stringify(transaction),
       );
-    }
 
     switch (transaction.status) {
       case "queued":
@@ -136,6 +121,20 @@ export class TransactionDB {
     return result;
   };
 
+  /**
+   * Deletes multiple transaction details by a list of queueIds.
+   * @param queueIds
+   * @returns number - The number of transaction details deleted.
+   */
+  static bulkDelete = async (queueIds: string[]) => {
+    if (queueIds.length === 0) {
+      return [];
+    }
+
+    const keys = queueIds.map(this.transactionDetailsKey);
+    return await redis.unlink(...keys);
+  };
+
   /**
    * Check if a transaction exists.
    * @param queueId
@@ -151,7 +150,7 @@ export class TransactionDB {
    * @param limit
    * @returns
    */
-  static listByStatus = async (args: {
+  static getTransactionListByStatus = async (args: {
     status: "queued" | "mined" | "cancelled" | "errored";
     page: number;
     limit: number;
@@ -180,21 +179,36 @@ export class TransactionDB {
    * Deletes transactions between a time range.
    * @param from Date?
    * @param to Date?
-   * @returns string[] List of queueIds
    */
-  static pruneTransactions = async (args: { from?: Date; to?: Date }) => {
+  static pruneTransactionLists = async (args: { from?: Date; to?: Date }) => {
     const { from, to } = args;
     const min = from ? toSeconds(from) : 0;
     const max = to ? toSeconds(to) : "+inf";
 
-    // Delete per-status sorted sets. Do not purge queued transactions.
+    // Delete per-status sorted sets.
     await redis
       .pipeline()
+      .zremrangebyscore(this.queuedTransactionsKey, min, max)
       .zremrangebyscore(this.minedTransactionsKey, min, max)
       .zremrangebyscore(this.cancelledTransactionsKey, min, max)
       .zremrangebyscore(this.erroredTransactionsKey, min, max)
       .exec();
   };
+
+  /**
+   * Prunes transaction details after `keep` transactions.
+   * Example: `keep=100` prunes all transaction details except the most recent 100.
+   * @param keep number - The count of recent transactions to not prune. All older transactions are pruned.
+   * @returns number - The number of transaction details pruned.
+   */
+  static pruneTransactionDetails = async (keep: number) => {
+    const queueIds = await redis.zrange(
+      this.queuedTransactionsKey,
+      0,
+      -keep - 1,
+    );
+    return await this.bulkDelete(queueIds);
+  };
 }
 
 const toSeconds = (timestamp: Date) => timestamp.getTime() / 1000;

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

@@ -43,7 +43,7 @@ export async function getAllTransactions(fastify: FastifyInstance) {
       const walletAddress = normalizeAddress(_walletAddress);
 
       // @TODO: This query is not optimized. Cap the results to the most recent 10k total transactions for performance reasons.
-      const { transactions } = await TransactionDB.listByStatus({
+      const { transactions } = await TransactionDB.getTransactionListByStatus({
         status: "queued",
         page: 1,
         limit: 10_000,

src/server/routes/transaction/getAll.ts

@@ -103,11 +103,12 @@ export async function getAllTx(fastify: FastifyInstance) {
     handler: async (request, reply) => {
       const { page, limit } = request.query;
 
-      const { transactions, totalCount } = await TransactionDB.listByStatus({
-        status: "queued",
-        page,
-        limit,
-      });
+      const { transactions, totalCount } =
+        await TransactionDB.getTransactionListByStatus({
+          status: "queued",
+          page,
+          limit,
+        });
 
       reply.status(StatusCodes.OK).send({
         result: {

src/server/routes/transaction/getAllDeployedContracts.ts

@@ -82,7 +82,7 @@ export async function getAllDeployedContracts(fastify: FastifyInstance) {
     handler: async (request, reply) => {
       const { page, limit } = request.query;
 
-      const { transactions } = await TransactionDB.listByStatus({
+      const { transactions } = await TransactionDB.getTransactionListByStatus({
         status: "queued",
         page,
         limit,

src/utils/env.ts

@@ -61,13 +61,20 @@ export const env = createEnv({
     ENABLE_HTTPS: boolSchema("false"),
     HTTPS_PASSPHRASE: z.string().default("thirdweb-engine"),
     TRUST_PROXY: z.boolean().default(false),
+    // REDIS_MAXMEMORY should be set below the amount of memory Redis has available
+    // to avoid running out of memory. If this limit is reached, keys are evicted
+    // with a "allkeys-lru" policy (removes least recently used keys).
+    REDIS_MAXMEMORY: z.string().default("900mb"),
     PRUNE_TRANSACTIONS: z
       .union([
         z.literal("true").transform(() => 7),
         z.literal("false").transform(() => 0),
         z.coerce.number().int(),
       ])
       .default(7),
+    // PRUNE_TRANSACTIONS_KEEP_COUNT defines the max transaction details to keep.
+    // In testing, storing about 400k consumes 1GB memory.
+    PRUNE_TRANSACTIONS_KEEP_COUNT: z.coerce.number().default(400_000),
     CLIENT_ANALYTICS_URL: z
       .union([UrlSchema, z.literal("")])
       .default("https://c.thirdweb.com/event"),
@@ -79,8 +86,8 @@ export const env = createEnv({
       .nonnegative()
       .default(0),
     REDIS_URL: z.string(),
-    SEND_TRANSACTION_QUEUE_CONCURRENCY: z.coerce.number().default(1500),
-    CONFIRM_TRANSACTION_QUEUE_CONCURRENCY: z.coerce.number().default(1500),
+    SEND_TRANSACTION_QUEUE_CONCURRENCY: z.coerce.number().default(2000),
+    CONFIRM_TRANSACTION_QUEUE_CONCURRENCY: z.coerce.number().default(2000),
     ENGINE_MODE: z
       .enum(["default", "sandbox", "server_only", "worker_only"])
       .default("default"),
@@ -103,6 +110,7 @@ export const env = createEnv({
     HTTPS_PASSPHRASE: process.env.HTTPS_PASSPHRASE,
     TRUST_PROXY: process.env.TRUST_PROXY,
     PRUNE_TRANSACTIONS: process.env.PRUNE_TRANSACTIONS,
+    PRUNE_TRANSACTIONS_KEEP_COUNT: process.env.PRUNE_TRANSACTIONS_KEEP_COUNT,
     CLIENT_ANALYTICS_URL: process.env.CLIENT_ANALYTICS_URL,
     SDK_BATCH_TIME_LIMIT: process.env.SDK_BATCH_TIME_LIMIT,
     SDK_BATCH_SIZE_LIMIT: process.env.SDK_BATCH_SIZE_LIMIT,
@@ -116,6 +124,7 @@ export const env = createEnv({
       process.env.CONFIRM_TRANSACTION_QUEUE_CONCURRENCY,
     ENGINE_MODE: process.env.ENGINE_MODE,
     GLOBAL_RATE_LIMIT_PER_MIN: process.env.GLOBAL_RATE_LIMIT_PER_MIN,
+    REDIS_MAXMEMORY: process.env.REDIS_MAXMEMORY,
   },
   onValidationError: (error: ZodError) => {
     console.error(

src/utils/redis/redis.ts

@@ -6,28 +6,23 @@ export const redis = new Redis(env.REDIS_URL, {
   enableAutoPipelining: true,
   maxRetriesPerRequest: null,
 });
-
-redis.on("error", (err) => () => {
+try {
+  await redis.config("SET", "maxmemory", env.REDIS_MAXMEMORY);
+} catch (error) {
   logger({
     level: "error",
-    message: `Redis error: ${err}`,
+    message: `Initializing Redis: ${error}`,
     service: "worker",
   });
-});
-redis.on("connect", () =>
-  logger({
-    level: "info",
-    message: "Redis connected",
-    service: "worker",
-  }),
-);
-redis.on("reconnecting", () =>
+}
+
+redis.on("error", (error) => () => {
   logger({
-    level: "info",
-    message: "Redis reconnecting",
+    level: "error",
+    message: `Redis error: ${error}`,
     service: "worker",
-  }),
-);
+  });
+});
 redis.on("ready", () => {
   logger({
     level: "info",

src/worker/queues/pruneTransactionsQueue.ts

@@ -13,5 +13,5 @@ _queue.setGlobalConcurrency(1);
 _queue.add("hourly-cron", "", {
   repeat: { pattern: "*/10 * * * *" },
   // Use a constant jobId to not insert multiple repeatable jobs.
-  jobId: "billing-reporter-hourly-cron",
+  jobId: "prune-transactions-cron",
 });

src/worker/tasks/pruneTransactionsWorker.ts

@@ -5,12 +5,18 @@ import { redis } from "../../utils/redis/redis";
 import { PRUNE_TRANSACTIONS_QUEUE_NAME } from "../queues/pruneTransactionsQueue";
 import { logWorkerExceptions } from "../queues/queues";
 
-const handler: Processor<any, void, string> = async (_: Job<string>) => {
+const handler: Processor<any, void, string> = async (job: Job<string>) => {
   // Purge transactions up to `PRUNE_TRANSACTIONS` days ago.
   const to = new Date();
   to.setDate(to.getDate() - env.PRUNE_TRANSACTIONS);
+  await TransactionDB.pruneTransactionLists({ to });
+  job.log(`Pruned transaction lists to ${to.toLocaleString()}.`);
 
-  await TransactionDB.pruneTransactions({ to });
+  // Prune transactions DB to the most recent `PRUNE_TRANSACTIONS_COUNT`.
+  const numPruned = await TransactionDB.pruneTransactionDetails(
+    env.PRUNE_TRANSACTIONS_KEEP_COUNT,
+  );
+  job.log(`Pruned ${numPruned} transaction details.`);
 };
 
 // Worker