fix(server): blob gc planning

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 }
+    )
+  );
+});

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) {