|
|
@@ -2,20 +2,20 @@
|
|
|
* 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 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`).
|
|
|
+ * cached listing read. The durable medium is one `projection_cache.json`
|
|
|
+ * per session under the cache's own configured root
|
|
|
+ * (`<root>/<session-id>/projection_cache.json`); the cache never consults
|
|
|
+ * the persistence layer.
|
|
|
*/
|
|
|
|
|
|
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 { join } from 'node:path'
|
|
|
import { Context } from '@deepseek-ai/cordis'
|
|
|
import { z } from 'zod'
|
|
|
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
|
|
|
-import type { Session, SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
|
|
|
+import type { Session, SessionEvent } from '@deepseek-ai/dsh-session'
|
|
|
import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection'
|
|
|
import type { ProjectionDefinition } from '@deepseek-ai/dsh-session-projection'
|
|
|
import SessionProjectionCache from '../src/index.ts'
|
|
|
@@ -55,27 +55,11 @@ const marksUnit = (stateVersion = 1) => ({
|
|
|
stateVersion,
|
|
|
}) satisfies ProjectionDefinition<'cache-test/marks', MarksState>
|
|
|
|
|
|
-/** One session's cache file inside its persistence directory. */
|
|
|
+/** One session's cache file under the cache's own root. */
|
|
|
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`)
|
|
|
- return {
|
|
|
- meta: { version: 0, id, createdAt: 0 },
|
|
|
- events: events.filter(event => event.seq >= fromSeq),
|
|
|
- }
|
|
|
- })
|
|
|
- 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). */
|
|
|
+/** Header shape for cachedSnapshot calls. */
|
|
|
const headerOf = (id: SessionId, createdAt = 0, cwd?: string) =>
|
|
|
({ version: 0, id, createdAt, ...cwd === undefined ? {} : { cwd } })
|
|
|
|
|
|
@@ -83,7 +67,6 @@ interface HarnessOptions {
|
|
|
root?: string
|
|
|
config?: { writeEveryEvents: number; writeIntervalMs: number }
|
|
|
stateVersion?: number
|
|
|
- logs?: Map<string, SessionEvent[]>
|
|
|
}
|
|
|
|
|
|
const contexts: Context[] = []
|
|
|
@@ -92,16 +75,16 @@ const roots: string[] = []
|
|
|
async function harness(options: HarnessOptions = {}) {
|
|
|
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(SessionStore)
|
|
|
await ctx.plugin(SessionProjectionRegistry)
|
|
|
ctx.sessionProjections.register(marksUnit(options.stateVersion))
|
|
|
- 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, root, logs, fiber, persistence, cache: ctx.sessionProjectionCache }
|
|
|
+ const fiber = await ctx.plugin(SessionProjectionCache, {
|
|
|
+ root,
|
|
|
+ ...options.config ?? { writeEveryEvents: 100, writeIntervalMs: 60_000 },
|
|
|
+ })
|
|
|
+ return { ctx, root, fiber, cache: ctx.sessionProjectionCache }
|
|
|
}
|
|
|
|
|
|
const mark = (session: Session, marks: string[]): SessionEvent =>
|
|
|
@@ -131,7 +114,7 @@ async function seedRecord(
|
|
|
rows: CheckpointRecord['rows'],
|
|
|
identity: CheckpointRecord['identity'] = { createdAt: 0 },
|
|
|
): Promise<void> {
|
|
|
- await mkdir(dirname(cachePath(root, SessionId(id))), { recursive: true })
|
|
|
+ await mkdir(join(root, id), { recursive: true })
|
|
|
await writeFile(cachePath(root, SessionId(id)), JSON.stringify({ identity, rows }))
|
|
|
}
|
|
|
|
|
|
@@ -192,18 +175,6 @@ describe('SessionProjectionCache write policy', () => {
|
|
|
expect((await storedRows(root, session.id))?.['cache-test/marks']?.val).toEqual({ marks: ['slow'] })
|
|
|
})
|
|
|
|
|
|
- it('serializes concurrent checkpoints so they land in call order', async () => {
|
|
|
- const { ctx, root } = await harness()
|
|
|
- const session = ctx.sessions.create(SessionId('chained'))
|
|
|
- const first = ctx.sessionProjectionCache.write(session)
|
|
|
- mark(session, ['a'])
|
|
|
- const second = ctx.sessionProjectionCache.write(session)
|
|
|
- await Promise.all([first, second])
|
|
|
- // Both cuts landed; the file holds the second (newer) cut (the mark is
|
|
|
- // the session's first event, seq 0).
|
|
|
- expect((await storedRows(root, session.id))?.['cache-test/marks']).toEqual({ ver: 1, seq: 0, val: { marks: ['a'] } })
|
|
|
- })
|
|
|
-
|
|
|
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.
|
|
|
@@ -239,36 +210,24 @@ describe('SessionProjectionCache write policy', () => {
|
|
|
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)
|
|
|
- // 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 })
|
|
|
+ await ctx.plugin(SessionProjectionCache, { root, writeEveryEvents: 100, writeIntervalMs: 60_000 })
|
|
|
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
|
|
|
const session = ctx.sessions.create(SessionId('fail-soft'))
|
|
|
+ // A directory where the cache file must land makes the atomic rename
|
|
|
+ // fail on the first write...
|
|
|
+ await mkdir(cachePath(root, session.id), { recursive: true })
|
|
|
mark(session, ['x'])
|
|
|
endTurn(session)
|
|
|
await settle()
|
|
|
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
|
|
|
+ // Self-heal: once the blocker clears, the next mandatory point writes.
|
|
|
+ await rm(cachePath(root, session.id), { recursive: true })
|
|
|
mark(session, ['y'])
|
|
|
endTurn(session)
|
|
|
await settle()
|
|
|
@@ -276,117 +235,22 @@ describe('SessionProjectionCache write policy', () => {
|
|
|
})
|
|
|
})
|
|
|
|
|
|
-describe('SessionProjectionCache cold read', () => {
|
|
|
- const storedLog = (marks: string[][]): SessionEvent[] => {
|
|
|
- const events: SessionEvent[] = [
|
|
|
- { type: 'turn/start', seq: 0, time: 0, data: { turn: 1 } },
|
|
|
- ]
|
|
|
- for (const m of marks) {
|
|
|
- events.push({ type: 'cache-test/mark', seq: events.length, time: events.length, data: { marks: m } })
|
|
|
- }
|
|
|
- events.push({ type: 'turn/end', seq: events.length, time: events.length, data: { turn: 1, reason: { kind: 'completed' } } })
|
|
|
- return events
|
|
|
- }
|
|
|
-
|
|
|
- it('serves a cold session from the cache row plus a bounded tail read, and writes the refresh back', async () => {
|
|
|
- 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).
|
|
|
- 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(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((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 root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
|
- roots.push(root)
|
|
|
- const logs = new Map([['bumped', storedLog([['a']])]])
|
|
|
- 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)
|
|
|
- expect(persistence.readFrom).toHaveBeenCalledWith(SessionId('bumped'), 0, undefined)
|
|
|
- })
|
|
|
-
|
|
|
- it('detects a log shrunk below the row watermark and degrades to one full re-read', async () => {
|
|
|
+describe('SessionProjectionCache listing read', () => {
|
|
|
+ it('serves identity-matching rows with the cut watermark and refuses unrelated ones', async () => {
|
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
|
roots.push(root)
|
|
|
- const logs = new Map([['shrunk', storedLog([['a']])]]) // seqs 0..2
|
|
|
- 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.
|
|
|
- expect(persistence.readFrom).toHaveBeenNthCalledWith(1, SessionId('shrunk'), 9, undefined)
|
|
|
- expect(persistence.readFrom).toHaveBeenNthCalledWith(2, SessionId('shrunk'), 0, undefined)
|
|
|
- })
|
|
|
-
|
|
|
- it('discards malformed persisted state and degrades to one full re-read', async () => {
|
|
|
- const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
|
- roots.push(root)
|
|
|
- const logs = new Map([['malformed', storedLog([['real']])]])
|
|
|
- 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(headerOf(SessionId('malformed')))
|
|
|
-
|
|
|
- expect(snapshot.values['cache-test/marks']).toEqual({ marks: ['real'] })
|
|
|
- expect(persistence.readFrom).toHaveBeenNthCalledWith(1, SessionId('malformed'), 1, undefined)
|
|
|
- expect(persistence.readFrom).toHaveBeenNthCalledWith(2, SessionId('malformed'), 0, undefined)
|
|
|
- })
|
|
|
-
|
|
|
- it('write-back failure is contained: the snapshot is still served', async () => {
|
|
|
- const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
|
- roots.push(root)
|
|
|
- const logs = new Map([['soft', storedLog([['a']])]])
|
|
|
- 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(() => {})
|
|
|
- // 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(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 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.
|
|
|
- await seedRecord(root, 'reborn', { 'cache-test/marks': { ver: 1, seq: 2, val: { marks: ['phantom'] } } }, { createdAt: 999 })
|
|
|
- const { cache, root: sameRoot } = await harness({ root, logs })
|
|
|
- // The caller holds the STALE header (the old lifecycle's createdAt): the
|
|
|
- // stale record passes the read-side filter, but the stored log's header
|
|
|
- // is the identity witness and rebinds the write-back to the real lifecycle.
|
|
|
- const snapshot = await cache.coldSnapshot(headerOf(SessionId('reborn'), 999))
|
|
|
- expect(snapshot.values['cache-test/marks']).toEqual({ marks: ['real'] })
|
|
|
- // The write-back rebinds the record to the actual log's identity.
|
|
|
- expect((await storedRecord(sameRoot, SessionId('reborn')))?.identity).toEqual({ createdAt: 0 })
|
|
|
+ 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(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(await cache.cachedSnapshot(headerOf(id, 777))).toBeUndefined()
|
|
|
+ // Unknown id: no block.
|
|
|
+ expect(await cache.cachedSnapshot(headerOf(SessionId('never-cached')))).toBeUndefined()
|
|
|
})
|
|
|
|
|
|
- it('cachedSnapshot returns undefined when every stored row is version-mismatched', async () => {
|
|
|
+ it('returns undefined when every stored row is version-mismatched', async () => {
|
|
|
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'] } } })
|
|
|
@@ -405,99 +269,12 @@ describe('SessionProjectionCache cold read', () => {
|
|
|
expect(await cache.cachedSnapshot(headerOf(id, 0))).toBeUndefined()
|
|
|
})
|
|
|
|
|
|
- it('dates an empty stored log at -1 in the zero-units topology', async () => {
|
|
|
+ it('returns undefined for a malformed cache file (refold from the log on the caller side)', async () => {
|
|
|
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(SessionStore)
|
|
|
- await ctx.plugin(SessionProjectionRegistry)
|
|
|
- ctx.provide('sessionPersistence', fakePersistence(root, logs) as never)
|
|
|
- await ctx.plugin(SessionProjectionCache, { writeEveryEvents: 100, writeIntervalMs: 60_000 })
|
|
|
- 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 root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
|
- roots.push(root)
|
|
|
- await seedRecord(root, 'listed', { 'cache-test/marks': { ver: 1, seq: 4, val: { marks: ['t'] } } })
|
|
|
+ await mkdir(join(root, 'malformed'), { recursive: true })
|
|
|
+ await writeFile(cachePath(root, SessionId('malformed')), 'not json at all')
|
|
|
const { cache } = await harness({ root })
|
|
|
- const id = SessionId('listed')
|
|
|
- // Matching header: values plus the watermark the client seeds under.
|
|
|
- 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(await cache.cachedSnapshot(headerOf(id, 777))).toBeUndefined()
|
|
|
- // Unknown id: no block.
|
|
|
- expect(await cache.cachedSnapshot(headerOf(SessionId('never-cached')))).toBeUndefined()
|
|
|
- })
|
|
|
-
|
|
|
- describe('SessionProjectionCache without a per-session directory (sqlite-style backend)', () => {
|
|
|
- it('write() no-ops and cachedSnapshot is undefined — no cache file is ever created', async () => {
|
|
|
- const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
|
- roots.push(root)
|
|
|
- const ctx = new Context()
|
|
|
- contexts.push(ctx)
|
|
|
- await ctx.plugin(SessionStore)
|
|
|
- await ctx.plugin(SessionProjectionRegistry)
|
|
|
- ctx.sessionProjections.register(marksUnit())
|
|
|
- ctx.provide('sessionPersistence', {
|
|
|
- readFrom: async () => { throw new Error('no log') },
|
|
|
- locate: () => undefined,
|
|
|
- } as never)
|
|
|
- await ctx.plugin(SessionProjectionCache, { writeEveryEvents: 100, writeIntervalMs: 60_000 })
|
|
|
- const session = ctx.sessions.create(SessionId('no-path'))
|
|
|
- mark(session, ['a'])
|
|
|
- endTurn(session)
|
|
|
- await settle()
|
|
|
- expect(await storedRows(root, session.id)).toBeUndefined()
|
|
|
- expect(await ctx.sessionProjectionCache.cachedSnapshot(headerOf(session.id))).toBeUndefined()
|
|
|
- // An explicit write is also a no-op: no checkpoint cut, no durability flush.
|
|
|
- await expect(ctx.sessionProjectionCache.write(session)).resolves.toBeUndefined()
|
|
|
- expect(await storedRows(root, session.id)).toBeUndefined()
|
|
|
- })
|
|
|
-
|
|
|
- it('coldSnapshot falls to the full-log rung when no cache path exists', async () => {
|
|
|
- const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-'))
|
|
|
- roots.push(root)
|
|
|
- const logs = new Map([['nopath', storedLog([['a']])]])
|
|
|
- const ctx = new Context()
|
|
|
- contexts.push(ctx)
|
|
|
- await ctx.plugin(SessionStore)
|
|
|
- await ctx.plugin(SessionProjectionRegistry)
|
|
|
- ctx.sessionProjections.register(marksUnit())
|
|
|
- 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`)
|
|
|
- return { meta: { version: 0, id, createdAt: 0 }, events: events.filter(event => event.seq >= fromSeq) }
|
|
|
- })
|
|
|
- ctx.provide('sessionPersistence', { readFrom, locate: () => undefined } as never)
|
|
|
- await ctx.plugin(SessionProjectionCache, { writeEveryEvents: 100, writeIntervalMs: 60_000 })
|
|
|
- const snapshot = await ctx.sessionProjectionCache.coldSnapshot(headerOf(SessionId('nopath')))
|
|
|
- expect(snapshot.values['cache-test/marks']).toEqual({ marks: ['a'] })
|
|
|
- // No cached rows: the restore floor is undefined, so the ladder is one
|
|
|
- // full probe read from seq 0 and no write-back happens.
|
|
|
- expect(readFrom).toHaveBeenCalledWith(SessionId('nopath'), 0, undefined)
|
|
|
- expect(await storedRows(root, SessionId('nopath'))).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 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(SessionStore)
|
|
|
- await ctx.plugin(SessionProjectionRegistry)
|
|
|
- ctx.provide('sessionPersistence', fakePersistence(root, logs) as never)
|
|
|
- await ctx.plugin(SessionProjectionCache, { writeEveryEvents: 100, writeIntervalMs: 60_000 })
|
|
|
- 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: {} })
|
|
|
+ expect(await cache.cachedSnapshot(headerOf(SessionId('malformed')))).toBeUndefined()
|
|
|
})
|
|
|
})
|
_Kerman