feat(cleanup): add bounded manual retention runs

Author: TheodoreSpeaks

apps/sim/app/api/cron/cleanup-soft-deletes/route.ts

@@ -1,18 +1,36 @@
 import { createLogger } from '@sim/logger'
 import { type NextRequest, NextResponse } from 'next/server'
+import { softDeletesCleanupContract } from '@/lib/api/contracts/cleanup'
+import { parseRequest } from '@/lib/api/server/validation'
 import { verifyCronAuth } from '@/lib/auth/internal'
-import { dispatchCleanupJobs } from '@/lib/billing/cleanup-dispatcher'
+import { dispatchBoundedCleanup, dispatchCleanupJobs } from '@/lib/billing/cleanup-dispatcher'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 
 export const dynamic = 'force-dynamic'
 
 const logger = createLogger('SoftDeleteCleanupAPI')
 
+/** Cron-secret maintenance protocol is global; workspace principal authorization does not apply. */
 export const GET = withRouteHandler(async (request: NextRequest) => {
   try {
     const authError = verifyCronAuth(request, 'soft-delete cleanup')
     if (authError) return authError
 
+    const parsed = await parseRequest(
+      softDeletesCleanupContract,
+      request,
+      {},
+      {
+        rejectDuplicateQueryValues: true,
+        rejectBlankQueryValues: true,
+      }
+    )
+    if (!parsed.success) return parsed.response
+    if (parsed.data.query) {
+      const result = await dispatchBoundedCleanup('cleanup-soft-deletes', parsed.data.query)
+      return NextResponse.json(result, { status: 202 })
+    }
+
     const result = await dispatchCleanupJobs('cleanup-soft-deletes')
 
     logger.info('Soft-delete cleanup jobs dispatched', result)

apps/sim/app/api/logs/cleanup/route.test.ts

@@ -0,0 +1,75 @@
+import { createMockRequest } from '@sim/testing'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { auth, bounded, scheduled } = vi.hoisted(() => ({
+  auth: vi.fn(),
+  bounded: vi.fn(),
+  scheduled: vi.fn(),
+}))
+vi.mock('@/lib/auth/internal', () => ({ verifyCronAuth: auth }))
+vi.mock('@/lib/billing/cleanup-dispatcher', () => ({
+  dispatchBoundedCleanup: bounded,
+  dispatchCleanupJobs: scheduled,
+}))
+
+import { GET as softDeletes } from '@/app/api/cron/cleanup-soft-deletes/route'
+import { GET as logs } from '@/app/api/logs/cleanup/route'
+
+for (const [path, GET, type, limit] of [
+  ['/api/logs/cleanup', logs, 'cleanup-logs', 'workflowLogs'],
+  ['/api/cron/cleanup-soft-deletes', softDeletes, 'cleanup-soft-deletes', 'workflows'],
+] as const) {
+  describe(path, () => {
+    beforeEach(() => {
+      vi.clearAllMocks()
+      auth.mockReturnValue(null)
+      bounded.mockResolvedValue({ runId: 'run-one', mode: 'bounded' })
+      scheduled.mockResolvedValue({
+        jobIds: ['batch-one'],
+        jobCount: 1,
+        chunkCount: 2,
+        workspaceCount: 3,
+      })
+    })
+    const request = (query = '') =>
+      createMockRequest('GET', undefined, {}, `http://localhost:3000${path}${query}`)
+    it('authenticates before parsing invalid limits', async () => {
+      auth.mockReturnValue(new Response(null, { status: 401 }))
+      expect((await GET(request('?unknown=1'))).status).toBe(401)
+      expect(bounded).not.toHaveBeenCalled()
+      expect(scheduled).not.toHaveBeenCalled()
+    })
+    it('keeps no-parameter scheduled dispatch unchanged', async () => {
+      const response = await GET(request())
+      expect(response.status).toBe(200)
+      expect(scheduled).toHaveBeenCalledWith(type)
+      expect(bounded).not.toHaveBeenCalled()
+    })
+    it('accepts one bounded run', async () => {
+      const response = await GET(request(`?${limit}=2&requestId=wave-1&dryRun=true`))
+      expect(response.status).toBe(202)
+      expect(bounded).toHaveBeenCalledWith(
+        type,
+        expect.objectContaining({
+          limits: expect.objectContaining({ [limit]: 2 }),
+          requestId: 'wave-1',
+          dryRun: true,
+          batchSize: 25,
+        })
+      )
+      expect(scheduled).not.toHaveBeenCalled()
+    })
+    it.each(['?dryRun=true', '?unknown=1', '?batchSize=3', `?${limit}=2&${limit}=3&requestId=r`])(
+      'rejects invalid query %s',
+      async (query) => {
+        expect((await GET(request(query))).status).toBe(400)
+        expect(bounded).not.toHaveBeenCalled()
+        expect(scheduled).not.toHaveBeenCalled()
+      }
+    )
+    it('reports a dispatch failure', async () => {
+      bounded.mockRejectedValue(new Error('Trigger unavailable'))
+      expect((await GET(request(`?${limit}=2&requestId=r`))).status).toBe(500)
+    })
+  })
+}

apps/sim/app/api/logs/cleanup/route.ts

@@ -1,18 +1,36 @@
 import { createLogger } from '@sim/logger'
 import { type NextRequest, NextResponse } from 'next/server'
+import { logsCleanupContract } from '@/lib/api/contracts/cleanup'
+import { parseRequest } from '@/lib/api/server/validation'
 import { verifyCronAuth } from '@/lib/auth/internal'
-import { dispatchCleanupJobs } from '@/lib/billing/cleanup-dispatcher'
+import { dispatchBoundedCleanup, dispatchCleanupJobs } from '@/lib/billing/cleanup-dispatcher'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 
 export const dynamic = 'force-dynamic'
 
 const logger = createLogger('LogsCleanupAPI')
 
+/** Cron-secret maintenance protocol is global; workspace principal authorization does not apply. */
 export const GET = withRouteHandler(async (request: NextRequest) => {
   try {
     const authError = verifyCronAuth(request, 'logs cleanup')
     if (authError) return authError
 
+    const parsed = await parseRequest(
+      logsCleanupContract,
+      request,
+      {},
+      {
+        rejectDuplicateQueryValues: true,
+        rejectBlankQueryValues: true,
+      }
+    )
+    if (!parsed.success) return parsed.response
+    if (parsed.data.query) {
+      const result = await dispatchBoundedCleanup('cleanup-logs', parsed.data.query)
+      return NextResponse.json(result, { status: 202 })
+    }
+
     const result = await dispatchCleanupJobs('cleanup-logs')
 
     logger.info('Log cleanup jobs dispatched', result)

apps/sim/background/cleanup-bounded.test.ts

@@ -0,0 +1,211 @@
+import {
+  dbChainMock,
+  dbChainMockFns,
+  hasMockCondition,
+  queueTableRows,
+  resetDbChainMock,
+  schemaMock,
+} from '@sim/testing'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { storage, prepareChat, executeChat, hardDelete, billing, decrement, reRoot } = vi.hoisted(
+  () => ({
+    storage: vi.fn(),
+    prepareChat: vi.fn(),
+    executeChat: vi.fn(),
+    hardDelete: vi.fn(),
+    billing: vi.fn(),
+    decrement: vi.fn(),
+    reRoot: vi.fn(),
+  })
+)
+vi.mock('@/background/cleanup-logs', () => ({ legacyLargeValuePredicate: vi.fn() }))
+vi.mock('@/background/cleanup-soft-deletes', () => ({
+  reRootActiveFolderChildrenUnguarded: reRoot,
+}))
+vi.mock('@/lib/uploads', () => ({
+  isUsingCloudStorage: () => true,
+  StorageService: { deleteFiles: storage },
+}))
+vi.mock('@/lib/cleanup/chat-cleanup', () => ({ prepareChatCleanup: prepareChat }))
+vi.mock('@/lib/knowledge/documents/service', () => ({ hardDeleteDocuments: hardDelete }))
+vi.mock('@/lib/billing/storage', () => ({
+  resolveStorageBillingContext: billing,
+  decrementStorageUsageForBillingContextInTx: decrement,
+}))
+
+import { BoundedCleanup, type CleanupTransaction } from '@/lib/cleanup/bounded'
+import type { CleanupType } from '@/lib/cleanup/bounded-types'
+import { runBoundedLogScope } from '@/background/cleanup-logs-bounded'
+import { runBoundedSoftDeleteScope } from '@/background/cleanup-soft-deletes-bounded'
+
+const scope = {
+  plan: 'free' as const,
+  workspaceIds: ['ws-one'],
+  retentionHours: 720,
+  label: 'test',
+  runGlobalHousekeeping: true,
+}
+function control(type: CleanupType, dryRun = true, limit = 1) {
+  return new BoundedCleanup(
+    { limits: { [type]: limit }, batchSize: 1, dryRun, requestId: 'test' },
+    async () => {},
+    Date.now,
+    async (query) => query(dbChainMock.db as CleanupTransaction)
+  )
+}
+beforeEach(() => {
+  vi.clearAllMocks()
+  resetDbChainMock()
+  storage.mockResolvedValue({ deleted: 1, failed: [] })
+  prepareChat.mockResolvedValue({ execute: executeChat })
+})
+
+describe('requested cleanup stages', () => {
+  const targets = [
+    ['workflowLogs', schemaMock.workflowExecutionLogs],
+    ['jobLogs', schemaMock.jobExecutionLogs],
+    ['largeValues', schemaMock.executionLargeValues],
+    ['legacyLargeValues', schemaMock.workspaceFiles],
+    ['orphanSnapshots', schemaMock.workflowExecutionSnapshots],
+    ['workflows', schemaMock.workflow],
+    ['chats', schemaMock.copilotChats],
+    ['legacyFiles', schemaMock.workspaceFile],
+    ['files', schemaMock.workspaceFiles],
+    ['knowledgeBases', schemaMock.knowledgeBase],
+    ['folders', schemaMock.folder],
+    ['userTables', schemaMock.userTableDefinitions],
+    ['memories', schemaMock.memory],
+    ['mcpServers', schemaMock.mcpServers],
+    ['workflowMcpServers', schemaMock.workflowMcpServer],
+    ['orphanKnowledgeBaseBindings', schemaMock.workspaceFiles],
+  ] as const
+  it.each(targets)(
+    'dry run of %s selects only that stage and has no side effects',
+    async (type, table) => {
+      queueTableRows(table, [{ id: 'root-one', key: 'key-one', files: [{ key: 'attached-file' }] }])
+      const run = control(type)
+      const runner =
+        targets.findIndex(([candidate]) => candidate === type) < 5
+          ? runBoundedLogScope
+          : runBoundedSoftDeleteScope
+      await runner(scope, run)
+      expect(run.progress.stages[type]?.selected).toBe(1)
+      expect(Object.keys(run.progress.stages)).toEqual([type])
+      expect(dbChainMockFns.delete).not.toHaveBeenCalled()
+      expect(dbChainMockFns.update).not.toHaveBeenCalled()
+      for (const effect of [storage, prepareChat, hardDelete, billing, decrement, reRoot])
+        expect(effect).not.toHaveBeenCalled()
+    }
+  )
+  it.each(['staleReferences', 'staleDependencies', 'largeValueTombstones'] as const)(
+    'dry run of %s uses SELECT only',
+    async (type) => {
+      dbChainMockFns.execute.mockResolvedValueOnce([{ id: 'metadata-one' }])
+      const run = control(type)
+      await runBoundedLogScope(scope, run)
+      expect(run.progress.stages[type]?.selected).toBe(1)
+      expect(dbChainMockFns.execute).toHaveBeenCalledTimes(1)
+      expect(dbChainMockFns.execute.mock.calls[0][0].strings.join('')).toMatch(/^SELECT /)
+    }
+  )
+  it('keeps the same log budget across workspace chunks', async () => {
+    const ids = Array.from({ length: 51 }, (_, index) => `ws-${index}`)
+    queueTableRows(schemaMock.jobExecutionLogs, [{ id: 'one' }])
+    queueTableRows(schemaMock.jobExecutionLogs, [])
+    queueTableRows(schemaMock.jobExecutionLogs, [{ id: 'two' }])
+    dbChainMockFns.returning.mockResolvedValue([{ id: 'deleted' }])
+    const run = control('jobLogs', false, 2)
+    await runBoundedLogScope({ ...scope, workspaceIds: ids }, run)
+    expect(run.progress.stages.jobLogs).toMatchObject({ selected: 2, deleted: 2 })
+    expect(dbChainMockFns.delete).toHaveBeenCalledTimes(2)
+  })
+  it('stops before root deletion if attached log storage fails', async () => {
+    queueTableRows(schemaMock.workflowExecutionLogs, [{ id: 'one', files: [{ key: 'blob' }] }])
+    storage.mockResolvedValue({ deleted: 0, failed: [{ key: 'blob', error: 'unavailable' }] })
+    const run = control('workflowLogs', false)
+    await expect(runBoundedLogScope(scope, run)).rejects.toThrow('storage deletions failed')
+    expect(dbChainMockFns.delete).not.toHaveBeenCalled()
+    expect(run.progress.stages.workflowLogs).toMatchObject({
+      selected: 1,
+      deleted: 0,
+      filesFailed: 1,
+    })
+  })
+  it('keeps workflow chat side effects even with a zero chat budget', async () => {
+    queueTableRows(schemaMock.workflow, [{ id: 'workflow-one' }])
+    queueTableRows(schemaMock.copilotChats, [{ id: 'child-chat' }])
+    queueTableRows(schemaMock.copilotChats, [])
+    dbChainMockFns.returning.mockResolvedValue([{ id: 'workflow-one' }])
+    const run = control('workflows', false)
+    await runBoundedSoftDeleteScope(scope, run)
+    expect(prepareChat).toHaveBeenCalledWith(
+      ['child-chat'],
+      'test',
+      expect.objectContaining({ type: 'workflows' })
+    )
+    expect(executeChat).toHaveBeenCalledTimes(1)
+    expect(run.progress.stages.workflows?.deleted).toBe(1)
+    expect(run.progress.stages.chats).toBeUndefined()
+  })
+  it('counts committed workflow deletion when backend cleanup subsequently fails', async () => {
+    queueTableRows(schemaMock.workflow, [{ id: 'workflow-one' }])
+    dbChainMockFns.returning.mockResolvedValue([{ id: 'workflow-one' }])
+    executeChat.mockRejectedValueOnce(new Error('backend failure'))
+    const run = control('workflows', false)
+    await expect(runBoundedSoftDeleteScope(scope, run)).rejects.toThrow('backend failure')
+    expect(run.progress.stages.workflows?.deleted).toBe(1)
+  })
+  it('applies the shared file budget across workspace and organization scopes', async () => {
+    const run = control('files', true, 2)
+    queueTableRows(schemaMock.workspaceFiles, [{ id: 'workspace-file' }])
+    queueTableRows(schemaMock.workspaceFiles, [])
+    await runBoundedSoftDeleteScope(scope, run)
+    queueTableRows(schemaMock.workspaceFiles, [{ id: 'organization-file' }])
+    await runBoundedSoftDeleteScope(
+      { ...scope, workspaceIds: [], organizationIds: ['org-one'] },
+      run
+    )
+    expect(run.progress.stages.files?.selected).toBe(2)
+    expect(storage).not.toHaveBeenCalled()
+  })
+})
+
+describe('bounded file billing', () => {
+  it('rechecks the exact payer workspace and decrements only bytes actually deleted', async () => {
+    queueTableRows(schemaMock.workspaceFiles, [
+      { id: 'file-one', key: 'blob', context: 'workspace', workspaceId: 'ws-one', sizeBytes: 100 },
+    ])
+    billing.mockResolvedValue({ workspaceId: 'ws-one' })
+    dbChainMockFns.returning.mockResolvedValue([{ id: 'file-one', sizeBytes: 40 }])
+    const run = control('files', false)
+    await runBoundedSoftDeleteScope({ ...scope, workspaceIds: ['ws-one', 'ws-two'] }, run)
+    expect(decrement).toHaveBeenCalledWith(
+      expect.anything(),
+      expect.objectContaining({ workspaceId: 'ws-one' }),
+      40
+    )
+    expect(
+      dbChainMockFns.where.mock.calls.some(([predicate]) =>
+        hasMockCondition(
+          predicate,
+          (condition) =>
+            condition.type === 'eq' &&
+            condition.left === schemaMock.workspaceFiles.workspaceId &&
+            condition.right === 'ws-one'
+        )
+      )
+    ).toBe(true)
+    expect(run.progress.stages.files?.deleted).toBe(1)
+  })
+  it('validates canonical sizes before deleting any storage', async () => {
+    queueTableRows(schemaMock.workspaceFiles, [
+      { id: 'file-one', key: 'blob', context: 'workspace', workspaceId: 'ws-one', sizeBytes: null },
+    ])
+    await expect(runBoundedSoftDeleteScope(scope, control('files', false))).rejects.toThrow(
+      'canonical size_bytes'
+    )
+    expect(storage).not.toHaveBeenCalled()
+    expect(dbChainMockFns.delete).not.toHaveBeenCalled()
+  })
+})