open-brainy/tests/helpers/durabilityKillMatrix.ts

/**
 * @module tests/helpers/durabilityKillMatrix
 * @description Shared machinery for the durability kill-matrix suite
 * (tests/integration/durability-kill-matrix.test.ts): open filesystem brains
 * with fully explicit durability (no background cadence, no embedder), arm
 * the generation store's test-only commit fault injector at one exact phase,
 * abandon a "crashed" brain the way a dead process would (its RAM is gone,
 * nothing flushes, nothing closes), and read the fact log / on-disk state the
 * recovery assertions pin.
 *
 * The crash model is PROCESS DEATH: in-memory state is lost, file bytes the
 * process already handed to the OS survive. One helper additionally models
 * POWER LOSS for a chosen entity by removing its canonical files — legal,
 * because single-op canonical writes are tmp+rename WITHOUT fsync, and a
 * rename that was never fsynced may surface as "no directory entry" after
 * power loss.
 */
import * as fs from 'node:fs'
import * as os from 'node:os'
import * as path from 'node:path'
import { Brainy } from '../../src/brainy.js'
import type { CommitFaultPhase, GenerationStore } from '../../src/db/generationStore.js'

/** The error a throwing fault injector uses to simulate a process crash. */
export class SimulatedCrash extends Error {
  constructor(phase: CommitFaultPhase) {
    super(`simulated process crash at ${phase}`)
    this.name = 'SimulatedCrash'
  }
}

/** Deterministic 384-dim vector so no test ever invokes the embedder. */
export function vec(seed: number): number[] {
  return Array.from({ length: 384 }, (_, i) => ((seed * 31 + i * 7) % 100) / 100)
}

/**
 * Map a readable label to a deterministic UUID-shaped id (entity ids must be
 * UUIDs — the sharded storage layout derives the shard from the UUID hex).
 */
export function uid(label: string): string {
  let h1 = 0x811c9dc5
  for (let i = 0; i < label.length; i++) {
    h1 = Math.imul(h1 ^ label.charCodeAt(i), 0x01000193) >>> 0
  }
  let h2 = 0xdeadbeef
  for (let i = label.length - 1; i >= 0; i--) {
    h2 = Math.imul(h2 ^ label.charCodeAt(i), 0x85ebca6b) >>> 0
  }
  const hex = h1.toString(16).padStart(8, '0') + h2.toString(16).padStart(8, '0')
  return `00000000-0000-4000-8000-${hex.slice(0, 12)}`
}

/** Create a fresh temp directory for one brain's storage root. */
export function makeTempDir(): string {
  return fs.mkdtempSync(path.join(os.tmpdir(), 'brainy-kill-matrix-'))
}

/**
 * Open a writer brain over `dir` with every implicit durability knob off:
 * persistence policy 'manual' (the engine never flushes on its own, so every
 * durable transition in a test is an explicit `flush()`/commit), deterministic
 * embeddings (tests always pass explicit vectors anyway), silent logs — and
 * `logAuthority: 'defer'` (the explicit opt-out of the 10.0.0 adopt-at-open
 * fleet default), so the durability POSTURE is explicit per row too: rows
 * pinning deferred/tree recovery semantics get exactly that, and at-ack rows
 * engage log authority via `flipToAtAck`. The fleet default's open-time
 * adoption would inject a baseline-backfill generation into every floor
 * computation and pre-flip every row.
 */
export async function openBrain(
  dir: string,
  opts?: { logAuthority?: 'adopt' | 'defer' }
): Promise<Brainy> {
  process.env.BRAINY_DETERMINISTIC_EMBEDDINGS = 'true'
  const brain = new Brainy({
    requireSubtype: false,
    storage: { type: 'filesystem', path: dir },
    silent: true,
    persistence: { policy: 'manual' },
    logAuthority: opts?.logAuthority ?? 'defer'
  })
  await brain.init()
  return brain
}

/** Typed access to the brain's private generation store (test injection point). */
export function storeOf(brain: Brainy): GenerationStore {
  return (brain as unknown as { generationStore: GenerationStore }).generationStore
}

/**
 * Arm the commit fault injector to simulate a process crash at EXACTLY one
 * phase (all other phases pass through untouched). Returns the list of phases
 * observed before (and including) the trip, so a test can assert the fault
 * actually fired where intended.
 */
export function armCrash(brain: Brainy, phase: CommitFaultPhase): { fired: CommitFaultPhase[] } {
  const fired: CommitFaultPhase[] = []
  storeOf(brain).setCommitFaultInjector((p) => {
    fired.push(p)
    if (p === phase) {
      throw new SimulatedCrash(p)
    }
  })
  return { fired }
}

/**
 * Abandon a crashed brain the way process death would: its buffered RAM state
 * is discarded and no background machinery may ever touch the storage
 * directory again (a dead process cannot flush). The fault injector stays
 * installed so any in-flight commit path still "crashes". Serialized behind
 * the store's commit mutex so an interleaved background flush cannot be
 * severed mid-section.
 *
 * NEVER calls close() — graceful close is exactly what a crash denies.
 */
export async function abandonAsCrashed(brain: Brainy): Promise<void> {
  const store = storeOf(brain) as unknown as {
    withMutex<R>(fn: () => Promise<R>): Promise<R>
    clearPendingFlushTimer(): void
    pendingGens: number[]
    pendingBuffer: Map<number, unknown>
  }
  await store.withMutex(async () => {
    store.clearPendingFlushTimer()
    store.pendingGens = []
    store.pendingBuffer.clear()
  })
}

/**
 * Every generation present in the brain's fact log, ascending — the suite's
 * "what does the log claim is committed" probe. Empty when no fact log exists.
 * A scan abort (gap detection) propagates — callers that PIN gap behavior
 * catch it themselves.
 */
export async function factGenerations(brain: Brainy): Promise<number[]> {
  const scan = brain.scanFacts({ fromGeneration: 1 })
  if (!scan) return []
  const gens: number[] = []
  for await (const batch of scan.batches()) {
    for (const fact of batch.facts) gens.push(fact.generation)
  }
  return gens.sort((a, b) => a - b)
}

/** An ENOSPC-shaped error, matching what a full disk surfaces from node:fs. */
export function enospcError(): NodeJS.ErrnoException {
  const err = new Error("ENOSPC: no space left on device, write") as NodeJS.ErrnoException
  err.code = 'ENOSPC'
  err.errno = -28
  err.syscall = 'write'
  return err
}

/**
 * Make the storage adapter's next raw-byte append (the fact-log append path)
 * fail once with ENOSPC, then restore the original — "the disk filled for one
 * append, then space was freed". Returns a probe telling how many appends
 * were failed.
 */
export function failNextAppendWithEnospc(brain: Brainy): { failed: () => number } {
  const storage = (brain as unknown as {
    storage: { appendRawBytes(p: string, b: Uint8Array): Promise<void> }
  }).storage
  const original = storage.appendRawBytes.bind(storage)
  let failures = 0
  storage.appendRawBytes = async (p: string, b: Uint8Array): Promise<void> => {
    storage.appendRawBytes = original
    failures++
    throw enospcError()
  }
  return { failed: () => failures }
}

/**
 * POWER-LOSS MODEL for one entity: remove its canonical noun files from the
 * storage root. Legal disk state — a single-op write's canonical bytes are
 * tmp+rename WITHOUT fsync (only `transact()` runs the write barrier), and an
 * un-fsynced rename may resolve to "no directory entry" after power loss.
 * Throws when nothing was removed (the caller's premise would be wrong).
 */
export function dropCanonicalNoun(dir: string, id: string): void {
  const removed: string[] = []
  const walk = (p: string): void => {
    for (const entry of fs.readdirSync(p, { withFileTypes: true })) {
      const full = path.join(p, entry.name)
      if (entry.isDirectory()) {
        if (entry.name === id) {
          fs.rmSync(full, { recursive: true, force: true })
          removed.push(full)
        } else {
          walk(full)
        }
      }
    }
  }
  const nounsRoot = path.join(dir, 'entities', 'nouns')
  if (fs.existsSync(nounsRoot)) walk(nounsRoot)
  if (removed.length === 0) {
    throw new Error(`power-loss model: no canonical files found for noun ${id} under ${nounsRoot}`)
  }
}

/** True when the staged record-set directory for `gen` exists on disk. */
export function generationDirExists(dir: string, gen: number): boolean {
  return fs.existsSync(path.join(dir, '_generations', String(gen)))
}