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
This commit is contained in:
David Snelling 2025-10-14 13:06:32 -07:00
parent 165def11a9
commit dcf20ffa1d
6 changed files with 184 additions and 83 deletions

4
package-lock.json generated
View file

@ -1,12 +1,12 @@
{ {
"name": "@soulcraft/brainy", "name": "@soulcraft/brainy",
"version": "3.43.0", "version": "3.43.1",
"lockfileVersion": 3, "lockfileVersion": 3,
"requires": true, "requires": true,
"packages": { "packages": {
"": { "": {
"name": "@soulcraft/brainy", "name": "@soulcraft/brainy",
"version": "3.43.0", "version": "3.43.1",
"license": "MIT", "license": "MIT",
"dependencies": { "dependencies": {
"@aws-sdk/client-s3": "^3.540.0", "@aws-sdk/client-s3": "^3.540.0",

View file

@ -737,6 +737,21 @@ export class Brainy<T = any> implements BrainyInterface<T> {
throw new Error(`Target entity ${params.to} not found`) 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 // Generate ID
const id = uuidv4() const id = uuidv4()
@ -1957,6 +1972,57 @@ export class Brainy<T = any> implements BrainyInterface<T> {
} }
} }
/**
* 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<void> {
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 * Efficient Pagination API - Production-scale pagination using index-first approach
* Automatically optimizes based on query type and applies pagination at the index level * Automatically optimizes based on query type and applies pagination at the index level

View file

@ -391,8 +391,9 @@ export class GraphAdjacencyIndex {
/** /**
* Flush dirty entries to cache * Flush dirty entries to cache
* CRITICAL FIX (v3.43.2): Now public so it can be called from brain.flush()
*/ */
private async flush(): Promise<void> { async flush(): Promise<void> {
if (this.dirtySourceIds.size === 0 && this.dirtyTargetIds.size === 0) { if (this.dirtySourceIds.size === 0 && this.dirtyTargetIds.size === 0) {
return return
} }

View file

@ -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 return result
} }
@ -590,10 +600,12 @@ export class ImportCoordinator {
if (options.createRelationships && row.relationships) { if (options.createRelationships && row.relationships) {
for (const rel of row.relationships) { for (const rel of row.relationships) {
try { 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 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 => const existingTarget = entities.find(e =>
e.name.toLowerCase() === rel.to.toLowerCase() e.name.toLowerCase() === rel.to.toLowerCase()
) )
@ -601,17 +613,19 @@ export class ImportCoordinator {
if (existingTarget) { if (existingTarget) {
targetEntityId = existingTarget.id targetEntityId = existingTarget.id
} else { } 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) { for (const otherRow of rows) {
const otherEntity = otherRow.entity || otherRow const otherEntity = otherRow.entity || otherRow
if (rel.to.toLowerCase().includes(otherEntity.name.toLowerCase()) || if (otherEntity.name.toLowerCase() === rel.to.toLowerCase()) {
otherEntity.name.toLowerCase().includes(rel.to.toLowerCase())) {
targetEntityId = otherEntity.id targetEntityId = otherEntity.id
break 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) { if (!targetEntityId) {
targetEntityId = await this.brain.add({ targetEntityId = await this.brain.add({
data: rel.to, data: rel.to,
@ -624,6 +638,7 @@ export class ImportCoordinator {
} }
}) })
// CRITICAL: Add to entities array so future searches find it
entities.push({ entities.push({
id: targetEntityId, id: targetEntityId,
name: rel.to, name: rel.to,

View file

@ -293,34 +293,40 @@ export class FileSystemStorage extends BaseStorage {
/** /**
* Get all nodes from storage * 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<HNSWNode[]> { protected async getAllNodes(): Promise<HNSWNode[]> {
await this.ensureInitialized() await this.ensureInitialized()
const allNodes: HNSWNode[] = [] const allNodes: HNSWNode[] = []
try { 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) { for (const file of files) {
if (file.endsWith('.json')) { // Extract ID from filename and use sharded path
const filePath = path.join(this.nounsDir, file) const id = file.replace('.json', '')
const data = await fs.promises.readFile(filePath, 'utf-8') const filePath = this.getNodePath(id)
const parsedNode = JSON.parse(data)
// Convert serialized connections back to Map<number, Set<string>> const data = await fs.promises.readFile(filePath, 'utf-8')
const connections = new Map<number, Set<string>>() const parsedNode = JSON.parse(data)
for (const [level, nodeIds] of Object.entries(
parsedNode.connections
)) {
connections.set(Number(level), new Set(nodeIds as string[]))
}
allNodes.push({ // Convert serialized connections back to Map<number, Set<string>>
id: parsedNode.id, const connections = new Map<number, Set<string>>()
vector: parsedNode.vector, for (const [level, nodeIds] of Object.entries(
connections, parsedNode.connections
level: parsedNode.level || 0 )) {
}) 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) { } catch (error: any) {
if (error.code !== 'ENOENT') { if (error.code !== 'ENOENT') {
@ -332,6 +338,7 @@ export class FileSystemStorage extends BaseStorage {
/** /**
* Get nodes by noun type * Get nodes by noun type
* CRITICAL FIX (v3.43.2): Now scans sharded subdirectories (depth=1)
* @param nounType The noun type to filter by * @param nounType The noun type to filter by
* @returns Promise that resolves to an array of nodes of the specified noun type * @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[] = [] const nouns: HNSWNode[] = []
try { 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) { for (const file of files) {
if (file.endsWith('.json')) { // Extract ID from filename and use sharded path
const filePath = path.join(this.nounsDir, file) const nodeId = file.replace('.json', '')
const data = await fs.promises.readFile(filePath, 'utf-8') const filePath = this.getNodePath(nodeId)
const parsedNode = JSON.parse(data)
// Filter by noun type using metadata const data = await fs.promises.readFile(filePath, 'utf-8')
const nodeId = parsedNode.id const parsedNode = JSON.parse(data)
const metadata = await this.getMetadata(nodeId)
if (metadata && metadata.noun === nounType) {
// Convert serialized connections back to Map<number, Set<string>>
const connections = new Map<number, Set<string>>()
for (const [level, nodeIds] of Object.entries(
parsedNode.connections
)) {
connections.set(Number(level), new Set(nodeIds as string[]))
}
nouns.push({ // Filter by noun type using metadata
id: parsedNode.id, const metadata = await this.getMetadata(nodeId)
vector: parsedNode.vector, if (metadata && metadata.noun === nounType) {
connections, // Convert serialized connections back to Map<number, Set<string>>
level: parsedNode.level || 0 const connections = new Map<number, Set<string>>()
}) 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) { } catch (error: any) {
@ -482,33 +491,39 @@ export class FileSystemStorage extends BaseStorage {
/** /**
* Get all edges from storage * 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<Edge[]> { protected async getAllEdges(): Promise<Edge[]> {
await this.ensureInitialized() await this.ensureInitialized()
const allEdges: Edge[] = [] const allEdges: Edge[] = []
try { 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) { for (const file of files) {
if (file.endsWith('.json')) { // Extract ID from filename and use sharded path
const filePath = path.join(this.verbsDir, file) const id = file.replace('.json', '')
const data = await fs.promises.readFile(filePath, 'utf-8') const filePath = this.getVerbPath(id)
const parsedEdge = JSON.parse(data)
// Convert serialized connections back to Map<number, Set<string>> const data = await fs.promises.readFile(filePath, 'utf-8')
const connections = new Map<number, Set<string>>() const parsedEdge = JSON.parse(data)
for (const [level, nodeIds] of Object.entries(
parsedEdge.connections
)) {
connections.set(Number(level), new Set(nodeIds as string[]))
}
allEdges.push({ // Convert serialized connections back to Map<number, Set<string>>
id: parsedEdge.id, const connections = new Map<number, Set<string>>()
vector: parsedEdge.vector, for (const [level, nodeIds] of Object.entries(
connections parsedEdge.connections
}) )) {
connections.set(Number(level), new Set(nodeIds as string[]))
} }
allEdges.push({
id: parsedEdge.id,
vector: parsedEdge.vector,
connections
})
} }
} catch (error: any) { } catch (error: any) {
if (error.code !== 'ENOENT') { if (error.code !== 'ENOENT') {
@ -974,20 +989,21 @@ export class FileSystemStorage extends BaseStorage {
totalSize = nounsDirSize + verbsDirSize + metadataDirSize + indexDirSize totalSize = nounsDirSize + verbsDirSize + metadataDirSize + indexDirSize
// Count files in each directory // CRITICAL FIX (v3.43.2): Use persisted counts instead of directory reads
const nounsCount = (await fs.promises.readdir(this.nounsDir)).filter( // This is O(1) instead of O(n), and handles sharded structure correctly
(file: string) => file.endsWith('.json') const nounsCount = this.totalNounCount
).length const verbsCount = this.totalVerbCount
const verbsCount = (await fs.promises.readdir(this.verbsDir)).filter(
(file: string) => file.endsWith('.json') // Count metadata files (these are NOT sharded)
).length
const metadataCount = ( const metadataCount = (
await fs.promises.readdir(this.metadataDir) await fs.promises.readdir(this.metadataDir)
).filter((file: string) => file.endsWith('.json')).length ).filter((file: string) => file.endsWith('.json')).length
// Count nouns by type using metadata // Use persisted entity counts by type (O(1) instead of scanning all files)
const nounTypeCounts: Record<string, number> = {} const nounTypeCounts: Record<string, number> = Object.fromEntries(this.entityCounts)
const metadataFiles = await fs.promises.readdir(this.metadataDir)
// 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) { for (const file of metadataFiles) {
if (file.endsWith('.json')) { if (file.endsWith('.json')) {
try { try {

View file

@ -223,26 +223,29 @@ describe('Brainy.relate()', () => {
describe('edge cases', () => { describe('edge cases', () => {
it('should handle duplicate relationships', async () => { it('should handle duplicate relationships', async () => {
// Act - Create same relationship twice // Act - Create same relationship twice
await brain.relate({ const id1 = await brain.relate({
from: entity1Id, from: entity1Id,
to: entity2Id, to: entity2Id,
type: 'worksWith', type: 'worksWith',
weight: 0.5 weight: 0.5
}) })
await brain.relate({ const id2 = await brain.relate({
from: entity1Id, from: entity1Id,
to: entity2Id, to: entity2Id,
type: 'worksWith', type: 'worksWith',
weight: 0.8 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 relations = await brain.getRelations({ from: entity1Id })
const matches = relations.filter(r => const matches = relations.filter(r =>
r.to === entity2Id && r.type === 'worksWith' 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 () => { it('should handle very long metadata', async () => {