From 57269564e7abcd1d3500c8421d3e7fc90ff85072 Mon Sep 17 00:00:00 2001 From: Maxime Beauchamp <15185355+baktun14@users.noreply.github.com> Date: Sat, 19 Sep 2026 01:21:29 +0400 Subject: [PATCH 1/3] fix(deployment): find orphaned deployments directly so the sweep can close them in minutes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An open, lease-less deployment holds the minimum escrow deposit until the cleanup sweep closes it. Bids expire about five minutes after a deployment is created, so one with no lease after ten minutes can never become active — but the sweep only ran at 10:30 and 22:30, so a trial user could wait half a day to get their credits back. The sweep could not simply run more often. It paged through every managed wallet and asked the chain database whether that one owner had orphans, so a run cost a query per wallet and grew with the wallet table rather than with the handful of real orphans. It now asks about a batch of owners at a time: one query screens five thousand wallets and comes back with only the orphans worth closing, which is a couple of dozen queries per run instead of tens of thousands. Closing is unchanged — still one transaction per owner, with the same already-closed drop, fee-grant refill and unsettleable handling. Two things the ten-minute cadence made worth guarding. The sweep now refuses to run when the indexer trails the chain far enough that a leased deployment would read as lease-less, because the staleness cutoff is derived from the last indexed height. And one owner's orphans are split across transactions, so a wallet holding many of them cannot build a close the chain rejects for gas on every run forever. The command also takes --dry-run, which reports what it would close and broadcasts nothing, so the new query can be watched in production before it is trusted to act. --- .helm/console-api-prod-mainnet-values.yaml | 5 +- .helm/console-api-staging-sandbox-values.yaml | 5 +- apps/api/src/app/console.ts | 3 +- .../user-wallet/user-wallet.repository.ts | 32 ++- .../top-up-deployments.controller.spec.ts | 2 +- .../top-up-deployments.controller.ts | 2 +- .../deployment.repository.integration.ts | 44 +++- .../deployment/deployment.repository.ts | 58 +++-- ...anaged-deployments-cleaner.service.spec.ts | 244 ++++++++++++------ ...ale-managed-deployments-cleaner.service.ts | 171 +++++++++--- .../src/deployment/types/state-deployments.ts | 4 +- .../test/seeders/stale-deployment.seeder.ts | 2 +- 12 files changed, 423 insertions(+), 149 deletions(-) diff --git a/.helm/console-api-prod-mainnet-values.yaml b/.helm/console-api-prod-mainnet-values.yaml index cb3e19350a..262f63ed7c 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 + # Offset off every other sweep's minute, mint-act's */10 included, so two jobs never contend for the signer at once. - 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..848759a16e 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 + # Offset off every other sweep's minute so two jobs never contend for the signer at once. - 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..b6b49db496 100644 --- a/apps/api/src/app/console.ts +++ b/apps/api/src/app/console.ts @@ -57,9 +57,10 @@ 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("-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.ts b/apps/api/src/billing/repositories/user-wallet/user-wallet.repository.ts index 879954c3b8..24e7b8f72a 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,30 @@ 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..399417348e 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 { @@ -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,32 @@ 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 of every managed wallet costs a query per + * batch instead of a query per wallet, and the batch that comes back is already only the orphans worth closing. + */ + 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..2c9adbc3c4 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.err).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,23 @@ 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("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 +197,120 @@ 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 when one of them fails", async () => { + const failure = new Error("some unexpected failure"); + const executeDerivedTx = vi.fn().mockRejectedValueOnce(failure).mockResolvedValue(buildOkTx()); + const { service, 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.err).toBe(true); + expect(errorLogger.error).toHaveBeenCalledWith(expect.objectContaining({ event: "DEPLOYMENT_CLEAN_UP_ERROR", error: failure })); + }); + + 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("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 +321,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 +332,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 }); + 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: 3 }); + const result = await service.cleanup({ concurrency: 1, dryRun: false }); - expect(blockRepository.getLatestProcessedHeight).toHaveBeenCalledTimes(1); - expect(deploymentRepository.findStaleDeployments).toHaveBeenCalledTimes(12); + 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("screens every wallet against the same cutoff", async () => { - const { service, deploymentRepository } = setup({ currentHeight: 1_000_000, pages: 3, walletsPerPage: 2 }); + it("sweeps when the indexer is within the tolerated lag", async () => { + const { service, deploymentRepository } = setup({ currentHeight: 1_000_000, chainHeight: 1_000_050 }); - 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); - }); - - 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) - }); - - await expect(service.cleanup({ concurrency: 1 })).resolves.toBeUndefined(); - - 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 +395,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 +433,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 +459,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..ea1fcc174c 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,47 @@ 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. */ 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 a sweep refuses to run: the staleness cutoff is derived from the last + * indexed height, so an indexer far enough behind reports a leased deployment as lease-less and the sweep closes a live one. + */ +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 +52,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,27 +64,97 @@ 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) - ); - }); + /** + * Asks the chain which of a batch of managed wallets owns an orphan rather than asking each wallet in turn, so the run + * costs a query per batch instead of a query per wallet and stays flat as the wallet table grows. + */ + async cleanup(options: CleanUpStaleDeploymentsParams): Promise> { + const indexedHeight = await this.#resolveFreshIndexedHeight(); - await Promise.all(cleanUpAllWallets); - }); + if (indexedHeight === undefined) { + return Err([new Error("Indexer is too far behind the chain to tell an orphan from a leased deployment")]); + } + + const staleBeforeHeight = indexedHeight - this.MAX_LIVE_BLOCKS; + const errors: unknown[] = []; + let screened = 0; + let owners = 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; + errors.push(...batch.errors); + } + + this.logger.info({ event: "DEPLOYMENT_CLEAN_UP_SWEEP_END", screened, owners, failed: errors.length, dryRun: options.dryRun }); + + return errors.length > 0 ? Err(errors) : Ok(undefined); } async cleanUpForWallet(wallet: UserWalletOutput, maxLiveBlocks: number = this.MAX_LIVE_BLOCKS) { - await this.#closeDeploymentsWithoutActiveLease(wallet, await this.#resolveStaleBeforeHeight(maxLiveBlocks)); + const staleBeforeHeight = await this.#resolveStaleBeforeHeight(maxLiveBlocks); + const managedWallet = { id: wallet.id, address: wallet.address! }; + const deployments = await this.deploymentRepository.findStaleDeployments({ owners: [managedWallet.address], staleBeforeHeight }); + + await this.#closeDeploymentsWithoutActiveLease(managedWallet, deployments); + } + + /** 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; errors: unknown[] }> { + const deployments = await this.deploymentRepository.findStaleDeployments({ + owners: wallets.map(wallet => wallet.address), + staleBeforeHeight + }); + const orphansByOwner = groupByOwner(deployments); + const errors: unknown[] = []; + + 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, errors }; + } + + const walletsByAddress = new Map(wallets.map(wallet => [wallet.address, wallet])); + + 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), + error => errors.push(error) + ); + }) + ); + } + + return { owners: orphansByOwner.size, errors }; + } + + /** The chain is the authority on where the tip is; the indexer only says how much of it this sweep can see. Undefined means it cannot see enough. */ + 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; } /** 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. */ @@ -67,22 +163,33 @@ export class StaleManagedDeploymentsCleanerService { } /** 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 - }); - - if (!remaining.length) { + 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); + + 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[]): 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,7 +205,7 @@ export class StaleManagedDeploymentsCleanerService { reason: "Deployment escrow cannot be settled yet; chain rejects close until it settles", owner: wallet.address }); - return; + return undefined; } throw failure; @@ -106,7 +213,7 @@ export class StaleManagedDeploymentsCleanerService { if (closedDeploymentsDropped >= MAX_CLOSED_DEPLOYMENT_DROPS) { 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 +221,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 +235,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 }; From 7b89003a130ffaaa7070538a7fa43d2d41700da7 Mon Sep 17 00:00:00 2001 From: Maxime Beauchamp <15185355+baktun14@users.noreply.github.com> Date: Sat, 19 Sep 2026 01:55:28 +0400 Subject: [PATCH 2/3] fix(deployment): keep one batch's failure and one owner's drop budget from spreading Three things the review caught in the inverted sweep. Screening a batch of owners ran unguarded, so a transient chain-database error on one batch threw out of the sweep loop and skipped every remaining batch for that run. Before the inversion each owner's query ran inside the error handler, so one failure was logged and the rest carried on. The batch query is now handled the same way. Chunking an owner's closes across transactions also reset the already-closed drop cap on every chunk, so an owner with a hundred orphans could drop and re-broadcast fifteen times in a run instead of three. The budget is now spent across all of an owner's transactions. The new JSDoc blocks ran to several lines, which the repo's comment rule does not allow. --- .../user-wallet/user-wallet.repository.ts | 5 +-- .../deployment/deployment.repository.ts | 5 +-- ...anaged-deployments-cleaner.service.spec.ts | 30 ++++++++++++++++ ...ale-managed-deployments-cleaner.service.ts | 34 +++++++++---------- 4 files changed, 48 insertions(+), 26 deletions(-) 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 24e7b8f72a..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 @@ -249,10 +249,7 @@ export class UserWalletRepository extends BaseRepository { let cursor: number | undefined; diff --git a/apps/api/src/deployment/repositories/deployment/deployment.repository.ts b/apps/api/src/deployment/repositories/deployment/deployment.repository.ts index 399417348e..a44071359a 100644 --- a/apps/api/src/deployment/repositories/deployment/deployment.repository.ts +++ b/apps/api/src/deployment/repositories/deployment/deployment.repository.ts @@ -121,10 +121,7 @@ export class DeploymentRepository { }); } - /** - * Owners reach the query as one array rather than one call each, so a sweep of every managed wallet costs a query per - * batch instead of a query per wallet, and the batch that comes back is already only the orphans worth closing. - */ + /** 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 []; 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 2c9adbc3c4..3a2bf9b639 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 @@ -179,6 +179,24 @@ describe(StaleManagedDeploymentsCleanerService.name, () => { 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()); @@ -303,6 +321,18 @@ describe(StaleManagedDeploymentsCleanerService.name, () => { expect(result.ok).toBe(true); }); + it("carries on to the next batch 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" })); + }); + 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()) 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 ea1fcc174c..b3588388bb 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 @@ -17,7 +17,7 @@ import { DeploymentRepository, type StaleDeployment, type StaleDeploymentsOutput 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. */ @@ -26,10 +26,7 @@ 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 a sweep refuses to run: the staleness cutoff is derived from the last - * indexed height, so an indexer far enough behind reports a leased deployment as lease-less and the sweep closes a live one. - */ +/** 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 { @@ -64,10 +61,7 @@ export class StaleManagedDeploymentsCleanerService { this.logger = createLogger({ context: StaleManagedDeploymentsCleanerService.name }); } - /** - * Asks the chain which of a batch of managed wallets owns an orphan rather than asking each wallet in turn, so the run - * costs a query per batch instead of a query per wallet and stays flat as the wallet table grows. - */ + /** Asks the chain which of a batch of wallets owns an orphan instead of asking each wallet in turn, so a run costs a query per batch, not per wallet. */ async cleanup(options: CleanUpStaleDeploymentsParams): Promise> { const indexedHeight = await this.#resolveFreshIndexedHeight(); @@ -109,12 +103,16 @@ export class StaleManagedDeploymentsCleanerService { staleBeforeHeight: number, options: CleanUpStaleDeploymentsParams ): Promise<{ owners: number; errors: unknown[] }> { - const deployments = await this.deploymentRepository.findStaleDeployments({ - owners: wallets.map(wallet => wallet.address), - staleBeforeHeight - }); - const orphansByOwner = groupByOwner(deployments); const errors: 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 => errors.push(error) + ); + const orphansByOwner = groupByOwner(deployments ?? []); if (options.dryRun) { for (const [owner, orphans] of orphansByOwner) { @@ -145,7 +143,7 @@ export class StaleManagedDeploymentsCleanerService { return { owners: orphansByOwner.size, errors }; } - /** The chain is the authority on where the tip is; the indexer only says how much of it this sweep can see. Undefined means it cannot see enough. */ + /** 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; @@ -173,7 +171,7 @@ export class StaleManagedDeploymentsCleanerService { let alreadyClosedCount = 0; for (const batch of chunk(deployments, MAX_CLOSES_PER_TX)) { - const dropped = await this.#closeBatch(wallet, batch); + const dropped = await this.#closeBatch(wallet, batch, MAX_CLOSED_DEPLOYMENT_DROPS - alreadyClosedCount); if (dropped === undefined) return; @@ -184,7 +182,7 @@ export class StaleManagedDeploymentsCleanerService { } /** 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[]): Promise { + async #closeBatch(wallet: ManagedWalletRef, deployments: StaleDeploymentsOutput[], dropBudget: number): Promise { let remaining = deployments; let closedDeploymentsDropped = 0; @@ -211,7 +209,7 @@ export class StaleManagedDeploymentsCleanerService { 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 undefined; } From 89b3d6509f818a86f021a9f921ea07f1d34c8d92 Mon Sep 17 00:00:00 2001 From: Maxime Beauchamp <15185355+baktun14@users.noreply.github.com> Date: Sun, 20 Sep 2026 17:16:06 +0400 Subject: [PATCH 3/3] fix(deployment): keep one owner's failed close from failing the whole cleanup run The sweep returned Err whenever one owner's close failed, and the CLI handler turns an Err into exit code 1. The console-api CronJob chart runs jobs with restartPolicy Never and the Job default of six retries, so a single failing owner re-ran the whole sweep several times per tick and kept the CronJob red for as long as that owner stayed broken. An orphan on an expired trial is the obvious case: the chain answers "fee allowance expired", the fee-refill retry only matches "not allowed to pay fees", so that close never succeeds. Now only a failure that left wallets unscreened fails the run. A failed close is still logged, and DEPLOYMENT_CLEAN_UP_SWEEP_END counts it as failedOwners next to screenFailures. The event also carries durationMs so a dry run shows what a pass costs. findManagedIteratively drove the sweep with no test of its own; it now has integration coverage. The single-wallet cutoff helper is inlined, since its comment still described the old per-wallet sweep, and the asHeight comment now says why the check exists with the height bound as a parameter instead of interpolated. The helm comments stop claiming two jobs can never contend for the signer: Forbid only keeps a job from overlapping itself. --- .helm/console-api-prod-mainnet-values.yaml | 2 +- .helm/console-api-staging-sandbox-values.yaml | 2 +- apps/api/src/app/console.ts | 2 +- .../user-wallet.repository.integration.ts | 64 +++++++++++++++++++ .../deployment/deployment.repository.ts | 2 +- ...anaged-deployments-cleaner.service.spec.ts | 21 ++++-- ...ale-managed-deployments-cleaner.service.ts | 43 +++++++------ 7 files changed, 108 insertions(+), 28 deletions(-) diff --git a/.helm/console-api-prod-mainnet-values.yaml b/.helm/console-api-prod-mainnet-values.yaml index 262f63ed7c..356bb4335b 100644 --- a/.helm/console-api-prod-mainnet-values.yaml +++ b/.helm/console-api-prod-mainnet-values.yaml @@ -19,7 +19,7 @@ jobs: - ./dist/instrumentation.js - ./dist/console.js - refill-wallets - # Offset off every other sweep's minute, mint-act's */10 included, so two jobs never contend for the signer at once. + # 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: "4,14,24,34,44,54 * * * *" # every 10 minutes concurrencyPolicy: Forbid diff --git a/.helm/console-api-staging-sandbox-values.yaml b/.helm/console-api-staging-sandbox-values.yaml index 848759a16e..d3facd68a8 100644 --- a/.helm/console-api-staging-sandbox-values.yaml +++ b/.helm/console-api-staging-sandbox-values.yaml @@ -17,7 +17,7 @@ jobs: - ./dist/instrumentation.js - ./dist/console.js - refill-wallets - # Offset off every other sweep's minute so two jobs never contend for the signer at once. + # Start minutes are offset off every other sweep's so no two jobs start on the same minute. - name: cleanup-stale-deployments schedule: "4,14,24,34,44,54 * * * *" # every 10 minutes concurrencyPolicy: Forbid diff --git a/apps/api/src/app/console.ts b/apps/api/src/app/console.ts index b6b49db496..5052d5c7e0 100644 --- a/apps/api/src/app/console.ts +++ b/apps/api/src/app/console.ts @@ -56,7 +56,7 @@ 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 () => { 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/deployment/repositories/deployment/deployment.repository.ts b/apps/api/src/deployment/repositories/deployment/deployment.repository.ts index a44071359a..0aa3e8a365 100644 --- a/apps/api/src/deployment/repositories/deployment/deployment.repository.ts +++ b/apps/api/src/deployment/repositories/deployment/deployment.repository.ts @@ -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}`); 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 3a2bf9b639..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 @@ -134,7 +134,7 @@ describe(StaleManagedDeploymentsCleanerService.name, () => { const result = await service.cleanup({ concurrency: 1, dryRun: false }); - expect(result.err).toBe(true); + 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 })); @@ -294,10 +294,10 @@ describe(StaleManagedDeploymentsCleanerService.name, () => { expect(logger.info).toHaveBeenCalledWith({ event: "DEPLOYMENT_CLEAN_UP_WOULD_CLOSE", owner: OWNER, dseqs: ["9"] }); }); - it("carries on to the other owners when one of them fails", async () => { + 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, errorLogger } = setup({ + const { service, logger, errorLogger } = setup({ orphans: [ { owner: "akash1a", dseq: "1" }, { owner: "akash1b", dseq: "2" } @@ -309,8 +309,9 @@ describe(StaleManagedDeploymentsCleanerService.name, () => { const result = await service.cleanup({ concurrency: 1, dryRun: false }); expect(executeDerivedTx).toHaveBeenCalledTimes(2); - expect(result.err).toBe(true); + 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 () => { @@ -321,7 +322,15 @@ describe(StaleManagedDeploymentsCleanerService.name, () => { expect(result.ok).toBe(true); }); - it("carries on to the next batch when screening one of them fails", async () => { + 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([]); @@ -330,7 +339,7 @@ describe(StaleManagedDeploymentsCleanerService.name, () => { expect(deploymentRepository.findStaleDeployments).toHaveBeenCalledTimes(2); expect(result.err).toBe(true); - expect(logger.info).toHaveBeenCalledWith(expect.objectContaining({ event: "DEPLOYMENT_CLEAN_UP_SWEEP_END" })); + 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 () => { 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 b3588388bb..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 @@ -61,8 +61,9 @@ export class StaleManagedDeploymentsCleanerService { this.logger = createLogger({ context: StaleManagedDeploymentsCleanerService.name }); } - /** Asks the chain which of a batch of wallets owns an orphan instead of asking each wallet in turn, so a run costs a query per batch, not per wallet. */ + /** 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) { @@ -70,9 +71,10 @@ export class StaleManagedDeploymentsCleanerService { } const staleBeforeHeight = indexedHeight - this.MAX_LIVE_BLOCKS; - const errors: unknown[] = []; + 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 }); @@ -81,16 +83,25 @@ export class StaleManagedDeploymentsCleanerService { screened += wallets.length; owners += batch.owners; - errors.push(...batch.errors); + failedOwners += batch.failedOwners; + screenErrors.push(...batch.screenErrors); } - this.logger.info({ event: "DEPLOYMENT_CLEAN_UP_SWEEP_END", screened, owners, failed: errors.length, dryRun: options.dryRun }); - - return errors.length > 0 ? Err(errors) : Ok(undefined); + 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) { - const staleBeforeHeight = 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 }); @@ -102,15 +113,15 @@ export class StaleManagedDeploymentsCleanerService { wallets: ManagedWalletRef[], staleBeforeHeight: number, options: CleanUpStaleDeploymentsParams - ): Promise<{ owners: number; errors: unknown[] }> { - const errors: unknown[] = []; + ): 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 => errors.push(error) + error => screenErrors.push(error) ); const orphansByOwner = groupByOwner(deployments ?? []); @@ -119,10 +130,11 @@ export class StaleManagedDeploymentsCleanerService { this.logger.info({ event: "DEPLOYMENT_CLEAN_UP_WOULD_CLOSE", owner, dseqs: orphans.map(orphan => orphan.dseq) }); } - return { owners: orphansByOwner.size, errors }; + 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( @@ -134,13 +146,13 @@ export class StaleManagedDeploymentsCleanerService { context: StaleManagedDeploymentsCleanerService.name }, () => this.#closeDeploymentsWithoutActiveLease(walletsByAddress.get(owner)!, orphans), - error => errors.push(error) + () => failedOwners++ ); }) ); } - return { owners: orphansByOwner.size, errors }; + return { owners: orphansByOwner.size, failedOwners, screenErrors }; } /** Undefined when the indexer cannot see enough of the chain for this sweep's cutoff to mean anything. */ @@ -155,11 +167,6 @@ export class StaleManagedDeploymentsCleanerService { return undefined; } - /** 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; - } - /** 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) {