fix(logs): guard export stream cancellation

Author: waleedlatif1

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

@@ -214,4 +214,26 @@ describe('GET /api/logs/export', () => {
     await reader.cancel()
     expect(dbChainMockFns.where).toHaveBeenCalledTimes(1)
   })
+
+  it('stops a pending pull cleanly when the reader cancels', async () => {
+    queueTableRows(workflowExecutionLogs, [logRow(0)])
+    let resolveMaterialization: ((value: unknown[]) => void) | undefined
+    mockMapWithConcurrency.mockImplementationOnce(
+      () =>
+        new Promise((resolve) => {
+          resolveMaterialization = resolve
+        })
+    )
+
+    const response = await GET(makeRequest())
+    const reader = response.body!.getReader()
+    await reader.read()
+
+    const pendingRead = reader.read()
+    await vi.waitFor(() => expect(mockMapWithConcurrency).toHaveBeenCalledTimes(1))
+    const cancellation = reader.cancel()
+    resolveMaterialization?.([{ message: 'message-0' }])
+
+    await expect(Promise.all([pendingRead, cancellation])).resolves.toBeDefined()
+  })
 })

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

@@ -177,22 +177,26 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       }
     })()
 
+    let cancelled = false
     const stream = new ReadableStream<Uint8Array>(
       {
         pull: async (controller) => {
           try {
             const next = await csvChunks.next()
+            if (cancelled) return
             if (next.done) {
               controller.close()
               return
             }
             controller.enqueue(next.value)
           } catch (error) {
+            if (cancelled) return
             logger.error('Export stream error', { error: getErrorMessage(error) })
             controller.error(error)
           }
         },
         cancel: async () => {
+          cancelled = true
           await csvChunks.return(undefined)
         },
       },