soulcraft/brainy
2
0
Fork 0
Code Issues 1 Pull requests Projects Releases 335 Packages Wiki Activity Actions

fix(storage): resolve count synchronization race condition across all storage adapters

Browse source 
Fixed critical bug where entity and relationship counts were not being tracked correctly
during add(), relate(), and import() operations. The root cause was a race condition where
count increment code tried to read metadata before it was saved to storage.

Core Fixes:
- Modified baseStorage.saveNounMetadata_internal to increment counts AFTER metadata is saved
- Modified baseStorage.saveVerbMetadata_internal to increment verb counts AFTER metadata is saved
- Added verb type to VerbMetadata to avoid circular dependency during count tracking
- Refactored verb count methods to prevent mutex deadlocks (synchronous base + async Safe wrapper)

Storage Adapter Cleanup:
- Removed broken count increment code from FileSystemStorage, GcsStorage, R2Storage, AzureBlobStorage
- Updated MemoryStorage comments to reflect centralized fix
- All count tracking now centralized in baseStorage (fixes ALL adapters automatically)

New Utilities:
- Added rebuildCounts utility to repair corrupted counts.json from actual storage data
- Added comprehensive integration tests for count synchronization across all operations

Verification:
- All 8 storage adapters verified (FileSystem, GCS, Memory, S3Compatible, R2, Azure, OPFS, TypeAware)
- All code paths verified (add, relate, import, batch, update, delete)
- 599 tests passing (no regressions)
- No deadlocks (tests complete in 6s vs 150s+)

Fixes #1 and #2 reported by Workshop team
This commit is contained in:
David Snelling 2025-10-21 10:58:44 -07:00
parent e5c56ed285
commit 798a6946d6
10 changed files with 598 additions and 76 deletions
Download patch file Download diff file Expand all files Collapse all files

2
src/brainy.ts
View file

@ -784,7 +784,9 @@ export class Brainy<T = any> implements BrainyInterface<T> {
return this.augmentationRegistry.execute('relate', params, async () => {
// v4.0.0: Prepare verb metadata
// CRITICAL (v4.1.2): Include verb type in metadata for count tracking
const verbMetadata = {
verb: params.type, // Store verb type for count synchronization
weight: params.weight ?? 1.0,
...(params.metadata || {}),
createdAt: Date.now()

7
src/storage/adapters/azureBlobStorage.ts
View file

@ -1057,11 +1057,8 @@ export class AzureBlobStorage extends BaseStorage {
// Update cache
this.verbCacheManager.set(edge.id, edge)
// Increment verb count
const metadata = await this.getVerbMetadata(edge.id)
if (metadata && metadata.type) {
await this.incrementVerbCount(metadata.type as string)
}
// Count tracking happens in baseStorage.saveVerbMetadata_internal (v4.1.2)
// This fixes the race condition where metadata didn't exist yet
this.logger.trace(`Edge ${edge.id} saved successfully`)
this.releaseBackpressure(true, requestId)

70
src/storage/adapters/baseStorageAdapter.ts
View file

@ -1015,47 +1015,63 @@ export abstract class BaseStorageAdapter implements StorageAdapter {
}
/**
* Increment verb count - O(1) operation with mutex protection
* Increment verb count - O(1) operation (v4.1.2: now synchronous)
* Protected by storage-specific mechanisms (mutex, distributed consensus, etc.)
* @param type The verb type
*/
protected async incrementVerbCount(type: string): Promise<void> {
protected incrementVerbCount(type: string): void {
this.verbCounts.set(type, (this.verbCounts.get(type) || 0) + 1)
this.totalVerbCount++
// Update cache
this.countCache.set('verbs_count', {
count: this.totalVerbCount,
timestamp: Date.now()
})
}
/**
* Thread-safe increment for verb counts (v4.1.2)
* Uses mutex for single-node, distributed consensus for multi-node
* @param type The verb type
*/
protected async incrementVerbCountSafe(type: string): Promise<void> {
const mutex = getGlobalMutex()
await mutex.runExclusive(`count-verb-${type}`, async () => {
this.verbCounts.set(type, (this.verbCounts.get(type) || 0) + 1)
this.totalVerbCount++
// Update cache
this.countCache.set('verbs_count', {
count: this.totalVerbCount,
timestamp: Date.now()
})
this.incrementVerbCount(type)
// Smart batching (v3.32.3+): Adapts to storage type
await this.scheduleCountPersist()
})
}
/**
* Decrement verb count - O(1) operation with mutex protection
* Decrement verb count - O(1) operation (v4.1.2: now synchronous)
* @param type The verb type
*/
protected async decrementVerbCount(type: string): Promise<void> {
protected decrementVerbCount(type: string): void {
const current = this.verbCounts.get(type) || 0
if (current > 1) {
this.verbCounts.set(type, current - 1)
} else {
this.verbCounts.delete(type)
}
if (this.totalVerbCount > 0) {
this.totalVerbCount--
}
// Update cache
this.countCache.set('verbs_count', {
count: this.totalVerbCount,
timestamp: Date.now()
})
}
/**
* Thread-safe decrement for verb counts (v4.1.2)
* @param type The verb type
*/
protected async decrementVerbCountSafe(type: string): Promise<void> {
const mutex = getGlobalMutex()
await mutex.runExclusive(`count-verb-${type}`, async () => {
const current = this.verbCounts.get(type) || 0
if (current > 1) {
this.verbCounts.set(type, current - 1)
} else {
this.verbCounts.delete(type)
}
if (this.totalVerbCount > 0) {
this.totalVerbCount--
}
// Update cache
this.countCache.set('verbs_count', {
count: this.totalVerbCount,
timestamp: Date.now()
})
this.decrementVerbCount(type)
// Smart batching (v3.32.3+): Adapts to storage type
await this.scheduleCountPersist()
})

31
src/storage/adapters/fileSystemStorage.ts
View file

@ -253,9 +253,6 @@ export class FileSystemStorage extends BaseStorage {
protected async saveNode(node: HNSWNode): Promise<void> {
await this.ensureInitialized()
// Check if this is a new node to update counts
const isNew = !(await this.fileExists(this.getNodePath(node.id)))
// Convert connections Map to a serializable format
// CRITICAL: Only save lightweight vector data (no metadata)
// Metadata is saved separately via saveNounMetadata() (2-file system)
@ -276,18 +273,8 @@ export class FileSystemStorage extends BaseStorage {
JSON.stringify(serializableNode, null, 2)
)
// Update counts for new nodes (v4.0.0: load metadata separately)
if (isNew) {
// v4.0.0: Get type from separate metadata storage
const metadata = await this.getNounMetadata(node.id)
const type = metadata?.noun || 'default'
this.incrementEntityCount(type)
// Persist counts periodically (every 10 operations for efficiency)
if (this.totalNounCount % 10 === 0) {
await this.persistCounts()
}
}
// Count tracking happens in baseStorage.saveNounMetadata_internal (v4.1.2)
// This fixes the race condition where metadata didn't exist yet
}
/**
@ -460,9 +447,6 @@ export class FileSystemStorage extends BaseStorage {
protected async saveEdge(edge: Edge): Promise<void> {
await this.ensureInitialized()
// Check if this is a new edge to update counts
const isNew = !(await this.fileExists(this.getVerbPath(edge.id)))
// Convert connections Map to a serializable format
// ARCHITECTURAL FIX (v3.50.1): Include core relational fields in verb vector file
// These fields are essential for 90% of operations - no metadata lookup needed
@ -489,15 +473,8 @@ export class FileSystemStorage extends BaseStorage {
JSON.stringify(serializableEdge, null, 2)
)
// Update verb count for new edges (production-scale optimizations)
if (isNew) {
this.totalVerbCount++
// Persist counts periodically (every 10 operations for efficiency)
if (this.totalVerbCount % 10 === 0) {
this.persistCounts() // Async persist, don't await
}
}
// Count tracking happens in baseStorage.saveVerbMetadata_internal (v4.1.2)
// This fixes the race condition where metadata didn't exist yet
}
/**

14
src/storage/adapters/gcsStorage.ts
View file

@ -488,11 +488,8 @@ export class GcsStorage extends BaseStorage {
}
// Note: Empty vectors are intentional during HNSW lazy mode - not logged
// Increment noun count
const metadata = await this.getNounMetadata(node.id)
if (metadata && metadata.type) {
await this.incrementEntityCountSafe(metadata.type as string)
}
// Count tracking happens in baseStorage.saveNounMetadata_internal (v4.1.2)
// This fixes the race condition where metadata didn't exist yet
this.logger.trace(`Node ${node.id} saved successfully`)
this.releaseBackpressure(true, requestId)
@ -858,11 +855,8 @@ export class GcsStorage extends BaseStorage {
// Update cache
this.verbCacheManager.set(edge.id, edge)
// Increment verb count
const metadata = await this.getVerbMetadata(edge.id)
if (metadata && metadata.type) {
await this.incrementVerbCount(metadata.type as string)
}
// Count tracking happens in baseStorage.saveVerbMetadata_internal (v4.1.2)
// This fixes the race condition where metadata didn't exist yet
this.logger.trace(`Edge ${edge.id} saved successfully`)
this.releaseBackpressure(true, requestId)

3
src/storage/adapters/memoryStorage.ts
View file

@ -79,7 +79,8 @@ export class MemoryStorage extends BaseStorage {
// Save the noun directly in the nouns map
this.nouns.set(noun.id, nounCopy)
// Note: Count tracking happens in saveNounMetadata since type info is in metadata now
// Count tracking happens in baseStorage.saveNounMetadata_internal (v4.1.2)
// This fixes the race condition where metadata didn't exist yet
}
/**

6
src/storage/adapters/r2Storage.ts
View file

@ -771,10 +771,8 @@ export class R2Storage extends BaseStorage {
this.verbCacheManager.set(edge.id, edge)
const metadata = await this.getVerbMetadata(edge.id)
if (metadata && metadata.type) {
await this.incrementVerbCount(metadata.type as string)
}
// Count tracking happens in baseStorage.saveVerbMetadata_internal (v4.1.2)
// This fixes the race condition where metadata didn't exist yet
this.releaseBackpressure(true, requestId)
} catch (error: any) {

53
src/storage/baseStorage.ts
View file

@ -925,12 +925,35 @@ export abstract class BaseStorage extends BaseStorageAdapter {
/**
* Internal method for saving noun metadata (v4.0.0: now typed)
* Uses routing logic to handle both UUIDs (sharded) and system keys (unsharded)
*
* CRITICAL (v4.1.2): Count synchronization happens here
* This ensures counts are updated AFTER metadata exists, fixing the race condition
* where storage adapters tried to read metadata before it was saved.
*
* @protected
*/
protected async saveNounMetadata_internal(id: string, metadata: NounMetadata): Promise<void> {
await this.ensureInitialized()
// Determine if this is a new entity by checking if metadata already exists
const keyInfo = this.analyzeKey(id, 'noun-metadata')
return this.writeObjectToPath(keyInfo.fullPath, metadata)
const existingMetadata = await this.readObjectFromPath(keyInfo.fullPath)
const isNew = !existingMetadata
// Save the metadata
await this.writeObjectToPath(keyInfo.fullPath, metadata)
// CRITICAL FIX (v4.1.2): Increment count for new entities
// This runs AFTER metadata is saved, guaranteeing type information is available
// Uses synchronous increment since storage operations are already serialized
// Fixes Bug #1: Count synchronization failure during add() and import()
if (isNew && metadata.noun) {
this.incrementEntityCount(metadata.noun)
// Persist counts asynchronously (fire and forget)
this.scheduleCountPersist().catch(() => {
// Ignore persist errors - will retry on next operation
})
}
}
/**
@ -965,12 +988,38 @@ export abstract class BaseStorage extends BaseStorageAdapter {
/**
* Internal method for saving verb metadata (v4.0.0: now typed)
* Uses routing logic to handle both UUIDs (sharded) and system keys (unsharded)
*
* CRITICAL (v4.1.2): Count synchronization happens here
* This ensures verb counts are updated AFTER metadata exists, fixing the race condition
* where storage adapters tried to read metadata before it was saved.
*
* Note: Verb type is now stored in both HNSWVerb (vector file) and VerbMetadata for count tracking
*
* @protected
*/
protected async saveVerbMetadata_internal(id: string, metadata: VerbMetadata): Promise<void> {
await this.ensureInitialized()
// Determine if this is a new verb by checking if metadata already exists
const keyInfo = this.analyzeKey(id, 'verb-metadata')
return this.writeObjectToPath(keyInfo.fullPath, metadata)
const existingMetadata = await this.readObjectFromPath(keyInfo.fullPath)
const isNew = !existingMetadata
// Save the metadata
await this.writeObjectToPath(keyInfo.fullPath, metadata)
// CRITICAL FIX (v4.1.2): Increment verb count for new relationships
// This runs AFTER metadata is saved
// Verb type is now stored in metadata (as of v4.1.2) to avoid loading HNSWVerb
// Uses synchronous increment since storage operations are already serialized
// Fixes Bug #2: Count synchronization failure during relate() and import()
if (isNew && (metadata as any).verb) {
this.incrementVerbCount((metadata as any).verb)
// Persist counts asynchronously (fire and forget)
this.scheduleCountPersist().catch(() => {
// Ignore persist errors - will retry on next operation
})
}
}
/**

154
src/utils/rebuildCounts.ts Normal file
View file

@ -0,0 +1,154 @@
/**
* Rebuild Counts Utility
*
* Scans storage and rebuilds counts.json from actual data
* Use this to fix databases affected by the v4.1.1 count synchronization bug
*
* NO MOCKS - Production-ready implementation
*/
import type { BaseStorage } from '../storage/baseStorage.js'
export interface RebuildCountsResult {
/** Total number of entities (nouns) found */
nounCount: number
/** Total number of relationships (verbs) found */
verbCount: number
/** Entity counts by type */
entityCounts: Map<string, number>
/** Verb counts by type */
verbCounts: Map<string, number>
/** Processing time in milliseconds */
duration: number
}
/**
* Rebuild counts.json from actual storage data
*
* This scans all entities and relationships in storage and reconstructs
* the counts index from scratch. Use this to fix count desynchronization.
*
* @param storage - The storage adapter to rebuild counts for
* @returns Promise that resolves to rebuild statistics
*
* @example
* ```typescript
* const brain = new Brainy({ storage: { type: 'filesystem', path: './brainy-data' } })
* await brain.init()
*
* const result = await rebuildCounts(brain.storage)
* console.log(`Rebuilt counts: ${result.nounCount} nouns, ${result.verbCount} verbs`)
* ```
*/
export async function rebuildCounts(storage: BaseStorage): Promise<RebuildCountsResult> {
const startTime = Date.now()
console.log('🔧 Rebuilding counts from storage...')
const entityCounts = new Map<string, number>()
const verbCounts = new Map<string, number>()
let totalNouns = 0
let totalVerbs = 0
// Scan all nouns using pagination
console.log('📊 Scanning entities...')
// Check if pagination method exists
const storageWithPagination = storage as any
if (typeof storageWithPagination.getNounsWithPagination !== 'function') {
throw new Error('Storage adapter does not support getNounsWithPagination')
}
let hasMore = true
let cursor: string | undefined
while (hasMore) {
const result: any = await storageWithPagination.getNounsWithPagination({ limit: 100, cursor })
for (const noun of result.items) {
const metadata = await storage.getNounMetadata(noun.id)
if (metadata?.noun) {
const entityType = metadata.noun
entityCounts.set(entityType, (entityCounts.get(entityType) || 0) + 1)
totalNouns++
}
}
hasMore = result.hasMore
cursor = result.nextCursor
}
console.log(` Found ${totalNouns} entities across ${entityCounts.size} types`)
// Scan all verbs using pagination
console.log('🔗 Scanning relationships...')
if (typeof storageWithPagination.getVerbsWithPagination !== 'function') {
throw new Error('Storage adapter does not support getVerbsWithPagination')
}
hasMore = true
cursor = undefined
while (hasMore) {
const result: any = await storageWithPagination.getVerbsWithPagination({ limit: 100, cursor })
for (const verb of result.items) {
if (verb.verb) {
const verbType = verb.verb
verbCounts.set(verbType, (verbCounts.get(verbType) || 0) + 1)
totalVerbs++
}
}
hasMore = result.hasMore
cursor = result.nextCursor
}
console.log(` Found ${totalVerbs} relationships across ${verbCounts.size} types`)
// Update storage adapter's in-memory counts FIRST
storageWithPagination.totalNounCount = totalNouns
storageWithPagination.totalVerbCount = totalVerbs
storageWithPagination.entityCounts = entityCounts
storageWithPagination.verbCounts = verbCounts
// Mark counts as pending persist (required for flushCounts to actually persist)
storageWithPagination.pendingCountPersist = true
storageWithPagination.pendingCountOperations = 1
// Persist counts using storage adapter's own persist method
// This ensures counts.json is written correctly (compressed or uncompressed)
await storageWithPagination.flushCounts()
const duration = Date.now() - startTime
console.log(`✅ Counts rebuilt successfully in ${duration}ms`)
console.log(` Entities: ${totalNouns}`)
console.log(` Relationships: ${totalVerbs}`)
console.log('')
console.log('Entity breakdown:')
entityCounts.forEach((count, entityType) => {
console.log(` ${entityType}: ${count}`)
})
if (verbCounts.size > 0) {
console.log('')
console.log('Relationship breakdown:')
verbCounts.forEach((count, verbType) => {
console.log(` ${verbType}: ${count}`)
})
}
return {
nounCount: totalNouns,
verbCount: totalVerbs,
entityCounts,
verbCounts,
duration
}
}