diff --git a/src/files/durable-byte-store.ts b/src/files/durable-byte-store.ts index 7e2a86277..029a0f1a3 100644 --- a/src/files/durable-byte-store.ts +++ b/src/files/durable-byte-store.ts @@ -1,3 +1,4 @@ +import { randomUUID } from "node:crypto"; import { createReadStream, createWriteStream } from "node:fs"; import { mkdir, rename, rm, stat } from "node:fs/promises"; import { once } from "node:events"; @@ -84,7 +85,7 @@ export function createLocalDurableByteStore(dir: string): DurableByteStore { } catch (error) { void error; } - const partPath = join(base, `${sha256}.${process.pid}.part`); + const partPath = join(base, `${sha256}.${randomUUID()}.part`); const out = createWriteStream(partPath); try { if (!out.write(data)) await once(out, "drain"); @@ -95,7 +96,12 @@ export function createLocalDurableByteStore(dir: string): DurableByteStore { await rm(partPath, { force: true }).catch(swallowAs("files: partial-file cleanup", undefined)); throw err; } - await rename(partPath, finalPath); + try { + await rename(partPath, finalPath); + } catch (err) { + await rm(partPath, { force: true }).catch(swallowAs("files: partial-file cleanup", undefined)); + throw err; + } return { blobKey, sizeBytes: data.length, sha256 }; }, diff --git a/test/file-artifact-store.test.ts b/test/file-artifact-store.test.ts index 8de0fc162..a2b01b7ff 100644 --- a/test/file-artifact-store.test.ts +++ b/test/file-artifact-store.test.ts @@ -1,6 +1,6 @@ import { test } from "node:test"; import assert from "node:assert/strict"; -import { mkdtemp, rm } from "node:fs/promises"; +import { mkdtemp, readdir, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { @@ -76,6 +76,21 @@ test("DurableByteStore (local-fs) round-trips binary intact across a fresh store } }); +test("DurableByteStore (local-fs) accepts concurrent identical writes", async () => { + const dir = await mkdtemp(join(tmpdir(), "docstore-")); + try { + const bytes = createLocalDurableByteStore(dir); + const big = Buffer.concat([PNG, Buffer.alloc(4 * 1024 * 1024, 7)]); + const results = await Promise.all(Array.from({ length: 8 }, () => bytes.put(big))); + for (const r of results) assert.equal(r.blobKey, results[0]!.blobKey); + assert.deepEqual(await drain(bytes as never, results[0]!.blobKey), big); + const leftovers = (await readdir(join(dir, "files"))).filter((name) => name.endsWith(".part")); + assert.deepEqual(leftovers, [], "no orphaned partial files survive the race"); + } finally { + await rm(dir, { recursive: true, force: true }); + } +}); + test("DurableByteStore enforces maxBytes mid-stream", async () => { const bytes = createMemoryDurableByteStore(); await assert.rejects(bytes.put(Buffer.alloc(1000), { maxBytes: 10 }), ByteSourceTooLargeError);