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

chore(8.0): step-7 follow-through — collapse remaining cloud branches + docs sweep (scaffold step 13)

Browse source 
Final cleanup pass for Brainy 8.0. Catches three categories of debt:

A. STEP-7 FOLLOW-THROUGH (rebuild-path collapse)

Step 7's bisect-reset (debugging a flaky test) lost the in-source edits
to three rebuild paths even though the commit message claimed they
shipped. Re-applied now:

- src/utils/metadataIndex.ts — collapsed the `isLocalStorage` /
  cloud-pagination branching. Local-load-all-at-once is the only path
  in 8.0. Removed ~150 LOC of paginated-cloud branching for both nouns
  and verbs, plus the safety counters (`consecutiveEmptyBatches`,
  `MAX_ITERATIONS`, etc.).
- src/hnsw/hnswIndex.ts — same simplification for HNSW rebuild. The
  paginated cloud path is gone; HNSW now loads all nodes at once.
  Removed ~85 LOC.
- src/graph/graphAdjacencyIndex.ts — same simplification for graph
  adjacency rebuild. Removed ~50 LOC.

The collapse is safe because cloud adapters were deleted in step 7;
`storageType === 'OPFSStorage'` (and similar) can never match now.

B. CLOUD-ONLY DOCS DELETED

- docs/operations/cost-optimization-aws-s3.md
- docs/operations/cost-optimization-azure.md
- docs/operations/cost-optimization-cloudflare-r2.md
- docs/operations/cost-optimization-gcs.md
- docs/operations/cloud-run-filestore-guide.md

(docs/deployment/* contained no cloud-specific files that needed deletion.)

C. STORAGE-ADAPTERS GUIDE REWRITTEN FOR 8.0

docs/guides/storage-adapters.md → fresh content reflecting the 8.0
reality:
- Two adapters: FileSystemStorage + MemoryStorage. Quick-start matrix.
- Cloud backup section explains the operator-tooling pattern (gsutil /
  aws s3 / rclone / azcopy) with the exact commands consumers will run.
- "Why no cloud adapters in 8.0?" section documents the four reasons
  per BR-BRAINY-80-STORAGE-SIMPLIFY.
- Migration recipe for 7.x cloud-adapter consumers: mount local disk →
  filesystem storage → operator backup cron.

Updated frontmatter description so soulcraft.com/docs renders the
correct preview.

NOT IN THIS COMMIT (deliberate, lower-priority)

- src/storage/cacheManager.ts still references StorageType.S3 /
  REMOTE_API / OPFS as dead branches (23 sites). The branches are never
  reached in 8.0, but cleaning them would cascade through 5 consumers.
  Defer to a follow-up if the dead code surfaces as a real maintenance
  issue.
- src/config/storageAutoConfig.ts keeps its StorageType enum + autodetect
  for 7.x compat surface. Same reason: rewriting cascades through
  zeroConfig, extensibleConfig, sharedConfigManager. Defer.
- docs/MIGRATION-V3-TO-V4.md and docs/DEVELOPER_LEARNING_PATH.md still
  reference cloud adapters as historical artefacts. That's accurate —
  they describe how things used to be. Left as-is.
- @deprecated audit in src/ (10 files) deferred — audit each individually
  in a future polish pass.

VERIFICATION

- npx tsc --noEmit: clean
- npm test: 1408 / 1409 (same pre-existing race-condition outstanding from
  step 7; no regressions from this cleanup)
This commit is contained in:
David Snelling 2026-06-09 15:05:02 -07:00
parent 780fb6444b
commit 9f9a41599e
13 changed files with 162 additions and 5288 deletions
Download patch file Download diff file Expand all files Collapse all files

120
src/graph/graphAdjacencyIndex.ts
View file

@ -628,105 +628,43 @@ export class GraphAdjacencyIndex implements GraphIndexProvider {
// Note: LSM-trees will be recreated from storage via their own initialization
// Verb data will be loaded on-demand via UnifiedCache
// Adaptive loading strategy based on storage type
// Brainy 8.0: storage is always local (filesystem or memory) per
// BR-BRAINY-80-STORAGE-SIMPLIFY. Load all verbs at once.
const storageType = this.storage?.constructor.name || ''
const isLocalStorage =
storageType === 'FileSystemStorage' ||
storageType === 'MemoryStorage' ||
storageType === 'OPFSStorage'
let totalVerbs = 0
if (isLocalStorage) {
// Local storage: Load all verbs at once to avoid repeated getAllShardedFiles() calls
prodLog.info(
`GraphAdjacencyIndex: Using optimized strategy - load all verbs at once (${storageType})`
)
prodLog.info(`GraphAdjacencyIndex: Load all verbs at once (${storageType})`)
const result = await this.storage.getVerbs({
pagination: { limit: 10000000 } // Effectively unlimited for local development
})
const result = await this.storage.getVerbs({
pagination: { limit: 10000000 } // Effectively unlimited for local storage
})
// Add each verb to index
for (const verb of result.items) {
// Convert HNSWVerbWithMetadata to GraphVerb format
const graphVerb: GraphVerb = {
id: verb.id,
sourceId: verb.sourceId,
targetId: verb.targetId,
vector: verb.vector,
source: verb.sourceId,
target: verb.targetId,
verb: verb.verb,
createdAt: { seconds: Math.floor(verb.createdAt / 1000), nanoseconds: (verb.createdAt % 1000) * 1000000 },
updatedAt: { seconds: Math.floor(verb.updatedAt / 1000), nanoseconds: (verb.updatedAt % 1000) * 1000000 },
createdBy: verb.createdBy || { augmentation: 'unknown', version: '0.0.0' },
service: verb.service,
data: verb.data,
embedding: verb.vector,
confidence: verb.confidence,
weight: verb.weight
}
await this.addVerb(graphVerb)
totalVerbs++
for (const verb of result.items) {
const graphVerb: GraphVerb = {
id: verb.id,
sourceId: verb.sourceId,
targetId: verb.targetId,
vector: verb.vector,
source: verb.sourceId,
target: verb.targetId,
verb: verb.verb,
createdAt: { seconds: Math.floor(verb.createdAt / 1000), nanoseconds: (verb.createdAt % 1000) * 1000000 },
updatedAt: { seconds: Math.floor(verb.updatedAt / 1000), nanoseconds: (verb.updatedAt % 1000) * 1000000 },
createdBy: verb.createdBy || { augmentation: 'unknown', version: '0.0.0' },
service: verb.service,
data: verb.data,
embedding: verb.vector,
confidence: verb.confidence,
weight: verb.weight
}
prodLog.info(
`GraphAdjacencyIndex: Loaded ${totalVerbs.toLocaleString()} verbs at once (local storage)`
)
} else {
// Cloud storage: Use pagination with native cloud APIs (efficient)
prodLog.info(
`GraphAdjacencyIndex: Using cloud pagination strategy (${storageType})`
)
let hasMore = true
let cursor: string | undefined = undefined
const batchSize = 1000
while (hasMore) {
const result = await this.storage.getVerbs({
pagination: { limit: batchSize, cursor }
})
// Add each verb to index
for (const verb of result.items) {
// Convert HNSWVerbWithMetadata to GraphVerb format
const graphVerb: GraphVerb = {
id: verb.id,
sourceId: verb.sourceId,
targetId: verb.targetId,
vector: verb.vector,
source: verb.sourceId,
target: verb.targetId,
verb: verb.verb,
createdAt: { seconds: Math.floor(verb.createdAt / 1000), nanoseconds: (verb.createdAt % 1000) * 1000000 },
updatedAt: { seconds: Math.floor(verb.updatedAt / 1000), nanoseconds: (verb.updatedAt % 1000) * 1000000 },
createdBy: verb.createdBy || { augmentation: 'unknown', version: '0.0.0' },
service: verb.service,
data: verb.data,
embedding: verb.vector,
confidence: verb.confidence,
weight: verb.weight
}
await this.addVerb(graphVerb)
totalVerbs++
}
hasMore = result.hasMore
cursor = result.nextCursor
// Progress logging
if (totalVerbs % 10000 === 0) {
prodLog.info(`GraphAdjacencyIndex: Indexed ${totalVerbs} verbs...`)
}
}
prodLog.info(
`GraphAdjacencyIndex: Loaded ${totalVerbs.toLocaleString()} verbs via pagination (cloud storage)`
)
await this.addVerb(graphVerb)
totalVerbs++
}
prodLog.info(
`GraphAdjacencyIndex: Loaded ${totalVerbs.toLocaleString()} verbs (${storageType})`
)
const rebuildTime = Date.now() - this.rebuildStartTime
const memoryUsage = this.calculateMemoryUsage()

104
src/hnsw/hnswIndex.ts
View file

@ -1470,20 +1470,15 @@ export class JsHnswVectorIndex implements VectorIndexProvider {
)
}
// Step 4: Adaptive loading strategy based on storage type
// FileSystem/Memory/OPFS: Load all at once (avoids repeated getAllShardedFiles() calls)
// Cloud (GCS/S3/R2): Use pagination (efficient native cloud APIs)
// Step 4: Load all HNSW nodes at once. Brainy 8.0 ships filesystem +
// memory storage only (per BR-BRAINY-80-STORAGE-SIMPLIFY); the
// cloud-pagination rebuild path was deleted alongside the cloud adapters.
const storageType = this.storage?.constructor.name || ''
const isLocalStorage = storageType === 'FileSystemStorage' ||
storageType === 'MemoryStorage' ||
storageType === 'OPFSStorage'
let loadedCount = 0
let totalCount: number | undefined = undefined
if (isLocalStorage) {
// Local storage: Load all nouns at once
prodLog.info(`HNSW: Using optimized strategy - load all nodes at once (${storageType})`)
{
prodLog.info(`HNSW: Load all nodes at once (${storageType})`)
const result: {
items: HNSWNoun[]
@ -1554,94 +1549,7 @@ export class JsHnswVectorIndex implements VectorIndexProvider {
options.onProgress(loadedCount, totalCount)
}
prodLog.info(`HNSW: Loaded ${loadedCount.toLocaleString()} nodes at once (local storage)`)
} else {
// Cloud storage: Use pagination with native cloud APIs
prodLog.info(`HNSW: Using cloud pagination strategy (${storageType})`)
let hasMore = true
let offset = 0 // Use offset-based pagination instead of cursor (bug fix for infinite loop)
while (hasMore) {
// Fetch batch of nouns from storage (cast needed as method is not in base interface)
const result: {
items: HNSWNoun[]
totalCount?: number
hasMore: boolean
nextCursor?: string
} = await (this.storage as any).getNounsWithPagination({
limit: batchSize,
offset // Pass offset for proper pagination (previously passed cursor which was ignored)
})
// Set total count on first batch
if (totalCount === undefined && result.totalCount !== undefined) {
totalCount = result.totalCount
}
// Process each noun in the batch
for (const nounData of result.items) {
try {
// Load HNSW graph data for this entity
const hnswData = await (this.storage as any).getVectorIndexData(nounData.id)
if (!hnswData) {
// No HNSW data - skip (might be entity added before persistence)
continue
}
// Determine if vector should be kept in memory
const keepVector = shouldPreload && this.vectorStorageMode !== 'lazy'
// Create noun object with restored connections
const noun: HNSWNoun = {
id: nounData.id,
vector: keepVector ? nounData.vector : [], // Preload if dataset is small and not lazy
connections: new Map(),
level: hnswData.level
}
// Restore SQ8 quantized data if quantization is enabled
if (this.quantizationEnabled && nounData.vector.length > 0) {
const sq8 = quantizeSQ8(nounData.vector)
noun.quantizedVector = sq8.quantized
noun.codebookMin = sq8.min
noun.codebookMax = sq8.max
}
// Restore connections from persisted data — compressed blob path
// first, legacy JSON-array fallback for indexes written before
// graph link compression landed.
await this.restoreNodeConnections(nounData.id, hnswData, noun)
// Add to in-memory index
this.nouns.set(nounData.id, noun)
// Track high-level nodes for O(1) entry point selection
if (noun.level >= 2 && noun.level <= this.MAX_TRACKED_LEVELS) {
if (!this.highLevelNodes.has(noun.level)) {
this.highLevelNodes.set(noun.level, new Set())
}
this.highLevelNodes.get(noun.level)!.add(nounData.id)
}
loadedCount++
} catch (error) {
// Log error but continue (robust error recovery)
console.error(`Failed to rebuild HNSW data for ${nounData.id}:`, error)
}
}
// Report progress
if (options.onProgress && totalCount !== undefined) {
options.onProgress(loadedCount, totalCount)
}
// Check for more data
hasMore = result.hasMore
offset += batchSize // Increment offset for next page
}
prodLog.info(`HNSW: Loaded ${loadedCount.toLocaleString()} nodes (${storageType})`)
}
// Step 5: CRITICAL - Recover entry point if missing)

243
src/utils/metadataIndex.ts
View file

@ -3056,21 +3056,13 @@ export class MetadataIndexManager implements MetadataIndexProvider {
// Clear chunk manager cache
this.chunkManager.clearCache()
// Adaptive rebuild strategy based on storage adapter
// FileSystem/Memory/OPFS: Load all at once (avoids getAllShardedFiles() overhead on every batch)
// Cloud (GCS/S3/R2): Use pagination with small batches (prevent socket exhaustion)
const storageType = this.storage.constructor.name
const isLocalStorage = storageType === 'FileSystemStorage' ||
storageType === 'MemoryStorage' ||
storageType === 'OPFSStorage'
let nounLimit: number
// Brainy 8.0 ships filesystem + memory storage only. Load all nouns
// at once — the cloud-storage paginated branch was deleted alongside
// the cloud adapters in step 7.
let totalNounsProcessed = 0
if (isLocalStorage) {
// Load all nouns at once for local storage
// Avoids repeated directory scans in getAllShardedFiles()
prodLog.info(`⚡ Using optimized strategy: load all nouns at once (local storage)`)
{
prodLog.info(`⚡ Loading all nouns at once (local storage)`)
const result = await this.storage.getNouns({
pagination: { offset: 0, limit: 1000000 } // Effectively unlimited
})
@ -3085,7 +3077,6 @@ export class MetadataIndexManager implements MetadataIndexProvider {
metadataBatch = await this.storage.getMetadataBatch(nounIds)
prodLog.info(`✅ Loaded ${metadataBatch.size}/${nounIds.length} metadata objects`)
} else {
// Fallback to individual calls
metadataBatch = new Map()
for (const id of nounIds) {
try {
@ -3097,131 +3088,21 @@ export class MetadataIndexManager implements MetadataIndexProvider {
}
}
// 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, true)
localCount++
// Periodic safety flush every 5000 entities to cap memory during rebuild
// (Periodic safety flush removed in 7.22.0 — the underlying
// flushDirtyMetadata was a no-op since the 7.20.0 column-store
// refactor. Column store handles its own flushing.)
}
}
totalNounsProcessed = result.items.length
prodLog.info(`✅ Indexed ${totalNounsProcessed} nouns`)
} else {
// Cloud storage: use conservative batching
nounLimit = 25
prodLog.info(`⚡ Using conservative batch size: ${nounLimit} items/batch (cloud storage)`)
let nounOffset = 0
let hasMoreNouns = true
let consecutiveEmptyBatches = 0
const MAX_ITERATIONS = 10000
let iterations = 0
while (hasMoreNouns && iterations < MAX_ITERATIONS) {
iterations++
const result = await this.storage.getNouns({
pagination: { offset: nounOffset, limit: nounLimit }
})
// CRITICAL SAFETY CHECK: Prevent infinite loop on empty results
if (result.items.length === 0) {
consecutiveEmptyBatches++
if (consecutiveEmptyBatches >= 3) {
prodLog.warn('⚠️ Breaking metadata rebuild loop: received 3 consecutive empty batches')
break
}
// If hasMore is true but items are empty, it's likely a bug
if (result.hasMore) {
prodLog.warn(`⚠️ Storage returned empty items but hasMore=true at offset ${nounOffset}`)
hasMoreNouns = false // Force exit
break
}
} else {
consecutiveEmptyBatches = 0 // Reset counter on non-empty batch
}
// CRITICAL FIX: Use batch metadata reading to prevent socket exhaustion
const nounIds = result.items.map(noun => noun.id)
let metadataBatch: Map<string, any>
if (this.storage.getMetadataBatch) {
// Use batch reading if available (prevents socket exhaustion)
prodLog.info(`📦 Processing metadata batch ${Math.floor(totalNounsProcessed / nounLimit) + 1} (${nounIds.length} items)...`)
metadataBatch = await this.storage.getMetadataBatch(nounIds)
const successRate = ((metadataBatch.size / nounIds.length) * 100).toFixed(1)
prodLog.info(`✅ Batch loaded ${metadataBatch.size}/${nounIds.length} metadata objects (${successRate}% success)`)
} else {
// Fallback to individual calls with strict concurrency control
prodLog.warn(`⚠️ FALLBACK: Storage adapter missing getMetadataBatch - using individual calls with concurrency limit`)
metadataBatch = new Map()
const CONCURRENCY_LIMIT = 3 // Very conservative limit
for (let i = 0; i < nounIds.length; i += CONCURRENCY_LIMIT) {
const batch = nounIds.slice(i, i + CONCURRENCY_LIMIT)
const batchPromises = batch.map(async (id) => {
try {
const metadata = await this.storage.getNounMetadata(id)
return { id, metadata }
} catch (error) {
prodLog.debug(`Failed to read metadata for ${id}:`, error)
return { id, metadata: null }
}
})
const batchResults = await Promise.all(batchPromises)
for (const { id, metadata } of batchResults) {
if (metadata) {
metadataBatch.set(id, metadata)
}
}
// Yield between batches to prevent socket exhaustion
await this.yieldToEventLoop()
}
}
// Process the metadata batch
for (const noun of result.items) {
const metadata = metadataBatch.get(noun.id)
if (metadata) {
// Skip flush during rebuild for performance
await this.addToIndex(noun.id, metadata, true, true)
}
}
// Yield after processing the entire batch
await this.yieldToEventLoop()
totalNounsProcessed += result.items.length
hasMoreNouns = result.hasMore
nounOffset += nounLimit
// Progress logging and event loop yield after each batch
if (totalNounsProcessed % 100 === 0 || !hasMoreNouns) {
prodLog.debug(`📊 Indexed ${totalNounsProcessed} nouns...`)
}
await this.yieldToEventLoop()
}
// Check iteration limits for cloud storage
if (iterations >= MAX_ITERATIONS) {
prodLog.error(`❌ Metadata noun rebuild hit maximum iteration limit (${MAX_ITERATIONS}). This indicates a bug in storage pagination.`)
}
}
// Rebuild verb metadata indexes - same strategy as nouns
// Rebuild verb metadata indexes — same single-pass local strategy.
let totalVerbsProcessed = 0
if (isLocalStorage) {
// Load all verbs at once for local storage
{
prodLog.info(`⚡ Loading all verbs at once (local storage)`)
const result = await this.storage.getVerbs({
pagination: { offset: 0, limit: 1000000 } // Effectively unlimited
@ -3229,7 +3110,6 @@ export class MetadataIndexManager implements MetadataIndexProvider {
prodLog.info(`📦 Loading ${result.items.length} verbs with metadata...`)
// Get all verb metadata at once
const verbIds = result.items.map(verb => verb.id)
let verbMetadataBatch: Map<string, any>
@ -3248,127 +3128,24 @@ export class MetadataIndexManager implements MetadataIndexProvider {
}
}
// 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, true)
verbLocalCount++
// (Periodic safety flush removed in 7.22.0 — see noun loop above.)
}
}
totalVerbsProcessed = result.items.length
prodLog.info(`✅ Indexed ${totalVerbsProcessed} verbs`)
} else {
// Cloud storage: use conservative batching
let verbOffset = 0
const verbLimit = 25
let hasMoreVerbs = true
let consecutiveEmptyVerbBatches = 0
let verbIterations = 0
const MAX_ITERATIONS = 10000
while (hasMoreVerbs && verbIterations < MAX_ITERATIONS) {
verbIterations++
const result = await this.storage.getVerbs({
pagination: { offset: verbOffset, limit: verbLimit }
})
// CRITICAL SAFETY CHECK: Prevent infinite loop on empty results
if (result.items.length === 0) {
consecutiveEmptyVerbBatches++
if (consecutiveEmptyVerbBatches >= 3) {
prodLog.warn('⚠️ Breaking verb metadata rebuild loop: received 3 consecutive empty batches')
break
}
// If hasMore is true but items are empty, it's likely a bug
if (result.hasMore) {
prodLog.warn(`⚠️ Storage returned empty verb items but hasMore=true at offset ${verbOffset}`)
hasMoreVerbs = false // Force exit
break
}
} else {
consecutiveEmptyVerbBatches = 0 // Reset counter on non-empty batch
}
// CRITICAL FIX: Use batch verb metadata reading to prevent socket exhaustion
const verbIds = result.items.map(verb => verb.id)
let verbMetadataBatch: Map<string, any>
if ((this.storage as any).getVerbMetadataBatch) {
// Use batch reading if available (prevents socket exhaustion)
verbMetadataBatch = await (this.storage as any).getVerbMetadataBatch(verbIds)
prodLog.debug(`📦 Batch loaded ${verbMetadataBatch.size}/${verbIds.length} verb metadata objects`)
} else {
// Fallback to individual calls with strict concurrency control
verbMetadataBatch = new Map()
const CONCURRENCY_LIMIT = 3 // Very conservative limit to prevent socket exhaustion
for (let i = 0; i < verbIds.length; i += CONCURRENCY_LIMIT) {
const batch = verbIds.slice(i, i + CONCURRENCY_LIMIT)
const batchPromises = batch.map(async (id) => {
try {
const metadata = await this.storage.getVerbMetadata(id)
return { id, metadata }
} catch (error) {
prodLog.debug(`Failed to read verb metadata for ${id}:`, error)
return { id, metadata: null }
}
})
const batchResults = await Promise.all(batchPromises)
for (const { id, metadata } of batchResults) {
if (metadata) {
verbMetadataBatch.set(id, metadata)
}
}
// Yield between batches to prevent socket exhaustion
await this.yieldToEventLoop()
}
}
// Process the verb metadata batch
for (const verb of result.items) {
const metadata = verbMetadataBatch.get(verb.id)
if (metadata) {
// Skip flush during rebuild for performance
await this.addToIndex(verb.id, metadata, true, true)
}
}
// Yield after processing the entire batch
await this.yieldToEventLoop()
totalVerbsProcessed += result.items.length
hasMoreVerbs = result.hasMore
verbOffset += verbLimit
// Progress logging and event loop yield after each batch
if (totalVerbsProcessed % 100 === 0 || !hasMoreVerbs) {
prodLog.debug(`🔗 Indexed ${totalVerbsProcessed} verbs...`)
}
await this.yieldToEventLoop()
}
// Check iteration limits for cloud storage
if (verbIterations >= MAX_ITERATIONS) {
prodLog.error(`❌ Metadata verb rebuild hit maximum iteration limit (${MAX_ITERATIONS}). This indicates a bug in storage pagination.`)
}
}
// Flush to storage with final yield. The column store's flush() handles
// tail-buffer-to-segment promotion + manifest persistence.
// Flush to storage. The column store's flush() handles tail-buffer-to-
// segment promotion + manifest persistence.
prodLog.debug('💾 Flushing metadata index to storage...')
await this.flush()
await this.yieldToEventLoop()
prodLog.info(`✅ Metadata index rebuild completed! Processed ${totalNounsProcessed} nouns and ${totalVerbsProcessed} verbs`)
prodLog.info(`🎯 Initial indexing may show minor socket timeouts - this is expected and doesn't affect data processing`)
} finally {
this.isRebuilding = false
}