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

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.