improvement(api): make multipart transfers stateless

Author: TheodoreSpeaks

apps/docs/openapi-v2-files-audit.json

@@ -319,7 +319,7 @@
       "post": {
         "operationId": "createFileUpload",
         "summary": "Create File Upload",
-        "description": "Create a durable multipart upload session. Every file uses this flow; a small file is a single part. The maximum file size is 5 GB.",
+        "description": "Create a stateless multipart upload session and signed upload token. Every file uses this flow; a small file is a single part. The maximum file size is 5 GB.",
         "tags": ["Files"],
         "requestBody": {
           "required": true,
@@ -339,31 +339,6 @@
       }
     },
     "/api/v2/files/uploads/{uploadId}": {
-      "get": {
-        "operationId": "getFileUpload",
-        "summary": "Get File Upload",
-        "description": "Read the durable state of a file upload session.",
-        "tags": ["Files"],
-        "parameters": [
-          {
-            "name": "uploadId",
-            "in": "path",
-            "required": true,
-            "schema": { "type": "string" }
-          },
-          { "$ref": "#/components/parameters/WorkspaceIdQuery" }
-        ],
-        "responses": {
-          "200": {
-            "description": "The upload session.",
-            "content": { "application/json": { "schema": {} } }
-          },
-          "401": { "$ref": "#/components/responses/Unauthorized" },
-          "404": { "$ref": "#/components/responses/NotFound" },
-          "429": { "$ref": "#/components/responses/RateLimited" },
-          "500": { "$ref": "#/components/responses/InternalError" }
-        }
-      },
       "delete": {
         "operationId": "abortFileUpload",
         "summary": "Abort File Upload",
@@ -376,7 +351,8 @@
             "required": true,
             "schema": { "type": "string" }
           },
-          { "$ref": "#/components/parameters/WorkspaceIdQuery" }
+          { "$ref": "#/components/parameters/WorkspaceIdQuery" },
+          { "$ref": "#/components/parameters/UploadTokenHeader" }
         ],
         "responses": {
           "200": {
@@ -404,7 +380,8 @@
             "required": true,
             "schema": { "type": "string" }
           },
-          { "$ref": "#/components/parameters/WorkspaceIdQuery" }
+          { "$ref": "#/components/parameters/WorkspaceIdQuery" },
+          { "$ref": "#/components/parameters/UploadTokenHeader" }
         ],
         "requestBody": {
           "required": true,
@@ -437,7 +414,8 @@
             "required": true,
             "schema": { "type": "string" }
           },
-          { "$ref": "#/components/parameters/WorkspaceIdQuery" }
+          { "$ref": "#/components/parameters/WorkspaceIdQuery" },
+          { "$ref": "#/components/parameters/UploadTokenHeader" }
         ],
         "requestBody": {
           "required": true,
@@ -1699,6 +1677,13 @@
           "example": "a91c4b2e-6d3f-4e8a-b5c7-0d9e2f1a8c64"
         }
       },
+      "UploadTokenHeader": {
+        "name": "upload-token",
+        "in": "header",
+        "required": true,
+        "description": "The signed token returned when the multipart upload was created.",
+        "schema": { "type": "string", "minLength": 1 }
+      },
       "FileIdPath": {
         "name": "fileId",
         "in": "path",

apps/docs/openapi-v2-tables.json

@@ -3494,7 +3494,7 @@
       "post": {
         "operationId": "createTableImport",
         "summary": "Create Table Import",
-        "description": "Create one durable import resource for either a new or existing table. Upload sources return multipart details; workspace-file sources start immediately.",
+        "description": "Create a table import. Upload sources return a stateless multipart token; workspace-file sources start immediately and both use table jobs for processing state.",
         "tags": ["Tables"],
         "requestBody": {
           "required": true,
@@ -3519,7 +3519,7 @@
       "get": {
         "operationId": "getTableImport",
         "summary": "Get Table Import",
-        "description": "Read upload, processing, progress, and terminal state using the same import id.",
+        "description": "Read processing progress and terminal state from the table job using the same import id.",
         "tags": ["Tables"],
         "parameters": [
           {
@@ -3553,7 +3553,8 @@
             "required": true,
             "schema": { "type": "string" }
           },
-          { "$ref": "#/components/parameters/WorkspaceIdQuery" }
+          { "$ref": "#/components/parameters/WorkspaceIdQuery" },
+          { "$ref": "#/components/parameters/OptionalUploadTokenHeader" }
         ],
         "responses": {
           "200": {
@@ -3581,7 +3582,8 @@
             "required": true,
             "schema": { "type": "string" }
           },
-          { "$ref": "#/components/parameters/WorkspaceIdQuery" }
+          { "$ref": "#/components/parameters/WorkspaceIdQuery" },
+          { "$ref": "#/components/parameters/UploadTokenHeader" }
         ],
         "requestBody": {
           "required": true,
@@ -3614,7 +3616,8 @@
             "required": true,
             "schema": { "type": "string" }
           },
-          { "$ref": "#/components/parameters/WorkspaceIdQuery" }
+          { "$ref": "#/components/parameters/WorkspaceIdQuery" },
+          { "$ref": "#/components/parameters/UploadTokenHeader" }
         ],
         "requestBody": {
           "required": true,
@@ -3884,6 +3887,20 @@
         },
         "description": "The unique identifier of the workspace that owns the table."
       },
+      "UploadTokenHeader": {
+        "name": "upload-token",
+        "in": "header",
+        "required": true,
+        "description": "The signed token returned for an upload-backed table import.",
+        "schema": { "type": "string", "minLength": 1 }
+      },
+      "OptionalUploadTokenHeader": {
+        "name": "upload-token",
+        "in": "header",
+        "required": false,
+        "description": "Required when canceling before upload completion; omitted when canceling a running table job.",
+        "schema": { "type": "string", "minLength": 1 }
+      },
       "LimitQuery": {
         "name": "limit",
         "in": "query",

apps/sim/app/api/cron/cleanup-stale-executions/route.ts

@@ -1,11 +1,5 @@
 import { asyncJobs, db } from '@sim/db'
-import {
-  tableImports,
-  tableJobs,
-  uploadSessions,
-  workflowDeploymentOperation,
-  workflowExecutionLogs,
-} from '@sim/db/schema'
+import { tableJobs, workflowDeploymentOperation, workflowExecutionLogs } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
 import { and, eq, exists, gt, inArray, lt, sql } from 'drizzle-orm'
@@ -16,7 +10,6 @@ import { JOB_RETENTION_HOURS, JOB_STATUS } from '@/lib/core/async-jobs'
 import { getMaxExecutionTimeout } from '@/lib/core/execution-limits'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { deleteFile } from '@/lib/uploads/core/storage-service'
-import { expireUploadSessions } from '@/lib/uploads/multipart-session/service'
 
 const logger = createLogger('CleanupStaleExecutions')
 
@@ -135,8 +128,6 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
     // place (no rollback); the user retries. Also prune long-settled terminal jobs so the table
     // doesn't grow unbounded (the latest job per table is what list/detail reads surface).
     let staleTableJobsMarkedFailed = 0
-    let stalePreparingImportsMarkedFailed = 0
-    let expiredUploadSessions = 0
     try {
       const now = new Date()
       const staleJobs = await db
@@ -151,80 +142,10 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
         .returning({ id: tableJobs.id })
 
       staleTableJobsMarkedFailed = staleJobs.length
-      if (staleJobs.length > 0) {
-        const now = new Date()
-        await db
-          .update(tableImports)
-          .set({
-            status: 'failed',
-            error: `Import terminated: no progress for more than ${STALE_THRESHOLD_MINUTES} minutes`,
-            completedAt: now,
-            updatedAt: now,
-          })
-          .where(
-            inArray(
-              tableImports.id,
-              staleJobs.map((job) => job.id)
-            )
-          )
-      }
       if (staleTableJobsMarkedFailed > 0) {
         logger.info(`Marked ${staleTableJobsMarkedFailed} stale table jobs as failed`)
       }
 
-      const stalePreparingImports = await db
-        .select({ id: tableImports.id })
-        .from(tableImports)
-        .where(
-          and(eq(tableImports.status, 'preparing'), lt(tableImports.updatedAt, staleThreshold))
-        )
-        .orderBy(tableImports.updatedAt)
-        .limit(100)
-      if (stalePreparingImports.length > 0) {
-        const failedImports = await db
-          .update(tableImports)
-          .set({
-            status: 'failed',
-            error: `Import terminated: preparation did not finish within ${STALE_THRESHOLD_MINUTES} minutes`,
-            completedAt: now,
-            updatedAt: now,
-          })
-          .where(
-            and(
-              eq(tableImports.status, 'preparing'),
-              inArray(
-                tableImports.id,
-                stalePreparingImports.map((record) => record.id)
-              )
-            )
-          )
-          .returning({ uploadSessionId: tableImports.uploadSessionId })
-        stalePreparingImportsMarkedFailed = failedImports.length
-
-        const uploadSessionIds = failedImports.flatMap((record) =>
-          record.uploadSessionId ? [record.uploadSessionId] : []
-        )
-        if (uploadSessionIds.length > 0) {
-          const uploads = await db
-            .select({ storageKey: uploadSessions.storageKey })
-            .from(uploadSessions)
-            .where(inArray(uploadSessions.id, uploadSessionIds))
-          for (const upload of uploads) {
-            await deleteFile({ key: upload.storageKey, context: 'table-import' }).catch((error) => {
-              logger.warn('Failed to delete source for a stale table import', {
-                storageKey: upload.storageKey,
-                error: toError(error).message,
-              })
-            })
-          }
-        }
-      }
-      if (stalePreparingImportsMarkedFailed > 0) {
-        logger.info(
-          `Marked ${stalePreparingImportsMarkedFailed} stale preparing table imports as failed`
-        )
-      }
-
       const terminalRetention = new Date(Date.now() - TABLE_JOB_RETENTION_HOURS * 60 * 60 * 1000)
       const pruned = await db
         .delete(tableJobs)
@@ -255,14 +176,6 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       })
     }
 
-    try {
-      expiredUploadSessions = await expireUploadSessions(new Date(), 100)
-    } catch (error) {
-      logger.error('Failed to expire multipart upload sessions:', {
-        error: toError(error).message,
-      })
-    }
-
     // Clean up stale pending jobs (never started, e.g., due to server crash before startJob())
     let stalePendingJobsMarkedFailed = 0
 
@@ -393,12 +306,6 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       tableJobs: {
         staleMarkedFailed: staleTableJobsMarkedFailed,
       },
-      tableImports: {
-        stalePreparingMarkedFailed: stalePreparingImportsMarkedFailed,
-      },
-      uploadSessions: {
-        expired: expiredUploadSessions,
-      },
       deploymentOperations: {
         pruned: deploymentOperationsPruned,
         retentionDays: DEPLOYMENT_OPERATION_RETENTION_DAYS,

apps/sim/app/api/files/uploads/[uploadId]/complete/route.ts

@@ -28,10 +28,11 @@ export const POST = withRouteHandler(async (request: NextRequest, context: Uploa
   const access = await requireWorkspaceWrite(user, workspaceId)
   if (access) return access
   try {
-    const upload = await getOwnedUploadSession({
+    const upload = getOwnedUploadSession({
       uploadId: parsed.data.params.uploadId,
       workspaceId,
       userId: user,
+      uploadToken: parsed.data.headers['upload-token'],
     })
     const metadata = upload.metadata as { folderId?: string | null }
     const completed = await completeUploadSession({
@@ -49,7 +50,7 @@ export const POST = withRouteHandler(async (request: NextRequest, context: Uploa
         return { value: registered.file.id, completedFileId: registered.file.id }
       },
     })
-    const fileId = completed.value ?? completed.session.completedFileId
+    const fileId = completed.value
     if (!fileId) throw new Error('Completed upload is missing its workspace file id')
     const file = await getWorkspaceFile(workspaceId, fileId, { throwOnError: true })
     if (!file) throw new Error(`Completed workspace file ${fileId} not found`)

apps/sim/app/api/files/uploads/[uploadId]/parts/route.ts

@@ -25,10 +25,11 @@ export const POST = withRouteHandler(async (request: NextRequest, context: Uploa
   const access = await requireWorkspaceWrite(user, workspaceId)
   if (access) return access
   try {
-    const upload = await getOwnedUploadSession({
+    const upload = getOwnedUploadSession({
       uploadId: parsed.data.params.uploadId,
       workspaceId,
       userId: user,
+      uploadToken: parsed.data.headers['upload-token'],
     })
     const parts = await createUploadPartUrls({
       session: upload,

apps/sim/app/api/files/uploads/[uploadId]/route.ts

@@ -1,11 +1,7 @@
 import { type NextRequest, NextResponse } from 'next/server'
-import {
-  abortWorkspaceFileUploadContract,
-  getWorkspaceFileUploadContract,
-} from '@/lib/api/contracts/upload-sessions'
+import { abortWorkspaceFileUploadContract } from '@/lib/api/contracts/upload-sessions'
 import { parseRequest } from '@/lib/api/server'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
-import { getWorkspaceFile } from '@/lib/uploads/contexts/workspace'
 import { abortUploadSession, getOwnedUploadSession } from '@/lib/uploads/multipart-session/service'
 import {
   requireUploadUser,
@@ -18,31 +14,6 @@ interface UploadRouteParams {
   params: Promise<{ uploadId: string }>
 }
 
-export const GET = withRouteHandler(async (request: NextRequest, context: UploadRouteParams) => {
-  const user = await requireUploadUser()
-  if (user instanceof NextResponse) return user
-  const parsed = await parseRequest(getWorkspaceFileUploadContract, request, context)
-  if (!parsed.success) return parsed.response
-  const { workspaceId } = parsed.data.query
-  const access = await requireWorkspaceWrite(user, workspaceId)
-  if (access) return access
-  try {
-    const upload = await getOwnedUploadSession({
-      uploadId: parsed.data.params.uploadId,
-      workspaceId,
-      userId: user,
-    })
-    const file = upload.completedFileId
-      ? await getWorkspaceFile(workspaceId, upload.completedFileId, { throwOnError: true })
-      : null
-    return NextResponse.json({ data: toV2FileUpload(upload, file) })
-  } catch (error) {
-    const classified = uploadSessionErrorResponse(error)
-    if (classified) return classified
-    throw error
-  }
-})
-
 export const DELETE = withRouteHandler(async (request: NextRequest, context: UploadRouteParams) => {
   const user = await requireUploadUser()
   if (user instanceof NextResponse) return user
@@ -52,10 +23,11 @@ export const DELETE = withRouteHandler(async (request: NextRequest, context: Upl
   const access = await requireWorkspaceWrite(user, workspaceId)
   if (access) return access
   try {
-    const upload = await getOwnedUploadSession({
+    const upload = getOwnedUploadSession({
       uploadId: parsed.data.params.uploadId,
       workspaceId,
       userId: user,
+      uploadToken: parsed.data.headers['upload-token'],
     })
     return NextResponse.json({ data: toV2FileUpload(await abortUploadSession(upload), null) })
   } catch (error) {