diff --git a/src/host/backfill.ts b/src/host/backfill.ts index 2ab0a74..cad6eb3 100644 --- a/src/host/backfill.ts +++ b/src/host/backfill.ts @@ -40,6 +40,18 @@ interface ProjectionCacheLike { coldSnapshot(header: SessionHeader, inheritedEventCount: unknown, events: readonly SessionEvent[]): unknown } +/** The sliced fold's extra faces (all re-proved at runtime; any absence keeps the + * single-shot path). `restore` and `hydrate` ride the registry the plugin already injects; + * `prepare` and `write` are the same faces session-query's observation handoff uses. */ +interface SlicedFoldFaces { + restore( + checkpoint: unknown, events: readonly SessionEvent[], baseSeq: number, header: SessionHeader, inheritedEventCount: unknown, + ): unknown + hydrate(session: unknown, checkpoint: unknown, events: readonly SessionEvent[], baseSeq: number): unknown + prepare(id: string, options: { seed: readonly SessionEvent[]; meta: unknown; inheritedEventCount: unknown; eventState: string }): unknown + write(session: unknown): Promise +} + /** One persistence read handle, as consumed. */ interface ReadHandleLike { header: SessionHeader @@ -56,6 +68,12 @@ interface PersistenceLike { * sustained read pressure stalls its own client handshake. */ const YIELD_MS = 100 +/** The sliced fold's granularity and inter-slice pause: at ~13µs/event a slice is a ~25ms + * synchronous block (measured ≤88ms including the per-slice state re-validation), so even a + * 30k-event session never lands on the event loop as one uninterruptible second. */ +const SLICE_EVENTS = 2_000 +const SLICE_YIELD_MS = 25 + /** The `name` family dsh raises for a durable log it cannot interpret: the persistence seam * translates the format edge's refusal into `SessionFormatUnsupportedError`, so classify by prefix and import no dsh symbol. */ const UNSUPPORTED_FORMAT = 'SessionFormatUnsupported' @@ -123,6 +141,60 @@ function isLive(sessions: Record | null, id: string): boolean { } } +/** The sliced fold's per-session outcome: the rows landed, the session went live mid-fold + * (it folds for itself — skip silently), or the pass was aborted mid-fold (stop quietly). */ +type SliceOutcome = 'done' | 'live' | 'aborted' + +/** One cold fold in event-count slices (issue #145): a monster session's whole-log restore + * would otherwise land as ONE uninterruptible synchronous block (measured ~1s per 30k events + * on a fast host) inside a boot-time window already crowded with a hundred other folds. + * Each slice seeds every unit from the previous slice's rows IN MEMORY (the registry's + * restore is the cache's own cold-read computation, so the composed result is byte-identical + * to one single-shot coldSnapshot), the event loop breathes between slices, and exactly ONE + * durable write-back lands at the end — a crash mid-fold leaves no partial rows, the same + * failure semantics as the single-shot path. */ +async function coldSnapshotSliced( + ctx: Context, + faces: SlicedFoldFaces, + sessions: Record | null, + log: { header: SessionHeader; inheritedEventCount: unknown; events: readonly SessionEvent[] }, + signal: AbortSignal, +): Promise { + const { header, inheritedEventCount, events } = log + let rows: unknown = {} + for (let end = SLICE_EVENTS; ; end += SLICE_EVENTS) { + const last = end >= events.length + const restored = asRecord(faces.restore(rows, last ? events : events.slice(0, end), 0, header, inheritedEventCount)) + rows = restored?.checkpoint ?? {} + if (last) break + await new Promise(resolve => setTimeout(resolve, SLICE_YIELD_MS)) + if (signal.aborted) return 'aborted' + } + let prepared: unknown + try { + prepared = faces.prepare(header.id, { + seed: events, + meta: structuredClone(header), + inheritedEventCount, + eventState: 'shared-frozen', + }) + } catch (error: unknown) { + // The store rejects an id with a live owner: the session attached mid-fold and now folds for itself. + if (isLive(sessions, header.id)) return 'live' + throw error + } + // The rows end at the log's last seq, so hydrate installs the cells without refolding. + faces.hydrate(prepared, rows, events, 0) + try { + await faces.write(prepared) + } catch (error: unknown) { + // Fail-soft, mirroring the cache's own write-back posture: a lost write costs a refold + // on the next pass, never a wrong row. + ctx.logger.warn(`dsh-context: projection backfill write-back for "${header.id}" failed (cache stays stale): ${messageOf(error)}`) + } + return 'done' +} + function isUnsupportedFormat(error: unknown): boolean { try { return String(asRecord(error)?.name).startsWith(UNSUPPORTED_FORMAT) @@ -179,6 +251,24 @@ export function watchActivityBackfill(ctx: Context, coldReads: ColdReadGate = ma if (persistence === null || typeof persistence.open !== 'function') return const sessions = asRecord(injected.sessions) + // The sliced fold's faces: the registry the plugin already injects, plus the sessions and + // cache write verbs. Any absence keeps every session on the single-shot coldSnapshot path. + // Bind at extraction: an unbound hand-off loses `this` on the real faces. + const registry = asRecord((ctx as unknown as Record).sessionProjections) + const sliceFaces: SlicedFoldFaces | null = registry !== null + && typeof registry.restore === 'function' + && typeof registry.hydrate === 'function' + && sessions !== null + && typeof sessions.prepare === 'function' + && typeof cache.write === 'function' + ? { + restore: (registry.restore as SlicedFoldFaces['restore']).bind(registry), + hydrate: (registry.hydrate as SlicedFoldFaces['hydrate']).bind(registry), + prepare: (sessions.prepare as SlicedFoldFaces['prepare']).bind(sessions), + write: (cache.write as SlicedFoldFaces['write']).bind(cache), + } + : null + const abort = new AbortController() const run = async (): Promise => { const listed = await (query as unknown as SessionQueryLike).listSessions(abort.signal) @@ -196,15 +286,19 @@ export function watchActivityBackfill(ctx: Context, coldReads: ColdReadGate = ma const admitted = await coldReads.admit(async () => { const log = await readColdLog(persistence as unknown as PersistenceLike, header.id, abort.signal) // The handle's header is authoritative (fixed at open); the listed one was its probe. + if (sliceFaces !== null && log.events.length > SLICE_EVENTS) { + return coldSnapshotSliced(ctx, sliceFaces, sessions, log, abort.signal) + } ;(cache as unknown as ProjectionCacheLike).coldSnapshot(log.header, log.inheritedEventCount, log.events) - return true + return 'done' as const }) if (admitted === undefined) { // Pressure does not fall within a pacing tick, so re-probing per session only delays the host's own work. paused = true break } - folded++ + if (admitted === 'aborted') return + if (admitted === 'done') folded++ } catch (error: unknown) { if (isUnsupportedFormat(error)) { // Permanent and source-side: the per-session detail drops to diff --git a/tests/host/backfill.spec.ts b/tests/host/backfill.spec.ts index 7d42c28..7ea5d58 100644 --- a/tests/host/backfill.spec.ts +++ b/tests/host/backfill.spec.ts @@ -3,8 +3,12 @@ import assert from 'node:assert/strict' import { describe, test } from 'vitest' import { Context } from '@deepseek-ai/cordis' +import { SessionProjectionRegistry } from '@deepseek-ai/dsh-session-projection' import type { SessionEvent } from '@deepseek-ai/dsh-session' import { BACKFILL_ROUTE, watchActivityBackfill } from '../../src/host/backfill' +import { createContextTimelineDefinition } from '../../src/host/timeline' +import { createContextActivityDefinition } from '../../src/host/activity' +import { createContextHeadersDefinition } from '../../src/host/headers' async function until(read: () => T | undefined, message: string): Promise { for (let i = 0; i < 300; i++) { @@ -761,3 +765,212 @@ describe('watchActivityBackfill', () => { assert.equal(lines.filter(l => l.text.includes('heap pressure')).length, 1, 'one line, not one per session') }) }) + +/** A contiguous fake log of the given length (restore validates seq continuity, nothing else). */ +function fakeEvents(length: number): SessionEvent[] { + return Array.from({ length }, (_, seq) => ({ type: 'todo/write', seq, time: seq + 1, data: {} })) as unknown as SessionEvent[] +} + +interface SliceRig { + route: RouteBox + writes: Map + prepares: string[] + coldSnapshots: string[] + restoreCalls: number[] +} + +/** One single-shot restore over the same log on a fresh registry — the sliced path's reference answer. */ +function referenceRows(events: SessionEvent[], id: string): unknown { + const ctx = new Context() + const registry = new SessionProjectionRegistry(ctx as never) + registry.register(createContextTimelineDefinition({}, () => true) as never) + registry.register(createContextActivityDefinition() as never) + registry.register(createContextHeadersDefinition() as never) + return registry.restore({}, events, 0 as never, { id, version: 1, createdAt: 1, cwd: `/repo/${id}`, isSeeded: false } as never, 0 as never).checkpoint +} + +/** Arm the backfill with the REAL registry + the plugin's REAL units, fake persistence serving + * `events`, and the sliced path's faces (prepare/write) wired to the registry the way dsh's + * SessionStore and SessionProjectionCache wire them. */ +function armSliced(ctx: Context, eventsById: Map, opts: { + liveOnPrepare?: Set + failPrepare?: Set + failWrite?: boolean + noPrepare?: boolean + hostileRestoreAt?: number +} = {}): SliceRig { + const route: RouteBox = {} + const state = fakeState([...eventsById.keys()].map(id => ({ header: { id, cwd: `/repo/${id}` } }))) + const rig: SliceRig = { + route, + writes: new Map(), + prepares: [], + coldSnapshots: state.coldSnapshots, + restoreCalls: [], + } + const registry = new SessionProjectionRegistry(ctx as never) + registry.register(createContextTimelineDefinition({}, () => true) as never) + registry.register(createContextActivityDefinition() as never) + registry.register(createContextHeadersDefinition() as never) + const realRestore = registry.restore.bind(registry) + registry.restore = ((...args: Parameters) => { + rig.restoreCalls.push((args[1] as SessionEvent[]).length) + if (opts.hostileRestoreAt !== undefined && rig.restoreCalls.length === opts.hostileRestoreAt) return undefined + return realRestore(...args) + }) as typeof registry.restore + ctx.provide('connection', connectionOf(route)) + ctx.provide('sessionQuery', { listSessions: async () => state.listed }) + ctx.provide('sessionProjectionCache', { + cachedSnapshot: () => undefined, + coldSnapshot: (header: { id: string }) => { + state.coldSnapshots.push(header.id) + return { asOfSeq: 0, values: {} } + }, + write: async (session: { id: string }) => { + if (opts.failWrite === true) throw new Error('disk full') + rig.writes.set(session.id, registry.checkpoint(session as never)) + }, + }) + ctx.provide('sessionPersistence', { + open: async (id: string) => { + state.coldReads.push(id) + return { + header: { id, version: 1, createdAt: 1, cwd: `/repo/${id}`, isSeeded: false }, + inheritedEventCount: 0, + read: async () => ({ events: eventsById.get(id) ?? [] }), + close: async () => {}, + } + }, + }) + const threwPrepare = new Set() + const sessionsFace: Record = { + get: (id: string) => (state.live.has(id) || threwPrepare.has(id) ? { id } : undefined), + } + if (opts.noPrepare !== true) { + sessionsFace.prepare = (id: string, o: { seed: SessionEvent[]; meta: unknown; inheritedEventCount: number }) => { + rig.prepares.push(id) + if (opts.liveOnPrepare?.has(id) === true) { + threwPrepare.add(id) + throw new Error(`session "${id}" already exists`) + } + if (opts.failPrepare?.has(id) === true) throw new Error(`session "${id}" has a corrupt header`) + return { + id, + header: o.meta, + inheritedEventCount: o.inheritedEventCount, + seq: o.seed.length, + snapshotEvents: () => o.seed, + } + } + } + ctx.provide('sessions', sessionsFace) + return rig +} + +describe('the sliced cold fold (issue #145)', () => { + test('a monster session folds in slices and writes back rows byte-identical to one single-shot restore', async () => { + const ctx = new Context() + const events = fakeEvents(5_001) // three slices: 2000 + 2000 + 1001 + const rig = armSliced(ctx, new Map([['monster', events]])) + const dispose = watchActivityBackfill(ctx) + await trigger(rig.route) + await until(() => (rig.writes.has('monster') ? true : undefined), 'the write-back landed') + dispose() + assert.deepEqual(rig.coldSnapshots, [], 'the sliced path never calls coldSnapshot') + assert.deepEqual(rig.prepares, ['monster']) + assert.ok(rig.restoreCalls.length >= 3, `expected ≥3 sliced restores, got ${rig.restoreCalls.join(',')}`) + assert.deepStrictEqual(rig.writes.get('monster'), referenceRows(events, 'monster'), 'sliced rows must equal the single-shot rows byte for byte') + }) + + test('a hostile restore answer (no checkpoint) refolds from init and still lands the exact rows', async () => { + const ctx = new Context() + const events = fakeEvents(4_001) + const rig = armSliced(ctx, new Map([['monster', events]]), { hostileRestoreAt: 2 }) + const dispose = watchActivityBackfill(ctx) + await trigger(rig.route) + await until(() => (rig.writes.has('monster') ? true : undefined), 'the write-back landed') + dispose() + assert.deepStrictEqual(rig.writes.get('monster'), referenceRows(events, 'monster'), 'a dropped slice answer degrades to a from-init refold — same rows') + }) + + test('a small session stays on the single-shot coldSnapshot path even when the slice faces exist', async () => { + const ctx = new Context() + const rig = armSliced(ctx, new Map([['small', fakeEvents(2_000)]])) + const dispose = watchActivityBackfill(ctx) + await trigger(rig.route) + await until(() => (rig.coldSnapshots.length === 1 ? true : undefined), 'the coldSnapshot ran') + dispose() + assert.deepEqual(rig.coldSnapshots, ['small']) + assert.deepEqual(rig.prepares, [], 'no prepare for a session at or under the slice threshold') + assert.deepEqual([...rig.writes.keys()], []) + }) + + test('missing slice faces fall back to coldSnapshot for the same monster session', async () => { + const ctx = new Context() + const rig = armSliced(ctx, new Map([['monster', fakeEvents(4_001)]], ), { noPrepare: true }) + const dispose = watchActivityBackfill(ctx) + await trigger(rig.route) + await until(() => (rig.coldSnapshots.length === 1 ? true : undefined), 'the fallback ran') + dispose() + assert.deepEqual(rig.coldSnapshots, ['monster'], 'no prepare face → single-shot path') + assert.deepEqual([...rig.writes.keys()], []) + }) + + test('a session going live mid-fold is skipped silently (no write, no warn)', async () => { + const ctx = new Context() + const rig = armSliced(ctx, new Map([['racer', fakeEvents(4_001)]]), { liveOnPrepare: new Set(['racer']) }) + const lines = captureLogs(ctx) + const dispose = watchActivityBackfill(ctx) + await trigger(rig.route) + await new Promise(resolve => setTimeout(resolve, 250)) + dispose() + assert.deepEqual(rig.prepares, ['racer']) + assert.deepEqual([...rig.writes.keys()], [], 'no write over a session that went live') + assert.deepEqual(lines.filter(l => l.level === 'warn'), [], 'a live race never warns') + }) + + test('a prepare failure with the session still cold warns and the run continues', async () => { + const ctx = new Context() + const rig = armSliced(ctx, new Map([['broken', fakeEvents(4_001)], ['good', fakeEvents(10)]]), { failPrepare: new Set(['broken']) }) + const lines = captureLogs(ctx) + const dispose = watchActivityBackfill(ctx) + await trigger(rig.route) + await until(() => (rig.coldSnapshots.length === 1 ? true : undefined), 'the small session folded') + await new Promise(resolve => setTimeout(resolve, 250)) + dispose() + assert.deepEqual([...rig.writes.keys()], []) + const warns = lines.filter(l => l.level === 'warn') + assert.equal(warns.length, 1, 'the genuine prepare failure keeps the per-session warn') + assert.match(warns[0].text, /"broken"/) + assert.match(warns[0].text, /corrupt header/) + assert.deepEqual(rig.coldSnapshots, ['good'], 'the next session still folds') + }) + + test('a rejecting write-back warns (fail-soft) and the session still counts as folded', async () => { + const ctx = new Context() + const rig = armSliced(ctx, new Map([['monster', fakeEvents(4_001)]]), { failWrite: true }) + const lines = captureLogs(ctx) + const dispose = watchActivityBackfill(ctx) + await trigger(rig.route) + await until(() => (lines.some(l => l.text.includes('write-back')) ? true : undefined), 'the write-back warning') + await new Promise(resolve => setTimeout(resolve, 250)) + dispose() + const warns = lines.filter(l => l.level === 'warn') + assert.equal(warns.length, 1) + assert.match(warns[0].text, /write-back for "monster" failed/) + assert.ok(lines.some(l => l.level === 'info' && l.text.includes('backfilled for 1 session(s)')), 'the fold itself succeeded') + }) + + test('dispose mid-fold aborts between slices: no write, no completion line', async () => { + const ctx = new Context() + const rig = armSliced(ctx, new Map([['monster', fakeEvents(6_001)]]) ) + const lines = captureLogs(ctx) + const dispose = watchActivityBackfill(ctx) + await trigger(rig.route) + await until(() => (rig.restoreCalls.length >= 1 ? true : undefined), 'the first slice folded') + dispose() + await new Promise(resolve => setTimeout(resolve, 250)) + assert.deepEqual([...rig.writes.keys()], [], 'nothing durable lands mid-fold') + assert.deepEqual(lines.filter(l => l.level === 'info' && l.text.includes('backfilled')), [], 'the pass stopped quietly') + }) +})