open-brainy/tests/unit/db/bounded-chains.test.ts

/**
 * @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
    // Relative, not absolute: under the adopt-at-open default the open-time
    // baseline backfill takes a generation of its own, so the first add is
    // NOT generation 1 — pin the deep generation to the first add's commit.
    let deepGen = 0
    for (let i = 0; i < N; i++) {
      await brain.add({ data: `doc ${i}`, type: NounType.Document, subtype: 'note', metadata: { i }, vector: VEC })
      if (i === 0) deepGen = brain.generation()
    }
    const R = brain.generation() // ≈ N (each add is its own generation)
    expect(R).toBeGreaterThanOrEqual(N)

    // 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 brain at the first add's generation holds exactly the
    // one user 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()
  })
})