From dcf20ffa1dc1a9de357d34484611d1111cd24c93 Mon Sep 17 00:00:00 2001 From: David Snelling Date: Tue, 14 Oct 2025 13:06:32 -0700 Subject: [PATCH] fix: resolve import hang and index rebuild data loss bugs Critical Bug Fixes (v3.43.2): Bug #1 - Import Infinite Loop: - Fix placeholder entity infinite loop in ImportCoordinator - Use exact matching instead of fuzzy .includes() for entity names - Search entities array (not rows) for existing placeholders - Add duplicate relationship prevention in brain.relate() Bug #2 - Index Rebuild File Discovery: - Fix fileSystemStorage to scan sharded subdirectories - Update getAllNodes() to use getAllShardedFiles() - Update getAllEdges() to use getAllShardedFiles() - Update getNodesByNounType() to use getAllShardedFiles() - Fix getStorageStatus() to use O(1) persisted counts Additional Improvements: - Add brain.flush() API for explicit index persistence - Make GraphAdjacencyIndex.flush() public - Add auto-flush at end of import pipeline - Update duplicate relationship test to expect deduplication Files Modified: - src/storage/adapters/fileSystemStorage.ts - src/import/ImportCoordinator.ts - src/brainy.ts - src/graph/graphAdjacencyIndex.ts - tests/unit/brainy/relate.test.ts --- package-lock.json | 4 +- src/brainy.ts | 66 ++++++++++ src/graph/graphAdjacencyIndex.ts | 3 +- src/import/ImportCoordinator.ts | 27 +++- src/storage/adapters/fileSystemStorage.ts | 150 ++++++++++++---------- tests/unit/brainy/relate.test.ts | 17 ++- 6 files changed, 184 insertions(+), 83 deletions(-) diff --git a/package-lock.json b/package-lock.json index a884d623..c9d6e9dd 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@soulcraft/brainy", - "version": "3.43.0", + "version": "3.43.1", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@soulcraft/brainy", - "version": "3.43.0", + "version": "3.43.1", "license": "MIT", "dependencies": { "@aws-sdk/client-s3": "^3.540.0", diff --git a/src/brainy.ts b/src/brainy.ts index 67c8b4c0..73568d66 100644 --- a/src/brainy.ts +++ b/src/brainy.ts @@ -737,6 +737,21 @@ export class Brainy implements BrainyInterface { throw new Error(`Target entity ${params.to} not found`) } + // CRITICAL FIX (v3.43.2): Check for duplicate relationships + // This prevents infinite loops where same relationship is created repeatedly + // Bug #1 showed incrementing verb counts (7→8→9...) indicating duplicates + const existingVerbs = await this.storage.getVerbsBySource(params.from) + const duplicate = existingVerbs.find(v => + v.targetId === params.to && + v.type === params.type + ) + + if (duplicate) { + // Relationship already exists - return existing ID instead of creating duplicate + console.log(`[DEBUG] Skipping duplicate relationship: ${params.from} → ${params.to} (${params.type})`) + return duplicate.id + } + // Generate ID const id = uuidv4() @@ -1957,6 +1972,57 @@ export class Brainy implements BrainyInterface { } } + /** + * Flush all indexes and caches to persistent storage + * CRITICAL FIX (v3.43.2): Ensures data survives server restarts + * + * Flushes all 4 core indexes: + * 1. Storage counts (entity/verb counts by type) + * 2. Metadata index (field indexes + EntityIdMapper) + * 3. Graph adjacency index (relationship cache) + * 4. HNSW vector index (no flush needed - saves directly) + * + * @example + * // Flush after bulk operations + * await brain.import('./data.xlsx') + * await brain.flush() + * + * // Flush before shutdown + * process.on('SIGTERM', async () => { + * await brain.flush() + * process.exit(0) + * }) + */ + async flush(): Promise { + await this.ensureInitialized() + + console.log('🔄 Flushing Brainy indexes and caches to disk...') + + const startTime = Date.now() + + // Flush all components in parallel for performance + await Promise.all([ + // 1. Flush storage adapter counts (entity/verb counts by type) + (async () => { + if (this.storage && typeof (this.storage as any).flushCounts === 'function') { + await (this.storage as any).flushCounts() + } + })(), + + // 2. Flush metadata index (field indexes + EntityIdMapper) + this.metadataIndex.flush(), + + // 3. Flush graph adjacency index (relationship cache) + // Note: Graph structure is already persisted via storage.saveVerb() calls + // This just flushes the in-memory cache for performance + this.graphIndex.flush() + ]) + + const elapsed = Date.now() - startTime + + console.log(`✅ All indexes flushed to disk in ${elapsed}ms`) + } + /** * Efficient Pagination API - Production-scale pagination using index-first approach * Automatically optimizes based on query type and applies pagination at the index level diff --git a/src/graph/graphAdjacencyIndex.ts b/src/graph/graphAdjacencyIndex.ts index a43f1b0f..873c17b1 100644 --- a/src/graph/graphAdjacencyIndex.ts +++ b/src/graph/graphAdjacencyIndex.ts @@ -391,8 +391,9 @@ export class GraphAdjacencyIndex { /** * Flush dirty entries to cache + * CRITICAL FIX (v3.43.2): Now public so it can be called from brain.flush() */ - private async flush(): Promise { + async flush(): Promise { if (this.dirtySourceIds.size === 0 && this.dirtyTargetIds.size === 0) { return } diff --git a/src/import/ImportCoordinator.ts b/src/import/ImportCoordinator.ts index 65bcb507..81844728 100644 --- a/src/import/ImportCoordinator.ts +++ b/src/import/ImportCoordinator.ts @@ -320,6 +320,16 @@ export class ImportCoordinator { ) } + // CRITICAL FIX (v3.43.2): Auto-flush all indexes before returning + // Ensures imported data survives server restarts + // Bug #5: Import data was only in memory, lost on restart + options.onProgress?.({ + stage: 'complete', + message: 'Flushing indexes to disk...' + }) + + await this.brain.flush() + return result } @@ -590,10 +600,12 @@ export class ImportCoordinator { if (options.createRelationships && row.relationships) { for (const rel of row.relationships) { try { - // Find or create target entity + // CRITICAL FIX (v3.43.2): Prevent infinite placeholder creation loop + // Find or create target entity using EXACT matching only let targetEntityId: string | undefined - // Check if target already exists in our entities list + // STEP 1: Check if target already exists in entities list (includes placeholders) + // This prevents creating duplicate placeholders - the root cause of Bug #1 const existingTarget = entities.find(e => e.name.toLowerCase() === rel.to.toLowerCase() ) @@ -601,17 +613,19 @@ export class ImportCoordinator { if (existingTarget) { targetEntityId = existingTarget.id } else { - // Try to find in other extracted entities + // STEP 2: Try to find in extraction results (rows) + // FIX: Use EXACT matching instead of fuzzy .includes() + // Fuzzy matching caused false matches (e.g., "Entity_29" matching "Entity_297") for (const otherRow of rows) { const otherEntity = otherRow.entity || otherRow - if (rel.to.toLowerCase().includes(otherEntity.name.toLowerCase()) || - otherEntity.name.toLowerCase().includes(rel.to.toLowerCase())) { + if (otherEntity.name.toLowerCase() === rel.to.toLowerCase()) { targetEntityId = otherEntity.id break } } - // If still not found, create placeholder entity + // STEP 3: If still not found, create placeholder entity ONCE + // The placeholder is added to entities array, so future searches will find it if (!targetEntityId) { targetEntityId = await this.brain.add({ data: rel.to, @@ -624,6 +638,7 @@ export class ImportCoordinator { } }) + // CRITICAL: Add to entities array so future searches find it entities.push({ id: targetEntityId, name: rel.to, diff --git a/src/storage/adapters/fileSystemStorage.ts b/src/storage/adapters/fileSystemStorage.ts index ccf3c622..24d42a8c 100644 --- a/src/storage/adapters/fileSystemStorage.ts +++ b/src/storage/adapters/fileSystemStorage.ts @@ -293,34 +293,40 @@ export class FileSystemStorage extends BaseStorage { /** * Get all nodes from storage + * CRITICAL FIX (v3.43.2): Now scans sharded subdirectories (depth=1) + * Previously only scanned flat directory, causing rebuild to find 0 entities */ protected async getAllNodes(): Promise { await this.ensureInitialized() const allNodes: HNSWNode[] = [] try { - const files = await fs.promises.readdir(this.nounsDir) + // FIX: Use sharded file discovery instead of flat directory read + // This scans all 256 shard subdirectories (00-ff) to find actual files + const files = await this.getAllShardedFiles(this.nounsDir) + for (const file of files) { - if (file.endsWith('.json')) { - const filePath = path.join(this.nounsDir, file) - const data = await fs.promises.readFile(filePath, 'utf-8') - const parsedNode = JSON.parse(data) + // Extract ID from filename and use sharded path + const id = file.replace('.json', '') + const filePath = this.getNodePath(id) - // Convert serialized connections back to Map> - const connections = new Map>() - for (const [level, nodeIds] of Object.entries( - parsedNode.connections - )) { - connections.set(Number(level), new Set(nodeIds as string[])) - } + const data = await fs.promises.readFile(filePath, 'utf-8') + const parsedNode = JSON.parse(data) - allNodes.push({ - id: parsedNode.id, - vector: parsedNode.vector, - connections, - level: parsedNode.level || 0 - }) + // Convert serialized connections back to Map> + const connections = new Map>() + for (const [level, nodeIds] of Object.entries( + parsedNode.connections + )) { + connections.set(Number(level), new Set(nodeIds as string[])) } + + allNodes.push({ + id: parsedNode.id, + vector: parsedNode.vector, + connections, + level: parsedNode.level || 0 + }) } } catch (error: any) { if (error.code !== 'ENOENT') { @@ -332,6 +338,7 @@ export class FileSystemStorage extends BaseStorage { /** * Get nodes by noun type + * CRITICAL FIX (v3.43.2): Now scans sharded subdirectories (depth=1) * @param nounType The noun type to filter by * @returns Promise that resolves to an array of nodes of the specified noun type */ @@ -340,32 +347,34 @@ export class FileSystemStorage extends BaseStorage { const nouns: HNSWNode[] = [] try { - const files = await fs.promises.readdir(this.nounsDir) + // FIX: Use sharded file discovery instead of flat directory read + const files = await this.getAllShardedFiles(this.nounsDir) + for (const file of files) { - if (file.endsWith('.json')) { - const filePath = path.join(this.nounsDir, file) - const data = await fs.promises.readFile(filePath, 'utf-8') - const parsedNode = JSON.parse(data) + // Extract ID from filename and use sharded path + const nodeId = file.replace('.json', '') + const filePath = this.getNodePath(nodeId) - // Filter by noun type using metadata - const nodeId = parsedNode.id - const metadata = await this.getMetadata(nodeId) - if (metadata && metadata.noun === nounType) { - // Convert serialized connections back to Map> - const connections = new Map>() - for (const [level, nodeIds] of Object.entries( - parsedNode.connections - )) { - connections.set(Number(level), new Set(nodeIds as string[])) - } + const data = await fs.promises.readFile(filePath, 'utf-8') + const parsedNode = JSON.parse(data) - nouns.push({ - id: parsedNode.id, - vector: parsedNode.vector, - connections, - level: parsedNode.level || 0 - }) + // Filter by noun type using metadata + const metadata = await this.getMetadata(nodeId) + if (metadata && metadata.noun === nounType) { + // Convert serialized connections back to Map> + const connections = new Map>() + for (const [level, nodeIds] of Object.entries( + parsedNode.connections + )) { + connections.set(Number(level), new Set(nodeIds as string[])) } + + nouns.push({ + id: parsedNode.id, + vector: parsedNode.vector, + connections, + level: parsedNode.level || 0 + }) } } } catch (error: any) { @@ -482,33 +491,39 @@ export class FileSystemStorage extends BaseStorage { /** * Get all edges from storage + * CRITICAL FIX (v3.43.2): Now scans sharded subdirectories (depth=1) + * Previously only scanned flat directory, causing rebuild to find 0 relationships */ protected async getAllEdges(): Promise { await this.ensureInitialized() const allEdges: Edge[] = [] try { - const files = await fs.promises.readdir(this.verbsDir) + // FIX: Use sharded file discovery instead of flat directory read + // This scans all 256 shard subdirectories (00-ff) to find actual files + const files = await this.getAllShardedFiles(this.verbsDir) + for (const file of files) { - if (file.endsWith('.json')) { - const filePath = path.join(this.verbsDir, file) - const data = await fs.promises.readFile(filePath, 'utf-8') - const parsedEdge = JSON.parse(data) + // Extract ID from filename and use sharded path + const id = file.replace('.json', '') + const filePath = this.getVerbPath(id) - // Convert serialized connections back to Map> - const connections = new Map>() - for (const [level, nodeIds] of Object.entries( - parsedEdge.connections - )) { - connections.set(Number(level), new Set(nodeIds as string[])) - } + const data = await fs.promises.readFile(filePath, 'utf-8') + const parsedEdge = JSON.parse(data) - allEdges.push({ - id: parsedEdge.id, - vector: parsedEdge.vector, - connections - }) + // Convert serialized connections back to Map> + const connections = new Map>() + for (const [level, nodeIds] of Object.entries( + parsedEdge.connections + )) { + connections.set(Number(level), new Set(nodeIds as string[])) } + + allEdges.push({ + id: parsedEdge.id, + vector: parsedEdge.vector, + connections + }) } } catch (error: any) { if (error.code !== 'ENOENT') { @@ -974,20 +989,21 @@ export class FileSystemStorage extends BaseStorage { totalSize = nounsDirSize + verbsDirSize + metadataDirSize + indexDirSize - // Count files in each directory - const nounsCount = (await fs.promises.readdir(this.nounsDir)).filter( - (file: string) => file.endsWith('.json') - ).length - const verbsCount = (await fs.promises.readdir(this.verbsDir)).filter( - (file: string) => file.endsWith('.json') - ).length + // CRITICAL FIX (v3.43.2): Use persisted counts instead of directory reads + // This is O(1) instead of O(n), and handles sharded structure correctly + const nounsCount = this.totalNounCount + const verbsCount = this.totalVerbCount + + // Count metadata files (these are NOT sharded) const metadataCount = ( await fs.promises.readdir(this.metadataDir) ).filter((file: string) => file.endsWith('.json')).length - // Count nouns by type using metadata - const nounTypeCounts: Record = {} - const metadataFiles = await fs.promises.readdir(this.metadataDir) + // Use persisted entity counts by type (O(1) instead of scanning all files) + const nounTypeCounts: Record = Object.fromEntries(this.entityCounts) + + // Skip the expensive metadata file scan since we have counts + const metadataFiles: string[] = [] // Empty array to skip the loop below for (const file of metadataFiles) { if (file.endsWith('.json')) { try { diff --git a/tests/unit/brainy/relate.test.ts b/tests/unit/brainy/relate.test.ts index 9df5ed60..8db9511c 100644 --- a/tests/unit/brainy/relate.test.ts +++ b/tests/unit/brainy/relate.test.ts @@ -223,26 +223,29 @@ describe('Brainy.relate()', () => { describe('edge cases', () => { it('should handle duplicate relationships', async () => { // Act - Create same relationship twice - await brain.relate({ + const id1 = await brain.relate({ from: entity1Id, to: entity2Id, type: 'worksWith', weight: 0.5 }) - - await brain.relate({ + + const id2 = await brain.relate({ from: entity1Id, to: entity2Id, type: 'worksWith', weight: 0.8 }) - - // Assert - Both relationships should exist + + // Assert - Should prevent duplicates (v3.43.2 bug fix) + // Second call should return existing relationship ID instead of creating duplicate + expect(id1).toBe(id2) + const relations = await brain.getRelations({ from: entity1Id }) - const matches = relations.filter(r => + const matches = relations.filter(r => r.to === entity2Id && r.type === 'worksWith' ) - expect(matches.length).toBe(2) + expect(matches.length).toBe(1) // Only one relationship should exist }) it('should handle very long metadata', async () => {