refactor(session-persistence): add borrowable prepared sessions

This commit is contained in:
imccyu
2026-08-25 06:05:50 +08:00
parent f8b0ca046f
commit 7f4cdc809c
10 changed files with 412 additions and 6 deletions
@@ -142,6 +142,10 @@ class TestPersistence extends SessionPersistence {
: Promise.resolve(stored)
}
borrowSession(_id: SessionId, _signal?: AbortSignal): ReturnType<SessionPersistence['borrowSession']> {
return Promise.reject(new Error('not used'))
}
async readFrom(
id: SessionId,
fromSeq: number,
@@ -24,6 +24,9 @@ class TestPersistence extends SessionPersistence {
inspect(_id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
return Promise.reject(new Error('not used'))
}
borrowSession(_id: SessionId, _signal?: AbortSignal): ReturnType<SessionPersistence['borrowSession']> {
return Promise.reject(new Error('not used'))
}
readFrom(_id: SessionId, _fromSeq: number): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
return Promise.reject(new Error('not used'))
}
@@ -17,8 +17,10 @@ import { randomBytes } from 'node:crypto'
import {
DEFAULT_PREPARED_SESSION_CACHE_SIZE, DEFAULT_WRITE_BATCH_MAX_DELAY_MS, MAX_WRITE_BATCH_DELAY_MS,
SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator, SessionFormatUnsupportedError,
type BorrowedSessionSource,
type PersistenceBackend, type SessionLocation, type SessionPersistenceSnapshot,
type SessionInspection, type SessionPersistenceRevision as PersistenceRevision, type SessionRawArtifact,
type SessionInspection,
type SessionPersistenceRevision as PersistenceRevision, type SessionRawArtifact,
type StoredPrefix,
} from '@deepseek-ai/dsh-session-persistence'
import type { Session, SessionEvent, SessionId, SessionHeader, SessionPreparation } from '@deepseek-ai/dsh-session'
@@ -197,6 +199,10 @@ export class JsonlSessionPersistence extends SessionPersistence implements Persi
return this.coordinator.inspect(id, signal)
}
override borrowSession(id: SessionId, signal?: AbortSignal): Promise<BorrowedSessionSource> {
return this.coordinator.borrowSession(id, signal)
}
// JSONL is sequential media: no loadStoredFrom hook, so the coordinator
// parses the stored prefix (both encodings) and skips forward to fromSeq.
readFrom(id: SessionId, fromSeq: number, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
@@ -17,6 +17,7 @@ import {
DEFAULT_PREPARED_SESSION_CACHE_SIZE,
DEFAULT_WRITE_BATCH_MAX_DELAY_MS,
MAX_WRITE_BATCH_DELAY_MS,
type BorrowedSessionSource,
PersistenceCoordinator,
SessionPersistence,
type SessionInspection,
@@ -119,6 +120,10 @@ export class SqliteSessionPersistence extends SessionPersistence {
return this.coordinator.inspect(id, signal)
}
override borrowSession(id: SessionId, signal?: AbortSignal): Promise<BorrowedSessionSource> {
return this.coordinator.borrowSession(id, signal)
}
readFrom(
id: SessionId,
fromSeq: number,
@@ -17,7 +17,8 @@ import {
} from '@deepseek-ai/dsh-session'
import type { Session, SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
import type { SessionInspection, SessionLocation } from './index.ts'
import type { BorrowedSessionSource, SessionInspection, SessionLocation } from './index.ts'
import { SessionPersistenceNotFoundError } from './errors.ts'
import type { SessionPersistenceRevision } from './revision.ts'
import { observeQueuedAbort, SessionPreparations } from './preparations.ts'
import type { SessionPreparationReservation } from './preparations.ts'
@@ -841,6 +842,64 @@ export class PersistenceCoordinator<TornMarker = unknown> {
}
}
/**
* Borrow one exact logical view while pinning its reusable prepared Session.
* @param id - persisted session to observe.
* @param signal - optional cancellation for preparation work.
* @returns a disposable observation retaining the prepared source.
*/
async borrowSession(id: SessionId, signal?: AbortSignal): Promise<BorrowedSessionSource> {
for (;;) {
signal?.throwIfAborted()
if (this.retirements.has(id)) await this.waitForRetirement(id, signal)
const live = this.ctx.sessions.get(id)
if (live !== undefined) {
return { source: 'live', inspection: this.inspectLive(live), [Symbol.dispose]: () => {} }
}
const observation = await this.preparations.borrow(
id,
() => this.serialize(id, () => this.prepareCore(id)),
signal,
)
const source = observation.source
try {
const attached = this.ctx.sessions.get(id)
if (attached !== undefined) {
observation[Symbol.dispose]()
return { source: 'live', inspection: this.inspectLive(attached), [Symbol.dispose]: () => {} }
}
const current = await this.serialize(
id,
() => this.isPreparedSourceCurrent(source, signal),
signal,
)
const published = this.ctx.sessions.get(id)
if (published !== undefined) {
observation[Symbol.dispose]()
return { source: 'live', inspection: this.inspectLive(published), [Symbol.dispose]: () => {} }
}
if (current || this.preparations.discardReady(id, source) === 'retained') {
return {
source: 'prepared',
inspection: source.inspection,
revision: source.revision,
preparedSession: source.session,
[Symbol.dispose]: () => { observation[Symbol.dispose]() },
}
}
} catch (error: unknown) {
observation[Symbol.dispose]()
signal?.throwIfAborted()
const attached = this.ctx.sessions.get(id)
if (attached !== undefined) {
return { source: 'live', inspection: this.inspectLive(attached), [Symbol.dispose]: () => {} }
}
throw error
}
observation[Symbol.dispose]()
}
}
/**
* Read the stored events from `fromSeq` onward, detached and non-mutating
* (the read-from-seq primitive behind the service's `readFrom`). Runs on
@@ -876,7 +935,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
throw error
}
signal?.throwIfAborted()
if (suffix === undefined) throw new Error(`session "${id}" not found`)
if (suffix === undefined) throw new SessionPersistenceNotFoundError(id)
this.assertStoredId(id, suffix.meta)
this.assertVersion(suffix.meta)
if (suffix.events.some(needsLegacyPrefix)) {
@@ -900,7 +959,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
signal?.throwIfAborted()
const stored = await this.backend.loadStored(id, signal)
signal?.throwIfAborted()
if (stored === undefined) throw new Error(`session "${id}" not found`)
if (stored === undefined) throw new SessionPersistenceNotFoundError(id)
this.assertStoredId(id, stored.meta)
this.assertVersion(stored.meta)
const events = snapshotStoredEvents(stored.events, id)
@@ -914,7 +973,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
/** Read, repair in memory, validate, and freeze one cold source once. */
private async prepareCore(id: SessionId): Promise<PreparedSessionSource<TornMarker>> {
const stored = await this.backend.loadStored(id)
if (stored === undefined) throw new Error(`session "${id}" not found`)
if (stored === undefined) throw new SessionPersistenceNotFoundError(id)
try {
const { meta, events, revision, tornMarker } = stored
this.assertStoredId(id, meta)
@@ -0,0 +1,12 @@
/** Stable failures exposed by the session-persistence service. */
import type { SessionId } from '@deepseek-ai/dsh-session'
/** The requested Session identity has no materialized durable log. */
export class SessionPersistenceNotFoundError extends Error {
/** @param sessionId - absent durable Session identity. */
constructor(readonly sessionId: SessionId) {
super(`session "${sessionId}" not found`)
this.name = 'SessionPersistenceNotFoundError'
}
}
@@ -13,6 +13,7 @@ import type { SessionPersistenceRevision } from './revision.ts'
// Re-export the metadata vocabulary so Consumers import it from the Service Definition.
export type { SessionHeader } from '@deepseek-ai/dsh-session'
export { SessionPersistenceRevision } from './revision.ts'
export { SessionPersistenceNotFoundError } from './errors.ts'
/** Lightweight immutable source identity returned without loading a full log. */
export interface SessionPersistenceSnapshot {
@@ -30,6 +31,26 @@ export interface SessionInspection {
readonly events: readonly SessionEvent[]
}
/** A borrowed exact Session source returned from a cold materialization or concurrent live owner. */
export type BorrowedSessionSource = Disposable & (
| {
/** A reusable unpublished Session is pinned until this observation is disposed. */
readonly source: 'prepared'
/** Immutable header and logical event prefix observed together. */
readonly inspection: SessionInspection
/** Durable revision represented by the prepared source. */
readonly revision: SessionPersistenceRevision
/** Exact unpublished Session retained for a later {@link prepare}. */
readonly preparedSession: Session
}
| {
/** A live Session won source resolution while the persistence read was starting. */
readonly source: 'live'
/** Immutable live header and event prefix observed together. */
readonly inspection: SessionInspection
}
)
/** A backend's own raw artifact text for one session, verbatim. */
export interface SessionRawArtifact {
/** The session header parsed from the artifact's own first line. */
@@ -209,6 +230,17 @@ export abstract class SessionPersistence extends Service {
*/
abstract inspect(id: SessionId, signal?: AbortSignal): Promise<SessionInspection>
/**
* Borrow one exact inspection while retaining any reusable prepared source.
* A cold observation must pin the exact prepared Session that a later
* {@link prepare} reserves. Implementations must not degrade this operation
* to a detached {@link inspect} result.
* @param id - persisted session to observe.
* @param signal - optional cancellation for preparation work.
* @returns a disposable immutable observation.
*/
abstract borrowSession(id: SessionId, signal?: AbortSignal): Promise<BorrowedSessionSource>
/**
* Read the stored events from `fromSeq` onward — the read-from-seq
* primitive for read models that resume from a watermark (e.g. a persisted
@@ -19,6 +19,13 @@ interface PreparationEntry<Source, CommitState> {
reservation?: SessionPreparationReservation<Source, CommitState>
reservationSettled?: Promise<void>
settleReservation?: () => void
pins: number
}
/** A borrowed prepared source that remains outside ready-entry eviction until released. */
export interface PreparationLease<Source> extends Disposable {
/** Shared immutable prepared source. */
readonly source: Source
}
/** One exclusively held prepared source and its committed persistence state. */
@@ -64,6 +71,51 @@ export class SessionPreparations<Source extends PreparedSource, CommitState> {
return source
}
/**
* Borrow one prepared source and pin its ready entry against LRU eviction.
* @param id - session identity.
* @param load - cold loader used when no entry exists.
* @param signal - optional cancellation signal while waiting.
* @returns a caller-owned observation lease.
*/
async borrow(
id: SessionId,
load: () => Promise<Source>,
signal?: AbortSignal,
): Promise<PreparationLease<Source>> {
const entry = this.entryFor(id, load)
const pinned = this.entries.get(id) === entry
if (pinned) entry.pins += 1
let loaded: Source
try {
loaded = signal === undefined
? await entry.result
: await observeQueuedAbort(entry.result, signal)
} catch (error: unknown) {
if (pinned && this.entries.get(id) === entry) {
entry.pins -= 1
if (entry.phase === 'ready') this.touch(entry)
}
throw error
}
const source = entry.source ?? loaded
if (this.entries.get(id) !== entry) {
return { source, [Symbol.dispose]: () => {} }
}
if (entry.phase === 'ready') this.touch(entry)
let released = false
return {
source,
[Symbol.dispose]: () => {
if (released) return
released = true
if (this.entries.get(id) !== entry) return
entry.pins -= 1
if (entry.phase === 'ready') this.touch(entry)
},
}
}
/**
* Reserve one ready source after committing its pending durable repair.
* @param id - session identity.
@@ -238,6 +290,7 @@ export class SessionPreparations<Source extends PreparedSource, CommitState> {
id,
result: deferred.promise,
phase: 'loading',
pins: 0,
}
this.entries.set(id, entry)
let loading: Promise<Source>
@@ -291,7 +344,7 @@ export class SessionPreparations<Source extends PreparedSource, CommitState> {
}
if (readyCount <= this.capacity) return
for (const [id, candidate] of this.entries) {
if (candidate.phase !== 'ready') continue
if (candidate.phase !== 'ready' || candidate.pins > 0) continue
this.entries.delete(id)
return
}
@@ -118,6 +118,10 @@ class MemoryPersistence extends SessionPersistence implements PersistenceBackend
.then(loaded => ({ meta: loaded.meta, events: [...loaded.events] }))
}
borrowSession(id: SessionId, signal?: AbortSignal): ReturnType<PersistenceCoordinator['borrowSession']> {
return this.coordinator.borrowSession(id, signal)
}
readFrom(id: SessionId, fromSeq: number, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
return this.coordinator.readFrom(id, fromSeq, signal)
}
@@ -1200,6 +1204,162 @@ describe('PersistenceCoordinator session preparations', () => {
})
describe('PersistenceCoordinator observation cancellation', () => {
it('borrows live Sessions before, during, and after cold source validation', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const backend = new ControlledBackend()
const afterBorrowId = SessionId('borrow-became-live-before-validation')
const afterValidationId = SessionId('borrow-became-live-after-validation')
for (const id of [afterBorrowId, afterValidationId]) {
backend.store.set(id, { meta: meta(id), events: oneTurnLog() })
}
let coordinator!: PersistenceCoordinator<never>
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
coordinator = new PersistenceCoordinator(inner, backend)
}, { inject: ['sessions'] }))
try {
const immediate = ctx.sessions.create(SessionId('borrow-already-live'))
const immediateSource = await coordinator.borrowSession(immediate.id)
expect(immediateSource).toMatchObject({ source: 'live', inspection: { meta: { id: immediate.id } } })
immediateSource[Symbol.dispose]()
const afterBorrow = Session.create(afterBorrowId, oneTurnLog(), meta(afterBorrowId))
const afterBorrowGet = vi.spyOn(ctx.sessions, 'get')
.mockReturnValueOnce(undefined)
.mockReturnValue(afterBorrow)
const attachedSource = await coordinator.borrowSession(afterBorrowId)
expect(attachedSource).toMatchObject({ source: 'live', inspection: { meta: { id: afterBorrowId } } })
attachedSource[Symbol.dispose]()
afterBorrowGet.mockRestore()
const afterValidation = Session.create(afterValidationId, oneTurnLog(), meta(afterValidationId))
const afterValidationGet = vi.spyOn(ctx.sessions, 'get')
.mockReturnValueOnce(undefined)
.mockReturnValueOnce(undefined)
.mockReturnValue(afterValidation)
const publishedSource = await coordinator.borrowSession(afterValidationId)
expect(publishedSource).toMatchObject({
source: 'live', inspection: { meta: { id: afterValidationId } },
})
publishedSource[Symbol.dispose]()
afterValidationGet.mockRestore()
} finally {
await fiber.dispose()
await ctx.fiber.dispose()
}
})
it('returns and releases a current prepared observation', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const backend = new ControlledBackend()
const id = SessionId('borrow-current-prepared')
backend.store.set(id, { meta: meta(id), events: oneTurnLog() })
let coordinator!: PersistenceCoordinator<never>
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
coordinator = new PersistenceCoordinator(inner, backend)
}, { inject: ['sessions'] }))
try {
const source = await coordinator.borrowSession(id)
expect(source).toMatchObject({ source: 'prepared', inspection: { meta: { id } } })
source[Symbol.dispose]()
} finally {
await fiber.dispose()
await ctx.fiber.dispose()
}
})
it('reloads a stale prepared observation and retains one claimed concurrently', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const backend = new ControlledBackend()
const staleId = SessionId('borrow-stale-prepared')
const retainedId = SessionId('borrow-retained-prepared')
for (const id of [staleId, retainedId]) {
backend.store.set(id, { meta: meta(id), events: oneTurnLog() })
}
let coordinator!: PersistenceCoordinator<never>
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
coordinator = new PersistenceCoordinator(inner, backend)
}, { inject: ['sessions'] }))
try {
const readRevision = backend.readStoredRevision.bind(backend)
const revision = vi.spyOn(backend, 'readStoredRevision')
.mockResolvedValueOnce(SessionPersistenceRevision('stale'))
.mockImplementation(readRevision)
const stale = await coordinator.borrowSession(staleId)
expect(stale.source).toBe('prepared')
expect(backend.loadAttempts).toBe(2)
stale[Symbol.dispose]()
revision.mockRestore()
const preparations = (coordinator as unknown as {
preparations: { discardReady: (id: SessionId, source: unknown) => string }
}).preparations
vi.spyOn(backend, 'readStoredRevision').mockResolvedValue(SessionPersistenceRevision('changed'))
const discard = vi.spyOn(preparations, 'discardReady').mockReturnValue('retained')
const retained = await coordinator.borrowSession(retainedId)
expect(retained.source).toBe('prepared')
expect(discard).toHaveBeenCalledOnce()
retained[Symbol.dispose]()
} finally {
await fiber.dispose()
await ctx.fiber.dispose()
}
})
it('falls back to a concurrently attached Session after revision validation fails', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const backend = new ControlledBackend()
const id = SessionId('borrow-failed-validation-became-live')
backend.store.set(id, { meta: meta(id), events: oneTurnLog() })
let coordinator!: PersistenceCoordinator<never>
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
coordinator = new PersistenceCoordinator(inner, backend)
}, { inject: ['sessions'] }))
const attached = Session.create(id, oneTurnLog(), meta(id))
const get = vi.spyOn(ctx.sessions, 'get')
.mockReturnValueOnce(undefined)
.mockReturnValueOnce(undefined)
.mockReturnValue(attached)
vi.spyOn(backend, 'readStoredRevision').mockRejectedValue(new Error('revision failed'))
try {
const source = await coordinator.borrowSession(id)
expect(source).toMatchObject({ source: 'live', inspection: { meta: { id } } })
source[Symbol.dispose]()
} finally {
get.mockRestore()
await fiber.dispose()
await ctx.fiber.dispose()
}
})
it('rethrows revision validation failure when no live Session won the race', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const backend = new ControlledBackend()
const id = SessionId('borrow-failed-validation')
backend.store.set(id, { meta: meta(id), events: oneTurnLog() })
const failure = new Error('revision failed')
vi.spyOn(backend, 'readStoredRevision').mockRejectedValue(failure)
let coordinator!: PersistenceCoordinator<never>
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
coordinator = new PersistenceCoordinator(inner, backend)
}, { inject: ['sessions'] }))
try {
await expect(coordinator.borrowSession(id)).rejects.toBe(failure)
} finally {
await fiber.dispose()
await ctx.fiber.dispose()
}
})
it('promptly rejects a queued inspect without invoking it and keeps the same-id chain healthy', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
@@ -160,6 +160,78 @@ describe('SessionPreparations inspection', () => {
})
})
describe('SessionPreparations borrowing', () => {
it('returns a detached lease when loading invalidates its own entry', async () => {
const preparations = new SessionPreparations<PreparedSource, string>(1)
const id = SessionId('borrow-invalidated-load')
const source = prepared(id)
const lease = await preparations.borrow(id, () => {
preparations.invalidate(id)
return Promise.resolve(source)
})
expect(lease.source).toBe(source)
expect(preparations.has(id)).toBe(false)
expect(() => { lease[Symbol.dispose]() }).not.toThrow()
})
it('releases pins after cancellation while loading and after readiness', async () => {
const preparations = new SessionPreparations<PreparedSource, string>(1)
const loadingId = SessionId('borrow-cancelled-loading')
const loading = Promise.withResolvers<PreparedSource>()
const loadingAbort = new AbortController()
const pending = preparations.borrow(loadingId, () => loading.promise, loadingAbort.signal)
loadingAbort.abort(new Error('cancelled while loading'))
await expect(pending).rejects.toThrow('cancelled while loading')
loading.resolve(prepared(loadingId))
await loading.promise
await Promise.resolve()
const readyId = SessionId('borrow-cancelled-ready')
const ready = prepared(readyId)
await preparations.inspect(readyId, () => Promise.resolve(ready))
const readyAbort = new AbortController()
readyAbort.abort(new Error('cancelled while ready'))
await expect(preparations.borrow(readyId, () => Promise.resolve(ready), readyAbort.signal))
.rejects.toThrow('cancelled while ready')
await preparations.inspect(SessionId('borrow-eviction'), () => Promise.resolve(prepared('borrow-eviction')))
expect(preparations.has(loadingId)).toBe(false)
})
it('makes borrowed lease disposal idempotent across ready, invalidated, and reserved entries', async () => {
const preparations = new SessionPreparations<PreparedSource, string>(3)
const ready = prepared('borrow-ready-release')
const readyLease = await preparations.borrow(ready.session.id, () => Promise.resolve(ready))
readyLease[Symbol.dispose]()
readyLease[Symbol.dispose]()
const invalidated = prepared('borrow-invalidated-release')
const invalidatedLease = await preparations.borrow(
invalidated.session.id,
() => Promise.resolve(invalidated),
)
preparations.invalidate(invalidated.session.id)
invalidatedLease[Symbol.dispose]()
const reserved = prepared('borrow-reserved-release')
const reservation = await preparations.reserve(
reserved.session.id,
() => Promise.resolve(reserved),
committed,
)
expect(reservation).toBeDefined()
const reservedLease = await preparations.borrow(
reserved.session.id,
() => Promise.resolve(prepared('unused')),
)
reservedLease[Symbol.dispose]()
preparations.release(reservation!, false)
})
})
describe('SessionPreparations reservation', () => {
it('waits for an existing reservation, republishes the exact Session, and attaches once', async () => {
const preparations = new SessionPreparations<PreparedSource, string>(2)