diff --git a/packages/session-projection/session-projection/src/index.ts b/packages/session-projection/session-projection/src/index.ts index e43a03d38b..c2974b565d 100644 --- a/packages/session-projection/session-projection/src/index.ts +++ b/packages/session-projection/session-projection/src/index.ts @@ -97,6 +97,27 @@ export interface ProjectionSnapshot { values: Partial } +/** + * One unit's checkpoint: its internal state (plain JSON by the unit + * contract), the seq of the last event folded into it, and the + * `stateVersion` that produced it — the persisted projection-cache row + * `(sessionId, key, stateVersion, observedSeq, state)` minus the two outer + * keys. A row is never authoritative, only a fold shortcut: `restore` + * discards it on a `stateVersion` mismatch or when it claims events past the + * stored log end. + */ +export interface ProjectionCheckpointRow { + /** The registering unit's `stateVersion` at fold time. */ + stateVersion: number + /** Seq of the last event folded into `state`; -1 for the empty log. */ + observedSeq: number + /** The unit's internal state — plain JSON per the unit contract. */ + state: unknown +} + +/** Checkpoint rows keyed by projection key (one session's persisted cache value). */ +export type ProjectionCheckpoint = Record + /** Type-erased unit view the drive machinery works with (the register seam already proved the typed contract). */ interface ErasedDefinition { key: string @@ -206,6 +227,113 @@ export class SessionProjectionRegistry extends Service { return { asOfSeq: session.seq - 1, values: values } } + /** + * State-level checkpoint of every registered unit for one session, read + * from the watermark cache (missing cells fold lazily over the in-memory + * log). This is the write side of the persisted projection cache: the + * returned rows are the `(key → {stateVersion, observedSeq, state})` part + * of the durable `(sessionId, key, stateVersion, observedSeq, state)` + * rows. States are the units' live references — plain JSON by the unit + * contract, treated as immutable; a durable writer snapshots them at its + * own boundary. + * @param session - the session whose unit states are checkpointed. + * @returns one row per registered key; empty when no unit is registered. + */ + checkpoint(session: Session): ProjectionCheckpoint { + const rows: ProjectionCheckpoint = {} + for (const registration of this.registrations.values()) { + const cell = this.cellFor(registration, session) + rows[registration.def.key] = { + stateVersion: registration.def.stateVersion, + observedSeq: cell.observedSeq, + state: cell.state, + } + } + return rows + } + + /** + * The stored seq a {@link restore} tail read over `checkpoint` must start + * at: one event BELOW the lowest usable watermark (a row is usable when + * its `stateVersion` matches the live unit; an absent or mismatched row + * pulls the floor to `0` — that key must refold the full log). The + * one-below anchor is load-bearing: the tail then proves how far the + * stored log still extends, so {@link restore} can detect a log that + * shrank below a row's watermark (crash-repair truncation) instead of + * serving the stale row as current — an empty tail read from the anchor + * yields an end below every watermark and the restore rejects for a full + * re-read. + * @param checkpoint - persisted rows for one session (possibly stale or empty). + * @returns the seq to hand the persistence `readFrom`, or `undefined` + * when no unit is registered (no read needed — {@link restore} would + * serve empty values regardless). + */ + restoreFloor(checkpoint: ProjectionCheckpoint): number | undefined { + let floor: number | undefined + for (const registration of this.registrations.values()) { + const row = checkpoint[registration.def.key] + const need = row !== undefined && row.stateVersion === registration.def.stateVersion + ? Math.max(row.observedSeq + 1, 0) + : 0 + floor = floor === undefined ? need : Math.min(floor, need) + } + return floor === undefined ? undefined : Math.max(floor - 1, 0) + } + + /** + * Cold read: fold every registered unit over a stored log suffix, seeding + * each from its checkpoint row when usable — the one read recipe (cached + * state + forward tail replay + `view`) applied without a live `Session`. + * Call with the events returned by a persistence + * `readFrom(id, restoreFloor(checkpoint))` and that same floor as + * `baseSeq`; the floor's one-below anchor makes the supplied end honest, + * so a shrunk log is detected here. A row is usable iff its + * `stateVersion` matches the live unit, it does not predate `baseSeq` + * (`observedSeq >= baseSeq - 1`), and it does not claim events past the + * supplied end (`observedSeq <= endSeq`); an unusable row is discarded + * and its key refolds from `init` — which is only sound over the full + * log, so a discarded row with `baseSeq > 0` throws (the caller re-reads + * from seq 0, e.g. after a crash-repair truncation shrank the log below + * a row's watermark). + * @param checkpoint - persisted rows for one session (possibly stale or empty). + * @param events - the stored events with `seq >= baseSeq`, in seq order. + * @param baseSeq - the seq `events` starts at (its first event's seq when non-empty). + * @returns the snapshot cut at the supplied log end (`asOfSeq` is the last + * supplied event's seq, `baseSeq - 1` for an empty tail) plus the + * refreshed checkpoint rows at that cut, ready for a durable write-back. + */ + restore(checkpoint: ProjectionCheckpoint, events: readonly SessionEvent[], baseSeq: number): + { snapshot: ProjectionSnapshot; checkpoint: ProjectionCheckpoint } { + const endSeq = events.at(-1)?.seq ?? baseSeq - 1 + const values: Record = {} + const refreshed: ProjectionCheckpoint = {} + for (const registration of this.registrations.values()) { + const def = registration.def + const row = checkpoint[def.key] + const usable = row !== undefined + && row.stateVersion === def.stateVersion + && row.observedSeq >= baseSeq - 1 + && row.observedSeq <= endSeq + if (!usable && baseSeq > 0) { + throw new Error( + `session projection ${JSON.stringify(def.key)} cannot restore from seq ${baseSeq}: ` + + 'its checkpoint row is missing, version-mismatched, or beyond the supplied log end; re-read from seq 0', + ) + } + let state = usable ? row.state : def.init() + const from = usable ? row.observedSeq : baseSeq - 1 + for (const event of events) { + if (event.seq > from) state = def.apply(state, event) + } + values[def.key] = def.schema.parse(def.view(state)) + refreshed[def.key] = { stateVersion: def.stateVersion, observedSeq: endSeq, state } + } + return { + snapshot: { asOfSeq: endSeq, values: values as ProjectionSnapshot['values'] }, + checkpoint: refreshed, + } + } + /** Fold one unit from init over `events`, producing a cell watermarked at the last folded event. */ private buildCell(def: ErasedDefinition, events: readonly SessionEvent[]): UnitCell { let state = def.init() diff --git a/packages/session-projection/session-projection/tests/registry.spec.ts b/packages/session-projection/session-projection/tests/registry.spec.ts index 06d7947caf..3c069f91d3 100644 --- a/packages/session-projection/session-projection/tests/registry.spec.ts +++ b/packages/session-projection/session-projection/tests/registry.spec.ts @@ -169,6 +169,128 @@ describe('SessionProjectionRegistry drive', () => { expect(ctx.sessionProjections.snapshot(session).values).toEqual({}) }) + it('checkpoints every registered unit with its stateVersion and per-cell watermark', async () => { + const { ctx, session } = await harness() + ctx.sessionProjections.register(marksUnit()) + ctx.sessionProjections.register({ ...countUnit(), stateVersion: 7 }) + const markEvent = mark(session, ['a']) + const rows = ctx.sessionProjections.checkpoint(session) + expect(rows['test/marks']).toEqual({ stateVersion: 1, observedSeq: markEvent.seq, state: { marks: ['a'] } }) + expect(rows['test/count']).toEqual({ stateVersion: 7, observedSeq: markEvent.seq, state: 1 }) + // Empty log: init-derived state at watermark -1. + const fresh = ctx.sessions.create() + expect(ctx.sessionProjections.checkpoint(fresh)['test/marks']).toEqual({ stateVersion: 1, observedSeq: -1, state: null }) + }) + + it('restoreFloor anchors one below the lowest usable watermark and at 0 for missing or mismatched rows', async () => { + const { ctx } = await harness() + expect(ctx.sessionProjections.restoreFloor({})).toBeUndefined() // no unit registered + ctx.sessionProjections.register(marksUnit()) + ctx.sessionProjections.register(countUnit()) + expect(ctx.sessionProjections.restoreFloor({})).toBe(0) + // Lowest usable watermark is count's 5 → the anchored tail starts AT 5 + // (one below the first needed seq 6), so the read proves seq 5 still exists. + expect(ctx.sessionProjections.restoreFloor({ + 'test/marks': { stateVersion: 1, observedSeq: 10, state: { marks: [] } }, + 'test/count': { stateVersion: 1, observedSeq: 5, state: 6 }, + })).toBe(5) + // A version-mismatched row forces that key back to a full refold. + expect(ctx.sessionProjections.restoreFloor({ + 'test/marks': { stateVersion: 2, observedSeq: 10, state: { marks: [] } }, + 'test/count': { stateVersion: 1, observedSeq: 5, state: 6 }, + })).toBe(0) + // A fresh (-1) row still needs the whole tail from 0. + expect(ctx.sessionProjections.restoreFloor({ + 'test/marks': { stateVersion: 1, observedSeq: -1, state: null }, + 'test/count': { stateVersion: 1, observedSeq: -1, state: 0 }, + })).toBe(0) + }) + + it('restore folds the tail past each usable row and refolds from init on version mismatch', async () => { + const { ctx } = await harness() + ctx.sessionProjections.register(marksUnit()) + ctx.sessionProjections.register(countUnit()) + const tail: SessionEvent[] = [ + { type: 'test/mark', seq: 3, time: 3, data: { marks: ['new'] } } as SessionEvent, + { type: 'turn/end', seq: 4, time: 4, data: { turn: 1, reason: { kind: 'completed' } } }, + ] + // marks row usable (watermark 2, tail starts at 3); count row mismatched — but + // a mismatch with baseSeq > 0 cannot silently refold: it throws for a re-read. + expect(() => ctx.sessionProjections.restore({ + 'test/marks': { stateVersion: 1, observedSeq: 2, state: { marks: ['old'] } }, + 'test/count': { stateVersion: 99, observedSeq: 2, state: 3 }, + }, tail, 3)).toThrow(/re-read from seq 0/) + // The full-log re-read (baseSeq 0) refolds the mismatched key from init. + const full: SessionEvent[] = [ + { type: 'turn/start', seq: 0, time: 0, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }, + { type: 'test/mark', seq: 1, time: 1, data: { marks: ['old'] } } as SessionEvent, + { type: 'test/mark', seq: 2, time: 2, data: { marks: ['old', '2'] } } as SessionEvent, + ...tail, + ] + const { snapshot, checkpoint } = ctx.sessionProjections.restore({ + 'test/marks': { stateVersion: 1, observedSeq: 2, state: { marks: ['old', '2'] } }, + 'test/count': { stateVersion: 99, observedSeq: 2, state: 3 }, + }, full, 0) + expect(snapshot.asOfSeq).toBe(4) + expect(snapshot.values['test/marks']).toEqual({ marks: ['new'] }) + expect(snapshot.values['test/count']).toBe(5) // refolded from init over all 5 events + // The refreshed rows sit at the served cut, ready for a durable write-back. + expect(checkpoint['test/marks']).toEqual({ stateVersion: 1, observedSeq: 4, state: { marks: ['new'] } }) + expect(checkpoint['test/count']).toEqual({ stateVersion: 1, observedSeq: 4, state: 5 }) + }) + + it('restore over a suffix folds only past each row watermark and serves an exact empty-tail cut', async () => { + const { ctx } = await harness() + ctx.sessionProjections.register(marksUnit()) + ctx.sessionProjections.register(countUnit()) + const rows = { + 'test/marks': { stateVersion: 1, observedSeq: 4, state: { marks: ['done'] } }, + 'test/count': { stateVersion: 1, observedSeq: 2, state: 3 }, + } + const tail: SessionEvent[] = [ + { type: 'turn/start', seq: 3, time: 3, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } }, + { type: 'turn/end', seq: 4, time: 4, data: { turn: 2, reason: { kind: 'completed' } } }, + ] + const { snapshot } = ctx.sessionProjections.restore(rows, tail, 3) + expect(snapshot.asOfSeq).toBe(4) + // marks already covers the tail (watermark 4): nothing re-applied. + expect(snapshot.values['test/marks']).toEqual({ marks: ['done'] }) + // count folds exactly seqs 3 and 4 on top of its checkpoint. + expect(snapshot.values['test/count']).toBe(5) + + // Empty tail (checkpoint is current): the cut sits at baseSeq - 1. + const { snapshot: current } = ctx.sessionProjections.restore({ + 'test/marks': { stateVersion: 1, observedSeq: 4, state: { marks: ['done'] } }, + 'test/count': { stateVersion: 1, observedSeq: 4, state: 5 }, + }, [], 5) + expect(current.asOfSeq).toBe(4) + expect(current.values['test/count']).toBe(5) + }) + + it('restore rejects a row claiming events past the supplied log end (shrunk log ⇒ re-read)', async () => { + const { ctx } = await harness() + ctx.sessionProjections.register(countUnit()) + const rows = { 'test/count': { stateVersion: 1, observedSeq: 9, state: 10 } } + // The anchored floor sits ON the watermark, so the tail read must return + // at least seq 9 from an intact log… + const floor = ctx.sessionProjections.restoreFloor(rows) + expect(floor).toBe(9) + // …an intact log serves the anchor event and the checkpoint stands as-is. + const anchor: SessionEvent = { type: 'turn/end', seq: 9, time: 9, data: { turn: 2, reason: { kind: 'completed' } } } + expect(ctx.sessionProjections.restore(rows, [anchor], 9).snapshot.values['test/count']).toBe(10) + // …while a log crash-repaired down to fewer events returns an empty tail: + // the row overreaches the proven end and a tail read cannot fix this key. + expect(() => ctx.sessionProjections.restore(rows, [], 9)).toThrow(/re-read from seq 0/) + // The full re-read discards the overreaching row and refolds from init. + const events: SessionEvent[] = [ + { type: 'turn/start', seq: 0, time: 0, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }, + { type: 'turn/end', seq: 1, time: 1, data: { turn: 1, reason: { kind: 'completed' } } }, + ] + const { snapshot } = ctx.sessionProjections.restore(rows, events, 0) + expect(snapshot.asOfSeq).toBe(1) + expect(snapshot.values['test/count']).toBe(2) + }) + it('fails loud when a unit view violates its own schema (async unit output is unrepresentable)', async () => { const { ctx, session } = await harness() ctx.sessionProjections.register({