diff --git a/packages/session-persistence/session-persistence/src/coordinator.ts b/packages/session-persistence/session-persistence/src/coordinator.ts index 46db6eee62..18bd884acd 100644 --- a/packages/session-persistence/session-persistence/src/coordinator.ts +++ b/packages/session-persistence/session-persistence/src/coordinator.ts @@ -255,10 +255,13 @@ export class PersistenceCoordinator { * @param id - the persisted session to reload. * @returns the header plus the event log, ending on a balanced `turn/end`. */ - load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { - const live = this.ctx.sessions.get(id) - if (live !== undefined) return this.loadLiveSnapshot(live) - return this.serialize(id, () => this.loadCore(id)) + async load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { + const selected = await this.serialize(id, async () => { + const live = this.ctx.sessions.get(id) + if (live !== undefined) return { live } + return { loaded: await this.loadCore(id) } + }) + return 'loaded' in selected ? selected.loaded : this.loadLiveSnapshot(selected.live) } private async loadCore(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { diff --git a/packages/session-persistence/session-persistence/tests/coordinator-contract.ts b/packages/session-persistence/session-persistence/tests/coordinator-contract.ts index 857e3e5a91..05b86c699d 100644 --- a/packages/session-persistence/session-persistence/tests/coordinator-contract.ts +++ b/packages/session-persistence/session-persistence/tests/coordinator-contract.ts @@ -121,6 +121,39 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise< } }) + it('rechecks live ownership after a cold load enters the per-id chain', async () => { + const fix = await makeFixture() + const { ctx, fiber } = await freshCtx(fix) + try { + const id = SessionId('queued-load-live-race') + const header = meta(id, WORK) + const start: SessionEvent = { + type: 'turn/start', + seq: 0, + time: 1, + data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }, + } + await ctx.sessionPersistence.create(header) + await ctx.sessionPersistence.append(id, [start]) + + const loading = ctx.sessionPersistence.load(id) + const live = ctx.sessions.create(id, { seed: [start], meta: header }) + await expect(loading).rejects.toThrow(/live turn is open/) + + live.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) + await ctx.sessions.flush(live) + const loaded = await ctx.sessionPersistence.load(id) + expect(loaded.events.map(event => event.type)).toEqual(['turn/start', 'turn/end']) + expect(loaded.events.at(-1)).toMatchObject({ + type: 'turn/end', + data: { reason: { kind: 'completed' } }, + }) + } finally { + await fiber.dispose() + await fix.cleanup() + } + }) + it('does not load an unmaterialized empty live session', async () => { const fix = await makeFixture() const { ctx, fiber } = await freshCtx(fix) diff --git a/packages/session-persistence/session-persistence/tests/persistence.spec.ts b/packages/session-persistence/session-persistence/tests/persistence.spec.ts index 1ac6977134..8aa446ca56 100644 --- a/packages/session-persistence/session-persistence/tests/persistence.spec.ts +++ b/packages/session-persistence/session-persistence/tests/persistence.spec.ts @@ -354,6 +354,7 @@ describe('PersistenceCoordinator retirement', () => { coordinator = new PersistenceCoordinator(inner, backend) }, { inject: ['sessions'] })) const appendGate = Promise.withResolvers() + const loadGate = Promise.withResolvers() try { const id = SessionId('retiring-buffered-owner') @@ -367,15 +368,20 @@ describe('PersistenceCoordinator retirement', () => { first.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) await vi.waitFor(() => { expect(backend.appendAttempts).toBe(1) }) await firstFiber.dispose() + const baselineLoads = backend.loadAttempts + backend.beforeLoadStored = async () => { await loadGate.promise } const coldLoad = coordinator.load(id) + appendGate.resolve(true) + await vi.waitFor(() => { expect(backend.loadAttempts).toBe(baselineLoads + 1) }) + let reuse!: Session await ctx.plugin(Object.assign((inner: Context) => { reuse = inner.sessions.create(id) }, { inject: ['sessions'] })) const reuseFlush = ctx.sessions.flush(reuse) - appendGate.resolve(true) + loadGate.resolve(true) await expect(coldLoad).resolves.toMatchObject({ events: [{ seq: 0 }, { seq: 1 }], }) @@ -385,6 +391,7 @@ describe('PersistenceCoordinator retirement', () => { }) } finally { appendGate.resolve(true) + loadGate.resolve(true) await backendFiber.dispose() await ctx.fiber.dispose() }