David Snelling
a859d6ecf8
The MVCC time-travel layer kept nounChains/verbChains as unbounded Map<id, number[]> — one resident chain per id ever touched across retained history (O(N) RAM, defeating billion-scale time travel). Replace with a hot-tail window + bounded cold LRU + a mutex-free bulk resolver, keeping resolveAt exactly correct. The most-recent W generations stay resident as full chains (the common recent-pin read is O(log), zero scan); deeper pins reconstruct one id's chain on demand into an L-bounded LRU. Reconstruction is LOCK-LIGHT — it holds no commit mutex, so a historical read never stalls writers; correctness holds because a live pin's answer is always > minPinnedGeneration, which compaction can never reclaim, and a concurrently-reclaimed gen below that is skipped. materializeAtGeneration routes through a new mutex-free resolveManyAt (one forward pass, O(R + |ids|)) — without it a deep-pin materialize both regresses to O(N*R) and deadlocks on snapshotWith's mutex. The cold cache is invalidated on re-touch to stay coherent. Resident RAM is O(W*d + L), independent of N. 11-case test (oracle-vs-bruteforce, held-Db across eviction + concurrent compaction, no-deadlock, write-path I/O-free); unit 1735 + integration 613 (db-mvcc/db-temporal/db-asof green).
542 lines
22 KiB
TypeScript
542 lines
22 KiB
TypeScript
/**
|
||
* @module tests/unit/db/bounded-chains
|
||
* @description Correctness + RAM-boundedness oracle for the bounded per-id
|
||
* generation history chains in `src/db/generationStore.ts` (GA #33, Approach C:
|
||
* hot-tail window + bounded cold LRU + a mutex-free bulk `resolveManyAt`).
|
||
*
|
||
* The old design kept ONE resident chain per id ever touched across retained
|
||
* history (`Map<id, number[]>`) → O(distinct-ids) RAM, which defeats
|
||
* billion-scale time travel. The replacement bounds RAM to O(W·d̄)/O(L) — a
|
||
* resident window over the newest `W` generations plus a bounded LRU of
|
||
* reconstructed deep-pin chains — while staying EXACTLY correct.
|
||
*
|
||
* These tests are the correctness oracle for that change:
|
||
*
|
||
* 1. Oracle vs brute-force: for every id × every pin, `resolveAt` must equal a
|
||
* brute-force scan of every reserved generation's persisted `tx.json`.
|
||
* 2. Bounded RAM independent of N: the resident structures stay O(W)/O(L) no
|
||
* matter how many distinct ids the corpus touched.
|
||
* 3. Held-Db across cold eviction + a concurrent compaction (lock-light).
|
||
* 4. Un-flushed pending single-ops resolve identically before and after flush.
|
||
* 5. A deep-pin materialize is O(R) `getDelta` (not O(N·R)) and never deadlocks
|
||
* inside the mutex-held reconciliation pass.
|
||
* 6. `resolveManyAt` ≡ per-id `resolveAt` for random id-sets × random pins.
|
||
* 7. The write path is I/O-free across window slides (the inverse-index ring
|
||
* supplies slid-out ids — zero `tx.json` reads on commit).
|
||
* 8. Compaction invalidates + rebuilds the window over survivors; a pin below
|
||
* the new horizon throws `GenerationCompactedError`.
|
||
*/
|
||
|
||
import { describe, it, expect, beforeEach, afterEach } from 'vitest'
|
||
import { MemoryStorage } from '../../../src/storage/adapters/memoryStorage.js'
|
||
import { GenerationStore } from '../../../src/db/generationStore.js'
|
||
import { GenerationCompactedError } from '../../../src/db/errors.js'
|
||
import { Brainy } from '../../../src/index.js'
|
||
import { NounType } from '../../../src/types/graphTypes.js'
|
||
import { createTestConfig, generateTestVector } from '../../helpers/test-factory.js'
|
||
|
||
/** A precomputed embedding so adds skip the (slow) embedding model — these tests
|
||
* exercise the generation layer, not semantics. */
|
||
const VEC = generateTestVector()
|
||
|
||
/** Deterministic UUID-shaped id (the sharded storage layout derives the shard
|
||
* from the UUID hex), unique per `n`. */
|
||
function uid(n: number): string {
|
||
return `00000000-0000-4000-8000-${n.toString(16).padStart(12, '0')}`
|
||
}
|
||
|
||
/** Stored-metadata fixture carrying a `version` the oracle compares on. */
|
||
function fixture(version: number): Record<string, unknown> {
|
||
return {
|
||
noun: NounType.Document,
|
||
subtype: 'note',
|
||
data: `payload-v${version}`,
|
||
version,
|
||
createdAt: 1000,
|
||
updatedAt: 1000 + version,
|
||
_rev: version
|
||
}
|
||
}
|
||
|
||
/** A small deterministic PRNG so the random oracle runs are reproducible. */
|
||
function mulberry32(seed: number): () => number {
|
||
let a = seed >>> 0
|
||
return () => {
|
||
a |= 0
|
||
a = (a + 0x6d2b79f5) | 0
|
||
let t = Math.imul(a ^ (a >>> 15), 1 | a)
|
||
t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t
|
||
return ((t ^ (t >>> 14)) >>> 0) / 4294967296
|
||
}
|
||
}
|
||
|
||
/** Reject after `ms` so a deadlock surfaces as a test failure, not a hang. */
|
||
function withTimeout<T>(p: Promise<T>, ms: number, label: string): Promise<T> {
|
||
return Promise.race([
|
||
p,
|
||
new Promise<T>((_, reject) =>
|
||
setTimeout(() => reject(new Error(`timed out after ${ms}ms (likely deadlock): ${label}`)), ms)
|
||
)
|
||
])
|
||
}
|
||
|
||
/** Normalize a `resolveAt` result to a comparable scalar. */
|
||
function norm(r: { source: string; metadata?: any }): string {
|
||
if (r.source === 'current') return 'current'
|
||
if (r.source === 'absent') return 'absent'
|
||
return `rec:${r.metadata?.version}`
|
||
}
|
||
|
||
describe('db/GenerationStore — bounded per-id chains (GA #33)', () => {
|
||
let storage: MemoryStorage
|
||
let store: GenerationStore
|
||
|
||
beforeEach(async () => {
|
||
storage = new MemoryStorage()
|
||
await storage.init()
|
||
store = new GenerationStore(storage)
|
||
await store.open()
|
||
})
|
||
|
||
/** Commit one single-op noun write (its own generation); returns the gen. */
|
||
async function writeNoun(id: string, version: number): Promise<number> {
|
||
const { generation } = await store.commitSingleOp({
|
||
touched: { nouns: [id] },
|
||
execute: async () => {
|
||
await storage.saveNounMetadata(id, fixture(version))
|
||
}
|
||
})
|
||
return generation
|
||
}
|
||
|
||
/** Commit one single-op verb write (raw, so before-images are captured). */
|
||
async function writeVerb(id: string, version: number): Promise<number> {
|
||
const { generation } = await store.commitSingleOp({
|
||
touched: { verbs: [id] },
|
||
execute: async () => {
|
||
await storage.writeVerbRaw(id, {
|
||
metadata: { ...fixture(version), verb: 'RelatedTo' },
|
||
vector: { sourceId: uid(900001), targetId: uid(900002), verb: 'RelatedTo' }
|
||
})
|
||
}
|
||
})
|
||
return generation
|
||
}
|
||
|
||
/** Brute-force `resolveAt`: scan EVERY reserved generation's persisted tx.json
|
||
* (the gens must be flushed) for the first one after `pin` that touched `id`,
|
||
* then read its before-image. Fully independent of the resident chains. */
|
||
async function bruteForce(
|
||
kind: 'noun' | 'verb',
|
||
id: string,
|
||
pin: number,
|
||
maxGen: number
|
||
): Promise<string> {
|
||
for (let g = pin + 1; g <= maxGen; g++) {
|
||
const delta = (await storage.readRawObject(`_generations/${g}/tx.json`)) as
|
||
| { nouns: string[]; verbs: string[] }
|
||
| null
|
||
if (!delta) continue
|
||
const touched = kind === 'noun' ? delta.nouns : delta.verbs
|
||
if (Array.isArray(touched) && touched.includes(id)) {
|
||
const rec = (await storage.readRawObject(`_generations/${g}/prev/${id}.json`)) as
|
||
| { metadata: any; vector: any }
|
||
| null
|
||
if (!rec || (rec.metadata === null && rec.vector === null)) return 'absent'
|
||
return `rec:${rec.metadata?.version}`
|
||
}
|
||
}
|
||
return 'current'
|
||
}
|
||
|
||
// ==========================================================================
|
||
// 1. Oracle vs brute-force — every id × every pin, nouns + verbs.
|
||
// ==========================================================================
|
||
it('resolveAt equals a brute-force tx.json scan for every id × every pin (recent, deep, boundary)', async () => {
|
||
const s = store as any
|
||
s.recentWindowGenerations = 8 // W
|
||
s.coldChainLruMax = 4 // L
|
||
|
||
const nounIds = Array.from({ length: 5 }, (_, i) => uid(100 + i))
|
||
const verbIds = Array.from({ length: 5 }, (_, i) => uid(200 + i))
|
||
const rand = mulberry32(1234)
|
||
let v = 0
|
||
|
||
// 60 overlapping single-op generations. Trigger a historical read at the
|
||
// halfway point so the SECOND half exercises the live slide path
|
||
// (extendChains + window advance), not just the lazy build.
|
||
for (let i = 0; i < 60; i++) {
|
||
if (i === 30) {
|
||
// Force the window to build so the rest of the writes drive slides.
|
||
await store.resolveAt('noun', nounIds[0], 1)
|
||
}
|
||
if (rand() < 0.5) {
|
||
await writeNoun(nounIds[Math.floor(rand() * nounIds.length)], ++v)
|
||
} else {
|
||
await writeVerb(verbIds[Math.floor(rand() * verbIds.length)], ++v)
|
||
}
|
||
}
|
||
// Flush so every reserved generation has a persisted tx.json for the oracle.
|
||
await store.flushPendingSingleOps()
|
||
const maxGen = store.generation()
|
||
|
||
// Window must have actually slid below the head (proves the slide path ran).
|
||
expect(s.windowLo).toBeGreaterThan(1)
|
||
expect(s.windowLo).toBe(Math.max(1, maxGen - 8 + 1))
|
||
|
||
// Every id × every pin from 0..maxGen — covers recent (≥windowLo), deep
|
||
// (<windowLo), and the exact window/horizon boundaries.
|
||
for (let pin = 0; pin <= maxGen; pin++) {
|
||
for (const id of nounIds) {
|
||
expect(norm(await store.resolveAt('noun', id, pin))).toBe(
|
||
await bruteForce('noun', id, pin, maxGen)
|
||
)
|
||
}
|
||
for (const id of verbIds) {
|
||
expect(norm(await store.resolveAt('verb', id, pin))).toBe(
|
||
await bruteForce('verb', id, pin, maxGen)
|
||
)
|
||
}
|
||
}
|
||
})
|
||
|
||
// ==========================================================================
|
||
// 1b. Cold-cache coherence: an id touched AFTER it was cold-cached must NOT be
|
||
// served stale (the current↔record transition the unbounded design got
|
||
// for free).
|
||
// ==========================================================================
|
||
it('a deep pin stays correct when the id is touched again after its cold chain was cached', async () => {
|
||
const s = store as any
|
||
s.recentWindowGenerations = 4
|
||
s.coldChainLruMax = 16
|
||
|
||
const X = uid(300)
|
||
const filler = (i: number) => uid(400 + i)
|
||
|
||
const g1 = await writeNoun(X, 1) // X = v1
|
||
// Push X out of the window with filler writes (X untouched).
|
||
for (let i = 0; i < 10; i++) await writeNoun(filler(i), 100 + i)
|
||
|
||
// Deep read at g1 BEFORE any later touch: X untouched after g1 → 'current'.
|
||
expect(norm(await store.resolveAt('noun', X, g1))).toBe('current')
|
||
expect((await store.resolveAt('noun', X, g1)).source).toBe('current') // now cached cold
|
||
|
||
// Now touch X again (v2). The stale cached cold chain would still answer
|
||
// 'current' → reading the NEW live value (v1 corrupted to v2). extendChains
|
||
// must invalidate the cold entry so the re-read resolves the g1 before-image.
|
||
await writeNoun(X, 2)
|
||
await store.flushPendingSingleOps()
|
||
|
||
const resolved = await store.resolveAt('noun', X, g1)
|
||
expect(resolved.source).toBe('record')
|
||
if (resolved.source === 'record') expect(resolved.metadata.version).toBe(1)
|
||
})
|
||
|
||
// ==========================================================================
|
||
// 2. Bounded RAM independent of N.
|
||
// ==========================================================================
|
||
it('resident window/cold structures stay O(W)/O(L), independent of distinct-id count', async () => {
|
||
const s = store as any
|
||
const W = 8
|
||
const L = 4
|
||
s.recentWindowGenerations = W
|
||
s.coldChainLruMax = L
|
||
|
||
const N = 1200 // distinct ids, one single-op each
|
||
for (let i = 0; i < N; i++) await writeNoun(uid(1000 + i), i)
|
||
await store.flushPendingSingleOps()
|
||
const maxGen = store.generation()
|
||
|
||
// Build the window (first historical read).
|
||
await store.resolveAt('noun', uid(1000), maxGen)
|
||
|
||
// The hot-tail window holds at most W generations' worth of ids; one id per
|
||
// gen here ⇒ exactly W resident recent chains + W inverse-index entries.
|
||
expect(s.recentNounChains.size).toBeLessThanOrEqual(W)
|
||
expect(s.windowNounDeltas.size).toBeLessThanOrEqual(W)
|
||
expect(s.recentNounChains.size).toBeLessThan(N)
|
||
|
||
// Deep-read many distinct ids → the cold LRU is capped at L regardless.
|
||
for (let i = 0; i < 40; i++) await store.resolveAt('noun', uid(1000 + i), 1)
|
||
expect(s.coldNounChains.size).toBeLessThanOrEqual(L)
|
||
})
|
||
|
||
// ==========================================================================
|
||
// 3. Held-Db across cold eviction + a concurrent compaction (lock-light).
|
||
// ==========================================================================
|
||
it('a held pin reads correctly across cold eviction AND a concurrent compact()', async () => {
|
||
const s = store as any
|
||
s.recentWindowGenerations = 4
|
||
s.coldChainLruMax = 4
|
||
|
||
const heldIds = [uid(500), uid(501), uid(502)]
|
||
const filler = (i: number) => uid(600 + i)
|
||
|
||
// Each held id gets value 10/11/12 at g_old, then a later mutation so a
|
||
// before-image exists to resolve.
|
||
const gOldVals = new Map<string, number>()
|
||
let pOld = 0
|
||
for (let k = 0; k < heldIds.length; k++) {
|
||
const val = 10 + k
|
||
pOld = await writeNoun(heldIds[k], val)
|
||
gOldVals.set(heldIds[k], val)
|
||
}
|
||
const gOld = pOld // pin: at gOld every held id holds its value above
|
||
store.pin(gOld)
|
||
|
||
// Mutate the held ids AFTER the pin (so resolveAt resolves a before-image),
|
||
// then slide the window well past gOld with lots of filler writes.
|
||
for (const id of heldIds) await writeNoun(id, 99)
|
||
for (let i = 0; i < 30; i++) await writeNoun(filler(i), 700 + i)
|
||
await store.flushPendingSingleOps()
|
||
|
||
// Populate then evict the held ids' cold chains by cycling other deep ids.
|
||
for (const id of heldIds) await store.resolveAt('noun', id, gOld)
|
||
for (let i = 0; i < 12; i++) await store.resolveAt('noun', filler(i), 1)
|
||
|
||
// Re-read the held ids WHILE a compaction (which reclaims everything ≤ the
|
||
// pin) runs concurrently. Lock-light reconstruction must skip the reclaimed
|
||
// gens (all ≤ minPinned, never an answer) and still match the at-gOld oracle.
|
||
const reads = heldIds.map((id) =>
|
||
withTimeout(store.resolveAt('noun', id, gOld), 5000, `held read ${id}`)
|
||
)
|
||
const compaction = withTimeout(store.compact(), 5000, 'concurrent compact')
|
||
const [r0, r1, r2] = await Promise.all(reads)
|
||
await compaction
|
||
|
||
for (const [id, res] of [
|
||
[heldIds[0], r0],
|
||
[heldIds[1], r1],
|
||
[heldIds[2], r2]
|
||
] as const) {
|
||
expect(res.source).toBe('record')
|
||
if (res.source === 'record') expect(res.metadata.version).toBe(gOldVals.get(id))
|
||
}
|
||
|
||
// And a fresh read after compaction still matches (reconstruct over survivors).
|
||
for (const id of heldIds) {
|
||
const res = await store.resolveAt('noun', id, gOld)
|
||
expect(res.source).toBe('record')
|
||
if (res.source === 'record') expect(res.metadata.version).toBe(gOldVals.get(id))
|
||
}
|
||
store.release(gOld)
|
||
})
|
||
|
||
// ==========================================================================
|
||
// 4. Un-flushed pending single-ops — flush-agnostic before-images.
|
||
// ==========================================================================
|
||
it('a now()-style pin reads pre-mutation before-images of un-flushed single-ops, identical after flush', async () => {
|
||
const X = uid(800)
|
||
const g1 = await writeNoun(X, 1) // flushed below to make g1 durable history
|
||
|
||
// More UN-FLUSHED single-ops mutating the same id.
|
||
await writeNoun(X, 2)
|
||
await writeNoun(X, 3)
|
||
expect(store.committedGeneration()).toBe(0) // nothing on disk yet
|
||
|
||
const before = await store.resolveAt('noun', X, g1)
|
||
expect(before.source).toBe('record')
|
||
if (before.source === 'record') expect(before.metadata.version).toBe(1)
|
||
|
||
// Flush → the answer is identical (resolution is flush-agnostic).
|
||
await store.flushPendingSingleOps()
|
||
const after = await store.resolveAt('noun', X, g1)
|
||
expect(norm(after)).toBe(norm(before))
|
||
})
|
||
|
||
// ==========================================================================
|
||
// 6. resolveManyAt ≡ per-id resolveAt (random id-sets × random pins).
|
||
// ==========================================================================
|
||
it('resolveManyAt equals per-id resolveAt for random id-sets across recent and deep pins', async () => {
|
||
const s = store as any
|
||
s.recentWindowGenerations = 6
|
||
s.coldChainLruMax = 8
|
||
|
||
const ids = Array.from({ length: 8 }, (_, i) => uid(2000 + i))
|
||
const rand = mulberry32(99)
|
||
let v = 0
|
||
for (let i = 0; i < 50; i++) await writeNoun(ids[Math.floor(rand() * ids.length)], ++v)
|
||
await store.flushPendingSingleOps()
|
||
const maxGen = store.generation()
|
||
|
||
/** Map a resolveManyAt firstAfter entry to the same scalar resolveAt yields. */
|
||
const derived = async (id: string, firstAfter: Map<string, number>): Promise<string> => {
|
||
const g = firstAfter.get(id)
|
||
if (g === undefined) return 'current'
|
||
const rec = await store.readGenerationRecord('noun', g, id)
|
||
if (!rec || (rec.metadata === null && rec.vector === null)) return 'absent'
|
||
return `rec:${(rec.metadata as any)?.version}`
|
||
}
|
||
|
||
for (let pin = 0; pin <= maxGen; pin++) {
|
||
// A random subset of ids for this pin.
|
||
const subset = new Set(ids.filter(() => rand() < 0.6))
|
||
if (subset.size === 0) subset.add(ids[0])
|
||
const many = await store.resolveManyAt('noun', subset, pin)
|
||
for (const id of subset) {
|
||
expect(await derived(id, many)).toBe(norm(await store.resolveAt('noun', id, pin)))
|
||
}
|
||
}
|
||
})
|
||
|
||
// ==========================================================================
|
||
// 7. Write path is I/O-free across window slides (the ring supplies slid-out
|
||
// ids — no tx.json reads on commit, even with W > deltaCacheMax).
|
||
// ==========================================================================
|
||
it('committing across window slides reads ZERO tx.json from disk (W > deltaCacheMax)', async () => {
|
||
const s = store as any
|
||
s.recentWindowGenerations = 6 // W
|
||
s.deltaCacheMax = 2 // W > deltaCacheMax: a naive slide would re-read tx.json
|
||
|
||
// Seed + build the window so subsequent commits drive slides.
|
||
for (let i = 0; i < 8; i++) await writeNoun(uid(3000 + i), i)
|
||
await store.flushPendingSingleOps()
|
||
await store.resolveAt('noun', uid(3000), store.generation())
|
||
|
||
// Count tx.json reads from this point on.
|
||
let txReads = 0
|
||
const realRead = storage.readRawObject.bind(storage)
|
||
;(storage as any).readRawObject = async (path: string) => {
|
||
if (typeof path === 'string' && path.endsWith('/tx.json')) txReads++
|
||
return realRead(path)
|
||
}
|
||
try {
|
||
// Many more commits, each sliding the window past evicted-from-cache gens.
|
||
for (let i = 0; i < 20; i++) await writeNoun(uid(3100 + i), 1000 + i)
|
||
} finally {
|
||
;(storage as any).readRawObject = realRead
|
||
}
|
||
expect(txReads).toBe(0)
|
||
})
|
||
|
||
// ==========================================================================
|
||
// 8. Compaction invalidates + rebuilds the window over survivors; a pin below
|
||
// the new horizon throws GenerationCompactedError.
|
||
// ==========================================================================
|
||
it('rebuilds the window over survivors after compaction and rejects pins below the horizon', async () => {
|
||
const s = store as any
|
||
s.recentWindowGenerations = 4
|
||
|
||
const X = uid(900)
|
||
const g0 = await writeNoun(X, 0)
|
||
for (let v = 1; v <= 6; v++) await writeNoun(X, v)
|
||
await store.flushPendingSingleOps()
|
||
const maxGen = store.generation()
|
||
|
||
// Build the window, then compact to keep only the 2 most recent gens.
|
||
await store.resolveAt('noun', X, maxGen)
|
||
const res = await store.compact({ maxGenerations: 2 })
|
||
expect(res.removedGenerations).toBeGreaterThan(0)
|
||
expect(s.windowReady).toBe(false) // invalidated
|
||
|
||
// A pin ABOVE the horizon rebuilds the window from survivors and resolves.
|
||
const aboveHorizon = res.horizon + 1
|
||
const resolved = await store.resolveAt('noun', X, aboveHorizon)
|
||
expect(['record', 'current']).toContain(resolved.source)
|
||
|
||
// A pin below the horizon is unreachable.
|
||
expect(() => store.assertReachable(g0)).toThrow(GenerationCompactedError)
|
||
})
|
||
})
|
||
|
||
// ============================================================================
|
||
// 5. Deep-pin materialize — O(R) getDelta (not O(N·R)) + no mutex re-entrancy.
|
||
// ============================================================================
|
||
describe('materializeAtGeneration — bounded & deadlock-free (GA #33)', () => {
|
||
let brain: Brainy
|
||
|
||
beforeEach(async () => {
|
||
brain = new Brainy(createTestConfig())
|
||
await brain.init()
|
||
})
|
||
afterEach(async () => {
|
||
await brain.close()
|
||
})
|
||
|
||
it('resolveManyAt + readGenerationRecord do NOT re-enter the commit mutex (safe inside snapshotWith)', async () => {
|
||
const store = (brain as any).generationStore
|
||
const id = await brain.add({ data: 'a', type: NounType.Document, subtype: 'note', vector: VEC })
|
||
await brain.add({ data: 'b', type: NounType.Document, subtype: 'note', vector: VEC })
|
||
const pin = 1
|
||
|
||
// snapshotWith HOLDS the commit mutex. If resolveManyAt/readGenerationRecord
|
||
// took the mutex, this section would deadlock — assert it completes promptly.
|
||
const out = await withTimeout(
|
||
store.snapshotWith(async () => {
|
||
const many = await store.resolveManyAt('noun', new Set([id]), pin)
|
||
const rec = many.has(id) ? await store.readGenerationRecord('noun', many.get(id), id) : null
|
||
return { many, rec }
|
||
}),
|
||
3000,
|
||
'snapshotWith + resolveManyAt/readGenerationRecord'
|
||
)
|
||
expect(out.many).toBeInstanceOf(Map)
|
||
})
|
||
|
||
it('a deep-pin materialize over many ids invokes getDelta O(R) (not O(N·R)) and matches a per-id oracle', async () => {
|
||
const store = (brain as any).generationStore
|
||
|
||
const N = 400
|
||
for (let i = 0; i < N; i++) {
|
||
await brain.add({ data: `doc ${i}`, type: NounType.Document, subtype: 'note', metadata: { i }, vector: VEC })
|
||
}
|
||
const R = brain.generation() // ≈ N (each add is its own generation)
|
||
expect(R).toBeGreaterThanOrEqual(N)
|
||
|
||
const deepGen = 1
|
||
|
||
// Count getDelta invocations during the materialize.
|
||
const realGetDelta = store.getDelta.bind(store)
|
||
let getDeltaCalls = 0
|
||
store.getDelta = async (g: number) => {
|
||
getDeltaCalls++
|
||
return realGetDelta(g)
|
||
}
|
||
|
||
let handle: any
|
||
try {
|
||
handle = await withTimeout(
|
||
(brain as any).materializeAtGeneration(deepGen),
|
||
15000,
|
||
'deep materialize'
|
||
)
|
||
} finally {
|
||
store.getDelta = realGetDelta
|
||
}
|
||
|
||
// O(R): a small constant number of ascending passes (changedBetween +
|
||
// resolveManyAt per kind), NOT a per-id rescan (which would be ~N·R).
|
||
expect(getDeltaCalls).toBeLessThan(R * 5)
|
||
expect(getDeltaCalls).toBeLessThan(N * N) // the regression guard
|
||
|
||
// The materialized at-gen-1 brain holds exactly the one entity that existed.
|
||
const atGen1 = await handle.find({ limit: N + 10 })
|
||
expect(atGen1.length).toBe(1)
|
||
await handle.close()
|
||
})
|
||
|
||
it('a materialize completes (no deadlock) under a forced concurrent commit', async () => {
|
||
const N = 120
|
||
for (let i = 0; i < N; i++) {
|
||
await brain.add({ data: `x ${i}`, type: NounType.Document, subtype: 'note', vector: VEC })
|
||
}
|
||
const deepGen = 1
|
||
|
||
// Fire the materialize and several commits together: the reconciliation pass
|
||
// (under snapshotWith's mutex) must resolve the raced ids via the mutex-free
|
||
// bulk path without deadlocking.
|
||
const matPromise = withTimeout(
|
||
(brain as any).materializeAtGeneration(deepGen),
|
||
15000,
|
||
'materialize under concurrent commits'
|
||
)
|
||
const commits = Promise.all(
|
||
Array.from({ length: 5 }, (_, i) =>
|
||
brain.add({ data: `concurrent ${i}`, type: NounType.Document, subtype: 'note', vector: VEC })
|
||
)
|
||
)
|
||
const [handle] = await Promise.all([matPromise, commits])
|
||
expect(handle).toBeTruthy()
|
||
await handle.close()
|
||
})
|
||
})
|