Create new production release

apps/api/src/api/services/ramp/base.service.test.ts

@@ -0,0 +1,95 @@
+import {beforeEach, describe, expect, it, mock} from "bun:test";
+import {Op} from "sequelize";
+import sequelize from "../../../config/database";
+import QuoteTicket from "../../../models/quoteTicket.model";
+import {BaseRampService} from "./base.service";
+
+const transaction = { id: "cleanup-test-transaction" };
+const expectedDeleteBatchSize = 500;
+
+type QuoteCleanupWhere = {
+  expiresAt?: {
+    [Op.lt]?: Date;
+  };
+  id?: {
+    [Op.in]?: string[];
+  };
+  status?: string;
+};
+
+type QuoteCleanupOptions = {
+  attributes?: string[];
+  limit?: number;
+  order?: string[][];
+  transaction?: unknown;
+  where: QuoteCleanupWhere;
+};
+
+const transactionMock = mock(async (callback: (tx: unknown) => Promise<number>) => callback(transaction));
+const queryMock = mock(async () => [{ acquired: true }]);
+const updateMock = mock(async (_values: unknown, _options: QuoteCleanupOptions) => [3]);
+const findAllMock = mock(async (_options: QuoteCleanupOptions) => [{ id: "expired-quote-1" }, { id: "expired-quote-2" }]);
+const destroyMock = mock(async (_options: QuoteCleanupOptions) => 2);
+
+sequelize.transaction = transactionMock as unknown as typeof sequelize.transaction;
+sequelize.query = queryMock as unknown as typeof sequelize.query;
+QuoteTicket.update = updateMock as unknown as typeof QuoteTicket.update;
+QuoteTicket.findAll = findAllMock as unknown as typeof QuoteTicket.findAll;
+QuoteTicket.destroy = destroyMock as unknown as typeof QuoteTicket.destroy;
+
+describe("BaseRampService.cleanupExpiredQuotes", () => {
+  let service: BaseRampService;
+
+  beforeEach(() => {
+    service = new BaseRampService();
+    transactionMock.mockClear();
+    queryMock.mockClear();
+    updateMock.mockClear();
+    findAllMock.mockClear();
+    destroyMock.mockClear();
+    queryMock.mockImplementation(async () => [{ acquired: true }]);
+    updateMock.mockImplementation(async () => [3]);
+    findAllMock.mockImplementation(async () => [{ id: "expired-quote-1" }, { id: "expired-quote-2" }]);
+    destroyMock.mockImplementation(async () => 2);
+  });
+
+  it("does not update or delete quotes when another cleanup worker holds the advisory lock", async () => {
+    queryMock.mockImplementation(async () => [{ acquired: false }]);
+
+    const handledCount = await service.cleanupExpiredQuotes();
+
+    expect(handledCount).toBe(0);
+    expect(queryMock).toHaveBeenCalledTimes(1);
+    expect(updateMock).not.toHaveBeenCalled();
+    expect(findAllMock).not.toHaveBeenCalled();
+    expect(destroyMock).not.toHaveBeenCalled();
+  });
+
+  it("marks expired pending quotes and deletes old expired quotes in a bounded id batch", async () => {
+    const handledCount = await service.cleanupExpiredQuotes();
+
+    expect(handledCount).toBe(5);
+    expect(updateMock).toHaveBeenCalledTimes(1);
+    expect(findAllMock).toHaveBeenCalledTimes(1);
+    expect(destroyMock).toHaveBeenCalledTimes(1);
+
+    const updateOptions = updateMock.mock.calls[0][1];
+    expect(updateOptions.transaction).toBe(transaction);
+    expect(updateOptions.where.status).toBe("pending");
+    expect(updateOptions.where.expiresAt![Op.lt]).toBeInstanceOf(Date);
+
+    const findOptions = findAllMock.mock.calls[0][0];
+    expect(findOptions.attributes).toEqual(["id"]);
+    expect(findOptions.limit).toBe(expectedDeleteBatchSize);
+    expect(findOptions.order).toEqual([["expiresAt", "ASC"]]);
+    expect(findOptions.transaction).toBe(transaction);
+    expect(findOptions.where.status).toBe("expired");
+    expect(findOptions.where.expiresAt![Op.lt]).toBeInstanceOf(Date);
+
+    const destroyOptions = destroyMock.mock.calls[0][0];
+    expect(destroyOptions.transaction).toBe(transaction);
+    expect(destroyOptions.where.id![Op.in]).toEqual(["expired-quote-1", "expired-quote-2"]);
+    expect(destroyOptions.where.status).toBe("expired");
+    expect(destroyOptions.where.expiresAt![Op.lt]).toBeInstanceOf(Date);
+  });
+});

apps/api/src/api/services/ramp/base.service.ts

@@ -1,23 +1,51 @@
 import { RampPhase } from "@vortexfi/shared";
-import { Op, Transaction } from "sequelize";
+import { Op, QueryTypes, Transaction } from "sequelize";
 import { v4 as uuidv4 } from "uuid";
 import sequelize from "../../../config/database";
 import logger from "../../../config/logger";
 import QuoteTicket from "../../../models/quoteTicket.model";
 import RampState, { RampStateAttributes } from "../../../models/rampState.model";
 import { StateMetadata } from "../phases/meta-state-types";
 
+const EXPIRED_QUOTE_DELETE_BATCH_SIZE = 1000;
+const QUOTE_CLEANUP_ADVISORY_LOCK_NAMESPACE = 918521;
+const QUOTE_CLEANUP_ADVISORY_LOCK_KEY = 1;
+
 export class BaseRampService {
   /**
    * Clean up expired quotes by expiring them or deleting them from the database
    */
   public async cleanupExpiredQuotes(): Promise<number> {
+    return sequelize.transaction(async transaction => {
+      const [lock] = await sequelize.query<{ acquired: boolean }>(
+        "SELECT pg_try_advisory_xact_lock(:namespace, :key) AS acquired",
+        {
+          replacements: {
+            key: QUOTE_CLEANUP_ADVISORY_LOCK_KEY,
+            namespace: QUOTE_CLEANUP_ADVISORY_LOCK_NAMESPACE
+          },
+          transaction,
+          type: QueryTypes.SELECT
+        }
+      );
+
+      if (!lock?.acquired) {
+        logger.info("Skipping expired quote cleanup because another worker holds the cleanup lock");
+        return 0;
+      }
+
+      return this.cleanupExpiredQuotesWithLock(transaction);
+    });
+  }
+
+  private async cleanupExpiredQuotesWithLock(transaction: Transaction): Promise<number> {
     // Make quotes older than 10 minutes expire
-    let [count] = await QuoteTicket.update(
+    const [expiredPendingCount] = await QuoteTicket.update(
       {
         status: "expired"
       },
       {
+        transaction,
         where: {
           expiresAt: {
             [Op.lt]: new Date()
@@ -27,9 +55,12 @@ export class BaseRampService {
       }
     );
 
-    // Delete quotes that have been expired for more than 90 days
     const ninetyDaysAgo = new Date(Date.now() - 90 * 24 * 60 * 60 * 1000);
-    count += await QuoteTicket.destroy({
+    const expiredQuoteBatch = await QuoteTicket.findAll({
+      attributes: ["id"],
+      limit: EXPIRED_QUOTE_DELETE_BATCH_SIZE,
+      order: [["expiresAt", "ASC"]],
+      transaction,
       where: {
         expiresAt: {
           [Op.lt]: ninetyDaysAgo
@@ -38,7 +69,25 @@ export class BaseRampService {
       }
     });
 
-    return count;
+    const expiredQuoteIds = expiredQuoteBatch.map(quote => quote.id);
+    if (expiredQuoteIds.length === 0) {
+      return expiredPendingCount;
+    }
+
+    const deletedExpiredCount = await QuoteTicket.destroy({
+      transaction,
+      where: {
+        expiresAt: {
+          [Op.lt]: ninetyDaysAgo
+        },
+        id: {
+          [Op.in]: expiredQuoteIds
+        },
+        status: "expired"
+      }
+    });
+
+    return expiredPendingCount + deletedExpiredCount;
   }
 
   /**

apps/api/src/database/migrations/030-add-expired-quote-cleanup-index.ts

@@ -0,0 +1,15 @@
+import { QueryInterface } from "sequelize";
+
+export async function up(queryInterface: QueryInterface): Promise<void> {
+  await queryInterface.sequelize.query(`
+    CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_quote_tickets_expired_expires_at
+    ON quote_tickets (expires_at)
+    WHERE status = 'expired';
+  `);
+}
+
+export async function down(queryInterface: QueryInterface): Promise<void> {
+  await queryInterface.sequelize.query(`
+    DROP INDEX CONCURRENTLY IF EXISTS idx_quote_tickets_expired_expires_at;
+  `);
+}

apps/api/src/models/quoteTicket.model.ts

@@ -205,6 +205,17 @@ QuoteTicket.init(
       {
         fields: ["flow_variant"],
         name: "idx_quote_tickets_flow_variant"
+      },
+      {
+        fields: ["expires_at"],
+        name: "idx_quote_tickets_expired_expires_at",
+        where: {
+          status: "expired"
+        }
+      },
+      {
+        fields: [{ name: "created_at", order: "DESC" }],
+        name: "idx_quote_tickets_created_at"
       }
     ],
     modelName: "QuoteTicket",