Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(session-persistence): close preparation races

Browse Source
This commit is contained in:
imccyu
2026-08-06 10:19:51 +08:00
parent 2f98f29992
commit e37aa41f87
10 changed files with 114 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
packages/cordis/tool-cordis/src/api-catalog.ts
View File

@@ -590,7 +590,7 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [
},
{
signature: 'abstract inspect(id: SessionId, signal?: AbortSignal): Promise<SessionInspection>',
jsDoc: '/**\n * Inspect an immutable logical session without committing recovery or\n * publishing it. A cold complete interrupted turn receives synthetic closers\n * in memory and a torn physical tail remains untouched. An already-live\n * Session instead yields its current immutable snapshot, which may contain an\n * open turn and its `session/end-seed` boundary. Coordinator-backed\n * implementations retain the exact cold unpublished Session for bounded\n * reuse by a later {@link prepare}, reloading it when its durable revision\n * changes; callers borrow only its immutable header and log. Continuous\n * external writers may delay revision convergence.\n * @param id - the persisted session to inspect.\n * @param signal - optional cancellation for queued and backend read work.\n * @returns the validated header and current logical event log.\n */',
jsDoc: '/**\n * Inspect an immutable logical session without committing recovery or\n * publishing it. A cold complete interrupted turn receives synthetic closers\n * in memory and a torn physical tail remains untouched. An already-live\n * Session instead yields its current immutable snapshot, which may contain an\n * open turn and its `session/end-seed` boundary. Coordinator-backed\n * implementations retain the exact cold unpublished Session for bounded\n * reuse by a later {@link prepare}. A stale ready source is reloaded; a source\n * already committing or reserved for resume remains exclusive, and inspection\n * may borrow its immutable view. Callers borrow only the immutable header and\n * log. Continuous external writers may delay revision convergence.\n * @param id - the persisted session to inspect.\n * @param signal - optional cancellation for queued and backend read work.\n * @returns the validated header and current logical event log.\n */',
},
{
signature: 'abstract readFrom(id: SessionId, fromSeq: number, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }>',

19
packages/session-persistence/session-persistence/src/coordinator.ts
View File

@@ -704,9 +704,10 @@ export class PersistenceCoordinator<TornMarker = unknown> {
/**
* Inspect a logical session without publishing it or committing recovery.
* Retained cold state is reloaded after its durable revision changes. Revision
* retries converge once the log is stable for one read/check round trip;
* continuous external writers may delay completion.
* A stale ready source is reloaded. A source already committing or reserved
* for resume remains exclusive, and inspection may borrow its immutable view.
* Revision retries converge once the log is stable for one read/check round
* trip; continuous external writers may delay completion.
* @param id - persisted session to inspect.
* @param signal - optional cancellation for preparation work.
* @returns immutable prepared metadata and events; a live view may have an open turn.
@@ -737,6 +738,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
return source.inspection
}
} catch (error: unknown) {
signal?.throwIfAborted()
const attached = this.ctx.sessions.get(id)
if (attached !== undefined) return this.inspectLive(attached)
throw error
@@ -859,14 +861,11 @@ export class PersistenceCoordinator<TornMarker = unknown> {
throw new Error(`session "${id}" already has a live persistence owner`)
}
if (!await this.isPreparedSourceCurrent(source)) return undefined
let committedSource = source
if (source.tornMarker !== undefined || source.closers.length > 0) {
await this.backend.commitRepair(source.inspection.meta, source.tornMarker, source.closers)
const revision = await this.backend.readStoredRevision(id)
if (revision === undefined) {
throw new Error(`session "${id}" disappeared after persistence repair`)
}
committedSource = { ...source, revision, tornMarker: undefined, closers: [] }
// The repair changed the durable revision. Reload the exact committed
// graph instead of associating the old in-memory view with a newer revision.
return undefined
}
const state = existing ?? {
meta: source.inspection.meta,
@@ -878,7 +877,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
state.materialized = true
this.states.set(id, state)
return {
source: committedSource,
source,
state,
}
}

7
packages/session-persistence/session-persistence/src/index.ts
View File

@@ -147,9 +147,10 @@ export abstract class SessionPersistence extends Service {
* 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}, reloading it when its durable revision
* changes; callers borrow only its immutable header and log. Continuous
* external writers may delay revision convergence.
* reuse by a later {@link prepare}. A stale ready source is reloaded; a source
* already committing or reserved for resume remains exclusive, and inspection
* may borrow its immutable view. Callers borrow only the immutable header and
* log. Continuous external writers may delay revision convergence.
* @param id - the persisted session to inspect.
* @param signal - optional cancellation for queued and backend read work.
* @returns the validated header and current logical event log.

7
packages/session-persistence/session-persistence/src/preparations.ts
View File

@@ -184,11 +184,10 @@ export class SessionPreparations<Source extends PreparedSource, CommitState> {
/**
* Discard a prepared view after the durable log changes.
* @param id - changed session identity.
* @param expected - when supplied, invalidate only that exact source.
*/
invalidate(id: SessionId, expected?: Source): void {
invalidate(id: SessionId): void {
const entry = this.entries.get(id)
if (entry !== undefined && (expected === undefined || entry.source === expected)) this.remove(entry)
if (entry !== undefined) this.remove(entry)
}
/**
@@ -219,7 +218,7 @@ export class SessionPreparations<Source extends PreparedSource, CommitState> {
/**
* Remove a completed entry for an already-serialized append adoption.
* @param id - adopted session identity.
* @returns the prepared source, or undefined when no entry exists.
* @returns the prepared source, or undefined when no ready entry exists.
*/
takeReady(id: SessionId): Source | undefined {
const entry = this.entries.get(id)

97
packages/session-persistence/session-persistence/tests/persistence.spec.ts
View File

@@ -875,7 +875,7 @@ describe('PersistenceCoordinator session preparations', () => {
second = await coordinator.prepare(id)
expect(second.session).toBe(first.session)
expect(backend.loadAttempts).toBe(1)
expect(backend.loadAttempts).toBe(2)
expect(backend.repairAttempts).toBe(1)
} finally {
second?.[Symbol.dispose]()
@@ -885,7 +885,52 @@ describe('PersistenceCoordinator session preparations', () => {
}
})
it('rejects preparation when storage disappears after repair', async () => {
it('reloads the committed graph when another writer appends after repair', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const backend = new ControlledBackend()
const id = SessionId('repair-external-append')
backend.store.set(id, {
meta: meta(id),
events: [{ type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } }],
})
const commitRepair = backend.commitRepair.bind(backend)
vi.spyOn(backend, 'commitRepair').mockImplementation(async (header, tornMarker, closers) => {
await commitRepair(header, tornMarker, closers)
const entry = backend.store.get(id)
if (entry === undefined) throw new Error('test repair must keep storage materialized')
const seq = entry.events.length
entry.events.push(
{ type: 'turn/start', seq, time: 3, data: { turn: 2 } },
{ type: 'turn/end', seq: seq + 1, time: 4, data: { turn: 2, reason: { kind: 'completed' } } },
)
})
let coordinator!: PersistenceCoordinator<never>
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
coordinator = new PersistenceCoordinator(inner, backend)
}, { inject: ['sessions'] }))
let preparation: Awaited<ReturnType<typeof coordinator.prepare>> | undefined
try {
preparation = await coordinator.prepare(id)
expect(preparation.session.events.map(event => event.type)).toEqual([
'turn/start',
'turn/end',
'turn/start',
'turn/end',
'session/end-seed',
])
expect(backend.loadAttempts).toBe(2)
expect(backend.repairAttempts).toBe(1)
} finally {
preparation?.[Symbol.dispose]()
await fiber.dispose()
await ctx.fiber.dispose()
}
})
it('rejects preparation when storage disappears during the post-repair reload', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const backend = new ControlledBackend()
@@ -894,24 +939,20 @@ describe('PersistenceCoordinator session preparations', () => {
meta: meta(id),
events: [{ type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } }],
})
const commitRepair = backend.commitRepair.bind(backend)
vi.spyOn(backend, 'commitRepair').mockImplementation(async (header, tornMarker, closers) => {
await commitRepair(header, tornMarker, closers)
backend.store.delete(id)
})
let coordinator!: PersistenceCoordinator<never>
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
coordinator = new PersistenceCoordinator(inner, backend)
}, { inject: ['sessions'] }))
try {
await coordinator.inspect(id)
const readStoredRevision = backend.readStoredRevision.bind(backend)
let revisionReads = 0
vi.spyOn(backend, 'readStoredRevision').mockImplementation((sessionId, signal) => {
revisionReads += 1
if (revisionReads === 2) return Promise.resolve(undefined)
return readStoredRevision(sessionId, signal)
})
await expect(coordinator.prepare(id)).rejects.toThrow(/disappeared after persistence repair/)
await expect(coordinator.prepare(id)).rejects.toThrow(/not found/)
expect(backend.repairAttempts).toBe(1)
expect(revisionReads).toBe(2)
expect(backend.loadAttempts).toBe(2)
} finally {
await fiber.dispose()
await ctx.fiber.dispose()
@@ -1089,6 +1130,36 @@ describe('PersistenceCoordinator observation cancellation', () => {
}
})
it('preserves inspect cancellation when the session concurrently becomes live', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const backend = new ControlledBackend()
const id = SessionId('cancelled-inspect-became-live')
backend.store.set(id, { meta: meta(id), events: oneTurnLog() })
const controller = new AbortController()
const reason = new Error('inspect cancelled while publishing')
backend.beforeLoadStored = async () => {
controller.abort(reason)
throw new Error('load stopped after cancellation')
}
let coordinator!: PersistenceCoordinator<never>
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
coordinator = new PersistenceCoordinator(inner, backend)
}, { inject: ['sessions'] }))
const live = Session.create(id, oneTurnLog(), meta(id))
const get = vi.spyOn(ctx.sessions, 'get')
.mockReturnValueOnce(undefined)
.mockReturnValueOnce(live)
try {
await expect(coordinator.inspect(id, controller.signal)).rejects.toBe(reason)
} finally {
get.mockRestore()
await fiber.dispose()
await ctx.fiber.dispose()
}
})
it('readFrom via the seek hook: serves the suffix, maps undefined to not-found, and relays hook failures by abort state', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)

2
packages/session-persistence/session-persistence/tests/preparations.spec.ts
View File

@@ -34,8 +34,6 @@ describe('SessionPreparations inspection', () => {
await expect(preparations.inspect(id, load)).resolves.toBe(source)
expect(load).toHaveBeenCalledOnce()
preparations.invalidate(id, prepared('different-source'))
expect(preparations.has(id)).toBe(true)
preparations.invalidate(id)
preparations.invalidate(id)
expect(preparations.has(id)).toBe(false)