From eeacfbde0d9cb9fb069680e9e0b6b1ffb3a03ca0 Mon Sep 17 00:00:00 2001 From: Marcel Ebert Date: Tue, 2 Jun 2026 10:02:05 +0200 Subject: [PATCH 1/2] Delete expired quotes in batches --- .../api/services/ramp/base.service.test.ts | 95 +++++++++++++++++++ .../api/src/api/services/ramp/base.service.ts | 59 +++++++++++- .../030-add-expired-quote-cleanup-index.ts | 15 +++ apps/api/src/models/quoteTicket.model.ts | 7 ++ 4 files changed, 171 insertions(+), 5 deletions(-) create mode 100644 apps/api/src/api/services/ramp/base.service.test.ts create mode 100644 apps/api/src/database/migrations/030-add-expired-quote-cleanup-index.ts diff --git a/apps/api/src/api/services/ramp/base.service.test.ts b/apps/api/src/api/services/ramp/base.service.test.ts new file mode 100644 index 000000000..27725359d --- /dev/null +++ b/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) => 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); + }); +}); diff --git a/apps/api/src/api/services/ramp/base.service.ts b/apps/api/src/api/services/ramp/base.service.ts index e76bda1ae..81e8299e7 100644 --- a/apps/api/src/api/services/ramp/base.service.ts +++ b/apps/api/src/api/services/ramp/base.service.ts @@ -1,5 +1,5 @@ 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"; @@ -7,17 +7,45 @@ 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 { + 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 { // 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; } /** diff --git a/apps/api/src/database/migrations/030-add-expired-quote-cleanup-index.ts b/apps/api/src/database/migrations/030-add-expired-quote-cleanup-index.ts new file mode 100644 index 000000000..cef566235 --- /dev/null +++ b/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 { + 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 { + await queryInterface.sequelize.query(` + DROP INDEX CONCURRENTLY IF EXISTS idx_quote_tickets_expired_expires_at; + `); +} diff --git a/apps/api/src/models/quoteTicket.model.ts b/apps/api/src/models/quoteTicket.model.ts index 6b2b90f4e..2d2179c77 100644 --- a/apps/api/src/models/quoteTicket.model.ts +++ b/apps/api/src/models/quoteTicket.model.ts @@ -205,6 +205,13 @@ QuoteTicket.init( { fields: ["flow_variant"], name: "idx_quote_tickets_flow_variant" + }, + { + fields: ["expires_at"], + name: "idx_quote_tickets_expired_expires_at", + where: { + status: "expired" + } } ], modelName: "QuoteTicket", From 917f81a1ac56150b0bb3bb862ba0d615094f67fb Mon Sep 17 00:00:00 2001 From: Marcel Ebert Date: Tue, 2 Jun 2026 10:02:49 +0200 Subject: [PATCH 2/2] Improve sorting quotes by created_at descending --- apps/api/src/models/quoteTicket.model.ts | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/apps/api/src/models/quoteTicket.model.ts b/apps/api/src/models/quoteTicket.model.ts index 2d2179c77..df07a5ce1 100644 --- a/apps/api/src/models/quoteTicket.model.ts +++ b/apps/api/src/models/quoteTicket.model.ts @@ -212,6 +212,10 @@ QuoteTicket.init( where: { status: "expired" } + }, + { + fields: [{ name: "created_at", order: "DESC" }], + name: "idx_quote_tickets_created_at" } ], modelName: "QuoteTicket",