From 867939ed500f7a62a025693096befe6e9f9c780d Mon Sep 17 00:00:00 2001 From: David Snelling Date: Wed, 8 Jul 2026 15:13:26 -0700 Subject: [PATCH] fix: atomic ifAbsent/upsert inserts + exact blob reference counts under concurrency MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The fast-follow to the ifRev CAS fix: the sweep for the same check-then-act class found two more instances, both now closed with the same discipline — the decision runs at the serialization point that guards the apply. add({ifAbsent}) / add({upsert}): the absence check ran before the commit mutex, so N concurrent same-id creates could all pass it and all write — the second silently overwriting the first, violating ifAbsent's "return the existing id WITHOUT writing" contract and upsert's merge-never-clobber contract. The insert leg now carries a must-be-absent commit precondition (the same conditional-commit primitive ifRev uses), verified under the commit mutex against the authoritative before-image. A losing caller takes its documented resolution instead of overwriting: ifAbsent returns the existing id with zero writes; upsert merges into the now-existing entity via update() (shared param mapping in upsertMergeParams so the planning-time branch and the conflict retry can never drift), with a bounded retry if a concurrent delete lands between the conflict and the merge. The planning-time checks remain as fast-fails. transact()'s batch-level ifAbsent/upsert keep planning-time semantics (documented, converges to a valid entity). BlobStorage: write()'s dedup decision (exists → increment refCount, absent → create at 1) and delete()'s decrement-then-remove were unserialized read-modify-writes over blob-meta:. Concurrent writes of identical content could lose references, so a later delete removed bytes another file still referenced (data loss), or leaked unreferenced blobs. All reference-count-bearing mutations now serialize through a per-hash InMemoryMutex — distinct content never contends; incrementRefCount/ decrementRefCount are documented lock-assumed internals. Both fixes are complete by construction in-process: storage enforces single-writer-per-directory, so the process is the whole concurrency domain. Tests (tests/integration/ifabsent-upsert-blob-concurrency.test.ts): 8-way ifAbsent storm advances the store by exactly one generation with _rev 1; 8-way upsert storm on an absent id yields one create + seven merges (_rev === 8); sequential ifAbsent/upsert semantics pinned unchanged; N identical concurrent blob writes → refCount === N; the blob survives until the true last reference drops; interleaved write/delete storm never loses a landed reference. --- src/brainy.ts | 117 ++++++++-- src/storage/blobStorage.ts | 124 +++++++---- .../ifabsent-upsert-blob-concurrency.test.ts | 207 ++++++++++++++++++ 3 files changed, 387 insertions(+), 61 deletions(-) create mode 100644 tests/integration/ifabsent-upsert-blob-concurrency.test.ts diff --git a/src/brainy.ts b/src/brainy.ts index 88212db5..82f95e81 100644 --- a/src/brainy.ts +++ b/src/brainy.ts @@ -337,6 +337,22 @@ interface PlannedTransact { createdNouns: Set } +/** + * Internal control-flow signal: an insert guarded by a must-be-absent + * precondition (`add({ ifAbsent })` / `add({ upsert })`) found the entity + * already present in the authoritative before-image under the commit mutex — + * a concurrent writer created it after the planning-time absence check. + * Never escapes `add()`: the caller converts it into the documented + * resolution (ifAbsent → return the existing id without writing; upsert → + * merge into the now-existing entity via `update()`). + */ +class InsertPreconditionExistsSignal extends Error { + constructor(readonly id: string) { + super(`insert precondition: entity ${id} already exists`) + this.name = 'InsertPreconditionExistsSignal' + } +} + /** * The main Brainy class - Clean, Beautiful, Powerful * REAL IMPLEMENTATION - No stubs, no mocks @@ -1639,7 +1655,10 @@ export class Brainy implements BrainyInterface { // 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 // (canonical) id without writing if the entity is already present (no throw, - // no overwrite). + // no overwrite). This check is a FAST-FAIL (skips the embedding cost); the + // authoritative absence check runs in the insert's commit precondition + // below, atomic with the apply, so a concurrent same-id create cannot make + // both callers write. if (params.id && params.ifAbsent) { const existing = await this.storage.getNounMetadata(id) if (existing) return id @@ -1650,21 +1669,12 @@ export class Brainy implements BrainyInterface { // delegate to update() so the supplied fields MERGE into the existing entity // (metadata merge, re-embed on changed data, _rev bump, createdAt preserved) // instead of the default destructive overwrite. When the id is absent, fall - // through to the normal insert below. + // through to the guarded insert below (whose commit precondition converts a + // concurrent same-id create into this same merge — never an overwrite). if (params.id && params.upsert) { const existing = await this.storage.getNounMetadata(id) if (existing) { - await this.update({ - id, - ...(params.data !== undefined && { data: params.data }), - ...(params.type !== undefined && { type: params.type }), - ...(params.subtype !== undefined && { subtype: params.subtype }), - ...(params.visibility !== undefined && { visibility: params.visibility }), - ...(params.metadata !== undefined && { metadata: params.metadata }), - ...(params.vector !== undefined && { vector: params.vector }), - ...(params.confidence !== undefined && { confidence: params.confidence }), - ...(params.weight !== undefined && { weight: params.weight }) - }) + await this.update(this.upsertMergeParams(id, params)) return id } } @@ -1736,7 +1746,22 @@ export class Brainy implements BrainyInterface { // Execute atomically with transaction system, generation-stamped as one // immutable Model-B generation (before-image = the absent/create sentinel). // All operations succeed or all rollback - prevents partial failures - await this.persistSingleOp({ nouns: [id] }, async (tx) => { + // ifAbsent/upsert insert leg: must-be-absent commit precondition, run + // under the commit mutex against the authoritative before-image — the + // planning-time absence checks above are fast-fails that can interleave + // with a concurrent same-id create. Exactly one concurrent create wins; + // every other caller takes its documented resolution instead of + // silently overwriting the winner. + const requireAbsent = Boolean(params.id && (params.ifAbsent || params.upsert)) + const insertPrecommit = requireAbsent + ? (before: CommitBeforeImages): void => { + if (before.nouns.get(id)?.metadata) { + throw new InsertPreconditionExistsSignal(id) + } + } + : undefined + + const runInsert: TransactionFunction = async (tx) => { // Operation 1: Save metadata FIRST (TypeAwareStorage caching) // isNew=true: skip pre-read for rollback (entity doesn't exist yet) tx.addOperation( @@ -1763,7 +1788,44 @@ export class Brainy implements BrainyInterface { tx.addOperation( new AddToMetadataIndexOperation(this.metadataIndex, id, entityForIndexing) ) - }) + } + + // Bounded retry closes the remaining interleavings without ever holding + // a lock across the loop: a lost insert race resolves to skip (ifAbsent) + // or merge (upsert); a merge that then hits a concurrent delete retries + // the insert. Each persistSingleOp call constructs fresh operations, so + // re-running the callback is safe. + const MAX_UPSERT_ATTEMPTS = 10 + for (let attempt = 0; ; attempt++) { + try { + await this.persistSingleOp({ nouns: [id] }, runInsert, insertPrecommit) + break + } catch (err) { + if (!(err instanceof InsertPreconditionExistsSignal)) { + throw err + } + // A concurrent writer created the entity between our absence check + // and the commit. + if (params.ifAbsent) { + // Documented contract: return the existing id without writing. + return id + } + // upsert: merge into the now-existing entity. A concurrent delete + // between this conflict and the merge throws EntityNotFoundError — + // loop back and retry the insert. + try { + await this.update(this.upsertMergeParams(id, params)) + return id + } catch (mergeErr) { + if ( + !(mergeErr instanceof EntityNotFoundError) || + attempt >= MAX_UPSERT_ATTEMPTS + ) { + throw mergeErr + } + } + } + } // Aggregation hook (outside transaction — derived data, can be reconstructed) if (this._aggregationIndex) { @@ -1773,6 +1835,31 @@ export class Brainy implements BrainyInterface { return id } + /** + * @description The `update()` param mapping `add({ upsert })` delegates to + * when the target id already exists: every supplied field merges into the + * existing entity (metadata merge, re-embed on changed data, `_rev` bump, + * `createdAt` preserved); absent fields are left untouched. Shared by the + * planning-time upsert branch and the commit-conflict retry so the two + * paths can never drift. + * @param id - The canonical entity id the upsert resolved to. + * @param params - The original `add()` params carrying the fields to merge. + * @returns The `UpdateParams` for the merging `update()` call. + */ + private upsertMergeParams(id: string, params: AddParams): UpdateParams { + return { + id, + ...(params.data !== undefined && { data: params.data }), + ...(params.type !== undefined && { type: params.type }), + ...(params.subtype !== undefined && { subtype: params.subtype }), + ...(params.visibility !== undefined && { visibility: params.visibility }), + ...(params.metadata !== undefined && { metadata: params.metadata }), + ...(params.vector !== undefined && { vector: params.vector }), + ...(params.confidence !== undefined && { confidence: params.confidence }), + ...(params.weight !== undefined && { weight: params.weight }) + } as UpdateParams + } + /** * Get an entity by ID * diff --git a/src/storage/blobStorage.ts b/src/storage/blobStorage.ts index 4d98c0bc..7cb569ec 100644 --- a/src/storage/blobStorage.ts +++ b/src/storage/blobStorage.ts @@ -15,6 +15,7 @@ import { createHash } from 'crypto' import { unwrapBinaryData } from './binaryDataCodec.js' +import { InMemoryMutex } from '../utils/mutex.js' /** * @description Key-value bridge the blob store persists through. Implemented @@ -164,6 +165,19 @@ export class BlobStorage { private readonly CACHE_MAX_SIZE = 100 * 1024 * 1024 // 100MB default private readonly COMPRESSION_THRESHOLD = 1024 // 1KB - don't compress smaller + /** + * Per-hash write serialization. Every reference-count-bearing mutation + * (`write`'s dedup check-then-act, `delete`'s decrement-then-maybe-remove) + * is a read-modify-write over `blob-meta:` — unserialized, two + * concurrent writes of identical content both saw "absent" and both wrote + * `refCount: 1` (one reference lost → a later delete removed bytes another + * file still referenced), and concurrent increments/decrements could drop + * counts. Keyed by hash, so distinct content never contends; the process + * is the whole concurrency domain (storage enforces single-writer per + * directory). + */ + private readonly hashLocks = new InMemoryMutex() + /** * @param adapter - Key-value bridge to persist through. * @param options - `cacheMaxSize` bounds the LRU read cache (bytes, @@ -222,45 +236,53 @@ export class BlobStorage { async write(data: Buffer, options: BlobWriteOptions = {}): Promise { const hash = BlobStorage.hash(data) - // Deduplication: identical content already stored — just add a reference. - if (await this.has(hash)) { - await this.incrementRefCount(hash) + // The dedup decision (exists → add a reference; absent → create with + // refCount 1) is check-then-act over the same metadata a concurrent + // same-content write mutates — serialized per hash so N concurrent + // writes of identical content yield exactly N references, never a lost + // count (a lost reference turns a later delete into premature removal + // of bytes another file still needs). + return this.hashLocks.runExclusive(hash, async () => { + // Deduplication: identical content already stored — just add a reference. + if (await this.has(hash)) { + await this.incrementRefCount(hash) + return hash + } + + await this.ensureCompressionReady() + + // Determine compression strategy + const compression = this.selectCompression(data, options) + + // Compress if needed + let finalData = data + let compressedSize = data.length + + if (compression === 'zstd' && this.zstdCompress) { + finalData = await this.zstdCompress(data) + compressedSize = finalData.length + } + + // Record the ACTUAL compression state, not the intended one — prevents + // corruption if compression failed to initialize. + const actualCompression = finalData === data ? 'none' : compression + const metadata: BlobMetadata = { + hash, + size: data.length, + compressedSize, + compression: actualCompression, + createdAt: Date.now(), + refCount: 1 + } + + await this.adapter.put(`blob:${hash}`, finalData) + await this.adapter.put(`blob-meta:${hash}`, Buffer.from(JSON.stringify(metadata))) + + // Write-through cache (caches the ORIGINAL bytes, not the compressed form) + this.addToCache(hash, data, metadata) + return hash - } - - await this.ensureCompressionReady() - - // Determine compression strategy - const compression = this.selectCompression(data, options) - - // Compress if needed - let finalData = data - let compressedSize = data.length - - if (compression === 'zstd' && this.zstdCompress) { - finalData = await this.zstdCompress(data) - compressedSize = finalData.length - } - - // Record the ACTUAL compression state, not the intended one — prevents - // corruption if compression failed to initialize. - const actualCompression = finalData === data ? 'none' : compression - const metadata: BlobMetadata = { - hash, - size: data.length, - compressedSize, - compression: actualCompression, - createdAt: Date.now(), - refCount: 1 - } - - await this.adapter.put(`blob:${hash}`, finalData) - await this.adapter.put(`blob-meta:${hash}`, Buffer.from(JSON.stringify(metadata))) - - // Write-through cache (caches the ORIGINAL bytes, not the compressed form) - this.addToCache(hash, data, metadata) - - return hash + }) } /** @@ -341,16 +363,22 @@ export class BlobStorage { * @param hash - The blob's SHA-256 hash. */ async delete(hash: string): Promise { - const refCount = await this.decrementRefCount(hash) + // Decrement-then-maybe-remove must be atomic per hash: without the lock, + // a concurrent write() could re-reference the content between our + // reaching zero and the physical delete — removing bytes a live + // reference still needs. + await this.hashLocks.runExclusive(hash, async () => { + const refCount = await this.decrementRefCount(hash) - // Only delete if no references remain - if (refCount > 0) { - return - } + // Only delete if no references remain + if (refCount > 0) { + return + } - await this.adapter.delete(`blob:${hash}`) - await this.adapter.delete(`blob-meta:${hash}`) - this.removeFromCache(hash) + await this.adapter.delete(`blob:${hash}`) + await this.adapter.delete(`blob-meta:${hash}`) + this.removeFromCache(hash) + }) } /** @@ -403,6 +431,8 @@ export class BlobStorage { /** * Increment the reference count for an existing blob. + * Caller MUST hold the per-hash lock ({@link hashLocks}) — this is a raw + * read-modify-write with no serialization of its own. */ private async incrementRefCount(hash: string): Promise { const metadata = await this.getMetadata(hash) @@ -417,6 +447,8 @@ export class BlobStorage { /** * Decrement the reference count for a blob (floored at zero). + * Caller MUST hold the per-hash lock ({@link hashLocks}) — this is a raw + * read-modify-write with no serialization of its own. */ private async decrementRefCount(hash: string): Promise { const metadata = await this.getMetadata(hash) diff --git a/tests/integration/ifabsent-upsert-blob-concurrency.test.ts b/tests/integration/ifabsent-upsert-blob-concurrency.test.ts new file mode 100644 index 00000000..66cfdf96 --- /dev/null +++ b/tests/integration/ifabsent-upsert-blob-concurrency.test.ts @@ -0,0 +1,207 @@ +/** + * @module tests/integration/ifabsent-upsert-blob-concurrency + * @description The 8.0.15 CAS fix's fast-follow: the two remaining + * check-then-act races of the same class, found in the post-fix sweep. + * + * 1. `add({ ifAbsent })` / `add({ upsert })` — the absence check ran before + * the commit mutex, so N concurrent same-id creates could ALL pass it and + * all write (the second overwriting the first, violating ifAbsent's + * "no overwrite" contract and upsert's "merge, never clobber" contract). + * Fixed with the same conditional-commit primitive as `ifRev`: the insert + * leg carries a must-be-absent precondition; a loser resolves to skip + * (ifAbsent) or merge (upsert) instead of overwriting. + * + * 2. `BlobStorage` reference counts — `write()`'s dedup decision and + * `delete()`'s decrement-then-remove were unserialized read-modify-writes + * over `blob-meta:`; concurrent same-content writes could lose + * references, turning a later delete into premature removal of bytes + * another file still needs. Fixed with a per-hash mutex. + * + * Deterministic concurrency assertions: + * - ifAbsent storm: the generation counter advances by EXACTLY 1 + * (pre-fix: one generation per losing writer), `_rev` stays 1. + * - upsert storm: final `_rev === N` (1 create + N−1 merges, each with an + * honest bump; pre-fix a late insert reset `_rev` to 1 and clobbered). + * - blob storm: N same-content writes → refCount === N; N deletes → gone, + * with the bytes readable until the last reference drops. + */ +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' +import { BlobStorage } from '../../src/storage/blobStorage.js' + +describe('ifAbsent/upsert insert race + blob refCount (conditional commit fast-follow)', () => { + 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-upsert-race-')) + 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 add({ifAbsent}) → exactly ONE write (generation +1), no overwrite', async () => { + const id = freshId() + const genBefore = brain.generationStore.generation() + + const ids = await Promise.all( + Array.from({ length: 8 }, (_, i) => + brain.add({ + id, + ifAbsent: true, + data: `contender ${i}`, + type: NounType.Thing, + metadata: { writer: i } + }) + ) + ) + + // Every caller resolves to the same canonical id. + expect(new Set(ids).size).toBe(1) + + // THE contract: exactly one write landed. Pre-fix every losing caller + // also wrote (its own generation), silently overwriting the winner. + expect(brain.generationStore.generation()).toBe(genBefore + 1) + + const after = await brain.get(id) + expect(after._rev).toBe(1) + expect(typeof after.metadata.writer).toBe('number') + }) + + it('sequential ifAbsent semantics unchanged: existing entity is returned untouched', async () => { + const id = freshId() + await brain.add({ id, data: 'original', type: NounType.Thing, metadata: { keep: true } }) + const returned = await brain.add({ + id, + ifAbsent: true, + data: 'impostor', + type: NounType.Thing, + metadata: { keep: false } + }) + expect(returned).toBe(id) + const after = await brain.get(id) + expect(after.metadata.keep).toBe(true) + expect(after.data).toBe('original') + }) + + it('N concurrent add({upsert}) on an absent id → 1 create + N-1 merges (final _rev === N)', async () => { + const id = freshId() + + await Promise.all( + Array.from({ length: 8 }, (_, i) => + brain.add({ + id, + upsert: true, + data: 'shared upsert target', + type: NounType.Thing, + metadata: { [`k${i}`]: true } + }) + ) + ) + + const after = await brain.get(id) + // Exactly one insert (rev 1) + seven merging update()s, each with an + // honest monotonic bump. Pre-fix a losing insert restamped _rev to 1 and + // destroyed every merge that had already applied. + expect(after._rev).toBe(8) + // The entity survived as ONE identity: createdAt from the single create, + // and at least the last-applied merge's key present. + expect(Object.keys(after.metadata).filter((k) => k.startsWith('k')).length) + .toBeGreaterThanOrEqual(1) + }) + + it('sequential upsert semantics unchanged: existing → merge, absent → create', async () => { + const id = freshId() + await brain.add({ id, upsert: true, data: 'v1', type: NounType.Thing, metadata: { a: 1 } }) + expect((await brain.get(id))._rev).toBe(1) // created + + // add() requires data or vector even on the merge path — supply a vector + // and no data, so `data` preservation is still observable. + const vec = Array.from({ length: 384 }, (_, i) => (i % 7) / 7 - 0.5) + await brain.add({ id, upsert: true, vector: vec, type: NounType.Thing, metadata: { b: 2 } }) + const after = await brain.get(id) + expect(after._rev).toBe(2) // merged, not overwritten + expect(after.metadata.a).toBe(1) + expect(after.metadata.b).toBe(2) + expect(after.data).toBe('v1') // unsupplied field preserved + }) +}) + +describe('BlobStorage refCount is exact under concurrency (per-hash mutex)', () => { + /** Minimal in-memory adapter — the real interface, no mocks of behavior. */ + function memAdapter() { + const kv = new Map() + return { + get: async (k: string) => kv.get(k), + put: async (k: string, v: Buffer) => void kv.set(k, v), + delete: async (k: string) => void kv.delete(k), + list: async (prefix: string) => + [...kv.keys()].filter((k) => k.startsWith(prefix)) + } + } + + it('N concurrent writes of IDENTICAL content → refCount === N (no lost references)', async () => { + const blobs = new BlobStorage(memAdapter() as any) + const payload = Buffer.from('identical content stored by N concurrent writers') + + const hashes = await Promise.all( + Array.from({ length: 10 }, () => blobs.write(payload)) + ) + expect(new Set(hashes).size).toBe(1) + const meta = await blobs.getMetadata(hashes[0]) + // Pre-fix: concurrent writers raced the dedup check — several wrote + // refCount:1 over each other and increments were lost. + expect(meta?.refCount).toBe(10) + }) + + it('the blob survives until the LAST reference drops — no premature deletion', async () => { + const blobs = new BlobStorage(memAdapter() as any) + const payload = Buffer.from('shared bytes, two referencing files') + + const hash = await blobs.write(payload) + await blobs.write(payload) // second reference (concurrent-equivalent path) + + await blobs.delete(hash) // drop one reference + expect((await blobs.read(hash)).toString()).toBe(payload.toString()) // still readable + expect((await blobs.getMetadata(hash))?.refCount).toBe(1) + + await blobs.delete(hash) // last reference + expect(await blobs.has(hash)).toBe(false) + await expect(blobs.read(hash)).rejects.toThrow() + }) + + it('interleaved write/delete storm converges to an exact count', async () => { + const blobs = new BlobStorage(memAdapter() as any) + const payload = Buffer.from('storm payload') + const hash = BlobStorage.hash(payload) + + // 12 writes and 5 deletes racing: net 7 references, blob alive. + await Promise.all([ + ...Array.from({ length: 12 }, () => blobs.write(payload)), + ...Array.from({ length: 5 }, () => blobs.delete(hash)) + ]) + const meta = await blobs.getMetadata(hash) + // Deletes against a not-yet-written hash floor at zero without deleting, + // so the net can only be >= 12 - 5. The exactness we require: counts are + // never LOST (each landed write is represented). + expect(meta?.refCount).toBeGreaterThanOrEqual(7) + expect(await blobs.has(hash)).toBe(true) + }) +})