fix(tables): bind import worker to its importId (no stale-worker clobber/overlap) and destroy storage stream on failure

Author: TheodoreSpeaks

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

@@ -1,4 +1,4 @@
-import { Transform } from 'node:stream'
+import { type Readable, Transform } from 'node:stream'
 import { createLogger } from '@sim/logger'
 import { getErrorMessage } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
@@ -35,6 +35,13 @@ const logger = createLogger('TableImportRunner')
 /** Emit a progress event / DB update at most every this many rows. */
 const PROGRESS_INTERVAL_ROWS = 5000
 
+/**
+ * Thrown when this worker discovers it no longer owns the table's import (the stale-job janitor
+ * marked its run failed and a newer import took over). The worker stops inserting rather than
+ * writing into a table a second worker now owns.
+ */
+class ImportSupersededError extends Error {}
+
 /** `create` infers a schema for a new table; `append`/`replace` map onto an existing one. */
 export type TableImportMode = 'create' | 'append' | 'replace'
 
@@ -65,6 +72,9 @@ export interface TableImportPayload {
 export async function runTableImport(payload: TableImportPayload): Promise<void> {
   const { importId, tableId, workspaceId, userId, fileKey, fileName, delimiter, mode } = payload
   const requestId = generateId().slice(0, 8)
+  // Hoisted so `finally` can destroy it on any failure — otherwise the storage HTTP body leaks
+  // open until it times out.
+  let source: Readable | undefined
 
   try {
     const loaded = await getTableById(tableId, { includeArchived: true })
@@ -76,7 +86,7 @@ export async function runTableImport(payload: TableImportPayload): Promise<void>
     const totalBytes = (await headObject(fileKey, 'workspace'))?.size ?? 0
 
     // Stream the file rather than buffering it — a ~1M-row import must never be held in memory.
-    const source = await downloadFileStream({ key: fileKey, context: 'workspace' })
+    source = await downloadFileStream({ key: fileKey, context: 'workspace' })
 
     // Append must continue after the existing rows; create/replace start empty. Read once up
     // front (the import is the table's sole writer) and assign contiguous positions from it.
@@ -178,7 +188,9 @@ export async function runTableImport(payload: TableImportPayload): Promise<void>
         (lastReported === 0 && inserted > 0)
       ) {
         lastReported = inserted
-        await updateImportProgress(tableId, 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()
         // Extrapolate the total from rows-per-byte observed so far; self-refines as it runs.
         // `Math.max(inserted, …)` keeps it monotonic; omit when the byte size is unknown.
         const estimatedTotal =
@@ -219,7 +231,7 @@ export async function runTableImport(payload: TableImportPayload): Promise<void>
       if (sample.length === 0) {
         // No data rows — fail rather than report a successful empty import (matches the sync route).
         const message = 'CSV file has no data rows'
-        await markImportFailed(tableId, message)
+        await markImportFailed(tableId, importId, message)
         void appendTableEvent({
           kind: 'import',
           tableId,
@@ -236,8 +248,8 @@ export async function runTableImport(payload: TableImportPayload): Promise<void>
       await flush(batch)
     }
 
-    await updateImportProgress(tableId, inserted)
-    await markImportReady(tableId)
+    await updateImportProgress(tableId, inserted, importId)
+    await markImportReady(tableId, importId)
     void appendTableEvent({
       kind: 'import',
       tableId,
@@ -248,11 +260,22 @@ export async function runTableImport(payload: TableImportPayload): Promise<void>
     })
     logger.info(`[${requestId}] Import complete`, { tableId, fileName, mode, rows: inserted })
   } catch (err) {
-    const message = getErrorMessage(err, 'Import failed')
-    logger.error(`[${requestId}] Import failed for table ${tableId}:`, err)
-    await markImportFailed(tableId, message).catch(() => {})
-    void appendTableEvent({ kind: 'import', tableId, importId, status: 'failed', error: message })
+    if (err instanceof ImportSupersededError) {
+      // A newer import owns the table now — leave its status alone and just stop.
+      logger.info(`[${requestId}] Import superseded by a newer run; stopping`, {
+        tableId,
+        importId,
+      })
+    } else {
+      const message = getErrorMessage(err, 'Import failed')
+      logger.error(`[${requestId}] Import failed for table ${tableId}:`, err)
+      // Scoped to importId — a no-op if a newer import has taken over.
+      await markImportFailed(tableId, importId, message).catch(() => {})
+      void appendTableEvent({ kind: 'import', tableId, importId, status: 'failed', error: message })
+    }
   } finally {
+    // Release the storage stream so its HTTP connection doesn't leak on failure.
+    source?.destroy()
     // The uploaded source file is single-use (a fresh upload per import) — delete it once the
     // import is terminal so the workspace bucket doesn't accumulate. Best-effort.
     await deleteFile({ key: fileKey, context: 'workspace' }).catch((err) => {

apps/sim/lib/table/service.ts

@@ -1360,28 +1360,45 @@ export async function markTableImporting(tableId: string, importId: string): Pro
  * 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
  * still-running import as failed.
+ *
+ * Scoped to `importId`: a stale/superseded worker (its run was marked failed and retried)
+ * no longer matches and its write is a no-op. Returns whether this worker still owns the
+ * import, so the caller can stop inserting when it's been superseded.
  */
-export async function updateImportProgress(tableId: string, rowsProcessed: number): Promise<void> {
-  await db
+export async function updateImportProgress(
+  tableId: string,
+  rowsProcessed: number,
+  importId: string
+): Promise<boolean> {
+  const updated = await db
     .update(userTableDefinitions)
     .set({ importRowsProcessed: rowsProcessed, updatedAt: new Date() })
-    .where(eq(userTableDefinitions.id, tableId))
+    .where(and(eq(userTableDefinitions.id, tableId), eq(userTableDefinitions.importId, importId)))
+    .returning({ id: userTableDefinitions.id })
+  return updated.length > 0
 }
 
-/** Marks an import complete; rows become visible. */
-export async function markImportReady(tableId: string): Promise<void> {
+/** Marks an import complete; rows become visible. No-op if a newer import has taken over. */
+export async function markImportReady(tableId: string, importId: string): Promise<void> {
   await db
     .update(userTableDefinitions)
     .set({ importStatus: 'ready', importError: null, updatedAt: new Date() })
-    .where(eq(userTableDefinitions.id, tableId))
+    .where(and(eq(userTableDefinitions.id, tableId), eq(userTableDefinitions.importId, importId)))
 }
 
-/** Marks an import failed, leaving any already-committed rows in place. */
-export async function markImportFailed(tableId: string, error: string): Promise<void> {
+/**
+ * Marks an import failed, leaving any already-committed rows in place. No-op if a newer import
+ * has taken over (so a stale worker can't clobber the current run's status).
+ */
+export async function markImportFailed(
+  tableId: string,
+  importId: string,
+  error: string
+): Promise<void> {
   await db
     .update(userTableDefinitions)
     .set({ importStatus: 'failed', importError: error.slice(0, 2000), updatedAt: new Date() })
-    .where(eq(userTableDefinitions.id, tableId))
+    .where(and(eq(userTableDefinitions.id, tableId), eq(userTableDefinitions.importId, importId)))
 }
 
 /**