feat: per-entity _rev + update({ ifRev }) CAS + add({ ifAbsent })
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<T> gains optional `_rev?: number`;
Result<T> 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<T> 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<T> gains optional `ifAbsent?: boolean`;
AddManyParams<T> 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
This commit is contained in:
parent
ccc1d74953
commit
bafb4e4caa
11 changed files with 726 additions and 9 deletions
|
|
@ -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<T = any> implements BrainyInterface<T> {
|
|||
// 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<T = any> implements BrainyInterface<T> {
|
|||
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<T = any> implements BrainyInterface<T> {
|
|||
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<T = any> implements BrainyInterface<T> {
|
|||
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<T = any> implements BrainyInterface<T> {
|
|||
|
||||
// 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<T> = {
|
||||
id,
|
||||
|
|
@ -1454,6 +1467,8 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
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<T = any> implements BrainyInterface<T> {
|
|||
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<T = any> implements BrainyInterface<T> {
|
|||
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<T = any> implements BrainyInterface<T> {
|
|||
|
||||
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({
|
||||
|
|
|
|||
|
|
@ -237,6 +237,11 @@ export interface HNSWNounWithMetadata {
|
|||
// USER DATA (top-level) - compatible with other types
|
||||
data?: Record<string, any>
|
||||
|
||||
// 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<string, unknown>
|
||||
}
|
||||
|
|
@ -265,7 +270,8 @@ export const STANDARD_ENTITY_FIELDS: ReadonlySet<string> = new Set([
|
|||
'updatedAt',
|
||||
'service',
|
||||
'createdBy',
|
||||
'data'
|
||||
'data',
|
||||
'_rev'
|
||||
])
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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<string, any> | 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<string, any> | 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<string, any> | 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,
|
||||
|
|
|
|||
46
src/transaction/RevisionConflictError.ts
Normal file
46
src/transaction/RevisionConflictError.ts
Normal file
|
|
@ -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
|
||||
}
|
||||
}
|
||||
|
|
@ -30,3 +30,5 @@ export {
|
|||
InvalidTransactionStateError,
|
||||
TransactionTimeoutError
|
||||
} from './errors.js'
|
||||
|
||||
export { RevisionConflictError } from './RevisionConflictError.js'
|
||||
|
|
|
|||
|
|
@ -49,6 +49,13 @@ export interface Entity<T = any> {
|
|||
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<T = any> {
|
|||
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<T>
|
||||
|
|
@ -197,6 +205,13 @@ export interface AddParams<T = any> {
|
|||
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<T = any> {
|
|||
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<T = any> {
|
|||
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
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue