diff --git a/apps/sim/app/api/logs/export/route.ts b/apps/sim/app/api/logs/export/route.ts index a2006538fd9..49c8887081b 100644 --- a/apps/sim/app/api/logs/export/route.ts +++ b/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({ 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) { diff --git a/apps/sim/app/api/logs/stats/route.ts b/apps/sim/app/api/logs/stats/route.ts index 88f33ff6b54..6072fc93d18 100644 --- a/apps/sim/app/api/logs/stats/route.ts +++ b/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`MIN(${workflowExecutionLogs.startedAt})`, - maxTime: sql`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`MIN(${workflowExecutionLogs.startedAt})`, + maxTime: sql`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 diff --git a/apps/sim/app/api/table/[tableId]/export/route.test.ts b/apps/sim/app/api/table/[tableId]/export/route.test.ts index b420b5f97da..6b2c304cecc 100644 --- a/apps/sim/app/api/table/[tableId]/export/route.test.ts +++ b/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 }) + }) +}) diff --git a/apps/sim/app/api/table/[tableId]/export/route.ts b/apps/sim/app/api/table/[tableId]/export/route.ts index 75208e3d982..532748261da 100644 --- a/apps/sim/app/api/table/[tableId]/export/route.ts +++ b/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 } diff --git a/apps/sim/background/cleanup-soft-deletes.test.ts b/apps/sim/background/cleanup-soft-deletes.test.ts index a6ede324481..41c4e9b8346 100644 --- a/apps/sim/background/cleanup-soft-deletes.test.ts +++ b/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 + }) => ({ 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() + }) +}) diff --git a/apps/sim/background/cleanup-soft-deletes.ts b/apps/sim/background/cleanup-soft-deletes.ts index ee01e297898..a7565260998 100644 --- a/apps/sim/background/cleanup-soft-deletes.ts +++ b/apps/sim/background/cleanup-soft-deletes.ts @@ -2,10 +2,12 @@ import { db } from '@sim/db' import { copilotChats, document, + embedding, knowledgeBase, mcpServers, memory, userTableDefinitions, + userTableRows, workflow, workflowFolder, workflowMcpServer, @@ -15,10 +17,13 @@ import { import { createLogger } from '@sim/logger' import { task } from '@trigger.dev/sdk' import { and, asc, eq, inArray, isNotNull, isNull, lt, sql } from 'drizzle-orm' +import type { PgColumn, PgTable } from 'drizzle-orm/pg-core' import type { CleanupJobPayload } from '@/lib/billing/cleanup-dispatcher' import { batchDeleteByWorkspaceAndTimestamp, chunkArray, + DEFAULT_BATCH_SIZE, + DEFAULT_WORKSPACE_CHUNK_SIZE, deleteRowsById, selectRowsByIdChunks, } from '@/lib/cleanup/batch-delete' @@ -128,13 +133,69 @@ async function cleanupWorkspaceFileStorage( return stats } +/** + * Deletes child rows referencing the given parent ids in bounded batches. + * Parents with huge `ON DELETE CASCADE` fan-outs (knowledge base → embeddings, + * table definition → rows) must have their children drained this way BEFORE the + * parent DELETE runs — one cascading statement can otherwise touch millions of + * rows and exceed the 60s role-level statement_timeout. + * + * Throws when a batch makes no progress so the caller's parent DELETE (and its + * unbounded cascade) is skipped for this run instead of timing out. + */ +async function deleteChildRowsInBatches( + childTable: PgTable, + childIdCol: PgColumn, + parentFkCol: PgColumn, + parentIds: string[], + label: string +): Promise { + let totalDeleted = 0 + for (const parentChunk of chunkArray(parentIds, DEFAULT_WORKSPACE_CHUNK_SIZE)) { + while (true) { + const rows = await db + .select({ id: sql`id` }) + .from(childTable) + .where(inArray(parentFkCol, parentChunk)) + .limit(DEFAULT_BATCH_SIZE) + if (rows.length === 0) break + + const result = await deleteRowsById( + childTable, + childIdCol, + rows.map((r) => r.id), + label + ) + totalDeleted += result.deleted + if (result.deleted === 0) { + throw new Error( + `[${label}] Child cleanup made no progress (${rows.length} rows selected, 0 deleted)` + ) + } + } + } + return totalDeleted +} + +interface CleanupTarget { + table: PgTable + softDeleteCol: PgColumn + wsCol: PgColumn + name: string + /** + * Drains child tables with a large `ON DELETE CASCADE` fan-out before the + * parent rows are deleted — see {@link deleteChildRowsInBatches}. + */ + prepareCascade?: (rows: Array<{ id: string }>, label: string) => Promise +} + /** * Tables cleaned by the generic workspace-scoped batched DELETE. Tables whose * hard-delete triggers external side effects (workflow → copilot chats cascade, * workspace files → S3 storage) are handled explicitly so the SELECT that drives * the external cleanup and the SELECT that drives the DB delete see the same rows. */ -const CLEANUP_TARGETS = [ +const CLEANUP_TARGETS: CleanupTarget[] = [ { table: workflowFolder, softDeleteCol: workflowFolder.archivedAt, @@ -146,12 +207,40 @@ const CLEANUP_TARGETS = [ softDeleteCol: knowledgeBase.deletedAt, wsCol: knowledgeBase.workspaceId, name: 'knowledgeBase', + prepareCascade: async (rows, label) => { + const kbIds = rows.map((r) => r.id) + // Embeddings first: they cascade from both knowledge_base and document, so + // draining them makes the subsequent document deletes cascade-free. + await deleteChildRowsInBatches( + embedding, + embedding.id, + embedding.knowledgeBaseId, + kbIds, + `${label}/embedding` + ) + await deleteChildRowsInBatches( + document, + document.id, + document.knowledgeBaseId, + kbIds, + `${label}/document` + ) + }, }, { table: userTableDefinitions, softDeleteCol: userTableDefinitions.archivedAt, wsCol: userTableDefinitions.workspaceId, name: 'userTableDefinitions', + prepareCascade: async (rows, label) => { + await deleteChildRowsInBatches( + userTableRows, + userTableRows.id, + userTableRows.tableId, + rows.map((r) => r.id), + `${label}/userTableRows` + ) + }, }, { table: memory, softDeleteCol: memory.deletedAt, wsCol: memory.workspaceId, name: 'memory' }, { @@ -166,7 +255,7 @@ const CLEANUP_TARGETS = [ wsCol: workflowMcpServer.workspaceId, name: 'workflowMcpServer', }, -] as const +] /** * Sweep abandoned knowledge-base ownership bindings. The presigned upload flow @@ -331,14 +420,17 @@ export async function runCleanupSoftDeletes(payload: CleanupJobPayload): Promise totalDeleted += multiContextFileResult.deleted for (const target of CLEANUP_TARGETS) { + const { prepareCascade } = target + const targetLabel = `${label}/${target.name}` const result = await batchDeleteByWorkspaceAndTimestamp({ tableDef: target.table, workspaceIdCol: target.wsCol, timestampCol: target.softDeleteCol, workspaceIds, retentionDate, - tableName: `${label}/${target.name}`, + tableName: targetLabel, requireTimestampNotNull: true, + onBatch: prepareCascade ? (rows) => prepareCascade(rows, targetLabel) : undefined, }) totalDeleted += result.deleted } diff --git a/apps/sim/lib/cleanup/batch-delete.ts b/apps/sim/lib/cleanup/batch-delete.ts index d523800cd59..0d3203f6cf1 100644 --- a/apps/sim/lib/cleanup/batch-delete.ts +++ b/apps/sim/lib/cleanup/batch-delete.ts @@ -186,6 +186,8 @@ export interface BatchDeleteOptions { tableName: string /** When true, also requires `timestampCol IS NOT NULL` (soft-delete semantics). */ requireTimestampNotNull?: boolean + /** Runs between SELECT and DELETE; receives the just-selected rows. A throw skips the DELETE. */ + onBatch?: (rows: Array<{ id: string }>) => Promise batchSize?: number maxBatches?: number workspaceChunkSize?: number