fix(api): classify transient database reads and client-caused failures instead of answering 500

apps/sim/app/api/files/serve/[...path]/route.ts

@@ -19,6 +19,7 @@ import type { StorageContext } from '@/lib/uploads/config'
 import { readOrganizationChatAttachment } from '@/lib/uploads/contexts/organization-assistant/application'
 import type { WorkspaceFileRecord } from '@/lib/uploads/contexts/workspace'
 import { parseWorkspaceFileKey } from '@/lib/uploads/contexts/workspace/file-manager'
+import { hasObjectNotFoundCause } from '@/lib/uploads/core/errors'
 import { downloadFile, downloadFileStream, headObject } from '@/lib/uploads/core/storage-service'
 import { resolveServableDocBytes } from '@/lib/uploads/documents/compile'
 import { DocCompileUserError } from '@/lib/uploads/documents/compile-error'
@@ -34,6 +35,7 @@ import {
   inferContextFromKey,
   isRenderableDocumentName,
   resolveEffectiveMimeType,
+  tryInferContextFromKey,
 } from '@/lib/uploads/utils/file'
 import { internalWorkspaceFileServeAuth } from '@/lib/workspace-files/api'
 import { fetchAuthorizedServableWorkspaceFileBuffer } from '@/lib/workspace-files/application/fetch-servable-workspace-file-buffer'
@@ -71,7 +73,11 @@ const logger = createLogger('FilesServeAPI')
  * its way out of, not something on call needs to look at.
  */
 function logServeFailure(message: string, error: unknown): void {
-  if (error instanceof FileNotFoundError || isPayloadSizeLimitError(error)) {
+  if (
+    error instanceof FileNotFoundError ||
+    hasObjectNotFoundCause(error) ||
+    isPayloadSizeLimitError(error)
+  ) {
     logger.info(message, { reason: getErrorMessage(error) })
     return
   }
@@ -389,6 +395,9 @@ export const GET = withRouteHandler(
         })
       }
 
+      /** No storage context owns a key without a known prefix, so no file can live there. */
+      if (!tryInferContextFromKey(cloudKey)) throw new FileNotFoundError('File not found')
+
       const isPublicByKeyPrefix =
         cloudKey.startsWith('profile-pictures/') ||
         cloudKey.startsWith('og-images/') ||
@@ -536,6 +545,10 @@ export const GET = withRouteHandler(
         return createFileErrorResponse(error)
       }
 
+      if (hasObjectNotFoundCause(error)) {
+        return createFileErrorResponse(new FileNotFoundError('File not found'))
+      }
+
       return createFileErrorResponse(
         error instanceof Error ? error : new Error('Failed to serve file')
       )

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

@@ -4,6 +4,7 @@ import { createLogger } from '@sim/logger'
 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 { validationErrorResponse } from '@/lib/api/server/validation'
 import { getSession } from '@/lib/auth'
 import { MATERIALIZE_CONCURRENCY, mapWithConcurrency } from '@/lib/core/utils/concurrency'
 import { formatCsvValue, toCsvLinePieces, toCsvRow } from '@/lib/core/utils/csv'
@@ -50,7 +51,9 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
 
     const userId = session.user.id
     const { searchParams } = new URL(request.url)
-    const params = LogFilterParamsSchema.parse(Object.fromEntries(searchParams.entries()))
+    const parsedParams = LogFilterParamsSchema.safeParse(Object.fromEntries(searchParams.entries()))
+    if (!parsedParams.success) return validationErrorResponse(parsedParams.error)
+    const params = parsedParams.data
 
     const selectColumns = {
       id: workflowExecutionLogs.id,

apps/sim/app/api/mothership/integrations/catalog/route.ts

@@ -1,9 +1,11 @@
 import { readIntegrationCatalogContract } from '@/lib/api/contracts/mothership-integrations'
 import {
   defineInternalJsonRoute,
+  internalErrorResponse,
   internalOrchestrationErrorPolicy,
   internalRateLimits,
 } from '@/lib/api/server/routes'
+import { categorizeMcpServerError, isMcpServerUnavailableError } from '@/lib/mcp/utils'
 import { internalCopilotAuth } from '@/lib/mothership/auth/internal'
 import {
   INTEGRATION_CATALOG_AUDIENCE,
@@ -18,7 +20,20 @@ export const POST = defineInternalJsonRoute({
   rateLimit: internalRateLimits.none({
     reason: 'Private bounded catalog discovery rechecks current authorization on every request.',
   }),
-  errorPolicy: internalOrchestrationErrorPolicy,
+  errorPolicy: {
+    /**
+     * A targeted MCP server that cannot be reached, or that answers `tools/list` with a
+     * protocol or HTTP error, is the server's failure, not Sim's.
+     */
+    project(error) {
+      const mcpFailure = categorizeMcpServerError(error)
+      if (mcpFailure) return internalErrorResponse(mcpFailure.status, { error: mcpFailure.message })
+      if (isMcpServerUnavailableError(error))
+        return internalErrorResponse(502, { error: 'MCP server did not answer usefully' })
+      return internalOrchestrationErrorPolicy.project(error)
+    },
+    unhandled: internalOrchestrationErrorPolicy.unhandled,
+  },
   mapInput: ({ body }) => body,
   useCase: readIntegrationCatalog,
 })

apps/sim/app/api/workspaces/[id]/background-work/route.ts

@@ -2,6 +2,7 @@ import { db } from '@sim/db'
 import { type NextRequest, NextResponse } from 'next/server'
 import { getWorkspaceBackgroundWorkContract } from '@/lib/api/contracts/workspace-fork'
 import { parseRequest } from '@/lib/api/server'
+import { orchestrationFailureResponse } from '@/lib/api/server/orchestration-response'
 import { getSession } from '@/lib/auth'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { listSurfacedBackgroundWork } from '@/ee/workspace-forking/lib/background-work/store'
@@ -21,7 +22,13 @@ export const GET = withRouteHandler(
 
     // The fork Activity feed is a fork feature: gate it behind the same forking-enabled +
     // workspace-admin check the other fork routes use, instead of a bare access check.
-    await assertWorkspaceAdminAccess(id, session.user.id)
+    try {
+      await assertWorkspaceAdminAccess(id, session.user.id)
+    } catch (error) {
+      const refused = orchestrationFailureResponse(error)
+      if (refused) return refused
+      throw error
+    }
 
     const { rows, nextCursor } = await listSurfacedBackgroundWork(db, id, { cursor, limit })
     return NextResponse.json({

apps/sim/app/api/workspaces/members/[id]/route.ts

@@ -2,6 +2,7 @@ import { removeWorkspaceMemberContract } from '@/lib/api/contracts/invitations'
 import {
   defineInternalJsonRoute,
   internalErrorResponse,
+  internalOrchestrationErrorPolicy,
   internalRateLimits,
   internalSessionAuth,
 } from '@/lib/api/server/routes'
@@ -32,7 +33,7 @@ export const DELETE = defineInternalJsonRoute({
         error instanceof NoWorkspaceAccessError
       )
         return internalErrorResponse(403, { error: 'Insufficient permissions' })
-      return null
+      return internalOrchestrationErrorPolicy.project(error)
     },
     unhandled: () => internalErrorResponse(500, { error: 'Failed to remove workspace member' }),
   },

apps/sim/lib/core/utils/stream-limits.test.ts

@@ -293,6 +293,56 @@ describe('stream limits', () => {
     }
   )
 
+  it.each([
+    ['a JSON body', 'application/json', '{"name":"example"}', true],
+    [
+      'a boundary the body never uses',
+      'multipart/form-data; boundary=declared',
+      '--other\r\n',
+      true,
+    ],
+    ['a streamed body the parser cannot decode', 'text/plain', 'name=example', false],
+  ])('rejects %s as invalid multipart input', async (_label, contentType, body, declareLength) => {
+    const bytes = new TextEncoder().encode(body)
+    const request = new Request('http://localhost/upload', {
+      method: 'POST',
+      headers: {
+        'content-type': contentType,
+        ...(declareLength ? { 'content-length': String(bytes.byteLength) } : {}),
+      },
+      body: declareLength ? bytes : streamFromChunks([bytes]),
+      duplex: 'half',
+    } as RequestInit)
+
+    await expect(
+      readFormDataWithLimit(request, { maxBytes: 1024 * 1024, label: 'multipart body' })
+    ).rejects.toBeInstanceOf(MultipartFieldValidationError)
+  })
+
+  it('leaves a body that was already read to fail as a server error', async () => {
+    const input = new FormData()
+    input.append('name', 'example')
+    const encoded = new Request('http://localhost/upload', { method: 'POST', body: input })
+    const bytes = new Uint8Array(await encoded.arrayBuffer())
+    const request = new Request('http://localhost/upload', {
+      method: 'POST',
+      headers: {
+        'content-type': encoded.headers.get('content-type') ?? '',
+        'content-length': String(bytes.byteLength),
+      },
+      body: bytes,
+    })
+    await request.text()
+
+    const failure = await readFormDataWithLimit(request, {
+      maxBytes: 1024 * 1024,
+      label: 'multipart body',
+    }).catch((error: unknown) => error)
+
+    expect(failure).toBeInstanceOf(TypeError)
+    expect(failure).not.toBeInstanceOf(MultipartFieldValidationError)
+  })
+
   it('rejects multipart streams without content-length once bytes exceed the limit', async () => {
     const request = new Request('http://localhost/upload', {
       method: 'POST',

apps/sim/lib/core/utils/stream-limits.ts

@@ -69,6 +69,7 @@ export interface ReadFormDataWithLimitRequest {
   method: string
   headers?: Headers
   body?: ReadableStream<Uint8Array> | null
+  bodyUsed?: boolean
   formData: () => Promise<FormData>
 }
 
@@ -138,7 +139,7 @@ export async function readFormDataWithLimit(
   assertContentLengthWithinLimit(request.headers, options.maxBytes, options.label)
 
   if (request.headers?.get('content-length') || !request.body) {
-    const formData = await request.formData()
+    const formData = await parseFormData(request)
     assertMultipartFieldsAreStorable(formData)
     return formData
   }
@@ -149,11 +150,33 @@ export async function readFormDataWithLimit(
     headers: request.headers,
     body: new Uint8Array(body),
   })
-  const formData = await boundedRequest.formData()
+  const formData = await parseFormData(boundedRequest)
   assertMultipartFieldsAreStorable(formData)
   return formData
 }
 
+/**
+ * The fetch body parser rejects a body that does not decode as the content type it
+ * declares (a JSON body, a boundary the body never uses) with a bare `TypeError`. That
+ * is the caller's malformed input, so it surfaces as the same validation failure every
+ * multipart route already answers with a 400. So does a declared-length body that ends
+ * early, since the client sent an incomplete request. A body some earlier code already
+ * consumed is a server bug, so that failure is left to surface as it is.
+ */
+async function parseFormData(
+  request: Pick<ReadFormDataWithLimitRequest, 'formData' | 'bodyUsed'>
+): Promise<FormData> {
+  if (request.bodyUsed) return request.formData()
+  try {
+    return await request.formData()
+  } catch (error) {
+    if (error instanceof TypeError) {
+      throw new MultipartFieldValidationError('Request body must be valid multipart form data')
+    }
+    throw error
+  }
+}
+
 export interface ReadStreamWithLimitOptions {
   maxBytes: number
   label: string

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

@@ -1,4 +1,5 @@
 import { loggerMock } from '@sim/testing'
+import { DrizzleQueryError } from 'drizzle-orm/errors'
 import { NextRequest, NextResponse } from 'next/server'
 import { describe, expect, it, vi } from 'vitest'
 import { HttpError } from '@/lib/core/utils/http-error'
@@ -142,4 +143,89 @@ describe('withRouteHandler', () => {
       await expect(response.json()).resolves.toEqual({ family: 'generic' })
     }
   )
+
+  describe('a database that is temporarily unavailable', () => {
+    const SQL = 'select "secret_column" from "workflow_execution_logs" where "id" = $1'
+
+    function failedQuery(code: string, message: string): DrizzleQueryError {
+      return new DrizzleQueryError(SQL, ['id-1'], Object.assign(new Error(message), { code }))
+    }
+
+    function familyRoute(error: unknown) {
+      return withRouteHandler(
+        async () => {
+          throw error
+        },
+        {
+          typedErrorResponse: ({ error: typed, status }) =>
+            NextResponse.json({ family: 'typed', error: typed.message }, { status }),
+          unhandledErrorResponse: () => NextResponse.json({ family: 'generic' }, { status: 500 }),
+        }
+      )
+    }
+
+    it.each([
+      [
+        'a replica recovery conflict',
+        failedQuery('40001', 'canceling statement due to conflict with recovery'),
+      ],
+      ['a statement timeout', failedQuery('57014', 'canceling statement due to statement timeout')],
+      [
+        'a connection lost under the query',
+        failedQuery('CONNECTION_CLOSED', 'write CONNECTION_CLOSED'),
+      ],
+      [
+        'a server restarting',
+        failedQuery('57P01', 'terminating connection due to administrator command'),
+      ],
+    ])('answers a read failing on %s with a retryable 503', async (_, error) => {
+      const response = await familyRoute(error)(
+        new NextRequest('http://localhost/api/test'),
+        undefined
+      )
+
+      expect(response.status).toBe(503)
+      expect(Number(response.headers.get('Retry-After'))).toBeGreaterThan(0)
+      const body = await response.json()
+      expect(body.family).toBe('typed')
+      expect(JSON.stringify(body)).not.toContain('secret_column')
+    })
+
+    it('answers a HEAD the same way as its GET', async () => {
+      const response = await familyRoute(
+        failedQuery('40001', 'canceling statement due to conflict with recovery')
+      )(new NextRequest('http://localhost/api/test', { method: 'HEAD' }), undefined)
+
+      expect(response.status).toBe(503)
+    })
+
+    it.each(['POST', 'PUT', 'PATCH', 'DELETE'])(
+      'leaves a %s unhandled, since an earlier statement may already have committed',
+      async (method) => {
+        const response = await familyRoute(
+          failedQuery('40001', 'canceling statement due to conflict with recovery')
+        )(new NextRequest('http://localhost/api/test', { method }), undefined)
+
+        expect(response.status).toBe(500)
+        expect(response.headers.get('Retry-After')).toBeNull()
+      }
+    )
+
+    it.each([
+      ['a user-requested cancel', failedQuery('57014', 'canceling statement due to user request')],
+      ['a permanent query error', failedQuery('42601', 'syntax error at or near ")"')],
+      [
+        'a socket reset outside any query',
+        Object.assign(new Error('read ECONNRESET'), { code: 'ECONNRESET' }),
+      ],
+    ])('leaves %s on a read unhandled', async (_, error) => {
+      const response = await familyRoute(error)(
+        new NextRequest('http://localhost/api/test'),
+        undefined
+      )
+
+      expect(response.status).toBe(500)
+      await expect(response.json()).resolves.toEqual({ family: 'generic' })
+    })
+  })
 })