fix(session-persistence): address preparation review feedback

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

@@ -22,6 +22,18 @@ import type { SessionPreparationReservation } from './preparations.ts'
 /** Default number of detached session preparations retained by a coordinator. */
 export const DEFAULT_PREPARED_SESSION_CACHE_SIZE = 5
 
+/** Durable session contents failed validation after a successful backend read. */
+export class SessionPersistenceCorruptionError extends Error {
+  /**
+   * @param message - stable corruption context.
+   * @param options - original validation failure.
+   */
+  constructor(message: string, options: ErrorOptions) {
+    super(message, options)
+    this.name = 'SessionPersistenceCorruptionError'
+  }
+}
+
 /** Coordinator policy supplied by a concrete persistence backend. */
 export interface PersistenceCoordinatorOptions {
   /** Maximum completed unpublished preparations retained for reuse. */
@@ -623,7 +635,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
       }
       const reservation = await this.preparations.reserve(
         id,
-        () => this.serialize(id, () => this.prepareCore(id, signal), signal),
+        () => this.serialize(id, () => this.prepareCore(id)),
         source => this.serialize(id, () => this.commitPrepared(source), signal),
         signal,
       )
@@ -674,16 +686,17 @@ export class PersistenceCoordinator<TornMarker = unknown> {
    * Inspect a logical session without publishing it or committing recovery.
    * @param id - persisted session to inspect.
    * @param signal - optional cancellation for preparation work.
-   * @returns immutable prepared metadata and balanced events.
+   * @returns immutable prepared metadata and events; a live view may have an open turn.
    */
   async inspect(id: SessionId, signal?: AbortSignal): Promise<SessionInspection> {
-    await this.waitForRetirement(id, signal)
+    signal?.throwIfAborted()
+    if (this.retirements.has(id)) await this.waitForRetirement(id, signal)
     const live = this.ctx.sessions.get(id)
     if (live !== undefined) return this.inspectLive(live)
     try {
       const source = await this.preparations.inspect(
         id,
-        () => this.serialize(id, () => this.prepareCore(id, signal), signal),
+        () => this.serialize(id, () => this.prepareCore(id)),
         signal,
       )
       const attached = this.ctx.sessions.get(id)
@@ -776,29 +789,36 @@ export class PersistenceCoordinator<TornMarker = unknown> {
     }
     signal?.throwIfAborted()
     if (stored === undefined) throw new Error(`session "${id}" not found`)
-    const { meta, events, tornMarker } = stored
-    this.assertStoredId(id, meta)
-    this.assertVersion(meta)
-    const storedEvents = adoptStoredEvents(events, id)
+    try {
+      const { meta, events, tornMarker } = stored
+      this.assertStoredId(id, meta)
+      this.assertVersion(meta)
+      const storedEvents = adoptStoredEvents(events, id)
 
-    // Preserve complete interrupted events and synthesize only missing closers.
-    const closers = interruptedTurnClosers(storedEvents).map(adoptSessionEvent)
-    const balanced = [...storedEvents, ...closers]
-    const session = this.ctx.sessions.prepare(id, {
-      seed: balanced,
-      meta,
-      seedSource: 'persistence',
-    })
-    const inspection: SessionInspection = Object.freeze({
-      meta: session.header,
-      events: Object.freeze(balanced),
-    })
-    return {
-      inspection,
-      session,
-      sessionLength: session.events.length,
-      tornMarker,
-      closers,
+      // Preserve complete interrupted events and synthesize only missing closers.
+      const closers = interruptedTurnClosers(storedEvents).map(adoptSessionEvent)
+      const balanced = [...storedEvents, ...closers]
+      const session = this.ctx.sessions.prepare(id, {
+        seed: balanced,
+        meta,
+        seedSource: 'persistence',
+      })
+      const inspection: SessionInspection = Object.freeze({
+        meta: session.header,
+        events: Object.freeze(balanced),
+      })
+      return {
+        inspection,
+        session,
+        sessionLength: session.events.length,
+        tornMarker,
+        closers,
+      }
+    } catch (error: unknown) {
+      throw new SessionPersistenceCorruptionError(
+        `stored session "${id}" failed validation: ${String(error)}`,
+        { cause: error },
+      )
     }
   }
 

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

@@ -31,7 +31,11 @@ export interface SessionInspection {
 }
 
 // The backend-agnostic write-path orchestration first-party backends compose.
-export { DEFAULT_PREPARED_SESSION_CACHE_SIZE, PersistenceCoordinator } from './coordinator.ts'
+export {
+  DEFAULT_PREPARED_SESSION_CACHE_SIZE,
+  PersistenceCoordinator,
+  SessionPersistenceCorruptionError,
+} from './coordinator.ts'
 export type {
   PersistenceBackend,
   PersistenceCoordinatorOptions,
@@ -134,14 +138,17 @@ export abstract class SessionPersistence extends Service {
   abstract load(id: SessionId): Promise<SessionInspection>
 
   /**
-   * Inspect an immutable balanced logical session without committing recovery
-   * or publishing it. A complete interrupted turn receives synthetic closers
-   * in memory and a torn physical tail remains untouched. Coordinator-backed
-   * implementations retain the exact unpublished Session for bounded reuse by
-   * a later {@link prepare}; callers borrow only its immutable header and log.
+   * Inspect an immutable logical session without committing recovery or
+   * publishing it. A cold complete interrupted turn receives synthetic closers
+   * in memory and a torn physical tail remains untouched. An already-live
+   * Session instead yields its current immutable snapshot, which may contain an
+   * open turn and its `session/end-seed` boundary. Coordinator-backed
+   * implementations retain the exact cold unpublished Session for bounded
+   * reuse by a later {@link prepare}; callers borrow only its immutable header
+   * and log.
    * @param id - the persisted session to inspect.
    * @param signal - optional cancellation for queued and backend read work.
-   * @returns the validated header and balanced logical event log.
+   * @returns the validated header and current logical event log.
    */
   abstract inspect(id: SessionId, signal?: AbortSignal): Promise<SessionInspection>
 

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

@@ -55,8 +55,8 @@ export class SessionPreparations<Source extends PreparedSource, CommitState> {
     load: () => Promise<Source>,
     signal?: AbortSignal,
   ): Promise<Source> {
-    const { entry, created } = this.entryFor(id, load)
-    const loaded = signal === undefined || created
+    const entry = this.entryFor(id, load)
+    const loaded = signal === undefined
       ? await entry.result
       : await observeQueuedAbort(entry.result, signal)
     const source = entry.source ?? loaded
@@ -78,10 +78,8 @@ export class SessionPreparations<Source extends PreparedSource, CommitState> {
     commit: (source: Source) => Promise<{ source: Source; state: CommitState }>,
     signal?: AbortSignal,
   ): Promise<SessionPreparationReservation<Source, CommitState> | undefined> {
-    const { entry, created } = this.entryFor(id, load)
-    await (signal === undefined || created
-      ? entry.result
-      : observeQueuedAbort(entry.result, signal))
+    const entry = this.entryFor(id, load)
+    await (signal === undefined ? entry.result : observeQueuedAbort(entry.result, signal))
     while (this.entries.get(id) === entry && entry.phase !== 'ready') {
       const settled = entry.reservationSettled
       /* v8 ignore next -- committing/reserved transitions install this waiter synchronously. */
@@ -132,7 +130,7 @@ export class SessionPreparations<Source extends PreparedSource, CommitState> {
       && entry.reservation !== undefined) {
       return entry.reservation
     }
-    throw new Error(`cannot publish session "${session.id}" while a persisted preparation exists`)
+    throw new Error(`cannot publish session "${session.id}": persisted state already owns this identity`)
   }
 
   /**
@@ -216,20 +214,37 @@ export class SessionPreparations<Source extends PreparedSource, CommitState> {
   private entryFor(
     id: SessionId,
     load: () => Promise<Source>,
-  ): { entry: PreparationEntry<Source, CommitState>; created: boolean } {
+  ): PreparationEntry<Source, CommitState> {
     const existing = this.entries.get(id)
-    if (existing !== undefined) return { entry: existing, created: false }
-    const result = Promise.resolve().then(load)
-    const entry: PreparationEntry<Source, CommitState> = { id, result, phase: 'loading' }
+    if (existing !== undefined) return existing
+    const deferred = Promise.withResolvers<Source>()
+    const entry: PreparationEntry<Source, CommitState> = {
+      id,
+      result: deferred.promise,
+      phase: 'loading',
+    }
     this.entries.set(id, entry)
-    void result.then((source) => {
-      if (this.entries.get(id) !== entry) return
-      entry.source = source
-      entry.phase = 'ready'
-    }, () => {
+    let loading: Promise<Source>
+    try {
+      // Start immediately so a same-tick serialized append queues behind this
+      // read. The deferred result settles only after the entry becomes ready.
+      loading = load()
+    } catch (error: unknown) {
       this.remove(entry)
+      deferred.reject(error)
+      return entry
+    }
+    void loading.then((source) => {
+      if (this.entries.get(id) === entry) {
+        entry.source = source
+        entry.phase = 'ready'
+      }
+      deferred.resolve(source)
+    }, (error: unknown) => {
+      this.remove(entry)
+      deferred.reject(error)
     })
-    return { entry, created: true }
+    return entry
   }
 
   private makeReady(entry: PreparationEntry<Source, CommitState>): void {