diff --git a/src/brainy.ts b/src/brainy.ts index 33dc5fed..c3f3e962 100644 --- a/src/brainy.ts +++ b/src/brainy.ts @@ -461,7 +461,7 @@ export class Brainy implements BrainyInterface { let flushedCount = 0 for (const instance of Brainy.instances) { if (instance.initialized) { - // Full flush: counts + metadata index + graph index + HNSW dirty nodes + // Flush all buffered data, then close to release resources (timers, handles) await Promise.all([ (async () => { if (instance.storage && typeof (instance.storage as any).flushCounts === 'function') { @@ -484,6 +484,24 @@ export class Brainy implements BrainyInterface { } })() ]) + // Close components to stop timers that would prevent clean process exit + await Promise.all([ + (async () => { + if (instance.graphIndex && typeof instance.graphIndex.close === 'function') { + await instance.graphIndex.close() + } + })(), + (async () => { + if (instance.index && typeof (instance.index as any).close === 'function') { + await (instance.index as any).close() + } + })(), + (async () => { + if (instance.metadataIndex && typeof (instance.metadataIndex as any).close === 'function') { + await (instance.metadataIndex as any).close() + } + })(), + ]) flushedCount++ } } @@ -6702,7 +6720,7 @@ export class Brainy implements BrainyInterface { * This ensures deferred persistence mode data is saved */ async close(): Promise { - // Flush ALL components before closing to prevent data loss + // Phase 1: Flush ALL components in parallel to persist buffered data // This is critical when cortex native providers buffer data in Rust memory await Promise.all([ // Flush HNSW dirty nodes (deferred persistence mode) @@ -6731,7 +6749,27 @@ export class Brainy implements BrainyInterface { })() ]) - // Deactivate plugins (safe — all data flushed above) + // Phase 2: Close components to release resources (timers, file handles) + // Data is already safe on disk from Phase 1 + await Promise.all([ + (async () => { + if (this.graphIndex && typeof this.graphIndex.close === 'function') { + await this.graphIndex.close() + } + })(), + (async () => { + if (this.index && typeof (this.index as any).close === 'function') { + await (this.index as any).close() + } + })(), + (async () => { + if (this.metadataIndex && typeof (this.metadataIndex as any).close === 'function') { + await (this.metadataIndex as any).close() + } + })(), + ]) + + // Deactivate plugins (safe — all data flushed and resources released above) await this.pluginRegistry.deactivateAll() // Restore console methods if silent mode was enabled @@ -6751,8 +6789,6 @@ export class Brainy implements BrainyInterface { } } - // Storage doesn't have close in current interface - // We'll just mark as not initialized this.initialized = false } diff --git a/src/graph/graphAdjacencyIndex.ts b/src/graph/graphAdjacencyIndex.ts index 3c1a4986..59daac7f 100644 --- a/src/graph/graphAdjacencyIndex.ts +++ b/src/graph/graphAdjacencyIndex.ts @@ -800,10 +800,21 @@ export class GraphAdjacencyIndex { const startTime = Date.now() - // Flush both LSM-trees - // Note: LSMTree.close() will handle flushing MemTable - // For now, we don't have an explicit flush method in LSMTree - // The MemTable will be flushed automatically when threshold is reached + // Flush all 4 LSM-trees in parallel (MemTables → SSTables on disk) + await Promise.all([ + this.lsmTreeSource.flush().then(() => { + prodLog.debug(`GraphAdjacencyIndex: Flushed source tree`) + }), + this.lsmTreeTarget.flush().then(() => { + prodLog.debug(`GraphAdjacencyIndex: Flushed target tree`) + }), + this.lsmTreeVerbsBySource.flush().then(() => { + prodLog.debug(`GraphAdjacencyIndex: Flushed verbs-by-source tree`) + }), + this.lsmTreeVerbsByTarget.flush().then(() => { + prodLog.debug(`GraphAdjacencyIndex: Flushed verbs-by-target tree`) + }), + ]) const elapsed = Date.now() - startTime @@ -819,10 +830,14 @@ export class GraphAdjacencyIndex { this.flushTimer = undefined } - // Close LSM-trees (will flush MemTables) + // Close all 4 LSM-trees (will flush MemTables to SSTables) if (this.initialized) { - await this.lsmTreeSource.close() - await this.lsmTreeTarget.close() + await Promise.all([ + this.lsmTreeSource.close(), + this.lsmTreeTarget.close(), + this.lsmTreeVerbsBySource.close(), + this.lsmTreeVerbsByTarget.close(), + ]) } prodLog.info('GraphAdjacencyIndex: Shutdown complete') diff --git a/src/graph/lsm/LSMTree.ts b/src/graph/lsm/LSMTree.ts index 7bfc50a7..20933769 100644 --- a/src/graph/lsm/LSMTree.ts +++ b/src/graph/lsm/LSMTree.ts @@ -296,18 +296,21 @@ export class LSMTree { async get(sourceId: string): Promise { const startTime = performance.now() - // Check MemTable first (hot data) + // Merge results from MemTable AND SSTables + // Data can span both after a flush (old data in SSTables, new in MemTable) + const allTargets = new Set() + + // Check MemTable (hot data) const memResult = this.memTable.get(sourceId) if (memResult !== null) { - return memResult + for (const target of memResult) { + allTargets.add(target) + } } // Check SSTables from newest to oldest - // Newer levels (L0, L1, L2) checked first for better cache locality const maxLevel = Math.max(...Array.from(this.sstablesByLevel.keys()), 0) - const allTargets = new Set() - for (let level = 0; level <= maxLevel; level++) { const sstables = this.sstablesByLevel.get(level) || [] @@ -608,16 +611,20 @@ export class LSMTree { } /** - * Flush MemTable and stop compaction - * Called during shutdown + * Flush MemTable to SSTables without closing + * Called by GraphAdjacencyIndex.flush() and brain.close() */ + async flush(): Promise { + if (!this.memTable.isEmpty()) { + await this.flushMemTable() + } + } + async close(): Promise { this.stopCompactionTimer() // Final MemTable flush - if (!this.memTable.isEmpty()) { - await this.flushMemTable() - } + await this.flush() prodLog.info('LSMTree: Closed successfully') } diff --git a/tests/unit/vfs-restart-fix.test.ts b/tests/unit/vfs-restart-fix.test.ts new file mode 100644 index 00000000..83f1fae3 --- /dev/null +++ b/tests/unit/vfs-restart-fix.test.ts @@ -0,0 +1,133 @@ +/** + * Regression test: VFS data persists across restart (close + reopen) + * + * Verifies the fix for GraphAdjacencyIndex.flush() and LSMTree.get() + * which previously caused graph relationships to be lost on restart + * because LSM MemTables were never flushed to SSTables on close. + */ +import { describe, it, expect } from 'vitest' +import { Brainy } from '../../src/brainy.js' +import { VerbType } from '../../src/types/graphTypes.js' +import { mkdirSync, rmSync } from 'fs' + +describe('VFS restart persistence', () => { + it('entities and relationships survive close + reopen', async () => { + const dir = '/tmp/brainy-restart-fix-' + Date.now() + mkdirSync(dir, { recursive: true }) + + let entityId: string | undefined + const rootId = '00000000-0000-0000-0000-000000000000' + + try { + // === SESSION 1: Write data === + let brain = new Brainy({ + storage: { type: 'filesystem', rootDirectory: dir }, + disableAutoRebuild: true, + plugins: [], + silent: true, + }) + await brain.init() + + await brain.vfs.writeFile('/chapter-1.txt', 'Once upon a time...') + + entityId = await brain.vfs.resolvePathToId('/chapter-1.txt') + expect(entityId).toBeTruthy() + + // Verify within same session + const entity1 = await brain.get(entityId!) + expect(entity1).not.toBeNull() + + const entries1 = await brain.vfs.readdir('/') + expect(entries1.length).toBeGreaterThan(0) + + await brain.close() + + // === SESSION 2: Read data after restart === + brain = new Brainy({ + storage: { type: 'filesystem', rootDirectory: dir }, + disableAutoRebuild: true, + plugins: [], + silent: true, + }) + await brain.init() + + // Entity should be retrievable + const entity2 = await brain.get(entityId!) + expect(entity2).not.toBeNull() + + // Root entity should exist + const rootEntity = await brain.get(rootId) + expect(rootEntity).not.toBeNull() + + // readdir should return the file + const entries2 = await brain.vfs.readdir('/') + expect(entries2.length).toBeGreaterThan(0) + expect(entries2).toContain('chapter-1.txt') + + // getRelations should find the Contains relationship + const relations2 = await brain.getRelations({ from: rootId, type: VerbType.Contains }) + expect(relations2.length).toBeGreaterThan(0) + + await brain.close() + } finally { + try { rmSync(dir, { recursive: true }) } catch {} + } + }) + + it('multiple files and subdirectories survive restart', async () => { + const dir = '/tmp/brainy-restart-multi-' + Date.now() + mkdirSync(dir, { recursive: true }) + + const rootId = '00000000-0000-0000-0000-000000000000' + + try { + // === SESSION 1: Write multiple files === + let brain = new Brainy({ + storage: { type: 'filesystem', rootDirectory: dir }, + disableAutoRebuild: true, + plugins: [], + silent: true, + }) + await brain.init() + + await brain.vfs.writeFile('/readme.txt', 'Project readme') + await brain.vfs.writeFile('/docs/guide.txt', 'User guide') + await brain.vfs.writeFile('/docs/api.txt', 'API reference') + + const entries1 = await brain.vfs.readdir('/') + expect(entries1.length).toBe(2) // readme.txt + docs/ + + const docEntries1 = await brain.vfs.readdir('/docs') + expect(docEntries1.length).toBe(2) // guide.txt + api.txt + + await brain.close() + + // === SESSION 2: Verify all data persisted === + brain = new Brainy({ + storage: { type: 'filesystem', rootDirectory: dir }, + disableAutoRebuild: true, + plugins: [], + silent: true, + }) + await brain.init() + + // Root children + const entries2 = await brain.vfs.readdir('/') + expect(entries2.length).toBe(2) + expect(entries2.sort()).toEqual(['docs', 'readme.txt']) + + // Subdirectory children + const docEntries2 = await brain.vfs.readdir('/docs') + expect(docEntries2.length).toBe(2) + expect(docEntries2.sort()).toEqual(['api.txt', 'guide.txt']) + + // Root relations + const rootRelations = await brain.getRelations({ from: rootId, type: VerbType.Contains }) + expect(rootRelations.length).toBeGreaterThan(0) + + await brain.close() + } finally { + try { rmSync(dir, { recursive: true }) } catch {} + } + }) +})