prune tx by count

Author: arcoraven

src/db/transactions/db.ts

@@ -128,7 +128,7 @@ export class TransactionDB {
    */
   static bulkDelete = async (queueIds: string[]) => {
     if (queueIds.length === 0) {
-      return [];
+      return 0;
     }
 
     const keys = queueIds.map(this.transactionDetailsKey);
@@ -176,38 +176,26 @@ export class TransactionDB {
   };
 
   /**
-   * Deletes transactions between a time range.
-   * @param from Date?
-   * @param to Date?
+   * Prunes transaction details and lists, keeping the latest `keep` amount/
+   * @param keep number - The max recent transactions to not prune.
+   * @returns number - The number of transactions pruned.
    */
-  static pruneTransactionLists = async (args: { from?: Date; to?: Date }) => {
-    const { from, to } = args;
-    const min = from ? toSeconds(from) : 0;
-    const max = to ? toSeconds(to) : "+inf";
+  static pruneTransactionDetailsAndLists = async (keep: number) => {
+    // Delete up to `keep - 1` index, inclusive.
+    const stop = -keep - 1;
+
+    const queueIds = await redis.zrange(this.queuedTransactionsKey, 0, stop);
+    const numPruned = await this.bulkDelete(queueIds);
 
-    // 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)
+      .zremrangebyrank(this.queuedTransactionsKey, 0, stop)
+      .zremrangebyrank(this.minedTransactionsKey, 0, stop)
+      .zremrangebyrank(this.cancelledTransactionsKey, 0, stop)
+      .zremrangebyrank(this.erroredTransactionsKey, 0, stop)
       .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);
+    return numPruned;
   };
 }
 

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

@@ -2,6 +2,7 @@ import { Static, Type } from "@sinclair/typebox";
 import { FastifyInstance } from "fastify";
 import { StatusCodes } from "http-status-codes";
 import { TransactionDB } from "../../../db/transactions/db";
+import { env } from "../../../utils/env";
 import { normalizeAddress } from "../../../utils/primitiveTypes";
 import { standardResponseSchema } from "../../schemas/sharedApiSchemas";
 import {
@@ -42,11 +43,10 @@ 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 10k total transactions for performance reasons.
       const { transactions } = await TransactionDB.getTransactionListByStatus({
         status: "queued",
         page: 1,
-        limit: 10_000,
+        limit: env.TRANSACTION_HISTORY_COUNT,
       });
       const filtered = transactions.filter(
         (t) => t.chainId === chainId && t.from === walletAddress,

src/server/schemas/wallet/index.ts

@@ -8,7 +8,7 @@ export const walletHeaderSchema = Type.Object({
   }),
   "x-idempotency-key": Type.Optional(
     Type.String({
-      description: `Multiple transactions submitted with the same idempotency key will not send a new transaction for ${env.PRUNE_TRANSACTIONS} day(s).`,
+      description: `Transactions submitted with the same idempotency key will be de-duplicated. Only the last ${env.TRANSACTION_HISTORY_COUNT} transactions are compared.`,
     }),
   ),
 });

src/utils/env.ts

@@ -65,16 +65,9 @@ export const env = createEnv({
     // 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.
+    // TRANSACTION_HISTORY_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),
+    TRANSACTION_HISTORY_COUNT: z.coerce.number().default(400_000),
     CLIENT_ANALYTICS_URL: z
       .union([UrlSchema, z.literal("")])
       .default("https://c.thirdweb.com/event"),
@@ -109,8 +102,7 @@ export const env = createEnv({
     ENABLE_HTTPS: process.env.ENABLE_HTTPS,
     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,
+    TRANSACTION_HISTORY_COUNT: process.env.TRANSACTION_HISTORY_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,

src/worker/queues/queues.ts

@@ -10,7 +10,7 @@ export const defaultJobOptions: JobsOptions = {
     count: 10_000,
   },
   // Purge failed jobs.
-  // These limits should be high to debug Redis instances.
+  // These limits should be sufficiently high to debug/retry.
   removeOnFail: {
     age: 30 * 24 * 60 * 60,
     count: 100_000,

src/worker/tasks/pruneTransactionsWorker.ts

@@ -6,15 +6,8 @@ import { PRUNE_TRANSACTIONS_QUEUE_NAME } from "../queues/pruneTransactionsQueue"
 import { logWorkerExceptions } from "../queues/queues";
 
 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()}.`);
-
-  // Prune transactions DB to the most recent `PRUNE_TRANSACTIONS_COUNT`.
-  const numPruned = await TransactionDB.pruneTransactionDetails(
-    env.PRUNE_TRANSACTIONS_KEEP_COUNT,
+  const numPruned = await TransactionDB.pruneTransactionDetailsAndLists(
+    env.TRANSACTION_HISTORY_COUNT,
   );
   job.log(`Pruned ${numPruned} transaction details.`);
 };