diff --git a/packages/session-persistence/session-persistence/src/coordinator.ts b/packages/session-persistence/session-persistence/src/coordinator.ts index 602ac43f72..46db6eee62 100644 --- a/packages/session-persistence/session-persistence/src/coordinator.ts +++ b/packages/session-persistence/session-persistence/src/coordinator.ts @@ -346,7 +346,7 @@ export class PersistenceCoordinator { ctx.effect(() => async () => { let disposeError: unknown try { - const errors = await settledErrors([...this.live.keys()].map(session => this.flushForDispose(session))) + const errors = await settledErrors([...this.live.keys()].map(session => this.flush(session))) while (this.chains.size > 0) await Promise.allSettled([...this.chains.values()]) if (errors.length > 0) { throw new AggregateError(errors, `${this.backend.name} dispose failed`) @@ -474,7 +474,12 @@ export class PersistenceCoordinator { if (suffix.length > 0) await this.appendCore(id, suffix) return } - throw new Error(`session "${id}" is already bound to a different live session in this backend (id collision)`) + const owner = this.live.get(tracked.owner) + if (!tracked.materialized && !owner?.pending.length) { + this.states.delete(id) + } else { + throw new Error(`session "${id}" is already bound to a different live session in this backend (id collision)`) + } } // case 2/3: an artifact at THIS cwd is adopted as a live prefix (or rejected @@ -530,18 +535,14 @@ export class PersistenceCoordinator { private async flush(session: Session): Promise { const live = this.initFor(session) await live.init + const overlapping = live.flush + if (overlapping !== undefined) await Promise.allSettled([overlapping]) while (live.flush !== undefined || live.pending.length > 0) { - await this.ensureFlush(session, live) + if (live.flush !== undefined) await live.flush + else await this.ensureFlush(session, live) } } - /** Let an eager attempt settle, then make one teardown-owned retry observable. */ - private async flushForDispose(session: Session): Promise { - const current = this.live.get(session)?.flush - if (current !== undefined) await Promise.allSettled([current]) - await this.flush(session) - } - /** Start an eager drain without exposing its failure to the synchronous append. */ private scheduleDrain(session: Session, live: LiveSessionState): void { void this.ensureFlush(session, live).catch((error: unknown) => { @@ -549,9 +550,8 @@ export class PersistenceCoordinator { }) } - /** Return the current drain, or start one for the complete pending batch. */ + /** Start one drain for the complete pending batch. */ private ensureFlush(session: Session, live: LiveSessionState): Promise { - if (live.flush !== undefined) return live.flush const flush = live.init .then(() => this.serialize(session.header.id, () => this.drain(session.header.id, live))) .finally(() => { live.flush = undefined }) diff --git a/packages/session-persistence/session-persistence/tests/persistence.spec.ts b/packages/session-persistence/session-persistence/tests/persistence.spec.ts index 7c15d52a4a..1ac6977134 100644 --- a/packages/session-persistence/session-persistence/tests/persistence.spec.ts +++ b/packages/session-persistence/session-persistence/tests/persistence.spec.ts @@ -234,6 +234,41 @@ describe('PersistenceCoordinator eager writes', () => { await ctx.fiber.dispose() } }) + + it('retries a failed overlapping eager write at the explicit flush barrier', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const backend = new ControlledBackend() + const appendGate = Promise.withResolvers() + backend.beforeAppend = async (attempt) => { + if (attempt === 1) { + await appendGate.promise + throw new Error('transient eager failure') + } + } + const fiber = await ctx.plugin(Object.assign((inner: Context) => { + new PersistenceCoordinator(inner, backend) + }, { inject: ['sessions'] })) + + try { + const session = ctx.sessions.create(SessionId('eager-flush-retry')) + await ctx.sessions.flush(session) + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + session.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) + await vi.waitFor(() => { expect(backend.appendAttempts).toBe(1) }) + + const barriers = [ctx.sessions.flush(session), ctx.sessions.flush(session)] + appendGate.resolve(true) + + await expect(Promise.all(barriers)).resolves.toEqual([undefined, undefined]) + expect(backend.appendAttempts).toBe(2) + expect(backend.store.get(session.id)?.events.map(event => event.seq)).toEqual([0, 1]) + } finally { + appendGate.resolve(true) + await fiber.dispose() + await ctx.fiber.dispose() + } + }) }) describe('PersistenceCoordinator retirement', () => { @@ -241,29 +276,32 @@ describe('PersistenceCoordinator retirement', () => { const ctx = new Context() await ctx.plugin(SessionStore) const backend = new ControlledBackend() - let coordinator!: PersistenceCoordinator const backendFiber = await ctx.plugin(Object.assign((inner: Context) => { - coordinator = new PersistenceCoordinator(inner, backend) + new PersistenceCoordinator(inner, backend) }, { inject: ['sessions'] })) + const loadGate = Promise.withResolvers() + backend.beforeLoadStored = async (attempt) => { + if (attempt === 1) await loadGate.promise + } try { const id = SessionId('retiring-lazy-owner') - let first!: Session const firstFiber = await ctx.plugin(Object.assign((inner: Context) => { - first = inner.sessions.create(id) + inner.sessions.create(id) }, { inject: ['sessions'] })) - await ctx.sessions.flush(first) + await vi.waitFor(() => { expect(backend.loadAttempts).toBe(1) }) await firstFiber.dispose() - const internals = coordinator as unknown as CoordinatorInternals - await vi.waitFor(() => { expect(internals.states.has(id)).toBe(false) }) let reuse!: Session await ctx.plugin(Object.assign((inner: Context) => { reuse = inner.sessions.create(id) }, { inject: ['sessions'] })) + const reuseFlush = ctx.sessions.flush(reuse) - await expect(ctx.sessions.flush(reuse)).resolves.toBeUndefined() + loadGate.resolve(true) + await expect(reuseFlush).resolves.toBeUndefined() } finally { + loadGate.resolve(true) await backendFiber.dispose() await ctx.fiber.dispose() }