fix(tables): make markTableImporting an atomic claim to close the concurrent-import TOCTOU race

Author: TheodoreSpeaks

apps/sim/app/api/table/[tableId]/import-async/route.test.ts

@@ -77,7 +77,7 @@ describe('POST /api/table/[tableId]/import-async', () => {
       authType: 'session',
     })
     mockCheckAccess.mockResolvedValue({ ok: true, table: buildTable() })
-    mockMarkTableImporting.mockResolvedValue(undefined)
+    mockMarkTableImporting.mockResolvedValue(true)
     mockRunTableImport.mockResolvedValue(undefined)
   })
 
@@ -104,6 +104,13 @@ describe('POST /api/table/[tableId]/import-async', () => {
     )
   })
 
+  it('returns 409 when the table is already importing (claim lost)', async () => {
+    mockMarkTableImporting.mockResolvedValue(false)
+    const response = await makeRequest(validBody)
+    expect(response.status).toBe(409)
+    expect(mockRunTableImport).not.toHaveBeenCalled()
+  })
+
   it('returns 401 when unauthenticated', async () => {
     hybridAuthMockFns.mockCheckSessionOrInternalAuth.mockResolvedValue({ success: false })
     const response = await makeRequest(validBody)

apps/sim/app/api/table/[tableId]/import-async/route.ts

@@ -49,22 +49,23 @@ export const POST = withRouteHandler(async (request: NextRequest, { params }: Ro
   if (table.archivedAt) {
     return NextResponse.json({ error: 'Cannot import into an archived table' }, { status: 400 })
   }
-  // Reject overlapping imports: a second worker would insert at colliding row positions.
-  if (table.importStatus === 'importing') {
-    return NextResponse.json(
-      { error: 'An import is already in progress for this table' },
-      { status: 409 }
-    )
-  }
 
   const ext = fileName.split('.').pop()?.toLowerCase()
   if (ext !== 'csv' && ext !== 'tsv') {
     return NextResponse.json({ error: 'Only CSV and TSV files are supported' }, { status: 400 })
   }
   const delimiter = ext === 'tsv' ? '\t' : ','
 
+  // Atomically claim the table — the single concurrency gate. If another import already holds it,
+  // this returns false (no overlapping workers writing colliding row positions).
   const importId = generateId()
-  await markTableImporting(tableId, importId)
+  const claimed = await markTableImporting(tableId, importId)
+  if (!claimed) {
+    return NextResponse.json(
+      { error: 'An import is already in progress for this table' },
+      { status: 409 }
+    )
+  }
 
   runDetached('table-import', () =>
     runTableImport({

apps/sim/lib/table/service.ts

@@ -17,7 +17,7 @@ import {
 import { createLogger } from '@sim/logger'
 import { getPostgresErrorCode } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
-import { and, count, eq, gt, gte, inArray, isNull, type SQL, sql } from 'drizzle-orm'
+import { and, count, eq, gt, gte, inArray, isNull, ne, or, type SQL, sql } from 'drizzle-orm'
 import { MATERIALIZE_CONCURRENCY, mapWithConcurrency } from '@/lib/core/utils/concurrency'
 import { generateRestoreName } from '@/lib/core/utils/restore-name'
 import type { DbOrTx } from '@/lib/db/types'
@@ -1341,9 +1341,14 @@ export async function setTableSchemaForImport(tableId: string, schema: TableSche
     .where(eq(userTableDefinitions.id, tableId))
 }
 
-/** Marks an existing table as undergoing an async import (rows hidden until ready). */
-export async function markTableImporting(tableId: string, importId: string): Promise<void> {
-  await db
+/**
+ * Atomically claims a table for an async import. The `import_status != 'importing'` guard makes
+ * this the single concurrency gate: of two racing kickoffs only one row-update matches, so only
+ * one wins (no TOCTOU between a separate status check and this write). Returns whether it claimed
+ * the table — the caller returns 409 when it didn't.
+ */
+export async function markTableImporting(tableId: string, importId: string): Promise<boolean> {
+  const updated = await db
     .update(userTableDefinitions)
     .set({
       importStatus: 'importing',
@@ -1353,7 +1358,17 @@ export async function markTableImporting(tableId: string, importId: string): Pro
       importStartedAt: new Date(),
       updatedAt: new Date(),
     })
-    .where(eq(userTableDefinitions.id, tableId))
+    .where(
+      and(
+        eq(userTableDefinitions.id, tableId),
+        or(
+          isNull(userTableDefinitions.importStatus),
+          ne(userTableDefinitions.importStatus, 'importing')
+        )
+      )
+    )
+    .returning({ id: userTableDefinitions.id })
+  return updated.length > 0
 }
 
 /**