Merge branch 'staging' into feat/migrate-byok-script

Author: Theodore Li

apps/sim/app/api/files/upload/route.ts

@@ -46,9 +46,10 @@ export async function POST(request: NextRequest) {
 
     const formData = await request.formData()
 
-    const files = formData.getAll('file') as File[]
+    const rawFiles = formData.getAll('file')
+    const files = rawFiles.filter((f): f is File => f instanceof File)
 
-    if (!files || files.length === 0) {
+    if (files.length === 0) {
       throw new InvalidRequestError('No files provided')
     }
 
@@ -73,7 +74,7 @@ export async function POST(request: NextRequest) {
     const uploadResults = []
 
     for (const file of files) {
-      const originalName = file.name
+      const originalName = file.name || 'untitled'
 
       if (!validateFileExtension(originalName)) {
         const extension = originalName.split('.').pop()?.toLowerCase() || 'unknown'

apps/sim/app/api/workflows/[id]/execute/route.ts

@@ -20,6 +20,10 @@ import {
 } from '@/lib/execution/call-chain'
 import { createExecutionEventWriter, setExecutionMeta } from '@/lib/execution/event-buffer'
 import { processInputFileFields } from '@/lib/execution/files'
+import {
+  registerManualExecutionAborter,
+  unregisterManualExecutionAborter,
+} from '@/lib/execution/manual-cancellation'
 import { preprocessExecution } from '@/lib/execution/preprocessing'
 import { LoggingSession } from '@/lib/logs/execution/logging-session'
 import {
@@ -845,6 +849,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
     const encoder = new TextEncoder()
     const timeoutController = createTimeoutAbortController(preprocessResult.executionTimeout?.sync)
     let isStreamClosed = false
+    let isManualAbortRegistered = false
 
     const eventWriter = createExecutionEventWriter(executionId)
     setExecutionMeta(executionId, {
@@ -857,6 +862,9 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
       async start(controller) {
         let finalMetaStatus: 'complete' | 'error' | 'cancelled' | null = null
 
+        registerManualExecutionAborter(executionId, timeoutController.abort)
+        isManualAbortRegistered = true
+
         const sendEvent = (event: ExecutionEvent) => {
           if (!isStreamClosed) {
             try {
@@ -1224,6 +1232,10 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
           })
           finalMetaStatus = 'error'
         } finally {
+          if (isManualAbortRegistered) {
+            unregisterManualExecutionAborter(executionId)
+            isManualAbortRegistered = false
+          }
           try {
             await eventWriter.close()
           } catch (closeError) {

apps/sim/app/api/workflows/[id]/executions/[executionId]/cancel/route.test.ts

@@ -0,0 +1,148 @@
+/**
+ * @vitest-environment node
+ */
+
+import { NextRequest } from 'next/server'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const mockCheckHybridAuth = vi.fn()
+const mockAuthorizeWorkflowByWorkspacePermission = vi.fn()
+const mockMarkExecutionCancelled = vi.fn()
+const mockAbortManualExecution = vi.fn()
+
+vi.mock('@sim/logger', () => ({
+  createLogger: () => ({ info: vi.fn(), warn: vi.fn(), error: vi.fn() }),
+}))
+
+vi.mock('@/lib/auth/hybrid', () => ({
+  checkHybridAuth: (...args: unknown[]) => mockCheckHybridAuth(...args),
+}))
+
+vi.mock('@/lib/execution/cancellation', () => ({
+  markExecutionCancelled: (...args: unknown[]) => mockMarkExecutionCancelled(...args),
+}))
+
+vi.mock('@/lib/execution/manual-cancellation', () => ({
+  abortManualExecution: (...args: unknown[]) => mockAbortManualExecution(...args),
+}))
+
+vi.mock('@/lib/workflows/utils', () => ({
+  authorizeWorkflowByWorkspacePermission: (params: unknown) =>
+    mockAuthorizeWorkflowByWorkspacePermission(params),
+}))
+
+import { POST } from './route'
+
+describe('POST /api/workflows/[id]/executions/[executionId]/cancel', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mockCheckHybridAuth.mockResolvedValue({ success: true, userId: 'user-1' })
+    mockAuthorizeWorkflowByWorkspacePermission.mockResolvedValue({ allowed: true })
+    mockAbortManualExecution.mockReturnValue(false)
+  })
+
+  it('returns success when cancellation was durably recorded', async () => {
+    mockMarkExecutionCancelled.mockResolvedValue({
+      durablyRecorded: true,
+      reason: 'recorded',
+    })
+
+    const response = await POST(
+      new NextRequest('http://localhost/api/workflows/wf-1/executions/ex-1/cancel', {
+        method: 'POST',
+      }),
+      {
+        params: Promise.resolve({ id: 'wf-1', executionId: 'ex-1' }),
+      }
+    )
+
+    expect(response.status).toBe(200)
+    await expect(response.json()).resolves.toEqual({
+      success: true,
+      executionId: 'ex-1',
+      redisAvailable: true,
+      durablyRecorded: true,
+      locallyAborted: false,
+      reason: 'recorded',
+    })
+  })
+
+  it('returns unsuccessful response when Redis is unavailable', async () => {
+    mockMarkExecutionCancelled.mockResolvedValue({
+      durablyRecorded: false,
+      reason: 'redis_unavailable',
+    })
+
+    const response = await POST(
+      new NextRequest('http://localhost/api/workflows/wf-1/executions/ex-1/cancel', {
+        method: 'POST',
+      }),
+      {
+        params: Promise.resolve({ id: 'wf-1', executionId: 'ex-1' }),
+      }
+    )
+
+    expect(response.status).toBe(200)
+    await expect(response.json()).resolves.toEqual({
+      success: false,
+      executionId: 'ex-1',
+      redisAvailable: false,
+      durablyRecorded: false,
+      locallyAborted: false,
+      reason: 'redis_unavailable',
+    })
+  })
+
+  it('returns unsuccessful response when Redis persistence fails', async () => {
+    mockMarkExecutionCancelled.mockResolvedValue({
+      durablyRecorded: false,
+      reason: 'redis_write_failed',
+    })
+
+    const response = await POST(
+      new NextRequest('http://localhost/api/workflows/wf-1/executions/ex-1/cancel', {
+        method: 'POST',
+      }),
+      {
+        params: Promise.resolve({ id: 'wf-1', executionId: 'ex-1' }),
+      }
+    )
+
+    expect(response.status).toBe(200)
+    await expect(response.json()).resolves.toEqual({
+      success: false,
+      executionId: 'ex-1',
+      redisAvailable: true,
+      durablyRecorded: false,
+      locallyAborted: false,
+      reason: 'redis_write_failed',
+    })
+  })
+
+  it('returns success when local fallback aborts execution without Redis durability', async () => {
+    mockMarkExecutionCancelled.mockResolvedValue({
+      durablyRecorded: false,
+      reason: 'redis_unavailable',
+    })
+    mockAbortManualExecution.mockReturnValue(true)
+
+    const response = await POST(
+      new NextRequest('http://localhost/api/workflows/wf-1/executions/ex-1/cancel', {
+        method: 'POST',
+      }),
+      {
+        params: Promise.resolve({ id: 'wf-1', executionId: 'ex-1' }),
+      }
+    )
+
+    expect(response.status).toBe(200)
+    await expect(response.json()).resolves.toEqual({
+      success: true,
+      executionId: 'ex-1',
+      redisAvailable: false,
+      durablyRecorded: false,
+      locallyAborted: true,
+      reason: 'redis_unavailable',
+    })
+  })
+})

apps/sim/app/api/workflows/[id]/executions/[executionId]/cancel/route.ts

@@ -2,6 +2,7 @@ import { createLogger } from '@sim/logger'
 import { type NextRequest, NextResponse } from 'next/server'
 import { checkHybridAuth } from '@/lib/auth/hybrid'
 import { markExecutionCancelled } from '@/lib/execution/cancellation'
+import { abortManualExecution } from '@/lib/execution/manual-cancellation'
 import { authorizeWorkflowByWorkspacePermission } from '@/lib/workflows/utils'
 
 const logger = createLogger('CancelExecutionAPI')
@@ -45,20 +46,27 @@ export async function POST(
 
     logger.info('Cancel execution requested', { workflowId, executionId, userId: auth.userId })
 
-    const marked = await markExecutionCancelled(executionId)
+    const cancellation = await markExecutionCancelled(executionId)
+    const locallyAborted = abortManualExecution(executionId)
 
-    if (marked) {
+    if (cancellation.durablyRecorded) {
       logger.info('Execution marked as cancelled in Redis', { executionId })
+    } else if (locallyAborted) {
+      logger.info('Execution cancelled via local in-process fallback', { executionId })
     } else {
-      logger.info('Redis not available, cancellation will rely on connection close', {
+      logger.warn('Execution cancellation was not durably recorded', {
         executionId,
+        reason: cancellation.reason,
       })
     }
 
     return NextResponse.json({
-      success: true,
+      success: cancellation.durablyRecorded || locallyAborted,
       executionId,
-      redisAvailable: marked,
+      redisAvailable: cancellation.reason !== 'redis_unavailable',
+      durablyRecorded: cancellation.durablyRecorded,
+      locallyAborted,
+      reason: cancellation.reason,
     })
   } catch (error: any) {
     logger.error('Failed to cancel execution', { workflowId, executionId, error: error.message })

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

@@ -87,32 +87,33 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{
     }
 
     const formData = await request.formData()
-    const file = formData.get('file') as File
+    const rawFile = formData.get('file')
 
-    if (!file) {
+    if (!rawFile || !(rawFile instanceof File)) {
       return NextResponse.json({ error: 'No file provided' }, { status: 400 })
     }
 
-    // Validate file size (100MB limit)
+    const fileName = rawFile.name || 'untitled'
+
     const maxSize = 100 * 1024 * 1024
-    if (file.size > maxSize) {
+    if (rawFile.size > maxSize) {
       return NextResponse.json(
-        { error: `File size exceeds 100MB limit (${(file.size / (1024 * 1024)).toFixed(2)}MB)` },
+        { error: `File size exceeds 100MB limit (${(rawFile.size / (1024 * 1024)).toFixed(2)}MB)` },
         { status: 400 }
       )
     }
 
-    const buffer = Buffer.from(await file.arrayBuffer())
+    const buffer = Buffer.from(await rawFile.arrayBuffer())
 
     const userFile = await uploadWorkspaceFile(
       workspaceId,
       session.user.id,
       buffer,
-      file.name,
-      file.type || 'application/octet-stream'
+      fileName,
+      rawFile.type || 'application/octet-stream'
     )
 
-    logger.info(`[${requestId}] Uploaded workspace file: ${file.name}`)
+    logger.info(`[${requestId}] Uploaded workspace file: ${fileName}`)
 
     recordAudit({
       workspaceId,
@@ -122,8 +123,8 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{
       action: AuditAction.FILE_UPLOADED,
       resourceType: AuditResourceType.FILE,
       resourceId: userFile.id,
-      resourceName: file.name,
-      description: `Uploaded file "${file.name}"`,
+      resourceName: fileName,
+      description: `Uploaded file "${fileName}"`,
       request,
     })
 

apps/sim/executor/constants.ts

@@ -439,7 +439,10 @@ export function isValidEnvVarName(name: string): boolean {
   return PATTERNS.ENV_VAR_NAME.test(name)
 }
 
-export function sanitizeFileName(fileName: string): string {
+export function sanitizeFileName(fileName: string | null | undefined): string {
+  if (!fileName || typeof fileName !== 'string') {
+    return 'untitled'
+  }
   return fileName.replace(/\s+/g, '-').replace(/[^a-zA-Z0-9.-]/g, '_')
 }