diff --git a/packages/feedback/message-feedback/tests/helpers.ts b/packages/feedback/message-feedback/tests/helpers.ts index f902ba38d7..f653ac6d63 100644 --- a/packages/feedback/message-feedback/tests/helpers.ts +++ b/packages/feedback/message-feedback/tests/helpers.ts @@ -142,6 +142,10 @@ class TestPersistence extends SessionPersistence { : Promise.resolve(stored) } + borrowSession(_id: SessionId, _signal?: AbortSignal): ReturnType { + return Promise.reject(new Error('not used')) + } + async readFrom( id: SessionId, fromSeq: number, diff --git a/packages/session/session-checkpoint-policy/tests/session-checkpoint-policy.spec.ts b/packages/session/session-checkpoint-policy/tests/session-checkpoint-policy.spec.ts index 21a04c90c2..851271aa9c 100644 --- a/packages/session/session-checkpoint-policy/tests/session-checkpoint-policy.spec.ts +++ b/packages/session/session-checkpoint-policy/tests/session-checkpoint-policy.spec.ts @@ -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 { + return Promise.reject(new Error('not used')) + } readFrom(_id: SessionId, _fromSeq: number): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { return Promise.reject(new Error('not used')) } diff --git a/packages/session/session-persistence-jsonl/src/index.ts b/packages/session/session-persistence-jsonl/src/index.ts index dab2a75657..4bed7aefb9 100644 --- a/packages/session/session-persistence-jsonl/src/index.ts +++ b/packages/session/session-persistence-jsonl/src/index.ts @@ -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 { + 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[] }> { diff --git a/packages/session/session-persistence-sqlite/src/index.ts b/packages/session/session-persistence-sqlite/src/index.ts index 50204be7ab..3d14d65b17 100644 --- a/packages/session/session-persistence-sqlite/src/index.ts +++ b/packages/session/session-persistence-sqlite/src/index.ts @@ -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 { + return this.coordinator.borrowSession(id, signal) + } + readFrom( id: SessionId, fromSeq: number, diff --git a/packages/session/session-persistence/src/coordinator.ts b/packages/session/session-persistence/src/coordinator.ts index 3b99afb723..37c9558137 100644 --- a/packages/session/session-persistence/src/coordinator.ts +++ b/packages/session/session-persistence/src/coordinator.ts @@ -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 { } } + /** + * 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 { + 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 { 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 { 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 { /** Read, repair in memory, validate, and freeze one cold source once. */ private async prepareCore(id: SessionId): Promise> { 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) diff --git a/packages/session/session-persistence/src/errors.ts b/packages/session/session-persistence/src/errors.ts new file mode 100644 index 0000000000..0731b0edc7 --- /dev/null +++ b/packages/session/session-persistence/src/errors.ts @@ -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' + } +} diff --git a/packages/session/session-persistence/src/index.ts b/packages/session/session-persistence/src/index.ts index e9c75107bb..627098c648 100644 --- a/packages/session/session-persistence/src/index.ts +++ b/packages/session/session-persistence/src/index.ts @@ -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 + /** + * 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 + /** * Read the stored events from `fromSeq` onward — the read-from-seq * primitive for read models that resume from a watermark (e.g. a persisted diff --git a/packages/session/session-persistence/src/preparations.ts b/packages/session/session-persistence/src/preparations.ts index 2a685f71f9..96ad4352f7 100644 --- a/packages/session/session-persistence/src/preparations.ts +++ b/packages/session/session-persistence/src/preparations.ts @@ -19,6 +19,13 @@ interface PreparationEntry { reservation?: SessionPreparationReservation reservationSettled?: Promise settleReservation?: () => void + pins: number +} + +/** A borrowed prepared source that remains outside ready-entry eviction until released. */ +export interface PreparationLease 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 { 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, + signal?: AbortSignal, + ): Promise> { + 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 { id, result: deferred.promise, phase: 'loading', + pins: 0, } this.entries.set(id, entry) let loading: Promise @@ -291,7 +344,7 @@ export class SessionPreparations { } 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 } diff --git a/packages/session/session-persistence/tests/persistence.spec.ts b/packages/session/session-persistence/tests/persistence.spec.ts index d82627ffe9..a596ed0b11 100644 --- a/packages/session/session-persistence/tests/persistence.spec.ts +++ b/packages/session/session-persistence/tests/persistence.spec.ts @@ -118,6 +118,10 @@ class MemoryPersistence extends SessionPersistence implements PersistenceBackend .then(loaded => ({ meta: loaded.meta, events: [...loaded.events] })) } + borrowSession(id: SessionId, signal?: AbortSignal): ReturnType { + 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 + 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 + 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 + 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 + 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 + 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) diff --git a/packages/session/session-persistence/tests/preparations.spec.ts b/packages/session/session-persistence/tests/preparations.spec.ts index 5e12f29a79..7635ffd223 100644 --- a/packages/session/session-persistence/tests/preparations.spec.ts +++ b/packages/session/session-persistence/tests/preparations.spec.ts @@ -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(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(1) + const loadingId = SessionId('borrow-cancelled-loading') + const loading = Promise.withResolvers() + 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(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(2)