perf(logs): batch keyset export reads

Author: waleedlatif1

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

@@ -104,9 +104,7 @@ describe('GET /api/logs/export', () => {
   })
 
   it('materializes one bounded page at a time while preserving CSV row order', async () => {
-    queueTableRows(workflowExecutionLogs, [logRow(0)])
-    queueTableRows(workflowExecutionLogs, [logRow(1)])
-    queueTableRows(workflowExecutionLogs, [logRow(2)])
+    queueTableRows(workflowExecutionLogs, [logRow(0), logRow(1), logRow(2)])
 
     const response = await GET(makeRequest())
     const lines = (await response.text()).trimEnd().split('\n')
@@ -119,23 +117,25 @@ describe('GET /api/logs/export', () => {
   })
 
   it('resumes full pages by startedAt and id without using OFFSET', async () => {
-    const last = logRow(0, { startedAtCursor: '2026-08-23 12:00:00.000123' })
+    const firstPage = Array.from({ length: 100 }, (_, index) => logRow(index))
+    firstPage[99] = logRow(99, { startedAtCursor: '2026-08-23 11:58:21.000123' })
+    const last = firstPage.at(-1)!
     const secondPage = [
-      logRow(1, {
+      logRow(100, {
         id: 'log-0000-second',
         startedAt: last.startedAt,
-        startedAtCursor: '2026-08-23 12:00:00.000122',
+        startedAtCursor: '2026-08-23 11:58:21.000122',
       }),
     ]
-    queueTableRows(workflowExecutionLogs, [last])
+    queueTableRows(workflowExecutionLogs, firstPage)
     queueTableRows(workflowExecutionLogs, secondPage)
 
     const response = await GET(makeRequest())
     const lines = (await response.text()).trimEnd().split('\n')
 
-    expect(lines).toHaveLength(3)
+    expect(lines).toHaveLength(102)
     expect(dbChainMockFns.offset).not.toHaveBeenCalled()
-    expect(dbChainMockFns.where).toHaveBeenCalledTimes(3)
+    expect(dbChainMockFns.where).toHaveBeenCalledTimes(2)
     expect(dbChainMockFns.orderBy).toHaveBeenNthCalledWith(
       1,
       expect.objectContaining({
@@ -169,7 +169,10 @@ describe('GET /api/logs/export', () => {
   })
 
   it('does not load the next database page until the current row is consumed', async () => {
-    queueTableRows(workflowExecutionLogs, [logRow(0)])
+    queueTableRows(
+      workflowExecutionLogs,
+      Array.from({ length: 100 }, (_, index) => logRow(index))
+    )
     queueTableRows(workflowExecutionLogs, [logRow(1)])
 
     const response = await GET(makeRequest())

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

@@ -5,7 +5,7 @@ import { getErrorMessage } from '@sim/utils/errors'
 import { and, desc, eq, lt, or, type SQL, sql } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import { getSession } from '@/lib/auth'
-import { MATERIALIZE_CONCURRENCY, mapWithConcurrency } from '@/lib/core/utils/concurrency'
+import { mapWithConcurrency } from '@/lib/core/utils/concurrency'
 import { formatCsvValue, toCsvRow } from '@/lib/core/utils/csv'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { materializeExecutionDataForDisplay } from '@/lib/logs/execution/trace-store'
@@ -14,7 +14,8 @@ import { expandFolderIdsWithDescendants } from '@/lib/logs/folder-expansion'
 import { checkWorkspaceAccess } from '@/lib/workspaces/permissions/utils'
 
 const logger = createLogger('LogsExportAPI')
-const LOG_EXPORT_PAGE_SIZE = 1
+const LOG_EXPORT_PAGE_SIZE = 100
+const LOG_EXPORT_MATERIALIZE_CONCURRENCY = 1
 
 export const revalidate = 0
 
@@ -119,18 +120,25 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
 
         if (!rows.length) break
 
-        for (let chunkStart = 0; chunkStart < rows.length; chunkStart += MATERIALIZE_CONCURRENCY) {
-          const chunk = rows.slice(chunkStart, chunkStart + MATERIALIZE_CONCURRENCY)
-          const materialized = await mapWithConcurrency(chunk, MATERIALIZE_CONCURRENCY, (row) =>
-            materializeExecutionDataForDisplay(
-              row.executionData as Record<string, unknown> | null,
-              {
-                workspaceId: params.workspaceId,
-                workflowId: row.workflowId,
-                executionId: row.executionId,
-                userId: session.user.id,
-              }
-            )
+        for (
+          let chunkStart = 0;
+          chunkStart < rows.length;
+          chunkStart += LOG_EXPORT_MATERIALIZE_CONCURRENCY
+        ) {
+          const chunk = rows.slice(chunkStart, chunkStart + LOG_EXPORT_MATERIALIZE_CONCURRENCY)
+          const materialized = await mapWithConcurrency(
+            chunk,
+            LOG_EXPORT_MATERIALIZE_CONCURRENCY,
+            (row) =>
+              materializeExecutionDataForDisplay(
+                row.executionData as Record<string, unknown> | null,
+                {
+                  workspaceId: params.workspaceId,
+                  workflowId: row.workflowId,
+                  executionId: row.executionId,
+                  userId: session.user.id,
+                }
+              )
           )
 
           for (let index = 0; index < chunk.length; index++) {