open-brainy/tests/integration/fold-checkpoint-bound.test.ts

/**
 * @module tests/integration/fold-checkpoint-bound
 * @description The fold-checkpoint bound (crash recovery's log fold, bounded):
 * `_system/fold-checkpoint.json` at generation G asserts every entity whose
 * latest fact is ≤ G has DURABLE canonical bytes — each stamp strictly follows
 * a canonical-sync barrier over every live entity touched since the last one
 * (stamp-after-data). An unclean open then folds only `(G, head]` instead of
 * the whole log. These pins prove the four load-bearing properties:
 *
 *   1. The stamp exists and tracks the committed watermark (flush + close).
 *   2. The fold is genuinely BOUNDED — facts ≤ G are skipped — while facts in
 *      `(G, head]` are re-applied even BELOW the manifest.
 *   3. A failed barrier NEVER advances the stamp (the bound can lag, growing
 *      a later fold — it can never overstate durability, losing a write).
 *   4. A pre-checkpoint brain (the 10.0 shape) bootstraps its chain at its
 *      first whole-log fold; a tree-authority brain never stamps at all.
 */
import { describe, it, expect, afterEach, vi } from 'vitest'
import * as fs from 'node:fs'
import * as zlib from 'node:zlib'
import { join } from 'node:path'
import { Brainy } from '../../src/brainy.js'
import { NounType } from '../../src/types/graphTypes.js'
import {
  abandonAsCrashed,
  armCrash,
  dropCanonicalNoun,
  makeTempDir,
  openBrain,
  storeOf
} from '../helpers/durabilityKillMatrix.js'

const CHECKPOINT = join('_system', 'fold-checkpoint.json')

/** Read the fold-checkpoint artifact's generation from disk, or null. */
function readCheckpoint(dir: string): number | null {
  for (const candidate of [join(dir, `${CHECKPOINT}.gz`), join(dir, CHECKPOINT)]) {
    if (!fs.existsSync(candidate)) continue
    const raw = fs.readFileSync(candidate)
    const text = candidate.endsWith('.gz') ? zlib.gunzipSync(raw).toString('utf8') : raw.toString('utf8')
    const parsed = JSON.parse(text) as { generation?: number }
    return Number.isSafeInteger(parsed.generation) ? (parsed.generation as number) : null
  }
  return null
}

function removeArtifact(dir: string, rel: string): void {
  for (const candidate of [join(dir, `${rel}.gz`), join(dir, rel)]) {
    fs.rmSync(candidate, { force: true })
  }
}

function committedOf(brain: Brainy): number {
  return (storeOf(brain) as unknown as { committed: number }).committed
}

describe('fold-checkpoint bound — crash recovery folds (checkpoint, head], never less durability than stamped', () => {
  const dirs: string[] = []
  const liveBrains: Brainy[] = []
  afterEach(async () => {
    vi.restoreAllMocks()
    for (const b of liveBrains.splice(0)) await b.close().catch(() => {})
    for (const d of dirs.splice(0)) fs.rmSync(d, { recursive: true, force: true })
  })
  function trackDir(): string {
    const dir = makeTempDir()
    dirs.push(dir)
    return dir
  }

  it('a fresh adopt brain stamps at flush and again at close — the stamp tracks the committed watermark', async () => {
    const dir = trackDir()
    const brain = await openBrain(dir, { logAuthority: 'adopt' })
    liveBrains.push(brain)
    expect(brain.logAuthority().authority).toBe('log')

    await brain.add({ data: 'first', type: NounType.Document, metadata: { n: 1 } })
    await brain.add({ data: 'second', type: NounType.Document, metadata: { n: 2 } })
    await brain.flush()
    const afterFlush = readCheckpoint(dir)
    expect(afterFlush).toBe(committedOf(brain))
    expect(afterFlush!).toBeGreaterThan(0)

    await brain.add({ data: 'third', type: NounType.Document, metadata: { n: 3 } })
    const closingCommit = liveBrains.pop()!
    await closingCommit.close()
    // Close flushes, so the stamp advanced with it — and the clean-shutdown
    // marker it writes afterward never vouches for bytes the stamp has not.
    expect(readCheckpoint(dir)).toBeGreaterThanOrEqual(afterFlush!)
  }, 120000)

  it('BOUNDED fold: facts ≤ checkpoint are skipped, facts in (checkpoint, head] are re-applied even below the manifest; a failed barrier retains the old bound', async () => {
    const dir = trackDir()
    const brain = await openBrain(dir, { logAuthority: 'adopt' })
    liveBrains.push(brain)

    // Window 1 — flushed and stamped: the checkpoint's covered past.
    const idA = await brain.add({ data: 'covered by the stamp', type: NounType.Document, metadata: { w: 1 } })
    await brain.flush()
    const checkpoint1 = readCheckpoint(dir)
    expect(checkpoint1).toBe(committedOf(brain))

    // Window 2 — committed BELOW a new manifest but with the checkpoint stamp
    // FAILING: the barrier throws once, so the manifest advances while the
    // stamp stays at checkpoint1 (pin 3: a failed barrier never advances it).
    const storage = (brain as unknown as {
      storage: { syncEntityCanonical(n: string[], v: string[]): Promise<void> }
    }).storage
    const realBarrier = storage.syncEntityCanonical.bind(storage)
    let failedOnce = false
    vi.spyOn(storage, 'syncEntityCanonical').mockImplementation(async (n: string[], v: string[]) => {
      if (!failedOnce) {
        failedOnce = true
        throw new Error('injected barrier failure (device hiccup)')
      }
      return realBarrier(n, v)
    })
    const idB = await brain.add({ data: 'below manifest, above checkpoint', type: NounType.Document, metadata: { w: 2 } })
    await brain.flush()
    expect(failedOnce).toBe(true)
    expect(readCheckpoint(dir)).toBe(checkpoint1) // stamp did NOT advance
    expect(committedOf(brain)).toBeGreaterThan(checkpoint1!) // manifest DID

    // Crash. Vaporize BOTH canonical records: idB's fact lives in
    // (checkpoint, manifest] — the bounded fold MUST restore it; idA's fact
    // is ≤ checkpoint — the fold must SKIP it (its loss here is synthetic:
    // the stamp's barrier fsynced it, a power cut cannot take it, and the
    // skip is exactly what makes the fold bounded instead of whole-log).
    await abandonAsCrashed(liveBrains.pop()!)
    dropCanonicalNoun(dir, idA)
    dropCanonicalNoun(dir, idB)

    const reopened = await openBrain(dir, { logAuthority: 'adopt' })
    liveBrains.push(reopened)
    const restoredB = await reopened.get(idB)
    expect(restoredB, 'a fact above the checkpoint is re-applied even below the manifest').not.toBeNull()
    const skippedA = await reopened.get(idA)
    expect(skippedA, 'a fact at-or-below the checkpoint is outside the fold — the bound is real').toBeNull()
    // And recovery re-stamped at its new committed watermark.
    expect(readCheckpoint(dir)).toBe(committedOf(reopened))
  }, 120000)

  it('a pre-checkpoint brain (the 10.0 shape) folds the WHOLE log once, then its chain is established', async () => {
    const dir = trackDir()
    const brain = await openBrain(dir, { logAuthority: 'adopt' })
    liveBrains.push(brain)
    const idA = await brain.add({ data: 'ten-point-oh resident', type: NounType.Document, metadata: { era: '10.0' } })
    await brain.flush()
    await liveBrains.pop()!.close()

    // Rewind the brain to the 10.0 shape: no checkpoint artifact, and an
    // unclean shutdown (marker gone) — exactly what an existing fleet brain
    // looks like at its first crash under 10.1.
    removeArtifact(dir, CHECKPOINT)
    removeArtifact(dir, join('_system', 'clean-shutdown.json'))
    dropCanonicalNoun(dir, idA)

    const reopened = await openBrain(dir, { logAuthority: 'adopt' })
    liveBrains.push(reopened)
    expect(await reopened.get(idA), 'no checkpoint ⇒ whole-log fold ⇒ every acked write restored').not.toBeNull()
    const stamped = readCheckpoint(dir)
    expect(stamped, 'the first whole-log fold is the chain’s base case — it stamps').toBe(committedOf(reopened))
  }, 120000)

  it('ARM-AT-FLIP: a non-fresh adoption founds the checkpoint immediately — the first post-flip boot folds BOUNDED, never whole-log', async () => {
    const dir = trackDir()
    // The production shape: a brain with history flips LIVE (no crash ever).
    const brain = await openBrain(dir, { logAuthority: 'defer' })
    liveBrains.push(brain)
    const preFlip = await brain.add({ data: 'pre-flip resident', type: NounType.Document, metadata: { era: 'tree' } })
    await brain.flush()
    expect(readCheckpoint(dir), 'no checkpoint before the flip').toBeNull()

    const report = await brain.adoptLogAuthority()
    expect(report.verdict).toBe('green')
    // THE PIN: the flip itself founded the checkpoint — no crash required.
    const founded = readCheckpoint(dir)
    expect(founded, 'checkpoint founded at flip').toBe(committedOf(brain))

    // First post-flip boot, unclean (the production first-restart shape):
    // a post-flip write above the checkpoint is restored FROM ITS AT-ACK FACT;
    // the pre-flip row (its baseline fact ≤ checkpoint, its bytes barrier-
    // synced at the flip) is OUTSIDE the fold — vaporizing it synthetically
    // proves the bound.
    //
    // THE CRASH IS ARMED, NOT RACED. The post-flip write "dies" at exactly
    // `singleop-after-fact-append`: its fact is in the log and at-ack synced,
    // and NO pending flush was ever scheduled — so the checkpoint provably
    // still reads the flip's stamp when the bytes are dropped. The earlier
    // shape (`add()` then abandon) raced the store's 50ms pending-flush
    // timer: on a loaded box the flush won, barrier-synced the row, advanced
    // the stamp over it — and the fold, CORRECTLY bounded, did not restore
    // bytes the test had synthetically destroyed after they were stamped
    // durable. The plant lane caught it; the engine was right, the pin was
    // timing-dependent.
    const postFlip = `post-flip-${Date.now().toString(36)}-0000-4000-8000-000000000000`
    const arm = armCrash(brain, 'singleop-after-fact-append')
    await expect(
      brain.add({ id: postFlip, data: 'post-flip write', type: NounType.Document, metadata: { era: 'log' } })
    ).rejects.toThrow('simulated process crash at singleop-after-fact-append')
    expect(arm.fired).toContain('singleop-after-fact-append')
    expect(readCheckpoint(dir), 'the stamp did not move — nothing flushed after the flip').toBe(founded)
    await abandonAsCrashed(liveBrains.pop()!)
    // The post-flip row's canonical bytes lived only in the pending tier's
    // RAM (written at flush, never reached) — the crash takes them for real;
    // nothing to drop. Only the pre-flip row is vaporized synthetically.
    dropCanonicalNoun(dir, preFlip)

    const reopened = await openBrain(dir, { logAuthority: 'adopt' })
    liveBrains.push(reopened)
    expect(await reopened.get(postFlip), 'above-checkpoint fact re-applied').not.toBeNull()
    expect(await reopened.get(preFlip), 'below-checkpoint fact skipped — the fold is bounded on the FIRST post-flip boot').toBeNull()
  }, 240000)

  it('a tree-authority brain never stamps a checkpoint', async () => {
    const dir = trackDir()
    const brain = await openBrain(dir, { logAuthority: 'defer' })
    liveBrains.push(brain)
    expect(brain.logAuthority().authority).not.toBe('log')
    await brain.add({ data: 'tree resident', type: NounType.Document, metadata: { n: 1 } })
    await brain.flush()
    await liveBrains.pop()!.close()
    expect(readCheckpoint(dir)).toBeNull()
  }, 120000)

  it('restore is an UNCLEAN event: the snapshot’s stamps do not survive — the reopen fold re-founds and re-stamps the restored state', async () => {
    const dir = trackDir()
    const brain = await openBrain(dir, { logAuthority: 'adopt' })
    liveBrains.push(brain)
    const idA = await brain.add({ data: 'survives the restore', type: NounType.Document, metadata: { n: 1 } })
    await brain.flush()

    const snapDir = join(trackDir(), 'snap')
    const db = brain.now()
    await (db as unknown as { persist(p: string): Promise<void> }).persist(snapDir)
    await (db as unknown as { release(): Promise<void> }).release()

    // Advance the live brain past the snapshot: a later write, a later flush,
    // a later checkpoint stamp — none of which may survive the restore.
    const idB = await brain.add({ data: 'must not survive', type: NounType.Document, metadata: { n: 2 } })
    await brain.flush()
    const stampBeforeRestore = readCheckpoint(dir)
    expect(stampBeforeRestore).toBe(committedOf(brain))

    // Unflushed traffic in flight at restore time — the quiesced swap discards
    // it under the mutex instead of letting its flush timer race the swap
    // (the ENOTEMPTY class).
    await brain.add({ data: 'in-flight at restore', type: NounType.Document, metadata: { n: 3 } })
    await brain.restore(snapDir, { confirm: true })

    expect(await brain.get(idA), 'snapshot state restored').not.toBeNull()
    expect(await brain.get(idB), 'post-snapshot state replaced').toBeNull()
    // The stamp on disk is the REOPEN FOLD's fresh assertion about the
    // restored (and now barrier-synced) bytes — at the restored watermark,
    // strictly below the pre-restore stamp that must not survive.
    const stampAfterRestore = readCheckpoint(dir)
    expect(stampAfterRestore).toBe(committedOf(brain))
    expect(stampAfterRestore!).toBeLessThan(stampBeforeRestore!)
  }, 120000)

  it('a delete rides the barrier: the tombstoned id is in the synced set and the stamp advances past it', async () => {
    const dir = trackDir()
    const brain = await openBrain(dir, { logAuthority: 'adopt' })
    liveBrains.push(brain)
    const id = await brain.add({ data: 'short-lived', type: NounType.Document, metadata: { n: 1 } })
    await brain.flush()

    const storage = (brain as unknown as {
      storage: { syncEntityCanonical(n: string[], v: string[]): Promise<void> }
    }).storage
    const seen: string[][] = []
    const realBarrier = storage.syncEntityCanonical.bind(storage)
    vi.spyOn(storage, 'syncEntityCanonical').mockImplementation(async (n: string[], v: string[]) => {
      seen.push([...n])
      return realBarrier(n, v)
    })

    await brain.remove(id)
    await brain.flush()
    expect(
      seen.some((nouns) => nouns.includes(id)),
      'the deleted id must reach the canonical barrier (absence is durable state too)'
    ).toBe(true)
    expect(readCheckpoint(dir)).toBe(committedOf(brain))
  }, 120000)
})