fix(files): restore CSV preview cancellation (#6596)

* fix(files): restore CSV preview cancellation

* fix

Author: icecrasher321

apps/sim/app/api/workspaces/[id]/files/[fileId]/csv-preview/route.test.ts

@@ -0,0 +1,81 @@
+/**
+ * @vitest-environment node
+ */
+import { authMockFns } from '@sim/testing'
+import { NextRequest } from 'next/server'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const mocks = vi.hoisted(() => ({
+  getSlice: vi.fn(),
+  readFile: vi.fn(),
+}))
+
+vi.mock('@/lib/file-parsers/csv-preview-slice', () => ({
+  getCsvPreviewSlice: mocks.getSlice,
+}))
+
+vi.mock('@/lib/workspace-files/application/read-workspace-file-record', () => ({
+  readWorkspaceFileContentRecord: {
+    operation: { id: 'files.read_content', minimumRole: 'read', workspaceApiKey: 'allow' },
+    execute: mocks.readFile,
+  },
+}))
+
+import { GET } from '@/app/api/workspaces/[id]/files/[fileId]/csv-preview/route'
+
+const WORKSPACE_ID = '7727ef3f-8cf6-4686-b063-2bb006a10785'
+const FILE_ID = 'wf_csv'
+const KEY = `workspace/${WORKSPACE_ID}/large.csv`
+const USER = { id: 'user-1' }
+const context = { params: Promise.resolve({ id: WORKSPACE_ID, fileId: FILE_ID }) }
+
+function request(signal?: AbortSignal): NextRequest {
+  return new NextRequest(
+    `http://localhost/api/workspaces/${WORKSPACE_ID}/files/${FILE_ID}/csv-preview?key=${encodeURIComponent(KEY)}`,
+    { signal }
+  )
+}
+
+describe('GET /api/workspaces/[id]/files/[fileId]/csv-preview', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    authMockFns.mockGetSession.mockResolvedValue({ user: USER, session: { id: 'session-1' } })
+    mocks.readFile.mockResolvedValue({ file: { id: FILE_ID, key: KEY } })
+    mocks.getSlice.mockResolvedValue({
+      headers: ['name'],
+      rows: [['Ada']],
+      truncated: false,
+    })
+  })
+
+  it('propagates client cancellation to the storage preview read', async () => {
+    const req = request()
+    const response = await GET(req, context)
+
+    expect(response.status).toBe(200)
+    expect(mocks.getSlice).toHaveBeenCalledWith({
+      key: KEY,
+      context: 'workspace',
+      signal: req.signal,
+    })
+  })
+
+  it('returns a cancellation response when the client disconnects during the storage read', async () => {
+    const controller = new AbortController()
+    const req = request(controller.signal)
+    mocks.getSlice.mockImplementation(async () => {
+      controller.abort()
+      throw Object.assign(new Error('Premature close'), {
+        code: 'ERR_STREAM_PREMATURE_CLOSE',
+      })
+    })
+
+    const response = await GET(req, context)
+
+    expect(response.status).toBe(499)
+    await expect(response.json()).resolves.toMatchObject({
+      error: 'Client cancelled request',
+      requestId: expect.any(String),
+    })
+  })
+})

apps/sim/app/api/workspaces/[id]/files/[fileId]/csv-preview/route.ts

@@ -15,10 +15,11 @@ export const GET = defineInternalJsonRoute({
   operation: csvPreviewWorkspaceFile.operation,
   rateLimit: internalRateLimits.none({ reason: 'Preserve existing internal CSV preview behavior' }),
   errorPolicy: internalFileErrorPolicies.default,
-  mapInput: ({ params, query }) => ({
+  mapInput: ({ params, query }, { request }) => ({
     fileId: params.fileId,
     assertedWorkspaceId: params.id,
     key: query.key,
+    signal: request.signal,
   }),
   useCase: csvPreviewWorkspaceFile,
   onSuccess: ({ result }) => {

apps/sim/lib/api/server/routes/internal-json-route.ts

@@ -365,6 +365,10 @@ export function defineInternalJsonRoute<
       }
     },
     {
+      clientAbortResponse: ({ requestId }) =>
+        createJsonErrorResponse(
+          internalErrorResponse(499, { error: 'Client cancelled request', requestId })
+        ),
       typedErrorResponse: ({ error, status, requestId }) =>
         NextResponse.json({ error: error.message, requestId }, { status }),
       unhandledErrorResponse: () =>

apps/sim/lib/api/server/routes/v2-json-route.test.ts

@@ -116,11 +116,12 @@ function createHandler(overrides: HandlerOverrides = {}) {
   })
 }
 
-function request(body: unknown = { value: 'ok' }): NextRequest {
+function request(body: unknown = { value: 'ok' }, signal?: AbortSignal): NextRequest {
   return new NextRequest('http://localhost/api/v2/widgets', {
     method: 'POST',
     headers: { 'content-type': 'application/json', 'x-api-key': 'secret' },
     body: JSON.stringify(body),
+    signal,
   })
 }
 
@@ -395,6 +396,25 @@ describe('defineV2JsonRoute', () => {
     expect(response.headers.get('X-RateLimit-Remaining')).toBe('99')
   })
 
+  it('renders a client disconnect through the v2 cancellation envelope', async () => {
+    const controller = new AbortController()
+    const response = await createHandler({
+      execute: async () => {
+        controller.abort()
+        throw Object.assign(new Error('Premature close'), {
+          code: 'ERR_STREAM_PREMATURE_CLOSE',
+        })
+      },
+    })(request(undefined, controller.signal))
+
+    expect(response.status).toBe(499)
+    await expect(response.json()).resolves.toEqual({
+      error: { code: 'CLIENT_CLOSED_REQUEST', message: 'Client cancelled request' },
+    })
+    expect(response.headers.get('Cache-Control')).toBe('private, no-store')
+    expect(response.headers.get('X-RateLimit-Remaining')).toBe('99')
+  })
+
   it('validates the presented response before onSuccess', async () => {
     const onSuccess = vi.fn()
     const response = await createHandler({

apps/sim/lib/api/server/routes/v2-json-route.ts

@@ -287,6 +287,7 @@ export function defineV2JsonRoute<
       }
     },
     {
+      clientAbortResponse: () => v2Error('CLIENT_CLOSED_REQUEST', 'Client cancelled request'),
       typedErrorResponse: ({ error }) => v2HttpError(error),
       unhandledErrorResponse: ({ error }) =>
         error instanceof V2RouteInfrastructureError

apps/sim/lib/core/utils/with-route-handler.test.ts

@@ -2,6 +2,7 @@
  * @vitest-environment node
  */
 
+import { loggerMock } from '@sim/testing'
 import { NextRequest, NextResponse } from 'next/server'
 import { describe, expect, it, vi } from 'vitest'
 import { HttpError } from '@/lib/core/utils/http-error'
@@ -17,6 +18,47 @@ class TestHttpError extends HttpError {
 }
 
 describe('withRouteHandler', () => {
+  it('classifies errors after a client disconnect without using the unhandled fallback', async () => {
+    const routeHandlerLogger = vi.mocked(loggerMock.createLogger).mock.results[
+      vi.mocked(loggerMock.createLogger).mock.calls.findIndex(([name]) => name === 'RouteHandler')
+    ]?.value
+    routeHandlerLogger?.info.mockClear()
+    routeHandlerLogger?.error.mockClear()
+
+    const controller = new AbortController()
+    const clientAbortResponse = vi.fn(() =>
+      NextResponse.json({ error: 'Client cancelled request' }, { status: 499 })
+    )
+    const unhandledErrorResponse = vi.fn(() =>
+      NextResponse.json({ error: 'Internal server error' }, { status: 500 })
+    )
+    const handler = withRouteHandler(
+      async () => {
+        controller.abort()
+        throw Object.assign(new Error('Premature close'), {
+          code: 'ERR_STREAM_PREMATURE_CLOSE',
+        })
+      },
+      { clientAbortResponse, unhandledErrorResponse }
+    )
+
+    const response = await handler(
+      new NextRequest('http://localhost/api/test', { signal: controller.signal }),
+      undefined
+    )
+
+    expect(response.status).toBe(499)
+    await expect(response.json()).resolves.toEqual({ error: 'Client cancelled request' })
+    expect(clientAbortResponse).toHaveBeenCalledOnce()
+    expect(unhandledErrorResponse).not.toHaveBeenCalled()
+    expect(routeHandlerLogger?.error).not.toHaveBeenCalled()
+    expect(routeHandlerLogger?.info).toHaveBeenCalledWith('Client closed request', {
+      duration: expect.any(Number),
+      status: 499,
+    })
+    expect(response.headers.get('x-request-id')).toBeTruthy()
+  })
+
   it('lets a route family render a typed error before its generic fallback', async () => {
     const unhandledErrorResponse = vi.fn(() =>
       NextResponse.json({ family: 'generic' }, { status: 500 })

apps/sim/lib/core/utils/with-route-handler.ts

@@ -25,6 +25,7 @@ interface RouteHandlerTypedErrorContext {
 }
 
 interface RouteHandlerOptions {
+  clientAbortResponse?: (context: RouteHandlerErrorContext) => NextResponse | Response
   typedErrorResponse?: (context: RouteHandlerTypedErrorContext) => NextResponse | Response
   unhandledErrorResponse?: (context: RouteHandlerErrorContext) => NextResponse | Response
 }
@@ -78,7 +79,9 @@ function applyResponseHeaders(
  * - Generates a unique request ID and stores it in AsyncLocalStorage so every
  *   logger in the request lifecycle automatically includes it
  * - Logs all 4xx and 5xx responses with method, path, status, duration
+ * - Classifies errors after a client disconnect as a normal 499 cancellation
  * - Catches unhandled errors, logs them, and returns a 500 with the request ID
+ * - Supports a route-family-specific client-abort response envelope
  * - Supports a route-family-specific unhandled-error response envelope
  * - Attaches `x-request-id`, plus the rate-limit headers when the route
  *   recorded a snapshot for the request
@@ -101,6 +104,15 @@ export function withRouteHandler<T>(
       } catch (error) {
         const duration = Date.now() - startTime
         const message = getErrorMessage(error, 'Unknown error')
+        if (request.signal.aborted) {
+          logger.info('Client closed request', { duration, status: 499 })
+          response = options.clientAbortResponse
+            ? options.clientAbortResponse({ error, requestId })
+            : new Response(null, { status: 499 })
+          applyResponseHeaders(response, request, requestId)
+          return response
+        }
+
         const typedError = readTypedError(error)
         if (typedError) {
           const typedStatus = typedError.statusCode

apps/sim/lib/file-parsers/csv-preview-slice.test.ts

@@ -100,4 +100,32 @@ describe('getCsvPreviewSlice', () => {
     expect(slice.truncated).toBe(true)
     expect(destroySpy).toHaveBeenCalled()
   })
+
+  it('destroys a source acquired after the request was already aborted', async () => {
+    const source = streamOf('a,b\n1,2\n')
+    const destroySpy = vi.spyOn(source, 'destroy')
+    const controller = new AbortController()
+    controller.abort()
+    mockDownloadFileStream.mockResolvedValue(source)
+
+    await expect(getCsvPreviewSlice({ ...args, signal: controller.signal })).rejects.toMatchObject({
+      name: 'AbortError',
+    })
+    expect(destroySpy).toHaveBeenCalled()
+  })
+
+  it('destroys an active source when the request is aborted', async () => {
+    const read = vi.fn()
+    const source = new Readable({ read })
+    const destroySpy = vi.spyOn(source, 'destroy')
+    const controller = new AbortController()
+    mockDownloadFileStream.mockResolvedValue(source)
+
+    const preview = getCsvPreviewSlice({ ...args, signal: controller.signal })
+    await vi.waitFor(() => expect(read).toHaveBeenCalled())
+    controller.abort()
+
+    await expect(preview).rejects.toMatchObject({ name: 'AbortError' })
+    expect(destroySpy).toHaveBeenCalled()
+  })
 })

apps/sim/lib/file-parsers/csv-preview-slice.ts

@@ -52,6 +52,10 @@ export async function getCsvPreviewSlice({
   signal,
 }: CsvPreviewSliceArgs): Promise<CsvPreviewSlice> {
   const source = await downloadFileStream({ key, context })
+  if (signal?.aborted) {
+    source.destroy()
+    signal.throwIfAborted()
+  }
   const onAbort = () => source.destroy()
   signal?.addEventListener('abort', onAbort, { once: true })
 
@@ -133,6 +137,9 @@ export async function getCsvPreviewSlice({
     piped.destroy()
     parser.destroy()
     return { headers, rows, truncated }
+  } catch (error) {
+    if (signal?.aborted) signal.throwIfAborted()
+    throw error
   } finally {
     signal?.removeEventListener('abort', onAbort)
     source.destroy()