David Snelling
d1651f986c
The generic reprojection engine (pure TS; the twin of the native
implementation — same frozen contract, one shared conformance intent):
register any ProjectionAdapter; advance(family, {budgetMs}) folds facts
from the adapter's own watermark to the head in installments ≤50ms with
real macrotask yields; foreground door traffic bumps the DoorSignal and
an in-flight advance yields within one installment ('preempted');
advanceAll round-robins families fairly. swap(family, buildAdapter) is
the doors-open migration primitive: the OLD projection keeps serving
while the new one builds beside it, the flip is atomic at parity, and a
concurrent second swap refuses typed. A fact the fold cannot apply
(typed ProjectionApplyError) is QUARANTINED — skipped, ledgered,
narrated per-doubling, exposed for refuse-affected-reads — the service
class law's fourth answer: never a wedged rebuild, never a silent skip.
The engine never writes stamps: each adapter owns its durability and its
stamp-after-data discipline. Upgrade, heal, and rebuild are now the same
machinery behind open doors.
FactLogSource wires any host's fact scan in one line
(factSourceFromHost(brain)); window-contract violations are loud.
Pins: 23 unit (budget resume without refold · preemption within one
installment · round-robin fairness under a skewed backlog · build-beside
visibility mid-swap · atomic flip · single-flight refusal · quarantine
skip/ledger/doubling · non-typed throw aborts · losing adapter
discarded) + 3 integration on a real brain (fold matches ground truth ·
doors answer mid-fold with the preemption path exercised · crash
mid-fold resumes from the stamp, never refolds).
Gates: unit 2054/2054 (157 files) · integration 820 (93 files) ·
conformance 31/31.
590 lines
24 KiB
TypeScript
590 lines
24 KiB
TypeScript
/**
|
||
* @module tests/unit/reprojection/reprojection-engine
|
||
* @description Spec-by-example for the pure-TS reprojection engine — the
|
||
* frozen contract mirrored from the native twin (a shared conformance suite
|
||
* runs against both, so the shapes pinned here are load-bearing):
|
||
*
|
||
* (a) register + advance folds a scripted source to caught-up with exact
|
||
* watermark/applied counts and adapter-owned stamping;
|
||
* (b) budget exhaustion answers mid-stream and a second advance RESUMES from
|
||
* the watermark — never a refold;
|
||
* (c) a door bump mid-advance preempts within one installment — pinned by
|
||
* MECHANISM (no further applyBatch after the bumping step), with only a
|
||
* generous wall-clock sanity bound;
|
||
* (d) advanceAll round-robins families at batch granularity — no starvation;
|
||
* (e) swap builds beside (the old adapter serves throughout), flips
|
||
* atomically at parity, refuses a concurrent swap with a typed error;
|
||
* (f) quarantine: a typed poison fact is skipped + ledgered, narration
|
||
* doubles, a NON-typed throw aborts loudly;
|
||
* (g) discard() lands on the LOSING adapter after a swap.
|
||
*/
|
||
import { describe, it, expect, vi, afterEach } from 'vitest'
|
||
import {
|
||
ReprojectionEngine,
|
||
DoorSignal,
|
||
ProjectionApplyError,
|
||
SwapInFlightError,
|
||
MAX_INSTALLMENT_MS,
|
||
type ProjectionAdapter,
|
||
type FactSource
|
||
} from '../../../src/reprojection/reprojectionEngine.js'
|
||
import { FactLogSource } from '../../../src/reprojection/factLogSource.js'
|
||
import type { CommitFact } from '../../../src/db/factLog.js'
|
||
import { prodLog } from '../../../src/utils/logger.js'
|
||
|
||
/** Build one committed fact for a generation. */
|
||
function fact(generation: number): CommitFact {
|
||
return {
|
||
generation,
|
||
timestamp: 1_700_000_000_000 + generation,
|
||
ops: [
|
||
{
|
||
kind: 'noun',
|
||
id: `id-${generation}`,
|
||
record: { metadata: { n: generation }, vector: null }
|
||
}
|
||
]
|
||
}
|
||
}
|
||
|
||
/** A scripted FactSource over a (possibly mutable) list of generations. */
|
||
function scriptedSource(gens: () => number[]): FactSource {
|
||
return {
|
||
async scan(from: number, limit: number): Promise<CommitFact[]> {
|
||
return gens()
|
||
.filter((g) => g > from)
|
||
.sort((x, y) => x - y)
|
||
.slice(0, limit)
|
||
.map(fact)
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* A recording in-memory adapter: stamps after data (the watermark advances
|
||
* only after a successful apply), applies idempotently (a Map keyed by
|
||
* generation), and can be scripted to poison (typed) or hard-fail (untyped)
|
||
* specific generations, or to run a hook inside applyBatch.
|
||
*/
|
||
class RecordingAdapter implements ProjectionAdapter {
|
||
readonly family: string
|
||
/** Generations per applyBatch call, in call order (empty arrays included). */
|
||
readonly batches: number[][] = []
|
||
/** The upTo passed to each applyBatch call, in call order. */
|
||
readonly upTos: number[] = []
|
||
/** Latest state per generation — idempotent under at-least-once delivery. */
|
||
readonly state = new Map<number, unknown>()
|
||
/** Generations that throw a typed ProjectionApplyError. */
|
||
readonly poison = new Set<number>()
|
||
/** Generations that throw a plain (untyped) Error. */
|
||
readonly hardFail = new Set<number>()
|
||
/** Runs inside applyBatch after validation, before the stamp. */
|
||
onApply?: (gens: number[]) => void | Promise<void>
|
||
discarded = 0
|
||
private wm: number | null
|
||
|
||
constructor(family: string, watermark: number | null = null) {
|
||
this.family = family
|
||
this.wm = watermark
|
||
}
|
||
|
||
watermark(): number | null {
|
||
return this.wm
|
||
}
|
||
|
||
async applyBatch(facts: CommitFact[], upTo: number): Promise<void> {
|
||
for (const [i, f] of facts.entries()) {
|
||
if (this.hardFail.has(f.generation)) {
|
||
throw new Error(`disk exploded at generation ${f.generation}`)
|
||
}
|
||
if (this.poison.has(f.generation)) {
|
||
throw new ProjectionApplyError({
|
||
generation: f.generation,
|
||
recordIndex: i,
|
||
cause: new Error(`unfoldable payload at ${f.generation}`)
|
||
})
|
||
}
|
||
}
|
||
for (const f of facts) this.state.set(f.generation, f.ops)
|
||
const gens = facts.map((f) => f.generation)
|
||
this.batches.push(gens)
|
||
this.upTos.push(upTo)
|
||
if (this.onApply) await this.onApply(gens)
|
||
this.wm = upTo // stamp-after-data
|
||
}
|
||
|
||
async discard(): Promise<void> {
|
||
this.discarded++
|
||
}
|
||
}
|
||
|
||
const range = (from: number, to: number): number[] =>
|
||
Array.from({ length: to - from + 1 }, (_, i) => from + i)
|
||
|
||
afterEach(() => {
|
||
vi.restoreAllMocks()
|
||
})
|
||
|
||
describe('reprojection engine — (a) register + advance to caught-up', () => {
|
||
it('folds a scripted source in order, adapter-stamped, with exact counts', async () => {
|
||
const source = scriptedSource(() => range(1, 7))
|
||
const engine = new ReprojectionEngine({ source, batchSize: 3 })
|
||
const adapter = new RecordingAdapter('a')
|
||
engine.register(adapter)
|
||
|
||
const result = await engine.advance('a', { budgetMs: 10_000 })
|
||
|
||
expect(result.status).toBe('caught-up')
|
||
expect(result.watermark).toBe(7)
|
||
expect(result.applied).toBe(7)
|
||
// Batch shape and the upTo handed to the adapter's own stamp.
|
||
expect(adapter.batches).toEqual([[1, 2, 3], [4, 5, 6], [7]])
|
||
expect(adapter.upTos).toEqual([3, 6, 7])
|
||
// The watermark is the ADAPTER's stamp — the engine never wrote one.
|
||
expect(adapter.watermark()).toBe(7)
|
||
expect(engine.getAdapter('a')).toBe(adapter)
|
||
})
|
||
|
||
it('honors upTo as an inclusive cap and answers caught-up at the cap', async () => {
|
||
const source = scriptedSource(() => range(1, 9))
|
||
const engine = new ReprojectionEngine({ source, batchSize: 3 })
|
||
const adapter = new RecordingAdapter('a')
|
||
engine.register(adapter)
|
||
|
||
const result = await engine.advance('a', { budgetMs: 10_000, upTo: 5 })
|
||
|
||
expect(result.status).toBe('caught-up')
|
||
expect(result.watermark).toBe(5)
|
||
expect(result.applied).toBe(5)
|
||
expect(adapter.batches.flat()).toEqual([1, 2, 3, 4, 5])
|
||
})
|
||
|
||
it('a caught-up family answers immediately with zero applied', async () => {
|
||
const source = scriptedSource(() => range(1, 4))
|
||
const engine = new ReprojectionEngine({ source, batchSize: 10 })
|
||
const adapter = new RecordingAdapter('a', 4) // already stamped to the head
|
||
engine.register(adapter)
|
||
|
||
const result = await engine.advance('a', { budgetMs: 10_000 })
|
||
|
||
expect(result).toEqual({ status: 'caught-up', watermark: 4, applied: 0 })
|
||
expect(adapter.batches).toEqual([])
|
||
})
|
||
|
||
it('refuses duplicate registration and unregistered families loudly', async () => {
|
||
const engine = new ReprojectionEngine({ source: scriptedSource(() => []) })
|
||
engine.register(new RecordingAdapter('a'))
|
||
expect(() => engine.register(new RecordingAdapter('a'))).toThrow(/already registered/)
|
||
await expect(engine.advance('ghost', { budgetMs: 0 })).rejects.toThrow(/not registered/)
|
||
})
|
||
})
|
||
|
||
describe('reprojection engine — (b) budget exhaustion resumes, never refolds', () => {
|
||
it('returns budget-exhausted mid-stream; the next advance resumes from the watermark', async () => {
|
||
const source = scriptedSource(() => range(1, 10))
|
||
const engine = new ReprojectionEngine({ source, batchSize: 2 })
|
||
const adapter = new RecordingAdapter('b')
|
||
engine.register(adapter)
|
||
|
||
// Zero budget: exactly ONE step of guaranteed progress, then the answer.
|
||
const first = await engine.advance('b', { budgetMs: 0 })
|
||
expect(first.status).toBe('budget-exhausted')
|
||
expect(first.watermark).toBe(2)
|
||
expect(first.applied).toBe(2)
|
||
expect(adapter.batches).toEqual([[1, 2]])
|
||
|
||
// The second advance RESUMES from the stamp — its first batch starts at 3.
|
||
const second = await engine.advance('b', { budgetMs: 10_000 })
|
||
expect(second.status).toBe('caught-up')
|
||
expect(second.watermark).toBe(10)
|
||
expect(second.applied).toBe(8)
|
||
expect(adapter.batches[1]).toEqual([3, 4])
|
||
// No refold: every generation delivered exactly once across both calls.
|
||
expect(adapter.batches.flat()).toEqual(range(1, 10))
|
||
})
|
||
})
|
||
|
||
describe('reprojection engine — (c) door bump preempts within one installment', () => {
|
||
it('a bump during a step yields preempted at that step boundary — no further applyBatch', async () => {
|
||
const source = scriptedSource(() => range(1, 12))
|
||
const engine = new ReprojectionEngine({ source, batchSize: 2 })
|
||
const adapter = new RecordingAdapter('c')
|
||
adapter.onApply = (gens) => {
|
||
if (gens[0] === 3) engine.doorSignal.bump() // door traffic mid-second-batch
|
||
}
|
||
engine.register(adapter)
|
||
|
||
const started = Date.now()
|
||
const result = await engine.advance('c', { budgetMs: 60_000 })
|
||
const elapsed = Date.now() - started
|
||
|
||
expect(result.status).toBe('preempted')
|
||
expect(result.watermark).toBe(4)
|
||
expect(result.applied).toBe(4)
|
||
// THE MECHANISM PIN: the batch that observed the bump was the LAST batch —
|
||
// preemption landed at the very next boundary, not after more work.
|
||
expect(adapter.batches).toEqual([[1, 2], [3, 4]])
|
||
// Generous wall-clock sanity only (the pin above carries the contract):
|
||
// two tiny batches plus one installment boundary sit far under 5s.
|
||
expect(elapsed).toBeLessThan(5_000)
|
||
expect(MAX_INSTALLMENT_MS).toBe(50)
|
||
|
||
// Resuming folds the rest — preemption lost nothing.
|
||
const resumed = await engine.advance('c', { budgetMs: 60_000 })
|
||
expect(resumed.status).toBe('caught-up')
|
||
expect(resumed.watermark).toBe(12)
|
||
expect(adapter.batches.flat()).toEqual(range(1, 12))
|
||
})
|
||
|
||
it('bumps are edge-triggered per advance: a stale bump never preempts', async () => {
|
||
const source = scriptedSource(() => range(1, 4))
|
||
const doorSignal = new DoorSignal()
|
||
const engine = new ReprojectionEngine({ source, doorSignal, batchSize: 2 })
|
||
const adapter = new RecordingAdapter('c2')
|
||
engine.register(adapter)
|
||
|
||
doorSignal.bump() // BEFORE the advance — belongs to earlier traffic
|
||
const result = await engine.advance('c2', { budgetMs: 10_000 })
|
||
expect(result.status).toBe('caught-up')
|
||
expect(result.watermark).toBe(4)
|
||
})
|
||
})
|
||
|
||
describe('reprojection engine — (d) advanceAll round-robin fairness', () => {
|
||
it('a one-batch family is served on the first round despite a huge backlog next to it', async () => {
|
||
const source = scriptedSource(() => range(1, 40))
|
||
const engine = new ReprojectionEngine({ source, batchSize: 5 })
|
||
const callOrder: string[] = []
|
||
const big = new RecordingAdapter('big') // 8 batches behind
|
||
const small = new RecordingAdapter('small', 35) // 1 batch behind
|
||
big.onApply = () => {
|
||
callOrder.push('big')
|
||
}
|
||
small.onApply = () => {
|
||
callOrder.push('small')
|
||
}
|
||
engine.register(big)
|
||
engine.register(small)
|
||
|
||
const results = await engine.advanceAll({ budgetMs: 10_000 })
|
||
|
||
expect(results.big).toEqual({ status: 'caught-up', watermark: 40, applied: 40 })
|
||
expect(results.small).toEqual({ status: 'caught-up', watermark: 40, applied: 5 })
|
||
// Fairness pin: 'small' folded its single batch on round ONE — it never
|
||
// waited behind 'big''s backlog.
|
||
expect(callOrder[1]).toBe('small')
|
||
expect(callOrder.filter((f) => f === 'small')).toHaveLength(1)
|
||
})
|
||
|
||
it('two full-backlog families interleave strictly, one batch each per round', async () => {
|
||
const source = scriptedSource(() => range(1, 40))
|
||
const engine = new ReprojectionEngine({ source, batchSize: 5 })
|
||
const callOrder: string[] = []
|
||
const first = new RecordingAdapter('first')
|
||
const second = new RecordingAdapter('second')
|
||
first.onApply = () => {
|
||
callOrder.push('first')
|
||
}
|
||
second.onApply = () => {
|
||
callOrder.push('second')
|
||
}
|
||
engine.register(first)
|
||
engine.register(second)
|
||
|
||
const results = await engine.advanceAll({ budgetMs: 10_000 })
|
||
|
||
expect(results.first.status).toBe('caught-up')
|
||
expect(results.second.status).toBe('caught-up')
|
||
// 8 rounds × (first, second): strict alternation — neither ever ran twice
|
||
// while the other waited.
|
||
expect(callOrder).toHaveLength(16)
|
||
for (let i = 0; i < callOrder.length; i += 2) {
|
||
expect(callOrder.slice(i, i + 2)).toEqual(['first', 'second'])
|
||
}
|
||
})
|
||
|
||
it('budget exhaustion mid-round reports every unfinished family at its own watermark', async () => {
|
||
const source = scriptedSource(() => range(1, 40))
|
||
const engine = new ReprojectionEngine({ source, batchSize: 5 })
|
||
const a = new RecordingAdapter('a')
|
||
const b = new RecordingAdapter('b')
|
||
engine.register(a)
|
||
engine.register(b)
|
||
|
||
const results = await engine.advanceAll({ budgetMs: 0 })
|
||
|
||
// Zero budget: the leading family gets its one guaranteed step, then the
|
||
// budget answer lands for everyone still mid-stream.
|
||
expect(results.a.status).toBe('budget-exhausted')
|
||
expect(results.b.status).toBe('budget-exhausted')
|
||
expect(results.a.applied + results.b.applied).toBeGreaterThanOrEqual(5)
|
||
// A later advanceAll resumes both to the head.
|
||
const finished = await engine.advanceAll({ budgetMs: 10_000 })
|
||
expect(finished.a.status).toBe('caught-up')
|
||
expect(finished.b.status).toBe('caught-up')
|
||
expect(a.batches.flat()).toEqual(range(1, 40))
|
||
expect(b.batches.flat()).toEqual(range(1, 40))
|
||
})
|
||
})
|
||
|
||
describe('reprojection engine — (e) swap: build-beside, atomic flip, single-flight', () => {
|
||
it('the old adapter serves at its own watermark throughout the build; the flip is atomic at parity', async () => {
|
||
const log = range(1, 20)
|
||
const source = scriptedSource(() => log)
|
||
const engine = new ReprojectionEngine({ source, batchSize: 4 })
|
||
const oldAdapter = new RecordingAdapter('e')
|
||
engine.register(oldAdapter)
|
||
await engine.advance('e', { budgetMs: 10_000 })
|
||
expect(oldAdapter.watermark()).toBe(20)
|
||
|
||
// The log grows after the old adapter stamped — the build must reach the
|
||
// HEAD (24), not merely the old watermark (20), before the flip.
|
||
log.push(21, 22, 23, 24)
|
||
|
||
const servingDuringBuild: Array<{ adapter: ProjectionAdapter | undefined; watermark: number | null }> = []
|
||
let replacement!: RecordingAdapter
|
||
const result = await engine.swap('e', async () => {
|
||
replacement = new RecordingAdapter('e')
|
||
replacement.onApply = () => {
|
||
servingDuringBuild.push({
|
||
adapter: engine.getAdapter('e'),
|
||
watermark: engine.getAdapter('e')!.watermark()
|
||
})
|
||
}
|
||
return replacement
|
||
})
|
||
|
||
// Build-beside pin: EVERY mid-build observation saw the OLD adapter,
|
||
// still serving, still at its own stamp.
|
||
expect(servingDuringBuild.length).toBeGreaterThan(0)
|
||
for (const seen of servingDuringBuild) {
|
||
expect(seen.adapter).toBe(oldAdapter)
|
||
expect(seen.watermark).toBe(20)
|
||
}
|
||
// The flip: the registry now serves the replacement, at parity with head.
|
||
expect(engine.getAdapter('e')).toBe(replacement)
|
||
expect(result.watermark).toBe(24)
|
||
expect(result.applied).toBe(24)
|
||
expect(replacement.batches.flat()).toEqual(range(1, 24))
|
||
})
|
||
|
||
it('a second concurrent swap on the same family refuses with the typed single-flight error', async () => {
|
||
const source = scriptedSource(() => range(1, 8))
|
||
const engine = new ReprojectionEngine({ source, batchSize: 4 })
|
||
engine.register(new RecordingAdapter('e2'))
|
||
|
||
let release!: () => void
|
||
const gate = new Promise<void>((resolve) => {
|
||
release = resolve
|
||
})
|
||
const inFlight = engine.swap('e2', async () => {
|
||
const building = new RecordingAdapter('e2')
|
||
building.onApply = () => gate // the build parks mid-fold
|
||
return building
|
||
})
|
||
|
||
// While the first swap builds, a second one is refused — typed.
|
||
const refusal = await engine.swap('e2', async () => new RecordingAdapter('e2')).catch((e) => e)
|
||
expect(refusal).toBeInstanceOf(SwapInFlightError)
|
||
expect((refusal as SwapInFlightError).family).toBe('e2')
|
||
|
||
release()
|
||
const done = await inFlight
|
||
expect(done.watermark).toBe(8)
|
||
// Single-flight released: a follow-up swap is admitted again.
|
||
const again = await engine.swap('e2', async () => new RecordingAdapter('e2'))
|
||
expect(again.watermark).toBe(8)
|
||
})
|
||
|
||
it('a failed build discards the partial replacement and leaves the old adapter serving', async () => {
|
||
const source = scriptedSource(() => range(1, 8))
|
||
const engine = new ReprojectionEngine({ source, batchSize: 4 })
|
||
const oldAdapter = new RecordingAdapter('e3')
|
||
engine.register(oldAdapter)
|
||
await engine.advance('e3', { budgetMs: 10_000 })
|
||
|
||
let failed!: RecordingAdapter
|
||
await expect(
|
||
engine.swap('e3', async () => {
|
||
failed = new RecordingAdapter('e3')
|
||
failed.hardFail.add(5) // an UNTYPED failure mid-build
|
||
return failed
|
||
})
|
||
).rejects.toThrow(/disk exploded/)
|
||
|
||
expect(failed.discarded).toBe(1) // the partial build was cleaned up
|
||
expect(oldAdapter.discarded).toBe(0)
|
||
expect(engine.getAdapter('e3')).toBe(oldAdapter) // still serving, untouched
|
||
expect(oldAdapter.watermark()).toBe(8)
|
||
})
|
||
})
|
||
|
||
describe('reprojection engine — (f) quarantine: the fourth answer class', () => {
|
||
it('a typed poison fact is skipped, ledgered, and the rest folds to quarantined', async () => {
|
||
const source = scriptedSource(() => range(1, 10))
|
||
const engine = new ReprojectionEngine({ source, batchSize: 4 })
|
||
const adapter = new RecordingAdapter('f')
|
||
adapter.poison.add(6)
|
||
engine.register(adapter)
|
||
|
||
const result = await engine.advance('f', { budgetMs: 10_000 })
|
||
|
||
expect(result.status).toBe('quarantined')
|
||
expect(result.watermark).toBe(10)
|
||
expect(result.applied).toBe(9) // every generation but the poison
|
||
expect(adapter.batches.flat().sort((x, y) => x - y)).toEqual([1, 2, 3, 4, 5, 7, 8, 9, 10])
|
||
expect(adapter.state.has(6)).toBe(false)
|
||
|
||
const ledger = engine.quarantined('f')
|
||
expect(ledger).toHaveLength(1)
|
||
expect(ledger[0].generation).toBe(6)
|
||
expect(ledger[0].error).toBeInstanceOf(ProjectionApplyError)
|
||
expect(ledger[0].error.recordIndex).toBe(1) // 6 sat at index 1 of [5..8]
|
||
expect(typeof ledger[0].at).toBe('number')
|
||
})
|
||
|
||
it('narration doubles: warns on the 1st, 2nd, and 4th quarantine — not the 3rd', async () => {
|
||
const warnSpy = vi.spyOn(prodLog, 'warn').mockImplementation(() => {})
|
||
const source = scriptedSource(() => range(1, 10))
|
||
const engine = new ReprojectionEngine({ source, batchSize: 10 })
|
||
const adapter = new RecordingAdapter('f2')
|
||
for (const g of [2, 4, 6, 8]) adapter.poison.add(g)
|
||
engine.register(adapter)
|
||
|
||
const result = await engine.advance('f2', { budgetMs: 10_000 })
|
||
|
||
expect(result.status).toBe('quarantined')
|
||
expect(result.watermark).toBe(10)
|
||
expect(result.applied).toBe(6)
|
||
expect(engine.quarantined('f2').map((q) => q.generation)).toEqual([2, 4, 6, 8])
|
||
const quarantineWarns = warnSpy.mock.calls.filter((c) => String(c[0]).includes('quarantined generation'))
|
||
// 4 entries, narrated at counts 1, 2, and 4 — the 3rd stayed quiet.
|
||
expect(quarantineWarns).toHaveLength(3)
|
||
expect(quarantineWarns.map((c) => String(c[0]))).toEqual([
|
||
expect.stringContaining('(1 quarantined total)'),
|
||
expect.stringContaining('(2 quarantined total)'),
|
||
expect.stringContaining('(4 quarantined total)')
|
||
])
|
||
})
|
||
|
||
it('an all-poison window still advances the stamp via an empty applyBatch', async () => {
|
||
const source = scriptedSource(() => range(1, 3))
|
||
const engine = new ReprojectionEngine({ source, batchSize: 3 })
|
||
const adapter = new RecordingAdapter('f3')
|
||
for (const g of [1, 2, 3]) adapter.poison.add(g)
|
||
engine.register(adapter)
|
||
|
||
const result = await engine.advance('f3', { budgetMs: 10_000 })
|
||
|
||
expect(result.status).toBe('quarantined')
|
||
expect(result.watermark).toBe(3)
|
||
expect(result.applied).toBe(0)
|
||
// The final call carried NO facts but a real upTo — the pure watermark
|
||
// advance past poison, stamped by the adapter itself.
|
||
expect(adapter.batches).toEqual([[]])
|
||
expect(adapter.upTos).toEqual([3])
|
||
expect(engine.quarantined('f3').map((q) => q.generation)).toEqual([1, 2, 3])
|
||
})
|
||
|
||
it('a NON-typed throw aborts the advance loudly — unknown failure is never poison', async () => {
|
||
const source = scriptedSource(() => range(1, 8))
|
||
const engine = new ReprojectionEngine({ source, batchSize: 4 })
|
||
const adapter = new RecordingAdapter('f4')
|
||
adapter.hardFail.add(5)
|
||
engine.register(adapter)
|
||
|
||
await expect(engine.advance('f4', { budgetMs: 10_000 })).rejects.toThrow(/disk exploded at generation 5/)
|
||
|
||
expect(adapter.watermark()).toBe(4) // the clean first batch landed; nothing after
|
||
expect(engine.quarantined('f4')).toEqual([]) // no ledger entry for an unknown failure
|
||
})
|
||
|
||
it('an adapter re-condemning an already-quarantined generation is refused loudly', async () => {
|
||
const source = scriptedSource(() => range(1, 4))
|
||
const engine = new ReprojectionEngine({ source, batchSize: 4 })
|
||
// A misbehaving adapter: always blames generation 3, even once it is
|
||
// filtered out of its batches.
|
||
const adapter: ProjectionAdapter = {
|
||
family: 'f5',
|
||
watermark: () => null,
|
||
applyBatch: async () => {
|
||
throw new ProjectionApplyError({ generation: 3, cause: new Error('always 3') })
|
||
},
|
||
discard: async () => {}
|
||
}
|
||
engine.register(adapter)
|
||
|
||
await expect(engine.advance('f5', { budgetMs: 10_000 })).rejects.toThrow(/ALREADY quarantined/)
|
||
expect(engine.quarantined('f5').map((q) => q.generation)).toEqual([3])
|
||
})
|
||
|
||
it('an adapter that never stamps is refused loudly instead of spinning', async () => {
|
||
const source = scriptedSource(() => range(1, 4))
|
||
const engine = new ReprojectionEngine({ source, batchSize: 2 })
|
||
const adapter: ProjectionAdapter = {
|
||
family: 'f6',
|
||
watermark: () => null, // never advances
|
||
applyBatch: async () => {},
|
||
discard: async () => {}
|
||
}
|
||
engine.register(adapter)
|
||
|
||
await expect(engine.advance('f6', { budgetMs: 10_000 })).rejects.toThrow(/not stamping/)
|
||
})
|
||
})
|
||
|
||
describe('reprojection engine — (g) discard lands on the losing adapter after a swap', () => {
|
||
it('the OLD adapter is discarded exactly once, after the flip; the winner is never discarded', async () => {
|
||
const source = scriptedSource(() => range(1, 6))
|
||
const engine = new ReprojectionEngine({ source, batchSize: 3 })
|
||
const losing = new RecordingAdapter('g')
|
||
engine.register(losing)
|
||
await engine.advance('g', { budgetMs: 10_000 })
|
||
expect(losing.discarded).toBe(0) // serving adapters are never discarded
|
||
|
||
let winner!: RecordingAdapter
|
||
await engine.swap('g', async () => {
|
||
winner = new RecordingAdapter('g')
|
||
winner.onApply = () => {
|
||
// Mid-build the loser still serves and is still intact.
|
||
expect(losing.discarded).toBe(0)
|
||
}
|
||
return winner
|
||
})
|
||
|
||
expect(losing.discarded).toBe(1)
|
||
expect(winner.discarded).toBe(0)
|
||
expect(engine.getAdapter('g')).toBe(winner)
|
||
})
|
||
})
|
||
|
||
describe('FactLogSource — the production source enforces the window contract', () => {
|
||
it('delegates to the injected callback and passes clean windows through', async () => {
|
||
const calls: Array<[number, number]> = []
|
||
const source = new FactLogSource(async (from, limit) => {
|
||
calls.push([from, limit])
|
||
return range(from + 1, Math.min(from + limit, 5)).map(fact)
|
||
})
|
||
const facts = await source.scan(2, 2)
|
||
expect(facts.map((f) => f.generation)).toEqual([3, 4])
|
||
expect(calls).toEqual([[2, 2]])
|
||
expect(await source.scan(5, 3)).toEqual([])
|
||
})
|
||
|
||
it('refuses out-of-contract callbacks loudly: oversize, non-ascending, at-or-below from', async () => {
|
||
const oversize = new FactLogSource(async () => range(1, 5).map(fact))
|
||
await expect(oversize.scan(0, 2)).rejects.toThrow(/contract violation/)
|
||
|
||
const unsorted = new FactLogSource(async () => [fact(3), fact(2)])
|
||
await expect(unsorted.scan(0, 10)).rejects.toThrow(/strictly ascending/)
|
||
|
||
const stale = new FactLogSource(async () => [fact(2)])
|
||
await expect(stale.scan(2, 10)).rejects.toThrow(/strictly ascending/)
|
||
})
|
||
|
||
it('validates its own window arguments', async () => {
|
||
const source = new FactLogSource(async () => [])
|
||
await expect(source.scan(-1, 5)).rejects.toThrow(/non-negative integer/)
|
||
await expect(source.scan(0, 0)).rejects.toThrow(/positive integer/)
|
||
})
|
||
})
|