open-brainy/tests/unit/db/fact-log-group-sync.test.ts

/**
 * @module tests/unit/db/fact-log-group-sync
 * @description Group commit on the fact log — the covering guarantee behind
 * durable-at-ack: concurrent callers of ensureSynced() share ONE covering
 * fsync (running + queued slots), a caller appending during a running sync
 * joins a sync that STARTS after its append (never the possibly-stale running
 * one), a solo writer syncs immediately, and at the brain level an at-ack
 * ack resolving means the write's fact is on disk.
 *
 * One pin is marked `.fails` (real finding, not a test bug): the at-ack
 * durability contract says an acked write's fact survives power loss, but
 * FactLog.open() truncates every fact beyond the store's committed
 * generation watermark — which only advances at the pending-tier flush. A
 * crash-shaped reopen (acks landed, flush never ran) therefore DISCARDS the
 * fsynced facts at open. See the test comment for the exact mechanism.
 */
import { describe, it, expect, beforeEach, afterEach } from 'vitest'
import { mkdtempSync, rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { Brainy } from '../../../src/index.js'
import { FileSystemStorage } from '../../../src/storage/adapters/fileSystemStorage.js'
import {
  FactLog,
  storageSupportsFactLog,
  type CommitFact,
  type FactLogStorage
} from '../../../src/db/factLog.js'

const UUID = (n: number): string =>
  `00000000-0000-4000-8000-${String(n).padStart(12, '0')}`

const fact = (generation: number): CommitFact => ({
  generation,
  timestamp: 1_700_000_000_000 + generation,
  ops: [
    {
      kind: 'noun',
      id: UUID(generation),
      record: { metadata: { noun: 'document', title: `doc ${generation}` }, vector: { v: [1, 2] } }
    }
  ]
})

/** Scan every fact from a FRESH reader log over the same directory. */
async function readBack(dir: string, committedHead: number): Promise<CommitFact[]> {
  const storage: any = new FileSystemStorage(dir)
  await storage.init()
  const reader = new FactLog(storage as FactLogStorage)
  await reader.open(committedHead)
  const facts: CommitFact[] = []
  const scan = reader.scanFacts()
  for await (const batch of scan.batches()) facts.push(...batch.facts)
  return facts
}

describe('fact log group commit — the covering fsync', () => {
  let dir: string
  let storage: any
  let log: FactLog

  beforeEach(async () => {
    dir = mkdtempSync(join(tmpdir(), 'brainy-group-sync-'))
    storage = new FileSystemStorage(dir)
    await storage.init()
    expect(storageSupportsFactLog(storage)).toBe(true)
    log = new FactLog(storage as FactLogStorage)
    await log.open(0)
  })

  afterEach(() => {
    rmSync(dir, { recursive: true, force: true })
  })

  it('many concurrent ensureSynced() callers share one covering fsync — every caller resolves, batching happened', async () => {
    for (let g = 1; g <= 10; g++) await log.append(fact(g))

    // Count REAL fsync batches at the storage boundary, with a small delay so
    // the concurrent callers genuinely overlap the running sync.
    let fsyncBatches = 0
    const origSync = storage.syncRawObjects.bind(storage)
    storage.syncRawObjects = async (paths: string[]) => {
      fsyncBatches++
      await new Promise((r) => setTimeout(r, 15))
      return origSync(paths)
    }

    const callers = Array.from({ length: 10 }, () => log.ensureSynced())
    await Promise.all(callers) // every caller resolves — no lost writer

    expect(fsyncBatches, 'callers shared a covering fsync').toBeLessThan(10)
    expect(fsyncBatches).toBeGreaterThanOrEqual(1)

    // Durable: a fresh reader over the same directory sees all 10 facts.
    const facts = await readBack(dir, 10)
    expect(facts.map((f) => f.generation)).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
  })

  it('an append during a RUNNING sync is covered by a sync that starts after it — never the stale running one', async () => {
    for (let g = 1; g <= 3; g++) await log.append(fact(g))

    // Gate the FIRST fsync so a sync is provably in flight.
    let fsyncBatches = 0
    let releaseGate!: () => void
    const gate = new Promise<void>((r) => {
      releaseGate = r
    })
    let gated = true
    const origSync = storage.syncRawObjects.bind(storage)
    storage.syncRawObjects = async (paths: string[]) => {
      fsyncBatches++
      if (gated) {
        gated = false
        await gate
      }
      return origSync(paths)
    }

    const p1 = log.ensureSynced() // sync A: snapshots gens 1..3, blocks in fsync
    await new Promise((r) => setTimeout(r, 10))
    expect(fsyncBatches, 'sync A is in flight').toBe(1)

    await log.append(fact(4)) // lands AFTER sync A snapshotted
    let p2Resolved = false
    const p2 = log.ensureSynced().then(() => {
      p2Resolved = true
    })

    // The covering guarantee: p2 must NOT resolve off the running sync (it
    // may have snapshotted before the append) — it waits for the queued one.
    await new Promise((r) => setTimeout(r, 25))
    expect(p2Resolved, 'p2 never joins the possibly-stale running sync').toBe(false)

    releaseGate()
    await p1
    await p2
    expect(p2Resolved).toBe(true)
    expect(fsyncBatches, 'the queued covering sync ran after the running one').toBe(2)

    // The late append is durable once p2 resolved.
    const facts = await readBack(dir, 4)
    expect(facts.map((f) => f.generation)).toEqual([1, 2, 3, 4])
  })

  it('a solo writer syncs immediately — one fsync, and a dirty-free ensureSynced adds none', async () => {
    // Count only covering syncs: the first append itself fsyncs the tail
    // manifest (the manifest-first flip), so instrument AFTER it.
    await log.append(fact(1))
    let fsyncBatches = 0
    const origSync = storage.syncRawObjects.bind(storage)
    storage.syncRawObjects = async (paths: string[]) => {
      fsyncBatches++
      return origSync(paths)
    }

    await log.ensureSynced()
    expect(fsyncBatches).toBe(1)

    // Nothing new appended: the covering sync finds nothing dirty.
    await log.ensureSynced()
    expect(fsyncBatches).toBe(1)
  })
})

describe('durable-at-ack through the brain (group commit end-to-end)', () => {
  const dirs: string[] = []
  const brains: any[] = []

  const openBrain = async (dir?: string): Promise<{ brain: any; dir: string }> => {
    process.env.BRAINY_DETERMINISTIC_EMBEDDINGS = 'true'
    const d = dir ?? mkdtempSync(join(tmpdir(), 'brainy-at-ack-'))
    if (!dir) dirs.push(d)
    const brain: any = new Brainy({
      storage: { type: 'filesystem', path: d },
      requireSubtype: false,
      silent: true,
      dimensions: 384
    })
    brains.push(brain)
    await brain.init()
    return { brain, dir: d }
  }

  afterEach(async () => {
    for (const b of brains.splice(0)) await b.close?.().catch(() => {})
    for (const d of dirs.splice(0)) rmSync(d, { recursive: true, force: true })
  })

  it('at-ack: N concurrent add() acks all resolve, every ack was covered by a log sync, and every fact is on disk after reopen', async () => {
    const { brain, dir } = await openBrain()
    // White-box: engage the at-ack durability mode directly (the guarded
    // authority flip that normally enables it is covered by the integration
    // suite — this test pins the durability machinery itself).
    brain.generationStore.setLogDurability('at-ack')

    const factLog = brain.generationStore.getFactLog()
    expect(factLog).not.toBeNull()
    let syncs = 0
    const origSync = factLog.sync.bind(factLog)
    factLog.sync = async () => {
      syncs++
      return origSync()
    }

    const ids: string[] = await Promise.all(
      Array.from({ length: 10 }, (_, i) =>
        brain.add({ data: `concurrent write ${i}`, type: 'document', metadata: { i } })
      )
    )
    expect(new Set(ids).size, 'every ack resolved with a distinct id').toBe(10)
    // Honest pin: single-op acks serialize under the commit mutex (append +
    // covering sync run inside it), so concurrent add() acks do not currently
    // share one fsync — cross-writer batching is the FactLog-layer property
    // pinned above. What must hold here: at least one covering sync ran, and
    // no ack resolved without the machinery engaged.
    expect(syncs).toBeGreaterThanOrEqual(1)
    expect(syncs).toBeLessThanOrEqual(10)

    await brain.close()
    const { brain: reopened } = await openBrain(dir)
    const scan = reopened.scanFacts()
    expect(scan).not.toBeNull()
    const liveFactIds = new Set<string>()
    for await (const batch of scan!.batches()) {
      for (const f of batch.facts) {
        for (const op of f.ops) if (op.kind === 'noun' && op.record !== null) liveFactIds.add(op.id)
      }
    }
    for (const id of ids) {
      expect(liveFactIds.has(id), `fact for acked write ${id} survives reopen`).toBe(true)
    }
  })

  // KNOWN GAP (marked .fails — remove the marker when fixed in src): the
  // at-ack contract is that an acked write's fact survives power loss. The
  // fsync at ack does put the fact's bytes on disk — but FactLog.open()
  // truncates every fact with generation > the store's committed watermark,
  // and that watermark only advances at the pending-tier flush
  // (flushPendingSingleOps). So on a crash-shaped reopen (acks landed, flush
  // never ran) the store logs "[FactLog] truncating N uncommitted fact(s)"
  // and DISCARDS the acked, fsynced facts. Until recovery treats the log as
  // authoritative past the tree's watermark (or the watermark goes durable
  // at ack), durable-at-ack does not survive the very crash it exists for.
  it.fails('at-ack CONTRACT: acked facts survive a crash-shaped reopen (no flush ever ran)', async () => {
    const { brain, dir } = await openBrain()
    brain.generationStore.setLogDurability('at-ack')
    // Crash simulation: the pending-tier durability flush never happens
    // (every trigger routes through flushPendingSingleOps), and the brain is
    // abandoned without close() — exactly the power-loss shape at-ack is for.
    brain.generationStore.flushPendingSingleOps = async () => {}

    const ids: string[] = []
    for (let i = 0; i < 5; i++) {
      ids.push(await brain.add({ data: `acked write ${i}`, type: 'document', metadata: { i } }))
    }

    // No flush, no close — reopen the directory as a new session.
    const { brain: reopened } = await openBrain(dir)
    const scan = reopened.scanFacts()
    expect(scan).not.toBeNull()
    const liveFactIds = new Set<string>()
    for await (const batch of scan!.batches()) {
      for (const f of batch.facts) {
        for (const op of f.ops) if (op.kind === 'noun' && op.record !== null) liveFactIds.add(op.id)
      }
    }
    for (const id of ids) {
      expect(liveFactIds.has(id), `acked fact ${id} survives the crash-shaped reopen`).toBe(true)
    }
  })
})