imccyu
eb1713f856
* fix(session-projection): keep released units in checkpoints Since the Cordis rc.11 teardown order unloads the Web layer before the base layer, Session Controller and harness-dev-context release their projection units before their live Sessions detach. The projection cache is still alive at that point and wrote a checkpoint holding only the units that were still registered, replacing the complete record. A blank Session then lost sessionListMetadata and listed as an untitled conversation after every graceful restart. A released unit now stays in the registry as a predecessor: it leaves event drives, snapshots, restores, and change notifications, while checkpoint() keeps each Session's last cell of it without advancing it. Registering the same key again replaces the predecessor and refolds the log. The projection cache keeps replacing whole records. * test(session-projection): assert frozen released units across unload tests Update the remaining unload tests that still expected a released unit to vanish from checkpoint(): user-approval, agent-loop inbox and scope lifecycle, time-context, and the agent registry now assert that the same Session keeps its pre-unload row, that a later event does not advance it, and that a Session created after the unload has no row. The token-meter and goal unload tests also append an event after the unload. Document that a retained handle read still advances a released cell that checkpoint() serves, and that restore() folds only live units, so its refreshed rows omit released units. * docs(session-projection): sync the ProjectionHandle type-equiv block * docs(session-projection): record frozen released units and their upgrade Add the upgrade guide for checkpoint() keeping a released unit's last row, update the registration-handles Agent Note with the frozen-cell decision and the retained predecessor, list both consequences under the registry README's known limitations, note in the projection cache README that a stored row can be frozen at its unit's release, and say that restoreFloor() ignores released predecessors. Update the agent-loop resource-lifecycle unload test, which read checkpoints through an alias, to the frozen-row semantics. * docs(upgrade-guide): mark the released-checkpoint snippet ignore-check * docs(session-projection): state the retained-handle path and narrow a rejected alternative * docs(upgrade-guide): re-record the released-checkpoint guide pairing
951 lines
41 KiB
TypeScript
951 lines
41 KiB
TypeScript
/**
|
|
* SessionProjectionCache behavior: mandatory-point writes (turn/end, detach),
|
|
* count/interval throttling between them, fail-soft durability (a failed
|
|
* write logs and stays stale, never throws into the event path), and the
|
|
* synchronous cached listing read. The durable medium is the
|
|
* `session_projcache` storage domain in per-record layout: one
|
|
* version-stamped document per session under the json backend root at
|
|
* `<root>/session_projcache/sessions/<id>.json`. Reads never touch the
|
|
* medium — they come from the domain's in-memory tables, which writes mutate
|
|
* only after durability.
|
|
*/
|
|
|
|
import { afterEach, describe, expect, it, vi } from 'vitest'
|
|
import { mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'
|
|
import { tmpdir } from 'node:os'
|
|
import { dirname, join } from 'node:path'
|
|
import { Context } from '@deepseek-ai/cordis'
|
|
import { z } from 'zod'
|
|
import SessionStore, {
|
|
SESSION_FORMAT_VERSION,
|
|
Session,
|
|
SessionId,
|
|
SessionLogOffset,
|
|
SessionSeq,
|
|
} from '@deepseek-ai/dsh-session'
|
|
import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
|
|
import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection'
|
|
import type { ProjectionDefinition } from '@deepseek-ai/dsh-session-projection'
|
|
import type { SessionProjectionSummaryExcluded } from '@deepseek-ai/dsh-session-projection/types'
|
|
import Storage from '@deepseek-ai/dsh-storage'
|
|
import {
|
|
apply as storageJsonApply, Config as storageJsonConfig, inject as storageJsonInject, name as storageJsonName,
|
|
} from '@deepseek-ai/dsh-storage-json'
|
|
import {
|
|
apply as storageDomainApply, Config as storageDomainConfig, inject as storageDomainInject, name as storageDomainName,
|
|
} from '@deepseek-ai/dsh-storage-domain'
|
|
import SessionProjectionCache from '../src/index.ts'
|
|
import { checkpointRecord, projectionCacheDomainSpec } from '../src/spec.ts'
|
|
import type { CheckpointRecord } from '../src/spec.ts'
|
|
|
|
declare module '@deepseek-ai/dsh-session-projection/types' {
|
|
interface SessionProjectionMap {
|
|
'cache-test/marks': { marks: string[] }
|
|
'cache-test/secondary-marks': { marks: string[] }
|
|
'cache-test/marks3': { marks: string[] }
|
|
'cache-test/summary-excluded': SessionProjectionSummaryExcluded<{ n: number }>
|
|
title: string | null
|
|
}
|
|
}
|
|
|
|
declare module '@deepseek-ai/dsh-session/types' {
|
|
interface SessionEventMap {
|
|
'cache-test/mark': { marks: string[] }
|
|
}
|
|
|
|
interface OutOfBandSessionEventMap {
|
|
'cache-test/mark': true
|
|
}
|
|
}
|
|
|
|
type MarksState = { marks: string[] } | null
|
|
const marksUnit = (stateVersion = 1) => ({
|
|
key: 'cache-test/marks',
|
|
stateSchema: z.object({ marks: z.array(z.string()) }).nullable(),
|
|
init: () => null,
|
|
apply: (state, event) => (event.type === 'cache-test/mark' ? (event).data : state),
|
|
wire: {
|
|
viewSchema: z.object({ marks: z.array(z.string()) }),
|
|
view: state => state ?? { marks: [] },
|
|
},
|
|
stateVersion,
|
|
}) satisfies ProjectionDefinition<'cache-test/marks', MarksState>
|
|
|
|
const marks3Unit = {
|
|
key: 'cache-test/marks3',
|
|
stateSchema: z.object({ marks: z.array(z.string()) }).nullable(),
|
|
init: () => null,
|
|
apply: state => state,
|
|
wire: {
|
|
viewSchema: z.object({ marks: z.array(z.string()) }),
|
|
view: state => state ?? { marks: [] },
|
|
},
|
|
stateVersion: 1,
|
|
} satisfies ProjectionDefinition<'cache-test/marks3', MarksState>
|
|
|
|
const secretUnit = {
|
|
key: 'cache-test/secret',
|
|
stateSchema: z.string(),
|
|
init: () => '',
|
|
apply: state => state,
|
|
stateVersion: 1,
|
|
} satisfies ProjectionDefinition<'cache-test/secret', string>
|
|
|
|
const secondaryMarksUnit = {
|
|
key: 'cache-test/secondary-marks',
|
|
stateSchema: z.object({ marks: z.array(z.string()) }).nullable(),
|
|
init: () => null,
|
|
apply: (state, event) => event.type === 'cache-test/mark' ? event.data : state,
|
|
wire: {
|
|
viewSchema: z.object({ marks: z.array(z.string()) }),
|
|
view: state => state ?? { marks: [] },
|
|
},
|
|
stateVersion: 1,
|
|
} satisfies ProjectionDefinition<'cache-test/secondary-marks', MarksState>
|
|
|
|
/** A summary-excluded unit: its rows must never reach the summary block. */
|
|
const summaryExcludedUnit = {
|
|
key: 'cache-test/summary-excluded',
|
|
stateSchema: z.object({ n: z.number() }),
|
|
init: () => ({ n: 0 }),
|
|
apply: state => state,
|
|
wire: { viewSchema: z.object({ n: z.number() }), view: state => state, summary: false },
|
|
stateVersion: 1,
|
|
} satisfies ProjectionDefinition<'cache-test/summary-excluded', { n: number }>
|
|
|
|
/** Mirrors the shipped title unit's storage face (stateVersion 1, bare-string state). */
|
|
const titleUnit = {
|
|
key: 'title',
|
|
stateSchema: z.string().nullable(),
|
|
init: () => null,
|
|
apply: state => state,
|
|
wire: { viewSchema: z.string().nullable(), view: state => state },
|
|
stateVersion: 1,
|
|
} satisfies ProjectionDefinition<'title', string | null>
|
|
|
|
/** One session's record document on the per-record medium. */
|
|
const recordPath = (root: string, id: Session['id']): string =>
|
|
join(root, projectionCacheDomainSpec.name, 'sessions', `${String(id)}.json`)
|
|
|
|
/** Header shape for cachedSnapshot calls. */
|
|
const headerOf = (id: SessionId, createdAt = 0, cwd?: string): SessionHeader =>
|
|
({ version: SESSION_FORMAT_VERSION, id, createdAt, isSeeded: false, ...cwd === undefined ? {} : { cwd } })
|
|
|
|
interface HarnessOptions {
|
|
root?: string
|
|
config?: { writeEveryEvents: number; writeIntervalMs: number }
|
|
stateVersion?: number
|
|
}
|
|
|
|
const contexts: Context[] = []
|
|
const roots: string[] = []
|
|
|
|
async function harness(options: HarnessOptions = {}) {
|
|
const root = options.root ?? await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
const ctx = new Context()
|
|
contexts.push(ctx)
|
|
// The cache opens its domain through the storage stack; the json backend
|
|
// lands the per-record tree under this tmp root.
|
|
await ctx.plugin(Storage)
|
|
await ctx.plugin({ name: storageJsonName, inject: storageJsonInject, apply: storageJsonApply, Config: storageJsonConfig }, { root })
|
|
await ctx.plugin({ name: storageDomainName, inject: storageDomainInject, apply: storageDomainApply, Config: storageDomainConfig }, { backend: 'json' })
|
|
await ctx.plugin(SessionStore)
|
|
await ctx.plugin(SessionProjectionRegistry)
|
|
await ctx.sessionProjections.register(marksUnit(options.stateVersion))
|
|
const fiber = await ctx.plugin(SessionProjectionCache, options.config ?? { writeEveryEvents: 100, writeIntervalMs: 60_000 })
|
|
return { ctx, root, fiber, cache: ctx.sessionProjectionCache }
|
|
}
|
|
|
|
const mark = (session: Session, marks: string[]): SessionEvent =>
|
|
session.append('cache-test/mark', { marks })
|
|
|
|
const endTurn = (session: Session): SessionEvent =>
|
|
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
|
|
|
/** Resolve after this Session's next durable cache replacement. */
|
|
function whenWritten(ctx: Context, id: SessionId): Promise<void> {
|
|
return new Promise((resolve) => {
|
|
const dispose = ctx.on('domain/changed', (change) => {
|
|
if (change.domain !== projectionCacheDomainSpec.name
|
|
|| change.table !== 'sessions' || change.key !== id || change.operation !== 'put') return
|
|
dispose()
|
|
resolve()
|
|
})
|
|
})
|
|
}
|
|
|
|
/** Register the secondary unit on its own fiber, so a test can release it before a Session detaches. */
|
|
function releasableSecondaryUnit(ctx: Context) {
|
|
return ctx.plugin(Object.assign(async (inner: Context) => {
|
|
await inner.sessionProjections.register(secondaryMarksUnit)
|
|
}, { inject: ['sessionProjections'] }))
|
|
}
|
|
|
|
/** Create a Session owned by its own fiber; disposing `owner` detaches it. */
|
|
async function ownedSession(ctx: Context, id: SessionId) {
|
|
let session: Session | undefined
|
|
const owner = await ctx.plugin(Object.assign((inner: Context) => {
|
|
session = inner.sessions.create(id)
|
|
}, { inject: ['sessions'] }))
|
|
if (session === undefined) throw new Error('session was not created')
|
|
return { session, owner }
|
|
}
|
|
|
|
/** The stored record for one session id (undefined = absent or unreadable). */
|
|
async function storedRecord(root: string, id: Session['id']): Promise<CheckpointRecord | undefined> {
|
|
try {
|
|
const document = JSON.parse(await readFile(recordPath(root, id), 'utf8')) as { record: unknown }
|
|
return checkpointRecord.parse(document.record)
|
|
} catch {
|
|
return undefined
|
|
}
|
|
}
|
|
|
|
/** The stored rows for one session id (undefined = absent or unreadable). */
|
|
async function storedRows(root: string, id: Session['id']): Promise<CheckpointRecord['rows'] | undefined> {
|
|
return (await storedRecord(root, id))?.rows
|
|
}
|
|
|
|
/** Pre-seed one session's record document with a stored checkpoint record. */
|
|
async function seedRecord(
|
|
root: string,
|
|
id: string,
|
|
rows: CheckpointRecord['rows'],
|
|
identity: CheckpointRecord['identity'] = {
|
|
formatVersion: SESSION_FORMAT_VERSION,
|
|
createdAt: 0,
|
|
isSeeded: false,
|
|
inheritedEventCount: SessionLogOffset(0),
|
|
},
|
|
): Promise<void> {
|
|
const path = recordPath(root, SessionId(id))
|
|
await mkdir(dirname(path), { recursive: true })
|
|
await writeFile(path, JSON.stringify({ version: projectionCacheDomainSpec.version, record: { identity, rows } }))
|
|
}
|
|
|
|
afterEach(async () => {
|
|
vi.useRealTimers()
|
|
await Promise.all(contexts.splice(0).map(ctx => ctx.fiber.dispose()))
|
|
await Promise.all(roots.splice(0).map(root => rm(root, { recursive: true, force: true, maxRetries: 10, retryDelay: 100 })))
|
|
})
|
|
|
|
describe('SessionProjectionCache write policy', () => {
|
|
it('writes a durable checkpoint at turn/end (mandatory point)', async () => {
|
|
const { ctx, root } = await harness()
|
|
// The interval cannot substitute for the mandatory turn/end trigger.
|
|
vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] })
|
|
const id = SessionId('turn-end')
|
|
const created = whenWritten(ctx, id)
|
|
const session = ctx.sessions.create(id)
|
|
mark(session, ['a'])
|
|
// The mark is throttled, so the creation cut has no marks folded.
|
|
await created
|
|
expect((await storedRows(root, session.id))?.['cache-test/marks']?.seq).toBe(-1)
|
|
const written = whenWritten(ctx, id)
|
|
const end = endTurn(session)
|
|
await written
|
|
expect((await storedRows(root, session.id))?.['cache-test/marks'])
|
|
.toEqual({ ver: 1, seq: end.seq, val: { marks: ['a'] } })
|
|
})
|
|
|
|
it('writes a checkpoint at session creation, capturing the seed-derived cut', async () => {
|
|
const { ctx, root } = await harness()
|
|
// A forked child seeded with its ancestor's title-like event: no
|
|
// conversation follows, yet the creation write must capture the fold so
|
|
// a crash or a live-held fork still lists the derived value.
|
|
const id = SessionId('seeded')
|
|
const created = whenWritten(ctx, id)
|
|
const session = ctx.sessions.create(id, {
|
|
seed: [{ type: 'cache-test/mark', seq: 0, time: 1, data: { marks: ['seed'] } }] as SessionEvent[],
|
|
})
|
|
await created
|
|
expect((await storedRows(root, session.id))?.['cache-test/marks']?.val)
|
|
.toEqual({ marks: ['seed'] })
|
|
})
|
|
|
|
it('writes at session disposal (detach, the live-to-cold moment)', async () => {
|
|
const { ctx, root } = await harness()
|
|
vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] })
|
|
const id = SessionId('detach')
|
|
const created = whenWritten(ctx, id)
|
|
// Sessions dispose with their owning fiber: create in a child plugin.
|
|
let session: Session | undefined
|
|
const owner = await ctx.plugin(Object.assign((inner: Context) => {
|
|
session = inner.sessions.create(id)
|
|
}, { inject: ['sessions'] }))
|
|
if (session === undefined) throw new Error('session was not created')
|
|
await created
|
|
mark(session, ['live'])
|
|
const written = whenWritten(ctx, id)
|
|
await owner.dispose()
|
|
await written
|
|
expect((await storedRows(root, id))?.['cache-test/marks']?.val).toEqual({ marks: ['live'] })
|
|
})
|
|
|
|
it('writes the last state of a unit released before the detach write', async () => {
|
|
const { ctx, root } = await harness()
|
|
const unitOwner = await releasableSecondaryUnit(ctx)
|
|
const id = SessionId('released-exact')
|
|
const created = whenWritten(ctx, id)
|
|
const { session, owner } = await ownedSession(ctx, id)
|
|
await created
|
|
mark(session, ['a'])
|
|
const ended = whenWritten(ctx, id)
|
|
const end = endTurn(session)
|
|
await ended
|
|
// A plugin unloading ahead of the Sessions it describes releases its unit first.
|
|
await unitOwner.dispose()
|
|
const detached = whenWritten(ctx, id)
|
|
await owner.dispose()
|
|
await detached
|
|
expect(await storedRows(root, id)).toEqual({
|
|
'cache-test/marks': { ver: 1, seq: end.seq, val: { marks: ['a'] } },
|
|
'cache-test/secondary-marks': { ver: 1, seq: end.seq, val: { marks: ['a'] } },
|
|
})
|
|
})
|
|
|
|
it('keeps a released unit at the event it reached while later events advance the live units', async () => {
|
|
const { ctx, root } = await harness()
|
|
const unitOwner = await releasableSecondaryUnit(ctx)
|
|
const id = SessionId('released-behind')
|
|
const created = whenWritten(ctx, id)
|
|
const { session, owner } = await ownedSession(ctx, id)
|
|
await created
|
|
mark(session, ['a'])
|
|
const ended = whenWritten(ctx, id)
|
|
const end = endTurn(session)
|
|
await ended
|
|
await unitOwner.dispose()
|
|
const later = mark(session, ['b'])
|
|
const detached = whenWritten(ctx, id)
|
|
await owner.dispose()
|
|
await detached
|
|
expect(await storedRows(root, id)).toEqual({
|
|
'cache-test/marks': { ver: 1, seq: later.seq, val: { marks: ['b'] } },
|
|
'cache-test/secondary-marks': { ver: 1, seq: end.seq, val: { marks: ['a'] } },
|
|
})
|
|
})
|
|
|
|
it('drops a stored row whose unit this process never registered', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
const meta = headerOf(SessionId('carried'), 9)
|
|
await seedRecord(root, meta.id, {
|
|
'cache-test/marks': { ver: 1, seq: -1, val: null },
|
|
'cache-test/secret': { ver: 1, seq: -1, val: 'written by another composition' },
|
|
}, {
|
|
formatVersion: SESSION_FORMAT_VERSION,
|
|
createdAt: 9,
|
|
isSeeded: false,
|
|
inheritedEventCount: SessionLogOffset(0),
|
|
})
|
|
const { ctx } = await harness({ root })
|
|
const events: SessionEvent[] = [{ type: 'cache-test/mark', seq: SessionSeq(0), time: 0, data: { marks: ['cold'] } }]
|
|
const written = whenWritten(ctx, meta.id)
|
|
ctx.sessionProjectionCache.coldSnapshot(meta, SessionLogOffset(0), events)
|
|
await written
|
|
expect(await storedRows(root, meta.id)).toEqual({
|
|
'cache-test/marks': { ver: 1, seq: 0, val: { marks: ['cold'] } },
|
|
})
|
|
})
|
|
|
|
it('never keeps a stored row of another lifecycle under the same Session id', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
const id = SessionId('recreated')
|
|
// Same id and cut as the Session created below, but another creation time.
|
|
await seedRecord(root, id, {
|
|
'cache-test/secondary-marks': { ver: 1, seq: -1, val: { marks: ['foreign'] } },
|
|
}, {
|
|
formatVersion: SESSION_FORMAT_VERSION,
|
|
createdAt: 1,
|
|
isSeeded: false,
|
|
inheritedEventCount: SessionLogOffset(0),
|
|
})
|
|
const { ctx } = await harness({ root })
|
|
const created = whenWritten(ctx, id)
|
|
ctx.sessions.create(id)
|
|
await created
|
|
expect(await storedRows(root, id)).toEqual({
|
|
'cache-test/marks': { ver: 1, seq: -1, val: null },
|
|
})
|
|
})
|
|
|
|
it('flushes when the in-turn event count reaches the configured threshold', async () => {
|
|
const { ctx, root } = await harness({ config: { writeEveryEvents: 3, writeIntervalMs: 60_000 } })
|
|
vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] })
|
|
const id = SessionId('count')
|
|
const created = whenWritten(ctx, id)
|
|
const session = ctx.sessions.create(id)
|
|
mark(session, ['1'])
|
|
mark(session, ['2'])
|
|
await created
|
|
expect((await storedRows(root, id))?.['cache-test/marks'])
|
|
.toEqual({ ver: 1, seq: -1, val: null }) // still the creation cut
|
|
const written = whenWritten(ctx, id)
|
|
mark(session, ['3'])
|
|
await written
|
|
expect((await storedRows(root, id))?.['cache-test/marks']?.val).toEqual({ marks: ['3'] })
|
|
})
|
|
|
|
it('flushes on the configured interval when the count threshold is not reached', async () => {
|
|
const { ctx, cache } = await harness({ config: { writeEveryEvents: 100, writeIntervalMs: 20 } })
|
|
const write = vi.spyOn(cache, 'write').mockResolvedValue()
|
|
vi.useFakeTimers()
|
|
const session = ctx.sessions.create(SessionId('interval'))
|
|
write.mockClear()
|
|
mark(session, ['slow'])
|
|
await vi.advanceTimersByTimeAsync(19)
|
|
expect(write).not.toHaveBeenCalled()
|
|
await vi.advanceTimersByTimeAsync(1)
|
|
expect(write).toHaveBeenCalledExactlyOnceWith(session)
|
|
})
|
|
|
|
it('write() on a never-dirty session checkpoints directly and rejects a non-JSON unit state', async () => {
|
|
const { ctx, root } = await harness()
|
|
// Never dirtied: no events — write() still lands the init-derived cut.
|
|
const clean = ctx.sessions.create(SessionId('clean-write'))
|
|
await ctx.sessionProjectionCache.write(clean)
|
|
expect((await storedRows(root, clean.id))?.['cache-test/marks']).toEqual({ ver: 1, seq: -1, val: null })
|
|
// A unit whose state violates the plain-JSON contract fails the write loud.
|
|
await ctx.sessionProjections.register({
|
|
key: 'cache-test/marks2',
|
|
stateSchema: z.custom<Map<string, string>>(() => true),
|
|
init: () => new Map<string, string>(),
|
|
apply: state => state,
|
|
stateVersion: 1,
|
|
})
|
|
await expect(ctx.sessionProjectionCache.write(clean)).rejects.toThrow('not losslessly JSON-serializable')
|
|
})
|
|
|
|
it('plugin disposal clears armed interval timers and leaves cleaned sessions alone', async () => {
|
|
vi.useFakeTimers()
|
|
const { ctx, root, fiber } = await harness({ config: { writeEveryEvents: 100, writeIntervalMs: 5000 } })
|
|
const armed = ctx.sessions.create(SessionId('armed'))
|
|
const cleaned = ctx.sessions.create(SessionId('cleaned'))
|
|
mark(armed, ['pending']) // timer armed, no write yet
|
|
mark(cleaned, ['done'])
|
|
endTurn(cleaned) // mandatory write; markClean leaves {pending: 0, timer: undefined} in the map
|
|
await vi.advanceTimersByTimeAsync(0)
|
|
await fiber.dispose()
|
|
// The armed timer died with the plugin: advancing time writes nothing.
|
|
await vi.advanceTimersByTimeAsync(10_000)
|
|
// Only the creation cut exists: the armed mark never wrote.
|
|
expect((await storedRows(root, armed.id))?.['cache-test/marks']?.seq).toBe(-1)
|
|
})
|
|
|
|
it('contains a durable write failure: logs a warning, event path unharmed, next write self-heals', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
const ctx = new Context()
|
|
contexts.push(ctx)
|
|
await ctx.plugin(Storage)
|
|
await ctx.plugin({ name: storageJsonName, inject: storageJsonInject, apply: storageJsonApply, Config: storageJsonConfig }, { root })
|
|
await ctx.plugin({ name: storageDomainName, inject: storageDomainInject, apply: storageDomainApply, Config: storageDomainConfig }, { backend: 'json' })
|
|
await ctx.plugin(SessionStore)
|
|
await ctx.plugin(SessionProjectionRegistry)
|
|
await ctx.sessionProjections.register(marksUnit())
|
|
await ctx.plugin(SessionProjectionCache, { writeEveryEvents: 100, writeIntervalMs: 60_000 })
|
|
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
|
|
// A directory where the record document must land makes the atomic
|
|
// rename fail — including the creation write, so no row ever lands.
|
|
const blocker = recordPath(root, SessionId('fail-soft'))
|
|
await mkdir(blocker, { recursive: true })
|
|
const session = ctx.sessions.create(SessionId('fail-soft'))
|
|
mark(session, ['x'])
|
|
endTurn(session)
|
|
// The failed creation/turn-end writes are fire-and-forget: wait for the
|
|
// warn (the write actually failed), then assert no row landed — the
|
|
// property under test is that a failed write leaves no partial row.
|
|
await vi.waitFor(() => {
|
|
expect(warn).toHaveBeenCalledWith(expect.stringContaining('turn/end write for "fail-soft" failed'))
|
|
}, { timeout: 5_000 })
|
|
expect(await storedRows(root, session.id)).toBeUndefined()
|
|
// Self-heal: once the blocker clears, the next mandatory point writes.
|
|
await rm(recordPath(root, session.id), { recursive: true })
|
|
vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] })
|
|
mark(session, ['y'])
|
|
const written = whenWritten(ctx, session.id)
|
|
endTurn(session)
|
|
await written
|
|
expect((await storedRows(root, session.id))?.['cache-test/marks']?.val).toEqual({ marks: ['y'] })
|
|
})
|
|
})
|
|
|
|
describe('SessionProjectionCache listing read', () => {
|
|
it('rejects a nonzero inherited cut for an unseeded header on the fold face', async () => {
|
|
const { cache } = await harness()
|
|
|
|
expect(() => cache.coldSnapshot(
|
|
headerOf(SessionId('invalid-unseeded-cut')),
|
|
SessionLogOffset(1),
|
|
[],
|
|
)).toThrow('unseeded projection-cache identity inherited event count must be 0')
|
|
})
|
|
|
|
it('serves rows with differing watermarks as one block at the lowest served watermark', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
await seedRecord(root, 'watermark-lower', {
|
|
'cache-test/marks': { ver: 1, seq: SessionSeq(4), val: { marks: ['primary'] } },
|
|
'cache-test/secondary-marks': { ver: 1, seq: SessionSeq(2), val: { marks: ['secondary'] } },
|
|
})
|
|
await seedRecord(root, 'watermark-higher', {
|
|
'cache-test/marks': { ver: 1, seq: SessionSeq(4), val: { marks: ['primary'] } },
|
|
'cache-test/secondary-marks': { ver: 1, seq: SessionSeq(6), val: { marks: ['secondary'] } },
|
|
})
|
|
const { ctx, cache } = await harness({ root })
|
|
await ctx.sessionProjections.register(secondaryMarksUnit)
|
|
|
|
// The block carries one watermark: the seq every served value has folded
|
|
// through at least, so the lower of the two rows names the block.
|
|
for (const [id, asOfSeq] of [['watermark-lower', 2], ['watermark-higher', 4]] as const) {
|
|
expect(cache.cachedSnapshot(headerOf(SessionId(id)))).toEqual({
|
|
asOfSeq,
|
|
values: {
|
|
'cache-test/marks': { marks: ['primary'] },
|
|
'cache-test/secondary-marks': { marks: ['secondary'] },
|
|
},
|
|
})
|
|
}
|
|
})
|
|
|
|
it('serves a seeded lifecycle by header alone while the fold face still refuses another cut', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
const id = SessionId('cut-identity')
|
|
await seedRecord(
|
|
root,
|
|
id,
|
|
{ 'cache-test/marks': { ver: 1, seq: SessionSeq(1), val: { marks: ['seed'] } } },
|
|
{
|
|
formatVersion: SESSION_FORMAT_VERSION,
|
|
createdAt: 0,
|
|
isSeeded: true,
|
|
inheritedEventCount: SessionLogOffset(2),
|
|
},
|
|
)
|
|
const { cache } = await harness({ root })
|
|
const seededHeader = { ...headerOf(id), isSeeded: true }
|
|
const log: SessionEvent[] = [
|
|
{ type: 'turn/start', seq: SessionSeq(0), time: 0, data: { turn: 1 } },
|
|
{ type: 'cache-test/mark', seq: SessionSeq(1), time: 1, data: { marks: ['logged'] } },
|
|
]
|
|
|
|
// The listing knows only the header: the record's lifecycle matches, so
|
|
// its rows are viewed without any cut.
|
|
expect(cache.cachedSnapshot(seededHeader))
|
|
.toEqual({ asOfSeq: 1, values: { 'cache-test/marks': { marks: ['seed'] } } })
|
|
// An unseeded header names another lifecycle even though the id matches.
|
|
expect(cache.cachedSnapshot(headerOf(id))).toBeUndefined()
|
|
// The fold face holds the exact cut: the matching cut seeds from the row,
|
|
// another cut refolds the whole log instead of continuing a foreign state.
|
|
expect(cache.coldSnapshot(seededHeader, SessionLogOffset(2), log).values['cache-test/marks'])
|
|
.toEqual({ marks: ['seed'] })
|
|
expect(cache.coldSnapshot(seededHeader, SessionLogOffset(1), log).values['cache-test/marks'])
|
|
.toEqual({ marks: ['logged'] })
|
|
})
|
|
|
|
it('serves a seeded lifecycle\'s predecessor title by header alone', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
const id = SessionId('seeded-predecessor')
|
|
await seedRecord(
|
|
root,
|
|
id,
|
|
{
|
|
'cache-test/marks': { ver: 1, seq: SessionSeq(3), val: { marks: ['fork'] } },
|
|
title: { ver: 1, seq: SessionSeq(3), val: 'forked title' },
|
|
},
|
|
{
|
|
formatVersion: SESSION_FORMAT_VERSION - 1,
|
|
createdAt: 0,
|
|
isSeeded: true,
|
|
inheritedEventCount: SessionLogOffset(5),
|
|
},
|
|
)
|
|
const { ctx, cache } = await harness({ root })
|
|
await ctx.sessionProjections.register(titleUnit)
|
|
const seededHeader = { ...headerOf(id), isSeeded: true }
|
|
|
|
// An older format generation never serves the full block, but its title
|
|
// survives the format edge, for a seeded lifecycle as for an unseeded one.
|
|
expect(cache.cachedSnapshot(seededHeader)).toBeUndefined()
|
|
expect(cache.cachedPredecessorTitle(seededHeader))
|
|
.toEqual({ asOfSeq: 3, values: { title: 'forked title' } })
|
|
// An unseeded header names another lifecycle: no title either.
|
|
expect(cache.cachedPredecessorTitle(headerOf(id))).toBeUndefined()
|
|
})
|
|
|
|
it('serves a creation-time checkpoint at the before-first-event cursor', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
await seedRecord(root, 'before-first-event', {
|
|
'cache-test/marks': { ver: 1, seq: -1, val: null },
|
|
})
|
|
const { cache } = await harness({ root })
|
|
|
|
expect(cache.cachedSnapshot(headerOf(SessionId('before-first-event'))))
|
|
.toEqual({ asOfSeq: -1, values: { 'cache-test/marks': { marks: [] } } })
|
|
})
|
|
|
|
it('refuses a checkpoint folded from another Session format generation', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
const id = SessionId('format-identity')
|
|
await seedRecord(
|
|
root,
|
|
id,
|
|
{ 'cache-test/marks': { ver: 1, seq: SessionSeq(0), val: { marks: ['stale'] } } },
|
|
{
|
|
formatVersion: SESSION_FORMAT_VERSION + 1,
|
|
createdAt: 0,
|
|
isSeeded: false,
|
|
inheritedEventCount: SessionLogOffset(0),
|
|
},
|
|
)
|
|
const { cache } = await harness({ root })
|
|
|
|
expect(cache.cachedSnapshot(headerOf(id))).toBeUndefined()
|
|
})
|
|
|
|
it('keeps host-only checkpoint state out of cached wire snapshots', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
await seedRecord(root, 'host-state', {
|
|
'cache-test/marks': { ver: 1, seq: SessionSeq(4), val: { marks: ['wire'] } },
|
|
'cache-test/secret': { ver: 1, seq: SessionSeq(4), val: 'private prompt text' },
|
|
})
|
|
const { ctx, cache } = await harness({ root })
|
|
await ctx.sessionProjections.register(secretUnit)
|
|
const header = headerOf(SessionId('host-state'))
|
|
|
|
expect(cache.cachedSnapshot(header)).toEqual({
|
|
asOfSeq: 4,
|
|
values: { 'cache-test/marks': { marks: ['wire'] } },
|
|
})
|
|
expect(JSON.stringify(cache.cachedSnapshot(header)))
|
|
.not.toContain('private prompt text')
|
|
})
|
|
|
|
it('serves lifecycle-matching rows as a cached block and refuses unrelated ones', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
await seedRecord(root, 'listed', {
|
|
'cache-test/marks': { ver: 1, seq: SessionSeq(4), val: { marks: ['t'] } },
|
|
})
|
|
const { cache } = await harness({ root })
|
|
const id = SessionId('listed')
|
|
// Matching header: the values at the stored row's own watermark.
|
|
expect(cache.cachedSnapshot(headerOf(id)))
|
|
.toEqual({ asOfSeq: 4, values: { 'cache-test/marks': { marks: ['t'] } } })
|
|
// A recreated id (different createdAt): the record is unrelated — no block.
|
|
expect(cache.cachedSnapshot(headerOf(id, 777))).toBeUndefined()
|
|
// Unknown id: no block.
|
|
expect(cache.cachedSnapshot(headerOf(SessionId('never-cached'))))
|
|
.toBeUndefined()
|
|
})
|
|
|
|
it('views every served wire row of one record in one cached block', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
await seedRecord(root, 'multi-row', {
|
|
'cache-test/marks': { ver: 1, seq: SessionSeq(4), val: { marks: ['a'] } },
|
|
'cache-test/marks3': { ver: 1, seq: SessionSeq(4), val: { marks: ['b'] } },
|
|
})
|
|
const { ctx, cache } = await harness({ root })
|
|
await ctx.sessionProjections.register(marks3Unit)
|
|
const block = cache.cachedSnapshot(headerOf(SessionId('multi-row')))
|
|
expect(block?.values).toEqual({
|
|
'cache-test/marks': { marks: ['a'] },
|
|
'cache-test/marks3': { marks: ['b'] },
|
|
})
|
|
expect(block?.asOfSeq).toBe(4)
|
|
})
|
|
|
|
it('returns undefined when the stored record version is not accepted', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
// Version 2 is neither current nor declared compatible, so the document
|
|
// is discarded at open and the record reads as absent.
|
|
const path = recordPath(root, SessionId('all-stale'))
|
|
await mkdir(dirname(path), { recursive: true })
|
|
await writeFile(path, JSON.stringify({
|
|
version: 2,
|
|
record: {
|
|
identity: {
|
|
formatVersion: SESSION_FORMAT_VERSION,
|
|
createdAt: 0,
|
|
isSeeded: false,
|
|
inheritedEventCount: 0,
|
|
},
|
|
rows: { 'cache-test/marks': { ver: 1, seq: 4, val: { marks: ['old'] } } },
|
|
},
|
|
}))
|
|
const { cache } = await harness({ root })
|
|
expect(cache.cachedSnapshot(headerOf(SessionId('all-stale'))))
|
|
.toBeUndefined()
|
|
})
|
|
|
|
it('serves a pre-lineage record (accepted old version) to an unseeded caller only', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
// A document stamped with an accepted older version whose identity
|
|
// predates the lineage fields: absent lineage reads as unseeded.
|
|
const path = recordPath(root, SessionId('pre-lineage'))
|
|
await mkdir(dirname(path), { recursive: true })
|
|
await writeFile(path, JSON.stringify({
|
|
version: 4,
|
|
record: {
|
|
identity: { formatVersion: SESSION_FORMAT_VERSION, createdAt: 0 },
|
|
rows: { 'cache-test/marks': { ver: 1, seq: 4, val: { marks: ['kept'] } } },
|
|
},
|
|
}))
|
|
const { cache } = await harness({ root })
|
|
const id = SessionId('pre-lineage')
|
|
// Unseeded caller: the absent lineage is exactly its identity — served.
|
|
expect(cache.cachedSnapshot(headerOf(id)))
|
|
.toEqual({ asOfSeq: 4, values: { 'cache-test/marks': { marks: ['kept'] } } })
|
|
// Seeded caller: the lineage-less record cannot vouch for a seeded lifecycle — refused.
|
|
expect(cache.cachedSnapshot({ ...headerOf(id), isSeeded: true }))
|
|
.toBeUndefined()
|
|
// The fold face reads the absent lineage as the unseeded cut 0: the
|
|
// unseeded caller continues from the row, the seeded caller refolds.
|
|
const log: SessionEvent[] = [0, 1, 2, 3, 4].map(seq => ({
|
|
type: 'cache-test/mark', seq: SessionSeq(seq), time: seq, data: { marks: [`m${seq}`] },
|
|
}))
|
|
expect(cache.coldSnapshot(headerOf(id), SessionLogOffset(0), log).values['cache-test/marks'])
|
|
.toEqual({ marks: ['kept'] })
|
|
expect(cache.coldSnapshot({ ...headerOf(id), isSeeded: true }, SessionLogOffset(1), log).values['cache-test/marks'])
|
|
.toEqual({ marks: ['m4'] })
|
|
})
|
|
|
|
it('refuses an accepted predecessor record without a Session format generation', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
const id = SessionId('pre-format-identity')
|
|
const path = recordPath(root, id)
|
|
await mkdir(dirname(path), { recursive: true })
|
|
await writeFile(path, JSON.stringify({
|
|
version: 6,
|
|
record: {
|
|
identity: { createdAt: 0, isSeeded: false, inheritedEventCount: 0 },
|
|
rows: { 'cache-test/marks': { ver: 1, seq: 4, val: { marks: ['unbound'] } } },
|
|
},
|
|
}))
|
|
const { cache } = await harness({ root })
|
|
|
|
expect(cache.cachedSnapshot(headerOf(id))).toBeUndefined()
|
|
})
|
|
|
|
it('returns undefined when every stored row is version-mismatched', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
// A current document whose rows all fail the live unit's stateVersion:
|
|
// the listing view is empty, so no block is served.
|
|
await seedRecord(root, 'row-stale', {
|
|
'cache-test/marks': { ver: 99, seq: SessionSeq(4), val: { marks: ['old'] } },
|
|
})
|
|
const { cache } = await harness({ root })
|
|
expect(cache.cachedSnapshot(headerOf(SessionId('row-stale'))))
|
|
.toBeUndefined()
|
|
})
|
|
|
|
it('binds identity on cwd too: a matching cwd serves, a moved session does not', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
await seedRecord(root, 'homed', {
|
|
'cache-test/marks': { ver: 1, seq: SessionSeq(2), val: { marks: ['w'] } },
|
|
}, {
|
|
formatVersion: SESSION_FORMAT_VERSION,
|
|
createdAt: 0,
|
|
cwd: '/work',
|
|
isSeeded: false,
|
|
inheritedEventCount: SessionLogOffset(0),
|
|
})
|
|
const { cache } = await harness({ root })
|
|
const id = SessionId('homed')
|
|
expect(cache.cachedSnapshot(headerOf(id, 0, '/work'))?.values['cache-test/marks'])
|
|
.toEqual({ marks: ['w'] })
|
|
expect(cache.cachedSnapshot(headerOf(id, 0, '/elsewhere'))).toBeUndefined()
|
|
expect(cache.cachedSnapshot(headerOf(id, 0))).toBeUndefined()
|
|
})
|
|
|
|
it('returns undefined for a malformed record document (refold from the log on the caller side)', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
const path = recordPath(root, SessionId('malformed'))
|
|
await mkdir(dirname(path), { recursive: true })
|
|
await writeFile(path, 'not json at all')
|
|
const { cache } = await harness({ root })
|
|
expect(cache.cachedSnapshot(headerOf(SessionId('malformed'))))
|
|
.toBeUndefined()
|
|
})
|
|
})
|
|
|
|
describe('SessionProjectionCache cold-read seeding', () => {
|
|
/** One session's event log: turn/start, one mark per group, turn/end. */
|
|
const storedLog = (marks: string[][]): SessionEvent[] => {
|
|
const events: SessionEvent[] = [
|
|
{ type: 'turn/start', seq: SessionSeq(0), time: 0, data: { turn: 1 } },
|
|
]
|
|
for (const m of marks) {
|
|
events.push({
|
|
type: 'cache-test/mark',
|
|
seq: SessionSeq(events.length),
|
|
time: events.length,
|
|
data: { marks: m },
|
|
})
|
|
}
|
|
events.push({
|
|
type: 'turn/end',
|
|
seq: SessionSeq(events.length),
|
|
time: events.length,
|
|
data: { turn: 1, reason: { kind: 'completed' } },
|
|
})
|
|
return events
|
|
}
|
|
|
|
it('hydratePrepared seeds from a matching row and retries from the exact log on a malformed one', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
// Records land on disk before the domain opens, so the in-memory table
|
|
// picks them up at init.
|
|
await seedRecord(root, 'prepared-seeded', {
|
|
'cache-test/marks': { ver: 1, seq: SessionSeq(1), val: { marks: ['cached'] } },
|
|
})
|
|
await seedRecord(root, 'prepared-fallback', {
|
|
'cache-test/marks': { ver: 1, seq: SessionSeq(1), val: { marks: 'malformed' } },
|
|
})
|
|
const { cache } = await harness({ root })
|
|
const events = storedLog([['fresh']])
|
|
|
|
// A matching row hydrates the prepared Session without a persistence read.
|
|
const seeded = headerOf(SessionId('prepared-seeded'))
|
|
const seededSession = Session.create(seeded.id, events, seeded)
|
|
expect(cache.hydratePrepared(seededSession, events)).toEqual({
|
|
asOfSeq: 2,
|
|
values: { 'cache-test/marks': { marks: ['cached'] } },
|
|
})
|
|
|
|
// A malformed row cannot seed the fold; hydration falls back to the
|
|
// exact log so a valid Session stays readable.
|
|
const fallback = headerOf(SessionId('prepared-fallback'))
|
|
const fallbackSession = Session.create(fallback.id, events, fallback)
|
|
expect(cache.hydratePrepared(fallbackSession, events)).toEqual({
|
|
asOfSeq: 2,
|
|
values: { 'cache-test/marks': { marks: ['fresh'] } },
|
|
})
|
|
|
|
// No row at all: hydrate from init over the exact log.
|
|
const bare = headerOf(SessionId('prepared-bare'))
|
|
const bareSession = Session.create(bare.id, events, bare)
|
|
expect(cache.hydratePrepared(bareSession, events)).toEqual({
|
|
asOfSeq: 2,
|
|
values: { 'cache-test/marks': { marks: ['fresh'] } },
|
|
})
|
|
})
|
|
|
|
it('coldSnapshot traverses the full log but applies only the events after each cached watermark', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
// A cached row covering the prefix through seq 2 (three applies folded).
|
|
await seedRecord(root, 'cold-snap', {
|
|
'cache-test/count': { ver: 1, seq: SessionSeq(2), val: 3 },
|
|
}, {
|
|
formatVersion: SESSION_FORMAT_VERSION,
|
|
createdAt: 9,
|
|
isSeeded: false,
|
|
inheritedEventCount: SessionLogOffset(0),
|
|
})
|
|
const { cache, ctx } = await harness({ root })
|
|
const apply = vi.fn((_state: number, _event: SessionEvent) => 1)
|
|
await ctx.sessionProjections.register({
|
|
key: 'cache-test/count',
|
|
stateSchema: z.number().int().nonnegative(),
|
|
init: () => 0,
|
|
apply,
|
|
stateVersion: 1,
|
|
} satisfies ProjectionDefinition<'cache-test/count', number>)
|
|
const meta = headerOf(SessionId('cold-snap'), 9)
|
|
const events = Array.from({ length: 5 }, (_, seq) => ({
|
|
type: 'cache-test/mark', seq: SessionSeq(seq), time: seq, data: { marks: [`m${seq}`] },
|
|
})) as SessionEvent[]
|
|
const refreshed = whenWritten(ctx, meta.id)
|
|
const snapshot = cache.coldSnapshot(meta, SessionLogOffset(0), events)
|
|
// The full log was traversed, but the fold applied only seqs 3 and 4.
|
|
expect(apply).toHaveBeenCalledTimes(2)
|
|
expect(apply.mock.calls.map(call => call[1].seq)).toEqual([3, 4])
|
|
expect(snapshot.asOfSeq).toBe(4)
|
|
// Host-only unit: folded but not served; the refreshed row is written
|
|
// back (fail-soft, fire-and-forget) once the write lands.
|
|
expect(Object.keys(snapshot.values)).not.toContain('cache-test/count')
|
|
await refreshed
|
|
expect((await storedRows(root, meta.id))?.['cache-test/count']?.seq).toBe(4)
|
|
// No cached row yet: the first cold read folds from init over the full
|
|
// log and creates the cache row (the `?? {}` seed path).
|
|
const fresh = headerOf(SessionId('cold-fresh'), 10)
|
|
const created = whenWritten(ctx, fresh.id)
|
|
cache.coldSnapshot(fresh, SessionLogOffset(0), events)
|
|
expect(apply).toHaveBeenCalledTimes(7) // 2 tail + 5 full
|
|
await created
|
|
expect((await storedRows(root, fresh.id))?.['cache-test/count']?.seq).toBe(4)
|
|
})
|
|
|
|
it('coldSnapshot write-back is fail-soft: a failed durable write logs and never throws', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
const ctx = new Context()
|
|
contexts.push(ctx)
|
|
await ctx.plugin(Storage)
|
|
await ctx.plugin({ name: storageJsonName, inject: storageJsonInject, apply: storageJsonApply, Config: storageJsonConfig }, { root })
|
|
await ctx.plugin({ name: storageDomainName, inject: storageDomainInject, apply: storageDomainApply, Config: storageDomainConfig }, { backend: 'json' })
|
|
await ctx.plugin(SessionStore)
|
|
await ctx.plugin(SessionProjectionRegistry)
|
|
await ctx.sessionProjections.register(marksUnit())
|
|
await ctx.plugin(SessionProjectionCache, { writeEveryEvents: 100, writeIntervalMs: 60_000 })
|
|
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
|
|
// A directory where the record document must land makes the write-back
|
|
// fail; the cold read itself still succeeds and never throws.
|
|
const meta = headerOf(SessionId('cold-fail'))
|
|
await mkdir(recordPath(root, meta.id), { recursive: true })
|
|
expect(ctx.sessionProjectionCache.coldSnapshot(meta, SessionLogOffset(0), [])).toBeDefined()
|
|
// The failed write-back is fire-and-forget: poll for the warn instead of
|
|
// assuming a fixed settle window (slow runners exceed it).
|
|
await vi.waitFor(() => {
|
|
expect(warn).toHaveBeenCalledWith(expect.stringContaining('cold-read write-back for "cold-fail" failed'))
|
|
}, { timeout: 5_000 })
|
|
})
|
|
})
|
|
|
|
describe('SessionProjectionCache summary reads', () => {
|
|
it('serves the summary block from stored rows and excludes the marked unit', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
roots.push(root)
|
|
await seedRecord(root, 'summary-watermark', {
|
|
'cache-test/marks': { ver: 1, seq: SessionSeq(5), val: { marks: ['primary'] } },
|
|
'cache-test/summary-excluded': { ver: 1, seq: SessionSeq(1), val: { n: 3 } },
|
|
})
|
|
const { ctx, cache } = await harness({ root })
|
|
await ctx.sessionProjections.register(summaryExcludedUnit)
|
|
const summaryKeys = ctx.sessionProjections.summaryKeys()
|
|
|
|
// The summary watermark follows the served rows only: the marked row sits
|
|
// at seq 1, and its absence must not drag the summary block down.
|
|
expect(cache.cachedSnapshot(headerOf(SessionId('summary-watermark')), summaryKeys)).toEqual({
|
|
asOfSeq: 5,
|
|
values: { 'cache-test/marks': { marks: ['primary'] } },
|
|
})
|
|
expect(cache.cachedSnapshot(headerOf(SessionId('summary-watermark')))).toEqual({
|
|
asOfSeq: 1,
|
|
values: {
|
|
'cache-test/marks': { marks: ['primary'] },
|
|
'cache-test/summary-excluded': { n: 3 },
|
|
},
|
|
})
|
|
// No record for this id: a key-subset read reports absence like the complete face.
|
|
expect(cache.cachedSnapshot(headerOf(SessionId('summary-absent')), summaryKeys)).toBeUndefined()
|
|
})
|
|
})
|