brainy/tests/integration/ifrev-concurrent-cas.test.ts

/**
 * @module tests/integration/ifrev-concurrent-cas
 * @description Regression for the P0 report that `update({ ifRev })` CAS was
 * not atomic under concurrent in-process calls: N concurrent updates all
 * carrying the same `ifRev` ALL fulfilled (0 conflicts, last-writer-wins) —
 * the check ran before the commit mutex, so interleaved callers all passed it
 * before any apply landed. In production that silently lost 7 of 8
 * ifRev-guarded ledger writes and let two workers both "acquire" an advisory
 * lock built on exactly-one-winner semantics.
 *
 * The fix is a conditional commit: the generation store's `precommit` hook
 * re-verifies `ifRev` against the authoritative before-image UNDER the commit
 * mutex, atomically with the apply (the per-record analogue of
 * `ifAtGeneration`, which always ran there). These tests pin:
 *  - exactly-one-winner for N concurrent same-rev `update({ ifRev })`
 *  - the same for `transact()` per-op ifRev (whole batch rejected)
 *  - `_rev` monotonicity under concurrent plain updates (no ifRev)
 *  - the CAS retry loop converging exactly (the ledger/credit-consume shape)
 *  - sequential conflict behavior unchanged
 */
import { describe, it, expect, beforeAll, afterAll } 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 } from '../../src/types/graphTypes.js'

const isRevConflict = (e: unknown): boolean =>
  (e as Error)?.name === 'RevisionConflictError' ||
  String(e).includes('RevisionConflict')

describe('ifRev CAS is atomic under concurrency (conditional commit)', () => {
  let dir: string
  let brain: any
  let seq = 0
  const freshId = (): string =>
    `00000000-0000-4000-8000-${(++seq).toString(16).padStart(12, '0')}`

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

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

  it('N concurrent update({ifRev}) → exactly 1 winner, N-1 RevisionConflictError', async () => {
    const id = await brain.add({
      id: freshId(),
      data: 'contended entity',
      type: NounType.Thing,
      metadata: { credits: 8 }
    })
    const rev = (await brain.get(id))._rev

    const results = await Promise.allSettled(
      Array.from({ length: 8 }, (_, i) =>
        brain.update({ id, metadata: { writer: i }, merge: false, ifRev: rev })
      )
    )

    const wins = results.filter((r) => r.status === 'fulfilled')
    const conflicts = results.filter(
      (r) => r.status === 'rejected' && isRevConflict(r.reason)
    )
    expect(wins.length).toBe(1)
    expect(conflicts.length).toBe(7)

    // Exactly one bump; the surviving metadata belongs to the single winner.
    const after = await brain.get(id)
    expect(after._rev).toBe(rev + 1)
    expect(typeof after.metadata.writer).toBe('number')
  })

  it('transact() per-op ifRev → exactly 1 batch wins, others rejected whole', async () => {
    const id = await brain.add({
      id: freshId(),
      data: 'transact contended',
      type: NounType.Thing,
      metadata: { state: 'initial' }
    })
    const rev = (await brain.get(id))._rev

    const results = await Promise.allSettled(
      Array.from({ length: 6 }, (_, i) =>
        brain.transact([
          { op: 'update', id, metadata: { batch: i }, merge: false, ifRev: rev }
        ])
      )
    )
    expect(results.filter((r) => r.status === 'fulfilled').length).toBe(1)
    expect(
      results.filter((r) => r.status === 'rejected' && isRevConflict(r.reason))
        .length
    ).toBe(5)
    expect((await brain.get(id))._rev).toBe(rev + 1)
  })

  it('_rev is monotonic under concurrent plain updates (no ifRev): N updates → +N', async () => {
    const id = await brain.add({
      id: freshId(),
      data: 'plain concurrent updates',
      type: NounType.Thing,
      metadata: { v: 0 }
    })
    const rev = (await brain.get(id))._rev

    await Promise.all(
      Array.from({ length: 8 }, (_, i) => brain.update({ id, metadata: { v: i } }))
    )
    // Every applied update gets its own honest bump (previously all stamped
    // the same stale rev+1 they computed before the commit).
    expect((await brain.get(id))._rev).toBe(rev + 8)
  })

  it('the CAS retry loop converges exactly (the ledger shape: 8 workers, 8 consumes)', async () => {
    const id = await brain.add({
      id: freshId(),
      data: 'credit budget',
      type: NounType.Thing,
      metadata: { credits: 8 }
    })

    // Each worker: read → CAS-decrement → retry on conflict. The docs' documented
    // pattern; with atomic CAS this MUST land on exactly 0 with zero lost updates.
    const consumeOne = async (): Promise<void> => {
      for (let attempt = 0; attempt < 50; attempt++) {
        const cur = await brain.get(id)
        if (cur.metadata.credits <= 0) throw new Error('budget exhausted')
        try {
          await brain.update({
            id,
            metadata: { ...cur.metadata, credits: cur.metadata.credits - 1 },
            merge: false,
            ifRev: cur._rev
          })
          return
        } catch (e) {
          if (!isRevConflict(e)) throw e
        }
      }
      throw new Error('no convergence after 50 attempts')
    }

    await Promise.all(Array.from({ length: 8 }, () => consumeOne()))
    expect((await brain.get(id)).metadata.credits).toBe(0)
  })

  it('sequential conflict behavior is unchanged (stale ifRev throws, fresh succeeds)', async () => {
    const id = await brain.add({
      id: freshId(),
      data: 'sequential control',
      type: NounType.Thing,
      metadata: { n: 1 }
    })
    const rev = (await brain.get(id))._rev

    await brain.update({ id, metadata: { n: 2 }, ifRev: rev })
    await expect(
      brain.update({ id, metadata: { n: 3 }, ifRev: rev })
    ).rejects.toMatchObject({ name: 'RevisionConflictError' })
    // Fresh rev succeeds.
    const fresh = (await brain.get(id))._rev
    await brain.update({ id, metadata: { n: 3 }, ifRev: fresh })
    expect((await brain.get(id)).metadata.n).toBe(3)
  })

  it('forward-ref add+update of the same entity in one transact() batch still works', async () => {
    const id = freshId()
    await brain.transact([
      { op: 'add', id, data: 'created in batch', type: NounType.Thing, metadata: { step: 1 } },
      { op: 'update', id, metadata: { step: 2 } }
    ])
    const after = await brain.get(id)
    expect(after.metadata.step).toBe(2)
    expect(after._rev).toBe(2) // add stamps 1, in-batch update sequences to 2
  })
})