From 1576d049ed0ecddea72b3064644513f1c976b7e9 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Sat, 8 Aug 2026 16:43:37 +0800 Subject: [PATCH] fix(session-persistence): close batching failure gaps --- .../session-persistence/src/coordinator.ts | 26 +++++++----- .../session-persistence/src/write-behind.ts | 8 +++- .../tests/persistence.spec.ts | 42 +++++++++++++++++++ .../tests/write-behind.spec.ts | 23 ++++++++++ 4 files changed, 87 insertions(+), 12 deletions(-) diff --git a/packages/session-persistence/session-persistence/src/coordinator.ts b/packages/session-persistence/session-persistence/src/coordinator.ts index 0eabaaa8c0..cf013887d3 100644 --- a/packages/session-persistence/session-persistence/src/coordinator.ts +++ b/packages/session-persistence/session-persistence/src/coordinator.ts @@ -1084,14 +1084,12 @@ export class PersistenceCoordinator { return restored } const seed = session.events.map(e => structuredClone(e)) - let init = Promise.resolve() const live: LiveSessionState = { - init, - writes: this.createWriteBehind(session, () => init), + init: Promise.resolve(), + writes: this.createWriteBehind(session, () => live.init), } this.live.set(session, live) - init = this.serialize(session.header.id, () => this.onCreated(session, seed)) - live.init = init + live.init = this.serialize(session.header.id, () => this.onCreated(session, seed)) live.init.catch(() => { /* observed by flush/dispose through the controller */ }) return live } @@ -1110,14 +1108,12 @@ export class PersistenceCoordinator { const suffix = session.events.slice(state.cursor).map(event => structuredClone(event)) this.preparations.attach(reservation) state.owner = session - let init = Promise.resolve() const live: LiveSessionState = { - init, - writes: this.createWriteBehind(session, () => init), + init: Promise.resolve(), + writes: this.createWriteBehind(session, () => live.init), } if (suffix.length > 0) { - init = this.serialize(session.id, () => this.appendCore(session.id, suffix)) - live.init = init + live.init = this.serialize(session.id, () => this.appendCore(session.id, suffix)) live.init.catch(() => { /* observed by flush/dispose through the controller */ }) } return live @@ -1240,7 +1236,15 @@ export class PersistenceCoordinator { private async flush(session: Session): Promise { const live = this.initFor(session) - await live.init + live.writes.cancelAutomaticWait() + try { + await live.init + } catch (error: unknown) { + // Admission is closed during retirement/teardown, but an ordinary flush + // may have raced one last enqueue while initialization was pending. + live.writes.cancelAutomaticWait() + throw error + } await live.writes.flush() } diff --git a/packages/session-persistence/session-persistence/src/write-behind.ts b/packages/session-persistence/session-persistence/src/write-behind.ts index 60665455dc..97797732f6 100644 --- a/packages/session-persistence/session-persistence/src/write-behind.ts +++ b/packages/session-persistence/session-persistence/src/write-behind.ts @@ -71,6 +71,12 @@ export class SessionWriteBehind { return barrier.promise } + /** Cancel the current automatic deadline without draining retained work. */ + cancelAutomaticWait(): void { + this.cancelTimer() + this.deadlineExpired = false + } + /** Start the one fixed window for the current pending prefix. */ private armTimer(): void { this.timer = setTimeout(() => { this.onDeadline() }, this.options.maxDelayMs) @@ -137,7 +143,7 @@ export class SessionWriteBehind { const operation = Promise.resolve().then(() => this.options.write(batch)) const active = operation .catch((error: unknown) => { - this.pending.unshift(...batch) + this.pending = batch.concat(this.pending) this.cancelTimer() this.deadlineExpired = false this.automaticPaused = true diff --git a/packages/session-persistence/session-persistence/tests/persistence.spec.ts b/packages/session-persistence/session-persistence/tests/persistence.spec.ts index 14b4f9b951..75455abe81 100644 --- a/packages/session-persistence/session-persistence/tests/persistence.spec.ts +++ b/packages/session-persistence/session-persistence/tests/persistence.spec.ts @@ -257,6 +257,48 @@ runCoordinatorContract('memory', async (): Promise => { }) describe('PersistenceCoordinator bounded writes', () => { + it('cancels the batching deadline when live initialization rejects', async () => { + vi.useFakeTimers() + const ctx = new Context() + await ctx.plugin(SessionStore) + const backend = new ControlledBackend() + const failure = new Error('initialization failed') + backend.beforeLoadStored = () => Promise.reject(failure) + const fiber = await ctx.plugin(Object.assign((inner: Context) => { + new PersistenceCoordinator(inner, backend, { + preparedSessionCacheSize: DEFAULT_PREPARED_SESSION_CACHE_SIZE, + writeBatchMaxDelayMs: MAX_WRITE_BATCH_DELAY_MS, + }) + }, { inject: ['sessions'] })) + + try { + const session = ctx.sessions.create(SessionId('bounded-init-failure')) + session.append('turn/start', { turn: 1 }) + + await expect(ctx.sessions.flush(session)).rejects.toBe(failure) + expect(vi.getTimerCount()).toBe(0) + try { + await fiber.dispose() + } catch { + // The initialization failure was already asserted at the flush boundary. + } + expect(vi.getTimerCount()).toBe(0) + } finally { + try { + await fiber.dispose() + } catch { + // The expected initialization failure was asserted above; cleanup only + // needs to release any remaining parent effects. + } + try { + await ctx.fiber.dispose() + } catch { + // The child failure was already asserted through the backend fiber. + } + vi.useRealTimers() + } + }) + it('starts a follow-up batch for events admitted during an in-flight write', async () => { const ctx = new Context() await ctx.plugin(SessionStore) diff --git a/packages/session-persistence/session-persistence/tests/write-behind.spec.ts b/packages/session-persistence/session-persistence/tests/write-behind.spec.ts index b93b3824ae..df445e0cf7 100644 --- a/packages/session-persistence/session-persistence/tests/write-behind.spec.ts +++ b/packages/session-persistence/session-persistence/tests/write-behind.spec.ts @@ -249,4 +249,27 @@ describe('SessionWriteBehind', () => { expect(batches).toEqual([[0], [0, 1]]) await controller.flush() }) + + it('retains a failed batch larger than the engine call-argument limit', async () => { + const failure = new Error('durability failed') + const batchSize = 150_000 + const sizes: number[] = [] + let attempt = 0 + const controller = new SessionWriteBehind({ + maxDelayMs: 200, + write: async (events) => { + sizes.push(events.length) + if (++attempt === 1) throw failure + }, + reportBackgroundFailure: vi.fn(), + }) + + for (let seq = 0; seq < batchSize; seq += 1) controller.enqueue(event(seq)) + await expect(controller.flush()).rejects.toBe(failure) + expect(controller.hasWork).toBe(true) + + await controller.flush() + expect(sizes).toEqual([batchSize, batchSize]) + expect(controller.hasWork).toBe(false) + }) })