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

fix: reserve cold loads across repair

Browse Source
This commit is contained in:
_Kerman
2026-07-23 20:58:20 +08:00
parent 816907ba2d
commit 4e5166828b
10 changed files with 79 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -462,7 +462,7 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [
},
{
signature: 'abstract load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }>',
jsDoc: '/**\n * Load a header and balanced contiguous log. A complete interrupted final\n * turn is preserved and durably closed with missing tool errors plus any open\n * step and turn boundaries; only a torn final record is discarded. Unknown\n * versions and corruption in the committed prefix reject. Implementations\n * MUST NOT crash-repair an identity still bound to a live Session: a balanced\n * live log may return as a durable snapshot, while an open live turn rejects.\n * @param id - the persisted session to reload.\n * @returns the header and a log ending on a balanced `turn/end`.\n */',
jsDoc: '/**\n * Load a header and balanced contiguous log. A complete interrupted final\n * turn is preserved and durably closed with missing tool errors plus any open\n * step and turn boundaries; only a torn final record is discarded. Unknown\n * versions and corruption in the committed prefix reject. Implementations\n * MUST NOT crash-repair an identity still bound to a live Session: a balanced\n * live log may return as a durable snapshot, while an open live turn rejects.\n * A coordinator-backed cold load reserves the identity across storage awaits,\n * so concurrent publication of a same-id live Session rejects.\n * @param id - the persisted session to reload.\n * @returns the header and a log ending on a balanced `turn/end`.\n */',
},
{
signature: 'abstract list(): Promise<SessionHeader[]>',

2
packages/session-persistence/session-persistence/README.md
View File

@@ -27,7 +27,7 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l
Each `session/event` copies its event into the session controller and starts an eager drain without blocking the producer. Concurrent notifications share the current drain; events admitted during a write remain pending and trigger the next batch. `session/flush` is an observation barrier that waits until the controller has no current or pending batch. An eager failure is logged and retains the batch; the next explicit flush or backend teardown retries it and surfaces failure to its caller.
Crash repair is cold-only. For a live id, `load(id)` snapshots the authoritative in-memory log, waits for that snapshot to become durable, and returns it only when balanced; an open live turn rejects instead of receiving synthetic interruption closers. A cold id follows storage repair normally. HMR adoption likewise reads through `loadStored`, applies the coordinator's cwd check, and never closes the active turn.
Crash repair is cold-only. For a live id, `load(id)` snapshots the authoritative in-memory log, waits for that snapshot to become durable, and returns it only when balanced; an open live turn rejects instead of receiving synthetic interruption closers. A cold load reserves its id across backend reads and repair writes, so concurrent publication of a same-id live `Session` rejects and rolls back. HMR adoption reads through `loadStored`, applies the coordinator's cwd check, and never closes the active turn.
When a live session emits `session/disposed`, the coordinator waits for its controller, serializes a final drain, then releases state owned by that exact `Session` object. Failed retirement leaves the controller in the live-session map, so backend teardown can retry it. Backend teardown stops event admission first, flushes every remaining controller, awaits per-id operations, and only then closes the storage handle.

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

@@ -154,6 +154,8 @@ export class PersistenceCoordinator<TornMarker = unknown> {
private states = new Map<SessionId, SessionState>()
/** Lifecycle and write-behind state keyed by the exact live Session. */
private live = new Map<Session, LiveSessionState>()
/** Cold loads currently reserving an id across backend reads and repair writes. */
private coldLoads = new Set<SessionId>()
/**
* Per-session serialization: every operation chains onto the prior one for the
* same id, so writes for one session never interleave. Keyed by session id.
@@ -251,7 +253,12 @@ export class PersistenceCoordinator<TornMarker = unknown> {
const selected = await this.serialize(id, async () => {
const live = this.ctx.sessions.get(id)
if (live !== undefined) return { live }
return { loaded: await this.loadCore(id) }
this.coldLoads.add(id)
try {
return { loaded: await this.loadCore(id) }
} finally {
this.coldLoads.delete(id)
}
})
return 'loaded' in selected ? selected.loaded : this.loadLiveSnapshot(selected.live)
}
@@ -372,7 +379,12 @@ export class PersistenceCoordinator<TornMarker = unknown> {
}, `${this.backend.name} write path`)
// Capture the header on creation and persist a fork's seed once.
ctx.on('session/created', (session) => { void this.initFor(session) })
ctx.on('session/created', (session) => {
if (this.coldLoads.has(session.id)) {
throw new Error(`cannot publish session "${session.id}" while its persisted history is loading`)
}
void this.initFor(session)
})
// Keep a persistence-owned copy of each frozen event and start an eager drain.
ctx.on('session/event', (session, event) => {
@@ -394,6 +406,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
/** Start and observe one disposed session's final drain. */
private retire(session: Session): void {
if (!this.live.has(session)) return
void this.retireCore(session).catch((error: unknown) => {
this.ctx.logger.warn(`${this.backend.name}: session "${session.id}" retirement failed: ${String(error)}`)
})

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

@@ -79,6 +79,8 @@ export abstract class SessionPersistence extends Service {
* versions and corruption in the committed prefix reject. Implementations
* MUST NOT crash-repair an identity still bound to a live Session: a balanced
* live log may return as a durable snapshot, while an open live turn rejects.
* A coordinator-backed cold load reserves the identity across storage awaits,
* so concurrent publication of a same-id live Session rejects.
* @param id - the persisted session to reload.
* @returns the header and a log ending on a balanced `turn/end`.
*/

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

@@ -293,6 +293,48 @@ describe('PersistenceCoordinator stored identity', () => {
await ctx.fiber.dispose()
}
})
it('reserves a cold id across asynchronous storage repair', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const backend = new ControlledBackend()
const id = SessionId('cold-load-reservation')
const header = meta(id)
const start: SessionEvent = {
type: 'turn/start',
seq: 0,
time: 1,
data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } },
}
backend.store.set(id, { meta: header, events: [start] })
const loadGate = Promise.withResolvers<boolean>()
backend.beforeLoadStored = async () => { await loadGate.promise }
let coordinator!: PersistenceCoordinator<never>
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
coordinator = new PersistenceCoordinator(inner, backend)
}, { inject: ['sessions'] }))
try {
const loading = coordinator.load(id)
await vi.waitFor(() => { expect(backend.loadAttempts).toBe(1) })
await expect(ctx.plugin(Object.assign((inner: Context) => {
inner.sessions.create(id, { seed: [start], meta: header })
}, { inject: ['sessions'] }))).rejects.toThrow(/persisted history is loading/)
expect(ctx.sessions.get(id)).toBeUndefined()
loadGate.resolve(true)
const loaded = await loading
expect(loaded.events.map(event => event.type)).toEqual(['turn/start', 'turn/end'])
const resumed = ctx.sessions.create(id, { seed: loaded.events, meta: loaded.meta })
await expect(ctx.sessions.flush(resumed)).resolves.toBeUndefined()
} finally {
loadGate.resolve(true)
await fiber.dispose()
await ctx.fiber.dispose()
}
})
})
describe('PersistenceCoordinator retirement', () => {
@@ -399,17 +441,20 @@ describe('PersistenceCoordinator retirement', () => {
appendGate.resolve(true)
await vi.waitFor(() => { expect(backend.loadAttempts).toBe(baselineLoads + 1) })
let reuse!: Session
await ctx.plugin(Object.assign((inner: Context) => {
reuse = inner.sessions.create(id)
}, { inject: ['sessions'] }))
const reuseFlush = ctx.sessions.flush(reuse)
await expect(ctx.plugin(Object.assign((inner: Context) => {
inner.sessions.create(id)
}, { inject: ['sessions'] }))).rejects.toThrow(/persisted history is loading/)
loadGate.resolve(true)
await expect(coldLoad).resolves.toMatchObject({
events: [{ seq: 0 }, { seq: 1 }],
})
await expect(reuseFlush).rejects.toThrow(/id collision/)
let reuse!: Session
await ctx.plugin(Object.assign((inner: Context) => {
reuse = inner.sessions.create(id)
}, { inject: ['sessions'] }))
await expect(ctx.sessions.flush(reuse)).rejects.toThrow(/id collision/)
await vi.waitFor(() => {
expect(backend.store.get(id)?.events.map(event => event.seq)).toEqual([0, 1])
})