feat(cleanup): add per-type row limits to existing jobs (#7842)

* feat(cleanup): add bounded manual retention runs

* fix(cleanup): coordinate deletion with resource eligibility

* fix(cleanup): persist storage retries and complete test fixtures

* fix(cleanup): guard storage retries by file generation

* improvement(cleanup): reuse existing jobs for row limits

* fix(cleanup): fail manual jobs on owner lookup errors

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,72 @@
+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({ triggered: true, runId: 'run-one', limits: { [limit]: 2 } })
+      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`))
+      expect(response.status).toBe(202)
+      expect(bounded).toHaveBeenCalledWith(type, { [limit]: 2 })
+      expect(await response.json()).toEqual({
+        triggered: true,
+        runId: 'run-one',
+        limits: { [limit]: 2 },
+      })
+      expect(scheduled).not.toHaveBeenCalled()
+    })
+    it.each(['?dryRun=true', '?unknown=1', '?batchSize=3', `?${limit}=2&${limit}=3`])(
+      '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`))).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-logs.test.ts

@@ -45,9 +45,12 @@ const {
   mockTask: vi.fn((config: unknown) => config),
 }))
 
-vi.mock('@trigger.dev/sdk', () => ({ task: mockTask }))
+vi.mock('@trigger.dev/sdk', () => ({ task: mockTask, queue: vi.fn((config) => config) }))
+
+vi.mock('@/lib/billing/cleanup-dispatcher', () => ({ runCleanupWithLimits: vi.fn() }))
 
 vi.mock('@/lib/cleanup/batch-delete', () => ({
+  consumeRowBudget: vi.fn(),
   batchDeleteByWorkspaceAndTimestamp: mockBatchDeleteByWorkspaceAndTimestamp,
   chunkedBatchDelete: mockChunkedBatchDelete,
 }))
@@ -199,7 +202,7 @@ describe('cleanup logs worker', () => {
 
   it('caps Trigger.dev concurrency for log cleanup tasks', () => {
     expect(cleanupLogsTask).toMatchObject({
-      queue: { concurrencyLimit: 2 },
+      queue: { name: 'retention-cleanup', concurrencyLimit: 1 },
     })
   })
 })