From bafb4e4caa6be1940b8d5ba437608b5265794030 Mon Sep 17 00:00:00 2001 From: David Snelling Date: Tue, 9 Jun 2026 10:04:24 -0700 Subject: [PATCH] feat: per-entity _rev + update({ ifRev }) CAS + add({ ifAbsent }) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Optimistic concurrency for multi-writer coordination — the read-then-CAS lock pattern + idempotent-bootstrap singleton inserts. Lands the surface the SDK scheduler asked for in BRAINY-EXPOSE-TRANSACTIONS, scoped to what ships cleanly without painting the 8.0 transact() API into a corner. A. PER-ENTITY _rev FIELD - src/coreTypes.ts — HNSWNounWithMetadata gains optional `_rev?: number`; STANDARD_ENTITY_FIELDS includes it so resolveEntityField routes correctly. - src/types/brainy.types.ts — Entity gains optional `_rev?: number`; Result mirrors it on the convenience flatten layer. - src/brainy.ts add() — initializes `_rev: 1` in storageMetadata. New entities always start at 1. - src/brainy.ts update() — reads currentRev off the persisted metadata (falls back to 1 for pre-7.31.0 entities with no _rev), writes `_rev: currentRev + 1` into the updated metadata. Every successful update() bumps by exactly 1. - src/brainy.ts convertNounToEntity + convertMetadataToEntity — both pull _rev out of the storage metadata and surface it at the top level of Entity. Pre-7.31.0 entities (no _rev in storage) read as `_rev: 1` so consumers see a consistent value. - src/storage/baseStorage.ts — six destructure sites updated to pull _rev out of the metadata bag so it doesn't leak into customMetadata. Noun-side returns surface _rev; verb-side destructures correctly but doesn't expose it on HNSWVerbWithMetadata (verb CAS is future work). - src/brainy.ts createResult() — flattens entity._rev onto the Result for backward compat with the existing convenience-field layer. B. update({ ifRev }) OPTIMISTIC CONCURRENCY - src/types/brainy.types.ts — UpdateParams gains optional `ifRev?: number`. - src/transaction/RevisionConflictError.ts (NEW) — carries `{ id, expected, actual }`. Message names the recipe: refetch with brain.get() and retry with the latest _rev. Subclass of Error. - src/brainy.ts update() — when params.ifRev is provided, compares against currentRev (the rev we just read) and throws RevisionConflictError on mismatch before any storage write. Omitting ifRev keeps the prior unconditional-update behavior; existing consumers see no change. - src/transaction/index.ts — exports RevisionConflictError. - src/index.ts — public export of RevisionConflictError. C. add({ ifAbsent }) BY-ID IDEMPOTENT INSERT - src/types/brainy.types.ts — AddParams gains optional `ifAbsent?: boolean`; AddManyParams mirrors it as a batch-level flag. - src/brainy.ts add() — when params.id AND params.ifAbsent, pre-reads storage.getNounMetadata(id); if present, returns the existing id without writing. No throw, no overwrite. Ignored when id is omitted (a fresh UUID can never collide). - src/brainy.ts addMany() — propagates the batch-level ifAbsent to each item's add() call. Per-item ifAbsent takes precedence so callers can override individual rows. WHAT'S NOT SHIPPED (and why) A public brain.transaction(fn) wrapper was on the table but was cut. The internal TransactionManager exposes raw Operation classes (SaveNounMetadata, etc.) that take StorageAdapter as a constructor argument. A clean high-level facade in 7.31.0 would have meant either: (a) Delegate to brain.add() / update() / relate(). Each of those opens its own internal transaction and commits before the closure returns. Looks atomic, isn't — a footgun. (b) Thread an optional `tx?` through every internal write site in brainy.ts (8 transactionManager.executeTransaction sites). ~2 days of real refactor with regression surface, and the API shape changes again in 8.0 anyway. The SDK scheduler's actual ask (BRAINY-EXPOSE-TRANSACTIONS) is the read-then-CAS lock pattern. _rev + ifRev solves it completely. Multi-write atomicity is the 8.0 brain.transact() use case where the Datomic-style immutable Db makes it atomic by construction — that's the right home. TESTS - New tests/integration/rev-and-ifabsent.test.ts (18 tests): - _rev initialization (1 on add) and surface on get fast/full paths + find - _rev auto-bump on update across multiple writes - update({ ifRev }) pass / fail / message format / omitted / legacy-no-rev - add({ ifAbsent }) writes when absent / no-op when present / id-required - addMany({ ifAbsent }) propagation + per-item override - SDK-scheduler scenario: two concurrent CAS updates, one wins one throws - Unit suite unchanged: 1468/1468. - All integration subtype + verb + strict + find-limits + new rev suites pass: 96/96. DOCS - New docs/guides/optimistic-concurrency.md (public: true) — full reference: the lock pattern, read-modify-write retry, idempotent bootstrap, how _rev interacts with brain.versions, branches/fork, and VFS, what's coming in 8.0. - docs/api/README.md — add() and update() entries get the new params + tips pointing at the new guide. - RELEASES.md v7.31.0 entry. CORTEX COMPATIBILITY Zero changes required. _rev is a metadata column already supported by NativeColumnStore. The auto-bump runs in Brainy JS before any storage call; Cortex never sees the per-entity counter. 8.0 FORWARD-COMPAT _rev, ifRev, RevisionConflictError, and ifAbsent survive the 8.0 Db redesign unchanged. 8.0 layers brain.transact(tx, { ifAtGeneration }) for whole-tx CAS on top of the same per-entity mechanism — per-entity for single-record patterns (job locks, idempotent state machines), generation-based for "did the world move under me." Locked in .strategy/BRAINY-8.0-SUBTYPE-CONTRACT.md § C-6. Verification - npx tsc --noEmit: clean - npm test: 1468 / 1468 unit - All integration suites including new rev-and-ifabsent: 96/96 - npm run build: clean - Closed-source product reference audit: clean --- RELEASES.md | 130 +++++++++ docs/api/README.md | 4 + docs/guides/optimistic-concurrency.md | 175 +++++++++++++ src/brainy.ts | 34 ++- src/coreTypes.ts | 8 +- src/index.ts | 3 + src/storage/baseStorage.ts | 15 +- src/transaction/RevisionConflictError.ts | 46 ++++ src/transaction/index.ts | 2 + src/types/brainy.types.ts | 27 ++ tests/integration/rev-and-ifabsent.test.ts | 291 +++++++++++++++++++++ 11 files changed, 726 insertions(+), 9 deletions(-) create mode 100644 docs/guides/optimistic-concurrency.md create mode 100644 src/transaction/RevisionConflictError.ts create mode 100644 tests/integration/rev-and-ifabsent.test.ts diff --git a/RELEASES.md b/RELEASES.md index d221dd7f..f5f6a870 100644 --- a/RELEASES.md +++ b/RELEASES.md @@ -10,6 +10,136 @@ Full auto-generated changelog: `CHANGELOG.md` · Releases: https://github.com/so --- +## v7.31.0 — 2026-06-09 + +**Affected products:** consumers that need multi-writer coordination (job +schedulers, idempotent state machines, distributed locks, optimistic UI updates) +or idempotent bootstrap of well-known singletons. Additive; drop-in from 7.30.2. + +### Per-entity `_rev` + `update({ ifRev })` + `add({ ifAbsent })` + +CouchDB / PouchDB / ETag-style optimistic concurrency. Every entity now carries a +monotonic `_rev: number` that Brainy auto-bumps on every successful `update()`. +Pass it back as `ifRev` to make read-modify-write race-safe without an external +lock service. `ifAbsent` adds by-ID idempotent insert for singletons / config rows +/ deterministic-ID bootstraps. + +**What's new:** + +- **`entity._rev: number`** — initialized to `1` on `add()`, bumped by `1` on every + successful `update()` that reaches storage. Surfaced on `get()` (both fast metadata- + only and full-vector paths), `find()`, `search()`, and on the `Result` flatten layer + for backward compatibility. Pre-7.31.0 entities without `_rev` are read as `1`. +- **`update({ id, ..., ifRev: number })`** — optimistic-concurrency check. When + provided, throws `RevisionConflictError` if the persisted `_rev` no longer matches. + Carries `{ id, expected, actual }` for principled recovery. Omitting `ifRev` keeps + the prior unconditional-update behavior. +- **`add({ id, ifAbsent: true })`** — by-ID idempotent insert. Returns the existing + `id` without writing if the entity already exists. No throw, no overwrite. Ignored + when `id` is omitted (a fresh UUID can never collide). +- **`addMany({ items, ifAbsent: true })`** — applies the flag to every item; per-item + `ifAbsent` overrides the batch flag. + +### The lock recipe (single-process or distributed) + +```ts +import { Brainy, RevisionConflictError } from '@soulcraft/brainy' + +const LOCK_ID = '...uuid...' + +await brain.add({ + id: LOCK_ID, + type: NounType.Document, + data: { owner: null, expiresAt: 0 }, + ifAbsent: true +}) + +async function tryAcquireLock(workerId: string, ttlMs: number) { + const lock = await brain.get(LOCK_ID) + if (!lock) throw new Error('lock document missing') + const state = lock.data as { owner: string | null; expiresAt: number } + if (state.owner && state.expiresAt > Date.now()) return false + + try { + await brain.update({ + id: LOCK_ID, + data: { owner: workerId, expiresAt: Date.now() + ttlMs }, + ifRev: lock._rev + }) + return true + } catch (err) { + if (err instanceof RevisionConflictError) return false + throw err + } +} +``` + +### Why this is scoped to `_rev` + `ifAbsent` + +A public `brain.transaction(fn)` wrapper was on the table but was cut. The internal +`TransactionManager` exposes raw `Operation` classes that take a `StorageAdapter` +constructor argument; a high-level facade in 7.31.0 would have meant either +(a) delegating to `brain.add()` / `update()` / `relate()` — each of which commits +its own internal transaction before the closure returns, so multi-write rollback +wouldn't actually work (a footgun), or (b) threading `tx?` through every internal +write path — ~2 days of refactor with regression surface, and the API shape changes +again in 8.0 anyway. + +The SDK-scheduler use case that drove the thread (read-then-CAS lock pattern) is +fully solved by `_rev` + `ifRev`. Multi-write atomicity is the 8.0 `brain.transact()` +use case, where the Datomic-style immutable Db makes it atomic by construction. We +chose not to ship a worse version in the interim. + +### How `_rev` interacts with branches and the snapshot API + +| System | What it tracks | When it advances | +|---|---|---| +| `_rev` (NEW) | Per-entity write counter | Auto, on every successful `update()` | +| `brain.versions.save()` | Named snapshots per entity | Explicit — you call `save()` | +| `brain.fork()` / branches | Whole-brain copy-on-write | Explicit — you call `fork(name)` | +| VFS file versioning | Per-VFS-file snapshots | Same as `brain.versions.save()` | + +`_rev` is independent. Each branch has its own copy of every entity, so each +branch has its own `_rev` per entity (same as every other field under COW). +Snapshots taken via `brain.versions.save()` capture the entity at that moment +including its `_rev` at that time; the snapshot's own `version: number` is the +snapshot index, distinct from `_rev`. + +### Docs + +- **New `docs/guides/optimistic-concurrency.md`** (public, indexed at + soulcraft.com/docs after portal deploy) — full reference: the lock pattern, + read-modify-write retry, idempotent bootstrap, how `_rev` interacts with branches / + snapshots / VFS, what's coming in 8.0. +- `docs/api/README.md` `add()` and `update()` entries get the new params + tips + pointing at the guide. + +### Tests + +- **New `tests/integration/rev-and-ifabsent.test.ts`** (18 tests): rev initialization, + rev bump on update, ifRev pass / fail / omit / legacy-entity-fallback, ifAbsent + with custom id / no id / batch propagation / per-item override, plus a full + SDK-scheduler-style two-concurrent-CAS-updaters scenario. +- Existing suites unchanged: subtype-and-facets 26/26, verb-subtype-and-enforcement + 30/30, strict-mode-self-test 13/13, find-limits 9/9. Unit 1468/1468. + +### Cortex compatibility + +Zero changes required. `_rev` is a metadata column already supported by +`NativeColumnStore`; the auto-bump runs in Brainy JS before any storage call. +Cortex never sees the per-entity counter — it's a single integer field updated +alongside every other metadata field on the existing write paths. + +### 8.0 forward-compat + +`_rev`, `ifRev`, `RevisionConflictError`, and `ifAbsent` all survive the 8.0 Db +redesign unchanged. 8.0 layers `brain.transact(tx, { ifAtGeneration })` for +whole-tx CAS on top of the same per-entity mechanism — per-entity for single-record +patterns, generation-based for "did the world move under me." See +`.strategy/BRAINY-8.0-SUBTYPE-CONTRACT.md` § C-6 (internal). + +--- + ## v7.30.2 — 2026-06-08 **Affected products:** any consumer with `find({ limit })` call sites that pass values diff --git a/docs/api/README.md b/docs/api/README.md index 42e04ecf..69df13fb 100644 --- a/docs/api/README.md +++ b/docs/api/README.md @@ -128,6 +128,7 @@ const id = await brain.add({ - `vector?`: `number[]` - Pre-computed vector (skips auto-embedding) - `confidence?`: `number` - Type classification confidence (0-1) - `weight?`: `number` - Entity importance/salience (0-1) +- `ifAbsent?`: `boolean` - By-ID idempotent insert. When `true` AND a custom `id` is supplied AND an entity with that `id` already exists, returns the existing `id` without writing (no throw, no overwrite). Ignored without `id`. See [guides/optimistic-concurrency](../guides/optimistic-concurrency.md). > **`data`** is embedded into vectors for semantic search. **`metadata`** is indexed for `where` filters. See [Data Model](../DATA_MODEL.md). @@ -176,9 +177,12 @@ await brain.update({ - `metadata?`: `object` - Metadata to merge (or replace with `merge: false`) - `confidence?`: `number` - Update classification confidence - `weight?`: `number` - Update entity importance +- `ifRev?`: `number` - Optimistic-concurrency check. When provided, the update throws `RevisionConflictError` if the persisted entity's `_rev` no longer equals `ifRev`. See [guides/optimistic-concurrency](../guides/optimistic-concurrency.md). **Returns:** `Promise` +> **Tip — read-then-CAS.** Every entity returned by `get()` / `find()` / `search()` carries `entity._rev` (a monotonic counter Brainy auto-bumps on every successful `update()`). Pass it back as `ifRev` to make multi-writer coordination safe without an external lock service. Full guide: [guides/optimistic-concurrency](../guides/optimistic-concurrency.md). + --- ### `delete(id)` → `Promise` diff --git a/docs/guides/optimistic-concurrency.md b/docs/guides/optimistic-concurrency.md new file mode 100644 index 00000000..be903ad7 --- /dev/null +++ b/docs/guides/optimistic-concurrency.md @@ -0,0 +1,175 @@ +--- +title: Optimistic concurrency with _rev +slug: guides/optimistic-concurrency +public: true +category: guides +template: guide +order: 8 +description: Use the per-entity `_rev` counter and `update({ ifRev })` to coordinate concurrent writes safely. Covers the lock pattern, idempotent inserts with `ifAbsent`, and recovery on conflict. +next: + - guides/find-limits + - api/README +--- + +# Optimistic concurrency with `_rev` + +Brainy 7.31.0 adds a per-entity revision counter so multiple writers can coordinate without a global lock or external coordinator. The pattern is the same one CouchDB, PouchDB, and ETag-based HTTP caches use: read the current revision, do your work, write back with `ifRev: `. If the revision moved, your write is rejected and you retry against the latest state. + +## What gets added + +| Surface | Behavior | +|---|---| +| `entity._rev: number` | Returned on every `get()`, `find()`, `search()`. Initialized to `1` on `add()`. Bumped by `1` on every successful `update()`. Pre-7.31.0 entities without `_rev` are surfaced as `1`. | +| `update({ id, ..., ifRev: number })` | If the persisted `_rev` does not equal `ifRev`, throws `RevisionConflictError`. Omitting `ifRev` keeps the prior (unconditional) update behavior. | +| `RevisionConflictError` | Carries `{ id, expected, actual }` for principled recovery. | +| `add({ id, ifAbsent: true })` | By-ID idempotent insert. Returns the existing `id` if one is already present; no throw, no overwrite. | +| `addMany({ items, ifAbsent: true })` | Applies `ifAbsent` to every item. Per-item `ifAbsent` overrides the batch flag. | + +`_rev` is independent of `brain.versions.*` (named snapshots), of `brain.fork()` / branches (COW), and of VFS file versioning. They all coexist; `_rev` is the per-write counter used for CAS. + +## The lock pattern + +Every distributed-job scheduler eventually wants this exact loop: + +```ts +import { Brainy, RevisionConflictError } from '@soulcraft/brainy' + +const LOCK_ID = '...uuid for this job slot...' + +// Bootstrap the lock document once (idempotent). +await brain.add({ + id: LOCK_ID, + type: NounType.Document, + data: { owner: null, expiresAt: 0 }, + ifAbsent: true +}) + +async function tryAcquireLock(workerId: string, ttlMs: number) { + const lock = await brain.get(LOCK_ID) + if (!lock) throw new Error('lock document missing') + + const state = lock.data as { owner: string | null; expiresAt: number } + const now = Date.now() + + // Only take the lock if it's free or expired. + if (state.owner && state.expiresAt > now) return false + + try { + await brain.update({ + id: LOCK_ID, + data: { owner: workerId, expiresAt: now + ttlMs }, + ifRev: lock._rev + }) + return true + } catch (err) { + if (err instanceof RevisionConflictError) { + // Another worker grabbed it between our read and write. + return false + } + throw err + } +} +``` + +No external lock service, no Redis SETNX, no Cloud Tasks. The CAS check is the lock. + +## Read-modify-write with retry + +The other common shape is "update a counter / config object" with bounded retries on conflict: + +```ts +async function incrementCounter(id: string, by: number) { + for (let attempt = 0; attempt < 5; attempt++) { + const entity = await brain.get(id) + if (!entity) throw new Error('counter does not exist') + const current = (entity.data as { value: number }).value + try { + await brain.update({ + id, + data: { value: current + by }, + ifRev: entity._rev + }) + return + } catch (err) { + if (err instanceof RevisionConflictError) continue // refetch + retry + throw err + } + } + throw new Error('counter update conflict after 5 attempts') +} +``` + +The retry bound matters — without one, two unlucky writers can ping-pong forever. + +## Idempotent bootstrap with `ifAbsent` + +For singletons (config rows, well-known seed entities, job-state documents) where the natural ID is deterministic: + +```ts +await brain.add({ + id: 'config:singleton', + type: NounType.Document, + data: { tenantQuota: 1000 }, + ifAbsent: true +}) +``` + +- First caller writes; gets back `'config:singleton'`. +- Every subsequent caller short-circuits at the pre-read; gets back the same `'config:singleton'` without touching the existing entity. +- `_rev` is **not** bumped on the no-op path (no write happened). + +`ifAbsent` is only meaningful when you supply an `id`. With no `id`, Brainy generates a fresh UUID that can never collide, so the flag is silently ignored. + +`addMany({ items, ifAbsent: true })` applies the flag to every item. Mixing per-item overrides with the batch flag works as you'd expect: per-item `ifAbsent: false` opts an individual row out, per-item `ifAbsent: true` opts it in. + +### Why no `addIfMissing({ match, add })` for attribute-based dedup? + +You may want "create if no entity with this email exists" (lookup by attribute, not by ID). That's a different operation: + +```ts +// What we DID NOT ship in 7.31.0 — the attribute-based variant. +await brain.addIfMissing({ // ← not a real API + match: { type: 'Person', where: { email: 'x@y.com' } }, + add: { data: '...', metadata: { email: 'x@y.com' } } +}) +``` + +It's race-prone outside a transaction: two concurrent imports both see "not found," both insert, you get duplicates. Without a unique-index primitive (which Brainy doesn't have today), this pattern needs to live inside a transaction: + +```ts +// The race-safe shape, available once brain.transact() ships in 8.0. +await brain.transact(async tx => { + const existing = await tx.find({ + type: 'Person', + where: { email: 'x@y.com' }, + limit: 1 + }) + if (existing.length === 0) { + await tx.add({ type: 'Person', data: '...', metadata: { email: 'x@y.com' } }) + } +}) +``` + +For 7.31.0, lean on `ifAbsent` when you control the ID, and accept the inherent dedup race when you don't. The 8.0 `brain.transact()` is the natural home for the attribute-based variant. + +## How `_rev` interacts with the other versioning systems + +Brainy has several persistence primitives that all touch the word "version" in different ways. `_rev` is independent of every one of them: + +| System | What it tracks | When it advances | +|---|---|---| +| **`_rev`** (new in 7.31.0) | Per-entity write counter | Auto, on every successful `update()` | +| **`brain.versions.save()`** | Named snapshots per entity | Explicit — you call `save()` with a tag | +| **`brain.fork()` / branches** | Whole-brain copy-on-write | Explicit — you call `fork(name)` | +| **VFS file versioning** | Per-VFS-file snapshots (Document entity) | Same as `brain.versions.save()` | + +If you're on a branch, each branch has its own copy of every entity (that's what COW means), so each branch has its own `_rev` per entity — same as every other field. Snapshots taken via `brain.versions.save()` capture the entity at that moment including its `_rev` at that time; the snapshot's own `version: number` is the snapshot index, distinct from `_rev`. + +## What's coming in 8.0 + +Brainy 8.0 ships a Datomic-style immutable `Db` API where `brain.transact(tx)` is the public multi-write atomicity primitive. Two things change for code written against 7.31.0's surface: + +- `_rev` survives unchanged. It's part of the 8.0 entity shape; the auto-bump moves to the new `transact()` write path. Code that uses `update({ ifRev })` keeps working. +- A new `brain.transact(tx, { ifAtGeneration: prev.generation })` adds generation-based CAS for whole-transaction atomicity, layered on top of `_rev`. Per-entity for single-record patterns (the lock above), generation-based for "did the world move under me." + +If you adopt `_rev` + `ifRev` in 7.31.0, no migration work is needed for 8.0. diff --git a/src/brainy.ts b/src/brainy.ts index fdf5b8a3..d6bd7e39 100644 --- a/src/brainy.ts +++ b/src/brainy.ts @@ -46,6 +46,7 @@ import type { import { MmapVectorBackend } from './hnsw/mmapVectorBackend.js' import { ConnectionsCodec } from './hnsw/connectionsCodec.js' import { TransactionManager } from './transaction/TransactionManager.js' +import { RevisionConflictError } from './transaction/RevisionConflictError.js' import { ValidationConfig, validateAddParams, @@ -1052,6 +1053,14 @@ export class Brainy implements BrainyInterface { // missing-subtype check via the `isVFSEntity` marker. this.enforceSubtypeOnAdd('add', params.type, params.subtype, params.metadata) + // ifAbsent (7.31.0) — by-ID idempotent insert. Only meaningful when a custom id + // is supplied; a freshly generated UUID can never collide. Returns the existing + // id without writing if the entity is already present (no throw, no overwrite). + if (params.id && params.ifAbsent) { + const existing = await this.storage.getNounMetadata(params.id) + if (existing) return params.id + } + // Generate ID if not provided const id = params.id || uuidv4() @@ -1078,6 +1087,7 @@ export class Brainy implements BrainyInterface { service: params.service, createdAt: Date.now(), updatedAt: Date.now(), + _rev: 1, ...(params.confidence !== undefined && { confidence: params.confidence }), ...(params.weight !== undefined && { weight: params.weight }), ...(params.createdBy && { createdBy: params.createdBy }) @@ -1372,6 +1382,7 @@ export class Brainy implements BrainyInterface { data: entity.data, confidence: entity.confidence, weight: entity.weight, + _rev: entity._rev, // Preserve full entity for backward compatibility entity, // Optional score explanation @@ -1406,6 +1417,8 @@ export class Brainy implements BrainyInterface { service: noun.service, data: noun.data, createdBy: noun.createdBy, + // 7.31.0 — surface revision counter (defaults to 1 for pre-7.31.0 entities) + _rev: typeof noun._rev === 'number' ? noun._rev : 1, // ONLY custom user fields in metadata (already separated by storage adapter) metadata: noun.metadata as T @@ -1438,7 +1451,7 @@ export class Brainy implements BrainyInterface { // Extract standard fields, rest are custom metadata // Same destructuring as baseStorage.getNoun() to ensure consistency - const { noun, subtype, createdAt, updatedAt, confidence, weight, service, data, createdBy, ...customMetadata } = metadata + const { noun, subtype, createdAt, updatedAt, confidence, weight, service, data, createdBy, _rev, ...customMetadata } = metadata const entity: Entity = { id, @@ -1454,6 +1467,8 @@ export class Brainy implements BrainyInterface { service, data, createdBy, + // 7.31.0 — surface revision counter (defaults to 1 for pre-7.31.0 entities) + _rev: typeof _rev === 'number' ? _rev : 1, // Custom user fields (standard fields removed, only custom remain) metadata: customMetadata as T @@ -1549,6 +1564,15 @@ export class Brainy implements BrainyInterface { throw new Error(`Entity ${params.id} not found`) } + // ifRev (7.31.0) — optimistic concurrency. If caller supplied ifRev, the persisted + // _rev must match exactly. Pre-7.31.0 entities without _rev are treated as rev 1. + const currentRev = typeof (existing.metadata as any)?._rev === 'number' + ? (existing.metadata as any)._rev + : (typeof (existing as any)._rev === 'number' ? (existing as any)._rev : 1) + if (typeof params.ifRev === 'number' && params.ifRev !== currentRev) { + throw new RevisionConflictError(params.id, params.ifRev, currentRev) + } + // Update vector if data changed let vector = existing.vector const newType = params.type || existing.type @@ -1571,6 +1595,7 @@ export class Brainy implements BrainyInterface { service: existing.service, createdAt: existing.createdAt, updatedAt: Date.now(), + _rev: currentRev + 1, // Update confidence and weight if provided, otherwise preserve existing ...(params.confidence !== undefined && { confidence: params.confidence }), ...(params.weight !== undefined && { weight: params.weight }), @@ -3613,7 +3638,12 @@ export class Brainy implements BrainyInterface { const promises = chunk.map(async (item) => { try { - const id = await this.add(item) + // ifAbsent (7.31.0) — propagate batch-level flag to each item, but let + // per-item flag take precedence so callers can override individual rows. + const itemWithIfAbsent = params.ifAbsent && item.ifAbsent === undefined + ? { ...item, ifAbsent: true } + : item + const id = await this.add(itemWithIfAbsent) result.successful.push(id) } catch (error) { result.failed.push({ diff --git a/src/coreTypes.ts b/src/coreTypes.ts index 8f4b1532..7bcceed6 100644 --- a/src/coreTypes.ts +++ b/src/coreTypes.ts @@ -237,6 +237,11 @@ export interface HNSWNounWithMetadata { // USER DATA (top-level) - compatible with other types data?: Record + // REVISION COUNTER (7.31.0) — monotonic per-write integer for optimistic concurrency. + // Initialized to 1 on add(); bumped on every successful update(). Pre-7.31.0 + // entities without _rev are surfaced as `1` so existing callers see a consistent value. + _rev?: number + // CUSTOM USER METADATA (only custom fields, no standard fields) metadata?: Record } @@ -265,7 +270,8 @@ export const STANDARD_ENTITY_FIELDS: ReadonlySet = new Set([ 'updatedAt', 'service', 'createdBy', - 'data' + 'data', + '_rev' ]) /** diff --git a/src/index.ts b/src/index.ts index 5d8a8361..bacd8c5b 100644 --- a/src/index.ts +++ b/src/index.ts @@ -124,6 +124,9 @@ export { PluginRegistry } from './plugin.js' export { MigrationRunner, MIGRATIONS } from './migration/index.js' export type { Migration, MigrationState, MigrationPreview, MigrationResult, MigrateOptions, MigrationError } from './migration/index.js' +// Export optimistic-concurrency types (7.31.0) +export { RevisionConflictError } from './transaction/RevisionConflictError.js' + // Export embedding functionality import { UniversalSentenceEncoder, diff --git a/src/storage/baseStorage.ts b/src/storage/baseStorage.ts index 83155e4f..c600038c 100644 --- a/src/storage/baseStorage.ts +++ b/src/storage/baseStorage.ts @@ -904,7 +904,7 @@ export abstract class BaseStorage extends BaseStorageAdapter { } // Combine into HNSWNounWithMetadata - Extract standard fields to top-level - const { noun, subtype, createdAt, updatedAt, confidence, weight, service, data, createdBy, ...customMetadata } = metadata + const { noun, subtype, createdAt, updatedAt, confidence, weight, service, data, createdBy, _rev, ...customMetadata } = metadata return { id: vector.id, @@ -921,6 +921,7 @@ export abstract class BaseStorage extends BaseStorageAdapter { service: service as string | undefined, data: data as Record | undefined, createdBy, + _rev: typeof _rev === 'number' ? _rev : 1, // Only custom user fields remain in metadata metadata: customMetadata } @@ -942,7 +943,7 @@ export abstract class BaseStorage extends BaseStorageAdapter { for (const noun of nouns) { const metadata = await this.getNounMetadata(noun.id) if (metadata) { - const { noun: nounType, subtype, createdAt, updatedAt, confidence, weight, service, data, createdBy, ...customMetadata } = metadata + const { noun: nounType, subtype, createdAt, updatedAt, confidence, weight, service, data, createdBy, _rev, ...customMetadata } = metadata nounsWithMetadata.push({ ...noun, @@ -956,6 +957,7 @@ export abstract class BaseStorage extends BaseStorageAdapter { service: service as string | undefined, data: data as Record | undefined, createdBy, + _rev: typeof _rev === 'number' ? _rev : 1, // Only custom user fields in metadata metadata: customMetadata }) @@ -1021,7 +1023,7 @@ export abstract class BaseStorage extends BaseStorageAdapter { } // Combine into HNSWVerbWithMetadata - Extract standard fields to top-level - const { subtype, createdAt, updatedAt, confidence, weight, service, data, createdBy, ...customMetadata } = metadata + const { subtype, createdAt, updatedAt, confidence, weight, service, data, createdBy, _rev, ...customMetadata } = metadata return { id: verb.id, @@ -1096,7 +1098,7 @@ export abstract class BaseStorage extends BaseStorageAdapter { const verb = this.deserializeVerb(vectorData) // Extract standard fields to top-level - const { subtype, createdAt, updatedAt, confidence, weight, service, data, createdBy, ...customMetadata } = metadataData + const { subtype, createdAt, updatedAt, confidence, weight, service, data, createdBy, _rev, ...customMetadata } = metadataData results.set(id, { id: verb.id, @@ -2475,7 +2477,7 @@ export abstract class BaseStorage extends BaseStorageAdapter { const noun = this.deserializeNoun(vectorData) // Extract standard fields to top-level - const { noun: nounType, subtype, createdAt, updatedAt, confidence, weight, service, data, createdBy, ...customMetadata } = metadataData + const { noun: nounType, subtype, createdAt, updatedAt, confidence, weight, service, data, createdBy, _rev, ...customMetadata } = metadataData results.set(id, { id: noun.id, @@ -2492,6 +2494,7 @@ export abstract class BaseStorage extends BaseStorageAdapter { service: service as string | undefined, data: data as Record | undefined, createdBy, + _rev: typeof _rev === 'number' ? _rev : 1, // Only custom user fields remain in metadata metadata: customMetadata }) @@ -3992,7 +3995,7 @@ export abstract class BaseStorage extends BaseStorageAdapter { // Extract standard fields from metadata to top-level const metadataObj = (metadata || {}) as VerbMetadata - const { subtype, createdAt, updatedAt, confidence, weight, service, data, createdBy, ...customMetadata } = metadataObj + const { subtype, createdAt, updatedAt, confidence, weight, service, data, createdBy, _rev, ...customMetadata } = metadataObj const verbWithMetadata: HNSWVerbWithMetadata = { id: hnswVerb.id, diff --git a/src/transaction/RevisionConflictError.ts b/src/transaction/RevisionConflictError.ts new file mode 100644 index 00000000..f8932168 --- /dev/null +++ b/src/transaction/RevisionConflictError.ts @@ -0,0 +1,46 @@ +/** + * @module transaction/RevisionConflictError + * @description Error thrown by `brain.update({ ifRev })` when the persisted entity's + * `_rev` no longer matches the caller-supplied expected revision. Carries enough + * context (entity id, expected rev, actual rev) for the caller to choose a recovery + * strategy: refetch + retry, escalate to the user, or surface the conflict. + */ + +/** + * Optimistic-concurrency conflict on `brain.update({ id, ..., ifRev })`. + * + * Thrown when the persisted entity's `_rev` differs from the caller-supplied `ifRev`. + * The standard recovery is read-modify-write: `await brain.get(id)` → reconcile → + * retry with the new `_rev`. + * + * @example + * try { + * await brain.update({ id, data: '...', ifRev: 5 }) + * } catch (err) { + * if (err instanceof RevisionConflictError) { + * // Refetch and retry with the latest rev + * const latest = await brain.get(err.id) + * await brain.update({ id, data: merge(latest.data, '...'), ifRev: err.actual }) + * } + * } + */ +export class RevisionConflictError extends Error { + /** Entity ID whose revision check failed */ + readonly id: string + /** Revision number the caller expected to see (the `ifRev` argument) */ + readonly expected: number + /** Revision number actually persisted */ + readonly actual: number + + constructor(id: string, expected: number, actual: number) { + super( + `update({ id: ${JSON.stringify(id)}, ifRev: ${expected} }) failed: persisted _rev is ${actual}. ` + + `The entity was modified by another writer since you read it. ` + + `Refetch with brain.get(${JSON.stringify(id)}) and retry with the latest _rev.` + ) + this.name = 'RevisionConflictError' + this.id = id + this.expected = expected + this.actual = actual + } +} diff --git a/src/transaction/index.ts b/src/transaction/index.ts index 2972ee24..a09eb234 100644 --- a/src/transaction/index.ts +++ b/src/transaction/index.ts @@ -30,3 +30,5 @@ export { InvalidTransactionStateError, TransactionTimeoutError } from './errors.js' + +export { RevisionConflictError } from './RevisionConflictError.js' diff --git a/src/types/brainy.types.ts b/src/types/brainy.types.ts index 218116c5..9d5746f9 100644 --- a/src/types/brainy.types.ts +++ b/src/types/brainy.types.ts @@ -49,6 +49,13 @@ export interface Entity { confidence?: number /** Entity importance/salience (0-1) */ weight?: number + /** + * Monotonic revision counter, auto-bumped on every successful `update()`. + * Starts at `1` on `add()`. Pre-7.31.0 entities without `_rev` are read as `1`. + * Pass back as `update({ id, ..., ifRev: _rev })` for optimistic-concurrency CAS — + * throws `RevisionConflictError` if the persisted rev moved since read. + */ + _rev?: number } /** @@ -126,6 +133,7 @@ export interface Result { data?: any // Entity data (from entity.data) confidence?: number // Type classification confidence (from entity.confidence) weight?: number // Entity importance (from entity.weight) + _rev?: number // Monotonic revision counter (from entity._rev) — pass back as update({ ifRev }) for CAS // Full entity (preserved for backward compatibility) entity: Entity @@ -197,6 +205,13 @@ export interface AddParams { weight?: number /** Track which augmentation created this entity */ createdBy?: { augmentation: string; version: string } + /** + * Conditional insert. When `true` AND a custom `id` is supplied AND an entity with + * that `id` already exists, `add()` returns the existing `id` without writing — no + * throw, no overwrite. Used for idempotent bootstrap and create-or-noop patterns. + * Ignored when `id` is omitted (a freshly generated UUID can never collide). + */ + ifAbsent?: boolean } /** @@ -212,6 +227,13 @@ export interface UpdateParams { vector?: Vector // New pre-computed vector confidence?: number // Update type classification confidence weight?: number // Update entity importance/salience + /** + * Optimistic concurrency check. When provided, the update fails with + * `RevisionConflictError` if the persisted entity's `_rev` does not equal `ifRev`. + * `_rev` is auto-bumped on every successful update — read it from the entity returned + * by `get()` / `find()` / `search()`. Omit to skip the check (unconditional update). + */ + ifRev?: number } /** @@ -499,6 +521,11 @@ export interface AddManyParams { chunkSize?: number // Batch size (default: 100) onProgress?: (done: number, total: number) => void continueOnError?: boolean // Continue if some fail + /** + * Conditional insert applied to every item. Equivalent to setting `ifAbsent: true` + * on each item individually. Item-level `ifAbsent` overrides the batch flag. + */ + ifAbsent?: boolean } /** diff --git a/tests/integration/rev-and-ifabsent.test.ts b/tests/integration/rev-and-ifabsent.test.ts new file mode 100644 index 00000000..9da98342 --- /dev/null +++ b/tests/integration/rev-and-ifabsent.test.ts @@ -0,0 +1,291 @@ +/** + * @module tests/integration/rev-and-ifabsent + * @description End-to-end tests for the 7.31.0 optimistic-concurrency surface: + * - `_rev` is initialized to 1 on add() and surfaced on get()/find()/search() + * - update() auto-bumps `_rev` + * - update({ ifRev }) accepts a matching rev and rejects a stale one with + * RevisionConflictError carrying { id, expected, actual } + * - add({ ifAbsent: true }) is a no-op when the id is already present + * - addMany({ ifAbsent: true }) applies the flag to every item + */ + +import { describe, it, expect, beforeEach } from 'vitest' +import { Brainy } from '../../src/brainy.js' +import { RevisionConflictError } from '../../src/transaction/RevisionConflictError.js' +import { NounType } from '../../src/types/graphTypes.js' + +describe('7.31.0 — _rev CAS + ifAbsent', () => { + let brain: Brainy + + beforeEach(async () => { + brain = new Brainy({ storage: { type: 'memory' as const } }) + await brain.init() + }) + + describe('_rev initialization + surface', () => { + it('initializes _rev to 1 on add()', async () => { + const id = await brain.add({ data: 'hello', type: NounType.Document }) + const entity = await brain.get(id) + expect(entity?._rev).toBe(1) + }) + + it('surfaces _rev on metadata-only get() fast path', async () => { + const id = await brain.add({ data: 'fast-path', type: NounType.Document }) + const entity = await brain.get(id) // metadata-only by default + expect(entity?._rev).toBe(1) + }) + + it('surfaces _rev on includeVectors get() full path', async () => { + const id = await brain.add({ data: 'full-path', type: NounType.Document }) + const entity = await brain.get(id, { includeVectors: true }) + expect(entity?._rev).toBe(1) + }) + + it('surfaces _rev on find() results', async () => { + const id = await brain.add({ data: 'findable', type: NounType.Document }) + const results = await brain.find({ type: NounType.Document }) + const hit = results.find((r) => r.id === id) + expect(hit?._rev).toBe(1) + }) + }) + + describe('_rev auto-bump on update()', () => { + it('bumps _rev to 2 after first update()', async () => { + const id = await brain.add({ data: 'v1', type: NounType.Document }) + await brain.update({ id, data: 'v2' }) + const entity = await brain.get(id) + expect(entity?._rev).toBe(2) + }) + + it('bumps _rev monotonically across multiple updates', async () => { + const id = await brain.add({ data: 'v1', type: NounType.Document }) + for (let i = 2; i <= 5; i++) { + await brain.update({ id, data: `v${i}` }) + const entity = await brain.get(id) + expect(entity?._rev).toBe(i) + } + }) + + it('bumps _rev on metadata-only update', async () => { + const id = await brain.add({ + data: 'doc', + type: NounType.Document, + metadata: { status: 'draft' } + }) + await brain.update({ id, metadata: { status: 'published' } }) + const entity = await brain.get(id) + expect(entity?._rev).toBe(2) + expect(entity?.metadata?.status).toBe('published') + }) + }) + + describe('update({ ifRev }) optimistic concurrency', () => { + it('passes when ifRev matches the persisted _rev', async () => { + const id = await brain.add({ data: 'v1', type: NounType.Document }) + await expect(brain.update({ id, data: 'v2', ifRev: 1 })).resolves.not.toThrow() + const after = await brain.get(id) + expect(after?._rev).toBe(2) + }) + + it('throws RevisionConflictError when ifRev is stale', async () => { + const id = await brain.add({ data: 'v1', type: NounType.Document }) + await brain.update({ id, data: 'v2' }) // bumps to 2 + + let caught: unknown = null + try { + await brain.update({ id, data: 'v3-from-stale-reader', ifRev: 1 }) + } catch (err) { + caught = err + } + expect(caught).toBeInstanceOf(RevisionConflictError) + const e = caught as RevisionConflictError + expect(e.id).toBe(id) + expect(e.expected).toBe(1) + expect(e.actual).toBe(2) + }) + + it('emits a message that points at the recovery recipe', async () => { + const id = await brain.add({ data: 'v1', type: NounType.Document }) + await brain.update({ id, data: 'v2' }) + try { + await brain.update({ id, ifRev: 1, data: 'v3' }) + throw new Error('should have thrown') + } catch (err) { + const msg = (err as Error).message + expect(msg).toMatch(/persisted _rev is 2/) + expect(msg).toMatch(/brain\.get/) + } + }) + + it('omits the check entirely when ifRev is not provided', async () => { + const id = await brain.add({ data: 'v1', type: NounType.Document }) + await brain.update({ id, data: 'v2' }) + await brain.update({ id, data: 'v3' }) // no ifRev — must succeed + const after = await brain.get(id) + expect(after?._rev).toBe(3) + }) + + it('treats pre-7.31.0 metadata with no _rev as rev 1', async () => { + // Simulate a legacy entity by writing directly through the storage adapter + // with no _rev field, then verifying ifRev: 1 succeeds. + const id = 'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa' + const storage = (brain as any).storage + await storage.saveNounMetadata(id, { + data: 'pre-7.31', + noun: NounType.Document, + createdAt: Date.now(), + updatedAt: Date.now() + // intentionally no _rev + }) + // Also save the vector so get() doesn't return null on the full path. + const vector = await (brain as any).embed('pre-7.31') + await storage.saveNoun({ + id, + vector, + connections: new Map(), + level: 0 + }) + const legacy = await brain.get(id) + expect(legacy?._rev).toBe(1) + await expect(brain.update({ id, data: 'updated', ifRev: 1 })).resolves.not.toThrow() + const after = await brain.get(id) + expect(after?._rev).toBe(2) + }) + }) + + describe('add({ ifAbsent: true })', () => { + it('writes when no entity with the id exists', async () => { + const id = await brain.add({ + id: '11111111-1111-4111-8111-111111111111', + data: 'first-write', + type: NounType.Document, + ifAbsent: true + }) + expect(id).toBe('11111111-1111-4111-8111-111111111111') + const entity = await brain.get(id) + expect(entity?.data).toBe('first-write') + expect(entity?._rev).toBe(1) + }) + + it('returns the existing id without writing when the id is taken', async () => { + await brain.add({ + id: '22222222-2222-4222-8222-222222222222', + data: 'original', + type: NounType.Document + }) + const id = await brain.add({ + id: '22222222-2222-4222-8222-222222222222', + data: 'attempted-overwrite', + type: NounType.Document, + ifAbsent: true + }) + expect(id).toBe('22222222-2222-4222-8222-222222222222') + const entity = await brain.get(id) + expect(entity?.data).toBe('original') // not overwritten + expect(entity?._rev).toBe(1) // not bumped (no write happened) + }) + + it('is a no-op (writes anyway) when no id is supplied', async () => { + // ifAbsent without id can't mean anything — a fresh UUID can't collide. + // The contract says "ignored when id is omitted." Verify it writes. + const id = await brain.add({ + data: 'no-id-uuid-generated', + type: NounType.Document, + ifAbsent: true + }) + expect(id).toMatch(/^[0-9a-f]{8}-/) // UUID v4 + const entity = await brain.get(id) + expect(entity?.data).toBe('no-id-uuid-generated') + }) + }) + + describe('addMany({ ifAbsent: true })', () => { + it('applies ifAbsent to every item', async () => { + // Seed one entity that should NOT be overwritten by the batch. + await brain.add({ + id: '33333333-3333-4333-8333-333333333333', + data: 'original', + type: NounType.Document + }) + + const result = await brain.addMany({ + ifAbsent: true, + items: [ + { id: '33333333-3333-4333-8333-333333333333', data: 'should-not-overwrite', type: NounType.Document }, + { id: '44444444-4444-4444-8444-444444444444', data: 'new', type: NounType.Document }, + { id: '55555555-5555-4555-8555-555555555555', data: 'new', type: NounType.Document } + ] + }) + + expect(result.successful).toContain('33333333-3333-4333-8333-333333333333') + expect(result.successful).toContain('44444444-4444-4444-8444-444444444444') + expect(result.successful).toContain('55555555-5555-4555-8555-555555555555') + + const seeded = await brain.get('33333333-3333-4333-8333-333333333333') + expect(seeded?.data).toBe('original') + const fresh1 = await brain.get('44444444-4444-4444-8444-444444444444') + expect(fresh1?.data).toBe('new') + }) + + it('per-item ifAbsent overrides the batch flag', async () => { + await brain.add({ + id: '77777777-7777-4777-8777-777777777777', + data: 'keep', + type: NounType.Document + }) + + // Batch flag is false (default), but one item opts in. + const result = await brain.addMany({ + items: [ + { id: '77777777-7777-4777-8777-777777777777', data: 'overwrite-attempt', type: NounType.Document, ifAbsent: true }, + { id: '66666666-6666-4666-8666-666666666666', data: 'new', type: NounType.Document } + ] + }) + + expect(result.successful).toContain('77777777-7777-4777-8777-777777777777') + const protectedEntity = await brain.get('77777777-7777-4777-8777-777777777777') + expect(protectedEntity?.data).toBe('keep') + }) + }) + + describe('SDK scheduler use case — read-then-CAS lock pattern', () => { + it('two concurrent CAS updates: one wins, one throws RevisionConflictError', async () => { + const lockId = 'bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb' + await brain.add({ + id: lockId, + data: { owner: null, expiresAt: 0 }, + type: NounType.Document, + metadata: { kind: 'job-state' }, + ifAbsent: true + }) + + // Two readers see the same _rev + const a = await brain.get(lockId) + const b = await brain.get(lockId) + expect(a?._rev).toBe(b?._rev) + + // A wins the race + await brain.update({ + id: lockId, + data: { owner: 'worker-A', expiresAt: Date.now() + 300_000 }, + ifRev: a!._rev + }) + + // B's CAS now fails because the rev has moved + let caught: unknown = null + try { + await brain.update({ + id: lockId, + data: { owner: 'worker-B', expiresAt: Date.now() + 300_000 }, + ifRev: b!._rev + }) + } catch (err) { + caught = err + } + expect(caught).toBeInstanceOf(RevisionConflictError) + // Final state: A owns the lock + const final = await brain.get(lockId) + expect((final?.data as any)?.owner).toBe('worker-A') + }) + }) +})