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.
This commit is contained in:
parent
7146ce3544
commit
867939ed50
3 changed files with 387 additions and 61 deletions
|
|
@ -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.
|
||||
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<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
|
||||
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<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)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue