diff --git a/.helm/console-api-prod-mainnet-values.yaml b/.helm/console-api-prod-mainnet-values.yaml index cb3e19350a..356bb4335b 100644 --- a/.helm/console-api-prod-mainnet-values.yaml +++ b/.helm/console-api-prod-mainnet-values.yaml @@ -19,8 +19,11 @@ jobs: - ./dist/instrumentation.js - ./dist/console.js - refill-wallets + # Start minutes are offset off every other sweep's, mint-act's */10 included, so no two jobs start on the same minute. - name: cleanup-stale-deployments - schedule: "30 10,22 * * *" # 10:30 AM and 10:30 PM every day + schedule: "4,14,24,34,44,54 * * * *" # every 10 minutes + concurrencyPolicy: Forbid + activeDeadlineSeconds: 540 command: - node - --require diff --git a/.helm/console-api-staging-sandbox-values.yaml b/.helm/console-api-staging-sandbox-values.yaml index feab8a17e4..d3facd68a8 100644 --- a/.helm/console-api-staging-sandbox-values.yaml +++ b/.helm/console-api-staging-sandbox-values.yaml @@ -17,8 +17,11 @@ jobs: - ./dist/instrumentation.js - ./dist/console.js - refill-wallets + # Start minutes are offset off every other sweep's so no two jobs start on the same minute. - name: cleanup-stale-deployments - schedule: "30 10 * * 2" # 10:30 AM every Tuesday + schedule: "4,14,24,34,44,54 * * * *" # every 10 minutes + concurrencyPolicy: Forbid + activeDeadlineSeconds: 540 command: - node - --require diff --git a/apps/api/src/app/console.ts b/apps/api/src/app/console.ts index fb163e5e89..5052d5c7e0 100644 --- a/apps/api/src/app/console.ts +++ b/apps/api/src/app/console.ts @@ -56,10 +56,11 @@ program program .command("cleanup-stale-deployments") .description("Close deployments without leases created at least 10min ago") - .option("-c, --concurrency ", "How many wallets is processed concurrently", value => z.number({ coerce: true }).optional().default(10).parse(value)) + .option("-c, --concurrency ", "How many owners' closes run concurrently", value => z.number({ coerce: true }).optional().default(10).parse(value)) + .option("-d, --dry-run", "Log which deployments would be closed without broadcasting", false) .action(async (options, command) => { await executeCliHandler(command.name(), async () => { - await container.resolve(TopUpDeploymentsController).cleanUpStaleDeployment(options); + return container.resolve(TopUpDeploymentsController).cleanUpStaleDeployment(options); }); }); diff --git a/apps/api/src/billing/repositories/user-wallet/user-wallet.repository.integration.ts b/apps/api/src/billing/repositories/user-wallet/user-wallet.repository.integration.ts index f1e2e5ddd8..9a579e65f7 100644 --- a/apps/api/src/billing/repositories/user-wallet/user-wallet.repository.integration.ts +++ b/apps/api/src/billing/repositories/user-wallet/user-wallet.repository.integration.ts @@ -314,6 +314,70 @@ describe(UserWalletRepository.name, () => { }); }); + describe("findManagedIteratively", () => { + it("yields every wallet that holds an address, in id order across batches", async () => { + const { createWallet, collectYielded } = await setupManagedWallets(); + const first = await createWallet(); + const second = await createWallet(); + const third = await createWallet(); + + const yielded = await collectYielded([first, second, third], 2); + + expect(yielded).toEqual([ + { id: first.id, address: first.address }, + { id: second.id, address: second.address }, + { id: third.id, address: third.address } + ]); + }); + + it("leaves out a wallet that has no address yet, since it owns nothing on chain", async () => { + const { createWallet, collectYielded } = await setupManagedWallets(); + const addressless = await createWallet({ withAddress: false }); + const managed = await createWallet(); + + const yielded = await collectYielded([addressless, managed], 10); + + expect(yielded).toEqual([{ id: managed.id, address: managed.address }]); + }); + + it("caps every batch at the batch size", async () => { + const { userWalletRepository, createWallet } = await setupManagedWallets(); + await createWallet(); + await createWallet(); + await createWallet(); + const sizes: number[] = []; + + for await (const batch of userWalletRepository.findManagedIteratively({ batchSize: 2 })) { + sizes.push(batch.length); + } + + expect(Math.max(...sizes)).toBeLessThanOrEqual(2); + }); + }); + + async function setupManagedWallets() { + const userRepository = container.resolve(UserRepository); + const userWalletRepository = container.resolve(UserWalletRepository); + + async function createWallet(input: { withAddress?: boolean } = {}) { + const user = await userRepository.create({ userId: faker.string.uuid() }); + return await userWalletRepository.create({ userId: user.id, address: input.withAddress === false ? undefined : createAkashAddress() }); + } + + async function collectYielded(wallets: { id: number }[], batchSize: number) { + const ids = new Set(wallets.map(wallet => wallet.id)); + const yielded = []; + + for await (const batch of userWalletRepository.findManagedIteratively({ batchSize })) { + yielded.push(...batch.filter(wallet => ids.has(wallet.id))); + } + + return yielded; + } + + return { userWalletRepository, createWallet, collectYielded }; + } + async function setupDomain() { const userRepository = container.resolve(UserRepository); const userWalletRepository = container.resolve(UserWalletRepository); diff --git a/apps/api/src/billing/repositories/user-wallet/user-wallet.repository.ts b/apps/api/src/billing/repositories/user-wallet/user-wallet.repository.ts index 879954c3b8..28fafd813f 100644 --- a/apps/api/src/billing/repositories/user-wallet/user-wallet.repository.ts +++ b/apps/api/src/billing/repositories/user-wallet/user-wallet.repository.ts @@ -1,6 +1,6 @@ import { Trace } from "@akashnetwork/instrumentation"; import subDays from "date-fns/subDays"; -import { and, count, eq, gt, inArray, isNotNull, isNull, lte, ne, notExists, or, sql } from "drizzle-orm"; +import { and, asc, count, eq, gt, inArray, isNotNull, isNull, lte, ne, notExists, or, sql } from "drizzle-orm"; import { singleton } from "tsyringe"; import { StripeTransactions } from "@src/billing/model-schemas"; @@ -30,6 +30,12 @@ export function isWalletInitialized(wallet: UserWalletOutput): wallet is WalletI return !!wallet.address; } +/** All a sweep needs to close a deployment on a wallet's behalf: the address it owns on chain and the index the signer derives it from. */ +export interface ManagedWalletRef { + id: UserWalletOutput["id"]; + address: string; +} + export interface TrialWindow { trialEndsAt: Date | null; trialDurationDays: number | null; @@ -243,6 +249,27 @@ export class UserWalletRepository extends BaseRepository { + let cursor: number | undefined; + + while (true) { + const batch = await this.cursor + .select({ id: this.table.id, address: this.table.address }) + .from(this.table) + .where(this.whereAccessibleBy(and(isNotNull(this.table.address), ...(cursor === undefined ? [] : [gt(this.table.id, cursor)])))) + .orderBy(asc(this.table.id)) + .limit(batchSize); + + if (!batch.length) return; + + yield batch as ManagedWalletRef[]; + + if (batch.length < batchSize) return; + cursor = batch[batch.length - 1].id; + } + } + async findFirst() { return this.findOneBy(); } diff --git a/apps/api/src/deployment/controllers/deployment/top-up-deployments.controller.spec.ts b/apps/api/src/deployment/controllers/deployment/top-up-deployments.controller.spec.ts index 95270ad003..6c9545dae7 100644 --- a/apps/api/src/deployment/controllers/deployment/top-up-deployments.controller.spec.ts +++ b/apps/api/src/deployment/controllers/deployment/top-up-deployments.controller.spec.ts @@ -69,7 +69,7 @@ describe(TopUpDeploymentsController.name, () => { describe("cleanUpStaleDeployment", () => { it("should call the service to clean up stale deployments", async () => { const { controller, staleDeploymentsCleanerService } = setup(); - const options = { concurrency: 5 }; + const options = { concurrency: 5, dryRun: false }; await controller.cleanUpStaleDeployment(options); diff --git a/apps/api/src/deployment/controllers/deployment/top-up-deployments.controller.ts b/apps/api/src/deployment/controllers/deployment/top-up-deployments.controller.ts index 00297f0b7b..e2c20b28bc 100644 --- a/apps/api/src/deployment/controllers/deployment/top-up-deployments.controller.ts +++ b/apps/api/src/deployment/controllers/deployment/top-up-deployments.controller.ts @@ -30,7 +30,7 @@ export class TopUpDeploymentsController { } async cleanUpStaleDeployment(options: CleanUpStaleDeploymentsParams) { - await this.staleDeploymentsCleanerService.cleanup(options); + return await this.staleDeploymentsCleanerService.cleanup(options); } async notifyExpiringDeployments(options: DryRunOptions) { diff --git a/apps/api/src/deployment/repositories/deployment/deployment.repository.integration.ts b/apps/api/src/deployment/repositories/deployment/deployment.repository.integration.ts index ee320dcedc..63c76de460 100644 --- a/apps/api/src/deployment/repositories/deployment/deployment.repository.integration.ts +++ b/apps/api/src/deployment/repositories/deployment/deployment.repository.integration.ts @@ -186,7 +186,7 @@ describe(DeploymentRepository.name, () => { const { repository, owner, base } = setup(); const deployment = await seedOpenDeployment(owner, { createdHeight: base - 1 }); - const found = await repository.findStaleDeployments({ owner, staleBeforeHeight: base }); + const found = await repository.findStaleDeployments({ owners: [owner], staleBeforeHeight: base }); expect(found.map(stale => String(stale.dseq))).toEqual([deployment.dseq]); }); @@ -196,7 +196,7 @@ describe(DeploymentRepository.name, () => { const deployment = await seedOpenDeployment(owner, { createdHeight: base - 1_000 }); await seedLease(deployment, { closedHeight: base - 1 }); - const found = await repository.findStaleDeployments({ owner, staleBeforeHeight: base }); + const found = await repository.findStaleDeployments({ owners: [owner], staleBeforeHeight: base }); expect(found.map(stale => String(stale.dseq))).toEqual([deployment.dseq]); }); @@ -207,7 +207,7 @@ describe(DeploymentRepository.name, () => { await seedLease(deployment, { gseq: 1, closedHeight: base - 500 }); await seedLease(deployment, { gseq: 2, closedHeight: base }); - const found = await repository.findStaleDeployments({ owner, staleBeforeHeight: base }); + const found = await repository.findStaleDeployments({ owners: [owner], staleBeforeHeight: base }); expect(found).toEqual([]); }); @@ -218,7 +218,7 @@ describe(DeploymentRepository.name, () => { await seedLease(deployment, { gseq: 1, closedHeight: base - 500 }); await seedLease(deployment, { gseq: 2 }); - const found = await repository.findStaleDeployments({ owner, staleBeforeHeight: base }); + const found = await repository.findStaleDeployments({ owners: [owner], staleBeforeHeight: base }); expect(found).toEqual([]); }); @@ -227,7 +227,7 @@ describe(DeploymentRepository.name, () => { const { repository, owner, base } = setup(); await seedOpenDeployment(owner, { createdHeight: base }); - const found = await repository.findStaleDeployments({ owner, staleBeforeHeight: base }); + const found = await repository.findStaleDeployments({ owners: [owner], staleBeforeHeight: base }); expect(found).toEqual([]); }); @@ -236,7 +236,7 @@ describe(DeploymentRepository.name, () => { const { repository, owner, base } = setup(); await seedOpenDeployment(owner, { createdHeight: base - 1_000, closedHeight: base - 500 }); - const found = await repository.findStaleDeployments({ owner, staleBeforeHeight: base }); + const found = await repository.findStaleDeployments({ owners: [owner], staleBeforeHeight: base }); expect(found).toEqual([]); }); @@ -245,10 +245,40 @@ describe(DeploymentRepository.name, () => { const { repository, owner, base } = setup(); await seedOpenDeployment(createAkashAddress(), { createdHeight: base - 1 }); - const found = await repository.findStaleDeployments({ owner, staleBeforeHeight: base }); + const found = await repository.findStaleDeployments({ owners: [owner], staleBeforeHeight: base }); expect(found).toEqual([]); }); + + it("returns the orphans of every owner it is given, each labelled with its owner", async () => { + const { repository, owner, base } = setup(); + const other = createAkashAddress(); + const mine = await seedOpenDeployment(owner, { createdHeight: base - 1 }); + const theirs = await seedOpenDeployment(other, { createdHeight: base - 1 }); + + const found = await repository.findStaleDeployments({ owners: [owner, other], staleBeforeHeight: base }); + + expect(found).toHaveLength(2); + expect(found).toContainEqual({ owner, dseq: mine.dseq }); + expect(found).toContainEqual({ owner: other, dseq: theirs.dseq }); + }); + + it("returns a deployment once however many of its leases closed before the cutoff", async () => { + const { repository, owner, base } = setup(); + const deployment = await seedOpenDeployment(owner, { createdHeight: base - 1_000 }); + await seedLease(deployment, { gseq: 1, closedHeight: base - 500 }); + await seedLease(deployment, { gseq: 2, closedHeight: base - 400 }); + + const found = await repository.findStaleDeployments({ owners: [owner], staleBeforeHeight: base }); + + expect(found).toEqual([{ owner, dseq: deployment.dseq }]); + }); + + it("queries nothing when given no owners", async () => { + const { repository, base } = setup(); + + await expect(repository.findStaleDeployments({ owners: [], staleBeforeHeight: base })).resolves.toEqual([]); + }); }); describe("countActiveByOwner", () => { diff --git a/apps/api/src/deployment/repositories/deployment/deployment.repository.ts b/apps/api/src/deployment/repositories/deployment/deployment.repository.ts index 6247d23ece..0aa3e8a365 100644 --- a/apps/api/src/deployment/repositories/deployment/deployment.repository.ts +++ b/apps/api/src/deployment/repositories/deployment/deployment.repository.ts @@ -8,7 +8,7 @@ import { CHAIN_DB } from "@src/chain"; export interface StaleDeploymentsOptions { staleBeforeHeight: number; - owner: string; + owners: string[]; } export interface ProviderCleanupOptions { @@ -27,7 +27,7 @@ export interface DeploymentActivityWindow { endDate: string; } -/** Heights reach the query through a literal, so anything but a plain integer is refused before it can be interpolated. */ +/** A missing height would bind as NULL and match nothing, so anything but a plain integer is refused before the query runs. */ function asHeight(value: number): number { if (!Number.isSafeInteger(value)) { throw new TypeError(`Expected a block height, received ${value}`); @@ -52,7 +52,11 @@ export interface DatabaseDeploymentListParams { } export interface StaleDeploymentsOutput { - dseq: number; + dseq: string; +} + +export interface StaleDeployment extends StaleDeploymentsOutput { + owner: string; } export interface DeploymentKey { @@ -117,32 +121,29 @@ export class DeploymentRepository { }); } - async findStaleDeployments(options: StaleDeploymentsOptions): Promise { - const deployments = await Deployment.findAll({ - attributes: ["dseq"], - include: [ - { - model: Lease, - attributes: [], - required: false - } - ], - where: { - owner: options.owner, - createdHeight: { - [Op.lt]: options.staleBeforeHeight - }, - closedHeight: null - }, - group: ["deployment.dseq"], - having: literal( - `COUNT("leases"."deploymentId") FILTER (WHERE "leases"."closedHeight" IS NULL) = 0 ` + - `AND COALESCE(MAX("leases"."closedHeight"), 0) < ${asHeight(options.staleBeforeHeight)}` - ), - raw: true - }); + /** Owners reach the query as one array rather than one call each, so a sweep costs a query per batch instead of a query per wallet. */ + async findStaleDeployments(options: StaleDeploymentsOptions): Promise { + if (options.owners.length === 0) return []; - return deployments ? (deployments as unknown as StaleDeploymentsOutput[]) : []; + const staleBeforeHeight = asHeight(options.staleBeforeHeight); + + return await this.#chainDb.query( + `/* deployment:staleByOwners */ + SELECT d."owner", d."dseq" + FROM deployment d + JOIN unnest($1::text[]) AS o(owner) ON d."owner" = o.owner + WHERE d."closedHeight" IS NULL + AND d."createdHeight" < $2 + AND NOT EXISTS ( + SELECT 1 FROM lease live + WHERE live."deploymentId" = d."id" AND live."closedHeight" IS NULL + ) + AND COALESCE((SELECT MAX(last."closedHeight") FROM lease last WHERE last."deploymentId" = d."id"), 0) < $2`, + { + bind: [options.owners, staleBeforeHeight], + type: QueryTypes.SELECT + } + ); } async findDeploymentsForProvider(options: ProviderCleanupOptions): Promise { diff --git a/apps/api/src/deployment/services/stale-managed-deployments-cleaner/stale-managed-deployments-cleaner.service.spec.ts b/apps/api/src/deployment/services/stale-managed-deployments-cleaner/stale-managed-deployments-cleaner.service.spec.ts index e742a7d362..7c9578baac 100644 --- a/apps/api/src/deployment/services/stale-managed-deployments-cleaner/stale-managed-deployments-cleaner.service.spec.ts +++ b/apps/api/src/deployment/services/stale-managed-deployments-cleaner/stale-managed-deployments-cleaner.service.spec.ts @@ -5,25 +5,27 @@ import { describe, expect, it, vi } from "vitest"; import { mock } from "vitest-mock-extended"; import type { BillingConfig } from "@src/billing/providers"; -import type { UserWalletRepository } from "@src/billing/repositories"; +import type { ManagedWalletRef, UserWalletRepository } from "@src/billing/repositories"; import type { ManagedUserWalletService, RpcMessageService } from "@src/billing/services"; import type { BillingConfigService } from "@src/billing/services/billing-config/billing-config.service"; import { ChainErrorService } from "@src/billing/services/chain-error/chain-error.service"; import type { ManagedSignerService } from "@src/billing/services/managed-signer/managed-signer.service"; import type { TxManagerService } from "@src/billing/services/tx-manager/tx-manager.service"; import type { BlockRepository } from "@src/chain/repositories/block.repository"; +import type { BlockHttpService } from "@src/chain/services/block-http/block-http.service"; import type { CreateLogger } from "@src/core/providers/logging.provider"; import { ErrorService } from "@src/core/services/error/error.service"; -import type { DeploymentRepository } from "@src/deployment/repositories/deployment/deployment.repository"; +import type { DeploymentRepository, StaleDeployment } from "@src/deployment/repositories/deployment/deployment.repository"; import { StaleManagedDeploymentsCleanerService } from "./stale-managed-deployments-cleaner.service"; import { createUserWallet } from "@test/seeders/user-wallet.seeder"; const UNSETTLEABLE_PANIC = "Query failed with (6): rpc error: code = Unknown desc = recovered: negative decimal coin amount: -2.000000000000000000"; +const OWNER = "akash1test"; const UNSETTLEABLE_LOG = { event: "DEPLOYMENT_CLEAN_UP_UNSETTLEABLE", reason: "Deployment escrow cannot be settled yet; chain rejects close until it settles", - owner: "akash1test" + owner: OWNER }; describe(StaleManagedDeploymentsCleanerService.name, () => { @@ -42,7 +44,7 @@ describe(StaleManagedDeploymentsCleanerService.name, () => { await service.cleanUpForWallet(wallet, 0); - expect(deploymentRepository.findStaleDeployments).toHaveBeenCalledWith({ owner: wallet.address, staleBeforeHeight: 1_000_000 }); + expect(deploymentRepository.findStaleDeployments).toHaveBeenCalledWith({ owners: [wallet.address], staleBeforeHeight: 1_000_000 }); }); it("reads the chain height itself when called for a single wallet", async () => { @@ -63,52 +65,62 @@ describe(StaleManagedDeploymentsCleanerService.name, () => { it("closes all stale deployments in a single derived tx", async () => { const closeMsg = { typeUrl: "/close", value: {} }; - const { service, managedSignerService, rpcMessageService, wallet } = setup({ staleDeployments: [{ dseq: 1 }, { dseq: 2 }] }); + const { service, managedSignerService, rpcMessageService, wallet } = setup({ staleDeployments: ["1", "2"] }); rpcMessageService.getCloseDeploymentMsg.mockReturnValue(closeMsg as never); await service.cleanUpForWallet(wallet, 0); expect(managedSignerService.executeDerivedTx).toHaveBeenCalledWith(wallet.id, [closeMsg, closeMsg]); }); + + it("splits a wallet's orphans across transactions so one close batch cannot outgrow a block", async () => { + const dseqs = Array.from({ length: 45 }, (_, index) => String(index + 1)); + const { service, managedSignerService, wallet } = setup({ staleDeployments: dseqs }); + + await service.cleanUpForWallet(wallet, 0); + + expect(managedSignerService.executeDerivedTx).toHaveBeenCalledTimes(3); + expect(managedSignerService.executeDerivedTx.mock.calls.map(([, messages]) => messages.length)).toEqual([20, 20, 5]); + }); }); describe("when a deployment is already closed on chain", () => { it("drops the closed deployment and closes the rest in a second broadcast", async () => { const executeDerivedTx = vi.fn().mockRejectedValueOnce(buildDeploymentClosedAppError(1)).mockResolvedValueOnce(buildOkTx()); - const { service, logger, wallet } = setup({ staleDeployments: [{ dseq: 1 }, { dseq: 2 }, { dseq: 3 }], executeDerivedTx }); + const { service, logger, wallet } = setup({ staleDeployments: ["1", "2", "3"], executeDerivedTx }); await service.cleanUpForWallet(wallet, 0); expect(executeDerivedTx).toHaveBeenCalledTimes(2); expect(executeDerivedTx).toHaveBeenLastCalledWith(wallet.id, [ - expect.objectContaining({ value: expect.objectContaining({ dseq: 1 }) }), - expect.objectContaining({ value: expect.objectContaining({ dseq: 3 }) }) + expect.objectContaining({ value: expect.objectContaining({ dseq: "1" }) }), + expect.objectContaining({ value: expect.objectContaining({ dseq: "3" }) }) ]); - expect(logger.info).toHaveBeenCalledWith({ event: "DEPLOYMENT_CLEAN_UP_ALREADY_CLOSED", owner: wallet.address, dseq: 2 }); - expect(logger.info).toHaveBeenCalledWith({ event: "DEPLOYMENT_CLEAN_UP_SUCCESS", owner: wallet.address, alreadyClosedCount: 1 }); + expect(logger.info).toHaveBeenCalledWith({ event: "DEPLOYMENT_CLEAN_UP_ALREADY_CLOSED", owner: OWNER, dseq: "2" }); + expect(logger.info).toHaveBeenCalledWith({ event: "DEPLOYMENT_CLEAN_UP_SUCCESS", owner: OWNER, alreadyClosedCount: 1 }); }); it("resolves quietly when the wallet's only orphan is already closed and the error carries no index", async () => { const executeDerivedTx = vi.fn().mockRejectedValueOnce(buildDeploymentClosedAppError()); - const { service, logger, errorLogger, wallet } = setup({ staleDeployments: [{ dseq: 7 }], executeDerivedTx }); + const { service, logger, errorLogger, wallet } = setup({ staleDeployments: ["7"], executeDerivedTx }); await service.cleanUpForWallet(wallet, 0); expect(executeDerivedTx).toHaveBeenCalledTimes(1); - expect(logger.info).toHaveBeenCalledWith({ event: "DEPLOYMENT_CLEAN_UP_ALREADY_CLOSED", owner: wallet.address, dseq: 7 }); - expect(logger.info).toHaveBeenCalledWith({ event: "DEPLOYMENT_CLEAN_UP_SUCCESS", owner: wallet.address, alreadyClosedCount: 1 }); + expect(logger.info).toHaveBeenCalledWith({ event: "DEPLOYMENT_CLEAN_UP_ALREADY_CLOSED", owner: OWNER, dseq: "7" }); + expect(logger.info).toHaveBeenCalledWith({ event: "DEPLOYMENT_CLEAN_UP_SUCCESS", owner: OWNER, alreadyClosedCount: 1 }); expect(logger.error).not.toHaveBeenCalled(); expect(errorLogger.error).not.toHaveBeenCalled(); }); it("reports success without an error when the whole batch is already closed", async () => { const executeDerivedTx = vi.fn().mockRejectedValue(buildDeploymentClosedAppError(0)); - const { service, logger, errorLogger, wallet } = setup({ staleDeployments: [{ dseq: 1 }, { dseq: 2 }], executeDerivedTx }); + const { service, logger, errorLogger, wallet } = setup({ staleDeployments: ["1", "2"], executeDerivedTx }); await service.cleanUpForWallet(wallet, 0); expect(executeDerivedTx).toHaveBeenCalledTimes(2); - expect(logger.info).toHaveBeenCalledWith({ event: "DEPLOYMENT_CLEAN_UP_SUCCESS", owner: wallet.address, alreadyClosedCount: 2 }); + expect(logger.info).toHaveBeenCalledWith({ event: "DEPLOYMENT_CLEAN_UP_SUCCESS", owner: OWNER, alreadyClosedCount: 2 }); expect(logger.error).not.toHaveBeenCalled(); expect(errorLogger.error).not.toHaveBeenCalled(); }); @@ -116,12 +128,13 @@ describe(StaleManagedDeploymentsCleanerService.name, () => { it("rethrows into the wallet error handler when the reported index falls outside the batch", async () => { const error = buildDeploymentClosedAppError(7); const { service, managedSignerService, logger, errorLogger } = setup({ - staleDeployments: [{ dseq: 1 }, { dseq: 2 }], + staleDeployments: ["1", "2"], executeDerivedTx: vi.fn().mockRejectedValue(error) }); - await expect(service.cleanup({ concurrency: 1 })).resolves.toBeUndefined(); + const result = await service.cleanup({ concurrency: 1, dryRun: false }); + expect(result.ok).toBe(true); expect(managedSignerService.executeDerivedTx).toHaveBeenCalledTimes(1); expect(logger.info).not.toHaveBeenCalledWith(expect.objectContaining({ event: "DEPLOYMENT_CLEAN_UP_ALREADY_CLOSED" })); expect(errorLogger.error).toHaveBeenCalledWith(expect.objectContaining({ event: "DEPLOYMENT_CLEAN_UP_ERROR", error })); @@ -130,14 +143,14 @@ describe(StaleManagedDeploymentsCleanerService.name, () => { it("stops after the drop limit without reporting an error when too many deployments turn out closed", async () => { const executeDerivedTx = vi.fn().mockRejectedValue(buildDeploymentClosedAppError(0)); const { service, logger, errorLogger, wallet } = setup({ - staleDeployments: [{ dseq: 1 }, { dseq: 2 }, { dseq: 3 }, { dseq: 4 }, { dseq: 5 }], + staleDeployments: ["1", "2", "3", "4", "5"], executeDerivedTx }); await service.cleanUpForWallet(wallet, 0); expect(executeDerivedTx).toHaveBeenCalledTimes(4); - expect(logger.warn).toHaveBeenCalledWith({ event: "DEPLOYMENT_CLEAN_UP_DROP_LIMIT", owner: wallet.address, remainingCount: 2 }); + expect(logger.warn).toHaveBeenCalledWith({ event: "DEPLOYMENT_CLEAN_UP_DROP_LIMIT", owner: OWNER, remainingCount: 2 }); expect(logger.info).not.toHaveBeenCalledWith(expect.objectContaining({ event: "DEPLOYMENT_CLEAN_UP_SUCCESS" })); expect(logger.error).not.toHaveBeenCalled(); expect(errorLogger.error).not.toHaveBeenCalled(); @@ -151,7 +164,7 @@ describe(StaleManagedDeploymentsCleanerService.name, () => { .mockRejectedValueOnce(buildDeploymentClosedAppError(0)) .mockResolvedValueOnce(buildOkTx()); const { service, logger, wallet } = setup({ - staleDeployments: [{ dseq: 1 }, { dseq: 2 }, { dseq: 3 }, { dseq: 4 }, { dseq: 5 }], + staleDeployments: ["1", "2", "3", "4", "5"], executeDerivedTx }); @@ -159,23 +172,41 @@ describe(StaleManagedDeploymentsCleanerService.name, () => { expect(executeDerivedTx).toHaveBeenCalledTimes(4); expect(executeDerivedTx).toHaveBeenLastCalledWith(wallet.id, [ - expect.objectContaining({ value: expect.objectContaining({ dseq: 4 }) }), - expect.objectContaining({ value: expect.objectContaining({ dseq: 5 }) }) + expect.objectContaining({ value: expect.objectContaining({ dseq: "4" }) }), + expect.objectContaining({ value: expect.objectContaining({ dseq: "5" }) }) ]); expect(logger.warn).not.toHaveBeenCalledWith(expect.objectContaining({ event: "DEPLOYMENT_CLEAN_UP_DROP_LIMIT" })); - expect(logger.info).toHaveBeenCalledWith({ event: "DEPLOYMENT_CLEAN_UP_SUCCESS", owner: wallet.address, alreadyClosedCount: 3 }); + expect(logger.info).toHaveBeenCalledWith({ event: "DEPLOYMENT_CLEAN_UP_SUCCESS", owner: OWNER, alreadyClosedCount: 3 }); + }); + + it("spends one drop budget across all of an owner's transactions rather than one per transaction", async () => { + const executeDerivedTx = vi + .fn() + .mockRejectedValueOnce(buildDeploymentClosedAppError(0)) + .mockRejectedValueOnce(buildDeploymentClosedAppError(0)) + .mockResolvedValueOnce(buildOkTx()) + .mockRejectedValueOnce(buildDeploymentClosedAppError(0)) + .mockRejectedValueOnce(buildDeploymentClosedAppError(0)); + const dseqs = Array.from({ length: 25 }, (_, index) => String(index + 1)); + const { service, logger, wallet } = setup({ staleDeployments: dseqs, executeDerivedTx }); + + await service.cleanUpForWallet(wallet, 0); + + expect(executeDerivedTx).toHaveBeenCalledTimes(5); + expect(logger.warn).toHaveBeenCalledWith(expect.objectContaining({ event: "DEPLOYMENT_CLEAN_UP_DROP_LIMIT", owner: OWNER })); + expect(logger.info).not.toHaveBeenCalledWith(expect.objectContaining({ event: "DEPLOYMENT_CLEAN_UP_SUCCESS" })); }); it("treats a landed tx that reverted on a closed deployment as a failure and drops it", async () => { const revertedTx = mock({ code: 8, hash: "tx-hash", rawLog: "failed to execute message; message index: 0: Deployment closed" }); const executeDerivedTx = vi.fn().mockResolvedValueOnce(revertedTx).mockResolvedValueOnce(buildOkTx()); - const { service, logger, wallet } = setup({ staleDeployments: [{ dseq: 1 }, { dseq: 2 }], executeDerivedTx }); + const { service, logger, wallet } = setup({ staleDeployments: ["1", "2"], executeDerivedTx }); await service.cleanUpForWallet(wallet, 0); expect(executeDerivedTx).toHaveBeenCalledTimes(2); - expect(logger.info).toHaveBeenCalledWith({ event: "DEPLOYMENT_CLEAN_UP_ALREADY_CLOSED", owner: wallet.address, dseq: 1 }); - expect(logger.info).toHaveBeenCalledWith({ event: "DEPLOYMENT_CLEAN_UP_SUCCESS", owner: wallet.address, alreadyClosedCount: 1 }); + expect(logger.info).toHaveBeenCalledWith({ event: "DEPLOYMENT_CLEAN_UP_ALREADY_CLOSED", owner: OWNER, dseq: "1" }); + expect(logger.info).toHaveBeenCalledWith({ event: "DEPLOYMENT_CLEAN_UP_SUCCESS", owner: OWNER, alreadyClosedCount: 1 }); }); it("composes the fee refill with the closed-deployment drop", async () => { @@ -184,35 +215,141 @@ describe(StaleManagedDeploymentsCleanerService.name, () => { .mockRejectedValueOnce(new Error("not allowed to pay fees")) .mockRejectedValueOnce(buildDeploymentClosedAppError(0)) .mockResolvedValueOnce(buildOkTx()); - const { service, managedUserWalletService, logger, wallet } = setup({ staleDeployments: [{ dseq: 1 }, { dseq: 2 }], executeDerivedTx }); + const { service, managedUserWalletService, logger, wallet } = setup({ staleDeployments: ["1", "2"], executeDerivedTx }); await service.cleanUpForWallet(wallet, 0); expect(managedUserWalletService.authorizeSpending).toHaveBeenCalledTimes(1); expect(executeDerivedTx).toHaveBeenCalledTimes(3); - expect(logger.info).toHaveBeenCalledWith({ event: "DEPLOYMENT_CLEAN_UP_SUCCESS", owner: wallet.address, alreadyClosedCount: 1 }); + expect(logger.info).toHaveBeenCalledWith({ event: "DEPLOYMENT_CLEAN_UP_SUCCESS", owner: OWNER, alreadyClosedCount: 1 }); }); it("logs the unsettleable event when the re-broadcast after a drop hits the escrow underflow", async () => { const executeDerivedTx = vi.fn().mockRejectedValueOnce(buildDeploymentClosedAppError(0)).mockRejectedValueOnce(buildUnsettleableAppError()); - const { service, logger, wallet } = setup({ staleDeployments: [{ dseq: 1 }, { dseq: 2 }], executeDerivedTx }); + const { service, logger, wallet } = setup({ staleDeployments: ["1", "2"], executeDerivedTx }); await service.cleanUpForWallet(wallet, 0); expect(executeDerivedTx).toHaveBeenCalledTimes(2); - expect(logger.info).toHaveBeenCalledWith({ event: "DEPLOYMENT_CLEAN_UP_ALREADY_CLOSED", owner: wallet.address, dseq: 1 }); + expect(logger.info).toHaveBeenCalledWith({ event: "DEPLOYMENT_CLEAN_UP_ALREADY_CLOSED", owner: OWNER, dseq: "1" }); expect(logger.error).toHaveBeenCalledWith(UNSETTLEABLE_LOG); }); }); describe("cleanup", () => { + it("asks the chain once per batch of wallets instead of once per wallet", async () => { + const { service, deploymentRepository, blockRepository } = setup({ walletBatches: [5, 5, 3] }); + + await service.cleanup({ concurrency: 3, dryRun: false }); + + expect(deploymentRepository.findStaleDeployments).toHaveBeenCalledTimes(3); + expect(blockRepository.getLatestProcessedHeight).toHaveBeenCalledTimes(1); + }); + + it("screens every wallet of a batch against the same cutoff in one call", async () => { + const { service, deploymentRepository } = setup({ currentHeight: 1_000_000, walletBatches: [3] }); + + await service.cleanup({ concurrency: 3, dryRun: false }); + + const [{ owners, staleBeforeHeight }] = deploymentRepository.findStaleDeployments.mock.calls[0]; + expect(owners).toHaveLength(3); + expect(staleBeforeHeight).toBeLessThan(1_000_000); + }); + + it("closes each owner's orphans in that owner's own transaction", async () => { + const { service, managedSignerService } = setup({ + orphans: [ + { owner: "akash1a", dseq: "1" }, + { owner: "akash1a", dseq: "2" }, + { owner: "akash1b", dseq: "3" } + ], + walletIdsByAddress: { akash1a: 11, akash1b: 22 } + }); + + await service.cleanup({ concurrency: 2, dryRun: false }); + + expect(managedSignerService.executeDerivedTx).toHaveBeenCalledTimes(2); + expect(managedSignerService.executeDerivedTx).toHaveBeenCalledWith(11, [ + expect.objectContaining({ value: expect.objectContaining({ dseq: "1" }) }), + expect.objectContaining({ value: expect.objectContaining({ dseq: "2" }) }) + ]); + expect(managedSignerService.executeDerivedTx).toHaveBeenCalledWith(22, [expect.objectContaining({ value: expect.objectContaining({ dseq: "3" }) })]); + }); + + it("broadcasts nothing for a batch that holds no orphan", async () => { + const { service, managedSignerService } = setup({ walletBatches: [4], orphans: [] }); + + await service.cleanup({ concurrency: 2, dryRun: false }); + + expect(managedSignerService.executeDerivedTx).not.toHaveBeenCalled(); + }); + + it("reports what it would close without broadcasting on a dry run", async () => { + const { service, managedSignerService, logger } = setup({ orphans: [{ owner: OWNER, dseq: "9" }] }); + + const result = await service.cleanup({ concurrency: 1, dryRun: true }); + + expect(result.ok).toBe(true); + expect(managedSignerService.executeDerivedTx).not.toHaveBeenCalled(); + expect(logger.info).toHaveBeenCalledWith({ event: "DEPLOYMENT_CLEAN_UP_WOULD_CLOSE", owner: OWNER, dseqs: ["9"] }); + }); + + it("carries on to the other owners and still succeeds when one of them fails", async () => { + const failure = new Error("some unexpected failure"); + const executeDerivedTx = vi.fn().mockRejectedValueOnce(failure).mockResolvedValue(buildOkTx()); + const { service, logger, errorLogger } = setup({ + orphans: [ + { owner: "akash1a", dseq: "1" }, + { owner: "akash1b", dseq: "2" } + ], + walletIdsByAddress: { akash1a: 11, akash1b: 22 }, + executeDerivedTx + }); + + const result = await service.cleanup({ concurrency: 1, dryRun: false }); + + expect(executeDerivedTx).toHaveBeenCalledTimes(2); + expect(result.ok).toBe(true); + expect(errorLogger.error).toHaveBeenCalledWith(expect.objectContaining({ event: "DEPLOYMENT_CLEAN_UP_ERROR", error: failure })); + expect(logger.info).toHaveBeenCalledWith(expect.objectContaining({ event: "DEPLOYMENT_CLEAN_UP_SWEEP_END", failedOwners: 1, screenFailures: 0 })); + }); + + it("succeeds without an error when every owner closes", async () => { + const { service } = setup({ orphans: [{ owner: OWNER, dseq: "1" }] }); + + const result = await service.cleanup({ concurrency: 1, dryRun: false }); + + expect(result.ok).toBe(true); + }); + + it("reports how long the sweep took", async () => { + const { service, logger } = setup(); + + await service.cleanup({ concurrency: 1, dryRun: false }); + + expect(logger.info).toHaveBeenCalledWith(expect.objectContaining({ event: "DEPLOYMENT_CLEAN_UP_SWEEP_END", durationMs: expect.any(Number) })); + }); + + it("carries on to the next batch but fails the run when screening one of them fails", async () => { + const screenFailure = new Error("chain db timed out"); + const { service, deploymentRepository, logger } = setup({ walletBatches: [2, 2] }); + deploymentRepository.findStaleDeployments.mockRejectedValueOnce(screenFailure).mockResolvedValueOnce([]); + + const result = await service.cleanup({ concurrency: 1, dryRun: false }); + + expect(deploymentRepository.findStaleDeployments).toHaveBeenCalledTimes(2); + expect(result.err).toBe(true); + expect(logger.info).toHaveBeenCalledWith(expect.objectContaining({ event: "DEPLOYMENT_CLEAN_UP_SWEEP_END", screenFailures: 1, failedOwners: 0 })); + }); + it("logs the unsettleable event and swallows the error without refilling fees or retrying", async () => { const { service, managedSignerService, managedUserWalletService, logger, errorLogger } = setup({ executeDerivedTx: vi.fn().mockRejectedValue(buildUnsettleableAppError()) }); - await expect(service.cleanup({ concurrency: 1 })).resolves.toBeUndefined(); + const result = await service.cleanup({ concurrency: 1, dryRun: false }); + expect(result.ok).toBe(true); expect(managedSignerService.executeDerivedTx).toHaveBeenCalledTimes(1); expect(managedUserWalletService.authorizeSpending).not.toHaveBeenCalled(); expect(logger.error).toHaveBeenCalledWith(UNSETTLEABLE_LOG); @@ -223,7 +360,7 @@ describe(StaleManagedDeploymentsCleanerService.name, () => { const executeDerivedTx = vi.fn().mockRejectedValueOnce(new Error("not allowed to pay fees")).mockResolvedValueOnce(buildOkTx()); const { service, managedUserWalletService, logger } = setup({ executeDerivedTx }); - await service.cleanup({ concurrency: 1 }); + await service.cleanup({ concurrency: 1, dryRun: false }); expect(managedUserWalletService.authorizeSpending).toHaveBeenCalledTimes(1); expect(executeDerivedTx).toHaveBeenCalledTimes(2); @@ -234,44 +371,35 @@ describe(StaleManagedDeploymentsCleanerService.name, () => { const executeDerivedTx = vi.fn().mockRejectedValueOnce(new Error("not allowed to pay fees")).mockRejectedValueOnce(buildUnsettleableAppError()); const { service, managedUserWalletService, logger } = setup({ executeDerivedTx }); - await expect(service.cleanup({ concurrency: 1 })).resolves.toBeUndefined(); + await service.cleanup({ concurrency: 1, dryRun: false }); expect(managedUserWalletService.authorizeSpending).toHaveBeenCalledTimes(1); expect(executeDerivedTx).toHaveBeenCalledTimes(2); expect(logger.error).toHaveBeenCalledWith(UNSETTLEABLE_LOG); }); + }); - it("reads the chain height once for the whole sweep", async () => { - const { service, blockRepository, deploymentRepository } = setup({ pages: 4, walletsPerPage: 3 }); - - await service.cleanup({ concurrency: 3 }); - - expect(blockRepository.getLatestProcessedHeight).toHaveBeenCalledTimes(1); - expect(deploymentRepository.findStaleDeployments).toHaveBeenCalledTimes(12); - }); - - it("screens every wallet against the same cutoff", async () => { - const { service, deploymentRepository } = setup({ currentHeight: 1_000_000, pages: 3, walletsPerPage: 2 }); + describe("when the indexer trails the chain", () => { + it("refuses to sweep rather than read a leased deployment as an orphan", async () => { + const { service, deploymentRepository, managedSignerService, logger } = setup({ currentHeight: 1_000_000, chainHeight: 1_000_500 }); - await service.cleanup({ concurrency: 2 }); + const result = await service.cleanup({ concurrency: 1, dryRun: false }); - const cutoffs = new Set(deploymentRepository.findStaleDeployments.mock.calls.map(([{ staleBeforeHeight }]) => staleBeforeHeight)); - expect(cutoffs.size).toBe(1); - expect([...cutoffs][0]).toBeLessThan(1_000_000); + expect(result.err).toBe(true); + expect(deploymentRepository.findStaleDeployments).not.toHaveBeenCalled(); + expect(managedSignerService.executeDerivedTx).not.toHaveBeenCalled(); + expect(logger.error).toHaveBeenCalledWith( + expect.objectContaining({ event: "DEPLOYMENT_CLEAN_UP_INDEXER_LAGGING", chainHeight: 1_000_500, indexedHeight: 1_000_000 }) + ); }); - it("rethrows unrelated errors into the wallet error handler without retrying", async () => { - const unexpectedError = new Error("some unexpected failure"); - const { service, managedSignerService, managedUserWalletService, logger, errorLogger } = setup({ - executeDerivedTx: vi.fn().mockRejectedValue(unexpectedError) - }); + it("sweeps when the indexer is within the tolerated lag", async () => { + const { service, deploymentRepository } = setup({ currentHeight: 1_000_000, chainHeight: 1_000_050 }); - await expect(service.cleanup({ concurrency: 1 })).resolves.toBeUndefined(); + const result = await service.cleanup({ concurrency: 1, dryRun: false }); - expect(managedSignerService.executeDerivedTx).toHaveBeenCalledTimes(1); - expect(managedUserWalletService.authorizeSpending).not.toHaveBeenCalled(); - expect(logger.error).not.toHaveBeenCalled(); - expect(errorLogger.error).toHaveBeenCalledWith(expect.objectContaining({ event: "DEPLOYMENT_CLEAN_UP_ERROR", error: unexpectedError })); + expect(result.ok).toBe(true); + expect(deploymentRepository.findStaleDeployments).toHaveBeenCalled(); }); }); @@ -306,28 +434,31 @@ describe(StaleManagedDeploymentsCleanerService.name, () => { function setup(input?: { currentHeight?: number; - staleDeployments?: { dseq: number }[]; + chainHeight?: number; + staleDeployments?: string[]; + orphans?: StaleDeployment[]; + walletBatches?: number[]; + walletIdsByAddress?: Record; executeDerivedTx?: ManagedSignerService["executeDerivedTx"]; - pages?: number; - walletsPerPage?: number; }) { - const wallet = createUserWallet({ id: 123, address: "akash1test" }); + const wallet = createUserWallet({ id: 123, address: OWNER }); + const orphans = input?.orphans ?? (input?.staleDeployments ?? ["456"]).map(dseq => ({ owner: OWNER, dseq })); + const walletIdsByAddress = input?.walletIdsByAddress ?? { [OWNER]: wallet.id }; - const walletPages = Array.from({ length: input?.pages ?? 1 }, (_, page) => - Array.from({ length: input?.walletsPerPage ?? 1 }, (_, index) => - page === 0 && index === 0 ? wallet : createUserWallet({ id: 1000 + page * 10 + index, address: `akash1owner${page}${index}` }) - ) - ); + const walletBatches: ManagedWalletRef[][] = input?.walletBatches + ? input.walletBatches.map((size, batch) => + Array.from({ length: size }, (_, index) => ({ id: 1000 + batch * 10 + index, address: `akash1owner${batch}${index}` })) + ) + : [Object.entries(walletIdsByAddress).map(([address, id]) => ({ id, address }))]; const userWalletRepository = mock({ - paginate: vi.fn(async (_options, cb) => { - for (const page of walletPages) { - await cb(page); - } - }) as UserWalletRepository["paginate"] + findManagedIteratively: vi.fn(async function* () { + for (const batch of walletBatches) yield batch; + }) as UserWalletRepository["findManagedIteratively"] }); const deploymentRepository = mock(); const blockRepository = mock(); + const blockHttpService = mock(); const rpcMessageService = mock(); const managedSignerService = mock({ executeDerivedTx: input?.executeDerivedTx ?? vi.fn().mockResolvedValue(buildOkTx()) @@ -341,14 +472,17 @@ describe(StaleManagedDeploymentsCleanerService.name, () => { const errorService = new ErrorService(createErrorLogger); const chainErrorService = new ChainErrorService(mock(), mock(), mock()); - blockRepository.getLatestProcessedHeight.mockResolvedValue(input?.currentHeight ?? 1_000_000); - deploymentRepository.findStaleDeployments.mockResolvedValue(input?.staleDeployments ?? [{ dseq: 456 }]); + const currentHeight = input?.currentHeight ?? 1_000_000; + blockRepository.getLatestProcessedHeight.mockResolvedValue(currentHeight); + blockHttpService.getCurrentHeight.mockResolvedValue(input?.chainHeight ?? currentHeight); + deploymentRepository.findStaleDeployments.mockImplementation(async ({ owners }) => orphans.filter(orphan => owners.includes(orphan.owner))); rpcMessageService.getCloseDeploymentMsg.mockImplementation((_address, dseq) => ({ typeUrl: "/close", value: { dseq } }) as never); const service = new StaleManagedDeploymentsCleanerService( userWalletRepository, deploymentRepository, blockRepository, + blockHttpService, rpcMessageService, managedSignerService, config, @@ -364,6 +498,7 @@ describe(StaleManagedDeploymentsCleanerService.name, () => { userWalletRepository, deploymentRepository, blockRepository, + blockHttpService, rpcMessageService, managedSignerService, managedUserWalletService, diff --git a/apps/api/src/deployment/services/stale-managed-deployments-cleaner/stale-managed-deployments-cleaner.service.ts b/apps/api/src/deployment/services/stale-managed-deployments-cleaner/stale-managed-deployments-cleaner.service.ts index a23430bf4d..a41b58b8e7 100644 --- a/apps/api/src/deployment/services/stale-managed-deployments-cleaner/stale-managed-deployments-cleaner.service.ts +++ b/apps/api/src/deployment/services/stale-managed-deployments-cleaner/stale-managed-deployments-cleaner.service.ts @@ -1,22 +1,44 @@ import type { EncodeObject } from "@cosmjs/proto-signing"; import { secondsInMinute } from "date-fns/constants"; +import { chunk } from "lodash"; +import { Err, Ok, Result } from "ts-results"; import { inject, singleton } from "tsyringe"; import { type BillingConfig, InjectBillingConfig } from "@src/billing/providers"; -import { UserWalletOutput, UserWalletRepository } from "@src/billing/repositories"; +import { type ManagedWalletRef, UserWalletOutput, UserWalletRepository } from "@src/billing/repositories"; import { ManagedUserWalletService, RpcMessageService } from "@src/billing/services"; import { ChainErrorService } from "@src/billing/services/chain-error/chain-error.service"; import { ManagedSignerService } from "@src/billing/services/managed-signer/managed-signer.service"; import { BlockRepository } from "@src/chain/repositories/block.repository"; +import { BlockHttpService } from "@src/chain/services/block-http/block-http.service"; import { type CreateLogger, LOGGER_FACTORY } from "@src/core"; import { ErrorService } from "@src/core/services/error/error.service"; -import { DeploymentRepository } from "@src/deployment/repositories/deployment/deployment.repository"; +import { DeploymentRepository, type StaleDeployment, type StaleDeploymentsOutput } from "@src/deployment/repositories/deployment/deployment.repository"; import { CleanUpStaleDeploymentsParams } from "@src/deployment/types/state-deployments"; import { averageBlockTime, COSMOS_TX_CODE_OK } from "@src/utils/constants"; -/** Bounds how many already-closed deployments one pass drops; the batch left after the last drop is still broadcast once. */ +/** Bounds how many already-closed deployments one owner's pass drops; the batch left after the last drop is still broadcast once. */ const MAX_CLOSED_DEPLOYMENT_DROPS = 3; +/** How many owners one chain query screens, which bounds both the query payload and the rows held in memory at once. */ +const WALLET_BATCH_SIZE = 5_000; + +/** Keeps one owner's orphans from growing into a transaction the chain refuses for gas, which no amount of retrying fixes. */ +const MAX_CLOSES_PER_TX = 20; + +/** How far the indexer may trail the chain before the sweep refuses to run, because a lagging indexer reads a leased deployment as lease-less. */ +const MAX_INDEXER_LAG_IN_BLOCKS = Math.floor((10 * secondsInMinute) / averageBlockTime); + +function groupByOwner(deployments: StaleDeployment[]): Map { + const byOwner = new Map(); + + for (const deployment of deployments) { + byOwner.set(deployment.owner, [...(byOwner.get(deployment.owner) ?? []), deployment]); + } + + return byOwner; +} + @singleton() export class StaleManagedDeploymentsCleanerService { private readonly MAX_LIVE_BLOCKS = Math.floor((10 * secondsInMinute) / averageBlockTime); @@ -27,6 +49,7 @@ export class StaleManagedDeploymentsCleanerService { private readonly userWalletRepository: UserWalletRepository, private readonly deploymentRepository: DeploymentRepository, private readonly blockRepository: BlockRepository, + private readonly blockHttpService: BlockHttpService, private readonly rpcMessageService: RpcMessageService, private readonly managedSignerService: ManagedSignerService, @InjectBillingConfig() private readonly config: BillingConfig, @@ -38,51 +61,140 @@ export class StaleManagedDeploymentsCleanerService { this.logger = createLogger({ context: StaleManagedDeploymentsCleanerService.name }); } - async cleanup(options: CleanUpStaleDeploymentsParams) { - const staleBeforeHeight = await this.#resolveStaleBeforeHeight(this.MAX_LIVE_BLOCKS); - - await this.userWalletRepository.paginate({ limit: options.concurrency || 10 }, async wallets => { - const cleanUpAllWallets = wallets.map(async wallet => { - await this.errorService.execWithErrorHandler( - { - wallet, - event: "DEPLOYMENT_CLEAN_UP_ERROR", - context: StaleManagedDeploymentsCleanerService.name - }, - () => this.#closeDeploymentsWithoutActiveLease(wallet, staleBeforeHeight) - ); - }); + /** One owner's failed close is logged and counted rather than failing the run, because a persistently failing owner would otherwise keep every run red. */ + async cleanup(options: CleanUpStaleDeploymentsParams): Promise> { + const startedAt = Date.now(); + const indexedHeight = await this.#resolveFreshIndexedHeight(); + + if (indexedHeight === undefined) { + return Err([new Error("Indexer is too far behind the chain to tell an orphan from a leased deployment")]); + } - await Promise.all(cleanUpAllWallets); + const staleBeforeHeight = indexedHeight - this.MAX_LIVE_BLOCKS; + const screenErrors: unknown[] = []; + let screened = 0; + let owners = 0; + let failedOwners = 0; + + this.logger.info({ event: "DEPLOYMENT_CLEAN_UP_SWEEP_START", staleBeforeHeight, dryRun: options.dryRun }); + + for await (const wallets of this.userWalletRepository.findManagedIteratively({ batchSize: WALLET_BATCH_SIZE })) { + const batch = await this.#cleanUpBatch(wallets, staleBeforeHeight, options); + + screened += wallets.length; + owners += batch.owners; + failedOwners += batch.failedOwners; + screenErrors.push(...batch.screenErrors); + } + + this.logger.info({ + event: "DEPLOYMENT_CLEAN_UP_SWEEP_END", + screened, + owners, + failedOwners, + screenFailures: screenErrors.length, + durationMs: Date.now() - startedAt, + dryRun: options.dryRun }); + + return screenErrors.length > 0 ? Err(screenErrors) : Ok(undefined); } async cleanUpForWallet(wallet: UserWalletOutput, maxLiveBlocks: number = this.MAX_LIVE_BLOCKS) { - await this.#closeDeploymentsWithoutActiveLease(wallet, await this.#resolveStaleBeforeHeight(maxLiveBlocks)); + const staleBeforeHeight = (await this.blockRepository.getLatestProcessedHeight()) - maxLiveBlocks; + const managedWallet = { id: wallet.id, address: wallet.address! }; + const deployments = await this.deploymentRepository.findStaleDeployments({ owners: [managedWallet.address], staleBeforeHeight }); + + await this.#closeDeploymentsWithoutActiveLease(managedWallet, deployments); } - /** Read once per sweep instead of per wallet: the tip is the same for every one of them, and the sweep walks the whole managed-wallet table. */ - async #resolveStaleBeforeHeight(maxLiveBlocks: number): Promise { - return (await this.blockRepository.getLatestProcessedHeight()) - maxLiveBlocks; + /** One query screens the whole batch, so a batch holding no orphan at all costs exactly that one query. */ + async #cleanUpBatch( + wallets: ManagedWalletRef[], + staleBeforeHeight: number, + options: CleanUpStaleDeploymentsParams + ): Promise<{ owners: number; failedOwners: number; screenErrors: unknown[] }> { + const screenErrors: unknown[] = []; + const deployments = await this.errorService.execWithErrorHandler( + { + event: "DEPLOYMENT_CLEAN_UP_SCREEN_ERROR", + context: StaleManagedDeploymentsCleanerService.name + }, + () => this.deploymentRepository.findStaleDeployments({ owners: wallets.map(wallet => wallet.address), staleBeforeHeight }), + error => screenErrors.push(error) + ); + const orphansByOwner = groupByOwner(deployments ?? []); + + if (options.dryRun) { + for (const [owner, orphans] of orphansByOwner) { + this.logger.info({ event: "DEPLOYMENT_CLEAN_UP_WOULD_CLOSE", owner, dseqs: orphans.map(orphan => orphan.dseq) }); + } + + return { owners: orphansByOwner.size, failedOwners: 0, screenErrors }; + } + + const walletsByAddress = new Map(wallets.map(wallet => [wallet.address, wallet])); + let failedOwners = 0; + + for (const group of chunk([...orphansByOwner], options.concurrency || 10)) { + await Promise.all( + group.map(async ([owner, orphans]) => { + await this.errorService.execWithErrorHandler( + { + owner, + event: "DEPLOYMENT_CLEAN_UP_ERROR", + context: StaleManagedDeploymentsCleanerService.name + }, + () => this.#closeDeploymentsWithoutActiveLease(walletsByAddress.get(owner)!, orphans), + () => failedOwners++ + ); + }) + ); + } + + return { owners: orphansByOwner.size, failedOwners, screenErrors }; } - /** Dropping a message and re-broadcasting is safe because both classified failures reject the tx whole: an estimate never lands, a non-zero code reverts. */ - async #closeDeploymentsWithoutActiveLease(wallet: UserWalletOutput, staleBeforeHeight: number) { - let remaining = await this.deploymentRepository.findStaleDeployments({ - owner: wallet.address!, - staleBeforeHeight - }); + /** Undefined when the indexer cannot see enough of the chain for this sweep's cutoff to mean anything. */ + async #resolveFreshIndexedHeight(): Promise { + const [chainHeight, indexedHeight] = await Promise.all([this.blockHttpService.getCurrentHeight(), this.blockRepository.getLatestProcessedHeight()]); + const lag = chainHeight - indexedHeight; + + if (lag <= MAX_INDEXER_LAG_IN_BLOCKS) return indexedHeight; + + this.logger.error({ event: "DEPLOYMENT_CLEAN_UP_INDEXER_LAGGING", chainHeight, indexedHeight, lag, maxLag: MAX_INDEXER_LAG_IN_BLOCKS }); + + return undefined; + } - if (!remaining.length) { + /** Dropping a message and re-broadcasting is safe because both classified failures reject the tx whole: an estimate never lands, a non-zero code reverts. */ + async #closeDeploymentsWithoutActiveLease(wallet: ManagedWalletRef, deployments: StaleDeploymentsOutput[]) { + if (!deployments.length) { return; } this.logger.info({ event: "DEPLOYMENT_CLEAN_UP", owner: wallet.address }); + let alreadyClosedCount = 0; + + for (const batch of chunk(deployments, MAX_CLOSES_PER_TX)) { + const dropped = await this.#closeBatch(wallet, batch, MAX_CLOSED_DEPLOYMENT_DROPS - alreadyClosedCount); + + if (dropped === undefined) return; + + alreadyClosedCount += dropped; + } + + this.logger.info({ event: "DEPLOYMENT_CLEAN_UP_SUCCESS", owner: wallet.address, alreadyClosedCount }); + } + + /** Returns how many already-closed deployments it dropped, or undefined when the wallet is left for the next run. */ + async #closeBatch(wallet: ManagedWalletRef, deployments: StaleDeploymentsOutput[], dropBudget: number): Promise { + let remaining = deployments; let closedDeploymentsDropped = 0; while (remaining.length) { - const messages = remaining.map(deployment => this.rpcMessageService.getCloseDeploymentMsg(wallet.address!, deployment.dseq)); + const messages = remaining.map(deployment => this.rpcMessageService.getCloseDeploymentMsg(wallet.address, deployment.dseq)); const failure = await this.closeDeployments(wallet, messages); if (!failure) { @@ -98,15 +210,15 @@ export class StaleManagedDeploymentsCleanerService { reason: "Deployment escrow cannot be settled yet; chain rejects close until it settles", owner: wallet.address }); - return; + return undefined; } throw failure; } - if (closedDeploymentsDropped >= MAX_CLOSED_DEPLOYMENT_DROPS) { + if (closedDeploymentsDropped >= dropBudget) { this.logger.warn({ event: "DEPLOYMENT_CLEAN_UP_DROP_LIMIT", owner: wallet.address, remainingCount: remaining.length }); - return; + return undefined; } this.logger.info({ event: "DEPLOYMENT_CLEAN_UP_ALREADY_CLOSED", owner: wallet.address, dseq: remaining[closedIndex].dseq }); @@ -114,11 +226,11 @@ export class StaleManagedDeploymentsCleanerService { closedDeploymentsDropped++; } - this.logger.info({ event: "DEPLOYMENT_CLEAN_UP_SUCCESS", owner: wallet.address, alreadyClosedCount: closedDeploymentsDropped }); + return closedDeploymentsDropped; } /** Returns the failure rather than throwing so the caller classifies a rejected estimate and a reverted tx alike. */ - private async closeDeployments(wallet: UserWalletOutput, messages: EncodeObject[]): Promise { + private async closeDeployments(wallet: ManagedWalletRef, messages: EncodeObject[]): Promise { try { await this.#broadcastClose(wallet.id, messages); return undefined; @@ -128,7 +240,7 @@ export class StaleManagedDeploymentsCleanerService { } await this.managedUserWalletService.authorizeSpending(this.managedSignerService, { - address: wallet.address!, + address: wallet.address, limits: { fees: this.config.FEE_ALLOWANCE_REFILL_AMOUNT } diff --git a/apps/api/src/deployment/types/state-deployments.ts b/apps/api/src/deployment/types/state-deployments.ts index 384a1c5b70..38c216b9c2 100644 --- a/apps/api/src/deployment/types/state-deployments.ts +++ b/apps/api/src/deployment/types/state-deployments.ts @@ -1,3 +1,3 @@ -import type { ConcurrencyOptions } from "@src/core/types/console"; +import type { ConcurrencyOptions, DryRunOptions } from "@src/core/types/console"; -export interface CleanUpStaleDeploymentsParams extends ConcurrencyOptions {} +export interface CleanUpStaleDeploymentsParams extends ConcurrencyOptions, DryRunOptions {} diff --git a/apps/api/test/seeders/stale-deployment.seeder.ts b/apps/api/test/seeders/stale-deployment.seeder.ts index cea7eeb069..1154527907 100644 --- a/apps/api/test/seeders/stale-deployment.seeder.ts +++ b/apps/api/test/seeders/stale-deployment.seeder.ts @@ -2,7 +2,7 @@ import { faker } from "@faker-js/faker"; import type { StaleDeploymentsOutput } from "@src/deployment/repositories/deployment/deployment.repository"; -export function createStaleDeployment({ dseq = faker.number.int({ min: 1, max: 99999999 }) }: Partial = {}): StaleDeploymentsOutput { +export function createStaleDeployment({ dseq = faker.string.numeric({ length: 8 }) }: Partial = {}): StaleDeploymentsOutput { return { dseq };