open-brainy/tests/integration/write-flow-production-shape.test.ts

/**
 * @module tests/integration/write-flow-production-shape
 * @description The production-shaped WRITE-FLOW gate leg. A downstream
 * deployment's release gate went all-green on snapshots and rehearsal reads
 * while two write-path defects (pad-frame constructibility, a counter rewind
 * after a successful append) waited in ordinary WRITE flows — deferred
 * embedding retries plus background history-flush concurrency wearing the
 * stacks. This leg runs that exact shape, permanently:
 *
 *   - concurrent mixed writes (adds, deferred-embed adds, updates, removes)
 *   - racing explicit flushes (the history tier's group commit, mid-traffic)
 *   - then the three laws: every ack is readable truth, the fact log is
 *     STRICTLY ascending end-to-end, and no write is ever refused.
 *
 * Part two crashes the brain mid-traffic (no close — RAM discarded) and
 * requires every acked write back after reopen: the at-ack contract under
 * the same production shape, not under a synthetic single write.
 */
import { describe, it, expect, afterEach } from 'vitest'
import * as fs from 'node:fs'
import { Brainy } from '../../src/brainy.js'
import { NounType } from '../../src/types/graphTypes.js'
import {
  abandonAsCrashed,
  factGenerations,
  makeTempDir,
  openBrain
} from '../helpers/durabilityKillMatrix.js'

describe('write-flow production shape — the pair gate leg from a consumer-reported miss', () => {
  const dirs: string[] = []
  const liveBrains: Brainy[] = []
  afterEach(async () => {
    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
  }

  async function runTrafficWave(
    brain: Brainy,
    wave: number,
    perWave: number
  ): Promise<{ kept: string[]; removed: string[] }> {
    const kept: string[] = []
    const removed: string[] = []
    const work: Promise<unknown>[] = []
    for (let i = 0; i < perWave; i++) {
      const n = wave * perWave + i
      if (i % 4 === 0) {
        // Deferred-embed add — the retry-marker flow that wore the defect.
        work.push(
          brain
            .add({ data: `deferred payload ${n}`, type: NounType.Document, metadata: { n, defer: true }, deferEmbedding: true })
            .then((id) => void kept.push(id))
        )
      } else if (i % 4 === 1) {
        // Add, then update it in the same wave (two generations, same id).
        work.push(
          brain.add({ data: `versioned payload ${n}`, type: NounType.Document, metadata: { n, v: 1 } }).then(async (id) => {
            kept.push(id)
            await brain.update({ id, metadata: { n, v: 2 } })
          })
        )
      } else if (i % 4 === 2) {
        // Add, then remove — a durable tombstone is an ack too.
        work.push(
          brain.add({ data: `ephemeral payload ${n}`, type: NounType.Document, metadata: { n } }).then(async (id) => {
            await brain.remove(id)
            removed.push(id)
          })
        )
      } else {
        work.push(
          brain.add({ data: `plain payload ${n}`, type: NounType.Document, metadata: { n } }).then((id) => void kept.push(id))
        )
      }
      // Race the history tier's group commit against live traffic.
      if (i % 5 === 3) work.push(brain.flush())
    }
    // NO REFUSALS: every promise must resolve — a single rejection here is
    // the refusal-loop costume this leg exists to catch.
    await Promise.all(work)
    return { kept, removed }
  }

  it('three waves of mixed traffic with racing flushes: every ack is truth, the log is strictly ascending, nothing refused', async () => {
    const dir = trackDir()
    const brain = await openBrain(dir, { logAuthority: 'adopt' })
    liveBrains.push(brain)
    expect(brain.logAuthority().authority).toBe('log')

    const kept: string[] = []
    const removed: string[] = []
    for (let wave = 0; wave < 3; wave++) {
      const result = await runTrafficWave(brain, wave, 20)
      kept.push(...result.kept)
      removed.push(...result.removed)
    }
    await brain.flush()

    for (const id of kept) {
      expect(await brain.get(id), `acked write ${id} must be readable truth`).not.toBeNull()
    }
    for (const id of removed) {
      expect(await brain.get(id), `acked remove ${id} must hold`).toBeNull()
    }

    const gens = await factGenerations(brain)
    expect(gens.length).toBeGreaterThan(0)
    for (let i = 1; i < gens.length; i++) {
      expect(gens[i], 'fact log strictly ascending end-to-end').toBeGreaterThan(gens[i - 1])
    }

    // Clean reopen: the same truth survives a restart.
    await liveBrains.pop()!.close()
    const reopened = await openBrain(dir, { logAuthority: 'adopt' })
    liveBrains.push(reopened)
    for (const id of kept.slice(0, 10)) {
      expect(await reopened.get(id)).not.toBeNull()
    }
  }, 240000)

  it('crash mid-traffic: every acked write survives the reopen (the at-ack law under the production shape)', async () => {
    const dir = trackDir()
    const brain = await openBrain(dir, { logAuthority: 'adopt' })
    liveBrains.push(brain)

    const { kept, removed } = await runTrafficWave(brain, 0, 24)
    // No close, no flush — the process "dies" holding its RAM.
    await abandonAsCrashed(liveBrains.pop()!)

    const reopened = await openBrain(dir, { logAuthority: 'adopt' })
    liveBrains.push(reopened)
    for (const id of kept) {
      expect(await reopened.get(id), `acked write ${id} must survive the crash`).not.toBeNull()
    }
    for (const id of removed) {
      expect(await reopened.get(id), `acked remove ${id} must survive the crash`).toBeNull()
    }
    const gens = await factGenerations(reopened)
    for (let i = 1; i < gens.length; i++) {
      expect(gens[i], 'fact log strictly ascending after recovery').toBeGreaterThan(gens[i - 1])
    }
  }, 240000)
})