fix(tables): close sync-import TOCTOU by claiming the atomic import gate

The sync import route checked importStatus from a checkAccess snapshot, then
parsed/validated/wrote seconds later without taking the atomic claim. A
concurrent async kickoff (markTableImporting) could slip into that window and
both writers would run together — for replace mode, two delete+insert passes
leave the table indeterminate.

Claim the same atomic gate (markTableImporting) right before the write and
release it in the finally (before the response returns, so a client refetch
never sees the transient status). A row-level FOR UPDATE was avoided on purpose:
it would invert lock order against the position advisory lock / row-count
trigger and risk a deadlock — markTableImporting is the established gate.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: TheodoreSpeaks

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

@@ -12,12 +12,16 @@ const {
   mockReplaceTableRowsWithTx,
   mockAddTableColumnsWithTx,
   mockDispatchAfterBatchInsert,
+  mockMarkTableImporting,
+  mockReleaseImportClaim,
 } = vi.hoisted(() => ({
   mockCheckAccess: vi.fn(),
   mockBatchInsertRowsWithTx: vi.fn(),
   mockReplaceTableRowsWithTx: vi.fn(),
   mockAddTableColumnsWithTx: vi.fn(),
   mockDispatchAfterBatchInsert: vi.fn(),
+  mockMarkTableImporting: vi.fn(),
+  mockReleaseImportClaim: vi.fn(),
 }))
 
 vi.mock('@sim/utils/id', () => ({
@@ -53,6 +57,8 @@ vi.mock('@/lib/table/service', () => ({
   replaceTableRowsWithTx: mockReplaceTableRowsWithTx,
   addTableColumnsWithTx: mockAddTableColumnsWithTx,
   dispatchAfterBatchInsert: mockDispatchAfterBatchInsert,
+  markTableImporting: mockMarkTableImporting,
+  releaseImportClaim: mockReleaseImportClaim,
 }))
 
 import { POST } from '@/app/api/table/[tableId]/import/route'
@@ -142,6 +148,8 @@ describe('POST /api/table/[tableId]/import', () => {
       data.rows.map((_, i) => ({ id: `row_${i}` }))
     )
     mockReplaceTableRowsWithTx.mockResolvedValue({ deletedCount: 0, insertedCount: 0 })
+    mockMarkTableImporting.mockResolvedValue(true)
+    mockReleaseImportClaim.mockResolvedValue(undefined)
     mockAddTableColumnsWithTx.mockImplementation(
       async (
         _trx,
@@ -168,6 +176,22 @@ describe('POST /api/table/[tableId]/import', () => {
     expect(response.status).toBe(401)
   })
 
+  it('returns 409 when a background import already holds the table (claim lost)', async () => {
+    mockMarkTableImporting.mockResolvedValueOnce(false)
+    const response = await callPost(createFormData(createCsvFile('name,age\nAlice,30')))
+    expect(response.status).toBe(409)
+    expect(mockBatchInsertRowsWithTx).not.toHaveBeenCalled()
+    expect(mockReplaceTableRowsWithTx).not.toHaveBeenCalled()
+    expect(mockReleaseImportClaim).not.toHaveBeenCalled()
+  })
+
+  it('releases the import claim after a successful write', async () => {
+    const response = await callPost(createFormData(createCsvFile('name,age\nAlice,30')))
+    expect(response.status).toBe(200)
+    expect(mockMarkTableImporting).toHaveBeenCalledWith('tbl_1', 'deadbeefcafef00d')
+    expect(mockReleaseImportClaim).toHaveBeenCalledWith('tbl_1', 'deadbeefcafef00d')
+  })
+
   it('returns 400 when the mode is invalid', async () => {
     const response = await callPost(
       createFormData(createCsvFile('name,age\nAlice,30'), { mode: 'bogus' })

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

@@ -29,6 +29,8 @@ import {
   createCsvParser,
   dispatchAfterBatchInsert,
   inferColumnType,
+  markTableImporting,
+  releaseImportClaim,
   replaceTableRowsWithTx,
   sanitizeName,
   type TableDefinition,
@@ -57,6 +59,7 @@ export const POST = withRouteHandler(async (request: NextRequest, { params }: Ro
   const requestId = generateRequestId()
   const { tableId } = tableIdParamsSchema.parse(await params)
   let fileStream: Readable | undefined
+  let claimedImportId: string | null = null
 
   try {
     const authResult = await checkSessionOrInternalAuth(request, { requireWorkflowId: false })
@@ -247,6 +250,19 @@ export const POST = withRouteHandler(async (request: NextRequest, { params }: Ro
 
     const coerced = coerceRowsForTable(rows, prospectiveTable.schema, validation.effectiveMap)
 
+    // Atomically claim the table before writing. The pre-check above reads a checkAccess snapshot
+    // taken before the parse/validation; a background import could claim the table in that window.
+    // markTableImporting is the single atomic gate (same one the async kickoff uses) — released in
+    // the finally so a sync import can't write concurrently with a background one (corrupts replace).
+    const syncImportId = generateId()
+    if (!(await markTableImporting(tableId, syncImportId))) {
+      return NextResponse.json(
+        { error: 'An import is already in progress for this table' },
+        { status: 409 }
+      )
+    }
+    claimedImportId = syncImportId
+
     if (mode === 'append') {
       if (prospectiveTable.rowCount + coerced.length > prospectiveTable.maxRows) {
         const deficit = prospectiveTable.rowCount + coerced.length - prospectiveTable.maxRows
@@ -407,5 +423,7 @@ export const POST = withRouteHandler(async (request: NextRequest, { params }: Ro
     )
   } finally {
     fileStream?.destroy()
+    // Release before the response returns, so a client refetch never observes the transient claim.
+    if (claimedImportId) await releaseImportClaim(tableId, claimedImportId).catch(() => {})
   }
 })

apps/sim/lib/table/service.ts

@@ -1371,6 +1371,24 @@ export async function markTableImporting(tableId: string, importId: string): Pro
   return updated.length > 0
 }
 
+/**
+ * Releases a claim taken by {@link markTableImporting} for a synchronous import — clears the
+ * import state back to idle. Scoped to `importId` so it only clears its own claim, never a newer
+ * run that may have taken over. A sync route claims, writes, then releases here in a `finally`.
+ */
+export async function releaseImportClaim(tableId: string, importId: string): Promise<void> {
+  await db
+    .update(userTableDefinitions)
+    .set({ importStatus: null, importId: null, importStartedAt: null, updatedAt: new Date() })
+    .where(
+      and(
+        eq(userTableDefinitions.id, tableId),
+        eq(userTableDefinitions.importId, importId),
+        eq(userTableDefinitions.importStatus, 'importing')
+      )
+    )
+}
+
 /**
  * Records import progress (rows processed so far). Also bumps `updatedAt` so the
  * stale-import janitor (`cleanup-stale-executions`) sees a live heartbeat and doesn't mark a