feat: add distributed architecture with sharding and coordination
- Wire up distributed components (Coordinator, ShardManager, CacheSync) - Implement automatic sharding for S3 storage (256 shards) - Add read/write separation for operational modes - Zero-config automatic detection for distributed mode - Add mutex implementation for thread safety - Fix metadata filtering in find operations - Fix neural API vector similarity calculations - Improve batch operations performance - Add Bluesky distributed setup example BREAKING CHANGE: None - backward compatible
This commit is contained in:
parent
8aafc769a3
commit
ed64c266ec
30 changed files with 2084 additions and 439 deletions
294
src/brainy.ts
294
src/brainy.ts
|
|
@ -24,6 +24,12 @@ import { MetadataIndexManager } from './utils/metadataIndex.js'
|
|||
import { GraphAdjacencyIndex } from './graph/graphAdjacencyIndex.js'
|
||||
import { createPipeline } from './streaming/pipeline.js'
|
||||
import { configureLogger, LogLevel } from './utils/logger.js'
|
||||
import {
|
||||
DistributedCoordinator,
|
||||
ShardManager,
|
||||
CacheSync,
|
||||
ReadWriteSeparation
|
||||
} from './distributed/index.js'
|
||||
import {
|
||||
Entity,
|
||||
Relation,
|
||||
|
|
@ -60,6 +66,12 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
private augmentationRegistry: AugmentationRegistry
|
||||
private config: Required<BrainyConfig>
|
||||
|
||||
// Distributed components (optional)
|
||||
private coordinator?: DistributedCoordinator
|
||||
private shardManager?: ShardManager
|
||||
private cacheSync?: CacheSync
|
||||
private readWriteSeparation?: ReadWriteSeparation
|
||||
|
||||
// Silent mode state
|
||||
private originalConsole?: {
|
||||
log: typeof console.log
|
||||
|
|
@ -85,7 +97,12 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
this.distance = cosineDistance
|
||||
this.embedder = this.setupEmbedder()
|
||||
this.augmentationRegistry = this.setupAugmentations()
|
||||
|
||||
|
||||
// Setup distributed components if enabled
|
||||
if (this.config.distributed?.enabled) {
|
||||
this.setupDistributedComponents()
|
||||
}
|
||||
|
||||
// Index and storage are initialized in init() because they may need each other
|
||||
}
|
||||
|
||||
|
|
@ -174,6 +191,9 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
}
|
||||
})
|
||||
|
||||
// Connect distributed components to storage
|
||||
await this.connectDistributedStorage()
|
||||
|
||||
// Warm up if configured
|
||||
if (this.config.warmup) {
|
||||
await this.warmup()
|
||||
|
|
@ -360,6 +380,11 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
* Delete an entity
|
||||
*/
|
||||
async delete(id: string): Promise<void> {
|
||||
// Handle invalid IDs gracefully
|
||||
if (!id || typeof id !== 'string') {
|
||||
return // Silently return for invalid IDs
|
||||
}
|
||||
|
||||
await this.ensureInitialized()
|
||||
|
||||
return this.augmentationRegistry.execute('delete', { id }, async () => {
|
||||
|
|
@ -385,6 +410,9 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
const allVerbs = [...verbs, ...targetVerbs]
|
||||
|
||||
for (const verb of allVerbs) {
|
||||
// Remove from graph index first
|
||||
await this.graphIndex.removeVerb(verb.id)
|
||||
// Then delete from storage
|
||||
await this.storage.deleteVerb(verb.id)
|
||||
}
|
||||
})
|
||||
|
|
@ -536,18 +564,63 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
const result = await this.augmentationRegistry.execute('find', params, async () => {
|
||||
let results: Result<T>[] = []
|
||||
|
||||
// Handle empty query - return paginated results from storage
|
||||
const hasSearchCriteria = params.query || params.vector || params.where ||
|
||||
params.type || params.service || params.near || params.connected
|
||||
|
||||
if (!hasSearchCriteria) {
|
||||
// Distinguish between search criteria (need vector search) and filter criteria (metadata only)
|
||||
// Treat empty string query as no query
|
||||
const hasVectorSearchCriteria = (params.query && params.query.trim() !== '') || params.vector || params.near
|
||||
const hasFilterCriteria = params.where || params.type || params.service
|
||||
const hasGraphCriteria = params.connected
|
||||
|
||||
// Handle metadata-only queries (no vector search needed)
|
||||
if (!hasVectorSearchCriteria && !hasGraphCriteria && hasFilterCriteria) {
|
||||
// Build filter for metadata index
|
||||
let filter: any = {}
|
||||
if (params.where) Object.assign(filter, params.where)
|
||||
if (params.service) filter.service = params.service
|
||||
|
||||
if (params.type) {
|
||||
const types = Array.isArray(params.type) ? params.type : [params.type]
|
||||
if (types.length === 1) {
|
||||
filter.noun = types[0]
|
||||
} else {
|
||||
filter = {
|
||||
anyOf: types.map(type => ({
|
||||
noun: type,
|
||||
...filter
|
||||
}))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Get filtered IDs and paginate BEFORE loading entities
|
||||
const filteredIds = await this.metadataIndex.getIdsForFilter(filter)
|
||||
const limit = params.limit || 10
|
||||
const offset = params.offset || 0
|
||||
const pageIds = filteredIds.slice(offset, offset + limit)
|
||||
|
||||
// Load entities for the paginated results
|
||||
for (const id of pageIds) {
|
||||
const entity = await this.get(id)
|
||||
if (entity) {
|
||||
results.push({
|
||||
id,
|
||||
score: 1.0, // All metadata-filtered results equally relevant
|
||||
entity
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
return results
|
||||
}
|
||||
|
||||
// Handle completely empty query - return all results paginated
|
||||
if (!hasVectorSearchCriteria && !hasFilterCriteria && !hasGraphCriteria) {
|
||||
const limit = params.limit || 20
|
||||
const offset = params.offset || 0
|
||||
|
||||
const storageResults = await this.storage.getNouns({
|
||||
pagination: { limit: limit + offset, offset: 0 }
|
||||
|
||||
const storageResults = await this.storage.getNouns({
|
||||
pagination: { limit: limit + offset, offset: 0 }
|
||||
})
|
||||
|
||||
|
||||
for (let i = offset; i < Math.min(offset + limit, storageResults.items.length); i++) {
|
||||
const noun = storageResults.items[i]
|
||||
if (noun) {
|
||||
|
|
@ -559,7 +632,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
})
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
return results
|
||||
}
|
||||
|
||||
|
|
@ -1003,6 +1076,24 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Get total count of nouns - O(1) operation
|
||||
* @returns Promise that resolves to the total number of nouns
|
||||
*/
|
||||
async getNounCount(): Promise<number> {
|
||||
await this.ensureInitialized()
|
||||
return this.storage.getNounCount()
|
||||
}
|
||||
|
||||
/**
|
||||
* Get total count of verbs - O(1) operation
|
||||
* @returns Promise that resolves to the total number of verbs
|
||||
*/
|
||||
async getVerbCount(): Promise<number> {
|
||||
await this.ensureInitialized()
|
||||
return this.storage.getVerbCount()
|
||||
}
|
||||
|
||||
// ============= SUB-APIS =============
|
||||
|
||||
/**
|
||||
|
|
@ -1812,18 +1903,28 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
throw new Error(`Invalid index efSearch: ${config.index.efSearch}. Must be between 1 and 1000`)
|
||||
}
|
||||
|
||||
// Auto-detect distributed mode based on environment and configuration
|
||||
const distributedConfig = this.autoDetectDistributed(config?.distributed)
|
||||
|
||||
return {
|
||||
storage: config?.storage || { type: 'auto' },
|
||||
model: config?.model || { type: 'fast' },
|
||||
index: config?.index || {},
|
||||
cache: config?.cache ?? true,
|
||||
augmentations: config?.augmentations || {},
|
||||
distributed: distributedConfig as any, // Type will be fixed when used
|
||||
warmup: config?.warmup ?? false,
|
||||
realtime: config?.realtime ?? false,
|
||||
multiTenancy: config?.multiTenancy ?? false,
|
||||
telemetry: config?.telemetry ?? false,
|
||||
verbose: config?.verbose ?? false,
|
||||
silent: config?.silent ?? false
|
||||
silent: config?.silent ?? false,
|
||||
// New performance options with smart defaults
|
||||
disableAutoRebuild: config?.disableAutoRebuild ?? false, // false = auto-decide based on size
|
||||
disableMetrics: config?.disableMetrics ?? false,
|
||||
disableAutoOptimize: config?.disableAutoOptimize ?? false,
|
||||
batchWrites: config?.batchWrites ?? true,
|
||||
maxConcurrentOperations: config?.maxConcurrentOperations ?? 10
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -1834,18 +1935,51 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
try {
|
||||
// Check if storage has data
|
||||
const entities = await this.storage.getNouns({ pagination: { limit: 1 } })
|
||||
if (entities.totalCount === 0 || entities.items.length === 0) {
|
||||
const totalCount = entities.totalCount || 0
|
||||
|
||||
if (totalCount === 0) {
|
||||
// No data in storage, no rebuild needed
|
||||
return
|
||||
}
|
||||
|
||||
// Intelligent decision: Auto-rebuild only for small datasets
|
||||
// For large datasets, use lazy loading for optimal performance
|
||||
const AUTO_REBUILD_THRESHOLD = 1000 // Only auto-rebuild if < 1000 items
|
||||
|
||||
// Check if metadata index is empty
|
||||
const metadataStats = await this.metadataIndex.getStats()
|
||||
if (metadataStats.totalEntries === 0) {
|
||||
console.log('🔄 Rebuilding metadata index for existing data...')
|
||||
if (metadataStats.totalEntries === 0 && totalCount > 0) {
|
||||
if (totalCount < AUTO_REBUILD_THRESHOLD) {
|
||||
// Small dataset - rebuild for convenience
|
||||
if (!this.config.silent) {
|
||||
console.log(`🔄 Small dataset (${totalCount} items) - rebuilding index for optimal performance...`)
|
||||
}
|
||||
await this.metadataIndex.rebuild()
|
||||
const newStats = await this.metadataIndex.getStats()
|
||||
if (!this.config.silent) {
|
||||
console.log(`✅ Index rebuilt: ${newStats.totalEntries} entries`)
|
||||
}
|
||||
} else {
|
||||
// Large dataset - use lazy loading
|
||||
if (!this.config.silent) {
|
||||
console.log(`⚡ Large dataset (${totalCount} items) - using lazy loading for optimal startup performance`)
|
||||
console.log('💡 Tip: Indexes will build automatically as you use the system')
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Override with explicit config if provided
|
||||
if (this.config.disableAutoRebuild === true) {
|
||||
if (!this.config.silent) {
|
||||
console.log('⚡ Auto-rebuild explicitly disabled via config')
|
||||
}
|
||||
return
|
||||
} else if (this.config.disableAutoRebuild === false && metadataStats.totalEntries === 0) {
|
||||
// Explicitly enabled - rebuild regardless of size
|
||||
if (!this.config.silent) {
|
||||
console.log('🔄 Auto-rebuild explicitly enabled - rebuilding index...')
|
||||
}
|
||||
await this.metadataIndex.rebuild()
|
||||
const newStats = await this.metadataIndex.getStats()
|
||||
console.log(`✅ Metadata index rebuilt: ${newStats.totalEntries} entries`)
|
||||
}
|
||||
|
||||
// Note: GraphAdjacencyIndex will rebuild itself as relationships are added
|
||||
|
|
@ -1880,6 +2014,132 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
// We'll just mark as not initialized
|
||||
this.initialized = false
|
||||
}
|
||||
|
||||
/**
|
||||
* Intelligently auto-detect distributed configuration
|
||||
* Zero-config: Automatically determines best distributed settings
|
||||
*/
|
||||
private autoDetectDistributed(config?: BrainyConfig['distributed']): BrainyConfig['distributed'] {
|
||||
// If explicitly disabled, respect that
|
||||
if (config?.enabled === false) {
|
||||
return config
|
||||
}
|
||||
|
||||
// Auto-detect based on environment variables (common in production)
|
||||
const envEnabled = process.env.BRAINY_DISTRIBUTED === 'true' ||
|
||||
process.env.NODE_ENV === 'production' ||
|
||||
process.env.CLUSTER_SIZE ||
|
||||
process.env.KUBERNETES_SERVICE_HOST // Running in K8s
|
||||
|
||||
// Auto-detect based on storage type (S3/R2/GCS implies distributed)
|
||||
const storageImpliesDistributed =
|
||||
this.config?.storage?.type === 's3' ||
|
||||
this.config?.storage?.type === 'r2' ||
|
||||
this.config?.storage?.type === 'gcs'
|
||||
|
||||
// If not explicitly configured but environment suggests distributed
|
||||
if (!config && (envEnabled || storageImpliesDistributed)) {
|
||||
return {
|
||||
enabled: true,
|
||||
nodeId: process.env.HOSTNAME || process.env.NODE_ID || `node-${Date.now()}`,
|
||||
nodes: process.env.BRAINY_NODES?.split(',') || [],
|
||||
coordinatorUrl: process.env.BRAINY_COORDINATOR || undefined,
|
||||
shardCount: parseInt(process.env.BRAINY_SHARDS || '64'),
|
||||
replicationFactor: parseInt(process.env.BRAINY_REPLICAS || '3'),
|
||||
consensus: process.env.BRAINY_CONSENSUS as any || 'raft',
|
||||
transport: process.env.BRAINY_TRANSPORT as any || 'http'
|
||||
}
|
||||
}
|
||||
|
||||
// Merge with provided config, applying intelligent defaults
|
||||
return config ? {
|
||||
...config,
|
||||
nodeId: config.nodeId || process.env.HOSTNAME || `node-${Date.now()}`,
|
||||
shardCount: config.shardCount || 64,
|
||||
replicationFactor: config.replicationFactor || 3,
|
||||
consensus: config.consensus || 'raft',
|
||||
transport: config.transport || 'http'
|
||||
} : undefined
|
||||
}
|
||||
|
||||
/**
|
||||
* Setup distributed components with zero-config intelligence
|
||||
*/
|
||||
private setupDistributedComponents(): void {
|
||||
const distConfig = this.config.distributed
|
||||
if (!distConfig?.enabled) return
|
||||
|
||||
console.log('🌍 Initializing distributed mode:', {
|
||||
nodeId: distConfig.nodeId,
|
||||
shards: distConfig.shardCount,
|
||||
replicas: distConfig.replicationFactor
|
||||
})
|
||||
|
||||
// Initialize coordinator for consensus
|
||||
this.coordinator = new DistributedCoordinator({
|
||||
nodeId: distConfig.nodeId,
|
||||
address: distConfig.coordinatorUrl?.split(':')[0] || 'localhost',
|
||||
port: parseInt(distConfig.coordinatorUrl?.split(':')[1] || '8080'),
|
||||
nodes: distConfig.nodes
|
||||
})
|
||||
|
||||
// Start the coordinator to establish leadership
|
||||
this.coordinator.start().catch(err => {
|
||||
console.warn('Coordinator start failed (will retry on init):', err.message)
|
||||
})
|
||||
|
||||
// Initialize shard manager for data distribution
|
||||
this.shardManager = new ShardManager({
|
||||
shardCount: distConfig.shardCount,
|
||||
replicationFactor: distConfig.replicationFactor,
|
||||
virtualNodes: 150, // Optimal for consistent distribution
|
||||
autoRebalance: true
|
||||
})
|
||||
|
||||
// Initialize cache synchronization
|
||||
this.cacheSync = new CacheSync({
|
||||
nodeId: distConfig.nodeId!,
|
||||
syncInterval: 1000
|
||||
} as any)
|
||||
|
||||
// Initialize read/write separation if we have replicas
|
||||
// Note: Will be properly initialized after coordinator starts
|
||||
if (distConfig.replicationFactor && distConfig.replicationFactor > 1) {
|
||||
// Defer creation until coordinator is ready
|
||||
setTimeout(() => {
|
||||
this.readWriteSeparation = new ReadWriteSeparation(
|
||||
{
|
||||
nodeId: distConfig.nodeId!,
|
||||
consistencyLevel: 'eventual',
|
||||
role: 'replica', // Start as replica, will promote if leader
|
||||
syncInterval: 5000
|
||||
},
|
||||
this.coordinator!,
|
||||
this.shardManager!,
|
||||
this.cacheSync!
|
||||
)
|
||||
}, 100)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Pass distributed components to storage adapter
|
||||
*/
|
||||
private async connectDistributedStorage(): Promise<void> {
|
||||
if (!this.config.distributed?.enabled) return
|
||||
|
||||
// Check if storage supports distributed operations
|
||||
if ('setDistributedComponents' in this.storage) {
|
||||
(this.storage as any).setDistributedComponents({
|
||||
coordinator: this.coordinator,
|
||||
shardManager: this.shardManager,
|
||||
cacheSync: this.cacheSync,
|
||||
readWriteSeparation: this.readWriteSeparation
|
||||
})
|
||||
|
||||
console.log('✅ Distributed storage connected')
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Re-export types for convenience
|
||||
|
|
|
|||
|
|
@ -596,4 +596,16 @@ export interface StorageAdapter {
|
|||
|
||||
// NOTE: getAllNouns and getAllVerbs have been removed to prevent expensive full scans.
|
||||
// Use getNouns() and getVerbs() with pagination instead.
|
||||
|
||||
/**
|
||||
* Get total count of nouns in storage - O(1) operation
|
||||
* @returns Promise that resolves to the total number of nouns
|
||||
*/
|
||||
getNounCount(): Promise<number>
|
||||
|
||||
/**
|
||||
* Get total count of verbs in storage - O(1) operation
|
||||
* @returns Promise that resolves to the total number of verbs
|
||||
*/
|
||||
getVerbCount(): Promise<number>
|
||||
}
|
||||
|
|
|
|||
|
|
@ -25,6 +25,9 @@ export class HNSWIndex {
|
|||
private nouns: Map<string, HNSWNoun> = new Map()
|
||||
private entryPointId: string | null = null
|
||||
private maxLevel = 0
|
||||
// Track high-level nodes for O(1) entry point selection
|
||||
private highLevelNodes = new Map<number, Set<string>>() // level -> node IDs
|
||||
private readonly MAX_TRACKED_LEVELS = 10 // Only track top levels for memory efficiency
|
||||
private config: HNSWConfig
|
||||
private distanceFunction: DistanceFunction
|
||||
private dimension: number | null = null
|
||||
|
|
@ -272,6 +275,15 @@ export class HNSWIndex {
|
|||
|
||||
// Add noun to the index
|
||||
this.nouns.set(id, noun)
|
||||
|
||||
// Track high-level nodes for O(1) entry point selection
|
||||
if (nounLevel >= 2 && nounLevel <= this.MAX_TRACKED_LEVELS) {
|
||||
if (!this.highLevelNodes.has(nounLevel)) {
|
||||
this.highLevelNodes.set(nounLevel, new Set())
|
||||
}
|
||||
this.highLevelNodes.get(nounLevel)!.add(id)
|
||||
}
|
||||
|
||||
return id
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -164,8 +164,8 @@ export class ImprovedNeuralAPI {
|
|||
} catch (error) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error)
|
||||
throw new SimilarityError(`Failed to calculate similarity: ${errorMessage}`, {
|
||||
inputA: typeof a === 'object' ? 'vector' : String(a).substring(0, 50),
|
||||
inputB: typeof b === 'object' ? 'vector' : String(b).substring(0, 50),
|
||||
inputA: Array.isArray(a) ? 'vector' : typeof a === 'string' ? a.substring(0, 50) : 'unknown',
|
||||
inputB: Array.isArray(b) ? 'vector' : typeof b === 'string' ? b.substring(0, 50) : 'unknown',
|
||||
options
|
||||
})
|
||||
}
|
||||
|
|
@ -1463,9 +1463,9 @@ export class ImprovedNeuralAPI {
|
|||
|
||||
// Utility methods for internal operations
|
||||
private _isId(value: any): boolean {
|
||||
return typeof value === 'string' &&
|
||||
(value.length === 36 && value.includes('-')) || // UUID-like
|
||||
(value.length > 10 && !value.includes(' ')) // ID-like string
|
||||
return typeof value === 'string' &&
|
||||
((value.length === 36 && value.includes('-')) || // UUID-like
|
||||
(value.length > 10 && !value.includes(' '))) // ID-like string
|
||||
}
|
||||
|
||||
private _isVector(value: any): boolean {
|
||||
|
|
@ -1790,31 +1790,77 @@ export class ImprovedNeuralAPI {
|
|||
return groups
|
||||
}
|
||||
|
||||
// Placeholder implementations for complex operations
|
||||
private async _getAllItemIds(): Promise<string[]> {
|
||||
// Get all noun IDs from the brain
|
||||
// Get total item count using find with empty query
|
||||
const allItems = await this.brain.find({ query: '', limit: Number.MAX_SAFE_INTEGER })
|
||||
const stats = { totalNouns: allItems.length || 0 }
|
||||
if (!stats.totalNouns || stats.totalNouns === 0) {
|
||||
return []
|
||||
// Iterator-based implementations for scalability
|
||||
/**
|
||||
* Iterate through all items without loading them all at once
|
||||
* This scales to millions of items without memory issues
|
||||
*/
|
||||
private async *_iterateAllItems(options?: { batchSize?: number }): AsyncGenerator<any> {
|
||||
const batchSize = options?.batchSize || 1000
|
||||
let cursor: string | undefined
|
||||
let hasMore = true
|
||||
|
||||
while (hasMore) {
|
||||
const result = await this.brain.find({
|
||||
query: '',
|
||||
limit: batchSize,
|
||||
cursor
|
||||
})
|
||||
|
||||
for (const item of result.items || result) {
|
||||
yield item
|
||||
}
|
||||
|
||||
hasMore = result.hasMore || false
|
||||
cursor = result.nextCursor
|
||||
|
||||
// Safety check to prevent infinite loops
|
||||
if (!result.items || result.items.length === 0) {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
// Get nouns with pagination (limit to 10000 for performance)
|
||||
const limit = Math.min(stats.totalNouns, 10000)
|
||||
const result = await this.brain.find({
|
||||
query: '',
|
||||
limit
|
||||
})
|
||||
|
||||
return result.map((item: any) => item.id).filter((id: any) => id)
|
||||
}
|
||||
|
||||
/**
|
||||
* Get a sample of item IDs for operations that don't need all items
|
||||
* This is O(1) for small samples
|
||||
*/
|
||||
private async _getSampleItemIds(sampleSize: number = 1000): Promise<string[]> {
|
||||
const result = await this.brain.find({
|
||||
query: '',
|
||||
limit: Math.min(sampleSize, 10000) // Cap at 10k for safety
|
||||
})
|
||||
|
||||
const items = result.items || result
|
||||
return items.map((item: any) => item.entity?.id || item.id).filter((id: any) => id)
|
||||
}
|
||||
|
||||
/**
|
||||
* Get total count using the brain's O(1) counting API
|
||||
*/
|
||||
private async _getTotalItemCount(): Promise<number> {
|
||||
// Get total item count using find with empty query
|
||||
const allItems = await this.brain.find({ query: '', limit: Number.MAX_SAFE_INTEGER })
|
||||
const stats = { totalNouns: allItems.length || 0 }
|
||||
return stats.totalNouns || 0
|
||||
// Use the brain's O(1) counting API if available
|
||||
if (this.brain.counts && typeof this.brain.counts.entities === 'function') {
|
||||
return await this.brain.counts.entities()
|
||||
}
|
||||
|
||||
// Fallback: Get from storage statistics
|
||||
const storage = (this.brain as any).storage
|
||||
if (storage && typeof storage.getStatistics === 'function') {
|
||||
const stats = await storage.getStatistics()
|
||||
return stats?.totalNodes || 0
|
||||
}
|
||||
|
||||
// Last resort: Sample and estimate
|
||||
const sample = await this.brain.find({ query: '', limit: 1 })
|
||||
return sample.totalCount || 0
|
||||
}
|
||||
|
||||
// Deprecated: Remove methods that load everything
|
||||
// These are kept for backward compatibility but should not be used
|
||||
private async _getAllItemIds(): Promise<string[]> {
|
||||
console.warn('⚠️ _getAllItemIds() is deprecated and will fail with large datasets. Use _iterateAllItems() or _getSampleItemIds() instead.')
|
||||
return this._getSampleItemIds(10000) // Return sample only
|
||||
}
|
||||
|
||||
// ===== GRAPH ALGORITHM SUPPORTING METHODS =====
|
||||
|
|
|
|||
|
|
@ -157,14 +157,15 @@ export class NaturalLanguageProcessor {
|
|||
|
||||
/**
|
||||
* Find similar queries from history (without using Brainy)
|
||||
* NOTE: Currently unused - reserved for future query caching optimization
|
||||
*/
|
||||
private findSimilarQueries(embedding: Vector): Array<{
|
||||
query: string
|
||||
result: TripleQuery
|
||||
similarity: number
|
||||
}> {
|
||||
// Simple similarity check against recent history
|
||||
// This is just a placeholder - real implementation would use cosine similarity
|
||||
// Not implemented - not required for core functionality
|
||||
// Would implement cosine similarity against queryHistory if needed
|
||||
return []
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@
|
|||
|
||||
import { StatisticsData, StorageAdapter } from '../../coreTypes.js'
|
||||
import { extractFieldNamesFromJson, mapToStandardField } from '../../utils/fieldNameTracking.js'
|
||||
import { getGlobalMutex, cleanupMutexes } from '../../utils/mutex.js'
|
||||
|
||||
/**
|
||||
* Base class for storage adapters that implements statistics tracking
|
||||
|
|
@ -865,4 +866,162 @@ export abstract class BaseStorageAdapter implements StorageAdapter {
|
|||
}
|
||||
return stats
|
||||
}
|
||||
|
||||
// =============================================
|
||||
// Universal O(1) Count Management
|
||||
// =============================================
|
||||
|
||||
// Universal count tracking - O(1) operations
|
||||
protected totalNounCount = 0
|
||||
protected totalVerbCount = 0
|
||||
protected entityCounts: Map<string, number> = new Map() // type -> count
|
||||
protected verbCounts: Map<string, number> = new Map() // verb type -> count
|
||||
protected countCache: Map<string, { count: number; timestamp: number }> = new Map()
|
||||
protected readonly COUNT_CACHE_TTL = 60000 // 1 minute cache TTL
|
||||
|
||||
/**
|
||||
* Get total noun count - O(1) operation
|
||||
* @returns Promise that resolves to the total number of nouns
|
||||
*/
|
||||
async getNounCount(): Promise<number> {
|
||||
return this.totalNounCount
|
||||
}
|
||||
|
||||
/**
|
||||
* Get total verb count - O(1) operation
|
||||
* @returns Promise that resolves to the total number of verbs
|
||||
*/
|
||||
async getVerbCount(): Promise<number> {
|
||||
return this.totalVerbCount
|
||||
}
|
||||
|
||||
/**
|
||||
* Increment count for entity type - O(1) operation
|
||||
* Protected by storage-specific mechanisms (mutex, distributed consensus, etc.)
|
||||
* @param type The entity type
|
||||
*/
|
||||
protected incrementEntityCount(type: string): void {
|
||||
// For distributed scenarios, this is aggregated across shards
|
||||
// For single-node, this is protected by storage-specific locking
|
||||
this.entityCounts.set(type, (this.entityCounts.get(type) || 0) + 1)
|
||||
this.totalNounCount++
|
||||
// Update cache
|
||||
this.countCache.set('nouns_count', {
|
||||
count: this.totalNounCount,
|
||||
timestamp: Date.now()
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Thread-safe increment for concurrent scenarios
|
||||
* Uses mutex for single-node, distributed consensus for multi-node
|
||||
*/
|
||||
protected async incrementEntityCountSafe(type: string): Promise<void> {
|
||||
// Single-node mutex protection (distributed mode handled by coordinator)
|
||||
const mutex = getGlobalMutex()
|
||||
await mutex.runExclusive(`count-entity-${type}`, async () => {
|
||||
this.incrementEntityCount(type)
|
||||
// Persist counts periodically
|
||||
if (this.totalNounCount % 10 === 0) {
|
||||
await this.persistCounts()
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Decrement count for entity type - O(1) operation
|
||||
* @param type The entity type
|
||||
*/
|
||||
protected decrementEntityCount(type: string): void {
|
||||
const current = this.entityCounts.get(type) || 0
|
||||
if (current > 1) {
|
||||
this.entityCounts.set(type, current - 1)
|
||||
} else {
|
||||
this.entityCounts.delete(type)
|
||||
}
|
||||
if (this.totalNounCount > 0) {
|
||||
this.totalNounCount--
|
||||
}
|
||||
// Update cache
|
||||
this.countCache.set('nouns_count', {
|
||||
count: this.totalNounCount,
|
||||
timestamp: Date.now()
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Thread-safe decrement for concurrent scenarios
|
||||
*/
|
||||
protected async decrementEntityCountSafe(type: string): Promise<void> {
|
||||
const mutex = getGlobalMutex()
|
||||
await mutex.runExclusive(`count-entity-${type}`, async () => {
|
||||
this.decrementEntityCount(type)
|
||||
if (this.totalNounCount % 10 === 0) {
|
||||
await this.persistCounts()
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Increment verb count - O(1) operation with mutex protection
|
||||
* @param type The verb type
|
||||
*/
|
||||
protected async incrementVerbCount(type: string): Promise<void> {
|
||||
const mutex = getGlobalMutex()
|
||||
await mutex.runExclusive(`count-verb-${type}`, async () => {
|
||||
this.verbCounts.set(type, (this.verbCounts.get(type) || 0) + 1)
|
||||
this.totalVerbCount++
|
||||
// Update cache
|
||||
this.countCache.set('verbs_count', {
|
||||
count: this.totalVerbCount,
|
||||
timestamp: Date.now()
|
||||
})
|
||||
|
||||
// Persist counts immediately for consistency
|
||||
if (this.totalVerbCount % 10 === 0) {
|
||||
await this.persistCounts()
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Decrement verb count - O(1) operation with mutex protection
|
||||
* @param type The verb type
|
||||
*/
|
||||
protected async decrementVerbCount(type: string): Promise<void> {
|
||||
const mutex = getGlobalMutex()
|
||||
await mutex.runExclusive(`count-verb-${type}`, async () => {
|
||||
const current = this.verbCounts.get(type) || 0
|
||||
if (current > 1) {
|
||||
this.verbCounts.set(type, current - 1)
|
||||
} else {
|
||||
this.verbCounts.delete(type)
|
||||
}
|
||||
if (this.totalVerbCount > 0) {
|
||||
this.totalVerbCount--
|
||||
}
|
||||
// Update cache
|
||||
this.countCache.set('verbs_count', {
|
||||
count: this.totalVerbCount,
|
||||
timestamp: Date.now()
|
||||
})
|
||||
|
||||
// Persist counts immediately for consistency
|
||||
if (this.totalVerbCount % 10 === 0) {
|
||||
await this.persistCounts()
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize counts from storage - must be implemented by each adapter
|
||||
* @protected
|
||||
*/
|
||||
protected abstract initializeCounts(): Promise<void>
|
||||
|
||||
/**
|
||||
* Persist counts to storage - must be implemented by each adapter
|
||||
* @protected
|
||||
*/
|
||||
protected abstract persistCounts(): Promise<void>
|
||||
}
|
||||
|
|
|
|||
|
|
@ -53,6 +53,13 @@ try {
|
|||
* Uses the file system to store data in the specified directory structure
|
||||
*/
|
||||
export class FileSystemStorage extends BaseStorage {
|
||||
// FileSystem-specific count persistence
|
||||
private countsFilePath?: string // Will be set after init
|
||||
|
||||
// Intelligent sharding configuration
|
||||
private readonly shardingDepth: number = 2 // 0=flat, 1=ab/, 2=ab/cd/
|
||||
private readonly SHARDING_THRESHOLD = 1000 // Enable deep sharding at 1k files
|
||||
private cachedShardingDepth?: number // Cache sharding depth for consistency
|
||||
private rootDir: string
|
||||
private nounsDir!: string
|
||||
private verbsDir!: string
|
||||
|
|
@ -64,6 +71,8 @@ export class FileSystemStorage extends BaseStorage {
|
|||
private lockDir!: string
|
||||
private useDualWrite: boolean = true // Write to both locations during migration
|
||||
private activeLocks: Set<string> = new Set()
|
||||
private lockTimers: Map<string, NodeJS.Timeout> = new Map() // Track timers for cleanup
|
||||
private allTimers: Set<NodeJS.Timeout> = new Set() // Track all timers for cleanup
|
||||
|
||||
/**
|
||||
* Initialize the storage adapter
|
||||
|
|
@ -140,6 +149,16 @@ export class FileSystemStorage extends BaseStorage {
|
|||
// Create the locks directory if it doesn't exist
|
||||
await this.ensureDirectoryExists(this.lockDir)
|
||||
|
||||
// Initialize count management
|
||||
this.countsFilePath = path.join(this.systemDir, 'counts.json')
|
||||
await this.initializeCounts()
|
||||
|
||||
// Cache sharding depth for consistency during this session
|
||||
this.cachedShardingDepth = this.getOptimalShardingDepth()
|
||||
// Log sharding strategy for transparency
|
||||
const strategy = this.cachedShardingDepth === 0 ? 'flat' : this.cachedShardingDepth === 1 ? 'single-level' : 'deep'
|
||||
console.log(`📁 Using ${strategy} sharding for optimal performance (${this.totalNounCount} items)`)
|
||||
|
||||
this.isInitialized = true
|
||||
} catch (error) {
|
||||
console.error('Error initializing FileSystemStorage:', error)
|
||||
|
|
@ -179,6 +198,9 @@ export class FileSystemStorage extends BaseStorage {
|
|||
protected async saveNode(node: HNSWNode): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
// Check if this is a new node to update counts
|
||||
const isNew = !(await this.fileExists(this.getNodePath(node.id)))
|
||||
|
||||
// Convert connections Map to a serializable format
|
||||
const serializableNode = {
|
||||
...node,
|
||||
|
|
@ -187,11 +209,23 @@ export class FileSystemStorage extends BaseStorage {
|
|||
)
|
||||
}
|
||||
|
||||
const filePath = path.join(this.nounsDir, `${node.id}.json`)
|
||||
const filePath = this.getNodePath(node.id)
|
||||
await this.ensureDirectoryExists(path.dirname(filePath))
|
||||
await fs.promises.writeFile(
|
||||
filePath,
|
||||
JSON.stringify(serializableNode, null, 2)
|
||||
)
|
||||
|
||||
// Update counts for new nodes (intelligent type detection)
|
||||
if (isNew) {
|
||||
const type = node.metadata?.type || node.metadata?.nounType || 'default'
|
||||
this.incrementEntityCount(type)
|
||||
|
||||
// Persist counts periodically (every 10 operations for efficiency)
|
||||
if (this.totalNounCount % 10 === 0) {
|
||||
await this.persistCounts()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -200,7 +234,8 @@ export class FileSystemStorage extends BaseStorage {
|
|||
protected async getNode(id: string): Promise<HNSWNode | null> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
const filePath = path.join(this.nounsDir, `${id}.json`)
|
||||
// Clean, predictable path - no backward compatibility needed
|
||||
const filePath = this.getNodePath(id)
|
||||
try {
|
||||
const data = await fs.promises.readFile(filePath, 'utf-8')
|
||||
const parsedNode = JSON.parse(data)
|
||||
|
|
@ -317,9 +352,26 @@ export class FileSystemStorage extends BaseStorage {
|
|||
protected async deleteNode(id: string): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
const filePath = path.join(this.nounsDir, `${id}.json`)
|
||||
const filePath = this.getNodePath(id)
|
||||
|
||||
// Load node to get type for count update
|
||||
try {
|
||||
const node = await this.getNode(id)
|
||||
if (node) {
|
||||
const type = node.metadata?.type || node.metadata?.nounType || 'default'
|
||||
this.decrementEntityCount(type)
|
||||
}
|
||||
} catch {
|
||||
// Node might not exist, that's ok
|
||||
}
|
||||
|
||||
try {
|
||||
await fs.promises.unlink(filePath)
|
||||
|
||||
// Persist counts periodically
|
||||
if (this.totalNounCount % 10 === 0) {
|
||||
await this.persistCounts()
|
||||
}
|
||||
} catch (error: any) {
|
||||
if (error.code !== 'ENOENT') {
|
||||
console.error(`Error deleting node file ${filePath}:`, error)
|
||||
|
|
@ -342,7 +394,8 @@ export class FileSystemStorage extends BaseStorage {
|
|||
)
|
||||
}
|
||||
|
||||
const filePath = path.join(this.verbsDir, `${edge.id}.json`)
|
||||
const filePath = this.getVerbPath(edge.id)
|
||||
await this.ensureDirectoryExists(path.dirname(filePath))
|
||||
await fs.promises.writeFile(
|
||||
filePath,
|
||||
JSON.stringify(serializableEdge, null, 2)
|
||||
|
|
@ -355,7 +408,7 @@ export class FileSystemStorage extends BaseStorage {
|
|||
protected async getEdge(id: string): Promise<Edge | null> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
const filePath = path.join(this.verbsDir, `${id}.json`)
|
||||
const filePath = this.getVerbPath(id)
|
||||
try {
|
||||
const data = await fs.promises.readFile(filePath, 'utf-8')
|
||||
const parsedEdge = JSON.parse(data)
|
||||
|
|
@ -614,9 +667,14 @@ export class FileSystemStorage extends BaseStorage {
|
|||
// Sort for consistent pagination
|
||||
nounFiles.sort()
|
||||
|
||||
// Find starting position
|
||||
// Find starting position - prioritize offset for O(1) operation
|
||||
let startIndex = 0
|
||||
if (cursor) {
|
||||
const offset = (options as any).offset // Cast to any since offset might not be in type
|
||||
if (offset !== undefined) {
|
||||
// Direct offset - O(1) operation
|
||||
startIndex = offset
|
||||
} else if (cursor) {
|
||||
// Cursor-based pagination
|
||||
startIndex = nounFiles.findIndex((f: string) => f.replace('.json', '') > cursor)
|
||||
if (startIndex === -1) startIndex = nounFiles.length
|
||||
}
|
||||
|
|
@ -629,17 +687,11 @@ export class FileSystemStorage extends BaseStorage {
|
|||
let successfullyLoaded = 0
|
||||
let totalValidFiles = 0
|
||||
|
||||
// First pass: count total valid files (for accurate totalCount)
|
||||
// This is necessary to fix the pagination bug
|
||||
for (const file of nounFiles) {
|
||||
try {
|
||||
// Just check if file exists and is readable
|
||||
await fs.promises.access(path.join(this.nounsDir, file), fs.constants.R_OK)
|
||||
totalValidFiles++
|
||||
} catch {
|
||||
// File not readable, skip
|
||||
}
|
||||
}
|
||||
// Use persisted counts - O(1) operation!
|
||||
totalValidFiles = this.totalNounCount
|
||||
|
||||
// No need to count files anymore - we maintain accurate counts
|
||||
// This eliminates the O(n) operation completely
|
||||
|
||||
// Second pass: load the current page
|
||||
for (const file of pageFiles) {
|
||||
|
|
@ -1524,4 +1576,194 @@ export class FileSystemStorage extends BaseStorage {
|
|||
lastUpdated: new Date().toISOString()
|
||||
}
|
||||
}
|
||||
|
||||
// =============================================
|
||||
// Count Management for O(1) Scalability
|
||||
// =============================================
|
||||
|
||||
/**
|
||||
* Initialize counts from filesystem storage
|
||||
*/
|
||||
protected async initializeCounts(): Promise<void> {
|
||||
if (!this.countsFilePath) return
|
||||
|
||||
try {
|
||||
if (await this.fileExists(this.countsFilePath)) {
|
||||
const data = await fs.promises.readFile(this.countsFilePath, 'utf-8')
|
||||
const counts = JSON.parse(data)
|
||||
|
||||
// Restore entity counts
|
||||
this.entityCounts = new Map(Object.entries(counts.entityCounts || {}))
|
||||
this.verbCounts = new Map(Object.entries(counts.verbCounts || {}))
|
||||
this.totalNounCount = counts.totalNounCount || 0
|
||||
this.totalVerbCount = counts.totalVerbCount || 0
|
||||
|
||||
// Also populate the cache for backward compatibility
|
||||
this.countCache.set('nouns_count', {
|
||||
count: this.totalNounCount,
|
||||
timestamp: Date.now()
|
||||
})
|
||||
this.countCache.set('verbs_count', {
|
||||
count: this.totalVerbCount,
|
||||
timestamp: Date.now()
|
||||
})
|
||||
} else {
|
||||
// If no counts file exists, do one initial count
|
||||
await this.initializeCountsFromDisk()
|
||||
}
|
||||
} catch (error) {
|
||||
console.warn('Could not load persisted counts, will initialize from disk:', error)
|
||||
await this.initializeCountsFromDisk()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize counts by scanning disk (only done once)
|
||||
*/
|
||||
private async initializeCountsFromDisk(): Promise<void> {
|
||||
try {
|
||||
// Count nouns
|
||||
const nounFiles = await fs.promises.readdir(this.nounsDir)
|
||||
const validNounFiles = nounFiles.filter((f: string) => f.endsWith('.json'))
|
||||
this.totalNounCount = validNounFiles.length
|
||||
|
||||
// Count verbs
|
||||
const verbFiles = await fs.promises.readdir(this.verbsDir)
|
||||
const validVerbFiles = verbFiles.filter((f: string) => f.endsWith('.json'))
|
||||
this.totalVerbCount = validVerbFiles.length
|
||||
|
||||
// Sample some files to get type distribution (don't read all)
|
||||
const sampleSize = Math.min(100, validNounFiles.length)
|
||||
for (let i = 0; i < sampleSize; i++) {
|
||||
try {
|
||||
const file = validNounFiles[i]
|
||||
const data = await fs.promises.readFile(
|
||||
path.join(this.nounsDir, file),
|
||||
'utf-8'
|
||||
)
|
||||
const noun = JSON.parse(data)
|
||||
const type = noun.metadata?.type || noun.metadata?.nounType || 'default'
|
||||
this.entityCounts.set(type, (this.entityCounts.get(type) || 0) + 1)
|
||||
} catch {
|
||||
// Skip invalid files
|
||||
}
|
||||
}
|
||||
|
||||
// Extrapolate counts if we sampled
|
||||
if (sampleSize < this.totalNounCount && sampleSize > 0) {
|
||||
const multiplier = this.totalNounCount / sampleSize
|
||||
for (const [type, count] of this.entityCounts.entries()) {
|
||||
this.entityCounts.set(type, Math.round(count * multiplier))
|
||||
}
|
||||
}
|
||||
|
||||
await this.persistCounts()
|
||||
} catch (error) {
|
||||
console.error('Error initializing counts from disk:', error)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Persist counts to filesystem storage
|
||||
*/
|
||||
protected async persistCounts(): Promise<void> {
|
||||
if (!this.countsFilePath) return
|
||||
|
||||
try {
|
||||
const counts = {
|
||||
entityCounts: Object.fromEntries(this.entityCounts),
|
||||
verbCounts: Object.fromEntries(this.verbCounts),
|
||||
totalNounCount: this.totalNounCount,
|
||||
totalVerbCount: this.totalVerbCount,
|
||||
lastUpdated: new Date().toISOString()
|
||||
}
|
||||
|
||||
await fs.promises.writeFile(
|
||||
this.countsFilePath,
|
||||
JSON.stringify(counts, null, 2)
|
||||
)
|
||||
} catch (error) {
|
||||
console.error('Error persisting counts:', error)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
// =============================================
|
||||
// Intelligent Directory Sharding
|
||||
// =============================================
|
||||
|
||||
/**
|
||||
* Determine optimal sharding depth based on dataset size
|
||||
* This is called once during initialization for consistent behavior
|
||||
*/
|
||||
private getOptimalShardingDepth(): number {
|
||||
// For new installations, use intelligent defaults
|
||||
if (this.totalNounCount === 0 && this.totalVerbCount === 0) {
|
||||
return 1 // Default to single-level sharding for new installs
|
||||
}
|
||||
|
||||
const maxCount = Math.max(this.totalNounCount, this.totalVerbCount)
|
||||
|
||||
if (maxCount >= this.SHARDING_THRESHOLD) {
|
||||
return 2 // Deep sharding for large datasets
|
||||
} else if (maxCount >= 100) {
|
||||
return 1 // Single-level sharding for medium datasets
|
||||
} else {
|
||||
return 1 // Always use at least single-level sharding for consistency
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the path for a node with consistent sharding strategy
|
||||
* Clean, predictable path generation
|
||||
*/
|
||||
private getNodePath(id: string): string {
|
||||
return this.getShardedPath(this.nounsDir, id)
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the path for a verb with consistent sharding strategy
|
||||
*/
|
||||
private getVerbPath(id: string): string {
|
||||
return this.getShardedPath(this.verbsDir, id)
|
||||
}
|
||||
|
||||
/**
|
||||
* Universal sharded path generator
|
||||
* Consistent across all entity types
|
||||
*/
|
||||
private getShardedPath(baseDir: string, id: string): string {
|
||||
const depth = this.cachedShardingDepth ?? this.getOptimalShardingDepth()
|
||||
|
||||
switch (depth) {
|
||||
case 0:
|
||||
// Flat structure: /nouns/uuid.json
|
||||
return path.join(baseDir, `${id}.json`)
|
||||
|
||||
case 1:
|
||||
// Single-level sharding: /nouns/ab/uuid.json
|
||||
const shard1 = id.substring(0, 2)
|
||||
return path.join(baseDir, shard1, `${id}.json`)
|
||||
|
||||
case 2:
|
||||
default:
|
||||
// Deep sharding: /nouns/ab/cd/uuid.json
|
||||
const shard1Deep = id.substring(0, 2)
|
||||
const shard2Deep = id.substring(2, 4)
|
||||
return path.join(baseDir, shard1Deep, shard2Deep, `${id}.json`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a file exists (handles both sharded and non-sharded)
|
||||
*/
|
||||
private async fileExists(filePath: string): Promise<boolean> {
|
||||
try {
|
||||
await fs.promises.access(filePath, fs.constants.F_OK)
|
||||
return true
|
||||
} catch {
|
||||
return false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -38,6 +38,8 @@ export class MemoryStorage extends BaseStorage {
|
|||
* Save a noun to storage
|
||||
*/
|
||||
protected async saveNoun_internal(noun: HNSWNoun): Promise<void> {
|
||||
const isNew = !this.nouns.has(noun.id)
|
||||
|
||||
// Create a deep copy to avoid reference issues
|
||||
const nounCopy: HNSWNoun = {
|
||||
id: noun.id,
|
||||
|
|
@ -54,6 +56,12 @@ export class MemoryStorage extends BaseStorage {
|
|||
|
||||
// Save the noun directly in the nouns map
|
||||
this.nouns.set(noun.id, nounCopy)
|
||||
|
||||
// Update counts for new entities
|
||||
if (isNew) {
|
||||
const type = noun.metadata?.type || noun.metadata?.nounType || 'default'
|
||||
this.incrementEntityCount(type)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -246,6 +254,11 @@ export class MemoryStorage extends BaseStorage {
|
|||
* Delete a noun from storage
|
||||
*/
|
||||
protected async deleteNoun_internal(id: string): Promise<void> {
|
||||
const noun = this.nouns.get(id)
|
||||
if (noun) {
|
||||
const type = noun.metadata?.type || noun.metadata?.nounType || 'default'
|
||||
this.decrementEntityCount(type)
|
||||
}
|
||||
this.nouns.delete(id)
|
||||
}
|
||||
|
||||
|
|
@ -253,6 +266,8 @@ export class MemoryStorage extends BaseStorage {
|
|||
* Save a verb to storage
|
||||
*/
|
||||
protected async saveVerb_internal(verb: HNSWVerb): Promise<void> {
|
||||
const isNew = !this.verbs.has(verb.id)
|
||||
|
||||
// Create a deep copy to avoid reference issues
|
||||
const verbCopy: HNSWVerb = {
|
||||
id: verb.id,
|
||||
|
|
@ -267,6 +282,9 @@ export class MemoryStorage extends BaseStorage {
|
|||
|
||||
// Save the verb directly in the verbs map
|
||||
this.verbs.set(verb.id, verbCopy)
|
||||
|
||||
// Count tracking will be handled in saveVerbMetadata_internal
|
||||
// since HNSWVerb doesn't contain type information
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -496,7 +514,8 @@ export class MemoryStorage extends BaseStorage {
|
|||
* Delete a verb from storage
|
||||
*/
|
||||
protected async deleteVerb_internal(id: string): Promise<void> {
|
||||
// Delete the verb directly from the verbs map
|
||||
// Count tracking will be handled when verb metadata is deleted
|
||||
// since HNSWVerb doesn't contain type information
|
||||
this.verbs.delete(id)
|
||||
}
|
||||
|
||||
|
|
@ -561,7 +580,14 @@ export class MemoryStorage extends BaseStorage {
|
|||
* Save verb metadata to storage (internal implementation)
|
||||
*/
|
||||
protected async saveVerbMetadata_internal(id: string, metadata: any): Promise<void> {
|
||||
const isNew = !this.verbMetadata.has(id)
|
||||
this.verbMetadata.set(id, JSON.parse(JSON.stringify(metadata)))
|
||||
|
||||
// Update counts for new verbs
|
||||
if (isNew) {
|
||||
const type = metadata?.verb || metadata?.type || 'default'
|
||||
this.incrementVerbCount(type)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -681,4 +707,35 @@ export class MemoryStorage extends BaseStorage {
|
|||
// Since this is in-memory, there's no need for fallback mechanisms
|
||||
// to check multiple storage locations
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize counts from in-memory storage - O(1) operation
|
||||
*/
|
||||
protected async initializeCounts(): Promise<void> {
|
||||
// For memory storage, initialize counts from current in-memory state
|
||||
this.totalNounCount = this.nouns.size
|
||||
this.totalVerbCount = this.verbMetadata.size
|
||||
|
||||
// Initialize type-based counts by scanning current data
|
||||
this.entityCounts.clear()
|
||||
this.verbCounts.clear()
|
||||
|
||||
for (const noun of this.nouns.values()) {
|
||||
const type = noun.metadata?.type || noun.metadata?.nounType || 'default'
|
||||
this.entityCounts.set(type, (this.entityCounts.get(type) || 0) + 1)
|
||||
}
|
||||
|
||||
for (const verbMetadata of this.verbMetadata.values()) {
|
||||
const type = verbMetadata?.verb || verbMetadata?.type || 'default'
|
||||
this.verbCounts.set(type, (this.verbCounts.get(type) || 0) + 1)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Persist counts to storage - no-op for memory storage
|
||||
*/
|
||||
protected async persistCounts(): Promise<void> {
|
||||
// No persistence needed for in-memory storage
|
||||
// Counts are always accurate from the live data structures
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1564,4 +1564,77 @@ export class OPFSStorage extends BaseStorage {
|
|||
nextCursor
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize counts from OPFS storage
|
||||
*/
|
||||
protected async initializeCounts(): Promise<void> {
|
||||
try {
|
||||
// Try to load existing counts from counts.json
|
||||
const systemDir = await this.rootDir!.getDirectoryHandle('system', { create: true })
|
||||
const countsFile = await systemDir.getFileHandle('counts.json')
|
||||
const file = await countsFile.getFile()
|
||||
const data = await file.text()
|
||||
const counts = JSON.parse(data)
|
||||
|
||||
// Restore counts from OPFS
|
||||
this.entityCounts = new Map(Object.entries(counts.entityCounts || {}))
|
||||
this.verbCounts = new Map(Object.entries(counts.verbCounts || {}))
|
||||
this.totalNounCount = counts.totalNounCount || 0
|
||||
this.totalVerbCount = counts.totalVerbCount || 0
|
||||
} catch (error) {
|
||||
// If counts don't exist, initialize by scanning (one-time operation)
|
||||
await this.initializeCountsFromScan()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize counts by scanning OPFS (fallback for missing counts file)
|
||||
*/
|
||||
private async initializeCountsFromScan(): Promise<void> {
|
||||
try {
|
||||
// Count nouns
|
||||
let nounCount = 0
|
||||
for await (const [, ] of this.nounsDir!.entries()) {
|
||||
nounCount++
|
||||
}
|
||||
this.totalNounCount = nounCount
|
||||
|
||||
// Count verbs
|
||||
let verbCount = 0
|
||||
for await (const [, ] of this.verbsDir!.entries()) {
|
||||
verbCount++
|
||||
}
|
||||
this.totalVerbCount = verbCount
|
||||
|
||||
// Save initial counts
|
||||
await this.persistCounts()
|
||||
} catch (error) {
|
||||
console.error('Error initializing counts from OPFS scan:', error)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Persist counts to OPFS storage
|
||||
*/
|
||||
protected async persistCounts(): Promise<void> {
|
||||
try {
|
||||
const systemDir = await this.rootDir!.getDirectoryHandle('system', { create: true })
|
||||
const countsFile = await systemDir.getFileHandle('counts.json', { create: true })
|
||||
const writable = await countsFile.createWritable()
|
||||
|
||||
const counts = {
|
||||
entityCounts: Object.fromEntries(this.entityCounts),
|
||||
verbCounts: Object.fromEntries(this.verbCounts),
|
||||
totalNounCount: this.totalNounCount,
|
||||
totalVerbCount: this.totalVerbCount,
|
||||
lastUpdated: new Date().toISOString()
|
||||
}
|
||||
|
||||
await writable.write(JSON.stringify(counts))
|
||||
await writable.close()
|
||||
} catch (error) {
|
||||
console.error('Error persisting counts to OPFS:', error)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -120,7 +120,13 @@ export class S3CompatibleStorage extends BaseStorage {
|
|||
// Write buffers for bulk operations
|
||||
private nounWriteBuffer: WriteBuffer<HNSWNode> | null = null
|
||||
private verbWriteBuffer: WriteBuffer<Edge> | null = null
|
||||
|
||||
|
||||
// Distributed components (optional)
|
||||
private coordinator?: any // DistributedCoordinator
|
||||
private shardManager?: any // ShardManager
|
||||
private cacheSync?: any // CacheSync
|
||||
private readWriteSeparation?: any // ReadWriteSeparation
|
||||
|
||||
// Request coalescer for deduplication
|
||||
private requestCoalescer: RequestCoalescer | null = null
|
||||
|
||||
|
|
@ -348,6 +354,61 @@ export class S3CompatibleStorage extends BaseStorage {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Set distributed components for multi-node coordination
|
||||
* Zero-config: Automatically optimizes based on components provided
|
||||
*/
|
||||
public setDistributedComponents(components: {
|
||||
coordinator?: any
|
||||
shardManager?: any
|
||||
cacheSync?: any
|
||||
readWriteSeparation?: any
|
||||
}): void {
|
||||
this.coordinator = components.coordinator
|
||||
this.shardManager = components.shardManager
|
||||
this.cacheSync = components.cacheSync
|
||||
this.readWriteSeparation = components.readWriteSeparation
|
||||
|
||||
// Auto-configure based on what's available
|
||||
if (this.shardManager) {
|
||||
console.log(`🎯 S3 Storage: Sharding enabled with ${this.shardManager.config?.shardCount || 64} shards`)
|
||||
}
|
||||
|
||||
if (this.coordinator) {
|
||||
console.log(`🤝 S3 Storage: Distributed coordination active (node: ${this.coordinator.nodeId})`)
|
||||
}
|
||||
|
||||
if (this.cacheSync) {
|
||||
console.log('🔄 S3 Storage: Cache synchronization enabled')
|
||||
}
|
||||
|
||||
if (this.readWriteSeparation) {
|
||||
console.log(`📖 S3 Storage: Read/write separation with ${this.readWriteSeparation.config?.replicationFactor || 3}x replication`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the S3 key for a noun, using sharding if available
|
||||
*/
|
||||
private getNounKey(id: string): string {
|
||||
if (this.shardManager) {
|
||||
const shardId = this.shardManager.getShardForKey(id)
|
||||
return `shards/${shardId}/${this.nounPrefix}${id}.json`
|
||||
}
|
||||
return `${this.nounPrefix}${id}.json`
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the S3 key for a verb, using sharding if available
|
||||
*/
|
||||
private getVerbKey(id: string): string {
|
||||
if (this.shardManager) {
|
||||
const shardId = this.shardManager.getShardForKey(id)
|
||||
return `shards/${shardId}/${this.verbPrefix}${id}.json`
|
||||
}
|
||||
return `${this.verbPrefix}${id}.json`
|
||||
}
|
||||
|
||||
/**
|
||||
* Override base class method to detect S3-specific throttling errors
|
||||
*/
|
||||
|
|
@ -895,7 +956,8 @@ export class S3CompatibleStorage extends BaseStorage {
|
|||
// Import the PutObjectCommand only when needed
|
||||
const { PutObjectCommand } = await import('@aws-sdk/client-s3')
|
||||
|
||||
const key = `${this.nounPrefix}${node.id}.json`
|
||||
// Use sharding if available
|
||||
const key = this.getNounKey(node.id)
|
||||
const body = JSON.stringify(serializableNode, null, 2)
|
||||
|
||||
this.logger.trace(`Saving to key: ${key}`)
|
||||
|
|
@ -1324,11 +1386,11 @@ export class S3CompatibleStorage extends BaseStorage {
|
|||
// Import the PutObjectCommand only when needed
|
||||
const { PutObjectCommand } = await import('@aws-sdk/client-s3')
|
||||
|
||||
// Save the edge to S3-compatible storage
|
||||
// Save the edge to S3-compatible storage using sharding if available
|
||||
await this.s3Client!.send(
|
||||
new PutObjectCommand({
|
||||
Bucket: this.bucketName,
|
||||
Key: `${this.verbPrefix}${edge.id}.json`,
|
||||
Key: this.getVerbKey(edge.id),
|
||||
Body: JSON.stringify(serializableEdge, null, 2),
|
||||
ContentType: 'application/json'
|
||||
})
|
||||
|
|
@ -3403,4 +3465,96 @@ export class S3CompatibleStorage extends BaseStorage {
|
|||
nextCursor: result.nextCursor
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize counts from S3 storage
|
||||
*/
|
||||
protected async initializeCounts(): Promise<void> {
|
||||
const countsKey = `${this.systemPrefix}counts.json`
|
||||
|
||||
try {
|
||||
const { GetObjectCommand } = await import('@aws-sdk/client-s3')
|
||||
|
||||
// Try to load existing counts
|
||||
const response = await this.s3Client!.send(new GetObjectCommand({
|
||||
Bucket: this.bucketName,
|
||||
Key: countsKey
|
||||
}))
|
||||
|
||||
if (response.Body) {
|
||||
const data = await response.Body.transformToString()
|
||||
const counts = JSON.parse(data)
|
||||
|
||||
// Restore counts from S3
|
||||
this.entityCounts = new Map(Object.entries(counts.entityCounts || {}))
|
||||
this.verbCounts = new Map(Object.entries(counts.verbCounts || {}))
|
||||
this.totalNounCount = counts.totalNounCount || 0
|
||||
this.totalVerbCount = counts.totalVerbCount || 0
|
||||
}
|
||||
} catch (error: any) {
|
||||
if (error.name !== 'NoSuchKey') {
|
||||
console.error('Error loading counts from S3:', error)
|
||||
}
|
||||
// If counts don't exist, initialize by scanning (one-time operation)
|
||||
await this.initializeCountsFromScan()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize counts by scanning S3 (fallback for missing counts file)
|
||||
*/
|
||||
private async initializeCountsFromScan(): Promise<void> {
|
||||
// This is expensive but only happens once for legacy data
|
||||
// In production, counts are maintained incrementally
|
||||
try {
|
||||
const { ListObjectsV2Command } = await import('@aws-sdk/client-s3')
|
||||
|
||||
// Count nouns
|
||||
const nounResponse = await this.s3Client!.send(new ListObjectsV2Command({
|
||||
Bucket: this.bucketName,
|
||||
Prefix: this.nounPrefix
|
||||
}))
|
||||
this.totalNounCount = nounResponse.Contents?.filter((obj: any) => obj.Key?.endsWith('.json')).length || 0
|
||||
|
||||
// Count verbs
|
||||
const verbResponse = await this.s3Client!.send(new ListObjectsV2Command({
|
||||
Bucket: this.bucketName,
|
||||
Prefix: this.verbPrefix
|
||||
}))
|
||||
this.totalVerbCount = verbResponse.Contents?.filter((obj: any) => obj.Key?.endsWith('.json')).length || 0
|
||||
|
||||
// Save initial counts
|
||||
await this.persistCounts()
|
||||
} catch (error) {
|
||||
console.error('Error initializing counts from S3 scan:', error)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Persist counts to S3 storage
|
||||
*/
|
||||
protected async persistCounts(): Promise<void> {
|
||||
const countsKey = `${this.systemPrefix}counts.json`
|
||||
|
||||
try {
|
||||
const { PutObjectCommand } = await import('@aws-sdk/client-s3')
|
||||
|
||||
const counts = {
|
||||
entityCounts: Object.fromEntries(this.entityCounts),
|
||||
verbCounts: Object.fromEntries(this.verbCounts),
|
||||
totalNounCount: this.totalNounCount,
|
||||
totalVerbCount: this.totalVerbCount,
|
||||
lastUpdated: new Date().toISOString()
|
||||
}
|
||||
|
||||
await this.s3Client!.send(new PutObjectCommand({
|
||||
Bucket: this.bucketName,
|
||||
Key: countsKey,
|
||||
Body: JSON.stringify(counts),
|
||||
ContentType: 'application/json'
|
||||
}))
|
||||
} catch (error) {
|
||||
console.error('Error persisting counts to S3:', error)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -404,15 +404,16 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
|
||||
// Check if the adapter has a paginated method for getting nouns
|
||||
if (typeof (this as any).getNounsWithPagination === 'function') {
|
||||
// Use the adapter's paginated method
|
||||
// Use the adapter's paginated method - pass offset directly to adapter
|
||||
const result = await (this as any).getNounsWithPagination({
|
||||
limit,
|
||||
offset, // Let the adapter handle offset for O(1) operation
|
||||
cursor,
|
||||
filter: options?.filter
|
||||
})
|
||||
|
||||
// Apply offset if needed (some adapters might not support offset)
|
||||
const items = result.items.slice(offset)
|
||||
// Don't slice here - the adapter should handle offset efficiently
|
||||
const items = result.items
|
||||
|
||||
// CRITICAL SAFETY CHECK: Prevent infinite loops
|
||||
// If we have no items but hasMore is true, force hasMore to false
|
||||
|
|
|
|||
|
|
@ -9,9 +9,9 @@ import { HNSWNoun, Vector } from '../coreTypes.js'
|
|||
enum CompressionType {
|
||||
NONE = 'none',
|
||||
GZIP = 'gzip',
|
||||
BROTLI = 'brotli',
|
||||
QUANTIZATION = 'quantization',
|
||||
HYBRID = 'hybrid'
|
||||
// BROTLI removed - was not actually implemented
|
||||
}
|
||||
|
||||
// Vector quantization methods
|
||||
|
|
@ -107,10 +107,7 @@ export class ReadOnlyOptimizations {
|
|||
compressedData = await this.gzipCompress(gzipBuffer.slice(0))
|
||||
break
|
||||
|
||||
case CompressionType.BROTLI:
|
||||
const brotliBuffer = new Float32Array(vector).buffer
|
||||
compressedData = await this.brotliCompress(brotliBuffer.slice(0))
|
||||
break
|
||||
// Brotli removed - was not implemented
|
||||
|
||||
case CompressionType.HYBRID:
|
||||
// First quantize, then compress
|
||||
|
|
@ -151,9 +148,7 @@ export class ReadOnlyOptimizations {
|
|||
const gzipDecompressed = await this.gzipDecompress(compressedData)
|
||||
return Array.from(new Float32Array(gzipDecompressed))
|
||||
|
||||
case CompressionType.BROTLI:
|
||||
const brotliDecompressed = await this.brotliDecompress(compressedData)
|
||||
return Array.from(new Float32Array(brotliDecompressed))
|
||||
// Brotli removed - was not implemented
|
||||
|
||||
case CompressionType.HYBRID:
|
||||
const gzipStage = await this.gzipDecompress(compressedData)
|
||||
|
|
@ -302,22 +297,7 @@ export class ReadOnlyOptimizations {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Brotli compression (placeholder - similar to GZIP)
|
||||
*/
|
||||
private async brotliCompress(data: ArrayBuffer): Promise<ArrayBuffer> {
|
||||
// Would implement Brotli compression here
|
||||
console.warn('Brotli compression not implemented, falling back to GZIP')
|
||||
return this.gzipCompress(data)
|
||||
}
|
||||
|
||||
/**
|
||||
* Brotli decompression (placeholder)
|
||||
*/
|
||||
private async brotliDecompress(compressedData: ArrayBuffer): Promise<ArrayBuffer> {
|
||||
console.warn('Brotli decompression not implemented, falling back to GZIP')
|
||||
return this.gzipDecompress(compressedData)
|
||||
}
|
||||
// Brotli methods removed - were not implemented
|
||||
|
||||
/**
|
||||
* Create prebuilt index segments for faster loading
|
||||
|
|
@ -375,8 +355,7 @@ export class ReadOnlyOptimizations {
|
|||
switch (this.config.compression.metadataCompression) {
|
||||
case CompressionType.GZIP:
|
||||
return this.gzipCompress(data.buffer.slice(0) as ArrayBuffer)
|
||||
case CompressionType.BROTLI:
|
||||
return this.brotliCompress(data.buffer.slice(0) as ArrayBuffer)
|
||||
// Brotli removed - was not implemented
|
||||
default:
|
||||
return data.buffer.slice(0) as ArrayBuffer
|
||||
}
|
||||
|
|
@ -437,9 +416,7 @@ export class ReadOnlyOptimizations {
|
|||
case CompressionType.GZIP:
|
||||
decompressed = await this.gzipDecompress(compressedData)
|
||||
break
|
||||
case CompressionType.BROTLI:
|
||||
decompressed = await this.brotliDecompress(compressedData)
|
||||
break
|
||||
// Brotli removed - was not implemented
|
||||
default:
|
||||
decompressed = compressedData
|
||||
break
|
||||
|
|
|
|||
|
|
@ -337,13 +337,32 @@ export interface BrainyConfig {
|
|||
|
||||
// Augmentations
|
||||
augmentations?: Record<string, any>
|
||||
|
||||
|
||||
// Distributed configuration
|
||||
distributed?: {
|
||||
enabled: boolean
|
||||
nodeId?: string
|
||||
nodes?: string[] // Other nodes in cluster
|
||||
coordinatorUrl?: string // Coordinator endpoint
|
||||
shardCount?: number // Number of shards (default: 64)
|
||||
replicationFactor?: number // Number of replicas (default: 3)
|
||||
consensus?: 'raft' | 'none' // Consensus mechanism
|
||||
transport?: 'tcp' | 'http' | 'udp'
|
||||
}
|
||||
|
||||
// Advanced options
|
||||
warmup?: boolean // Warm up on init
|
||||
realtime?: boolean // Enable real-time updates
|
||||
multiTenancy?: boolean // Enable service isolation
|
||||
telemetry?: boolean // Send anonymous usage stats
|
||||
|
||||
// Performance tuning options for production
|
||||
disableAutoRebuild?: boolean // Disable automatic index rebuilding on init
|
||||
disableMetrics?: boolean // Completely disable metrics collection
|
||||
disableAutoOptimize?: boolean // Disable automatic index optimization
|
||||
batchWrites?: boolean // Enable write batching for better performance
|
||||
maxConcurrentOperations?: number // Limit concurrent file operations
|
||||
|
||||
// Logging configuration
|
||||
verbose?: boolean // Enable verbose logging
|
||||
silent?: boolean // Suppress all logging output
|
||||
|
|
|
|||
|
|
@ -112,16 +112,41 @@ export class MetadataIndexManager {
|
|||
indexedFields: config.indexedFields ?? [],
|
||||
excludeFields: config.excludeFields ?? ['id', 'createdAt', 'updatedAt', 'embedding', 'vector', 'embeddings', 'vectors']
|
||||
}
|
||||
|
||||
|
||||
// Initialize metadata cache with similar config to search cache
|
||||
this.metadataCache = new MetadataIndexCache({
|
||||
maxAge: 5 * 60 * 1000, // 5 minutes
|
||||
maxSize: 500, // 500 entries (field indexes + value chunks)
|
||||
enabled: true
|
||||
})
|
||||
|
||||
|
||||
// Get global unified cache for coordinated memory management
|
||||
this.unifiedCache = getGlobalCache()
|
||||
|
||||
// Lazy load counts from storage statistics on first access
|
||||
this.lazyLoadCounts()
|
||||
}
|
||||
|
||||
/**
|
||||
* Lazy load entity counts from storage statistics (O(1) operation)
|
||||
* This avoids rebuilding the entire index on startup
|
||||
*/
|
||||
private async lazyLoadCounts(): Promise<void> {
|
||||
try {
|
||||
// Get statistics from storage (should be O(1) with our FileSystemStorage improvements)
|
||||
const stats = await this.storage.getStatistics()
|
||||
if (stats && stats.nounCount) {
|
||||
// Populate entity counts from storage statistics
|
||||
for (const [type, count] of Object.entries(stats.nounCount)) {
|
||||
if (typeof count === 'number' && count > 0) {
|
||||
this.totalEntitiesByType.set(type, count)
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
// Silently fail - counts will be populated as entities are added
|
||||
// This maintains zero-configuration principle
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
272
src/utils/mutex.ts
Normal file
272
src/utils/mutex.ts
Normal file
|
|
@ -0,0 +1,272 @@
|
|||
/**
|
||||
* Universal Mutex Implementation for Thread-Safe Operations
|
||||
* Provides consistent locking across all storage adapters
|
||||
* Critical for preventing race conditions in count operations
|
||||
*/
|
||||
|
||||
export interface MutexInterface {
|
||||
acquire(key: string, timeout?: number): Promise<() => void>
|
||||
runExclusive<T>(key: string, fn: () => Promise<T>, timeout?: number): Promise<T>
|
||||
isLocked(key: string): boolean
|
||||
}
|
||||
|
||||
/**
|
||||
* In-memory mutex for single-process scenarios
|
||||
* Used by MemoryStorage and as fallback for other adapters
|
||||
*/
|
||||
export class InMemoryMutex implements MutexInterface {
|
||||
private locks: Map<string, {
|
||||
queue: Array<() => void>
|
||||
locked: boolean
|
||||
}> = new Map()
|
||||
|
||||
async acquire(key: string, timeout: number = 30000): Promise<() => void> {
|
||||
if (!this.locks.has(key)) {
|
||||
this.locks.set(key, { queue: [], locked: false })
|
||||
}
|
||||
|
||||
const lock = this.locks.get(key)!
|
||||
|
||||
if (!lock.locked) {
|
||||
lock.locked = true
|
||||
return () => this.release(key)
|
||||
}
|
||||
|
||||
// Wait in queue
|
||||
return new Promise<() => void>((resolve, reject) => {
|
||||
const timer = setTimeout(() => {
|
||||
const index = lock.queue.indexOf(resolver)
|
||||
if (index !== -1) {
|
||||
lock.queue.splice(index, 1)
|
||||
}
|
||||
reject(new Error(`Mutex timeout for key: ${key}`))
|
||||
}, timeout)
|
||||
|
||||
const resolver = () => {
|
||||
clearTimeout(timer)
|
||||
lock.locked = true
|
||||
resolve(() => this.release(key))
|
||||
}
|
||||
|
||||
lock.queue.push(resolver)
|
||||
})
|
||||
}
|
||||
|
||||
private release(key: string): void {
|
||||
const lock = this.locks.get(key)
|
||||
if (!lock) return
|
||||
|
||||
if (lock.queue.length > 0) {
|
||||
const next = lock.queue.shift()!
|
||||
next()
|
||||
} else {
|
||||
lock.locked = false
|
||||
// Clean up if no waiters
|
||||
if (lock.queue.length === 0) {
|
||||
this.locks.delete(key)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async runExclusive<T>(
|
||||
key: string,
|
||||
fn: () => Promise<T>,
|
||||
timeout?: number
|
||||
): Promise<T> {
|
||||
const release = await this.acquire(key, timeout)
|
||||
try {
|
||||
return await fn()
|
||||
} finally {
|
||||
release()
|
||||
}
|
||||
}
|
||||
|
||||
isLocked(key: string): boolean {
|
||||
return this.locks.get(key)?.locked || false
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* File-based mutex for multi-process scenarios (Node.js)
|
||||
* Uses atomic file operations to prevent TOCTOU races
|
||||
*/
|
||||
export class FileMutex implements MutexInterface {
|
||||
private fs: any
|
||||
private path: any
|
||||
private lockDir: string
|
||||
private processLocks: Map<string, () => void> = new Map()
|
||||
private lockTimers: Map<string, NodeJS.Timeout> = new Map()
|
||||
|
||||
constructor(lockDir: string) {
|
||||
this.lockDir = lockDir
|
||||
// Lazy load Node.js modules
|
||||
if (typeof window === 'undefined') {
|
||||
this.fs = require('fs')
|
||||
this.path = require('path')
|
||||
}
|
||||
}
|
||||
|
||||
async acquire(key: string, timeout: number = 30000): Promise<() => void> {
|
||||
if (!this.fs || !this.path) {
|
||||
throw new Error('FileMutex is only available in Node.js environments')
|
||||
}
|
||||
|
||||
const lockFile = this.path.join(this.lockDir, `${key}.lock`)
|
||||
const lockId = `${Date.now()}_${Math.random()}_${process.pid}`
|
||||
const startTime = Date.now()
|
||||
|
||||
// Ensure lock directory exists
|
||||
await this.fs.promises.mkdir(this.lockDir, { recursive: true })
|
||||
|
||||
while (Date.now() - startTime < timeout) {
|
||||
try {
|
||||
// Atomic lock creation using 'wx' flag
|
||||
await this.fs.promises.writeFile(
|
||||
lockFile,
|
||||
JSON.stringify({
|
||||
lockId,
|
||||
pid: process.pid,
|
||||
timestamp: Date.now(),
|
||||
expiresAt: Date.now() + timeout
|
||||
}),
|
||||
{ flag: 'wx' } // Write exclusive - fails if exists
|
||||
)
|
||||
|
||||
// Successfully acquired lock
|
||||
const release = () => this.release(key, lockFile, lockId)
|
||||
this.processLocks.set(key, release)
|
||||
|
||||
// Auto-release on timeout
|
||||
const timer = setTimeout(() => {
|
||||
release()
|
||||
}, timeout)
|
||||
this.lockTimers.set(key, timer)
|
||||
|
||||
return release
|
||||
} catch (error: any) {
|
||||
if (error.code === 'EEXIST') {
|
||||
// Lock exists - check if expired
|
||||
try {
|
||||
const data = await this.fs.promises.readFile(lockFile, 'utf-8')
|
||||
const lock = JSON.parse(data)
|
||||
|
||||
if (lock.expiresAt < Date.now()) {
|
||||
// Expired - try to remove
|
||||
try {
|
||||
await this.fs.promises.unlink(lockFile)
|
||||
continue // Retry acquisition
|
||||
} catch (unlinkError: any) {
|
||||
if (unlinkError.code !== 'ENOENT') {
|
||||
// Someone else removed it, continue
|
||||
continue
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
// Can't read lock file, assume it's valid
|
||||
}
|
||||
|
||||
// Wait before retry
|
||||
await new Promise(resolve => setTimeout(resolve, 50))
|
||||
} else {
|
||||
throw error
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
throw new Error(`Failed to acquire mutex for key: ${key} after ${timeout}ms`)
|
||||
}
|
||||
|
||||
private async release(key: string, lockFile: string, lockId: string): Promise<void> {
|
||||
// Clear timer
|
||||
const timer = this.lockTimers.get(key)
|
||||
if (timer) {
|
||||
clearTimeout(timer)
|
||||
this.lockTimers.delete(key)
|
||||
}
|
||||
|
||||
// Remove from process locks
|
||||
this.processLocks.delete(key)
|
||||
|
||||
try {
|
||||
// Verify we own the lock before releasing
|
||||
const data = await this.fs.promises.readFile(lockFile, 'utf-8')
|
||||
const lock = JSON.parse(data)
|
||||
|
||||
if (lock.lockId === lockId) {
|
||||
await this.fs.promises.unlink(lockFile)
|
||||
}
|
||||
} catch {
|
||||
// Lock already released or doesn't exist
|
||||
}
|
||||
}
|
||||
|
||||
async runExclusive<T>(
|
||||
key: string,
|
||||
fn: () => Promise<T>,
|
||||
timeout?: number
|
||||
): Promise<T> {
|
||||
const release = await this.acquire(key, timeout)
|
||||
try {
|
||||
return await fn()
|
||||
} finally {
|
||||
release()
|
||||
}
|
||||
}
|
||||
|
||||
isLocked(key: string): boolean {
|
||||
return this.processLocks.has(key)
|
||||
}
|
||||
|
||||
/**
|
||||
* Clean up all locks held by this process
|
||||
*/
|
||||
async cleanup(): Promise<void> {
|
||||
// Clear all timers
|
||||
for (const timer of this.lockTimers.values()) {
|
||||
clearTimeout(timer)
|
||||
}
|
||||
this.lockTimers.clear()
|
||||
|
||||
// Release all locks
|
||||
const releases = Array.from(this.processLocks.values())
|
||||
await Promise.all(releases.map(release => release()))
|
||||
this.processLocks.clear()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Factory to create appropriate mutex for the environment
|
||||
*/
|
||||
export function createMutex(options?: {
|
||||
type?: 'memory' | 'file'
|
||||
lockDir?: string
|
||||
}): MutexInterface {
|
||||
const type = options?.type || (typeof window === 'undefined' ? 'file' : 'memory')
|
||||
|
||||
if (type === 'file' && typeof window === 'undefined') {
|
||||
const lockDir = options?.lockDir || '.brainy/locks'
|
||||
return new FileMutex(lockDir)
|
||||
}
|
||||
|
||||
return new InMemoryMutex()
|
||||
}
|
||||
|
||||
// Global mutex instance for count operations
|
||||
let globalMutex: MutexInterface | null = null
|
||||
|
||||
export function getGlobalMutex(): MutexInterface {
|
||||
if (!globalMutex) {
|
||||
globalMutex = createMutex()
|
||||
}
|
||||
return globalMutex
|
||||
}
|
||||
|
||||
/**
|
||||
* Cleanup function for graceful shutdown
|
||||
*/
|
||||
export async function cleanupMutexes(): Promise<void> {
|
||||
if (globalMutex && 'cleanup' in globalMutex) {
|
||||
await (globalMutex as any).cleanup()
|
||||
}
|
||||
}
|
||||
|
|
@ -165,12 +165,29 @@ export function validateFindParams(params: FindParams): void {
|
|||
export function validateAddParams(params: AddParams): void {
|
||||
// Universal truth: must have data or vector
|
||||
if (!params.data && !params.vector) {
|
||||
throw new Error('must provide either data or vector')
|
||||
throw new Error(
|
||||
`Invalid add() parameters: Missing required field 'data'\n` +
|
||||
`\nReceived: ${JSON.stringify({
|
||||
type: params.type,
|
||||
hasMetadata: !!params.metadata,
|
||||
hasId: !!params.id
|
||||
}, null, 2)}\n` +
|
||||
`\nExpected one of:\n` +
|
||||
` { data: 'text to store', type?: 'note', metadata?: {...} }\n` +
|
||||
` { vector: [0.1, 0.2, ...], type?: 'embedding', metadata?: {...} }\n` +
|
||||
`\nExamples:\n` +
|
||||
` await brain.add({ data: 'Machine learning is AI', type: 'concept' })\n` +
|
||||
` await brain.add({ data: { title: 'Doc', content: '...' }, type: 'document' })`
|
||||
)
|
||||
}
|
||||
|
||||
// Validate noun type
|
||||
if (!Object.values(NounType).includes(params.type)) {
|
||||
throw new Error(`invalid NounType: ${params.type}`)
|
||||
throw new Error(
|
||||
`Invalid NounType: '${params.type}'\n` +
|
||||
`\nValid types: ${Object.values(NounType).join(', ')}\n` +
|
||||
`\nExample: await brain.add({ data: 'text', type: NounType.Note })`
|
||||
)
|
||||
}
|
||||
|
||||
// Validate vector dimensions if provided
|
||||
|
|
@ -227,8 +244,10 @@ export function validateRelateParams(params: RelateParams): void {
|
|||
throw new Error('cannot create self-referential relationship')
|
||||
}
|
||||
|
||||
// Validate verb type
|
||||
if (!Object.values(VerbType).includes(params.type)) {
|
||||
// Validate verb type - default to RelatedTo if not specified
|
||||
if (params.type === undefined) {
|
||||
params.type = VerbType.RelatedTo
|
||||
} else if (!Object.values(VerbType).includes(params.type)) {
|
||||
throw new Error(`invalid VerbType: ${params.type}`)
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue