/** * @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 => { 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 }) })