From ab2493af0295e376ac90dced4b782d8cf839f166 Mon Sep 17 00:00:00 2001 From: David Snelling Date: Sun, 1 Feb 2026 17:55:40 -0800 Subject: [PATCH] fix: flush graph LSM-trees on close to prevent data loss across restarts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit GraphAdjacencyIndex.flush() was a no-op — LSM MemTables were never written to SSTables for datasets under the 100K auto-flush threshold. This caused readdir, getRelations, and getDescendants to return empty results after close + reopen. Three fixes: - LSMTree.get(): merge MemTable + SSTable results (data spans both after flush, old early-return missed SSTable data) - GraphAdjacencyIndex.flush(): actually flush all 4 LSM-trees - GraphAdjacencyIndex.close(): close all 4 trees (was only closing 2) Also: brain.close() and shutdown hooks now call close() on graphIndex, HNSW index, and metadataIndex to release timers and file handles. --- src/brainy.ts | 46 ++++++++-- src/graph/graphAdjacencyIndex.ts | 29 +++++-- src/graph/lsm/LSMTree.ts | 27 +++--- tests/unit/vfs-restart-fix.test.ts | 133 +++++++++++++++++++++++++++++ 4 files changed, 213 insertions(+), 22 deletions(-) create mode 100644 tests/unit/vfs-restart-fix.test.ts 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 {} + } + }) +})