David Snelling
ab2493af02
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.
860 lines
28 KiB
TypeScript
860 lines
28 KiB
TypeScript
/**
|
||
* GraphAdjacencyIndex - Billion-Scale Graph Traversal Engine
|
||
*
|
||
* NOW SCALES TO BILLIONS: LSM-tree storage reduces memory from 500GB to 1.3GB
|
||
* for 1 billion relationships while maintaining sub-5ms neighbor lookups.
|
||
*
|
||
* NO FALLBACKS - NO MOCKS - REAL PRODUCTION CODE
|
||
* Handles billions of relationships with sustainable memory usage
|
||
*/
|
||
|
||
import { GraphVerb, StorageAdapter } from '../coreTypes.js'
|
||
import { UnifiedCache, getGlobalCache } from '../utils/unifiedCache.js'
|
||
import { prodLog } from '../utils/logger.js'
|
||
import { LSMTree } from './lsm/LSMTree.js'
|
||
|
||
export interface GraphIndexConfig {
|
||
maxIndexSize?: number // Default: 100000
|
||
rebuildThreshold?: number // Default: 0.1
|
||
autoOptimize?: boolean // Default: true
|
||
flushInterval?: number // Default: 30000ms
|
||
}
|
||
|
||
export interface GraphIndexStats {
|
||
totalRelationships: number
|
||
sourceNodes: number
|
||
targetNodes: number
|
||
memoryUsage: number // in bytes
|
||
lastRebuild: number
|
||
rebuildTime: number // in ms
|
||
}
|
||
|
||
/**
|
||
* GraphAdjacencyIndex - Billion-scale adjacency list with LSM-tree storage
|
||
*
|
||
* Core innovation: LSM-tree for disk-based storage with bloom filter optimization
|
||
* Memory efficient: 385x less memory (1.3GB vs 500GB for 1B relationships)
|
||
* Performance: Sub-5ms neighbor lookups with bloom filter optimization
|
||
*/
|
||
export class GraphAdjacencyIndex {
|
||
// LSM-tree storage for outgoing and incoming edges
|
||
private lsmTreeSource: LSMTree // sourceId -> targetIds (outgoing edges)
|
||
private lsmTreeTarget: LSMTree // targetId -> sourceIds (incoming edges)
|
||
|
||
// LSM-tree storage for verb ID lookups (billion-scale optimization)
|
||
private lsmTreeVerbsBySource: LSMTree // sourceId -> verbIds
|
||
private lsmTreeVerbsByTarget: LSMTree // targetId -> verbIds
|
||
|
||
// ID-only tracking for billion-scale memory optimization
|
||
// Previous: Map<string, GraphVerb> stored full objects (128GB @ 1B verbs)
|
||
// Now: Set<string> stores only IDs (~100KB @ 1B verbs) = 1,280,000x reduction
|
||
private verbIdSet = new Set<string>()
|
||
|
||
// Infrastructure integration
|
||
private storage: StorageAdapter
|
||
private unifiedCache: UnifiedCache
|
||
private config: Required<GraphIndexConfig>
|
||
|
||
// Performance optimization
|
||
private isRebuilding = false
|
||
private flushTimer?: NodeJS.Timeout
|
||
private rebuildStartTime = 0
|
||
private totalRelationshipsIndexed = 0
|
||
|
||
// Production-scale relationship counting by type
|
||
private relationshipCountsByType = new Map<string, number>()
|
||
|
||
// Initialization flag
|
||
private initialized = false
|
||
|
||
/**
|
||
* Check if index is initialized and ready for use
|
||
*/
|
||
get isInitialized(): boolean {
|
||
return this.initialized
|
||
}
|
||
|
||
constructor(storage: StorageAdapter, config: GraphIndexConfig = {}) {
|
||
this.storage = storage
|
||
this.config = {
|
||
maxIndexSize: config.maxIndexSize ?? 100000,
|
||
rebuildThreshold: config.rebuildThreshold ?? 0.1,
|
||
autoOptimize: config.autoOptimize ?? true,
|
||
flushInterval: config.flushInterval ?? 30000
|
||
}
|
||
|
||
// Create LSM-trees for source and target indexes
|
||
this.lsmTreeSource = new LSMTree(storage, {
|
||
memTableThreshold: 100000,
|
||
storagePrefix: 'graph-lsm-source',
|
||
enableCompaction: true
|
||
})
|
||
|
||
this.lsmTreeTarget = new LSMTree(storage, {
|
||
memTableThreshold: 100000,
|
||
storagePrefix: 'graph-lsm-target',
|
||
enableCompaction: true
|
||
})
|
||
|
||
// Create LSM-trees for verb ID lookups (billion-scale optimization)
|
||
this.lsmTreeVerbsBySource = new LSMTree(storage, {
|
||
memTableThreshold: 100000,
|
||
storagePrefix: 'graph-lsm-verbs-source',
|
||
enableCompaction: true
|
||
})
|
||
|
||
this.lsmTreeVerbsByTarget = new LSMTree(storage, {
|
||
memTableThreshold: 100000,
|
||
storagePrefix: 'graph-lsm-verbs-target',
|
||
enableCompaction: true
|
||
})
|
||
|
||
// Use SAME UnifiedCache as MetadataIndexManager for coordinated memory management
|
||
this.unifiedCache = getGlobalCache()
|
||
|
||
prodLog.info('GraphAdjacencyIndex initialized with LSM-tree storage (4 LSM-trees total)')
|
||
}
|
||
|
||
/**
|
||
* Initialize the graph index (lazy initialization)
|
||
* Added defensive auto-rebuild check for verbIdSet consistency
|
||
*/
|
||
private async ensureInitialized(): Promise<void> {
|
||
if (this.initialized) {
|
||
return
|
||
}
|
||
|
||
await this.lsmTreeSource.init()
|
||
await this.lsmTreeTarget.init()
|
||
await this.lsmTreeVerbsBySource.init()
|
||
await this.lsmTreeVerbsByTarget.init()
|
||
|
||
// Defensive check - if LSM-trees have data but verbIdSet is empty,
|
||
// the index was created without proper rebuild (shouldn't happen with singleton
|
||
// pattern but protects against edge cases and future refactoring)
|
||
const lsmTreeSize = this.lsmTreeVerbsBySource.size()
|
||
if (lsmTreeSize > 0 && this.verbIdSet.size === 0) {
|
||
prodLog.warn(
|
||
`GraphAdjacencyIndex: LSM-trees have ${lsmTreeSize} relationships but verbIdSet is empty. ` +
|
||
`Triggering auto-rebuild to restore consistency.`
|
||
)
|
||
// Note: We don't await rebuild() here to avoid infinite loop
|
||
// (rebuild calls ensureInitialized). Instead, we'll populate verbIdSet
|
||
// by loading all verb IDs from storage.
|
||
await this.populateVerbIdSetFromStorage()
|
||
}
|
||
|
||
// Start auto-flush timer after initialization
|
||
this.startAutoFlush()
|
||
|
||
this.initialized = true
|
||
}
|
||
|
||
/**
|
||
* Populate verbIdSet from storage without full rebuild
|
||
* Lighter weight than full rebuild - only loads verb IDs, not all verb data
|
||
* @private
|
||
*/
|
||
private async populateVerbIdSetFromStorage(): Promise<void> {
|
||
prodLog.info('GraphAdjacencyIndex: Populating verbIdSet from storage...')
|
||
const startTime = Date.now()
|
||
|
||
// Use pagination to load all verb IDs
|
||
let hasMore = true
|
||
let cursor: string | undefined = undefined
|
||
let count = 0
|
||
|
||
while (hasMore) {
|
||
const result = await this.storage.getVerbs({
|
||
pagination: { limit: 10000, cursor }
|
||
})
|
||
|
||
for (const verb of result.items) {
|
||
this.verbIdSet.add(verb.id)
|
||
// Also update counts
|
||
const verbType = verb.verb || 'unknown'
|
||
this.relationshipCountsByType.set(
|
||
verbType,
|
||
(this.relationshipCountsByType.get(verbType) || 0) + 1
|
||
)
|
||
count++
|
||
}
|
||
|
||
hasMore = result.hasMore
|
||
cursor = result.nextCursor
|
||
}
|
||
|
||
const elapsed = Date.now() - startTime
|
||
prodLog.info(`GraphAdjacencyIndex: Populated verbIdSet with ${count} verb IDs in ${elapsed}ms`)
|
||
}
|
||
|
||
/**
|
||
* Core API - Neighbor lookup with LSM-tree storage
|
||
*
|
||
* O(log n) with bloom filter optimization (90% of queries skip disk I/O)
|
||
* Added pagination support for high-degree nodes
|
||
*
|
||
* @param id Entity ID to get neighbors for
|
||
* @param optionsOrDirection Optional: direction string OR options object
|
||
* @returns Array of neighbor IDs (paginated if limit/offset specified)
|
||
*
|
||
* @example
|
||
* // Get all neighbors (backward compatible)
|
||
* const all = await graphIndex.getNeighbors(id)
|
||
*
|
||
* @example
|
||
* // Get outgoing neighbors (backward compatible)
|
||
* const out = await graphIndex.getNeighbors(id, 'out')
|
||
*
|
||
* @example
|
||
* // Get first 50 outgoing neighbors (new API)
|
||
* const page1 = await graphIndex.getNeighbors(id, { direction: 'out', limit: 50 })
|
||
*
|
||
* @example
|
||
* // Paginate through neighbors
|
||
* const page1 = await graphIndex.getNeighbors(id, { limit: 100, offset: 0 })
|
||
* const page2 = await graphIndex.getNeighbors(id, { limit: 100, offset: 100 })
|
||
*/
|
||
async getNeighbors(
|
||
id: string,
|
||
optionsOrDirection?: {
|
||
direction?: 'in' | 'out' | 'both'
|
||
limit?: number
|
||
offset?: number
|
||
} | 'in' | 'out' | 'both'
|
||
): Promise<string[]> {
|
||
await this.ensureInitialized()
|
||
|
||
// Normalize old API (direction string) to new API (options object)
|
||
const options = typeof optionsOrDirection === 'string'
|
||
? { direction: optionsOrDirection }
|
||
: (optionsOrDirection || {})
|
||
|
||
const startTime = performance.now()
|
||
const direction = options.direction || 'both'
|
||
const neighbors = new Set<string>()
|
||
|
||
// Query LSM-trees with bloom filter optimization
|
||
if (direction !== 'in') {
|
||
const outgoing = await this.lsmTreeSource.get(id)
|
||
if (outgoing) {
|
||
outgoing.forEach(neighborId => neighbors.add(neighborId))
|
||
}
|
||
}
|
||
|
||
if (direction !== 'out') {
|
||
const incoming = await this.lsmTreeTarget.get(id)
|
||
if (incoming) {
|
||
incoming.forEach(neighborId => neighbors.add(neighborId))
|
||
}
|
||
}
|
||
|
||
// Convert to array for pagination
|
||
let result = Array.from(neighbors)
|
||
|
||
// Apply pagination if requested
|
||
if (options?.limit !== undefined || options?.offset !== undefined) {
|
||
const offset = options.offset || 0
|
||
const limit = options.limit !== undefined ? options.limit : result.length
|
||
result = result.slice(offset, offset + limit)
|
||
}
|
||
|
||
const elapsed = performance.now() - startTime
|
||
|
||
// Performance assertion - should be sub-5ms with LSM-tree
|
||
if (elapsed > 5.0) {
|
||
prodLog.warn(`GraphAdjacencyIndex: Slow neighbor lookup for ${id}: ${elapsed.toFixed(2)}ms`)
|
||
}
|
||
|
||
return result
|
||
}
|
||
|
||
/**
|
||
* Get verb IDs by source - Billion-scale optimization for getVerbsBySource
|
||
*
|
||
* O(log n) LSM-tree lookup with bloom filter optimization
|
||
* Filters out deleted verb IDs (tombstone deletion workaround)
|
||
* Added pagination support for entities with many relationships
|
||
*
|
||
* @param sourceId Source entity ID
|
||
* @param options Optional configuration
|
||
* @param options.limit Maximum number of verb IDs to return (default: all)
|
||
* @param options.offset Number of verb IDs to skip (default: 0)
|
||
* @returns Array of verb IDs originating from this source (excluding deleted, paginated if requested)
|
||
*
|
||
* @example
|
||
* // Get all verb IDs (backward compatible)
|
||
* const all = await graphIndex.getVerbIdsBySource(sourceId)
|
||
*
|
||
* @example
|
||
* // Get first 50 verb IDs
|
||
* const page1 = await graphIndex.getVerbIdsBySource(sourceId, { limit: 50 })
|
||
*
|
||
* @example
|
||
* // Paginate through verb IDs
|
||
* const page1 = await graphIndex.getVerbIdsBySource(sourceId, { limit: 100, offset: 0 })
|
||
* const page2 = await graphIndex.getVerbIdsBySource(sourceId, { limit: 100, offset: 100 })
|
||
*/
|
||
async getVerbIdsBySource(
|
||
sourceId: string,
|
||
options?: {
|
||
limit?: number
|
||
offset?: number
|
||
}
|
||
): Promise<string[]> {
|
||
await this.ensureInitialized()
|
||
|
||
const startTime = performance.now()
|
||
const verbIds = await this.lsmTreeVerbsBySource.get(sourceId)
|
||
const elapsed = performance.now() - startTime
|
||
|
||
// Performance assertion - should be sub-5ms with LSM-tree
|
||
if (elapsed > 5.0) {
|
||
prodLog.warn(`GraphAdjacencyIndex: Slow getVerbIdsBySource for ${sourceId}: ${elapsed.toFixed(2)}ms`)
|
||
}
|
||
|
||
// Filter out deleted verb IDs (tombstone deletion workaround)
|
||
// LSM-tree retains all IDs, but verbIdSet tracks deletions
|
||
const allIds = verbIds || []
|
||
let result = allIds.filter(id => this.verbIdSet.has(id))
|
||
|
||
// Apply pagination if requested
|
||
if (options?.limit !== undefined || options?.offset !== undefined) {
|
||
const offset = options.offset || 0
|
||
const limit = options.limit !== undefined ? options.limit : result.length
|
||
result = result.slice(offset, offset + limit)
|
||
}
|
||
|
||
return result
|
||
}
|
||
|
||
/**
|
||
* Get verb IDs by target - Billion-scale optimization for getVerbsByTarget
|
||
*
|
||
* O(log n) LSM-tree lookup with bloom filter optimization
|
||
* Filters out deleted verb IDs (tombstone deletion workaround)
|
||
* Added pagination support for popular target entities
|
||
*
|
||
* @param targetId Target entity ID
|
||
* @param options Optional configuration
|
||
* @param options.limit Maximum number of verb IDs to return (default: all)
|
||
* @param options.offset Number of verb IDs to skip (default: 0)
|
||
* @returns Array of verb IDs pointing to this target (excluding deleted, paginated if requested)
|
||
*
|
||
* @example
|
||
* // Get all verb IDs (backward compatible)
|
||
* const all = await graphIndex.getVerbIdsByTarget(targetId)
|
||
*
|
||
* @example
|
||
* // Get first 50 verb IDs
|
||
* const page1 = await graphIndex.getVerbIdsByTarget(targetId, { limit: 50 })
|
||
*
|
||
* @example
|
||
* // Paginate through verb IDs
|
||
* const page1 = await graphIndex.getVerbIdsByTarget(targetId, { limit: 100, offset: 0 })
|
||
* const page2 = await graphIndex.getVerbIdsByTarget(targetId, { limit: 100, offset: 100 })
|
||
*/
|
||
async getVerbIdsByTarget(
|
||
targetId: string,
|
||
options?: {
|
||
limit?: number
|
||
offset?: number
|
||
}
|
||
): Promise<string[]> {
|
||
await this.ensureInitialized()
|
||
|
||
const startTime = performance.now()
|
||
const verbIds = await this.lsmTreeVerbsByTarget.get(targetId)
|
||
const elapsed = performance.now() - startTime
|
||
|
||
// Performance assertion - should be sub-5ms with LSM-tree
|
||
if (elapsed > 5.0) {
|
||
prodLog.warn(`GraphAdjacencyIndex: Slow getVerbIdsByTarget for ${targetId}: ${elapsed.toFixed(2)}ms`)
|
||
}
|
||
|
||
// Filter out deleted verb IDs (tombstone deletion workaround)
|
||
// LSM-tree retains all IDs, but verbIdSet tracks deletions
|
||
const allIds = verbIds || []
|
||
let result = allIds.filter(id => this.verbIdSet.has(id))
|
||
|
||
// Apply pagination if requested
|
||
if (options?.limit !== undefined || options?.offset !== undefined) {
|
||
const offset = options.offset || 0
|
||
const limit = options.limit !== undefined ? options.limit : result.length
|
||
result = result.slice(offset, offset + limit)
|
||
}
|
||
|
||
return result
|
||
}
|
||
|
||
/**
|
||
* Get verb from cache or storage - Billion-scale memory optimization
|
||
* Uses UnifiedCache with LRU eviction instead of storing all verbs in memory
|
||
*
|
||
* @param verbId Verb ID to retrieve
|
||
* @returns GraphVerb or null if not found
|
||
*/
|
||
async getVerbCached(verbId: string): Promise<GraphVerb | null> {
|
||
const cacheKey = `graph:verb:${verbId}`
|
||
|
||
// Try to get from cache, load if not present
|
||
const verb = await this.unifiedCache.get(cacheKey, async () => {
|
||
// Load from storage (fallback if not in cache)
|
||
const loadedVerb = await this.storage.getVerb(verbId)
|
||
|
||
// Cache the loaded verb with metadata
|
||
if (loadedVerb) {
|
||
this.unifiedCache.set(cacheKey, loadedVerb, 'other', 128, 50) // 128 bytes estimated size, 50ms rebuild cost
|
||
}
|
||
|
||
return loadedVerb
|
||
})
|
||
|
||
return verb
|
||
}
|
||
|
||
/**
|
||
* Batch get multiple verbs with caching
|
||
*
|
||
* **Performance**: Eliminates N+1 pattern for verb loading
|
||
* - Current: N × getVerbCached() = N × 50ms on GCS = 250ms for 5 verbs
|
||
* - Batched: 1 × getVerbsBatchCached() = 1 × 50ms on GCS = 50ms (**5x faster**)
|
||
*
|
||
* **Use cases:**
|
||
* - relate() duplicate checking (check multiple existing relationships)
|
||
* - Loading relationship chains
|
||
* - Pre-loading verbs for analysis
|
||
*
|
||
* **Cache behavior:**
|
||
* - Checks UnifiedCache first (fast path)
|
||
* - Batch-loads uncached verbs from storage
|
||
* - Caches loaded verbs for future access
|
||
*
|
||
* @param verbIds Array of verb IDs to fetch
|
||
* @returns Map of verbId → GraphVerb (only successful reads included)
|
||
*
|
||
*/
|
||
async getVerbsBatchCached(verbIds: string[]): Promise<Map<string, GraphVerb>> {
|
||
const results = new Map<string, GraphVerb>()
|
||
const uncached: string[] = []
|
||
|
||
// Phase 1: Check cache for each verb
|
||
for (const verbId of verbIds) {
|
||
const cacheKey = `graph:verb:${verbId}`
|
||
const cached = this.unifiedCache.getSync(cacheKey)
|
||
|
||
if (cached) {
|
||
results.set(verbId, cached)
|
||
} else {
|
||
uncached.push(verbId)
|
||
}
|
||
}
|
||
|
||
// Phase 2: Batch-load uncached verbs from storage
|
||
if (uncached.length > 0 && this.storage.getVerbsBatch) {
|
||
const loadedVerbs = await this.storage.getVerbsBatch(uncached)
|
||
|
||
for (const [verbId, verb] of loadedVerbs.entries()) {
|
||
const cacheKey = `graph:verb:${verbId}`
|
||
// Cache the loaded verb with metadata
|
||
// Note: HNSWVerbWithMetadata is compatible with GraphVerb (both interfaces)
|
||
this.unifiedCache.set(cacheKey, verb as any, 'other', 128, 50) // 128 bytes estimated size, 50ms rebuild cost
|
||
results.set(verbId, verb as any)
|
||
}
|
||
}
|
||
|
||
return results
|
||
}
|
||
|
||
/**
|
||
* Get total relationship count - O(1) operation
|
||
*/
|
||
size(): number {
|
||
// Use LSM-tree size for accurate count
|
||
return this.lsmTreeSource.size()
|
||
}
|
||
|
||
/**
|
||
* Get relationship count by type - O(1) operation using existing tracking
|
||
*/
|
||
getRelationshipCountByType(type: string): number {
|
||
return this.relationshipCountsByType.get(type) || 0
|
||
}
|
||
|
||
/**
|
||
* Get total relationship count - O(1) operation
|
||
*/
|
||
getTotalRelationshipCount(): number {
|
||
return this.verbIdSet.size
|
||
}
|
||
|
||
/**
|
||
* Get all relationship types and their counts - O(1) operation
|
||
*/
|
||
getAllRelationshipCounts(): Map<string, number> {
|
||
return new Map(this.relationshipCountsByType)
|
||
}
|
||
|
||
/**
|
||
* Get relationship statistics with enhanced counting information
|
||
*/
|
||
getRelationshipStats(): {
|
||
totalRelationships: number
|
||
relationshipsByType: Record<string, number>
|
||
uniqueSourceNodes: number
|
||
uniqueTargetNodes: number
|
||
totalNodes: number
|
||
} {
|
||
const totalRelationships = this.lsmTreeSource.size()
|
||
const relationshipsByType = Object.fromEntries(this.relationshipCountsByType)
|
||
|
||
// Get stats from LSM-trees
|
||
const sourceStats = this.lsmTreeSource.getStats()
|
||
const targetStats = this.lsmTreeTarget.getStats()
|
||
|
||
// Note: Exact unique node counts would require full LSM-tree scan
|
||
// Using verbIdSet (ID-only tracking) for memory efficiency
|
||
const uniqueSourceNodes = this.verbIdSet.size
|
||
const uniqueTargetNodes = this.verbIdSet.size
|
||
const totalNodes = this.verbIdSet.size
|
||
|
||
return {
|
||
totalRelationships,
|
||
relationshipsByType,
|
||
uniqueSourceNodes,
|
||
uniqueTargetNodes,
|
||
totalNodes
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Add relationship to index using LSM-tree storage
|
||
*/
|
||
async addVerb(verb: GraphVerb): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
const startTime = performance.now()
|
||
|
||
// Track verb ID (memory-efficient: IDs only, full objects loaded on-demand via UnifiedCache)
|
||
this.verbIdSet.add(verb.id)
|
||
|
||
// Add to LSM-trees (outgoing and incoming edges)
|
||
await this.lsmTreeSource.add(verb.sourceId, verb.targetId)
|
||
await this.lsmTreeTarget.add(verb.targetId, verb.sourceId)
|
||
|
||
// Add to verbId tracking LSM-trees (billion-scale optimization for getVerbsBySource/Target)
|
||
await this.lsmTreeVerbsBySource.add(verb.sourceId, verb.id)
|
||
await this.lsmTreeVerbsByTarget.add(verb.targetId, verb.id)
|
||
|
||
// Update type-specific counts atomically
|
||
const verbType = verb.type || 'unknown'
|
||
this.relationshipCountsByType.set(
|
||
verbType,
|
||
(this.relationshipCountsByType.get(verbType) || 0) + 1
|
||
)
|
||
|
||
const elapsed = performance.now() - startTime
|
||
this.totalRelationshipsIndexed++
|
||
|
||
// Performance assertion
|
||
if (elapsed > 10.0) {
|
||
prodLog.warn(`GraphAdjacencyIndex: Slow addVerb for ${verb.id}: ${elapsed.toFixed(2)}ms`)
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Remove relationship from index
|
||
* Note: LSM-tree edges persist (tombstone deletion not yet implemented)
|
||
* Only removes from verb cache and updates counts
|
||
*/
|
||
async removeVerb(verbId: string): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
// Load verb from cache/storage to get type info
|
||
const verb = await this.getVerbCached(verbId)
|
||
if (!verb) return
|
||
|
||
const startTime = performance.now()
|
||
|
||
// Remove from verb ID set
|
||
this.verbIdSet.delete(verbId)
|
||
|
||
// Update type-specific counts atomically
|
||
const verbType = verb.type || 'unknown'
|
||
const currentCount = this.relationshipCountsByType.get(verbType) || 0
|
||
if (currentCount > 1) {
|
||
this.relationshipCountsByType.set(verbType, currentCount - 1)
|
||
} else {
|
||
this.relationshipCountsByType.delete(verbType)
|
||
}
|
||
|
||
// Note: LSM-tree edges persist
|
||
// Full tombstone deletion can be implemented via compaction
|
||
// For now, removed verbs won't appear in queries (verbIndex check)
|
||
|
||
const elapsed = performance.now() - startTime
|
||
|
||
// Performance assertion
|
||
if (elapsed > 5.0) {
|
||
prodLog.warn(`GraphAdjacencyIndex: Slow removeVerb for ${verbId}: ${elapsed.toFixed(2)}ms`)
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Rebuild entire index from storage
|
||
* Critical for cold starts and data consistency
|
||
*/
|
||
async rebuild(): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
if (this.isRebuilding) {
|
||
prodLog.warn('GraphAdjacencyIndex: Rebuild already in progress')
|
||
return
|
||
}
|
||
|
||
this.isRebuilding = true
|
||
this.rebuildStartTime = Date.now()
|
||
|
||
try {
|
||
prodLog.info('GraphAdjacencyIndex: Starting rebuild with LSM-tree...')
|
||
|
||
// Clear current index
|
||
this.verbIdSet.clear()
|
||
this.totalRelationshipsIndexed = 0
|
||
// CRITICAL FIX - Clear relationship counts to prevent accumulation
|
||
this.relationshipCountsByType.clear()
|
||
|
||
// Note: LSM-trees will be recreated from storage via their own initialization
|
||
// Verb data will be loaded on-demand via UnifiedCache
|
||
|
||
// Adaptive loading strategy based on storage type
|
||
const storageType = this.storage?.constructor.name || ''
|
||
const isLocalStorage =
|
||
storageType === 'FileSystemStorage' ||
|
||
storageType === 'MemoryStorage' ||
|
||
storageType === 'OPFSStorage'
|
||
|
||
let totalVerbs = 0
|
||
|
||
if (isLocalStorage) {
|
||
// Local storage: Load all verbs at once to avoid repeated getAllShardedFiles() calls
|
||
prodLog.info(
|
||
`GraphAdjacencyIndex: Using optimized strategy - load all verbs at once (${storageType})`
|
||
)
|
||
|
||
const result = await this.storage.getVerbs({
|
||
pagination: { limit: 10000000 } // Effectively unlimited for local development
|
||
})
|
||
|
||
// Add each verb to index
|
||
for (const verb of result.items) {
|
||
// Convert HNSWVerbWithMetadata to GraphVerb format
|
||
const graphVerb: GraphVerb = {
|
||
id: verb.id,
|
||
sourceId: verb.sourceId,
|
||
targetId: verb.targetId,
|
||
vector: verb.vector,
|
||
source: verb.sourceId,
|
||
target: verb.targetId,
|
||
verb: verb.verb,
|
||
createdAt: { seconds: Math.floor(verb.createdAt / 1000), nanoseconds: (verb.createdAt % 1000) * 1000000 },
|
||
updatedAt: { seconds: Math.floor(verb.updatedAt / 1000), nanoseconds: (verb.updatedAt % 1000) * 1000000 },
|
||
createdBy: verb.createdBy || { augmentation: 'unknown', version: '0.0.0' },
|
||
service: verb.service,
|
||
data: verb.data,
|
||
embedding: verb.vector,
|
||
confidence: verb.confidence,
|
||
weight: verb.weight
|
||
}
|
||
await this.addVerb(graphVerb)
|
||
totalVerbs++
|
||
}
|
||
|
||
prodLog.info(
|
||
`GraphAdjacencyIndex: Loaded ${totalVerbs.toLocaleString()} verbs at once (local storage)`
|
||
)
|
||
} else {
|
||
// Cloud storage: Use pagination with native cloud APIs (efficient)
|
||
prodLog.info(
|
||
`GraphAdjacencyIndex: Using cloud pagination strategy (${storageType})`
|
||
)
|
||
|
||
let hasMore = true
|
||
let cursor: string | undefined = undefined
|
||
const batchSize = 1000
|
||
|
||
while (hasMore) {
|
||
const result = await this.storage.getVerbs({
|
||
pagination: { limit: batchSize, cursor }
|
||
})
|
||
|
||
// Add each verb to index
|
||
for (const verb of result.items) {
|
||
// Convert HNSWVerbWithMetadata to GraphVerb format
|
||
const graphVerb: GraphVerb = {
|
||
id: verb.id,
|
||
sourceId: verb.sourceId,
|
||
targetId: verb.targetId,
|
||
vector: verb.vector,
|
||
source: verb.sourceId,
|
||
target: verb.targetId,
|
||
verb: verb.verb,
|
||
createdAt: { seconds: Math.floor(verb.createdAt / 1000), nanoseconds: (verb.createdAt % 1000) * 1000000 },
|
||
updatedAt: { seconds: Math.floor(verb.updatedAt / 1000), nanoseconds: (verb.updatedAt % 1000) * 1000000 },
|
||
createdBy: verb.createdBy || { augmentation: 'unknown', version: '0.0.0' },
|
||
service: verb.service,
|
||
data: verb.data,
|
||
embedding: verb.vector,
|
||
confidence: verb.confidence,
|
||
weight: verb.weight
|
||
}
|
||
await this.addVerb(graphVerb)
|
||
totalVerbs++
|
||
}
|
||
|
||
hasMore = result.hasMore
|
||
cursor = result.nextCursor
|
||
|
||
// Progress logging
|
||
if (totalVerbs % 10000 === 0) {
|
||
prodLog.info(`GraphAdjacencyIndex: Indexed ${totalVerbs} verbs...`)
|
||
}
|
||
}
|
||
|
||
prodLog.info(
|
||
`GraphAdjacencyIndex: Loaded ${totalVerbs.toLocaleString()} verbs via pagination (cloud storage)`
|
||
)
|
||
}
|
||
|
||
const rebuildTime = Date.now() - this.rebuildStartTime
|
||
const memoryUsage = this.calculateMemoryUsage()
|
||
|
||
prodLog.info(`GraphAdjacencyIndex: Rebuild complete in ${rebuildTime}ms`)
|
||
prodLog.info(` - Total relationships: ${totalVerbs}`)
|
||
prodLog.info(` - Memory usage: ${(memoryUsage / 1024 / 1024).toFixed(1)}MB`)
|
||
prodLog.info(` - LSM-tree stats:`, this.lsmTreeSource.getStats())
|
||
|
||
} finally {
|
||
this.isRebuilding = false
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Calculate current memory usage (LSM-tree mostly on disk)
|
||
*/
|
||
private calculateMemoryUsage(): number {
|
||
let bytes = 0
|
||
|
||
// LSM-tree memory (MemTable + bloom filters + zone maps)
|
||
const sourceStats = this.lsmTreeSource.getStats()
|
||
const targetStats = this.lsmTreeTarget.getStats()
|
||
|
||
bytes += sourceStats.memTableMemory
|
||
bytes += targetStats.memTableMemory
|
||
|
||
// Verb ID set (memory-efficient: IDs only, ~8 bytes per ID pointer)
|
||
// Previous verbIndex Map stored full objects (128 bytes each = 128GB @ 1B verbs)
|
||
// Now: verbIdSet stores only IDs (~8 bytes each = ~100KB @ 1B verbs) = 1,280,000x reduction
|
||
bytes += this.verbIdSet.size * 8
|
||
|
||
// Note: Bloom filters and zone maps are in LSM-tree MemTable memory
|
||
// Full verb objects loaded on-demand via UnifiedCache with LRU eviction
|
||
|
||
return bytes
|
||
}
|
||
|
||
/**
|
||
* Get comprehensive statistics
|
||
*/
|
||
getStats(): GraphIndexStats {
|
||
const sourceStats = this.lsmTreeSource.getStats()
|
||
const targetStats = this.lsmTreeTarget.getStats()
|
||
|
||
return {
|
||
totalRelationships: this.size(),
|
||
sourceNodes: sourceStats.sstableCount,
|
||
targetNodes: targetStats.sstableCount,
|
||
memoryUsage: this.calculateMemoryUsage(),
|
||
lastRebuild: this.rebuildStartTime,
|
||
rebuildTime: this.isRebuilding ? Date.now() - this.rebuildStartTime : 0
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Start auto-flush timer
|
||
*/
|
||
private startAutoFlush(): void {
|
||
this.flushTimer = setInterval(async () => {
|
||
await this.flush()
|
||
}, this.config.flushInterval)
|
||
}
|
||
|
||
/**
|
||
* Flush LSM-tree MemTables to disk
|
||
* CRITICAL FIX: Now public so it can be called from brain.flush()
|
||
*/
|
||
async flush(): Promise<void> {
|
||
if (!this.initialized) {
|
||
return
|
||
}
|
||
|
||
const startTime = Date.now()
|
||
|
||
// 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
|
||
|
||
prodLog.debug(`GraphAdjacencyIndex: Flush completed in ${elapsed}ms`)
|
||
}
|
||
|
||
/**
|
||
* Clean shutdown
|
||
*/
|
||
async close(): Promise<void> {
|
||
if (this.flushTimer) {
|
||
clearInterval(this.flushTimer)
|
||
this.flushTimer = undefined
|
||
}
|
||
|
||
// Close all 4 LSM-trees (will flush MemTables to SSTables)
|
||
if (this.initialized) {
|
||
await Promise.all([
|
||
this.lsmTreeSource.close(),
|
||
this.lsmTreeTarget.close(),
|
||
this.lsmTreeVerbsBySource.close(),
|
||
this.lsmTreeVerbsByTarget.close(),
|
||
])
|
||
}
|
||
|
||
prodLog.info('GraphAdjacencyIndex: Shutdown complete')
|
||
}
|
||
|
||
/**
|
||
* Check if index is healthy
|
||
*/
|
||
isHealthy(): boolean {
|
||
if (!this.initialized) {
|
||
return false
|
||
}
|
||
|
||
return (
|
||
!this.isRebuilding &&
|
||
this.lsmTreeSource.isHealthy() &&
|
||
this.lsmTreeTarget.isHealthy()
|
||
)
|
||
}
|
||
}
|