diff --git a/packages/backend/server/src/core/storage/__tests__/blob-job.spec.ts b/packages/backend/server/src/core/storage/__tests__/blob-job.spec.ts index 2b433ad2a1..842e005603 100644 --- a/packages/backend/server/src/core/storage/__tests__/blob-job.spec.ts +++ b/packages/backend/server/src/core/storage/__tests__/blob-job.spec.ts @@ -156,3 +156,60 @@ test('doc blob refs sweep continues after one workspace fails', async t => { ) ); }); + +test('blob cleanup planning drains each workspace cursor before continuing', async t => { + t.context.db.workspace.findMany.resolves([ + { id: 'workspace-1', sid: 1 }, + { id: 'workspace-2', sid: 2 }, + ]); + t.context.runtime.planUnreferencedWorkspaceBlobs + .onFirstCall() + .resolves({ + runId: 'run-1', + scannedBlobs: 100, + candidatesMarked: 100, + nextCursor: 'cursor-1', + }) + .onSecondCall() + .resolves({ + runId: 'run-2', + scannedBlobs: 50, + candidatesMarked: 40, + nextCursor: null, + }) + .onThirdCall() + .resolves({ + runId: 'run-3', + scannedBlobs: 1, + candidatesMarked: 0, + nextCursor: null, + }); + + await t.context.job.planUnreferencedWorkspaceBlobsBySid({ + workspaceLimit: 2, + gracePeriodDays: 14, + limit: 100, + }); + + t.is(t.context.runtime.planUnreferencedWorkspaceBlobs.callCount, 3); + t.deepEqual(t.context.runtime.planUnreferencedWorkspaceBlobs.firstCall.args, [ + 'workspace-1', + 14, + 100, + ]); + t.deepEqual( + t.context.runtime.planUnreferencedWorkspaceBlobs.secondCall.args, + ['workspace-1', 14, 100] + ); + t.deepEqual(t.context.runtime.planUnreferencedWorkspaceBlobs.thirdCall.args, [ + 'workspace-2', + 14, + 100, + ]); + t.true( + t.context.queue.add.calledWith( + 'backendRuntime.planUnreferencedWorkspaceBlobsBySid', + { lastSid: 2, workspaceLimit: 2, gracePeriodDays: 14, limit: 100 } + ) + ); +}); diff --git a/packages/backend/server/src/core/storage/blob-job.ts b/packages/backend/server/src/core/storage/blob-job.ts index 235daf59b3..2d25ce462e 100644 --- a/packages/backend/server/src/core/storage/blob-job.ts +++ b/packages/backend/server/src/core/storage/blob-job.ts @@ -275,14 +275,7 @@ export class StorageBlobJob { return; } - const result = await this.rt.planUnreferencedWorkspaceBlobs( - workspaceId, - gracePeriodDays, - limit - ); - this.logger.log( - `planned blob cleanup workspace=${workspaceId} run=${result.runId} candidates=${result.candidatesMarked} scanned=${result.scannedBlobs}` - ); + await this.drainBlobCleanupPlanning(workspaceId, gracePeriodDays, limit); } @OnJob('backendRuntime.planUnreferencedWorkspaceBlobsBySid') @@ -314,13 +307,11 @@ export class StorageBlobJob { for (const workspace of workspaces) { try { - const result = await this.rt.planUnreferencedWorkspaceBlobs( + await this.drainBlobCleanupPlanning( workspace.id, gracePeriodDays, - limit - ); - this.logger.log( - `planned blob cleanup workspace=${workspace.id} sid=${workspace.sid} run=${result.runId} candidates=${result.candidatesMarked} scanned=${result.scannedBlobs}` + limit, + { sid: workspace.sid } ); } catch (err) { this.logger.error( @@ -409,6 +400,27 @@ export class StorageBlobJob { } } + private async drainBlobCleanupPlanning( + workspaceId: string, + gracePeriodDays: number, + limit: number, + context: { sid?: number } = {} + ) { + for (;;) { + const result = await this.rt.planUnreferencedWorkspaceBlobs( + workspaceId, + gracePeriodDays, + limit + ); + this.logger.log( + `planned blob cleanup workspace=${workspaceId}${context.sid === undefined ? '' : ` sid=${context.sid}`} run=${result.runId} candidates=${result.candidatesMarked} scanned=${result.scannedBlobs}` + ); + if (!result.nextCursor) { + break; + } + } + } + private async hasObjectStorage(operation: string) { const health = await this.rt.health(); if (health.provider) {