Give each local byte-store write a unique temp path

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 };
     },
 

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);