/** * Storage-based Discovery for Zero-Config Distributed Brainy * Uses shared storage (S3/GCS/R2) as coordination point * REAL PRODUCTION CODE - No mocks, no stubs! */ import { EventEmitter } from 'node:events' import * as os from 'node:os' import { StorageAdapter } from '../coreTypes.js' export interface NodeInfo { id: string endpoint: string hostname: string started: number lastSeen: number role: 'primary' | 'replica' | 'candidate' shards: string[] capacity: { cpu: number memory: number storage: number } stats: { nouns: number verbs: number queries: number latency: number } } export interface ClusterConfig { version: number created: number updated: number leader: string | null nodes: Record shards: { count: number assignments: Record // shardId -> [primaryNode, ...replicas] } settings: { replicationFactor: number shardCount: number autoRebalance: boolean minNodes: number maxNodesPerShard: number } } export class StorageDiscovery extends EventEmitter { private nodeId: string private storage: StorageAdapter private nodeInfo: NodeInfo private clusterConfig: ClusterConfig | null = null private heartbeatInterval: NodeJS.Timeout | null = null private discoveryInterval: NodeJS.Timeout | null = null private endpoint: string = '' private isRunning: boolean = false private readonly HEARTBEAT_INTERVAL = 5000 // 5 seconds private readonly DISCOVERY_INTERVAL = 2000 // 2 seconds private readonly NODE_TIMEOUT = 30000 // 30 seconds until node considered dead private readonly CLUSTER_PATH = '_cluster' constructor(storage: StorageAdapter, nodeId?: string) { super() this.storage = storage this.nodeId = nodeId || this.generateNodeId() // Initialize node info with REAL system data this.nodeInfo = { id: this.nodeId, endpoint: '', // Will be set when HTTP server starts hostname: os.hostname(), started: Date.now(), lastSeen: Date.now(), role: 'candidate', shards: [], capacity: { cpu: os.cpus().length, memory: Math.floor(os.totalmem() / 1024 / 1024), // MB storage: 0 // Will be updated based on actual usage }, stats: { nouns: 0, verbs: 0, queries: 0, latency: 0 } } } /** * Start discovery and registration */ async start(httpPort: number): Promise { if (this.isRunning) return this.clusterConfig! this.isRunning = true // Set our endpoint this.endpoint = await this.detectEndpoint(httpPort) this.nodeInfo.endpoint = this.endpoint // Try to load existing cluster config this.clusterConfig = await this.loadClusterConfig() if (!this.clusterConfig) { // We're the first node - initialize cluster await this.initializeCluster() } else { // Join existing cluster await this.joinCluster() } // Start heartbeat to keep our node alive this.startHeartbeat() // Start discovery to find other nodes this.startDiscovery() this.emit('started', this.nodeInfo) return this.clusterConfig! } /** * Stop discovery and unregister */ async stop(): Promise { if (!this.isRunning) return this.isRunning = false // Stop intervals if (this.heartbeatInterval) { clearInterval(this.heartbeatInterval) this.heartbeatInterval = null } if (this.discoveryInterval) { clearInterval(this.discoveryInterval) this.discoveryInterval = null } // Remove ourselves from cluster await this.leaveCluster() this.emit('stopped') } /** * Initialize a new cluster (we're the first node) */ private async initializeCluster(): Promise { console.log(`[${this.nodeId}] Initializing new cluster as first node`) this.nodeInfo.role = 'primary' this.clusterConfig = { version: 1, created: Date.now(), updated: Date.now(), leader: this.nodeId, nodes: { [this.nodeId]: this.nodeInfo }, shards: { count: 64, // Default shard count assignments: {} }, settings: { replicationFactor: 3, shardCount: 64, autoRebalance: true, minNodes: 1, maxNodesPerShard: 5 } } // Assign all shards to ourselves initially for (let i = 0; i < this.clusterConfig.shards.count; i++) { const shardId = `shard-${i.toString().padStart(3, '0')}` this.clusterConfig.shards.assignments[shardId] = [this.nodeId] this.nodeInfo.shards.push(shardId) } // Save cluster config await this.saveClusterConfig() // Register ourselves await this.registerNode() this.emit('clusterInitialized', this.clusterConfig) } /** * Join an existing cluster */ private async joinCluster(): Promise { console.log(`[${this.nodeId}] Joining existing cluster`) if (!this.clusterConfig) throw new Error('No cluster config') // Add ourselves to the cluster this.clusterConfig.nodes[this.nodeId] = this.nodeInfo // Determine our role based on cluster state const nodeCount = Object.keys(this.clusterConfig.nodes).length if (!this.clusterConfig.leader || !this.clusterConfig.nodes[this.clusterConfig.leader]) { // No leader or leader is gone - trigger election await this.triggerLeaderElection() } else { // Become replica this.nodeInfo.role = 'replica' } // Register ourselves await this.registerNode() // Request shard assignment if auto-rebalance is enabled if (this.clusterConfig.settings.autoRebalance) { await this.requestShardAssignment() } this.emit('clusterJoined', this.clusterConfig) } /** * Leave cluster cleanly */ private async leaveCluster(): Promise { if (!this.clusterConfig) return console.log(`[${this.nodeId}] Leaving cluster`) // Remove ourselves from node registry try { // Mark as deleted rather than actually deleting const deadNode = { noun: 'Document', ...this.nodeInfo, lastSeen: 0, status: 'inactive' as const } await this.storage.saveMetadata(`${this.CLUSTER_PATH}/nodes/${this.nodeId}.json`, deadNode) } catch (err) { // Ignore errors during shutdown } // If we're the leader, trigger new election if (this.clusterConfig.leader === this.nodeId) { this.clusterConfig.leader = null await this.saveClusterConfig() } this.emit('clusterLeft') } /** * Register node in storage */ private async registerNode(): Promise { const path = `${this.CLUSTER_PATH}/nodes/${this.nodeId}.json` await this.storage.saveMetadata(path, { noun: 'Document', ...this.nodeInfo }) // Also update registry await this.updateNodeRegistry(this.nodeId) } /** * Heartbeat to keep node alive */ private startHeartbeat(): void { this.heartbeatInterval = setInterval(async () => { try { this.nodeInfo.lastSeen = Date.now() await this.registerNode() // Also update cluster config if we're the leader if (this.clusterConfig && this.clusterConfig.leader === this.nodeId) { await this.saveClusterConfig() } } catch (err) { console.error(`[${this.nodeId}] Heartbeat failed:`, err) } }, this.HEARTBEAT_INTERVAL) } /** * Discover other nodes and monitor health */ private startDiscovery(): void { this.discoveryInterval = setInterval(async () => { try { await this.discoverNodes() await this.checkNodeHealth() // Check if we need to rebalance if (this.shouldRebalance()) { await this.triggerRebalance() } } catch (err) { console.error(`[${this.nodeId}] Discovery failed:`, err) } }, this.DISCOVERY_INTERVAL) } /** * Discover nodes from storage */ private async discoverNodes(): Promise { try { // Since we can't list arbitrary paths, we'll use a registry approach // Each node registers in a central registry file const registry = await this.loadNodeRegistry() const now = Date.now() let updated = false for (const nodeId of registry) { if (nodeId === this.nodeId) continue try { const nodeInfoData = await this.storage.getMetadata( `${this.CLUSTER_PATH}/nodes/${nodeId}.json` ) const nodeInfo = nodeInfoData as unknown as NodeInfo // Check if node is alive if (now - nodeInfo.lastSeen < this.NODE_TIMEOUT) { if (!this.clusterConfig!.nodes[nodeId]) { // New node discovered! console.log(`[${this.nodeId}] Discovered new node: ${nodeId}`) this.clusterConfig!.nodes[nodeId] = nodeInfo updated = true this.emit('nodeDiscovered', nodeInfo) } else { // Update existing node info this.clusterConfig!.nodes[nodeId] = nodeInfo } } } catch (err) { // Node file might be corrupted or deleted console.warn(`[${this.nodeId}] Failed to read node ${nodeId}:`, err) } } if (updated) { this.clusterConfig!.version++ this.clusterConfig!.updated = Date.now() } } catch (err) { // Storage might be unavailable console.error(`[${this.nodeId}] Failed to discover nodes:`, err) } } /** * Load node registry from storage */ private async loadNodeRegistry(): Promise { try { const registry = await this.storage.getMetadata(`${this.CLUSTER_PATH}/registry.json`) as any return registry?.nodes || [] } catch (err) { return [] } } /** * Update node registry in storage */ private async updateNodeRegistry(add?: string, remove?: string): Promise { try { let registry = await this.loadNodeRegistry() if (add && !registry.includes(add)) { registry.push(add) } if (remove) { registry = registry.filter(id => id !== remove) } await this.storage.saveMetadata(`${this.CLUSTER_PATH}/registry.json`, { noun: 'Document', nodes: registry, updated: Date.now() }) } catch (err) { console.error(`[${this.nodeId}] Failed to update registry:`, err) } } /** * Check health of known nodes */ private async checkNodeHealth(): Promise { if (!this.clusterConfig) return const now = Date.now() const deadNodes: string[] = [] for (const [nodeId, nodeInfo] of Object.entries(this.clusterConfig.nodes)) { if (nodeId === this.nodeId) continue if (now - nodeInfo.lastSeen > this.NODE_TIMEOUT) { console.log(`[${this.nodeId}] Node ${nodeId} is dead (last seen ${now - nodeInfo.lastSeen}ms ago)`) deadNodes.push(nodeId) } } // Remove dead nodes for (const nodeId of deadNodes) { delete this.clusterConfig.nodes[nodeId] this.emit('nodeLost', nodeId) // If dead node was leader, trigger election if (this.clusterConfig.leader === nodeId) { await this.triggerLeaderElection() } } if (deadNodes.length > 0) { // Trigger rebalance to reassign shards from dead nodes await this.triggerRebalance() } } /** * Load cluster configuration from storage */ private async loadClusterConfig(): Promise { try { const config = await this.storage.getMetadata(`${this.CLUSTER_PATH}/config.json`) return config as unknown as ClusterConfig } catch (err) { // No cluster config exists yet return null } } /** * Save cluster configuration to storage */ private async saveClusterConfig(): Promise { if (!this.clusterConfig) return await this.storage.saveMetadata( `${this.CLUSTER_PATH}/config.json`, { noun: 'Document', ...this.clusterConfig } ) } /** * Trigger leader election (simplified - not full Raft) */ private async triggerLeaderElection(): Promise { console.log(`[${this.nodeId}] Triggering leader election`) // Simple election: node with lowest ID wins // In production, use proper Raft consensus const activeNodes = Object.entries(this.clusterConfig!.nodes) .filter(([_, info]) => Date.now() - info.lastSeen < this.NODE_TIMEOUT) .sort(([a], [b]) => a.localeCompare(b)) if (activeNodes.length > 0) { const [leaderId, leaderInfo] = activeNodes[0] this.clusterConfig!.leader = leaderId if (leaderId === this.nodeId) { console.log(`[${this.nodeId}] Became leader`) this.nodeInfo.role = 'primary' this.emit('becameLeader') } else { console.log(`[${this.nodeId}] Node ${leaderId} is the new leader`) this.nodeInfo.role = 'replica' this.emit('leaderElected', leaderId) } await this.saveClusterConfig() } } /** * Request shard assignment for this node */ private async requestShardAssignment(): Promise { if (!this.clusterConfig) return // Calculate how many shards each node should have const nodeCount = Object.keys(this.clusterConfig.nodes).length const shardsPerNode = Math.ceil(this.clusterConfig.shards.count / nodeCount) // Find shards that need assignment const unassignedShards: string[] = [] for (let i = 0; i < this.clusterConfig.shards.count; i++) { const shardId = `shard-${i.toString().padStart(3, '0')}` if (!this.clusterConfig.shards.assignments[shardId] || this.clusterConfig.shards.assignments[shardId].length === 0) { unassignedShards.push(shardId) } } // Assign some shards to ourselves const ourShare = unassignedShards.slice(0, shardsPerNode) for (const shardId of ourShare) { this.clusterConfig.shards.assignments[shardId] = [this.nodeId] this.nodeInfo.shards.push(shardId) } if (ourShare.length > 0) { console.log(`[${this.nodeId}] Assigned ${ourShare.length} shards`) await this.saveClusterConfig() } } /** * Check if rebalancing is needed */ private shouldRebalance(): boolean { if (!this.clusterConfig || !this.clusterConfig.settings.autoRebalance) { return false } // Check if shards are evenly distributed const nodeCount = Object.keys(this.clusterConfig.nodes).length if (nodeCount <= 1) return false const targetShardsPerNode = Math.ceil(this.clusterConfig.shards.count / nodeCount) const variance = 2 // Allow some variance for (const nodeInfo of Object.values(this.clusterConfig.nodes)) { const shardCount = nodeInfo.shards.length if (Math.abs(shardCount - targetShardsPerNode) > variance) { return true } } return false } /** * Trigger shard rebalancing */ private async triggerRebalance(): Promise { // Only leader can trigger rebalance if (this.clusterConfig?.leader !== this.nodeId) return console.log(`[${this.nodeId}] Triggering shard rebalance`) // This will be implemented with actual data migration // For now, just redistribute shard assignments await this.redistributeShards() this.emit('rebalanceTriggered') } /** * Redistribute shards among active nodes */ private async redistributeShards(): Promise { if (!this.clusterConfig) return const activeNodes = Object.keys(this.clusterConfig.nodes) .filter(id => Date.now() - this.clusterConfig!.nodes[id].lastSeen < this.NODE_TIMEOUT) if (activeNodes.length === 0) return const shardsPerNode = Math.ceil(this.clusterConfig.shards.count / activeNodes.length) const newAssignments: Record = {} // Clear current shard assignments from nodes for (const nodeInfo of Object.values(this.clusterConfig.nodes)) { nodeInfo.shards = [] } // Redistribute shards let nodeIndex = 0 for (let i = 0; i < this.clusterConfig.shards.count; i++) { const shardId = `shard-${i.toString().padStart(3, '0')}` const primaryNode = activeNodes[nodeIndex % activeNodes.length] // Assign primary newAssignments[shardId] = [primaryNode] this.clusterConfig.nodes[primaryNode].shards.push(shardId) // Assign replicas const replicas: string[] = [] for (let r = 1; r < Math.min(this.clusterConfig.settings.replicationFactor, activeNodes.length); r++) { const replicaNode = activeNodes[(nodeIndex + r) % activeNodes.length] if (replicaNode !== primaryNode) { replicas.push(replicaNode) } } if (replicas.length > 0) { newAssignments[shardId].push(...replicas) } nodeIndex++ } this.clusterConfig.shards.assignments = newAssignments this.clusterConfig.version++ this.clusterConfig.updated = Date.now() await this.saveClusterConfig() console.log(`[${this.nodeId}] Rebalanced ${this.clusterConfig.shards.count} shards across ${activeNodes.length} nodes`) } /** * Detect our public endpoint */ private async detectEndpoint(port: number): Promise { // Try to detect public IP const interfaces = os.networkInterfaces() let ip = '127.0.0.1' // Find first non-internal IPv4 address for (const iface of Object.values(interfaces)) { if (!iface) continue for (const addr of iface) { if (addr.family === 'IPv4' && !addr.internal) { ip = addr.address break } } } // In cloud environments, might need to detect public IP differently if (process.env.PUBLIC_IP) { ip = process.env.PUBLIC_IP } else if (process.env.KUBERNETES_SERVICE_HOST) { // In Kubernetes, use pod IP ip = process.env.POD_IP || ip } return `http://${ip}:${port}` } /** * Generate unique node ID */ private generateNodeId(): string { const hostname = os.hostname() const pid = process.pid const random = Math.random().toString(36).substring(2, 8) return `${hostname}-${pid}-${random}` } /** * Get current cluster configuration */ getClusterConfig(): ClusterConfig | null { return this.clusterConfig } /** * Get active nodes */ getActiveNodes(): NodeInfo[] { if (!this.clusterConfig) return [] const now = Date.now() return Object.values(this.clusterConfig.nodes) .filter(node => now - node.lastSeen < this.NODE_TIMEOUT) } /** * Get shards assigned to this node */ getMyShards(): string[] { return this.nodeInfo.shards } /** * Update node statistics */ updateStats(stats: Partial): void { Object.assign(this.nodeInfo.stats, stats) } }