fix(db): bound log stats/export and KB cleanup queries for 60s statement_timeout

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

@@ -85,22 +85,38 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       })
     }
 
+    const pageSize = 1000
+
+    /**
+     * Fetches one page via keyset pagination on (startedAt, id): each page is an
+     * index seek past the last emitted row, where OFFSET would re-scan and discard
+     * every prior row (O(N²) across the stream, blowing the 60s statement_timeout
+     * on deep pages).
+     */
+    const fetchPage = (cursor: { startedAt: Date; id: string } | null) => {
+      const pageConditions = cursor
+        ? and(
+            conditions,
+            sql`(${workflowExecutionLogs.startedAt}, ${workflowExecutionLogs.id}) < (${cursor.startedAt}, ${cursor.id})`
+          )
+        : conditions
+      return dbReplica
+        .select(selectColumns)
+        .from(workflowExecutionLogs)
+        .leftJoin(workflow, eq(workflowExecutionLogs.workflowId, workflow.id))
+        .where(pageConditions)
+        .orderBy(desc(workflowExecutionLogs.startedAt), desc(workflowExecutionLogs.id))
+        .limit(pageSize)
+    }
+
     const encoder = new TextEncoder()
     const stream = new ReadableStream<Uint8Array>({
       start: async (controller) => {
         controller.enqueue(encoder.encode(`${header}\n`))
-        const pageSize = 1000
-        let offset = 0
+        let cursor: { startedAt: Date; id: string } | null = null
         try {
           while (true) {
-            const rows = await dbReplica
-              .select(selectColumns)
-              .from(workflowExecutionLogs)
-              .leftJoin(workflow, eq(workflowExecutionLogs.workflowId, workflow.id))
-              .where(conditions)
-              .orderBy(desc(workflowExecutionLogs.startedAt))
-              .limit(pageSize)
-              .offset(offset)
+            const rows = await fetchPage(cursor)
 
             if (!rows.length) break
 
@@ -157,7 +173,8 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
               controller.enqueue(encoder.encode(`${line}\n`))
             }
 
-            offset += pageSize
+            const last = rows[rows.length - 1]
+            cursor = { startedAt: last.startedAt, id: last.id }
           }
           controller.close()
         } catch (e: any) {

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

@@ -1,7 +1,7 @@
 import { dbReplica } from '@sim/db'
 import { workflow, workflowExecutionLogs } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
-import { and, eq, sql } from 'drizzle-orm'
+import { and, eq, gte, lte, sql } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import {
   type DashboardStatsResponse,
@@ -19,6 +19,13 @@ import { checkWorkspaceAccess } from '@/lib/workspaces/permissions/utils'
 
 const logger = createLogger('LogsStatsAPI')
 
+/**
+ * Hard default window for open-ended stats requests. Production roles run with
+ * a 60s statement_timeout, so the aggregations must never scan a workspace's
+ * entire log history.
+ */
+const DEFAULT_STATS_WINDOW_MS = 30 * 24 * 60 * 60 * 1000
+
 export const revalidate = 0
 
 export const GET = withRouteHandler(async (request: NextRequest) => {
@@ -63,19 +70,37 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       }
 
       const commonFilters = buildFilterConditions(params, { useSimpleLevelFilter: true })
-      const whereCondition = commonFilters ? and(workspaceFilter, commonFilters) : workspaceFilter
 
-      const boundsQuery = await dbReplica
-        .select({
-          minTime: sql<string>`MIN(${workflowExecutionLogs.startedAt})`,
-          maxTime: sql<string>`MAX(${workflowExecutionLogs.startedAt})`,
-        })
-        .from(workflowExecutionLogs)
-        .leftJoin(workflow, eq(workflowExecutionLogs.workflowId, workflow.id))
-        .where(whereCondition)
+      const now = new Date()
+      const windowEnd = params.endDate ? new Date(params.endDate) : now
+      const windowStart = params.startDate
+        ? new Date(params.startDate)
+        : new Date(windowEnd.getTime() - DEFAULT_STATS_WINDOW_MS)
+      const windowFilter = and(
+        gte(workflowExecutionLogs.startedAt, windowStart),
+        lte(workflowExecutionLogs.startedAt, windowEnd)
+      )
+      const whereCondition = commonFilters
+        ? and(workspaceFilter, windowFilter, commonFilters)
+        : and(workspaceFilter, windowFilter)
+
+      // The workflow join only matters when a filter references workflow columns.
+      const needsWorkflowJoin = Boolean(
+        params.workflowIds || params.folderIds || params.workflowName || params.folderName
+      )
+      const boundsSelection = {
+        minTime: sql<string>`MIN(${workflowExecutionLogs.startedAt})`,
+        maxTime: sql<string>`MAX(${workflowExecutionLogs.startedAt})`,
+      }
+      const boundsQuery = needsWorkflowJoin
+        ? await dbReplica
+            .select(boundsSelection)
+            .from(workflowExecutionLogs)
+            .leftJoin(workflow, eq(workflowExecutionLogs.workflowId, workflow.id))
+            .where(whereCondition)
+        : await dbReplica.select(boundsSelection).from(workflowExecutionLogs).where(whereCondition)
 
       const bounds = boundsQuery[0]
-      const now = new Date()
 
       let startTime: Date
       let endTime: Date

apps/sim/app/api/table/[tableId]/export/route.test.ts

@@ -66,13 +66,18 @@ describe('table export route — id→name translation', () => {
     })
     mockCheckAccess.mockResolvedValue({ ok: true, table: buildTable() })
     // Row data is keyed by stable column id (`col_email`), not the display name.
-    mockQueryRows.mockResolvedValue({
-      rows: [{ id: 'r1', data: { col_email: 'a@b.c', legacy: 'x' }, executions: {}, position: 0 }],
-      rowCount: 1,
-      totalCount: 1,
-      limit: 1000,
-      offset: 0,
-    })
+    // The export loop terminates on an empty page, so the mock must drain.
+    mockQueryRows
+      .mockResolvedValueOnce({
+        rows: [
+          { id: 'r1', data: { col_email: 'a@b.c', legacy: 'x' }, executions: {}, position: 0 },
+        ],
+        rowCount: 1,
+        totalCount: 1,
+        limit: 1000,
+        offset: 0,
+      })
+      .mockResolvedValue({ rows: [], rowCount: 0, totalCount: 1, limit: 1000, offset: 0 })
   })
 
   it('CSV: header uses display names and cell values resolve from id-keyed data', async () => {
@@ -93,3 +98,71 @@ describe('table export route — id→name translation', () => {
     expect(JSON.stringify(parsed)).not.toContain('col_email')
   })
 })
+
+describe('table export route — keyset pagination', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    hybridAuthMockFns.mockCheckSessionOrInternalAuth.mockResolvedValue({
+      success: true,
+      userId: 'user-1',
+      authType: 'session',
+    })
+    mockCheckAccess.mockResolvedValue({ ok: true, table: buildTable() })
+  })
+
+  it('drives the after cursor from the last row instead of offset paging', async () => {
+    const page = (ids: string[]) => ({
+      rows: ids.map((id, i) => ({
+        id,
+        data: { col_email: `${id}@x`, legacy: 'x' },
+        executions: {},
+        position: i,
+        orderKey: `k-${id}`,
+      })),
+      rowCount: ids.length,
+      totalCount: null,
+      limit: 1000,
+      offset: 0,
+    })
+    mockQueryRows
+      .mockResolvedValueOnce(page(['r1']))
+      .mockResolvedValueOnce(page(['r2']))
+      .mockResolvedValue(page([]))
+
+    const res = await callGet('csv')
+    expect(res.status).toBe(200)
+    const body = await res.text()
+    expect(body.trim().split('\n')).toEqual(['email,legacy', 'r1@x,x', 'r2@x,x'])
+
+    expect(mockQueryRows).toHaveBeenCalledTimes(3)
+    expect(mockQueryRows.mock.calls[0][1]).toMatchObject({ after: undefined, includeTotal: false })
+    expect(mockQueryRows.mock.calls[1][1]).toMatchObject({ after: { orderKey: 'k-r1', id: 'r1' } })
+    expect(mockQueryRows.mock.calls[2][1]).toMatchObject({ after: { orderKey: 'k-r2', id: 'r2' } })
+  })
+
+  it('falls back to offset paging for legacy rows without an order key', async () => {
+    const legacyPage = (ids: string[]) => ({
+      rows: ids.map((id, i) => ({
+        id,
+        data: { col_email: `${id}@x`, legacy: 'x' },
+        executions: {},
+        position: i,
+      })),
+      rowCount: ids.length,
+      totalCount: null,
+      limit: 1000,
+      offset: 0,
+    })
+    mockQueryRows
+      .mockResolvedValueOnce(legacyPage(['r1']))
+      .mockResolvedValueOnce(legacyPage(['r2']))
+      .mockResolvedValue(legacyPage([]))
+
+    const res = await callGet('csv')
+    expect(res.status).toBe(200)
+
+    expect(mockQueryRows).toHaveBeenCalledTimes(3)
+    expect(mockQueryRows.mock.calls[1][1]).toMatchObject({ after: undefined, offset: 1 })
+    expect(mockQueryRows.mock.calls[2][1]).toMatchObject({ after: undefined, offset: 2 })
+  })
+})

apps/sim/app/api/table/[tableId]/export/route.ts

@@ -8,6 +8,7 @@ import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { buildNameById, getColumnId, rowDataIdToName } from '@/lib/table/column-keys'
 import { queryRows } from '@/lib/table/rows/service'
+import type { TableRowsCursor } from '@/lib/table/types'
 import { accessError, checkAccess } from '@/app/api/table/utils'
 
 const logger = createLogger('TableExport')
@@ -65,15 +66,23 @@ export const GET = withRouteHandler(async (request: NextRequest, { params }: Rou
           controller.enqueue(encoder.encode('['))
         }
 
+        // Keyset pagination on the default (order_key, id) order: `after` makes each
+        // page an index seek, where OFFSET re-scans every prior row (O(N²) across a
+        // full drain, blowing the 60s statement_timeout on large tables). Legacy rows
+        // without an order key fall back to offset paging, mirroring the client's
+        // cursor derivation in getNextTableRowsPageParam.
+        let after: TableRowsCursor | undefined
         let offset = 0
         let firstJsonRow = true
         while (true) {
           const result = await queryRows(
             table,
-            { limit: EXPORT_BATCH_SIZE, offset, includeTotal: false },
+            { limit: EXPORT_BATCH_SIZE, after, offset, includeTotal: false },
             requestId
           )
 
+          if (result.rows.length === 0) break
+
           for (const row of result.rows) {
             if (format === 'csv') {
               const values = columns.map((c) => formatCsvValue(row.data[getColumnId(c)]))
@@ -87,7 +96,8 @@ export const GET = withRouteHandler(async (request: NextRequest, { params }: Rou
             }
           }
 
-          if (result.rows.length < EXPORT_BATCH_SIZE) break
+          const last = result.rows[result.rows.length - 1]
+          after = last.orderKey ? { orderKey: last.orderKey, id: last.id } : undefined
           offset += result.rows.length
         }
 

apps/sim/background/cleanup-soft-deletes.test.ts

@@ -28,10 +28,15 @@ const {
   const mockSelect = vi.fn(() => ({ from: mockFrom }))
 
   return {
-    mockBatchDeleteByWorkspaceAndTimestamp: vi.fn(async () => ({ deleted: 0, failed: 0 })),
+    mockBatchDeleteByWorkspaceAndTimestamp: vi.fn(
+      async (_opts: {
+        tableName: string
+        onBatch?: (rows: Array<{ id: string }>) => Promise<void>
+      }) => ({ deleted: 0, failed: 0 })
+    ),
     mockDeleteFileMetadata: vi.fn(async () => true),
     mockDeleteFiles: vi.fn(async () => ({ deleted: 0, failed: [] as Array<{ key: string }> })),
-    mockDeleteRowsById: vi.fn(async () => ({ deleted: 0, failed: 0 })),
+    mockDeleteRowsById: vi.fn(async (..._args: unknown[]) => ({ deleted: 0, failed: 0 })),
     mockIsUsingCloudStorage: vi.fn(() => true),
     mockLimit,
     mockOrderBy,
@@ -53,10 +58,12 @@ vi.mock('@sim/db/schema', () => {
   return {
     copilotChats: table(['id', 'workflowId']),
     document: table(['id', 'storageKey', 'knowledgeBaseId']),
+    embedding: table(['id', 'knowledgeBaseId', 'documentId']),
     knowledgeBase: table(softCols),
     mcpServers: table(softCols),
     memory: table(softCols),
     userTableDefinitions: table(softCols),
+    userTableRows: table(['id', 'tableId']),
     workflow: table(softCols),
     workflowFolder: table(softCols),
     workflowMcpServer: table(softCols),
@@ -83,6 +90,8 @@ vi.mock('drizzle-orm', () => ({
 }))
 
 vi.mock('@/lib/cleanup/batch-delete', () => ({
+  DEFAULT_BATCH_SIZE: 2000,
+  DEFAULT_WORKSPACE_CHUNK_SIZE: 50,
   batchDeleteByWorkspaceAndTimestamp: mockBatchDeleteByWorkspaceAndTimestamp,
   chunkArray: (items: string[], size: number) => {
     const chunks: string[][] = []
@@ -161,3 +170,78 @@ describe('cleanup soft deletes — orphan KB binding sweep', () => {
     expect(mockDeleteFileMetadata).not.toHaveBeenCalled()
   })
 })
+
+describe('cleanup soft deletes — cascade pre-drain', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mockIsUsingCloudStorage.mockReturnValue(true)
+    mockLimit.mockResolvedValue([])
+  })
+
+  /** Runs the job once and returns the onBatch hook wired for the given target. */
+  async function captureOnBatch(name: string) {
+    await runCleanupSoftDeletes(basePayload)
+    const call = mockBatchDeleteByWorkspaceAndTimestamp.mock.calls.find(
+      ([opts]) => opts.tableName === `free/1/${name}`
+    )
+    expect(call).toBeDefined()
+    return call![0].onBatch
+  }
+
+  it('knowledgeBase drains embedding then document rows before the KB delete', async () => {
+    const onBatch = await captureOnBatch('knowledgeBase')
+    expect(onBatch).toBeDefined()
+
+    vi.clearAllMocks()
+    mockDeleteRowsById.mockResolvedValue({ deleted: 1, failed: 0 })
+    // embedding: one page then drained; document: one page then drained.
+    mockLimit
+      .mockResolvedValueOnce([{ id: 'emb-1' }] as never)
+      .mockResolvedValueOnce([])
+      .mockResolvedValueOnce([{ id: 'doc-1' }] as never)
+      .mockResolvedValueOnce([])
+
+    await onBatch!([{ id: 'kb-1' }])
+
+    expect(mockDeleteRowsById).toHaveBeenCalledTimes(2)
+    expect(mockDeleteRowsById.mock.calls[0][2]).toEqual(['emb-1'])
+    expect(mockDeleteRowsById.mock.calls[0][3]).toBe('free/1/knowledgeBase/embedding')
+    expect(mockDeleteRowsById.mock.calls[1][2]).toEqual(['doc-1'])
+    expect(mockDeleteRowsById.mock.calls[1][3]).toBe('free/1/knowledgeBase/document')
+  })
+
+  it('userTableDefinitions drains user_table_rows before the definition delete', async () => {
+    const onBatch = await captureOnBatch('userTableDefinitions')
+    expect(onBatch).toBeDefined()
+
+    vi.clearAllMocks()
+    mockDeleteRowsById.mockResolvedValue({ deleted: 1, failed: 0 })
+    mockLimit.mockResolvedValueOnce([{ id: 'row-1' }] as never).mockResolvedValueOnce([])
+
+    await onBatch!([{ id: 'tbl-1' }])
+
+    expect(mockDeleteRowsById).toHaveBeenCalledTimes(1)
+    expect(mockDeleteRowsById.mock.calls[0][2]).toEqual(['row-1'])
+    expect(mockDeleteRowsById.mock.calls[0][3]).toBe('free/1/userTableDefinitions/userTableRows')
+  })
+
+  it('throws when child deletion makes no progress so the parent delete is skipped', async () => {
+    const onBatch = await captureOnBatch('knowledgeBase')
+
+    vi.clearAllMocks()
+    mockDeleteRowsById.mockResolvedValue({ deleted: 0, failed: 1 })
+    mockLimit.mockResolvedValue([{ id: 'emb-stuck' }] as never)
+
+    await expect(onBatch!([{ id: 'kb-1' }])).rejects.toThrow(/no progress/)
+    expect(mockDeleteRowsById).toHaveBeenCalledTimes(1)
+  })
+
+  it('targets without a large cascade pass no onBatch hook', async () => {
+    await runCleanupSoftDeletes(basePayload)
+    const call = mockBatchDeleteByWorkspaceAndTimestamp.mock.calls.find(
+      ([opts]) => opts.tableName === 'free/1/memory'
+    )
+    expect(call).toBeDefined()
+    expect(call![0].onBatch).toBeUndefined()
+  })
+})