/** * @module graph/lsm/LSMTree * @description Log-Structured Merge tree for the JS (open-core) graph store — the * fallback used when no native graph provider is registered. Verb-id postings are * buffered in an in-memory MemTable, flushed to immutable sorted SSTables, and * background-compacted; bloom filters give fast negative lookups. * * Architecture: * - MemTable: in-memory write buffer (flush threshold configurable, default 100K) * - SSTables: immutable sorted segments, persisted via the StorageAdapter * - Bloom filters: in-memory membership pre-checks * - Compaction: background merge of SSTables (reclaims superseded segments) * * Complexity: O(1) MemTable writes; O(log n) reads with bloom-filter pre-filtering. * Note: this JS implementation loads its SSTables into memory after open (it is the * fallback engine). Billion-scale, on-disk-resident operation is the native graph * provider's role behind the provider boundary, not this fallback's; no absolute * memory/latency figures are claimed here without a cited benchmark. */ import { StorageAdapter } from '../../coreTypes.js' import { SSTable, SSTableEntry } from './SSTable.js' import { prodLog } from '../../utils/logger.js' /** * LSMTree configuration */ export interface LSMTreeConfig { /** * MemTable flush threshold (number of relationships) * Default: 100000 (100K relationships, ~24MB RAM) */ memTableThreshold?: number /** * Maximum number of SSTables at each level before compaction * Default: 10 */ maxSSTablesPerLevel?: number /** * Storage key prefix for SSTables * Default: 'graph-lsm' */ storagePrefix?: string /** * Enable background compaction * Default: true */ enableCompaction?: boolean /** * Compaction interval in milliseconds * Default: 60000 (1 minute) */ compactionInterval?: number } /** * In-memory write buffer (MemTable) * Stores recent writes before flushing to SSTable */ class MemTable { /** * sourceId → targetIds */ private data: Map> /** * Number of relationships in MemTable */ private count: number constructor() { this.data = new Map() this.count = 0 } /** * Add a relationship */ add(sourceId: string, targetId: string): void { if (!this.data.has(sourceId)) { this.data.set(sourceId, new Set()) } const targets = this.data.get(sourceId)! if (!targets.has(targetId)) { targets.add(targetId) this.count++ } } /** * Get targets for a sourceId */ get(sourceId: string): string[] | null { const targets = this.data.get(sourceId) return targets ? Array.from(targets) : null } /** * Get all entries as Map for flushing */ getAll(): Map> { return this.data } /** * Get number of relationships */ size(): number { return this.count } /** * Check if empty */ isEmpty(): boolean { return this.count === 0 } /** * Clear all data */ clear(): void { this.data.clear() this.count = 0 } /** * Estimate memory usage */ estimateMemoryUsage(): number { let bytes = 0 this.data.forEach((targets, sourceId) => { bytes += sourceId.length * 2 // UTF-16 bytes += targets.size * 40 // ~40 bytes per UUID }) return bytes } } /** * Manifest - Tracks all SSTables and their levels */ /** * Persisted manifest payload as written by `saveManifest()` (the `data` field * of the manifest metadata record, after a JSON round-trip). */ type PersistedManifestData = { sstables?: Record lastCompaction?: number totalRelationships?: number } /** * Persisted SSTable payload as written by `flushMemTable()`/`compact()` (the * `data` field of an SSTable metadata record): serialized bytes stored as a * plain number array so they survive JSON round-trips. */ type PersistedSSTableData = { type: string data: number[] } interface Manifest { /** * Map of SSTable ID to level */ sstables: Map /** * Last compaction time */ lastCompaction: number /** * Total number of relationships */ totalRelationships: number } /** * LSMTree - Main LSM-tree implementation * * Provides efficient graph storage with: * - Fast writes via MemTable * - Efficient reads via bloom filters and binary search * - Automatic compaction to maintain performance * - Integration with any StorageAdapter */ export class LSMTree { /** * Storage adapter for persistence */ private storage: StorageAdapter /** * Configuration */ private config: Required /** * In-memory write buffer */ private memTable: MemTable /** * Loaded SSTables grouped by level * Level 0: Fresh from MemTable (smallest, most recent) * Level 1-6: Progressively larger, older, merged files */ private sstablesByLevel: Map /** * Manifest tracking all SSTables */ private manifest: Manifest /** * Compaction timer */ private compactionTimer?: NodeJS.Timeout /** * Whether compaction is currently running */ private isCompacting: boolean /** * Whether LSMTree has been initialized */ private initialized: boolean constructor(storage: StorageAdapter, config: LSMTreeConfig = {}) { this.storage = storage this.config = { memTableThreshold: config.memTableThreshold ?? 100000, maxSSTablesPerLevel: config.maxSSTablesPerLevel ?? 10, storagePrefix: config.storagePrefix ?? 'graph-lsm', enableCompaction: config.enableCompaction ?? true, compactionInterval: config.compactionInterval ?? 60000 } this.memTable = new MemTable() this.sstablesByLevel = new Map() this.manifest = { sstables: new Map(), lastCompaction: Date.now(), totalRelationships: 0 } this.isCompacting = false this.initialized = false } /** * Initialize the LSMTree * Loads manifest and prepares for operations */ async init(): Promise { if (this.initialized) { return } try { // Load manifest from storage await this.loadManifest() // Start compaction timer if enabled if (this.config.enableCompaction) { this.startCompactionTimer() } this.initialized = true prodLog.info('LSMTree: Initialized successfully') } catch (error) { prodLog.error('LSMTree: Initialization failed', error) throw error } } /** * Add a relationship to the LSM-tree * @param sourceId Source node ID * @param targetId Target node ID */ async add(sourceId: string, targetId: string): Promise { const startTime = performance.now() // Add to MemTable this.memTable.add(sourceId, targetId) this.manifest.totalRelationships++ // Check if MemTable needs flushing if (this.memTable.size() >= this.config.memTableThreshold) { await this.flushMemTable() } const elapsed = performance.now() - startTime // Performance assertion - writes should be fast if (elapsed > 10.0) { prodLog.warn(`LSMTree: Slow write operation: ${elapsed.toFixed(2)}ms`) } } /** * Get targets for a sourceId * Checks MemTable first, then SSTables with bloom filter optimization * * @param sourceId Source node ID * @returns Array of target IDs, or null if not found */ async get(sourceId: string): Promise { const startTime = performance.now() // 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) { for (const target of memResult) { allTargets.add(target) } } // Check SSTables from newest to oldest const maxLevel = Math.max(...Array.from(this.sstablesByLevel.keys()), 0) for (let level = 0; level <= maxLevel; level++) { const sstables = this.sstablesByLevel.get(level) || [] for (const sstable of sstables) { // Quick check: Is sourceId in range? if (!sstable.isInRange(sourceId)) { continue } // Quick check: Does bloom filter say it might be here? if (!sstable.mightContain(sourceId)) { continue } // Binary search in SSTable const targets = sstable.get(sourceId) if (targets) { for (const target of targets) { allTargets.add(target) } } } } const elapsed = performance.now() - startTime // Performance assertion - reads should be fast if (elapsed > 5.0) { prodLog.warn(`LSMTree: Slow read operation for ${sourceId}: ${elapsed.toFixed(2)}ms`) } return allTargets.size > 0 ? Array.from(allTargets) : null } /** * Flush MemTable to a new L0 SSTable */ private async flushMemTable(): Promise { if (this.memTable.isEmpty()) { return } const startTime = Date.now() prodLog.info(`LSMTree: Flushing MemTable (${this.memTable.size()} relationships)`) try { // Create SSTable from MemTable const sstable = SSTable.fromMap(this.memTable.getAll(), 0) // Serialize and save to storage const data = sstable.serialize() const storageKey = `${this.config.storagePrefix}-${sstable.metadata.id}` await this.storage.saveMetadata(storageKey, { noun: 'thing', // Required for NounMetadata data: { type: 'lsm-sstable', data: Array.from(data) // Convert Uint8Array to number[] for JSON storage } }) // Add to L0 SSTables if (!this.sstablesByLevel.has(0)) { this.sstablesByLevel.set(0, []) } this.sstablesByLevel.get(0)!.push(sstable) // Update manifest this.manifest.sstables.set(sstable.metadata.id, 0) await this.saveManifest() // Clear MemTable this.memTable.clear() const elapsed = Date.now() - startTime prodLog.info(`LSMTree: MemTable flushed in ${elapsed}ms`) // Check if L0 needs compaction const l0Count = this.sstablesByLevel.get(0)?.length || 0 if (l0Count >= this.config.maxSSTablesPerLevel) { // Trigger compaction asynchronously setImmediate(() => this.compact(0)) } } catch (error) { prodLog.error('LSMTree: Failed to flush MemTable', error) throw error } } /** * Compact a level by merging SSTables * @param level Level to compact */ private async compact(level: number): Promise { if (this.isCompacting) { prodLog.debug('LSMTree: Compaction already in progress, skipping') return } this.isCompacting = true const startTime = Date.now() try { const sstables = this.sstablesByLevel.get(level) || [] if (sstables.length < this.config.maxSSTablesPerLevel) { this.isCompacting = false return } prodLog.info(`LSMTree: Compacting L${level} (${sstables.length} SSTables)`) // Merge all SSTables at this level const merged = SSTable.merge(sstables, level + 1) // Serialize and save merged SSTable const data = merged.serialize() const storageKey = `${this.config.storagePrefix}-${merged.metadata.id}` await this.storage.saveMetadata(storageKey, { noun: 'thing', // Required for NounMetadata data: { type: 'lsm-sstable', data: Array.from(data) } }) // Reclaim the compacted-away SSTables: drop them from the manifest AND // delete their persisted payloads, so the system channel does not grow // unbounded with graph write volume. (deleteMetadata is idempotent, so a // never-persisted memtable-only SSTable is a harmless no-op.) for (const sstable of sstables) { const oldKey = `${this.config.storagePrefix}-${sstable.metadata.id}` this.manifest.sstables.delete(sstable.metadata.id) try { await this.storage.deleteMetadata(oldKey) } catch (error) { // A reclaim failure must not abort compaction (the merged SSTable is // already durable and the manifest no longer references the old one); // surface it so a persistent leak is visible rather than silent. prodLog.warn(`LSMTree: failed to delete compacted SSTable ${sstable.metadata.id}`, error) } } // Update in-memory structures this.sstablesByLevel.set(level, []) if (!this.sstablesByLevel.has(level + 1)) { this.sstablesByLevel.set(level + 1, []) } this.sstablesByLevel.get(level + 1)!.push(merged) // Update manifest this.manifest.sstables.set(merged.metadata.id, level + 1) this.manifest.lastCompaction = Date.now() await this.saveManifest() const elapsed = Date.now() - startTime prodLog.info(`LSMTree: Compaction complete in ${elapsed}ms`) // Check if next level needs compaction const nextLevelCount = this.sstablesByLevel.get(level + 1)?.length || 0 if (nextLevelCount >= this.config.maxSSTablesPerLevel && level < 6) { // Trigger next level compaction setImmediate(() => this.compact(level + 1)) } } catch (error) { prodLog.error(`LSMTree: Compaction failed for L${level}`, error) } finally { this.isCompacting = false } } /** * Start background compaction timer */ private startCompactionTimer(): void { this.compactionTimer = setInterval(() => { // Check each level for compaction needs for (let level = 0; level < 6; level++) { const count = this.sstablesByLevel.get(level)?.length || 0 if (count >= this.config.maxSSTablesPerLevel) { this.compact(level) break // Only compact one level per interval } } }, this.config.compactionInterval) // Background compaction must never keep the host process alive — // close() compacts/flushes deterministically; this interval is best-effort. if (this.compactionTimer && typeof this.compactionTimer.unref === 'function') { this.compactionTimer.unref() } } /** * Stop background compaction timer */ private stopCompactionTimer(): void { if (this.compactionTimer) { clearInterval(this.compactionTimer) this.compactionTimer = undefined } } /** * Load manifest from storage */ private async loadManifest(): Promise { try { const metadata = await this.storage.getMetadata(`${this.config.storagePrefix}-manifest`) if (metadata && metadata.data) { // Storage boundary: `data` is the JSON manifest payload written by // saveManifest(); re-typed from the metadata channel's `unknown`. const data = metadata.data as PersistedManifestData this.manifest.sstables = new Map(Object.entries(data.sstables || {})) this.manifest.lastCompaction = data.lastCompaction || Date.now() // Load SSTables from storage BEFORE publishing the persisted count. // If the SSTable load throws, `size()` must keep reporting 0 — a tree // that claims its persisted relationships while holding none serves // silent-empty traversals as truth (the cold-load swallow class), and // downstream self-heal keys off the honest 0. await this.loadSSTables() this.manifest.totalRelationships = data.totalRelationships || 0 } } catch (error) { // Reset anything partially loaded — an honest empty tree triggers the // rebuild/self-heal paths; a half-loaded one masks them. (An absent // manifest on a fresh store also lands here: empty is correct.) this.manifest.sstables = new Map() this.manifest.totalRelationships = 0 this.sstablesByLevel.clear() prodLog.debug( `LSMTree(${this.config.storagePrefix}): no loadable manifest/SSTables — starting empty ` + `(${error instanceof Error ? error.message : String(error)})` ) } } /** * Load SSTables from storage based on manifest */ private async loadSSTables(): Promise { const loadPromises: Promise[] = [] this.manifest.sstables.forEach((level, sstableId) => { const loadPromise = (async () => { try { const storageKey = `${this.config.storagePrefix}-${sstableId}` const metadata = await this.storage.getMetadata(storageKey) if (metadata && metadata.data) { // Storage boundary: `data` is the JSON SSTable payload written by // flushMemTable()/compact(); re-typed from `unknown`. const data = metadata.data as PersistedSSTableData if (data.type === 'lsm-sstable') { // Convert number[] back to Uint8Array const uint8Data = new Uint8Array(data.data) const sstable = SSTable.deserialize(uint8Data) if (!this.sstablesByLevel.has(level)) { this.sstablesByLevel.set(level, []) } this.sstablesByLevel.get(level)!.push(sstable) } } } catch (error) { prodLog.warn(`LSMTree: Failed to load SSTable ${sstableId}`, error) } })() loadPromises.push(loadPromise) }) await Promise.all(loadPromises) prodLog.info(`LSMTree: Loaded ${this.manifest.sstables.size} SSTables`) } /** * Save manifest to storage */ private async saveManifest(): Promise { try { await this.storage.saveMetadata( `${this.config.storagePrefix}-manifest`, { noun: 'thing', // Required for NounMetadata data: { sstables: Object.fromEntries(this.manifest.sstables), lastCompaction: this.manifest.lastCompaction, totalRelationships: this.manifest.totalRelationships } } ) } catch (error) { prodLog.error('LSMTree: Failed to save manifest', error) throw error } } /** * Get statistics about the LSM-tree */ getStats(): { memTableSize: number memTableMemory: number sstableCount: number sstablesByLevel: Record totalRelationships: number lastCompaction: number } { const sstablesByLevel: Record = {} this.sstablesByLevel.forEach((sstables, level) => { sstablesByLevel[level] = sstables.length }) return { memTableSize: this.memTable.size(), memTableMemory: this.memTable.estimateMemoryUsage(), sstableCount: this.manifest.sstables.size, sstablesByLevel, totalRelationships: this.manifest.totalRelationships, lastCompaction: this.manifest.lastCompaction } } /** * 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 await this.flush() prodLog.info('LSMTree: Closed successfully') } /** * Get total relationship count */ size(): number { return this.manifest.totalRelationships } /** * Check if LSM-tree is healthy */ isHealthy(): boolean { return this.initialized && !this.isCompacting } }