brainy/tests/benchmarks/model-b-scalability.spike.ts
David Snelling 5c3bb2c864 feat(8.0): Model-B per-write generation-stamping + adaptive retention knob
Every write — transact() AND single-op add/update/remove/relate — is now its
own immutable generation (Model-B), so a now() pin always freezes and
asOf/since/diff/history/transactionLog reflect single-ops exactly like
transacts. Closes the Model-A hole where pins did not freeze against single-op
writes.

Generation-stamping:
- GenerationStore.commitSingleOp: a one-operation commitTransaction with
  deferred durability. Wired into add/update/remove/relate/updateRelation/
  unrelate + removeMany (the *Many and VFS paths delegate to these).
- Async group-commit (flushPendingSingleOps): the live write is acknowledged
  immediately; its before-image is buffered in an in-memory pending tier that
  resolveAt/chains/changedBetween/tx-log read like on-disk generations, so the
  synchronous now() freezes with no forced flush. One fsync per window
  (triggers: size / 50ms timer / flush / close / transact / compactHistory).
- Crash recovery is drop-without-restore for group-commit generations (marked
  groupCommit:true): a crash mid-flush discards the partial generation and
  never restores its before-images, which would otherwise revert the
  already-acknowledged live write.
- Init-time infrastructure (the VFS root) is the un-versioned generation-0
  baseline: a fresh brain reports generation()===0 and an empty
  transactionLog(); the first user write is generation 1.
- Historical find()/related() overlay bound is the full reserved watermark
  (generation()), so un-flushed single-op writes are overlaid too.

Retention knob:
- config `history` -> `retention`: 'all' | 'adaptive' |
  { maxGenerations?, maxAge?, maxBytes?, budgetBytes?, autoCompact? }; unset ->
  adaptive (disk/RAM pressure, zero-config). CompactHistoryOptions floors ->
  caps (retainGenerations->maxGenerations, retainMs->maxAge, +maxBytes):
  reclaim oldest-unpinned while ANY cap is exceeded; pins always exempt.
- brain.setRetentionBudget(bytes) drives the adaptive byte budget at runtime
  (a coordinator's fair-share input). Per-generation bytes recorded in each
  delta enable historyBytes() introspection without a storage size API.

Tests: per-write generation resolution, pin freeze vs add/update/remove,
drop-without-restore corruption-trap (fault injector), clean-reopen replay,
maxBytes/maxAge/no-cap reclamation, retention-then-reopen. 107 db/generation/
temporal tests green, tsc clean. Docs (ADR-001, consistency-model, snapshots
guide, api reference, RELEASES) updated to per-write granularity + retention.
2026-06-22 15:19:58 -07:00

196 lines
10 KiB
TypeScript

/**
* @module tests/benchmarks/model-b-scalability.spike
* @description SPIKE (throwaway, spike/model-b-write-perf branch) — the DECISION-CRITICAL
* half of the Model B evaluation. The write-perf spike showed writes are cheap at
* durability parity; the adversarial review showed the REAL risk is structural and
* grows with the number of retained generations N:
*
* 1. READ-vs-depth — resolveAt/changedBetween LINEARLY scan the global committedGens
* array → historical reads (asOf/get, diff, since) become O(N) = O(database-age).
* 2. REOPEN-vs-depth — GenerationStore.open enumerates+sorts EVERY generation dir → O(N) cold start.
* 3. RAM-vs-depth — deltaCache holds a Set per committed generation → O(N) resident heap.
* 4. COMPACTION cost — the only thing that bounds 1-3; time it + the post-compaction footprint.
*
* Each is measured at two depths (small, large). A ~linear slope CONFIRMS that brainy-TS's
* current global-committedGens design does NOT scale under Model B (every write = a generation),
* and that the canonical Model-B history must be a PER-ID INDEX (mirroring cor's delta_history
* BTreeMap, O(log n)) rather than a global scan. Memory backend is used where the scan/RAM cost
* is backend-independent (committedGens is in-memory); filesystem only where reopen/compaction
* touch disk.
*
* Run (force GC available): node --expose-gc node_modules/.bin/tsx tests/benchmarks/model-b-scalability.spike.ts
* small scale: SPIKE_SCALE=small ...
*/
import { rmSync, mkdirSync, existsSync } from 'node:fs'
import { execSync } from 'node:child_process'
import { Brainy } from '../../src/index.js'
import { NounType } from '../../src/types/graphTypes.js'
import type { BrainyConfig } from '../../src/types/brainy.types.js'
// Silence brainy's verbose per-brain logging (floods stdout, slows the run).
const _log = console.log.bind(console)
const say = (...a: any[]) => _log(...a)
for (const m of ['log', 'info', 'debug', 'warn'] as const) (console as any)[m] = () => {}
const SMALL = process.env.SPIKE_SCALE === 'small'
const DIM = 384
const FS_ROOT = '/media/dpsifr/storage/home/Projects/brainy/.spike-data'
// Two depths to read the slope. Memory legs can go deep cheaply; fs legs (fsync per gen) stay smaller.
const DEPTHS_MEM = SMALL ? [500, 5_000] : [2_000, 20_000]
const DEPTHS_FS = SMALL ? [300, 1_500] : [1_000, 6_000]
const READ_REPEATS = SMALL ? 20 : 50
function vec(i: number): number[] {
const v = new Array<number>(DIM)
let mag = 0
for (let d = 0; d < DIM; d++) { const x = Math.sin((i + 1) * 0.001 + d * 0.1); v[d] = x; mag += x * x }
mag = Math.sqrt(mag) || 1
for (let d = 0; d < DIM; d++) v[d] /= mag
return v
}
async function mkBrain(backend: 'memory' | 'filesystem', dir?: string): Promise<Brainy> {
const storage: BrainyConfig['storage'] =
backend === 'memory' ? { type: 'memory' } : { type: 'filesystem', path: dir! }
const brain = new Brainy({
requireSubtype: false,
storage,
eagerEmbeddings: false, // never load WASM — we pass precomputed vectors
// Keep history fully retained for the measurement (no mid-run compaction skewing depth).
retention: { autoCompact: false } as any,
index: { m: 16, efConstruction: 200, efSearch: 50 },
cache: { maxSize: 1000, ttl: 3600 }
} as BrainyConfig)
brain.use({ name: 'const-embedder', activate: async (ctx: any) => { ctx.registerProvider('embeddings', async () => vec(0)); return true } } as any)
await brain.init()
return brain
}
function payload(i: number) {
return { type: NounType.Document, subtype: 'note', data: `doc ${i}`, vector: vec(i), metadata: { i, batch: 'spike' } }
}
function freshDir(name: string): string {
const dir = `${FS_ROOT}/${name}`
if (existsSync(dir)) rmSync(dir, { recursive: true, force: true })
mkdirSync(dir, { recursive: true })
return dir
}
function allocBytes(dir: string): number { try { return parseInt(execSync(`du -s --block-size=1 ${dir}`, { encoding: 'utf8' }).split(/\s+/)[0], 10) } catch { return -1 } }
const SEED = 32 // small working set; X (id[0]) is the never-updated probe entity
/** Build N committed generations via transact([1]). X = ids[0] is seeded then NEVER updated, so a
* read of X at the oldest pin must scan all N generations to confirm it is unchanged (worst case). */
async function buildGenerations(b: Brainy, nGens: number): Promise<{ ids: string[]; g0: number; gN: number }> {
const ids: string[] = []
// Seed via ONE transact so g0 is a committed generation (asOf needs a committed gen).
const seedOps = Array.from({ length: SEED }, (_, i) => ({ op: 'add' as const, ...payload(i) }))
const seedDb = await b.transact(seedOps as any)
for (const id of (await b.find({ type: NounType.Document, limit: SEED }))) ids.push(id.id)
await seedDb.release?.()
const g0 = b.generation()
// N generations, each a single-op-equivalent transact([1]) update on the CHURN set (ids[1..]),
// never touching X=ids[0].
for (let k = 0; k < nGens; k++) {
const id = ids[1 + (k % (ids.length - 1))]
const d = await b.transact([{ op: 'update', id, metadata: { k } }] as any)
await d.release?.()
}
return { ids, g0, gN: b.generation() }
}
function median(xs: number[]): number { const s = [...xs].sort((a, b) => a - b); return s[Math.floor(s.length / 2)] }
async function timeMs(fn: () => Promise<void>, repeats: number): Promise<number> {
const ts: number[] = []
for (let r = 0; r < repeats; r++) { const t = process.hrtime.bigint(); await fn(); ts.push(Number(process.hrtime.bigint() - t) / 1e6) }
return median(ts)
}
async function main() {
say(`[scale] start (scale=${SMALL ? 'small' : 'full'}); gc=${typeof (global as any).gc === 'function' ? 'on' : 'OFF (run with node --expose-gc)'}`)
if (existsSync(FS_ROOT)) rmSync(FS_ROOT, { recursive: true, force: true })
mkdirSync(FS_ROOT, { recursive: true })
const results: any = { readVsDepth: [], reopenVsDepth: [], ramVsDepth: [], compaction: [] }
// ---- 1. READ latency vs history depth (memory; scan is in-memory) ----
say(`\n========== READ latency vs history depth (the O(database-age) test) ==========`)
for (const N of DEPTHS_MEM) {
const b = await mkBrain('memory')
const { ids, g0, gN } = await buildGenerations(b, N)
const X = ids[0]
// asOf(oldest).get(X): resolveAt(X,g0) scans all N committedGens (X never touched) → expect O(N).
const tAsOfGet = await timeMs(async () => { const db = await b.asOf(g0); await db.get(X); await db.release?.() }, READ_REPEATS)
// diff(g0,gN): changedBetween scans the full (g0,gN] range → expect O(N).
const tDiff = await timeMs(async () => { await b.diff(g0, gN) }, Math.max(5, READ_REPEATS / 5))
await b.close()
say(`N=${N.toLocaleString()} gens: asOf(oldest).get(X) = ${tAsOfGet.toFixed(3)} ms | diff(g0,gN) = ${tDiff.toFixed(2)} ms`)
results.readVsDepth.push({ N, asOfGetMs: tAsOfGet, diffMs: tDiff })
}
{
const [a, c] = results.readVsDepth
if (a && c) say(` → asOf-get scaling ${a.N}${c.N} (${(c.N / a.N).toFixed(0)}x depth): ${(c.asOfGetMs / a.asOfGetMs).toFixed(1)}x slower (linear≈${(c.N / a.N).toFixed(0)}x → confirms O(N) scan)`)
}
// ---- 2. RAM resident vs history depth (memory; deltaCache holds a Set per gen) ----
say(`\n========== RAM (heapUsed) vs history depth (deltaCache O(N) Sets) ==========`)
for (const N of DEPTHS_MEM) {
if ((global as any).gc) (global as any).gc()
const before = process.memoryUsage().heapUsed
const b = await mkBrain('memory')
await buildGenerations(b, N)
if ((global as any).gc) (global as any).gc()
const after = process.memoryUsage().heapUsed
const perGen = (after - before) / N
await b.close()
say(`N=${N.toLocaleString()} gens: heapUsed +${((after - before) / 1024 / 1024).toFixed(1)} MiB (${perGen.toFixed(0)} bytes/generation resident)`)
results.ramVsDepth.push({ N, deltaBytes: after - before, bytesPerGen: perGen })
}
// ---- 3. COLD-REOPEN vs history depth (filesystem; open() walks every gen dir) ----
say(`\n========== COLD-REOPEN time vs history depth (O(generations) dir walk) ==========`)
for (const N of DEPTHS_FS) {
const dir = freshDir(`reopen-${N}`)
let b = await mkBrain('filesystem', dir)
await buildGenerations(b, N)
await b.close()
const tOpen = await timeMs(async () => { const rb = await mkBrain('filesystem', dir); await rb.close() }, SMALL ? 3 : 5)
rmSync(dir, { recursive: true, force: true })
say(`N=${N.toLocaleString()} gens: cold reopen = ${tOpen.toFixed(0)} ms`)
results.reopenVsDepth.push({ N, reopenMs: tOpen })
}
{
const [a, c] = results.reopenVsDepth
if (a && c) say(` → reopen scaling ${a.N}${c.N} (${(c.N / a.N).toFixed(0)}x depth): ${(c.reopenMs / a.reopenMs).toFixed(1)}x slower`)
}
// ---- 4. COMPACTION cost + footprint (filesystem) ----
say(`\n========== COMPACTION cost (the bound on 1-3) ==========`)
{
const N = DEPTHS_FS[DEPTHS_FS.length - 1]
const dir = freshDir(`compact-${N}`)
const b = await mkBrain('filesystem', dir)
await buildGenerations(b, N)
await b.flush()
const beforeBytes = allocBytes(`${dir}/_generations`)
const t = process.hrtime.bigint()
const res = await b.compactHistory({ maxGenerations: 100 } as any)
const compMs = Number(process.hrtime.bigint() - t) / 1e6
await b.flush()
const afterBytes = allocBytes(`${dir}/_generations`)
await b.close()
rmSync(dir, { recursive: true, force: true })
say(`N=${N.toLocaleString()} gens → compactHistory({maxGenerations:100}): ${compMs.toFixed(0)} ms, removed ${(res as any)?.removedGenerations ?? '?'} gens`)
say(` _generations footprint: ${(beforeBytes / 1024 / 1024).toFixed(1)} MiB → ${(afterBytes / 1024 / 1024).toFixed(1)} MiB allocated`)
results.compaction.push({ N, compactMs: compMs, removed: (res as any)?.removedGenerations, beforeBytes, afterBytes })
}
const fs = await import('node:fs')
fs.writeFileSync('/media/dpsifr/storage/home/Projects/brainy/tests/benchmarks/_scalability-results.json', JSON.stringify(results, null, 2))
say(`\nRaw results → tests/benchmarks/_scalability-results.json`)
if (existsSync(FS_ROOT)) rmSync(FS_ROOT, { recursive: true, force: true })
}
main().then(() => { say('[scale] done.'); process.exit(0) }).catch((e) => { console.error(e); process.exit(1) })