refactor: eagerly persist session events
This commit is contained in:
_Kerman
@@ -458,7 +458,7 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [
|
||||
},
|
||||
{
|
||||
signature: 'abstract append(id: SessionId, events: readonly SessionEvent[]): Promise<void>',
|
||||
jsDoc: '/**\n * Durably persist a batch of events (called from the write-behind drain at\n * the `session/flush` checkpoint). Honors the append-only and contiguous-seq\n * contracts: the first event\'s `seq` MUST equal the stored next-seq (after\n * `load` has durably closed any interrupted turn). Rejects non-JSON-\n * serializable `event.data` with an error naming the offending event type.\n * @param id - the session the batch belongs to.\n * @param events - the contiguous batch to persist, in seq order.\n */',
|
||||
jsDoc: '/**\n * Durably persist a batch of events. Honors the append-only and contiguous-\n * seq contracts: the first event\'s `seq` MUST equal the stored next-seq\n * (after `load` has durably closed any interrupted turn). Rejects non-JSON-\n * serializable `event.data` with an error naming the offending event type.\n * @param id - the session the batch belongs to.\n * @param events - the contiguous batch to persist, in seq order.\n */',
|
||||
},
|
||||
{
|
||||
signature: 'abstract load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }>',
|
||||
|
||||
@@ -40,7 +40,7 @@ A root belongs to one encoding. Startup discovery and targeted lookup reject the
|
||||
|
||||
## Write path
|
||||
|
||||
The plugin buffers frozen session events and drains them on flush or disposal. A per-session cursor prevents resumed sessions from re-appending stored events, and live sessions are seeded when the plugin loads. Operations for one session are serialized; disposal waits for initialization and the final drain so no write lands after teardown.
|
||||
The plugin copies frozen session events into one controller per live session and starts an eager drain. Concurrent events share the current write; events admitted during it form a follow-up batch, while `session/flush` waits until both current and pending batches are durable. A per-session cursor prevents resumed sessions from re-appending stored events, and live sessions are seeded when the plugin loads. Operations for one session are serialized; disposal drains every retained controller before teardown.
|
||||
|
||||
## Model Experience
|
||||
|
||||
|
||||
@@ -31,7 +31,7 @@ interface Config {
|
||||
|
||||
## Write path
|
||||
|
||||
Like the JSONL backend, the plugin also installs the `session/event` → buffer → `session/flush` drain: it copies each already-frozen event into a persistence-owned buffer, persists a fork's seed once on `session/created`, keeps a per-session write cursor so a resumed session never re-appends stored events, and seeds existing live sessions on apply (HMR does not replay `session/created`). Dispose awaits every in-flight init + final drain and then closes the database, so no write lands after teardown.
|
||||
Like the JSONL backend, the plugin copies each frozen `session/event` into one controller per live session and starts an eager drain. Concurrent events share the current transaction; events admitted during it form a follow-up batch, while `session/flush` waits until both current and pending batches are durable. The controller persists a fork's seed once, keeps a write cursor so resume never re-appends stored events, and seeds live sessions on apply because HMR does not replay `session/created`. Dispose drains every retained controller before closing the database.
|
||||
|
||||
## Model Experience
|
||||
|
||||
|
||||
@@ -10,7 +10,7 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l
|
||||
|---|---|
|
||||
| `locate(meta): SessionLocation \| undefined` | Resolve an absolute per-session artifact target without I/O or materialization. Backends without an independent local artifact return `undefined`. |
|
||||
| `create(meta): Promise<void>` | Register a new session's metadata. MAY defer the physical write until the first `append` (lazy materialization). |
|
||||
| `append(id, events): Promise<void>` | Durably persist a batch (from the `session/flush` drain). Append-only; first event `seq` == stored next-seq after any repair; rejects non-JSON-serializable data naming the offending type. |
|
||||
| `append(id, events): Promise<void>` | Durably persist a batch. Append-only; first event `seq` == stored next-seq after any repair; rejects non-JSON-serializable data naming the offending type. |
|
||||
| `load(id): Promise<{ meta; events }>` | Reload meta + log. Preserves an interrupted (unclosed) final turn and closes it with synthetic closers — an error `tool/result` per unanswered `tool-call`, then `step/end?`+`turn/end {interrupted}` (a turn can be huge — never truncated); only a torn tail fragment is dropped. Events contiguous (`events[i].seq === i`); rejects a committed-region gap/parse error or unknown `version`. |
|
||||
| `list(): Promise<SessionHeader[]>` | Lightweight listing from metadata, no full-log parse. A zero-event lazily-materialized session is absent from `list`. |
|
||||
|
||||
@@ -23,11 +23,11 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l
|
||||
|
||||
## The write coordinator
|
||||
|
||||
`PersistenceCoordinator` owns per-id state, write-behind buffers and serialization, the `session/event` → `session/flush` drain, lazy materialization, crash-tail repair, session adoption, and quiescent disposal. A first-party backend composes one, implements the small `PersistenceBackend` storage hook interface, and delegates its four public service methods. JSONL and SQLite therefore share lifecycle correctness while retaining different storage primitives; see the [coordinator Agent Note](../../../.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md).
|
||||
`PersistenceCoordinator` owns per-id serialization, one eager write controller per live session, lazy materialization, crash-tail repair, session adoption, and quiescent disposal. A first-party backend composes one, implements the small `PersistenceBackend` storage hook interface, and delegates its four public service methods. JSONL and SQLite therefore share lifecycle correctness while retaining different storage primitives; see the [coordinator Agent Note](../../../.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md) and [flush-controller simplification](../../../.agents/notes/implemented/simplification/2026-07-23-collapse-persistence-flush-state.md).
|
||||
|
||||
The coordinator implements durability at checkpoints but does not select their schedule. Persisted deployments explicitly compose [`dsh-session-checkpoint-policy`](../session-checkpoint-policy) when they want request-, tool-dispatch-, and completed-step recovery boundaries; omitting it leaves the loop's coarser checkpoints intact.
|
||||
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.
|
||||
|
||||
When a live session emits `session/disposed`, the coordinator waits for its initialization, serializes a final buffer drain, then releases every map entry owned by that exact `Session` object. A failed final drain keeps the pending buffer for backend teardown to retry. Backend teardown stops event admission first, awaits all in-flight session retirements and remaining per-id operations, drains any retained buffers, and only then closes the storage handle.
|
||||
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.
|
||||
|
||||
The side-effect-free `locate` query remains backend-owned because it describes storage topology rather than write orchestration.
|
||||
|
||||
|
||||
@@ -99,6 +99,13 @@ interface SessionState {
|
||||
owner?: Session
|
||||
}
|
||||
|
||||
/** One live session's initialization and eager write-behind controller. */
|
||||
interface LiveSessionState {
|
||||
pending: SessionEvent[]
|
||||
init: Promise<void>
|
||||
flush: Promise<void> | undefined
|
||||
}
|
||||
|
||||
/** Collect the rejection reasons from a set of promises (none-throwing). */
|
||||
async function settledErrors(promises: Iterable<Promise<unknown>>): Promise<unknown[]> {
|
||||
const settled = await Promise.allSettled([...promises])
|
||||
@@ -153,21 +160,13 @@ function assertSupportedEvents(events: readonly SessionEvent[], id: SessionId):
|
||||
export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
/** Backend bookkeeping keyed by session id (NOT the live Session object). */
|
||||
private states = new Map<SessionId, SessionState>()
|
||||
/** Write-behind buffers keyed by the live Session (write path). */
|
||||
private buffers = new Map<Session, SessionEvent[]>()
|
||||
/** Lifecycle and write-behind state keyed by the exact live Session. */
|
||||
private live = new Map<Session, LiveSessionState>()
|
||||
/**
|
||||
* 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.
|
||||
*/
|
||||
private chains = new Map<SessionId, Promise<unknown>>()
|
||||
/**
|
||||
* Init promises keyed by live session object, preventing an id-reusing
|
||||
* replacement from inheriting stale initialization. Flush is the public
|
||||
* observation boundary; callers do not inspect this bookkeeping directly.
|
||||
*/
|
||||
private inits = new Map<Session, Promise<void>>()
|
||||
/** Final drains started by fire-and-forget session disposal notifications. */
|
||||
private retirements = new Set<Promise<void>>()
|
||||
|
||||
constructor(private ctx: Context, private backend: PersistenceBackend<TornMarker>) {
|
||||
this.installWritePath()
|
||||
@@ -290,7 +289,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
* public methods must NOT call each other (deadlock); they call the unserialized
|
||||
* `*Core` helpers instead.
|
||||
*/
|
||||
private serialize<T>(id: SessionId, op: () => Promise<T>): Promise<T> {
|
||||
private serialize<T>(id: SessionId, op: () => Promise<T> | T): Promise<T> {
|
||||
const prior = this.chains.get(id) ?? Promise.resolve()
|
||||
const next = prior.then(op, op)
|
||||
// Keep the chain alive but swallow this op's rejection for the NEXT waiter
|
||||
@@ -331,15 +330,10 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
// reverse registration order, so event admission closes before this final
|
||||
// drain reaches quiescence and closes the backend.
|
||||
ctx.effect(() => async () => {
|
||||
await this.awaitRetirements()
|
||||
|
||||
let disposeError: unknown
|
||||
try {
|
||||
const errors = [
|
||||
...await settledErrors(this.inits.values()),
|
||||
...await settledErrors([...this.buffers.keys()].map(s => this.flush(s))),
|
||||
...await settledErrors(this.chains.values()),
|
||||
]
|
||||
const errors = await settledErrors([...this.live.keys()].map(session => this.flushForDispose(session)))
|
||||
while (this.chains.size > 0) await Promise.allSettled([...this.chains.values()])
|
||||
if (errors.length > 0) {
|
||||
throw new AggregateError(errors, `${this.backend.name} dispose failed`)
|
||||
}
|
||||
@@ -360,25 +354,20 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
}
|
||||
}, `${this.backend.name} write path`)
|
||||
|
||||
// Capture the header on creation; persist a fork's seed once. Record the init
|
||||
// promise so flush/dispose can await it (onCreated is async).
|
||||
// Capture the header on creation and persist a fork's seed once.
|
||||
ctx.on('session/created', (session) => { void this.initFor(session) })
|
||||
|
||||
// Session emits an owned frozen event. Keep a persistence-owned copy anyway
|
||||
// so the write-behind queue owns exactly the record it will flush rather than
|
||||
// retaining a product-layer record by identity. Serializability is guaranteed
|
||||
// at the source, so structuredClone is safe.
|
||||
// Keep a persistence-owned copy of each frozen event and start an eager drain.
|
||||
ctx.on('session/event', (session, event) => {
|
||||
let buffer = this.buffers.get(session)
|
||||
if (!buffer) this.buffers.set(session, buffer = [])
|
||||
buffer.push(structuredClone(event))
|
||||
const live = this.initFor(session)
|
||||
live.pending.push(structuredClone(event))
|
||||
if (live.flush === undefined) this.scheduleDrain(session, live)
|
||||
})
|
||||
|
||||
// Drain to the backend at the durability checkpoint.
|
||||
// Callers use flush as the observation barrier for the eager write path.
|
||||
ctx.on('session/flush', session => this.flush(session))
|
||||
|
||||
// Session disposal is observe-only, so the coordinator observes the
|
||||
// detached task itself and backend teardown awaits quiescence.
|
||||
// Session disposal is observe-only, so retirement contains its own failure.
|
||||
ctx.on('session/disposed', (session) => { this.retire(session) })
|
||||
|
||||
// HMR: a hot reload does not replay session/created, so seed existing live
|
||||
@@ -386,52 +375,33 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
for (const session of ctx.sessions.list()) void this.initFor(session)
|
||||
}
|
||||
|
||||
/** Start, observe, and track one disposed session's final drain. */
|
||||
/** Start and observe one disposed session's final drain. */
|
||||
private retire(session: Session): void {
|
||||
const task = this.retireCore(session)
|
||||
this.retirements.add(task)
|
||||
const settled = (): void => { this.retirements.delete(task) }
|
||||
void task.then(settled, (error: unknown) => {
|
||||
settled()
|
||||
void this.retireCore(session).catch((error: unknown) => {
|
||||
this.ctx.logger.warn(`${this.backend.name}: session "${session.id}" retirement failed: ${String(error)}`)
|
||||
})
|
||||
}
|
||||
|
||||
/** Drain and release state owned by one exact disposed Session lifecycle. */
|
||||
private async retireCore(session: Session): Promise<void> {
|
||||
await this.inits.get(session)
|
||||
|
||||
await this.flush(session)
|
||||
const id = session.header.id
|
||||
await this.serialize(id, async () => {
|
||||
await this.drain(session)
|
||||
this.buffers.delete(session)
|
||||
this.inits.delete(session)
|
||||
await this.serialize(id, () => {
|
||||
this.live.delete(session)
|
||||
if (this.states.get(id)?.owner === session) this.states.delete(id)
|
||||
})
|
||||
}
|
||||
|
||||
/** Await every retirement admitted before listener teardown. */
|
||||
private async awaitRetirements(): Promise<void> {
|
||||
while (this.retirements.size > 0) {
|
||||
await Promise.allSettled([...this.retirements])
|
||||
}
|
||||
}
|
||||
|
||||
/** Start (once) the async init for a session and remember its promise. */
|
||||
private initFor(session: Session): Promise<void> {
|
||||
const existing = this.inits.get(session)
|
||||
/** Return the one lifecycle controller for a live session, creating it if needed. */
|
||||
private initFor(session: Session): LiveSessionState {
|
||||
const existing = this.live.get(session)
|
||||
if (existing) return existing
|
||||
// Snapshot the seed SYNCHRONOUSLY — initFor runs inside the `session/created`
|
||||
// emit, before any later append invalidates the public array snapshot. Events
|
||||
// are already frozen; cloning gives persistence independent ownership.
|
||||
const seed = session.events.map(e => structuredClone(e))
|
||||
const p = this.onCreated(session, seed)
|
||||
// Attach a no-op rejection handler so a failing init does not surface as an
|
||||
// unhandled rejection if no flush observes `p` before it rejects. The REAL
|
||||
// error is still delivered: flush/dispose await the same `p` from the map.
|
||||
p.catch(() => { /* observed by flush/dispose via the stored promise */ })
|
||||
this.inits.set(session, p)
|
||||
return p
|
||||
const live: LiveSessionState = { pending: [], init: Promise.resolve(), flush: undefined }
|
||||
this.live.set(session, live)
|
||||
live.init = this.serialize(session.header.id, () => this.onCreated(session, seed))
|
||||
live.init.catch(() => { /* observed by flush/dispose through the controller */ })
|
||||
return live
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -452,7 +422,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
*
|
||||
* Cases, by whether this backend tracks the id and whether an artifact exists:
|
||||
* 1. Already tracked → no-op (or claim ownerless state if the seed matches,
|
||||
* or reclaim a truly-abandoned id, else reject as a collision).
|
||||
* else reject as a collision).
|
||||
* 2. Not tracked, an artifact EXISTS at this cwd and is a seq-aligned PREFIX
|
||||
* of the live events → ADOPT it (HMR/reload), persisting any live suffix.
|
||||
* 3. Not tracked, an artifact EXISTS but is NOT a prefix → REJECT (collision).
|
||||
@@ -487,17 +457,10 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
// Persist the seed SUFFIX beyond the persisted prefix. Constructor seed
|
||||
// events never emit session/event, so the buffer never sees them.
|
||||
const suffix = seed.slice(tracked.cursor)
|
||||
if (suffix.length > 0) await this.append(id, suffix)
|
||||
if (suffix.length > 0) await this.appendCore(id, suffix)
|
||||
return
|
||||
}
|
||||
// Owned by a DIFFERENT live session. Reclaim ONLY a truly-abandoned id
|
||||
// (never materialized, no pending buffer); else it is a real collision.
|
||||
const ownerBuffer = this.buffers.get(tracked.owner)
|
||||
if (!tracked.materialized && !ownerBuffer?.length) {
|
||||
this.states.delete(id)
|
||||
} else {
|
||||
throw new Error(`session "${id}" is already bound to a different live session in this backend (id collision)`)
|
||||
}
|
||||
throw new Error(`session "${id}" is already bound to a different live session in this backend (id collision)`)
|
||||
}
|
||||
|
||||
// case 2/3: an artifact at THIS cwd is adopted as a live prefix (or rejected
|
||||
@@ -509,20 +472,20 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
// Do NOT route through loadCore(): that crash-repairs open turns as
|
||||
// interrupted, which is wrong for HMR while the live Session is still the
|
||||
// authority and may append the real step/turn end later.
|
||||
await this.serialize(id, () => this.adoptLivePrefix(session, seed, live))
|
||||
await this.adoptLivePrefix(session, seed, live)
|
||||
return
|
||||
}
|
||||
|
||||
// case 4: a genuinely new session. Register its meta (lazy), then persist its
|
||||
// seed (events present at creation time) once.
|
||||
const meta: SessionHeader = { ...session.header }
|
||||
await this.create(meta)
|
||||
await this.createCore(meta)
|
||||
// Bind this state to the live session so a later DIFFERENT session reusing
|
||||
// the id is detected as a collision (case 1) rather than silently no-opped.
|
||||
const created = this.states.get(id)
|
||||
/* v8 ignore next -- create() always sets the state for the id */
|
||||
if (created !== undefined) created.owner = session
|
||||
if (seed.length > 0) await this.append(id, seed)
|
||||
if (seed.length > 0) await this.appendCore(id, seed)
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -551,36 +514,48 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
}
|
||||
|
||||
private async flush(session: Session): Promise<void> {
|
||||
// Wait for the session's init (onCreated) so the state/cursor and any
|
||||
// fork-seed persistence are in place before draining. Awaiting the same
|
||||
// promise initFor stored also surfaces an init failure (e.g. a collision)
|
||||
// here, where the caller of session/flush observes it.
|
||||
await this.inits.get(session)
|
||||
// Serialize the WHOLE drain (read cursor → append → splice) on the per-session
|
||||
// chain so two concurrent flushes cannot both read the same cursor and
|
||||
// seq-mismatch on the second append.
|
||||
await this.serialize(session.header.id, () => this.drain(session))
|
||||
const live = this.initFor(session)
|
||||
await live.init
|
||||
while (live.flush !== undefined || live.pending.length > 0) {
|
||||
await this.ensureFlush(session, live)
|
||||
}
|
||||
}
|
||||
|
||||
/** Drain a session's write buffer to the backend. Caller serializes this per id. */
|
||||
private async drain(session: Session): Promise<void> {
|
||||
const buffer = this.buffers.get(session)
|
||||
if (!buffer?.length) return
|
||||
// Copy WITHOUT removing: the buffer is the only durable-pending copy of these
|
||||
// events. Drain it only AFTER the append commits; events pushed during the
|
||||
// await sit past batch.length and survive the prefix splice, so a
|
||||
// retry/dispose re-drains the rest.
|
||||
const batch = buffer.slice()
|
||||
const state = this.states.get(session.header.id)
|
||||
// Only append events at or beyond the write cursor (a resumed session's seed
|
||||
// is already stored). flush awaits the init above, which always sets state,
|
||||
// so the `?? 0` fallback is a defensive guard that never fires in practice.
|
||||
/** Let an eager attempt settle, then make one teardown-owned retry observable. */
|
||||
private async flushForDispose(session: Session): Promise<void> {
|
||||
const current = this.live.get(session)?.flush
|
||||
if (current !== undefined) await Promise.allSettled([current])
|
||||
await this.flush(session)
|
||||
}
|
||||
|
||||
/** Start an eager drain without exposing its failure to the synchronous append. */
|
||||
private scheduleDrain(session: Session, live: LiveSessionState): void {
|
||||
void this.ensureFlush(session, live).catch((error: unknown) => {
|
||||
this.ctx.logger.warn(`${this.backend.name}: eager drain for session "${session.id}" failed (buffered events retained): ${String(error)}`)
|
||||
})
|
||||
}
|
||||
|
||||
/** Return the current drain, or start one for the complete pending batch. */
|
||||
private ensureFlush(session: Session, live: LiveSessionState): Promise<void> {
|
||||
if (live.flush !== undefined) return live.flush
|
||||
const flush = live.init
|
||||
.then(() => this.serialize(session.header.id, () => this.drain(session.header.id, live)))
|
||||
.finally(() => { live.flush = undefined })
|
||||
live.flush = flush
|
||||
void flush.then(() => {
|
||||
if (live.pending.length > 0) this.scheduleDrain(session, live)
|
||||
}, () => {})
|
||||
return flush
|
||||
}
|
||||
|
||||
/** Drain one stable prefix; events admitted during the write remain pending. */
|
||||
private async drain(id: SessionId, live: LiveSessionState): Promise<void> {
|
||||
const batch = live.pending.slice()
|
||||
const state = this.states.get(id)
|
||||
/* v8 ignore next -- state is always set by the awaited init before flush */
|
||||
const cursor = state?.cursor ?? 0
|
||||
const fresh = batch.filter(e => e.seq >= cursor)
|
||||
// appendCore (NOT the serialized append) — drain already runs inside the
|
||||
// per-session chain, so re-entering via append() would deadlock.
|
||||
if (fresh.length > 0) await this.appendCore(session.header.id, fresh)
|
||||
buffer.splice(0, batch.length)
|
||||
await this.appendCore(id, fresh)
|
||||
live.pending.splice(0, batch.length)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -63,10 +63,9 @@ export abstract class SessionPersistence extends Service {
|
||||
abstract create(meta: SessionHeader): Promise<void>
|
||||
|
||||
/**
|
||||
* Durably persist a batch of events (called from the write-behind drain at
|
||||
* the `session/flush` checkpoint). Honors the append-only and contiguous-seq
|
||||
* contracts: the first event's `seq` MUST equal the stored next-seq (after
|
||||
* `load` has durably closed any interrupted turn). Rejects non-JSON-
|
||||
* Durably persist a batch of events. Honors the append-only and contiguous-
|
||||
* seq contracts: the first event's `seq` MUST equal the stored next-seq
|
||||
* (after `load` has durably closed any interrupted turn). Rejects non-JSON-
|
||||
* serializable `event.data` with an error naming the offending event type.
|
||||
* @param id - the session the batch belongs to.
|
||||
* @param events - the contiguous batch to persist, in seq order.
|
||||
|
||||
@@ -48,10 +48,8 @@ interface MemoryConfig { store?: MemoryStore }
|
||||
/** Test-only view of the coordinator containers whose retirement is the contract under test. */
|
||||
interface CoordinatorInternals {
|
||||
states: Map<unknown, unknown>
|
||||
buffers: Map<unknown, unknown>
|
||||
live: Map<unknown, { pending: unknown[]; flush: Promise<void> | undefined }>
|
||||
chains: Map<unknown, unknown>
|
||||
inits: Map<unknown, unknown>
|
||||
retirements: Set<Promise<void>>
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -204,6 +202,40 @@ runCoordinatorContract('memory', async (): Promise<CoordinatorFixture> => {
|
||||
}
|
||||
})
|
||||
|
||||
describe('PersistenceCoordinator eager writes', () => {
|
||||
it('starts a follow-up batch for events admitted during an in-flight write', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const backend = new ControlledBackend()
|
||||
const appendGate = Promise.withResolvers<boolean>()
|
||||
backend.beforeAppend = async (attempt) => {
|
||||
if (attempt === 1) await appendGate.promise
|
||||
}
|
||||
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
|
||||
new PersistenceCoordinator(inner, backend)
|
||||
}, { inject: ['sessions'] }))
|
||||
|
||||
try {
|
||||
const session = ctx.sessions.create(SessionId('eager-follow-up'))
|
||||
await ctx.sessions.flush(session)
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
await vi.waitFor(() => { expect(backend.appendAttempts).toBe(1) })
|
||||
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
appendGate.resolve(true)
|
||||
|
||||
await vi.waitFor(() => {
|
||||
expect(backend.appendAttempts).toBe(2)
|
||||
expect(backend.store.get(session.id)?.events.map(event => event.seq)).toEqual([0, 1])
|
||||
})
|
||||
} finally {
|
||||
appendGate.resolve(true)
|
||||
await fiber.dispose()
|
||||
await ctx.fiber.dispose()
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
describe('PersistenceCoordinator retirement', () => {
|
||||
it('a retiring unmaterialized owner without buffered events releases its id', async () => {
|
||||
const ctx = new Context()
|
||||
@@ -233,7 +265,6 @@ describe('PersistenceCoordinator retirement', () => {
|
||||
await ctx.plugin(Object.assign((inner: Context) => {
|
||||
reuse = inner.sessions.create(id)
|
||||
}, { inject: ['sessions'] }))
|
||||
await vi.waitFor(() => { expect(backend.loadAttempts).toBe(baselineLoads + 2) })
|
||||
|
||||
loadGate.resolve(true)
|
||||
await expect(blockingLoad).rejects.toThrow(/not found/)
|
||||
@@ -275,10 +306,11 @@ describe('PersistenceCoordinator retirement', () => {
|
||||
await ctx.plugin(Object.assign((inner: Context) => {
|
||||
reuse = inner.sessions.create(id)
|
||||
}, { inject: ['sessions'] }))
|
||||
await expect(ctx.sessions.flush(reuse)).rejects.toThrow(/bound to a different live session/)
|
||||
const reuseFlush = ctx.sessions.flush(reuse)
|
||||
|
||||
loadGate.resolve(true)
|
||||
await expect(blockingLoad).rejects.toThrow(/not found/)
|
||||
await expect(reuseFlush).rejects.toThrow(/id collision/)
|
||||
await vi.waitFor(() => {
|
||||
expect(backend.store.get(id)?.events.map(event => event.seq)).toEqual([0, 1])
|
||||
})
|
||||
@@ -346,8 +378,9 @@ describe('PersistenceCoordinator retirement', () => {
|
||||
coordinator = new PersistenceCoordinator(inner, backend)
|
||||
}, { inject: ['sessions'] }))
|
||||
const internals = coordinator as unknown as CoordinatorInternals
|
||||
backend.beforeAppend = async (attempt) => {
|
||||
if (attempt === 1) {
|
||||
let retryEnabled = false
|
||||
backend.beforeAppend = async () => {
|
||||
if (!retryEnabled) {
|
||||
backend.lifecycle.push('append-failed')
|
||||
throw new Error('transient append failure')
|
||||
}
|
||||
@@ -364,17 +397,18 @@ describe('PersistenceCoordinator retirement', () => {
|
||||
await sessionFiber.dispose()
|
||||
|
||||
await vi.waitFor(() => {
|
||||
expect(backend.appendAttempts).toBe(1)
|
||||
expect(internals.retirements.size).toBe(0)
|
||||
expect(backend.appendAttempts).toBeGreaterThanOrEqual(1)
|
||||
expect([...internals.live.values()][0]?.pending).toEqual(expect.arrayContaining([
|
||||
expect.objectContaining({ seq: 0 }),
|
||||
expect.objectContaining({ seq: 1 }),
|
||||
]))
|
||||
})
|
||||
expect([...internals.buffers.values()]).toEqual([expect.arrayContaining([
|
||||
expect.objectContaining({ seq: 0 }),
|
||||
expect.objectContaining({ seq: 1 }),
|
||||
])])
|
||||
|
||||
retryEnabled = true
|
||||
await backendFiber.dispose()
|
||||
expect(backend.store.get(SessionId('retry-retirement'))?.events.map(event => event.seq)).toEqual([0, 1])
|
||||
expect(backend.lifecycle).toEqual(['append-failed', 'append-committed', 'close'])
|
||||
expect(backend.lifecycle.at(-2)).toBe('append-committed')
|
||||
expect(backend.lifecycle.at(-1)).toBe('close')
|
||||
} finally {
|
||||
await backendFiber.dispose()
|
||||
await ctx.fiber.dispose()
|
||||
@@ -407,7 +441,8 @@ describe('PersistenceCoordinator retirement', () => {
|
||||
await sessionFiber.dispose()
|
||||
await vi.waitFor(() => {
|
||||
expect(backend.appendAttempts).toBe(1)
|
||||
expect(internals.retirements.size).toBe(1)
|
||||
expect(internals.live.size).toBe(1)
|
||||
expect([...internals.live.values()][0]?.flush).toBeInstanceOf(Promise)
|
||||
})
|
||||
|
||||
let disposed = false
|
||||
@@ -426,6 +461,47 @@ describe('PersistenceCoordinator retirement', () => {
|
||||
await ctx.fiber.dispose()
|
||||
}
|
||||
})
|
||||
|
||||
it('backend teardown waits for a detached public append before close', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const backend = new ControlledBackend()
|
||||
let coordinator!: PersistenceCoordinator<never>
|
||||
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
|
||||
coordinator = new PersistenceCoordinator(inner, backend)
|
||||
}, { inject: ['sessions'] }))
|
||||
const appendGate = Promise.withResolvers<boolean>()
|
||||
backend.beforeAppend = async () => {
|
||||
backend.lifecycle.push('append-started')
|
||||
await appendGate.promise
|
||||
backend.lifecycle.push('append-committed')
|
||||
}
|
||||
|
||||
try {
|
||||
const id = SessionId('inflight-public-append')
|
||||
await coordinator.create(meta(id))
|
||||
const append = coordinator.append(id, [{
|
||||
type: 'turn/start',
|
||||
seq: 0,
|
||||
time: 1,
|
||||
data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } },
|
||||
}])
|
||||
await vi.waitFor(() => { expect(backend.appendAttempts).toBe(1) })
|
||||
|
||||
let disposed = false
|
||||
const teardown = fiber.dispose().then(() => { disposed = true })
|
||||
await Promise.resolve()
|
||||
expect(disposed).toBe(false)
|
||||
|
||||
appendGate.resolve(true)
|
||||
await Promise.all([append, teardown])
|
||||
expect(backend.lifecycle).toEqual(['append-started', 'append-committed', 'close'])
|
||||
} finally {
|
||||
appendGate.resolve(true)
|
||||
await fiber.dispose()
|
||||
await ctx.fiber.dispose()
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
describe('SessionPersistence service registration', () => {
|
||||
@@ -555,11 +631,9 @@ describe('SessionPersistence service registration', () => {
|
||||
expect(ctx.sessions.list()).toHaveLength(0)
|
||||
expect({
|
||||
states: coordinator.states.size,
|
||||
buffers: coordinator.buffers.size,
|
||||
live: coordinator.live.size,
|
||||
chains: coordinator.chains.size,
|
||||
inits: coordinator.inits.size,
|
||||
retirements: coordinator.retirements.size,
|
||||
}).toEqual({ states: 0, buffers: 0, chains: 0, inits: 0, retirements: 0 })
|
||||
}).toEqual({ states: 0, live: 0, chains: 0 })
|
||||
})
|
||||
} finally {
|
||||
await fiber.dispose()
|
||||
|
||||
Reference in New Issue
Block a user