test(session): cover remaining preparation races

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<typeof import('node:fs/promises')>()
+  return {
+    ...actual,
+    stat: (async (...args: Parameters<typeof actual.stat>) => {
+      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)

packages/session-persistence/session-persistence/src/coordinator.ts

@@ -802,19 +802,8 @@ export class PersistenceCoordinator<TornMarker = unknown> {
   }
 
   /** Read, repair in memory, validate, and freeze one cold source once. */
-  private async prepareCore(
-    id: SessionId,
-    signal?: AbortSignal,
-  ): Promise<PreparedSessionSource<TornMarker>> {
-    signal?.throwIfAborted()
-    let stored: StoredPrefix<TornMarker> | 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<PreparedSessionSource<TornMarker>> {
+    const stored = await this.backend.loadStored(id)
     if (stored === undefined) throw new Error(`session "${id}" not found`)
     try {
       const { meta, events, revision, tornMarker } = stored

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<never>
@@ -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<never>
+    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)