open-brainy/tests/integration/transact-durability-barrier.test.ts

/**
 * @module tests/integration/transact-durability-barrier
 * @description Regression for a consumer-reported durability gap: a committed
 * `transact()` reported success while its canonical entity writes were still
 * only in the page cache (tmp+rename, not yet fsync'd), whereas the generation
 * counter/manifest WERE fsync'd — so a hard kill could leave the counter ahead
 * of the persisted entity bytes ("phantom progress" for any generation-based
 * consumer resuming from the counter).
 *
 * Fix: `commitTransaction` opens a write barrier before running the planned
 * operations and flushes it (fsync of every canonical write + the parent dir of
 * every canonical delete) BEFORE advancing the counter and writing the manifest.
 * These tests assert the ordering property directly by recording the storage's
 * `syncRawObjects` calls: the entity writes are fsync'd in an earlier call than
 * the manifest, and a precommit-rejected batch opens no barrier and advances
 * nothing.
 */
import { describe, it, expect, beforeEach, afterEach } from 'vitest'
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 { NounType, VerbType } from '../../src/types/graphTypes.js'
import { MemoryStorage } from '../../src/storage/adapters/memoryStorage.js'

const MANIFEST_REL = '_system/manifest.json'

describe('transact durability barrier — entity writes fsync before the counter advances', () => {
  let dir: string
  let brain: any
  let syncCalls: string[][]
  let beginCount: number
  let flushCount: number

  beforeEach(async () => {
    process.env.BRAINY_DETERMINISTIC_EMBEDDINGS = 'true'
    dir = fs.mkdtempSync(path.join(os.tmpdir(), 'brainy-durability-'))
    brain = new Brainy({
      requireSubtype: false,
      storage: { type: 'filesystem', path: dir },
      dimensions: 384,
      silent: true
    })
    await brain.init()

    // Drain the pending tier BEFORE instrumenting: the adopt-at-open fleet
    // default re-commits the init-time baseline as a buffered single-op
    // generation, and transact() flushes buffered single-ops first — that
    // flush's manifest sync would otherwise be recorded ahead of the
    // transact's own commit point and break the first-index ordering pins.
    await brain.flush()

    // Instrument the real filesystem storage: record every fsync batch in order,
    // and count barrier open/flush, delegating to the originals.
    syncCalls = []
    beginCount = 0
    flushCount = 0
    const storage = brain['storage']
    const origSync = storage.syncRawObjects.bind(storage)
    storage.syncRawObjects = async (paths: string[]) => {
      syncCalls.push([...paths])
      return origSync(paths)
    }
    const origBegin = storage.beginWriteBarrier.bind(storage)
    storage.beginWriteBarrier = () => {
      beginCount++
      return origBegin()
    }
    const origFlush = storage.flushWriteBarrier.bind(storage)
    storage.flushWriteBarrier = async () => {
      flushCount++
      return origFlush()
    }
  })

  afterEach(async () => {
    await brain.close()
    fs.rmSync(dir, { recursive: true, force: true })
  })

  /** Index of the first sync call whose paths satisfy `pred` (or -1). */
  const findSync = (pred: (p: string) => boolean): number =>
    syncCalls.findIndex((paths) => paths.some(pred))

  it('a committed transact fsyncs the entity write BEFORE the manifest', async () => {
    const id = '00000000-0000-4000-8000-00000000ab01'
    await brain.transact([{ op: 'add', id, data: 'durable entity', type: NounType.Thing }])

    // The barrier opened and flushed exactly once.
    expect(beginCount).toBe(1)
    expect(flushCount).toBe(1)

    // The entity's canonical metadata write was fsync'd…
    const entitySyncIdx = findSync((p) => p.includes(`/${id}/`) && p.includes('entities/nouns'))
    expect(entitySyncIdx).toBeGreaterThanOrEqual(0)
    // …and the manifest (the commit point) was fsync'd in a LATER call.
    const manifestSyncIdx = findSync((p) => p === MANIFEST_REL)
    expect(manifestSyncIdx).toBeGreaterThanOrEqual(0)
    expect(entitySyncIdx).toBeLessThan(manifestSyncIdx)
  })

  it('a multi-op transact (add + relate) fsyncs both endpoints and the edge before the manifest', async () => {
    const a = '00000000-0000-4000-8000-00000000ab02'
    const b = '00000000-0000-4000-8000-00000000ab03'
    await brain.transact([
      { op: 'add', id: a, data: 'A', type: NounType.Thing },
      { op: 'add', id: b, data: 'B', type: NounType.Thing },
      { op: 'relate', from: a, to: b, type: VerbType.Contains }
    ])

    expect(flushCount).toBe(1)
    const aIdx = findSync((p) => p.includes(`/${a}/`))
    const bIdx = findSync((p) => p.includes(`/${b}/`))
    const manifestIdx = findSync((p) => p === MANIFEST_REL)
    expect(aIdx).toBeGreaterThanOrEqual(0)
    expect(bIdx).toBeGreaterThanOrEqual(0)
    expect(manifestIdx).toBeGreaterThanOrEqual(0)
    expect(Math.max(aIdx, bIdx)).toBeLessThan(manifestIdx)

    // Sanity: the data is actually there and traverses.
    expect((await brain.related(a)).map((r: any) => r.to)).toContain(b)
  })

  it('a precommit-rejected batch opens no barrier and advances nothing', async () => {
    const anchor = await brain.add({ id: '00000000-0000-4000-8000-00000000ab04', data: 'anchor', type: NounType.Thing })
    const genBefore = brain.generationStore.generation()
    syncCalls.length = 0
    beginCount = 0
    flushCount = 0

    const never = '00000000-0000-4000-8000-00000000ab05'
    await expect(
      brain.transact([
        { op: 'add', id: never, data: 'never', type: NounType.Thing },
        { op: 'update', id: anchor, metadata: { poke: 1 }, ifRev: 999 }
      ])
    ).rejects.toMatchObject({ name: 'RevisionConflictError' })

    // Precommit throws before the barrier opens (it lives just before execute):
    // the barrier never opens, nothing is flushed, and the generation counter is
    // unchanged. (A manifest sync CAN appear here from transact() flushing the
    // anchor's previously-buffered single-op — that is the anchor's deferred
    // durability, not this rejected batch committing; beginCount is the clean
    // signal that this batch's commit path never reached execute.)
    expect(beginCount).toBe(0)
    expect(flushCount).toBe(0)
    expect(brain.generationStore.generation()).toBe(genBefore)
    expect(await brain.get(never)).toBeNull()
  })

  it('MemoryStorage does not implement the barrier (optional-chaining no-op)', () => {
    const mem = new MemoryStorage() as any
    // In-memory has no durability seam; the generation store treats the absent
    // barrier methods as no-ops via optional chaining.
    expect(mem.beginWriteBarrier).toBeUndefined()
    expect(mem.flushWriteBarrier).toBeUndefined()
  })
})