fix: atomic ifAbsent/upsert inserts + exact blob reference counts under concurrency

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:<hash>. 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

@@ -337,6 +337,22 @@ interface PlannedTransact {
   createdNouns: Set<string>
 }
 
+/**
+ * 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<T = any> implements BrainyInterface<T> {
     // 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<T = any> implements BrainyInterface<T> {
     // 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({
+        await this.update(this.upsertMergeParams(id, params))
-          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 })
-        })
         return id
       }
     }
@@ -1736,7 +1746,22 @@ export class Brainy<T = any> implements BrainyInterface<T> {
       // 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<void> = 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<T = any> implements BrainyInterface<T> {
         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<T = any> implements BrainyInterface<T> {
       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<T>): UpdateParams<T> {
+    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<T>
+  }
+
   /**
    * Get an entity by ID
    *

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:<hash>` — 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<string> {
     const hash = BlobStorage.hash(data)
 
-    // Deduplication: identical content already stored — just add a reference.
+    // The dedup decision (exists → add a reference; absent → create with
-    if (await this.has(hash)) {
+    // refCount 1) is check-then-act over the same metadata a concurrent
-      await this.incrementRefCount(hash)
+    // 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<void> {
-    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
+      // Only delete if no references remain
-    if (refCount > 0) {
+      if (refCount > 0) {
-      return
+        return
-    }
+      }
 
-    await this.adapter.delete(`blob:${hash}`)
+      await this.adapter.delete(`blob:${hash}`)
-    await this.adapter.delete(`blob-meta:${hash}`)
+      await this.adapter.delete(`blob-meta:${hash}`)
-    this.removeFromCache(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<number> {
     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<number> {
     const metadata = await this.getMetadata(hash)

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:<hash>`; 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<string, Buffer>()
+    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)
+  })
+})