- Version-coupling cold-init: getBrainyVersion() returned a stale build-time
default ('3.14.0') on the FIRST synchronous call — the one loadPlugins() makes
to build context.version — because the package.json read was async. A native
provider declaring a realistic `>=8.0.0` range was therefore rejected on cold
init. version.ts now reads package.json synchronously (8.0 targets Node-like
runtimes only); added a cold-init coupling regression with a `^8.0.0` plugin.
- No-silent-failures on the highest-fan-in read path: getNouns()/getVerbs()
converted a storage read failure into a success-shaped empty page, which the
cold-start rebuild then read as "store empty" and skipped the rebuild — booting
a permanently-empty index with no signal (the same silent-failure class as the
phantom bug). Both now re-throw a named BrainyError; the rebuild path re-throws
read failures (fail loud) and records a queryable degraded state, surfaced via
checkHealth(), for non-fatal rebuild hiccups.
- LSM SSTable leak: graph compaction wrote a merged SSTable but only dropped the
old ones from the manifest, orphaning their payloads forever ("In production
we'd add a cleanup mechanism"). Added StorageAdapter.deleteMetadata() and now
reclaim each compacted-away SSTable.
- Documented the two headline methods add() and find() (the only undocumented
public methods on the class).
Full gate green: build, unit 1512, integration 607.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
674 lines
18 KiB
TypeScript
674 lines
18 KiB
TypeScript
/**
|
|
* @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<string, Set<string>>
|
|
|
|
/**
|
|
* 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<string, Set<string>> {
|
|
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<string, number>
|
|
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<string, number>
|
|
|
|
/**
|
|
* 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<LSMTreeConfig>
|
|
|
|
/**
|
|
* 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<number, SSTable[]>
|
|
|
|
/**
|
|
* 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<void> {
|
|
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<void> {
|
|
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<string[] | null> {
|
|
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<string>()
|
|
|
|
// 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<void> {
|
|
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<void> {
|
|
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)
|
|
}
|
|
|
|
/**
|
|
* Stop background compaction timer
|
|
*/
|
|
private stopCompactionTimer(): void {
|
|
if (this.compactionTimer) {
|
|
clearInterval(this.compactionTimer)
|
|
this.compactionTimer = undefined
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Load manifest from storage
|
|
*/
|
|
private async loadManifest(): Promise<void> {
|
|
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()
|
|
this.manifest.totalRelationships = data.totalRelationships || 0
|
|
|
|
// Load SSTables from storage
|
|
await this.loadSSTables()
|
|
}
|
|
} catch (error) {
|
|
prodLog.debug('LSMTree: No existing manifest found, starting fresh')
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Load SSTables from storage based on manifest
|
|
*/
|
|
private async loadSSTables(): Promise<void> {
|
|
const loadPromises: Promise<void>[] = []
|
|
|
|
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<void> {
|
|
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<number, number>
|
|
totalRelationships: number
|
|
lastCompaction: number
|
|
} {
|
|
const sstablesByLevel: Record<number, number> = {}
|
|
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<void> {
|
|
if (!this.memTable.isEmpty()) {
|
|
await this.flushMemTable()
|
|
}
|
|
}
|
|
|
|
async close(): Promise<void> {
|
|
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
|
|
}
|
|
}
|