fix(session-persistence): close batching failure gaps

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

@@ -1084,14 +1084,12 @@ export class PersistenceCoordinator<TornMarker = unknown> {
       return restored
     }
     const seed = session.events.map(e => structuredClone(e))
-    let init = Promise.resolve()
     const live: LiveSessionState = {
-      init,
+      init: Promise.resolve(),
-      writes: this.createWriteBehind(session, () => init),
+      writes: this.createWriteBehind(session, () => live.init),
     }
     this.live.set(session, live)
-    init = this.serialize(session.header.id, () => this.onCreated(session, seed))
+    live.init = this.serialize(session.header.id, () => this.onCreated(session, seed))
-    live.init = init
     live.init.catch(() => { /* observed by flush/dispose through the controller */ })
     return live
   }
@@ -1110,14 +1108,12 @@ export class PersistenceCoordinator<TornMarker = unknown> {
     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,
+      init: Promise.resolve(),
-      writes: this.createWriteBehind(session, () => init),
+      writes: this.createWriteBehind(session, () => live.init),
     }
     if (suffix.length > 0) {
-      init = this.serialize(session.id, () => this.appendCore(session.id, suffix))
+      live.init = this.serialize(session.id, () => this.appendCore(session.id, suffix))
-      live.init = init
       live.init.catch(() => { /* observed by flush/dispose through the controller */ })
     }
     return live
@@ -1240,7 +1236,15 @@ export class PersistenceCoordinator<TornMarker = unknown> {
 
   private async flush(session: Session): Promise<void> {
     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()
   }
 

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

packages/session-persistence/session-persistence/tests/persistence.spec.ts

@@ -257,6 +257,48 @@ runCoordinatorContract('memory', async (): Promise<CoordinatorFixture> => {
 })
 
 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)

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)
+  })
 })