From f0868d6a7083fc16e59cda3c86da3b8c3a39dde2 Mon Sep 17 00:00:00 2001 From: imccyu Date: Thu, 6 Aug 2026 04:29:39 +0800 Subject: [PATCH] test(session): cover remaining preparation races --- .../tests/jsonl.spec.ts | 32 ++++++++++++++ .../session-persistence/src/coordinator.ts | 15 +------ .../tests/persistence.spec.ts | 44 ++++++++++++++++++- 3 files changed, 77 insertions(+), 14 deletions(-) diff --git a/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts b/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts index 8d0e920b8c..4c191aa0cd 100644 --- a/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts +++ b/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts @@ -13,6 +13,25 @@ import { import { runPersistenceContract, meta, oneTurnLog, appendLog } from '../../session-persistence/tests/contract.ts' import { runCoordinatorContract, type CoordinatorFixture } from '../../session-persistence/tests/coordinator-contract.ts' +const statRace = vi.hoisted(() => ({ + path: undefined as string | undefined, + reads: 0, +})) + +vi.mock('node:fs/promises', async (importOriginal) => { + const actual = await importOriginal() + return { + ...actual, + stat: (async (...args: Parameters) => { + const identity = await actual.stat(...args) + if (String(args[0]) !== statRace.path || !('mtimeNs' in identity)) return identity + statRace.reads += 1 + if (statRace.reads !== 2) return identity + return { ...identity, mtimeNs: identity.mtimeNs + 1n } + }) as typeof actual.stat, + } +}) + let root: string const dirs: string[] = [] @@ -54,6 +73,8 @@ function rawLogPath(root: string, cwd: string | undefined, id: SessionId): strin } afterEach(async () => { + statRace.path = undefined + statRace.reads = 0 vi.restoreAllMocks() for (const d of dirs.splice(0)) await rm(d, { recursive: true, force: true }) }) @@ -266,6 +287,17 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => { expect(await persistence.readStoredRevision(SessionId('missing-revision'))).toBeUndefined() }) + it('retries a full-prefix read when the file revision changes during the read', async () => { + const m = meta('stored-prefix-revision-race') + await ctx.sessionPersistence.create(m) + await ctx.sessionPersistence.append(m.id, oneTurnLog()) + const persistence = ctx.sessionPersistence as SessionPersistenceJsonl + statRace.path = rawLogPath(root, m.cwd, m.id) + + await expect(persistence.loadStored(m.id)).resolves.toMatchObject({ events: oneTurnLog() }) + expect(statRace.reads).toBe(4) + }) + it('handles revision-stat races and errors after log discovery', async () => { const m = meta('stored-revision-race') await ctx.sessionPersistence.create(m) diff --git a/packages/session-persistence/session-persistence/src/coordinator.ts b/packages/session-persistence/session-persistence/src/coordinator.ts index 3baa7dcce7..7cf0d50d48 100644 --- a/packages/session-persistence/session-persistence/src/coordinator.ts +++ b/packages/session-persistence/session-persistence/src/coordinator.ts @@ -802,19 +802,8 @@ export class PersistenceCoordinator { } /** Read, repair in memory, validate, and freeze one cold source once. */ - private async prepareCore( - id: SessionId, - signal?: AbortSignal, - ): Promise> { - signal?.throwIfAborted() - let stored: StoredPrefix | undefined - try { - stored = await this.backend.loadStored(id, signal) - } catch (error: unknown) { - if (signal?.aborted) signal.throwIfAborted() - throw error - } - signal?.throwIfAborted() + private async prepareCore(id: SessionId): Promise> { + const stored = await this.backend.loadStored(id) if (stored === undefined) throw new Error(`session "${id}" not found`) try { const { meta, events, revision, tornMarker } = stored diff --git a/packages/session-persistence/session-persistence/tests/persistence.spec.ts b/packages/session-persistence/session-persistence/tests/persistence.spec.ts index 16cadd8eeb..a0f936a6a2 100644 --- a/packages/session-persistence/session-persistence/tests/persistence.spec.ts +++ b/packages/session-persistence/session-persistence/tests/persistence.spec.ts @@ -441,8 +441,9 @@ describe('PersistenceCoordinator session preparations', () => { const prepareId = SessionId('prepare-became-live') const loadId = SessionId('load-became-live') const inspectId = SessionId('inspect-became-live') + const validatedInspectId = SessionId('validated-inspect-became-live') const failedInspectId = SessionId('failed-inspect-became-live') - for (const id of [prepareId, loadId, inspectId]) { + for (const id of [prepareId, loadId, inspectId, validatedInspectId]) { backend.store.set(id, { meta: meta(id), events: oneTurnLog() }) } let coordinator!: PersistenceCoordinator @@ -472,6 +473,15 @@ describe('PersistenceCoordinator session preparations', () => { await expect(coordinator.inspect(inspectId)).resolves.toMatchObject({ meta: { id: inspectId } }) inspectGet.mockRestore() + const validatedInspectLive = Session.create(validatedInspectId, oneTurnLog(), meta(validatedInspectId)) + const validatedInspectGet = vi.spyOn(ctx.sessions, 'get') + .mockReturnValueOnce(undefined) + .mockReturnValueOnce(undefined) + .mockReturnValueOnce(validatedInspectLive) + await expect(coordinator.inspect(validatedInspectId)) + .resolves.toMatchObject({ meta: { id: validatedInspectId } }) + validatedInspectGet.mockRestore() + const failedInspectLive = Session.create(failedInspectId, oneTurnLog(), meta(failedInspectId)) backend.beforeLoadStored = () => Promise.reject(new Error('load failed')) const failedInspectGet = vi.spyOn(ctx.sessions, 'get') @@ -705,6 +715,38 @@ describe('PersistenceCoordinator session preparations', () => { } }) + it('retries cold append adoption when the prepared revision becomes stale', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const backend = new ControlledBackend() + const id = SessionId('append-adoption-revision-refresh') + backend.store.set(id, { meta: meta(id), events: oneTurnLog() }) + const readStoredRevision = backend.readStoredRevision.bind(backend) + vi.spyOn(backend, 'readStoredRevision') + .mockResolvedValueOnce(SessionPersistenceRevision('stale-revision')) + .mockImplementation(readStoredRevision) + let coordinator!: PersistenceCoordinator + const fiber = await ctx.plugin(Object.assign((inner: Context) => { + coordinator = new PersistenceCoordinator(inner, backend) + }, { inject: ['sessions'] })) + + try { + await coordinator.append(id, [{ + type: 'turn/start', + seq: oneTurnLog().length, + time: 7, + data: { turn: 2 }, + }]) + + expect(backend.loadAttempts).toBe(2) + expect(backend.appendAttempts).toBe(1) + expect(backend.store.get(id)?.events).toHaveLength(oneTurnLog().length + 1) + } finally { + await fiber.dispose() + await ctx.fiber.dispose() + } + }) + it('inspects an open live turn without balancing it', async () => { const ctx = new Context() await ctx.plugin(SessionStore)