perf: optimize init() and rebuild performance
- Remove detectAndRepairCorruption from init() hot path — was loading all metadata chunks sequentially on startup. Now available via checkHealth() and repairIndex() methods. - Short-circuit warmCache on empty workspace — skip 4-6 wasted storage reads. - Parallelize MetadataIndex.init() and getGraphIndex() via Promise.all(). - Defer metadata writes during rebuild to batch boundaries (every 5000 entities) instead of flushing per-entity. - Skip pre-reads for new entities in transactions — saves 2 storage round-trips per add() on cloud storage.
This commit is contained in:
parent
cd875294ad
commit
35cb674157
3 changed files with 113 additions and 40 deletions
|
|
@ -270,15 +270,14 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
// Setup index now that we have storage
|
||||
this.index = this.setupIndex()
|
||||
|
||||
// Initialize core metadata index
|
||||
// Initialize metadata index and graph index in parallel
|
||||
// Both only need storage — they are independent of each other
|
||||
this.metadataIndex = new MetadataIndexManager(this.storage)
|
||||
await this.metadataIndex.init()
|
||||
|
||||
// Get GraphAdjacencyIndex from storage (SINGLETON pattern)
|
||||
// Storage owns the single instance, Brainy accesses it via getGraphIndex()
|
||||
// This fixes the dual-ownership bug where Brainy and Storage had separate instances
|
||||
// causing verbIdSet to be out of sync and VFS tree queries to fail
|
||||
this.graphIndex = await (this.storage as any).getGraphIndex()
|
||||
const [, graphIndex] = await Promise.all([
|
||||
this.metadataIndex.init(),
|
||||
(this.storage as any).getGraphIndex()
|
||||
])
|
||||
this.graphIndex = graphIndex
|
||||
|
||||
// Rebuild indexes if needed for existing data
|
||||
await this.rebuildIndexesIfNeeded()
|
||||
|
|
@ -688,18 +687,20 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
// All operations succeed or all rollback - prevents partial failures
|
||||
await this.transactionManager.executeTransaction(async (tx) => {
|
||||
// Operation 1: Save metadata FIRST (TypeAwareStorage caching)
|
||||
// isNew=true: skip pre-read for rollback (entity doesn't exist yet)
|
||||
tx.addOperation(
|
||||
new SaveNounMetadataOperation(this.storage, id, storageMetadata)
|
||||
new SaveNounMetadataOperation(this.storage, id, storageMetadata, true)
|
||||
)
|
||||
|
||||
// Operation 2: Save vector data
|
||||
// isNew=true: skip pre-read for rollback (entity doesn't exist yet)
|
||||
tx.addOperation(
|
||||
new SaveNounOperation(this.storage, {
|
||||
id,
|
||||
vector,
|
||||
connections: new Map(),
|
||||
level: 0
|
||||
})
|
||||
}, true)
|
||||
)
|
||||
|
||||
// Operation 3: Add to HNSW index (after entity saved)
|
||||
|
|
@ -6500,6 +6501,39 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check health of metadata indexes
|
||||
*
|
||||
* Returns validation result indicating whether indexes are healthy
|
||||
* or corrupted (e.g., from the update() field asymmetry bug).
|
||||
*
|
||||
* This check was previously run on every init(), causing significant
|
||||
* overhead on cloud storage (90+ sequential reads for 30-field datasets).
|
||||
* Now available as an on-demand diagnostic method.
|
||||
*/
|
||||
async checkHealth(): Promise<{
|
||||
healthy: boolean
|
||||
avgEntriesPerEntity: number
|
||||
entityCount: number
|
||||
indexEntryCount: number
|
||||
recommendation: string | null
|
||||
}> {
|
||||
await this.ensureInitialized()
|
||||
return this.metadataIndex.validateConsistency()
|
||||
}
|
||||
|
||||
/**
|
||||
* Detect and repair corrupted metadata indexes
|
||||
*
|
||||
* Runs corruption detection and auto-rebuilds if corruption is found.
|
||||
* This is the equivalent of the old init()-time corruption check,
|
||||
* now available as an explicit operation.
|
||||
*/
|
||||
async repairIndex(): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
await this.metadataIndex.detectAndRepairCorruption()
|
||||
}
|
||||
|
||||
/**
|
||||
* Close and cleanup
|
||||
*
|
||||
|
|
|
|||
|
|
@ -27,12 +27,15 @@ export class SaveNounMetadataOperation implements Operation {
|
|||
constructor(
|
||||
private readonly storage: StorageAdapter,
|
||||
private readonly id: string,
|
||||
private readonly metadata: NounMetadata
|
||||
private readonly metadata: NounMetadata,
|
||||
private readonly isNew: boolean = false
|
||||
) {}
|
||||
|
||||
async execute(): Promise<RollbackAction> {
|
||||
// Get existing metadata (for rollback)
|
||||
const previousMetadata = await this.storage.getNounMetadata(this.id)
|
||||
// Skip read for new entities — nothing to rollback to (saves 1 storage round-trip)
|
||||
const previousMetadata = this.isNew
|
||||
? null
|
||||
: await this.storage.getNounMetadata(this.id)
|
||||
|
||||
// Save new metadata
|
||||
await this.storage.saveNounMetadata(this.id, this.metadata)
|
||||
|
|
@ -65,12 +68,15 @@ export class SaveNounOperation implements Operation {
|
|||
|
||||
constructor(
|
||||
private readonly storage: StorageAdapter,
|
||||
private readonly noun: HNSWNoun
|
||||
private readonly noun: HNSWNoun,
|
||||
private readonly isNew: boolean = false
|
||||
) {}
|
||||
|
||||
async execute(): Promise<RollbackAction> {
|
||||
// Get existing noun (for rollback)
|
||||
const previousNoun = await this.storage.getNoun(this.noun.id)
|
||||
// Skip read for new entities — nothing to rollback to (saves 1 storage round-trip)
|
||||
const previousNoun = this.isNew
|
||||
? null
|
||||
: await this.storage.getNoun(this.noun.id)
|
||||
|
||||
// Save new noun
|
||||
await this.storage.saveNoun(this.noun)
|
||||
|
|
|
|||
|
|
@ -218,30 +218,35 @@ export class MetadataIndexManager {
|
|||
// Initialize EntityIdMapper (loads UUID ↔ integer mappings from storage)
|
||||
await this.idMapper.init()
|
||||
|
||||
// Warm the cache with common fields (lazy loading optimization)
|
||||
// This loads the 'noun' sparse index which is needed for type counts
|
||||
await this.warmCache()
|
||||
// Short-circuit: skip warmCache, lazyLoadCounts, syncTypeCountsToFixed for empty workspace
|
||||
// On empty workspace, there are no fields to warm — avoids 4-6 wasted storage reads (404s)
|
||||
const hasFields = this.fieldIndexes.size > 0
|
||||
|
||||
// Load type counts AFTER warmCache (sparse index is now cached)
|
||||
// Previously called in constructor without await and read from wrong source
|
||||
await this.lazyLoadCounts()
|
||||
if (hasFields) {
|
||||
// Warm the cache with common fields (lazy loading optimization)
|
||||
// This loads the 'noun' sparse index which is needed for type counts
|
||||
await this.warmCache()
|
||||
|
||||
// Phase 1b: Sync loaded counts to fixed-size arrays
|
||||
// Now correctly happens AFTER lazyLoadCounts() finishes
|
||||
this.syncTypeCountsToFixed()
|
||||
// Load type counts AFTER warmCache (sparse index is now cached)
|
||||
// Previously called in constructor without await and read from wrong source
|
||||
await this.lazyLoadCounts()
|
||||
|
||||
// Detect index corruption and auto-rebuild if necessary
|
||||
// The update() field asymmetry bug caused indexes to accumulate stale entries
|
||||
// This check runs on startup to detect and repair corrupted indexes automatically
|
||||
await this.detectAndRepairCorruption()
|
||||
// Phase 1b: Sync loaded counts to fixed-size arrays
|
||||
// Now correctly happens AFTER lazyLoadCounts() finishes
|
||||
this.syncTypeCountsToFixed()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Detect index corruption and automatically repair via rebuild
|
||||
* This catches the update() field asymmetry bug that causes 7 fields to accumulate per update
|
||||
* Corruption threshold: 100 avg metadata entries/entity, excluding __words__ (expected ~30)
|
||||
*
|
||||
* Removed from init() hot path for performance. Call explicitly via:
|
||||
* - brain.checkHealth() — returns health status
|
||||
* - brain.repairIndex() — runs detection + auto-repair
|
||||
*/
|
||||
private async detectAndRepairCorruption(): Promise<void> {
|
||||
async detectAndRepairCorruption(): Promise<void> {
|
||||
const validation = await this.validateConsistency()
|
||||
|
||||
if (!validation.healthy) {
|
||||
|
|
@ -1508,7 +1513,7 @@ export class MetadataIndexManager {
|
|||
* @param entityOrMetadata - Either full entity structure or plain metadata (backward compat)
|
||||
* @param skipFlush - Skip automatic flush (used during batch operations)
|
||||
*/
|
||||
async addToIndex(id: string, entityOrMetadata: any, skipFlush: boolean = false): Promise<void> {
|
||||
async addToIndex(id: string, entityOrMetadata: any, skipFlush: boolean = false, deferWrites: boolean = false): Promise<void> {
|
||||
const fields = this.extractIndexableFields(entityOrMetadata)
|
||||
|
||||
// Sanity check for excessive indexed fields (indicates possible data issue)
|
||||
|
|
@ -1559,7 +1564,11 @@ export class MetadataIndexManager {
|
|||
|
||||
// Flush all dirty chunks and sparse indices accumulated during this add operation
|
||||
// This batches writes that were previously sequential per-field into a single concurrent flush
|
||||
await this.flushDirtyMetadata()
|
||||
// During rebuild (deferWrites=true), defer flushes to periodic batch boundaries (every 5000 entities)
|
||||
// to avoid N × flush I/O amplification. The rebuild() method handles periodic + final flushes.
|
||||
if (!deferWrites) {
|
||||
await this.flushDirtyMetadata()
|
||||
}
|
||||
|
||||
// Adaptive auto-flush based on usage patterns
|
||||
if (!skipFlush) {
|
||||
|
|
@ -3103,10 +3112,16 @@ export class MetadataIndexManager {
|
|||
}
|
||||
|
||||
// Process all nouns
|
||||
let localCount = 0
|
||||
for (const noun of result.items) {
|
||||
const metadata = metadataBatch.get(noun.id)
|
||||
if (metadata) {
|
||||
await this.addToIndex(noun.id, metadata, true)
|
||||
await this.addToIndex(noun.id, metadata, true, true)
|
||||
localCount++
|
||||
// Periodic safety flush every 5000 entities to cap memory during rebuild
|
||||
if (localCount % 5000 === 0) {
|
||||
await this.flushDirtyMetadata()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -3192,14 +3207,18 @@ export class MetadataIndexManager {
|
|||
const metadata = metadataBatch.get(noun.id)
|
||||
if (metadata) {
|
||||
// Skip flush during rebuild for performance
|
||||
await this.addToIndex(noun.id, metadata, true)
|
||||
await this.addToIndex(noun.id, metadata, true, true)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// Yield after processing the entire batch
|
||||
await this.yieldToEventLoop()
|
||||
|
||||
|
||||
totalNounsProcessed += result.items.length
|
||||
// Periodic safety flush every 5000 entities to cap memory during rebuild
|
||||
if (totalNounsProcessed % 5000 === 0) {
|
||||
await this.flushDirtyMetadata()
|
||||
}
|
||||
hasMoreNouns = result.hasMore
|
||||
nounOffset += nounLimit
|
||||
|
||||
|
|
@ -3248,10 +3267,16 @@ export class MetadataIndexManager {
|
|||
}
|
||||
|
||||
// Process all verbs
|
||||
let verbLocalCount = 0
|
||||
for (const verb of result.items) {
|
||||
const metadata = verbMetadataBatch.get(verb.id)
|
||||
if (metadata) {
|
||||
await this.addToIndex(verb.id, metadata, true)
|
||||
await this.addToIndex(verb.id, metadata, true, true)
|
||||
verbLocalCount++
|
||||
// Periodic safety flush every 5000 entities to cap memory during rebuild
|
||||
if (verbLocalCount % 5000 === 0) {
|
||||
await this.flushDirtyMetadata()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -3332,14 +3357,18 @@ export class MetadataIndexManager {
|
|||
const metadata = verbMetadataBatch.get(verb.id)
|
||||
if (metadata) {
|
||||
// Skip flush during rebuild for performance
|
||||
await this.addToIndex(verb.id, metadata, true)
|
||||
await this.addToIndex(verb.id, metadata, true, true)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// Yield after processing the entire batch
|
||||
await this.yieldToEventLoop()
|
||||
|
||||
|
||||
totalVerbsProcessed += result.items.length
|
||||
// Periodic safety flush every 5000 entities to cap memory during rebuild
|
||||
if (totalVerbsProcessed % 5000 === 0) {
|
||||
await this.flushDirtyMetadata()
|
||||
}
|
||||
hasMoreVerbs = result.hasMore
|
||||
verbOffset += verbLimit
|
||||
|
||||
|
|
@ -3356,6 +3385,10 @@ export class MetadataIndexManager {
|
|||
}
|
||||
}
|
||||
|
||||
// Flush remaining dirty chunks/sparse indices accumulated during rebuild
|
||||
// (deferWrites=true prevented per-entity flushes, so dirty data accumulated)
|
||||
await this.flushDirtyMetadata()
|
||||
|
||||
// Flush to storage with final yield
|
||||
prodLog.debug('💾 Flushing metadata index to storage...')
|
||||
await this.flush()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue