soulcraft/brainy
1
0
Fork 0
Code Issues 1 Pull requests Projects Releases 335 Packages Wiki Activity Actions

fix: ifRev CAS is atomic — the revision check now runs under the commit mutex

Browse source 
N concurrent update({ ifRev }) calls carrying the same expected revision all
fulfilled (zero RevisionConflictErrors, last-writer-wins) — the check ran
before the commit mutex, so interleaved callers all passed it before any apply
landed. Sequential calls conflicted correctly, which hid the race. A production
cutover rehearsal caught it: 8 parallel ifRev-guarded ledger writes all
"succeeded" and 7 were silently lost. Advisory locks built on exactly-one-winner
semantics could hand the same lock to two workers.

Fix: conditional commit. commitSingleOp and commitTransaction accept an
optional precommit(beforeImages) precondition, invoked under the commit mutex
against the just-read authoritative before-images and before anything is staged
or applied — the per-record analogue of ifAtGeneration, which always ran there.
A throw aborts the commit atomically (generation reservation returned, zero
staging I/O in the transaction path). The store stays domain-ignorant; update()
and transact() supply the ifRev predicate:

- update(): the planning-time check remains as a fast-fail (avoids embedding
  cost on an obviously stale expectation); the authoritative check re-verifies
  the before-image's _rev and re-stamps the update's _rev from it, so the
  counter is monotonic even for concurrent non-CAS updates (N plain updates
  advance _rev by N). CAS against a concurrently-removed entity now throws
  EntityNotFoundError instead of silently resurrecting it; plain updates keep
  their last-writer-wins re-create semantics.
- transact(): per-op ifRev expectations registered during planning
  (PlannedTransact.casUpdates) are re-verified in the batch's precommit; any
  conflict rejects the whole batch before staging. Multiple updates to one
  entity sequence through a running rev; ids the batch itself adds
  (PlannedTransact.createdNouns — add resets the rev baseline even on
  overwrite) keep their exact plan-time sequencing.

Regression suite (tests/integration/ifrev-concurrent-cas.test.ts): 8 parallel
same-rev updates → exactly 1 winner + 7 conflicts; same for transact batches;
_rev monotonicity; the documented read→CAS→retry ledger loop converging exactly
(8 workers, 8 decrements, 0 lost); sequential behavior unchanged; forward-ref
add+update in one batch unchanged.
This commit is contained in:
David Snelling 2026-07-08 14:53:34 -07:00
parent b6c4d693cd
commit 9a3d1bd494
3 changed files with 393 additions and 13 deletions
Download patch file Download diff file Expand all files Collapse all files

151
src/brainy.ts
View file

@ -166,7 +166,7 @@ import {
type ImportOptions,
type ImportResult
} from './db/portableGraph.js'
import { GenerationStore } from './db/generationStore.js'
import { GenerationStore, type CommitBeforeImages } from './db/generationStore.js'
import { isDeterministicEmbedMode } from './embeddings/deterministicEmbedMode.js'
import { GenerationConflictError } from './db/errors.js'
import { BrainyError, GraphIndexNotReadyError, MetadataIndexNotReadyError, MigrationInProgressError } from './errors/brainyError.js'
@ -314,6 +314,27 @@ interface PlannedTransact {
touchedVerbs: string[]
/** Aggregation-index hooks to run after the commit point. */
postCommit: Array<() => void>
/**
* One entry per staged `{ op: 'update' }`, re-verified UNDER the commit
* mutex (the generation store's `precommit`) so per-op `ifRev` CAS is
* atomic with the apply — planning-time checks can interleave with
* concurrent writers. `updatedMetadata` is the staged metadata object (held
* by reference by the staged operation), re-stamped there with the
* authoritative `_rev`. A conflict rejects the WHOLE batch.
*/
casUpdates: Array<{
id: string
ifRev?: number
updatedMetadata: { _rev?: number }
}>
/**
* Ids whose revision baseline THIS batch resets via an `add` op (a fresh
* create, or add's overwrite semantics which restamp `_rev: 1`). Updates to
* these ids sequence against in-batch state no concurrent writer can touch,
* so their plan-time CAS checks and rev stamps are already exact — the
* commit precondition leaves them untouched.
*/
createdNouns: Set<string>
}
/**
@ -1515,12 +1536,29 @@ export class Brainy<T = any> implements BrainyInterface<T> {
*/
private async persistSingleOp(
touched: { nouns?: string[]; verbs?: string[] },
run: TransactionFunction<void>
run: TransactionFunction<void>,
precommit?: (before: CommitBeforeImages) => void
): Promise<void> {
if (!this._generationStampingActive) {
// Init-time / infrastructure baseline write (e.g. the VFS root): apply
// WITHOUT creating a generation. Generation 0 is the freshly-materialized
// brain (bootstrap included); the first USER write is generation 1.
// Bootstrap writes are single-threaded, so a supplied precondition is
// honored against directly-read before-images (no commit mutex exists
// on this path — nothing to race).
if (precommit) {
const nouns = new Map<string, { kind: 'noun'; metadata: unknown; vector: unknown }>()
for (const id of touched.nouns ?? []) {
const prev = await this.storage.readNounRaw(id)
nouns.set(id, { kind: 'noun', metadata: prev.metadata, vector: prev.vector })
}
const verbs = new Map<string, { kind: 'verb'; metadata: unknown; vector: unknown }>()
for (const id of touched.verbs ?? []) {
const prev = await this.storage.readVerbRaw(id)
verbs.set(id, { kind: 'verb', metadata: prev.metadata, vector: prev.vector })
}
precommit({ nouns, verbs } as CommitBeforeImages)
}
await this.generationStore.runWithoutGeneration(() =>
this.transactionManager.executeTransaction(run)
)
@ -1528,6 +1566,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
}
await this.generationStore.commitSingleOp({
touched,
precommit,
execute: () => this.transactionManager.executeTransaction(run)
})
}
@ -2492,7 +2531,11 @@ export class Brainy<T = any> implements BrainyInterface<T> {
// ifRev (7.31.0) — optimistic concurrency. If caller supplied ifRev, the persisted
// _rev must match exactly. `_rev` is reserved: every read path surfaces it ONLY
// top-level (entities without one are read as rev 1), so that is the one place
// to look.
// to look. This check is purely a FAST-FAIL (avoids paying the embedding
// cost on an obviously stale expectation) — the authoritative check runs
// in the commit precondition below, under the commit mutex, where it is
// atomic with the apply. Concurrent same-rev updates all pass HERE but
// exactly one survives THERE.
const currentRev = typeof existing._rev === 'number' ? existing._rev : 1
if (typeof params.ifRev === 'number' && params.ifRev !== currentRev) {
throw new RevisionConflictError(params.id, params.ifRev, currentRev)
@ -2568,6 +2611,35 @@ export class Brainy<T = any> implements BrainyInterface<T> {
metadata: newMetadata
}
// Authoritative CAS + honest rev stamp, run by the generation store
// UNDER the commit mutex against the just-read before-image — the one
// point where check-and-apply is atomic (the fast-fail above can
// interleave with concurrent writers; this cannot). Also re-stamps
// `_rev` from the authoritative base so the counter stays monotonic
// even for concurrent non-CAS updates. The staged operations below
// capture `updatedMetadata` by reference, so the re-stamp lands.
const casPrecommit = (before: CommitBeforeImages): void => {
const beforeMeta = before.nouns.get(params.id)?.metadata as
| { _rev?: unknown }
| null
| undefined
if (!beforeMeta) {
// A concurrent remove() deleted the entity after our read. A CAS
// caller gets the truth; a plain update keeps its long-standing
// last-writer-wins semantics (the staged write re-creates it).
if (typeof params.ifRev === 'number') {
throw new EntityNotFoundError(params.id)
}
return
}
const authoritativeRev =
typeof beforeMeta._rev === 'number' ? beforeMeta._rev : 1
if (typeof params.ifRev === 'number' && params.ifRev !== authoritativeRev) {
throw new RevisionConflictError(params.id, params.ifRev, authoritativeRev)
}
updatedMetadata._rev = authoritativeRev + 1
}
// Execute atomically with transaction system, generation-stamped as one
// immutable Model-B generation (before-image = the entity's prior state).
await this.persistSingleOp({ nouns: [params.id] }, async (tx) => {
@ -2627,7 +2699,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
tx.addOperation(
new AddToMetadataIndexOperation(this.metadataIndex, params.id, entityForIndexing)
)
})
}, casPrecommit)
// Aggregation hook (outside transaction — derived data)
if (this._aggregationIndex) {
@ -6901,10 +6973,55 @@ export class Brainy<T = any> implements BrainyInterface<T> {
const plan = await this.planTransact(ops)
// Authoritative per-op ifRev CAS, run by the generation store UNDER the
// commit mutex against the just-read before-images — atomic with the
// apply (the planning-time checks above are fast-fails that can
// interleave with concurrent writers). Any conflict rejects the WHOLE
// batch before anything is staged or applied. Sequencing rules:
// - runningRev tracks multiple updates to the SAME entity in one batch
// (each op CASes against — and bumps — its predecessor's result).
// - An entity the batch itself creates (null before-image, forward-ref
// add+update) keeps its plan-sequenced stamps: no concurrent writer
// can have moved a rev that did not exist, so plan-time was exact.
const casPrecommit = (before: CommitBeforeImages): void => {
const runningRev = new Map<string, number>()
for (const cas of plan.casUpdates) {
// An id this batch adds owns its revision baseline (the add restamps
// `_rev: 1` even on overwrite) — plan-time sequencing is exact, no
// concurrent writer can move a baseline the batch itself sets.
if (plan.createdNouns.has(cas.id)) {
continue
}
const beforeMeta = before.nouns.get(cas.id)?.metadata as
| { _rev?: unknown }
| null
| undefined
if (!beforeMeta && !runningRev.has(cas.id)) {
// A pre-existing entity a concurrent remove() deleted after
// planning: a CAS caller gets the truth; a plain update keeps its
// last-writer-wins re-create semantics (plan-stamped rev).
if (typeof cas.ifRev === 'number') {
throw new EntityNotFoundError(cas.id)
}
continue
}
const base =
runningRev.get(cas.id) ??
(typeof beforeMeta?._rev === 'number' ? beforeMeta._rev : 1)
if (typeof cas.ifRev === 'number' && cas.ifRev !== base) {
throw new RevisionConflictError(cas.id, cas.ifRev, base)
}
const next = base + 1
cas.updatedMetadata._rev = next
runningRev.set(cas.id, next)
}
}
const { generation, timestamp } = await this.generationStore.commitTransaction({
touched: { nouns: plan.touchedNouns, verbs: plan.touchedVerbs },
meta: options?.meta,
ifAtGeneration: options?.ifAtGeneration,
precommit: casPrecommit,
execute: async () => {
await this.transactionManager.executeTransaction(async (tx) => {
for (const operation of plan.operations) {
@ -8045,7 +8162,9 @@ export class Brainy<T = any> implements BrainyInterface<T> {
ids: [],
touchedNouns: [],
touchedVerbs: [],
postCommit: []
postCommit: [],
casUpdates: [],
createdNouns: new Set()
}
for (const op of ops) {
@ -8180,6 +8299,11 @@ export class Brainy<T = any> implements BrainyInterface<T> {
? !state.nouns.has(id) && (await this.storage.getNounMetadata(id)) === null
: true
// Either way (fresh create OR overwrite) this add resets the id's revision
// baseline (`_rev: 1` below), so later in-batch updates to it sequence
// against batch-local state — see PlannedTransact.createdNouns.
plan.createdNouns.add(id)
const now = Date.now()
const storageMetadata = {
...params.metadata,
@ -8273,8 +8397,10 @@ export class Brainy<T = any> implements BrainyInterface<T> {
}
// ifRev CAS — identical resolution to update(); a conflict rejects the
// WHOLE batch before anything is staged or applied. `_rev` is reserved:
// every read path surfaces it ONLY top-level.
// WHOLE batch. This planning-time check is purely a FAST-FAIL (it can
// interleave with concurrent writers); the authoritative re-verify runs in
// transact()'s commit precondition under the commit mutex, via the
// plan.casUpdates entry registered below.
const currentRev = typeof existing._rev === 'number' ? existing._rev : 1
if (typeof params.ifRev === 'number' && params.ifRev !== currentRev) {
throw new RevisionConflictError(params.id, params.ifRev, currentRev)
@ -8324,6 +8450,17 @@ export class Brainy<T = any> implements BrainyInterface<T> {
visibility: params.visibility ?? existing.visibility
})
}
// Register for the authoritative under-mutex CAS re-verify + rev re-stamp
// (see PlannedTransact.casUpdates). The staged UpdateNounMetadataOperation
// holds `updatedMetadata` by reference, so the precommit re-stamp lands in
// what execute() writes.
plan.casUpdates.push({
id: params.id,
...(typeof params.ifRev === 'number' && { ifRev: params.ifRev }),
updatedMetadata
})
const entityForIndexing = {
id: params.id,
vector,

68
src/db/generationStore.ts
View file

@ -44,6 +44,20 @@ import type {
TxLogEntry
} from './types.js'
/**
* The byte-identical before-images of every id a commit touches, read UNDER
* the commit mutex immediately before the write applies. Passed to a commit's
* optional `precommit` hook so callers can enforce compare-and-swap
* preconditions (e.g. the per-entity `ifRev` check) atomically with the apply
* — the same conditional-commit idea as `ifAtGeneration`, generalized to
* arbitrary per-record predicates. An absent id maps to the create sentinel
* (`metadata: null, vector: null`).
*/
export interface CommitBeforeImages {
nouns: Map<string, GenerationRecord>
verbs: Map<string, GenerationRecord>
}
/** Storage-root-relative path of the persisted generation counter. */
export const GENERATION_COUNTER_PATH = '_system/generation.json'
/** Storage-root-relative path of the commit manifest. */
@ -546,6 +560,11 @@ export class GenerationStore {
touched: TouchedIds
meta?: Record<string, unknown>
ifAtGeneration?: number
/** Optional compare-and-swap precondition over the {@link CommitBeforeImages},
* run under the commit mutex BEFORE anything is staged or applied — the
* per-record analogue of `ifAtGeneration`. A throw aborts the whole batch:
* the generation reservation is returned and no staging I/O has happened. */
precommit?: (before: CommitBeforeImages) => void
execute: () => Promise<void>
}): Promise<{ generation: number; timestamp: number }> {
return this.withMutex(async () => {
@ -575,20 +594,37 @@ export class GenerationStore {
try {
// -- 3. Before-images + delta (the durable undo log) ------------------
const stagedPaths: string[] = []
let recordBytes = 0 // serialized record-set size, for retention accounting
// Read every before-image FIRST, then run the caller's CAS
// precondition against them, and only then stage to disk — so a
// conflicting batch aborts with zero staging I/O. The maps hold the
// byte-identical records the staged files are written from.
const nounBefore = new Map<string, GenerationRecord>()
for (const id of nouns) {
const prev = await this.storage.readNounRaw(id)
nounBefore.set(id, { kind: 'noun', metadata: prev.metadata, vector: prev.vector })
}
const verbBefore = new Map<string, GenerationRecord>()
for (const id of verbs) {
const prev = await this.storage.readVerbRaw(id)
verbBefore.set(id, { kind: 'verb', metadata: prev.metadata, vector: prev.vector })
}
// Conditional commit: the per-record CAS analogue of the
// `ifAtGeneration` check above, but against authoritative
// before-images under the mutex. A throw lands in the catch below,
// which returns the generation reservation; nothing was applied.
args.precommit?.({ nouns: nounBefore, verbs: verbBefore })
const stagedPaths: string[] = []
let recordBytes = 0 // serialized record-set size, for retention accounting
for (const [id, record] of nounBefore) {
const recordPath = `${dir}/prev/${id}.json`
const record: GenerationRecord = { kind: 'noun', metadata: prev.metadata, vector: prev.vector }
await this.storage.writeRawObject(recordPath, record)
recordBytes += serializedBytes(record)
stagedPaths.push(recordPath)
}
for (const id of verbs) {
const prev = await this.storage.readVerbRaw(id)
for (const [id, record] of verbBefore) {
const recordPath = `${dir}/prev/${id}.json`
const record: GenerationRecord = { kind: 'verb', metadata: prev.metadata, vector: prev.vector }
await this.storage.writeRawObject(recordPath, record)
recordBytes += serializedBytes(record)
stagedPaths.push(recordPath)
@ -705,11 +741,18 @@ export class GenerationStore {
*
* @param args.touched - The ids this write creates/updates/deletes, by kind.
* @param args.execute - Runs the single-op's existing operation batch.
* @param args.precommit - Optional compare-and-swap precondition, invoked
* under the commit mutex with the just-read {@link CommitBeforeImages} and
* BEFORE `execute()`. A throw aborts the commit atomically: the generation
* reservation is returned and nothing is applied or buffered. This is the
* one place a per-record CAS (e.g. `ifRev`) is race-free — any check done
* before this mutex can interleave with a concurrent writer.
* @returns The reserved generation and its timestamp.
*/
async commitSingleOp(args: {
touched: { nouns?: string[]; verbs?: string[] }
execute: () => Promise<void>
precommit?: (before: CommitBeforeImages) => void
}): Promise<{ generation: number; timestamp: number }> {
return this.withMutex(async () => {
const nouns = args.touched.nouns ? [...new Set(args.touched.nouns)] : []
@ -732,6 +775,19 @@ export class GenerationStore {
verbBefore.set(id, { kind: 'verb', metadata: prev.metadata, vector: prev.vector })
}
// Conditional commit: the caller's CAS precondition runs here — under
// the mutex, against the authoritative before-images, before any write.
// A throw aborts cleanly: return the generation reservation (nothing was
// applied or buffered) and surface the conflict to the caller.
if (args.precommit) {
try {
args.precommit({ nouns: nounBefore, verbs: verbBefore })
} catch (err) {
if (this.counter === gen) this.counter = gen - 1
throw err
}
}
// Execute the live write. inTransact suppresses the storage bump hook so
// the counter is not double-advanced (this generation already reserved
// it) — the same suppression transact() uses.

187
tests/integration/ifrev-concurrent-cas.test.ts Normal file
View file

@ -0,0 +1,187 @@
/**
* @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
})
})