|
|
@@ -2,21 +2,25 @@
|
|
|
* 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
|
|
|
- * cold-read ladder (cached row + readFrom tail + registry restore +
|
|
|
+ * cold-read ladder (cached file + readFrom tail + registry restore +
|
|
|
* write-back; version bump and shrunk-log rows degrade to a full re-read).
|
|
|
+ * The durable medium is one `projection_cache.json` per session inside the
|
|
|
+ * session's persistence directory (resolved via `sessionPersistence.locate`).
|
|
|
*/
|
|
|
|
|
|
import { afterEach, describe, expect, it, vi } from 'vitest'
|
|
|
+import { mkdtemp, mkdir, 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 Storage from '@deepseek-ai/dsh-storage'
|
|
|
-import { DomainFacility } from '@deepseek-ai/dsh-storage-domain'
|
|
|
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
|
|
|
-import type { Session, SessionEvent } from '@deepseek-ai/dsh-session'
|
|
|
+import type { Session, 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 { MemoryMediaPool, MemoryStorageBackend } from '../../../storage/storage-domain/tests/helpers/memory-backend.ts'
|
|
|
import SessionProjectionCache from '../src/index.ts'
|
|
|
+import { checkpointRecord } from '../src/spec.ts'
|
|
|
+import type { CheckpointRecord } from '../src/spec.ts'
|
|
|
|
|
|
declare module '@deepseek-ai/dsh-session-projection/types' {
|
|
|
interface SessionProjectionStateMap {
|
|
|
@@ -51,8 +55,12 @@ const marksUnit = (stateVersion = 1) => ({
|
|
|
stateVersion,
|
|
|
}) satisfies ProjectionDefinition<'cache-test/marks', MarksState>
|
|
|
|
|
|
-/** A persistence double serving readFrom over a fixed per-id stored log (headers stamp createdAt 0). */
|
|
|
-function fakePersistence(logs: Map<string, SessionEvent[]>) {
|
|
|
+/** One session's cache file inside its persistence directory. */
|
|
|
+const cachePath = (root: string, id: Session['id']): string =>
|
|
|
+ join(root, String(id), 'projection_cache.json')
|
|
|
+
|
|
|
+/** A persistence double serving locate + readFrom over a fixed per-id stored log (headers stamp createdAt 0). */
|
|
|
+function fakePersistence(root: string, logs: Map<string, SessionEvent[]>) {
|
|
|
const readFrom = vi.fn(async (id: SessionId, fromSeq: number) => {
|
|
|
const events = logs.get(String(id))
|
|
|
if (events === undefined) throw new Error(`session "${id}" not found`)
|
|
|
@@ -61,7 +69,10 @@ function fakePersistence(logs: Map<string, SessionEvent[]>) {
|
|
|
events: events.filter(event => event.seq >= fromSeq),
|
|
|
}
|
|
|
})
|
|
|
- return { readFrom }
|
|
|
+ return {
|
|
|
+ readFrom,
|
|
|
+ locate: (meta: SessionHeader) => ({ kind: 'jsonl', path: join(root, String(meta.id), 'session.jsonl') }),
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
/** Header shape for cachedSnapshot calls (fake logs stamp createdAt 0, no cwd). */
|
|
|
@@ -69,31 +80,28 @@ const headerOf = (id: SessionId, createdAt = 0, cwd?: string) =>
|
|
|
({ version: 0, id, createdAt, ...cwd === undefined ? {} : { cwd } })
|
|
|
|
|
|
interface HarnessOptions {
|
|
|
- pool?: MemoryMediaPool
|
|
|
+ root?: string
|
|
|
config?: { writeEveryEvents: number; writeIntervalMs: number }
|
|
|
stateVersion?: number
|
|
|
logs?: Map<string, SessionEvent[]>
|
|
|
}
|
|
|
|
|
|
const contexts: Context[] = []
|
|
|
+const roots: string[] = []
|
|
|
|
|
|
async function harness(options: HarnessOptions = {}) {
|
|
|
- const pool = options.pool ?? new MemoryMediaPool()
|
|
|
+ const root = options.root ?? await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
|
+ roots.push(root)
|
|
|
const logs = options.logs ?? new Map<string, SessionEvent[]>()
|
|
|
const ctx = new Context()
|
|
|
contexts.push(ctx)
|
|
|
- await ctx.plugin(Storage)
|
|
|
- ctx.storage.backend.register('memory', new MemoryStorageBackend(pool))
|
|
|
- const facility = new DomainFacility(ctx, { backend: 'memory', routes: {} })
|
|
|
- ctx.storage.mount('domain', facility)
|
|
|
- ctx.provide('storageDomain', facility)
|
|
|
await ctx.plugin(SessionStore)
|
|
|
await ctx.plugin(SessionProjectionRegistry)
|
|
|
ctx.sessionProjections.register(marksUnit(options.stateVersion))
|
|
|
- const persistence = fakePersistence(logs)
|
|
|
+ const persistence = fakePersistence(root, logs)
|
|
|
ctx.provide('sessionPersistence', persistence as never)
|
|
|
const fiber = await ctx.plugin(SessionProjectionCache, options.config ?? { writeEveryEvents: 100, writeIntervalMs: 60_000 })
|
|
|
- return { ctx, pool, logs, fiber, persistence, cache: ctx.sessionProjectionCache }
|
|
|
+ return { ctx, root, logs, fiber, persistence, cache: ctx.sessionProjectionCache }
|
|
|
}
|
|
|
|
|
|
const mark = (session: Session, marks: string[]): SessionEvent =>
|
|
|
@@ -102,42 +110,54 @@ const mark = (session: Session, marks: string[]): SessionEvent =>
|
|
|
const endTurn = (session: Session): SessionEvent =>
|
|
|
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
|
|
|
|
|
-/** The stored medium record for one session id (undefined = never written). */
|
|
|
-function storedRecord(pool: MemoryMediaPool, id: Session['id']) {
|
|
|
- return pool.media.get('session_projcache')?.tables.get('sessions')?.get(String(id)) as
|
|
|
- {
|
|
|
- identity: { createdAt: number; cwd?: string }
|
|
|
- rows: Record<string, { ver: number; seq: number; val: unknown }>
|
|
|
- } | undefined
|
|
|
+/** The stored record for one session id (undefined = absent or unreadable). */
|
|
|
+async function storedRecord(root: string, id: Session['id']): Promise<CheckpointRecord | undefined> {
|
|
|
+ try {
|
|
|
+ return checkpointRecord.parse(JSON.parse(await readFile(cachePath(root, id), 'utf8')))
|
|
|
+ } 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
|
|
|
}
|
|
|
|
|
|
-/** The stored medium rows for one session id (undefined = never written). */
|
|
|
-function storedRows(pool: MemoryMediaPool, id: Session['id']) {
|
|
|
- return storedRecord(pool, id)?.rows
|
|
|
+/** Pre-seed one session's cache file with a stored checkpoint record. */
|
|
|
+async function seedRecord(
|
|
|
+ root: string,
|
|
|
+ id: string,
|
|
|
+ rows: CheckpointRecord['rows'],
|
|
|
+ identity: CheckpointRecord['identity'] = { createdAt: 0 },
|
|
|
+): Promise<void> {
|
|
|
+ await mkdir(dirname(cachePath(root, SessionId(id))), { recursive: true })
|
|
|
+ await writeFile(cachePath(root, SessionId(id)), JSON.stringify({ identity, rows }))
|
|
|
}
|
|
|
|
|
|
-/** Wait until queued fail-soft writes (event-listener fire-and-forget) drain. */
|
|
|
-const settle = () => new Promise(resolve => setTimeout(resolve, 0))
|
|
|
+/** Wait until queued fail-soft writes (event-listener fire-and-forget over real fs I/O) drain. */
|
|
|
+const settle = () => new Promise(resolve => setTimeout(resolve, 40))
|
|
|
|
|
|
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 })))
|
|
|
})
|
|
|
|
|
|
describe('SessionProjectionCache write policy', () => {
|
|
|
it('writes a durable checkpoint at turn/end (mandatory point)', async () => {
|
|
|
- const { ctx, pool } = await harness()
|
|
|
+ const { ctx, root } = await harness()
|
|
|
const session = ctx.sessions.create(SessionId('turn-end'))
|
|
|
mark(session, ['a'])
|
|
|
- expect(storedRows(pool, session.id)).toBeUndefined() // throttled: no write yet
|
|
|
+ expect(await storedRows(root, session.id)).toBeUndefined() // throttled: no write yet
|
|
|
const end = endTurn(session)
|
|
|
await settle()
|
|
|
- const rows = storedRows(pool, session.id)
|
|
|
+ const rows = await storedRows(root, session.id)
|
|
|
expect(rows?.['cache-test/marks']).toEqual({ ver: 1, seq: end.seq, val: { marks: ['a'] } })
|
|
|
})
|
|
|
|
|
|
it('writes at session disposal (detach, the live-to-cold moment)', async () => {
|
|
|
- const { ctx, pool } = await harness()
|
|
|
+ const { ctx, root } = await harness()
|
|
|
// Sessions dispose with their owning fiber: create in a child plugin.
|
|
|
let session: Session | undefined
|
|
|
const owner = await ctx.plugin(Object.assign((inner: Context) => {
|
|
|
@@ -147,39 +167,37 @@ describe('SessionProjectionCache write policy', () => {
|
|
|
mark(session, ['live'])
|
|
|
await owner.dispose()
|
|
|
await settle()
|
|
|
- expect(storedRows(pool, session.id)?.['cache-test/marks']?.val).toEqual({ marks: ['live'] })
|
|
|
+ expect((await storedRows(root, session.id))?.['cache-test/marks']?.val).toEqual({ marks: ['live'] })
|
|
|
})
|
|
|
|
|
|
it('flushes when the in-turn event count reaches the configured threshold', async () => {
|
|
|
- const { ctx, pool } = await harness({ config: { writeEveryEvents: 3, writeIntervalMs: 60_000 } })
|
|
|
+ const { ctx, root } = await harness({ config: { writeEveryEvents: 3, writeIntervalMs: 60_000 } })
|
|
|
const session = ctx.sessions.create(SessionId('count'))
|
|
|
mark(session, ['1'])
|
|
|
mark(session, ['2'])
|
|
|
await settle()
|
|
|
- expect(storedRows(pool, session.id)).toBeUndefined()
|
|
|
+ expect(await storedRows(root, session.id)).toBeUndefined()
|
|
|
mark(session, ['3'])
|
|
|
await settle()
|
|
|
- expect(storedRows(pool, session.id)?.['cache-test/marks']?.val).toEqual({ marks: ['3'] })
|
|
|
+ expect((await storedRows(root, session.id))?.['cache-test/marks']?.val).toEqual({ marks: ['3'] })
|
|
|
})
|
|
|
|
|
|
it('flushes on the configured interval when the count threshold is not reached', async () => {
|
|
|
- vi.useFakeTimers()
|
|
|
- const { ctx, pool } = await harness({ config: { writeEveryEvents: 100, writeIntervalMs: 250 } })
|
|
|
+ const { ctx, root } = await harness({ config: { writeEveryEvents: 100, writeIntervalMs: 20 } })
|
|
|
const session = ctx.sessions.create(SessionId('interval'))
|
|
|
mark(session, ['slow'])
|
|
|
- await vi.advanceTimersByTimeAsync(249)
|
|
|
- expect(storedRows(pool, session.id)).toBeUndefined()
|
|
|
- await vi.advanceTimersByTimeAsync(1)
|
|
|
- await vi.advanceTimersByTimeAsync(0)
|
|
|
- expect(storedRows(pool, session.id)?.['cache-test/marks']?.val).toEqual({ marks: ['slow'] })
|
|
|
+ await new Promise(resolve => setTimeout(resolve, 10)) // before the interval
|
|
|
+ expect(await storedRows(root, session.id)).toBeUndefined()
|
|
|
+ await settle() // past the interval; the fire-and-forget write lands
|
|
|
+ expect((await storedRows(root, session.id))?.['cache-test/marks']?.val).toEqual({ marks: ['slow'] })
|
|
|
})
|
|
|
|
|
|
it('write() on a never-dirty session checkpoints directly and rejects a non-JSON unit state', async () => {
|
|
|
- const { ctx, pool } = await harness()
|
|
|
+ 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(storedRows(pool, clean.id)?.['cache-test/marks']).toEqual({ ver: 1, seq: -1, val: null })
|
|
|
+ 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.
|
|
|
ctx.sessionProjections.register({
|
|
|
key: 'cache-test/marks2',
|
|
|
@@ -193,7 +211,7 @@ describe('SessionProjectionCache write policy', () => {
|
|
|
|
|
|
it('plugin disposal clears armed interval timers and leaves cleaned sessions alone', async () => {
|
|
|
vi.useFakeTimers()
|
|
|
- const { ctx, pool, fiber } = await harness({ config: { writeEveryEvents: 100, writeIntervalMs: 5000 } })
|
|
|
+ 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
|
|
|
@@ -203,24 +221,46 @@ describe('SessionProjectionCache write policy', () => {
|
|
|
await fiber.dispose()
|
|
|
// The armed timer died with the plugin: advancing time writes nothing.
|
|
|
await vi.advanceTimersByTimeAsync(10_000)
|
|
|
- expect(storedRows(pool, armed.id)).toBeUndefined()
|
|
|
+ expect(await storedRows(root, armed.id)).toBeUndefined()
|
|
|
})
|
|
|
|
|
|
it('contains a durable write failure: logs a warning, event path unharmed, next write self-heals', async () => {
|
|
|
- const { ctx, pool } = await harness()
|
|
|
+ const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
|
+ roots.push(root)
|
|
|
+ // A file where a directory is needed makes the first write fail...
|
|
|
+ await writeFile(join(root, 'blocked'), '')
|
|
|
+ const logs = new Map<string, SessionEvent[]>()
|
|
|
+ const ctx = new Context()
|
|
|
+ contexts.push(ctx)
|
|
|
+ await ctx.plugin(SessionStore)
|
|
|
+ await ctx.plugin(SessionProjectionRegistry)
|
|
|
+ ctx.sessionProjections.register(marksUnit())
|
|
|
+ let block = true
|
|
|
+ ctx.provide('sessionPersistence', {
|
|
|
+ readFrom: async (id: SessionId, fromSeq: number) => {
|
|
|
+ const events = logs.get(String(id))
|
|
|
+ if (events === undefined) throw new Error(`session "${id}" not found`)
|
|
|
+ return { meta: { version: 0, id, createdAt: 0 }, events: events.filter(event => event.seq >= fromSeq) }
|
|
|
+ },
|
|
|
+ // ...and the locate seam can be un-blocked to let the next write succeed.
|
|
|
+ locate: (meta: SessionHeader) => block
|
|
|
+ ? { kind: 'jsonl', path: join(root, 'blocked', String(meta.id), 'session.jsonl') }
|
|
|
+ : { kind: 'jsonl', path: join(root, String(meta.id), 'session.jsonl') },
|
|
|
+ } as never)
|
|
|
+ await ctx.plugin(SessionProjectionCache, { writeEveryEvents: 100, writeIntervalMs: 60_000 })
|
|
|
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
|
|
|
const session = ctx.sessions.create(SessionId('fail-soft'))
|
|
|
mark(session, ['x'])
|
|
|
- pool.failNextWrites = 1
|
|
|
endTurn(session)
|
|
|
await settle()
|
|
|
- expect(storedRows(pool, session.id)).toBeUndefined()
|
|
|
+ expect(await storedRows(root, session.id)).toBeUndefined()
|
|
|
expect(warn).toHaveBeenCalledWith(expect.stringContaining('turn/end write for "fail-soft" failed'))
|
|
|
// Self-heal: the next mandatory point writes the current cut.
|
|
|
+ block = false
|
|
|
mark(session, ['y'])
|
|
|
endTurn(session)
|
|
|
await settle()
|
|
|
- expect(storedRows(pool, session.id)?.['cache-test/marks']?.val).toEqual({ marks: ['y'] })
|
|
|
+ expect((await storedRows(root, session.id))?.['cache-test/marks']?.val).toEqual({ marks: ['y'] })
|
|
|
})
|
|
|
})
|
|
|
|
|
|
@@ -236,43 +276,31 @@ describe('SessionProjectionCache cold read', () => {
|
|
|
return events
|
|
|
}
|
|
|
|
|
|
- /** Pre-seed the medium with one stored checkpoint record (before the domain opens). */
|
|
|
- function seedRow(
|
|
|
- pool: MemoryMediaPool,
|
|
|
- id: string,
|
|
|
- row: { ver: number; seq: number; val: unknown },
|
|
|
- identity: { createdAt: number; cwd?: string } = { createdAt: 0 },
|
|
|
- ): void {
|
|
|
- pool.versions.set('session_projcache', 3)
|
|
|
- pool.media.set('session_projcache', {
|
|
|
- tables: new Map([['sessions', new Map([[id, { identity, rows: { 'cache-test/marks': row } }]])]]),
|
|
|
- global: null,
|
|
|
- })
|
|
|
- }
|
|
|
-
|
|
|
it('serves a cold session from the cache row plus a bounded tail read, and writes the refresh back', async () => {
|
|
|
- const pool = new MemoryMediaPool()
|
|
|
+ const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
|
+ roots.push(root)
|
|
|
const logs = new Map([['cold', storedLog([['a'], ['a', 'b']])]])
|
|
|
// A warm-era checkpoint at watermark 1 (only ['a'] folded).
|
|
|
- seedRow(pool, 'cold', { ver: 1, seq: 1, val: { marks: ['a'] } })
|
|
|
- const { cache, persistence, pool: samePool } = await harness({ pool, logs })
|
|
|
+ await seedRecord(root, 'cold', { 'cache-test/marks': { ver: 1, seq: 1, val: { marks: ['a'] } } })
|
|
|
+ const { cache, persistence, root: sameRoot } = await harness({ root, logs })
|
|
|
const id = SessionId('cold')
|
|
|
- const snapshot = await cache.coldSnapshot(id)
|
|
|
+ const snapshot = await cache.coldSnapshot(headerOf(id))
|
|
|
expect(snapshot.values['cache-test/marks']).toEqual({ marks: ['a', 'b'] })
|
|
|
expect(snapshot.asOfSeq).toBe(3)
|
|
|
// The tail read was bounded by the anchored floor (watermark 1 -> floor 1), not 0.
|
|
|
expect(persistence.readFrom).toHaveBeenCalledWith(id, 1, undefined)
|
|
|
// Write-back: the stored row advanced to the served cut.
|
|
|
- expect(storedRows(samePool, id)?.['cache-test/marks'])
|
|
|
+ expect((await storedRows(sameRoot, id))?.['cache-test/marks'])
|
|
|
.toEqual({ ver: 1, seq: 3, val: { marks: ['a', 'b'] } })
|
|
|
})
|
|
|
|
|
|
it('discards a version-mismatched row and refolds the full log', async () => {
|
|
|
- const pool = new MemoryMediaPool()
|
|
|
+ const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
|
+ roots.push(root)
|
|
|
const logs = new Map([['bumped', storedLog([['a']])]])
|
|
|
- seedRow(pool, 'bumped', { ver: 1, seq: 2, val: { marks: ['stale'] } })
|
|
|
- const { cache, persistence } = await harness({ pool, logs, stateVersion: 2 })
|
|
|
- const snapshot = await cache.coldSnapshot(SessionId('bumped'))
|
|
|
+ await seedRecord(root, 'bumped', { 'cache-test/marks': { ver: 1, seq: 2, val: { marks: ['stale'] } } })
|
|
|
+ const { cache, persistence } = await harness({ root, logs, stateVersion: 2 })
|
|
|
+ const snapshot = await cache.coldSnapshot(headerOf(SessionId('bumped')))
|
|
|
expect(snapshot.values['cache-test/marks']).toEqual({ marks: ['a'] })
|
|
|
// Mismatch pulls the floor to 0: one full read, no second pass needed.
|
|
|
expect(persistence.readFrom).toHaveBeenCalledTimes(1)
|
|
|
@@ -280,11 +308,12 @@ describe('SessionProjectionCache cold read', () => {
|
|
|
})
|
|
|
|
|
|
it('detects a log shrunk below the row watermark and degrades to one full re-read', async () => {
|
|
|
- const pool = new MemoryMediaPool()
|
|
|
+ const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
|
+ roots.push(root)
|
|
|
const logs = new Map([['shrunk', storedLog([['a']])]]) // seqs 0..2
|
|
|
- seedRow(pool, 'shrunk', { ver: 1, seq: 9, val: { marks: ['ghost'] } })
|
|
|
- const { cache, persistence } = await harness({ pool, logs })
|
|
|
- const snapshot = await cache.coldSnapshot(SessionId('shrunk'))
|
|
|
+ await seedRecord(root, 'shrunk', { 'cache-test/marks': { ver: 1, seq: 9, val: { marks: ['ghost'] } } })
|
|
|
+ const { cache, persistence } = await harness({ root, logs })
|
|
|
+ const snapshot = await cache.coldSnapshot(headerOf(SessionId('shrunk')))
|
|
|
expect(snapshot.values['cache-test/marks']).toEqual({ marks: ['a'] })
|
|
|
expect(snapshot.asOfSeq).toBe(2)
|
|
|
// Anchored tail read (floor 9) came back empty -> full re-read from 0.
|
|
|
@@ -293,12 +322,13 @@ describe('SessionProjectionCache cold read', () => {
|
|
|
})
|
|
|
|
|
|
it('discards malformed persisted state and degrades to one full re-read', async () => {
|
|
|
- const pool = new MemoryMediaPool()
|
|
|
+ const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
|
+ roots.push(root)
|
|
|
const logs = new Map([['malformed', storedLog([['real']])]])
|
|
|
- seedRow(pool, 'malformed', { ver: 1, seq: 1, val: { marks: 'not-an-array' } })
|
|
|
- const { cache, persistence } = await harness({ pool, logs })
|
|
|
+ await seedRecord(root, 'malformed', { 'cache-test/marks': { ver: 1, seq: 1, val: { marks: 'not-an-array' } } })
|
|
|
+ const { cache, persistence } = await harness({ root, logs })
|
|
|
|
|
|
- const snapshot = await cache.coldSnapshot(SessionId('malformed'))
|
|
|
+ const snapshot = await cache.coldSnapshot(headerOf(SessionId('malformed')))
|
|
|
|
|
|
expect(snapshot.values['cache-test/marks']).toEqual({ marks: ['real'] })
|
|
|
expect(persistence.readFrom).toHaveBeenNthCalledWith(1, SessionId('malformed'), 1, undefined)
|
|
|
@@ -306,101 +336,103 @@ describe('SessionProjectionCache cold read', () => {
|
|
|
})
|
|
|
|
|
|
it('write-back failure is contained: the snapshot is still served', async () => {
|
|
|
- const pool = new MemoryMediaPool()
|
|
|
+ const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
|
+ roots.push(root)
|
|
|
const logs = new Map([['soft', storedLog([['a']])]])
|
|
|
- const { ctx, cache } = await harness({ pool, logs })
|
|
|
+ const cacheFile = cachePath(root, SessionId('soft'))
|
|
|
+ await seedRecord(root, 'soft', { 'cache-test/marks': { ver: 1, seq: 0, val: { marks: [] } } })
|
|
|
+ const { ctx, cache } = await harness({ root, logs })
|
|
|
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
|
|
|
- pool.failNextWrites = 1
|
|
|
- const snapshot = await cache.coldSnapshot(SessionId('soft'))
|
|
|
+ // A directory where the cache file must land makes the atomic rename fail;
|
|
|
+ // the served snapshot is unaffected.
|
|
|
+ await rm(cacheFile)
|
|
|
+ await mkdir(cacheFile)
|
|
|
+ const snapshot = await cache.coldSnapshot(headerOf(SessionId('soft')))
|
|
|
expect(snapshot.values['cache-test/marks']).toEqual({ marks: ['a'] })
|
|
|
expect(warn).toHaveBeenCalledWith(expect.stringContaining('cold-read write-back for "soft" failed'))
|
|
|
})
|
|
|
|
|
|
it('rejects for a session with no persisted log', async () => {
|
|
|
const { cache } = await harness()
|
|
|
- await expect(cache.coldSnapshot(SessionId('absent'))).rejects.toThrow('not found')
|
|
|
+ await expect(cache.coldSnapshot(headerOf(SessionId('absent')))).rejects.toThrow('not found')
|
|
|
})
|
|
|
|
|
|
it('discards a record bound to a different log lifecycle and refolds from the actual log', async () => {
|
|
|
- const pool = new MemoryMediaPool()
|
|
|
+ const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
|
+ roots.push(root)
|
|
|
const logs = new Map([['reborn', storedLog([['real']])]]) // stored header stamps createdAt 0
|
|
|
// A checkpoint from a PRIOR lifecycle of the same id (different createdAt):
|
|
|
// its rows pass every watermark check, but the identity does not match.
|
|
|
- seedRow(pool, 'reborn', { ver: 1, seq: 2, val: { marks: ['phantom'] } }, { createdAt: 999 })
|
|
|
- const { cache, pool: samePool } = await harness({ pool, logs })
|
|
|
- const snapshot = await cache.coldSnapshot(SessionId('reborn'))
|
|
|
+ await seedRecord(root, 'reborn', { 'cache-test/marks': { ver: 1, seq: 2, val: { marks: ['phantom'] } } }, { createdAt: 999 })
|
|
|
+ const { cache, root: sameRoot } = await harness({ root, logs })
|
|
|
+ const snapshot = await cache.coldSnapshot(headerOf(SessionId('reborn')))
|
|
|
expect(snapshot.values['cache-test/marks']).toEqual({ marks: ['real'] })
|
|
|
// The write-back rebinds the record to the actual log's identity.
|
|
|
- expect(storedRecord(samePool, SessionId('reborn'))?.identity).toEqual({ createdAt: 0 })
|
|
|
+ expect((await storedRecord(sameRoot, SessionId('reborn')))?.identity).toEqual({ createdAt: 0 })
|
|
|
})
|
|
|
|
|
|
it('cachedSnapshot returns undefined when every stored row is version-mismatched', async () => {
|
|
|
- const pool = new MemoryMediaPool()
|
|
|
- seedRow(pool, 'all-stale', { ver: 99, seq: 4, val: { marks: ['old'] } })
|
|
|
- const { cache } = await harness({ pool })
|
|
|
- expect(cache.cachedSnapshot(headerOf(SessionId('all-stale')))).toBeUndefined()
|
|
|
+ const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
|
+ roots.push(root)
|
|
|
+ await seedRecord(root, 'all-stale', { 'cache-test/marks': { ver: 99, seq: 4, val: { marks: ['old'] } } })
|
|
|
+ const { cache } = await harness({ root })
|
|
|
+ expect(await cache.cachedSnapshot(headerOf(SessionId('all-stale')))).toBeUndefined()
|
|
|
})
|
|
|
|
|
|
it('binds identity on cwd too: a matching cwd serves, a moved session does not', async () => {
|
|
|
- const pool = new MemoryMediaPool()
|
|
|
- seedRow(pool, 'homed', { ver: 1, seq: 2, val: { marks: ['w'] } }, { createdAt: 0, cwd: '/work' })
|
|
|
- const { cache } = await harness({ pool })
|
|
|
+ const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
|
+ roots.push(root)
|
|
|
+ await seedRecord(root, 'homed', { 'cache-test/marks': { ver: 1, seq: 2, val: { marks: ['w'] } } }, { createdAt: 0, cwd: '/work' })
|
|
|
+ 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()
|
|
|
+ expect((await cache.cachedSnapshot(headerOf(id, 0, '/work')))?.values['cache-test/marks']).toEqual({ marks: ['w'] })
|
|
|
+ expect(await cache.cachedSnapshot(headerOf(id, 0, '/elsewhere'))).toBeUndefined()
|
|
|
+ expect(await cache.cachedSnapshot(headerOf(id, 0))).toBeUndefined()
|
|
|
})
|
|
|
|
|
|
it('dates an empty stored log at -1 in the zero-units topology', async () => {
|
|
|
- const pool = new MemoryMediaPool()
|
|
|
+ const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
|
+ roots.push(root)
|
|
|
const logs = new Map([['empty', [] as SessionEvent[]]])
|
|
|
const ctx = new Context()
|
|
|
contexts.push(ctx)
|
|
|
- await ctx.plugin(Storage)
|
|
|
- ctx.storage.backend.register('memory', new MemoryStorageBackend(pool))
|
|
|
- const facility = new DomainFacility(ctx, { backend: 'memory', routes: {} })
|
|
|
- ctx.storage.mount('domain', facility)
|
|
|
- ctx.provide('storageDomain', facility)
|
|
|
await ctx.plugin(SessionStore)
|
|
|
await ctx.plugin(SessionProjectionRegistry)
|
|
|
- ctx.provide('sessionPersistence', fakePersistence(logs) as never)
|
|
|
+ ctx.provide('sessionPersistence', fakePersistence(root, logs) as never)
|
|
|
await ctx.plugin(SessionProjectionCache, { writeEveryEvents: 100, writeIntervalMs: 60_000 })
|
|
|
- await expect(ctx.sessionProjectionCache.coldSnapshot(SessionId('empty')))
|
|
|
+ await expect(ctx.sessionProjectionCache.coldSnapshot(headerOf(SessionId('empty'))))
|
|
|
.resolves.toEqual({ asOfSeq: -1, values: {} })
|
|
|
})
|
|
|
|
|
|
it('cachedSnapshot serves identity-matching rows with the cut watermark and refuses unrelated ones', async () => {
|
|
|
- const pool = new MemoryMediaPool()
|
|
|
- seedRow(pool, 'listed', { ver: 1, seq: 4, val: { marks: ['t'] } })
|
|
|
- const { cache } = await harness({ pool })
|
|
|
+ const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
|
+ roots.push(root)
|
|
|
+ await seedRecord(root, 'listed', { 'cache-test/marks': { ver: 1, seq: 4, val: { marks: ['t'] } } })
|
|
|
+ const { cache } = await harness({ root })
|
|
|
const id = SessionId('listed')
|
|
|
// Matching header: values plus the watermark the client seeds under.
|
|
|
- expect(cache.cachedSnapshot(headerOf(id))).toEqual({ asOfSeq: 4, values: { 'cache-test/marks': { marks: ['t'] } } })
|
|
|
+ expect(await 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()
|
|
|
+ expect(await cache.cachedSnapshot(headerOf(id, 777))).toBeUndefined()
|
|
|
// Unknown id: no block.
|
|
|
- expect(cache.cachedSnapshot(headerOf(SessionId('never-cached')))).toBeUndefined()
|
|
|
+ expect(await cache.cachedSnapshot(headerOf(SessionId('never-cached')))).toBeUndefined()
|
|
|
})
|
|
|
|
|
|
it('holds the not-found contract with zero registered units, and dates the empty cut for a present log', async () => {
|
|
|
// Same composition minus any registered unit: restoreFloor is undefined,
|
|
|
// yet coldSnapshot must still reject for an absent log (probe read) and
|
|
|
// serve an empty cut at the stored end for a present one.
|
|
|
- const pool = new MemoryMediaPool()
|
|
|
+ const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
|
+ roots.push(root)
|
|
|
const logs = new Map([['bare', storedLog([['a']])]]) // seqs 0..2
|
|
|
const ctx = new Context()
|
|
|
contexts.push(ctx)
|
|
|
- await ctx.plugin(Storage)
|
|
|
- ctx.storage.backend.register('memory', new MemoryStorageBackend(pool))
|
|
|
- const facility = new DomainFacility(ctx, { backend: 'memory', routes: {} })
|
|
|
- ctx.storage.mount('domain', facility)
|
|
|
- ctx.provide('storageDomain', facility)
|
|
|
await ctx.plugin(SessionStore)
|
|
|
await ctx.plugin(SessionProjectionRegistry)
|
|
|
- ctx.provide('sessionPersistence', fakePersistence(logs) as never)
|
|
|
+ ctx.provide('sessionPersistence', fakePersistence(root, logs) as never)
|
|
|
await ctx.plugin(SessionProjectionCache, { writeEveryEvents: 100, writeIntervalMs: 60_000 })
|
|
|
- await expect(ctx.sessionProjectionCache.coldSnapshot(SessionId('absent'))).rejects.toThrow('not found')
|
|
|
- await expect(ctx.sessionProjectionCache.coldSnapshot(SessionId('bare')))
|
|
|
+ await expect(ctx.sessionProjectionCache.coldSnapshot(headerOf(SessionId('absent')))).rejects.toThrow('not found')
|
|
|
+ await expect(ctx.sessionProjectionCache.coldSnapshot(headerOf(SessionId('bare'))))
|
|
|
.resolves.toEqual({ asOfSeq: 2, values: {} })
|
|
|
})
|
|
|
})
|
_Kerman