From 681e8eef95576b753e44b10942844aef128d92f7 Mon Sep 17 00:00:00 2001 From: Blue <3067670134@qq.com> Date: Wed, 16 Sep 2026 10:52:07 +0800 Subject: [PATCH] =?UTF-8?q?fix(storage):=20=E4=BF=AE=E5=A4=8D=20JSON=20?= =?UTF-8?q?=E5=B9=B6=E5=8F=91=E5=86=99=E5=85=A5=E5=85=B1=E4=BA=AB=E6=9A=82?= =?UTF-8?q?=E5=AD=98=E6=96=87=E4=BB=B6=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../json-file-store.concurrent.test.ts | 66 +++++++++++++++++++ .../shared/src/storage/json-file-store.ts | 57 ++++++++++++++-- 2 files changed, 116 insertions(+), 7 deletions(-) create mode 100644 packages/shared/src/storage/json-file-store.concurrent.test.ts diff --git a/packages/shared/src/storage/json-file-store.concurrent.test.ts b/packages/shared/src/storage/json-file-store.concurrent.test.ts new file mode 100644 index 00000000..30e44cba --- /dev/null +++ b/packages/shared/src/storage/json-file-store.concurrent.test.ts @@ -0,0 +1,66 @@ +import { describe, expect, it } from 'bun:test'; +import { mkdir, mkdtemp, readdir, readFile, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { JsonFileStore } from './json-file-store.ts'; + +async function withStore(work: (store: JsonFileStore, root: string) => Promise): Promise { + const root = await mkdtemp(join(tmpdir(), 'folio-json-file-store-')); + try { + await work(new JsonFileStore(root), root); + } finally { + await rm(root, { recursive: true, force: true }); + } +} + +function payload(id: number) { + return { id, body: `${id}:`.repeat(256) }; +} + +describe('JsonFileStore concurrent writes', () => { + it('allows concurrent writers to the same file', async () => { + await withStore(async (store, root) => { + const writes = Array.from({ length: 32 }, (_, id) => store.write('state.json', payload(id))); + + await Promise.all(writes); + + const result = JSON.parse(await readFile(join(root, 'state.json'), 'utf8')) as ReturnType; + expect(result.id).toBeGreaterThanOrEqual(0); + expect(result.id).toBeLessThan(32); + expect(result.body).toBe(payload(result.id).body); + expect((await readdir(root)).filter((name) => name.endsWith('.tmp'))).toEqual([]); + }); + }); + + it('supports concurrent writers from separate store instances', async () => { + await withStore(async (_store, root) => { + const stores = [new JsonFileStore(root), new JsonFileStore(root)]; + const writes = Array.from({ length: 32 }, (_, id) => stores[id % stores.length]!.write('state.json', payload(id))); + + await Promise.all(writes); + + const result = JSON.parse(await readFile(join(root, 'state.json'), 'utf8')) as ReturnType; + expect(result.id).toBeGreaterThanOrEqual(0); + expect(result.id).toBeLessThan(32); + expect(result.body).toBe(payload(result.id).body); + expect((await readdir(root)).filter((name) => name.endsWith('.tmp'))).toEqual([]); + }); + }); + + it('cleans up its temporary file when publishing fails', async () => { + await withStore(async (store, root) => { + await mkdir(join(root, 'occupied')); + + let error: unknown; + try { + await store.write('occupied', { ok: true }); + } catch (caught) { + error = caught; + } + + expect(error).toMatchObject({ code: 'STORAGE_WRITE_FAILED' }); + expect((await readdir(root)).filter((name) => name.endsWith('.tmp'))).toEqual([]); + expect((await readdir(join(root, 'occupied')))).toEqual([]); + }); + }); +}); diff --git a/packages/shared/src/storage/json-file-store.ts b/packages/shared/src/storage/json-file-store.ts index 3044f526..c4e3e3af 100644 --- a/packages/shared/src/storage/json-file-store.ts +++ b/packages/shared/src/storage/json-file-store.ts @@ -1,13 +1,36 @@ -import { mkdir, readFile, rename, unlink, writeFile } from 'node:fs/promises'; +import { mkdir, open, readFile, rename, unlink } from 'node:fs/promises'; +import { randomUUID } from 'node:crypto'; import { dirname, join } from 'node:path'; import { createCodeError } from '../agent/errors.ts'; +const publishLocks = new Map>(); + +/** + * Serialize only the final replacement for a target. Temporary files can be + * prepared concurrently, while Windows gets a deterministic replacement + * order when several store instances publish the same target at once. + */ +async function publish(target: string, tmp: string): Promise { + const previous = publishLocks.get(target) ?? Promise.resolve(); + let current: Promise; + current = previous + .catch(() => undefined) + .then(() => rename(tmp, target)); + publishLocks.set(target, current); + try { + await current; + } finally { + if (publishLocks.get(target) === current) publishLocks.delete(target); + } +} + /** * Minimal atomic JSON persistence backed by a directory of files. * - * Writes go to `.tmp` first and are renamed into place, so a crash - * mid-write never leaves a truncated file. This is the V1 storage substrate; - * repositories can later be swapped for SQLite without touching callers. + * Writes go to an exclusively created, per-write temporary file in the target + * directory and are renamed into place, so concurrent writers never share a + * staging file. This is the V1 storage substrate; repositories can later be + * swapped for SQLite without touching callers. */ export class JsonFileStore { private readonly rootDir: string; @@ -41,16 +64,36 @@ export class JsonFileStore { async write(file: string, data: unknown): Promise { const target = join(this.rootDir, file); - const tmp = `${target}.tmp`; + const tmp = `${target}.${randomUUID()}.tmp`; + let ownsTemp = false; try { + const contents = `${JSON.stringify(data, null, 2)}\n`; await mkdir(dirname(target), { recursive: true }); - await writeFile(tmp, `${JSON.stringify(data, null, 2)}\n`, 'utf8'); - await rename(tmp, target); + const handle = await open(tmp, 'wx', 0o600); + ownsTemp = true; + try { + await handle.writeFile(contents, 'utf8'); + } finally { + await handle.close(); + } + await publish(target, tmp); + ownsTemp = false; } catch (error) { throw createCodeError( 'STORAGE_WRITE_FAILED', `Failed to write ${file}: ${error instanceof Error ? error.message : String(error)}` ); + } finally { + if (ownsTemp) { + // Only clean up this writer's file; never remove the last good target + // or another writer's staging file. Preserve the original write error. + await unlink(tmp).catch((cleanupError: unknown) => { + const code = (cleanupError as NodeJS.ErrnoException)?.code; + if (code !== 'ENOENT') { + console.warn('[JsonFileStore] Temporary-file cleanup failed:', code ?? 'UNKNOWN'); + } + }); + } } }