fix(tables): gate import ownership every batch and stop canceled imports reappearing

- Worker checked run ownership only at the progress cadence (~every 5k rows), so
  a canceled/superseded import could insert several more batches (incl. the final
  partial batch) before stopping. Move the updateImportProgress ownership gate to
  the top of every flush — a run that lost the table stops within one batch.
- A list/dialog import canceled mid-upload left the server row `importing` until
  the in-flight server cancel landed; hydration re-seeded it from useTablesList,
  so the dismissed import flickered back. Flag the real table id canceled on the
  mid-upload cancel path, skip re-seeding flagged tables in hydration, and clear
  the flag once the server import is terminal.

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

Author: TheodoreSpeaks

apps/sim/app/workspace/[workspaceId]/tables/components/import-progress-menu/use-hydrate-import-tray.ts

@@ -33,6 +33,9 @@ export function useHydrateImportTray(workspaceId: string | undefined): void {
     for (const table of tables) {
       if (table.importStatus === 'importing') {
         if (tray.entries[table.id]) continue
+        // Canceled mid-upload: the server row stays `importing` until the in-flight server cancel
+        // lands. Don't re-seed an entry the user already dismissed — it would flicker back.
+        if (tray.isCanceled(table.id)) continue
         tray.upsert({
           tableId: table.id,
           workspaceId,
@@ -57,6 +60,12 @@ export function useHydrateImportTray(workspaceId: string | undefined): void {
           })
         }
       }
+
+      // Once the server import reaches a terminal state, drop any lingering cancel flag so it can
+      // never suppress a future re-import of the same table.
+      if (table.importStatus !== 'importing' && tray.isCanceled(table.id)) {
+        tray.consumeCanceled(table.id)
+      }
     }
   }, [workspaceId, tables])
 }

apps/sim/app/workspace/[workspaceId]/tables/tables.tsx

@@ -445,7 +445,10 @@ export function Tables() {
                 })
                 useImportTrayStore.getState().dismiss(pendingId)
                 if (result?.tableId && useImportTrayStore.getState().consumeCanceled(pendingId)) {
-                  // Canceled mid-upload — the worker just started; cancel it server-side.
+                  // Canceled mid-upload — the worker just started. Flag the real table id so
+                  // hydration won't re-seed it from the still-`importing` server row while the
+                  // server cancel is in flight, then cancel it server-side.
+                  useImportTrayStore.getState().cancel(result.tableId)
                   void cancelTableImport(workspaceId, result.tableId, result.importId).catch(
                     () => {}
                   )

apps/sim/lib/table/import-runner.ts

@@ -176,21 +176,26 @@ export async function runTableImport(payload: TableImportPayload): Promise<void>
 
     const flush = async (rows: Record<string, unknown>[]) => {
       if (rows.length === 0 || !schema || !headerToColumn) return
+      // Ownership gate on *every* batch, before inserting: once this run loses the table (cancel,
+      // supersede, or the stale-import janitor), `updateImportProgress` returns false and we stop
+      // immediately — a superseded worker must never write another batch into a table a newer
+      // import now owns. (Throttling this to the progress cadence let up to PROGRESS_INTERVAL_ROWS
+      // extra rows land after cancel.) The write also keeps the persisted row count fresh.
+      const owns = await updateImportProgress(tableId, inserted, importId)
+      if (!owns) throw new ImportSupersededError()
       const coerced = coerceRowsForTable(rows, schema, headerToColumn)
       inserted += await bulkInsertImportBatch(
         { tableId, workspaceId, userId, rows: coerced, startPosition: basePosition + inserted },
         { ...table, schema },
         requestId
       )
-      // Emit after the first batch lands, then every interval, so the bar appears early.
+      // Emit after the first batch lands, then every interval, so the bar appears early without
+      // flooding the SSE stream (the ownership/progress write above is what runs every batch).
       if (
         inserted - lastReported >= PROGRESS_INTERVAL_ROWS ||
         (lastReported === 0 && inserted > 0)
       ) {
         lastReported = inserted
-        // Heartbeat + ownership check: if a newer import has taken over this table, stop.
-        const owns = await updateImportProgress(tableId, inserted, importId)
-        if (!owns) throw new ImportSupersededError()
         // Exact, monotonic completion from bytes consumed — no wobbly row estimate.
         const percent =
           totalBytes > 0 ? Math.min(99, Math.round((bytesRead / totalBytes) * 100)) : undefined