fix(uploads): commit local objects atomically, so "it exists" means "it finished"

headObject now answers local storage from `stat` instead of reporting every
object as missing. That is the honest answer and the byte-range paths need it,
but it also brought ten existing callers to life on self-hosted and dev
deployments — including three that read "the object is there" as "the copy
finished" and skip the work: the workspace fork copier, the KB document copier,
and the table snapshot cache.

Those guards were written against cloud backends, where an interrupted PUT
leaves no object at all. The local path was a bare `writeFile`, which truncates
before it writes, so a crash mid-write would have left a prefix under a key
those callers now treat as complete — a fork retry marking a torn file copied.

Writes now land on a sibling `.partial` path and are renamed onto the target;
`rename` within a directory is atomic on POSIX and replaces in one step, so the
key names either the previous bytes or the whole new ones. A failed write or
rename removes the temp file and rethrows. Local multipart accumulates and
commits through this same path, so snapshots are covered too.

Reviewed the other seven callers: all strictly improve. materialize-file now
charges quota against the real on-disk size rather than trusting the row, the
CSV import progress bar is no longer permanently indeterminate on local, and
the TikTok size probe skips its counting pass. Their comments claimed local
returns null; refreshed.

Tests assert the call sequence rather than an observed torn read — a timing
test passes against a plain `writeFile` too, since even a multi-megabyte write
resolves inside one macrotask, so it would be a test that cannot fail.
Confirmed all four go red when the temp-and-rename is reverted.

Author: waleedlatif1

apps/sim/app/api/tools/tiktok/upload-video-draft/upload.ts

@@ -138,8 +138,9 @@ async function countStoredFileBytes(options: StoredFileOptions): Promise<number>
 }
 
 /**
- * Resolves the authoritative object size. Cloud storage uses provider metadata; local storage
- * and providers without HEAD support are counted with a bounded, zero-accumulation pass.
+ * Resolves the authoritative object size. Cloud storage uses provider metadata and local
+ * storage stats the file; a provider that reports no size at all falls back to a bounded,
+ * zero-accumulation counting pass.
  */
 export async function getStoredVideoSize(options: StoredFileOptions): Promise<number> {
   throwIfAborted(options.signal)

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

@@ -129,8 +129,9 @@ export async function runTableImport(payload: TableImportPayload): Promise<void>
       return fresh ?? undefined
     }
 
-    // Total byte size for the progress estimate — a cheap HEAD, no download. May be null on
-    // the local dev provider, in which case the bar stays indeterminate (rows still show).
+    // Total byte size for the progress estimate — a cheap HEAD, no download. Every backend
+    // answers it, including local storage (which stats the file), so the bar is real; a
+    // genuinely absent object still yields 0 and leaves it indeterminate.
     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.

apps/sim/lib/uploads/core/storage-service.local-atomicity.test.ts

@@ -0,0 +1,122 @@
+/**
+ * The local-filesystem backend must commit an object atomically, the way every
+ * cloud backend does.
+ *
+ * `headObject` answers local storage from `stat` rather than reporting it as
+ * missing, so callers that read "the object is there" as "the copy finished" —
+ * the workspace fork copier, the KB document copier, the table snapshot cache —
+ * are live on self-hosted and dev deployments. That inference is only sound if a
+ * half-written file can never appear under the final key, which means the bytes
+ * must land on a sibling path and be renamed over the target: `rename` within a
+ * directory is atomic on POSIX, `writeFile` truncates first and is not.
+ *
+ * The assertion is on the call sequence rather than on an observed torn read.
+ * A timing-based test here passes against a plain `writeFile` too — the write of
+ * even a multi-megabyte buffer resolves well inside one macrotask — so it would
+ * be a test that cannot fail. This one fails the moment the temp-and-rename is
+ * replaced by a direct write.
+ *
+ * @vitest-environment node
+ *
+ * Under `isolate: false` the storage-service module may already be cached and
+ * bound to the real `@/lib/uploads/config` namespace, so a per-file `vi.mock` of
+ * that path would never reach it. This file patches the real namespace in place
+ * (the `USE_*` flags are value exports read at call time) and restores it after,
+ * matching `storage-service.blob-connection-string.test.ts`.
+ */
+import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { fs } = vi.hoisted(() => ({
+  fs: {
+    writeFile: vi.fn(async () => {}),
+    rename: vi.fn(async () => {}),
+    mkdir: vi.fn(async () => undefined),
+    rm: vi.fn(async () => {}),
+    stat: vi.fn(async () => ({ size: 11, isFile: () => true })),
+    readFile: vi.fn(async () => Buffer.alloc(0)),
+    unlink: vi.fn(async () => {}),
+    readdir: vi.fn(async () => []),
+    access: vi.fn(async () => {}),
+  },
+}))
+
+vi.mock('fs/promises', () => ({ ...fs, default: fs }))
+vi.mock('node:fs/promises', () => ({ ...fs, default: fs }))
+
+import * as uploadsConfig from '@/lib/uploads/config'
+import { UPLOAD_DIR_SERVER } from '@/lib/uploads/core/setup.server'
+import { uploadFile } from '@/lib/uploads/core/storage-service'
+
+const KEY = 'workspace/ws-1/doc.txt'
+const TARGET = `${UPLOAD_DIR_SERVER}/${KEY}`
+
+const CLOUD_FLAGS = ['USE_S3_STORAGE', 'USE_BLOB_STORAGE', 'USE_GCS_STORAGE'] as const
+const originalFlags = new Map(
+  CLOUD_FLAGS.map((flag) => [flag, uploadsConfig[flag] as boolean] as const)
+)
+
+function setCloudFlags(value: boolean) {
+  for (const flag of CLOUD_FLAGS) {
+    Object.defineProperty(uploadsConfig, flag, { value, configurable: true })
+  }
+}
+
+beforeEach(() => {
+  vi.clearAllMocks()
+  setCloudFlags(false)
+})
+
+afterAll(() => {
+  for (const [flag, value] of originalFlags) {
+    Object.defineProperty(uploadsConfig, flag, { value, configurable: true })
+  }
+})
+
+function upload() {
+  return uploadFile({
+    file: Buffer.from('hello world', 'utf8'),
+    fileName: 'doc.txt',
+    contentType: 'text/plain',
+    context: 'workspace',
+    customKey: KEY,
+    preserveKey: true,
+  })
+}
+
+describe('local filesystem uploads', () => {
+  it('writes to a sibling temp path and renames it onto the target', async () => {
+    await upload()
+
+    expect(fs.writeFile).toHaveBeenCalledTimes(1)
+    const [writtenPath] = fs.writeFile.mock.calls[0] as [string]
+    expect(writtenPath).toMatch(new RegExp(`^${TARGET}\\.[A-Za-z0-9_-]+\\.partial$`))
+    expect(fs.rename).toHaveBeenCalledWith(writtenPath, TARGET)
+  })
+
+  it('never writes directly to the key readers resolve', async () => {
+    await upload()
+
+    const targets = fs.writeFile.mock.calls.map(([path]) => path)
+    expect(targets).not.toContain(TARGET)
+  })
+
+  it('removes the temp file and surfaces the error when the write fails', async () => {
+    const failure = new Error('ENOSPC: no space left on device')
+    fs.writeFile.mockRejectedValueOnce(failure)
+
+    await expect(upload()).rejects.toThrow(failure)
+
+    const [writtenPath] = fs.writeFile.mock.calls[0] as [string]
+    expect(fs.rename).not.toHaveBeenCalled()
+    expect(fs.rm).toHaveBeenCalledWith(writtenPath, { force: true })
+  })
+
+  it('removes the temp file when the rename itself fails', async () => {
+    fs.rename.mockRejectedValueOnce(new Error('EXDEV: cross-device link'))
+
+    await expect(upload()).rejects.toThrow('EXDEV')
+
+    const [writtenPath] = fs.writeFile.mock.calls[0] as [string]
+    expect(fs.rm).toHaveBeenCalledWith(writtenPath, { force: true })
+  })
+})

apps/sim/lib/uploads/core/storage-service.ts

@@ -2,7 +2,7 @@ import type { Readable } from 'node:stream'
 import { randomBytes } from 'crypto'
 import { createLogger } from '@sim/logger'
 import { getErrorMessage } from '@sim/utils/errors'
-import { generateId } from '@sim/utils/id'
+import { generateId, generateShortId } from '@sim/utils/id'
 import { assertKnownSizeWithinLimit } from '@/lib/core/utils/stream-limits'
 import {
   getStorageConfig,
@@ -210,7 +210,7 @@ export async function uploadFile(options: UploadFileOptions): Promise<FileInfo>
     return uploadResult
   }
 
-  const { writeFile, mkdir } = await import('fs/promises')
+  const { writeFile, mkdir, rename, rm } = await import('fs/promises')
   const { join, dirname } = await import('path')
   const { UPLOAD_DIR_SERVER } = await import('./setup.server')
 
@@ -220,7 +220,25 @@ export async function uploadFile(options: UploadFileOptions): Promise<FileInfo>
 
   await mkdir(dirname(filesystemPath), { recursive: true })
 
-  await writeFile(filesystemPath, file)
+  /**
+   * Write to a sibling temp file, then rename over the target.
+   *
+   * Every cloud backend commits an object atomically — an interrupted PUT leaves
+   * nothing behind — and callers rely on that: the fork copier and the KB
+   * document copier both treat "the object is there" as "the copy finished" and
+   * skip re-copying. A bare `writeFile` here would let a crash mid-write leave a
+   * truncated file that satisfies that check forever. `rename` within the same
+   * directory is atomic on POSIX and replaces the target in one step, so the key
+   * either names the previous bytes or the complete new ones and never a prefix.
+   */
+  const pendingPath = `${filesystemPath}.${generateShortId(12)}.partial`
+  try {
+    await writeFile(pendingPath, file)
+    await rename(pendingPath, filesystemPath)
+  } catch (error) {
+    await rm(pendingPath, { force: true }).catch(() => {})
+    throw error
+  }
 
   if (metadata && persistMetadata) {
     await insertFileMetadataHelper(