From 9b4a802ff34b1cc8c45957f7185f853037eb82ec Mon Sep 17 00:00:00 2001 From: Pierre Date: Sun, 1 Feb 2026 13:39:36 +0100 Subject: [PATCH] leader discovery fix and code cleaning --- docs/vfs-rebuild-architecture.md | 527 ++++++++++++++++++ packages/meta-fuse-core/src/api/APIServer.ts | 9 +- packages/meta-fuse-core/src/index.ts | 4 - packages/meta-fuse-core/src/kv/IKVClient.ts | 49 +- packages/meta-fuse-core/src/kv/KVManager.ts | 118 ++-- .../meta-fuse-core/src/kv/LeaderClient.ts | 223 ++++++++ .../meta-fuse-core/src/kv/LeaderDiscovery.ts | 276 --------- .../meta-fuse-core/src/kv/ServiceDiscovery.ts | 137 +++-- 8 files changed, 928 insertions(+), 415 deletions(-) create mode 100644 docs/vfs-rebuild-architecture.md create mode 100644 packages/meta-fuse-core/src/kv/LeaderClient.ts delete mode 100644 packages/meta-fuse-core/src/kv/LeaderDiscovery.ts diff --git a/docs/vfs-rebuild-architecture.md b/docs/vfs-rebuild-architecture.md new file mode 100644 index 0000000..54c3da6 --- /dev/null +++ b/docs/vfs-rebuild-architecture.md @@ -0,0 +1,527 @@ +# VFS Rebuild Architecture + +## Overview + +This document describes the architecture and design philosophy for the Virtual Filesystem (VFS) rebuild process in MetaMesh. The system uses a **two-phase strategy** to provide fast startup times while maintaining data accuracy through continuous validation. + +## Design Philosophy + +The VFS rebuild architecture is built on three core principles: + +| Component | Role | Responsibility | +|-----------|------|----------------| +| **etcd** | Distributed Metadata Archive | Permanent record of all files (even if deleted from disk) | +| **In-Memory Database** | Current VFS State | Only files present on disk RIGHT NOW | +| **Discovery** | Source of Truth | Validates file existence and cleans stale entries | + +### Key Insight + +> **etcd stores the historical truth** (what files have existed) +> **In-memory DB stores the current truth** (what files exist NOW) +> **Discovery enforces reality** (makes memory match disk state) + +This separation enables: +- **Fast cold starts** (1-2 seconds for 10,000 files) +- **Distributed metadata sharing** via meta-orbit (etcd never deletes) +- **Accurate VFS state** (discovery continuously validates) +- **Graceful handling of offline storage** (network drives, removable media) + +--- + +## Two-Phase Strategy + +### Phase 1: Fast Bootstrap (Startup) + +**Goal**: Make VFS available as quickly as possible + +**Implementation**: `WatchedFileProcessor.rebuildVFSFromEtcd()` + +```typescript +async rebuildVFSFromEtcd(): Promise { + // 1. Fetch all hash IDs from etcd + const hashIds = await etcdClient.getAllHashIds(); + + // 2. Batch fetch ALL metadata in parallel (no disk I/O) + const metadataList = await Promise.all( + hashIds.map(hashId => etcdClient.getMetadataFlat(hashId)) + ); + + // 3. Load into in-memory DB without validation + for (const metadata of metadataList) { + metadata._lastVerified = 0; // Mark as unverified + this.fileProcessor.getDatabase().set(metadata.filePath, metadata); + } + + // 4. Build VFS immediately (1-2 seconds) + await this.finalize(); +} +``` + +**Performance**: +- 10,000 files: ~1-2 seconds (previously 60+ seconds) +- Zero disk I/O during startup +- VFS immediately usable (may contain stale entries) + +**Characteristics**: +- ✅ Fast startup +- ✅ No blocking on disk validation +- ✅ All files appear immediately +- ⚠️ May include deleted/offline files temporarily + +--- + +### Phase 2: Continuous Validation (Discovery) + +**Goal**: Make in-memory DB match current disk reality + +**Implementation**: Discovery scans folders progressively and validates file existence + +```typescript +// Progressive scan: Process folders one at a time +for (const folder of folderList) { + const discoveredFilesInFolder = new Set(); + + // Track discovered files for this folder + async function* trackDiscoveredFiles(stream: AsyncGenerator) { + for await (const filePath of stream) { + discoveredFilesInFolder.add(filePath); // Track what exists + yield filePath; + } + } + + const discoveryStream = folderWatcher.discoverFiles([folder]); + await pipeline.start(trackDiscoveredFiles(discoveryStream)); + + // Progressive cleanup: Remove stale entries after each folder + await fileProcessor.cleanupStaleEntries([folder], discoveredFilesInFolder); +} +``` + +**Cleanup Logic**: `WatchedFileProcessor.cleanupStaleEntries()` + +```typescript +async cleanupStaleEntries(scannedFolders: string[], discoveredFiles: Set) { + for (const folder of scannedFolders) { + // Find all in-memory files for this folder + const filesToCheck = db.getAllFilesStartingWith(folder); + + // Remove files not seen during scan + for (const filePath of filesToCheck) { + if (!discoveredFiles.has(filePath)) { + // File no longer exists → remove from memory + db.delete(filePath); + vfs.removeFileByRealPath(filePath); + stateManager.removeFile(filePath); + } + } + } + + // Rebuild VFS if anything changed + await this.finalize(); +} +``` + +**Validation Rules** (applied progressively per folder): + +| Scenario | Action | VFS Update | +|----------|--------|------------| +| File found during scan | Update `_lastVerified` timestamp, keep in memory | No change | +| File not found in scanned folder | Remove from in-memory DB, keep in etcd | VFS rebuilt incrementally | +| Empty folder scanned | Remove ALL child paths from memory immediately | VFS rebuilt (files disappear) | +| Folder missing/inaccessible | Remove ALL child paths from memory immediately | VFS rebuilt (files disappear) | + +**Key Behavior**: When a NAS/network drive is disconnected (folder becomes empty/inaccessible), all files from that folder are **progressively removed from the VFS as discovery proceeds** through each folder. + +**Characteristics**: +- ✅ Discovery is authoritative +- ✅ Progressive cleanup (files disappear as folders are scanned) +- ✅ Handles deleted files automatically +- ✅ Handles offline network drives (files removed immediately when folder is scanned) +- ✅ No manual TTL/expiry needed +- ✅ Real-time VFS updates during discovery + +--- + +## Architecture Comparison + +### Old Approach (Disk Validation at Startup) + +``` +┌─────────────────────────────────────────────┐ +│ Startup: rebuildVFSFromEtcd() │ +│ │ +│ For each file in etcd (sequential): │ +│ 1. Fetch metadata from etcd (1-5ms) │ +│ 2. Check if file exists (fs.access) │ +│ - Local disk: 1-10ms │ +│ - Network drive: 10-100ms │ +│ 3. Add to memory if exists │ +│ │ +│ Result: 60-120 seconds for 10k files │ +└─────────────────────────────────────────────┘ +``` + +**Problems**: +- ❌ Slow startup (blocks on disk I/O) +- ❌ Sequential processing (no parallelization) +- ❌ Network latency multiplies wait time +- ❌ VFS unavailable during validation + +### New Approach (Fast Load + Progressive Discovery) + +``` +┌─────────────────────────────────────────────┐ +│ Startup: rebuildVFSFromEtcd() │ +│ │ +│ 1. Batch fetch ALL metadata (parallel) │ +│ → 10k files in ~500ms │ +│ │ +│ 2. Load into memory (no disk checks) │ +│ → 10k files in ~200ms │ +│ │ +│ 3. Build VFS │ +│ → Ready in 1-2 seconds total │ +│ │ +│ Result: VFS immediately usable │ +└─────────────────────────────────────────────┘ + +┌─────────────────────────────────────────────┐ +│ Background: Progressive Discovery Scan │ +│ │ +│ For each folder in watch list: │ +│ 1. Walk folder and track discovered files│ +│ 2. Update _lastVerified for found files │ +│ 3. Remove stale entries for this folder │ +│ 4. Rebuild VFS (files disappear) │ +│ 5. Move to next folder │ +│ │ +│ Result: Files progressively disappear as │ +│ folders are scanned (real-time) │ +└─────────────────────────────────────────────┘ +``` + +**Benefits**: +- ✅ 10-50x faster startup +- ✅ Parallel etcd fetching +- ✅ No disk I/O blocking +- ✅ VFS immediately available +- ✅ Progressive cleanup (real-time feedback) +- ✅ Files disappear as folders are scanned + +--- + +## Implementation Details + +### File Metadata Structure + +```typescript +interface FileMetadata { + filePath: string; + cid_midhash256: string; // Permanent file ID + title: string; + // ... other metadata fields + + // Validation tracking (optional) + _lastVerified?: number; // Timestamp of last disk validation +} +``` + +The `_lastVerified` field: +- Set to `0` during fast load (unverified) +- Updated to `Date.now()` when discovery sees the file +- Used for debugging and monitoring (not enforced) + +### State Management + +**In-Memory Database** (`FileProcessor.database`): +- `Map` +- Cleared on restart +- Reflects only currently accessible files + +**etcd Storage**: +- `/file/midhash256:{hash}/property` - Permanent metadata +- Never deleted (even if file removed from disk) +- Shared across meta-orbit network + +**Virtual Filesystem** (`VirtualFileSystem`): +- Generated from in-memory database +- Rebuilt after cleanup operations +- Accessible via FUSE/WebDAV + +### Discovery Integration + +Discovery runs in two modes: + +**1. Manual Scan** (startup or triggered via API) - Progressive: +```typescript +// Process folders one at a time +for (const folder of folderList) { + const discoveredFiles = new Set(); + const discoveryStream = folderWatcher.discoverFiles([folder]); + + // Track and process files for this folder + await pipeline.start(trackFiles(discoveryStream, discoveredFiles)); + + // Progressive cleanup: Remove stale entries immediately + await fileProcessor.cleanupStaleEntries([folder], discoveredFiles); + // → VFS rebuilt, files from disconnected NAS disappear +} +``` + +**2. Watch Mode** (continuous monitoring): +```typescript +folderWatcher.watch(folderList, { + onAdd: (filePath) => { + // File added → process and add to memory + pipeline.processFile(filePath); + }, + onUnlink: (filePath) => { + // File removed → remove from memory immediately + fileProcessor.deleteFile(filePath); + } +}); +``` + +--- + +## Performance Characteristics + +### Startup Performance + +| Dataset Size | Old Approach | New Approach | Speedup | +|--------------|-------------|--------------|---------| +| 1,000 files | 5-10s | 0.5-1s | 10x | +| 10,000 files | 60-120s | 1-2s | 60x | +| 100,000 files | 10-20m | 10-15s | 60-80x | + +### Storage Type Impact + +| Storage Type | Old Approach | New Approach | Notes | +|--------------|-------------|--------------|-------| +| Local SSD | 5-10s | 1-2s | Validation was fast, but sequential | +| Local HDD | 20-40s | 1-2s | Seek time penalty eliminated | +| Network (NFS/SMB) | 60-300s | 1-2s | Latency eliminated from startup | +| Offline/Slow Network | Timeout/Fail | 1-2s | Gracefully handles offline storage | + +### Memory Usage + +- **In-Memory DB**: ~500 bytes/file (metadata only) +- **10,000 files**: ~5 MB +- **100,000 files**: ~50 MB +- **Negligible overhead** compared to old approach + +--- + +## Handling Edge Cases + +### Deleted Files + +**Scenario**: User deletes a file after startup but before discovery scans + +**Old Behavior**: File appears in VFS until next full rebuild + +**New Behavior**: +1. Fast load: File appears in VFS (loaded from etcd) +2. Discovery scan: File not found → removed from memory +3. VFS rebuilt without the file +4. etcd still has metadata (for meta-orbit sharing) + +### Offline Network Drives + +**Scenario**: NFS mount becomes unavailable + +**Old Behavior**: Startup hangs or fails with timeout + +**New Behavior**: +1. Fast load: All files appear in VFS (1-2s) +2. Discovery scan: Folder inaccessible → removes all child paths +3. When drive comes back online: Next discovery re-adds files + +### Moved Files + +**Scenario**: User moves file outside watch folders + +**Old Behavior**: File appears in VFS until manual cleanup + +**New Behavior**: +1. Discovery scans old location → file not found → removed from memory +2. If moved within watch folders: Discovered as new file +3. etcd has both old and new paths (deduplication via midhash256) + +### Partial Scans + +**Scenario**: Discovery interrupted mid-scan (crash, kill signal) + +**Safety**: Cleanup only runs after scan completes successfully +- Incomplete scan → no cleanup +- Memory state unchanged +- Next scan will complete and clean up + +--- + +## Monitoring and Debugging + +### Startup Logs + +``` +[VFS Rebuild] Starting fast-load from etcd... +[VFS Rebuild] Found 10000 files in etcd +[VFS Rebuild] Fetching metadata in parallel... +[VFS Rebuild] Loaded 9856/10000 files into memory (unverified) +[VFS Rebuild] VFS rebuilt in 1523ms +[VFS Rebuild] Discovery will validate file existence in background +``` + +### Discovery Logs (Progressive Mode) + +``` +[Discovery] Progressive scan mode: 3 folders +[Discovery] Scanning folder: /data/watch/local +[Discovery] Folder scan complete: /data/watch/local. Found 128 files. Running cleanup... +[Cleanup] Starting cleanup for 1 scanned folders... +[Cleanup] Removed 0 stale entries in 2ms +[Cleanup] Cleanup complete for folder: /data/watch/local + +[Discovery] Scanning folder: /data/watch/smb-nas +[Discovery] Folder scan complete: /data/watch/smb-nas. Found 0 files. Running cleanup... +[Cleanup] Starting cleanup for 1 scanned folders... +[Cleanup] Removed stale entry: /data/watch/smb-nas/Movies/deleted_movie.mkv +[Cleanup] Removed stale entry: /data/watch/smb-nas/Movies/another_movie.mkv +[Cleanup] Removed 13621 stale entries in 87ms +[Cleanup] Rebuilding VFS after cleanup... +Virtual filesystem update start +Duplicate find took 1ms - found 0 hash groups, 21 title groups +Generating virtual structure for 128 files +Virtual filesystem update took 167ms +[Cleanup] Cleanup complete for folder: /data/watch/smb-nas + +[Discovery] Scanning folder: /data/watch/downloads +[Discovery] Folder scan complete: /data/watch/downloads. Found 45 files. Running cleanup... +[Cleanup] Starting cleanup for 1 scanned folders... +[Cleanup] Removed 0 stale entries in 1ms +[Cleanup] Cleanup complete for folder: /data/watch/downloads + +[Discovery] All folders scanned and cleaned up +``` + +**Progressive Behavior**: Notice how the NAS folder (`/data/watch/smb-nas`) with 0 discovered files immediately triggers removal of 13,621 stale entries and VFS rebuild. Users see files disappear in real-time as each folder completes scanning. + +### API Endpoints + +Monitor VFS state via Unified API: + +```bash +# VFS statistics +curl http://localhost:3000/api/fuse/stats + +# Processing status +curl http://localhost:3000/api/processing/status + +# Trigger manual scan + cleanup +curl -X POST http://localhost:3000/api/scan/trigger +``` + +--- + +## Best Practices + +### For Operators + +1. **Initial Startup**: First boot may show all historical files (even deleted ones) + - Wait for first discovery scan to clean up + - Or run manual scan: `POST /api/scan/trigger` + +2. **Network Drives**: If drives are slow or offline at startup + - VFS still loads in 1-2 seconds + - Files appear/disappear as drives come online/offline + - No manual intervention needed + +3. **Monitoring**: Watch for cleanup logs to understand disk changes + ``` + [Cleanup] Removed 622 stale entries + ``` + Large numbers indicate significant file deletions + +### For Developers + +1. **Adding New Metadata Sources**: Always update both: + - etcd (permanent record) + - In-memory DB (current state) + +2. **Testing Cleanup**: Simulate file deletions + ```bash + # Delete files from watch folder + rm -rf /data/watch/Movies/test_movie.mkv + + # Trigger scan + curl -X POST http://localhost:3000/api/scan/trigger + + # Verify cleanup in logs + docker logs meta-mesh-dev-container | grep Cleanup + ``` + +3. **Performance Testing**: Benchmark with large datasets + ```bash + # Time startup + docker restart meta-mesh-dev-container + docker logs -f meta-mesh-dev-container | grep "VFS rebuilt" + + # Time discovery + cleanup + time curl -X POST http://localhost:3000/api/scan/trigger + ``` + +--- + +## Future Optimizations + +### Potential Improvements + +1. **Incremental Cleanup**: Instead of scanning all folders, track modified folders + - Use filesystem events (inotify) to detect folder changes + - Only clean up modified folders + +2. **Background Validation Queue**: Validate files lazily over time + - Prioritize recently accessed files + - Validate cold files during idle periods + +3. **Smart Caching**: Cache validation results + - LRU cache with TTL (5-10 minutes) + - Reduces redundant disk checks for frequently accessed files + +4. **Partial VFS Rebuild**: Only rebuild changed directories + - Track which folders had files removed + - Rebuild only affected VFS subtrees + +### Not Recommended + +❌ **TTL-based expiry**: Requires guessing timeout values, fails with slow scans + +❌ **Storing validation state in etcd**: Wrong tool for transient state, creates write amplification + +❌ **Immediate removal on unlink events**: Race conditions with discovery, inconsistent state + +--- + +## Related Documentation + +- [Streaming Pipeline Architecture](streaming-pipeline-architecture.md) - File processing flow +- [etcd Architecture](../../packages/meta-mesh/doc/architecture/etcd-architecture.md) - Storage design +- [FUSE API](../../packages/meta-mesh/doc/architecture/fuse-api.md) - Virtual filesystem API + +--- + +## Summary + +The VFS rebuild architecture achieves **10-50x faster startup** by separating concerns: + +- **etcd** = Permanent distributed metadata archive +- **In-memory DB** = Current filesystem state +- **Discovery** = Authoritative source of truth + +This design enables: +- ⚡ Sub-2-second cold starts (10k files) +- 🔄 Continuous accuracy through discovery validation +- 🌐 Distributed metadata sharing via meta-orbit +- 📦 Graceful handling of offline/slow storage + +The key insight: **Fast bootstrap + continuous validation > Blocking validation at startup** diff --git a/packages/meta-fuse-core/src/api/APIServer.ts b/packages/meta-fuse-core/src/api/APIServer.ts index 32c9d17..14aa6a2 100644 --- a/packages/meta-fuse-core/src/api/APIServer.ts +++ b/packages/meta-fuse-core/src/api/APIServer.ts @@ -303,9 +303,11 @@ export class APIServer { const info = this.kvManager.getLeaderInfo(); if (info) { leaderInfo = { - host: info.host, - api: info.api, - http: info.http, + hostname: info.hostname, + baseUrl: info.baseUrl, + apiUrl: info.apiUrl, + redisUrl: info.redisUrl, + webdavUrl: info.webdavUrl, }; } } @@ -314,7 +316,6 @@ export class APIServer { services, current: 'meta-fuse', leader: leaderInfo, - isLeader: false, // meta-fuse is never the leader }); } diff --git a/packages/meta-fuse-core/src/index.ts b/packages/meta-fuse-core/src/index.ts index 20c7525..8b85a40 100644 --- a/packages/meta-fuse-core/src/index.ts +++ b/packages/meta-fuse-core/src/index.ts @@ -73,12 +73,8 @@ async function main(): Promise { metaCorePath: config.metaCorePath, filesPath: config.filesPath, serviceName: 'meta-fuse', - version: config.serviceVersion, apiPort: config.apiPort, - redisUrl: config.redisUrl, - redisPrefix: config.redisPrefix, baseUrl: config.baseUrl, - capabilities: ['read', 'vfs'], }); let redisClient: RedisClient | null = null; diff --git a/packages/meta-fuse-core/src/kv/IKVClient.ts b/packages/meta-fuse-core/src/kv/IKVClient.ts index d4d5a7d..db697ae 100644 --- a/packages/meta-fuse-core/src/kv/IKVClient.ts +++ b/packages/meta-fuse-core/src/kv/IKVClient.ts @@ -58,17 +58,24 @@ export interface IKVClient { } /** - * Leader lock file content format + * Leader lock file content format (kv-leader.info) + * Written by meta-core leader, read by other services */ export interface LeaderLockInfo { /** Hostname of the leader */ - host: string; + hostname: string; + + /** Base URL for the leader service (e.g., http://localhost:8180) */ + baseUrl: string; + + /** meta-core API URL (port 9000) */ + apiUrl: string; /** Redis connection URL (e.g., redis://10.0.1.50:6379) */ - api: string; + redisUrl: string; - /** HTTP API URL for the leader service */ - http: string; + /** WebDAV URL for file access */ + webdavUrl: string; /** Timestamp when leadership was acquired */ timestamp: number; @@ -78,36 +85,22 @@ export interface LeaderLockInfo { } /** - * Service registration info stored in JSON file + * Service registration info (simplified) + * Full URLs are obtained via the /urls API endpoint */ export interface ServiceInfo { - /** Service name */ + /** Service name (e.g., 'meta-sort', 'meta-fuse') */ name: string; - /** Service version */ - version: string; - - /** HTTP API URL (e.g., http://10.0.1.50:3000) */ - api: string; - - /** Service status */ - status: 'starting' | 'running' | 'stopping' | 'stopped'; - - /** Process ID */ - pid: number; - /** Hostname */ hostname: string; - /** Service start time (ISO string) */ - startedAt: string; + /** Base URL for the service */ + baseUrl: string; - /** Last heartbeat time (ISO string) */ + /** Current status */ + status: 'running' | 'stale' | 'stopped'; + + /** Last heartbeat timestamp (ISO string) */ lastHeartbeat: string; - - /** Service capabilities */ - capabilities: string[]; - - /** Named endpoints */ - endpoints: Record; } diff --git a/packages/meta-fuse-core/src/kv/KVManager.ts b/packages/meta-fuse-core/src/kv/KVManager.ts index 5cf69e5..1c9f6b6 100644 --- a/packages/meta-fuse-core/src/kv/KVManager.ts +++ b/packages/meta-fuse-core/src/kv/KVManager.ts @@ -2,16 +2,17 @@ * KV Manager for meta-fuse (FOLLOWER-only) * * Simplified KV manager that: - * 1. Discovers the leader via lock file + * 1. Discovers the leader via LeaderClient (reads kv-leader.info) * 2. Creates Redis client connection * 3. Handles reconnection on leader failure * * Unlike meta-sort's KVManager, this version never spawns Redis - - * it only connects as a client. + * it only connects as a client. Leader election is handled by meta-core. */ +import { hostname } from 'os'; import { Logger } from 'tslog'; -import { LeaderDiscovery } from './LeaderDiscovery.js'; +import { LeaderClient } from './LeaderClient.js'; import { ServiceDiscovery } from './ServiceDiscovery.js'; import { RedisClient } from './RedisClient.js'; import type { LeaderLockInfo } from './IKVClient.js'; @@ -28,9 +29,6 @@ interface KVManagerConfig { /** Service name (for logging) */ serviceName?: string; - /** Service version */ - version?: string; - /** HTTP API port */ apiPort?: number; @@ -43,13 +41,13 @@ interface KVManagerConfig { /** Base URL for service discovery (overrides auto-detected URL) */ baseUrl?: string; - /** Service capabilities */ - capabilities?: string[]; + /** meta-core API URL (e.g., http://localhost:9000) */ + metaCoreUrl?: string; } export class KVManager { private config: Required; - private leaderDiscovery: LeaderDiscovery | null = null; + private leaderClient: LeaderClient | null = null; private serviceDiscovery: ServiceDiscovery | null = null; private redisClient: RedisClient | null = null; private isStarted = false; @@ -62,12 +60,11 @@ export class KVManager { constructor(config: KVManagerConfig) { this.config = { serviceName: 'meta-fuse', - version: '1.0.0', apiPort: 3000, redisUrl: '', redisPrefix: 'meta-sort:', baseUrl: '', - capabilities: ['read', 'vfs'], + metaCoreUrl: '', ...config }; } @@ -83,24 +80,15 @@ export class KVManager { logger.info(`Starting KVManager for ${this.config.serviceName}...`); - // Initialize service discovery - const apiUrl = this.config.baseUrl || `http://localhost:${this.config.apiPort}`; + // Initialize service discovery (meta-core handles actual registration) + // We keep this for service discovery (reading other services) + const apiUrl = this.config.baseUrl || `http://${hostname()}:${this.config.apiPort}`; this.serviceDiscovery = new ServiceDiscovery({ metaCorePath: this.config.metaCorePath, serviceName: this.config.serviceName, - version: this.config.version, - apiUrl, - baseUrl: this.config.baseUrl || undefined, - capabilities: this.config.capabilities, - endpoints: { - health: '/api/fuse/health', - dashboard: '/', - stats: '/api/fuse/stats', - } + apiUrl }); - - // Start service discovery - await this.serviceDiscovery.start(); + // Note: We don't start ServiceDiscovery - meta-core handles registration // If direct Redis URL provided, skip leader discovery if (this.config.redisUrl) { @@ -110,29 +98,56 @@ export class KVManager { return; } - // Use leader discovery - this.leaderDiscovery = new LeaderDiscovery({ - metaCorePath: this.config.metaCorePath + // Use leader client + this.leaderClient = new LeaderClient({ + metaCorePath: this.config.metaCorePath, + metaCoreUrl: this.config.metaCoreUrl || undefined }); - // Set up leader discovery callbacks - this.leaderDiscovery.onLeaderFound(async (info: LeaderLockInfo) => { - logger.info(`Connecting to leader at ${info.api}...`); - await this.connectToRedis(info.api); + // Set up leader change callback + this.leaderClient.onChange(async () => { + logger.info('Leader changed, reconnecting...'); + await this.reconnect(); }); - this.leaderDiscovery.onLeaderLost(async () => { - logger.warn('Leader lost, disconnecting...'); - await this.disconnectRedis(); - this.notifyDisconnect(); - }); + // Wait for leader and connect + try { + const leaderInfo = await this.leaderClient.waitForLeader(30000); + logger.info(`Connecting to Redis at ${leaderInfo.redisUrl}...`); + await this.connectToRedis(leaderInfo.redisUrl); + } catch (error: any) { + logger.error(`Failed to connect to leader: ${error.message}`); + throw error; + } - // Start leader discovery - await this.leaderDiscovery.start(); + // Start watching for leader changes + this.leaderClient.startWatching(); this.isStarted = true; } + /** + * Reconnect to Redis after leader change + */ + private async reconnect(): Promise { + if (this.isShuttingDown || !this.leaderClient) { + return; + } + + // Disconnect existing + await this.disconnectRedis(); + this.notifyDisconnect(); + + // Wait for new leader and connect + try { + const leaderInfo = await this.leaderClient.waitForLeader(30000); + logger.info(`Reconnecting to Redis at ${leaderInfo.redisUrl}...`); + await this.connectToRedis(leaderInfo.redisUrl); + } catch (error: any) { + logger.error(`Failed to reconnect: ${error.message}`); + } + } + /** * Stop the KV manager */ @@ -142,18 +157,18 @@ export class KVManager { logger.info('Stopping KVManager...'); this.isShuttingDown = true; - // Stop service discovery + // Stop leader client + if (this.leaderClient) { + this.leaderClient.close(); + this.leaderClient = null; + } + + // Stop service discovery (if it was started) if (this.serviceDiscovery) { await this.serviceDiscovery.stop(); this.serviceDiscovery = null; } - // Stop leader discovery - if (this.leaderDiscovery) { - await this.leaderDiscovery.stop(); - this.leaderDiscovery = null; - } - // Disconnect Redis await this.disconnectRedis(); @@ -313,10 +328,17 @@ export class KVManager { } /** - * Get leader info (if using leader discovery) + * Get leader info */ getLeaderInfo(): LeaderLockInfo | null { - return this.leaderDiscovery?.getLeaderInfo() ?? null; + return this.leaderClient?.getCachedLeaderInfo() ?? null; + } + + /** + * Get leader client + */ + getLeaderClient(): LeaderClient | null { + return this.leaderClient; } /** diff --git a/packages/meta-fuse-core/src/kv/LeaderClient.ts b/packages/meta-fuse-core/src/kv/LeaderClient.ts new file mode 100644 index 0000000..86bfc5a --- /dev/null +++ b/packages/meta-fuse-core/src/kv/LeaderClient.ts @@ -0,0 +1,223 @@ +/** + * LeaderClient - Client for reading leader info from meta-core + * + * This replaces the old LeaderDiscovery.ts by delegating leader election + * to meta-core (Go sidecar) and just reading the results. + * + * Features: + * - Reads kv-leader.info file for leader URLs + * - Calls meta-core /urls API for current leader URLs + * - Watches for leader changes via file system watcher + */ + +import { promises as fs } from 'fs'; +import { watch, FSWatcher } from 'fs'; +import { dirname } from 'path'; +import { Logger } from 'tslog'; +import type { LeaderLockInfo } from './IKVClient.js'; + +const logger = new Logger({ name: 'LeaderClient' }); + +export interface LeaderClientConfig { + /** Path to META_CORE_VOLUME (e.g., /meta-core) */ + metaCorePath: string; + + /** meta-core API URL (e.g., http://meta-core:9000 or http://localhost:9000) */ + metaCoreUrl?: string; +} + +export interface URLsResponse { + hostname: string; + baseUrl: string; + apiUrl: string; + redisUrl: string; + webdavUrl: string; + isLeader: boolean; +} + +export class LeaderClient { + private config: LeaderClientConfig; + private infoFilePath: string; + private leaderInfo: LeaderLockInfo | null = null; + private watcher: FSWatcher | null = null; + private onChangeCallbacks: (() => void)[] = []; + + constructor(config: LeaderClientConfig) { + this.config = config; + this.infoFilePath = `${config.metaCorePath}/locks/kv-leader.info`; + } + + /** + * Read leader info from the info file + */ + async getLeaderInfo(): Promise { + try { + const content = await fs.readFile(this.infoFilePath, 'utf-8'); + this.leaderInfo = JSON.parse(content) as LeaderLockInfo; + return this.leaderInfo; + } catch (error: any) { + logger.error(`Failed to read leader info: ${error.message}`); + return null; + } + } + + /** + * Get Redis URL from leader info + */ + async getRedisUrl(): Promise { + const info = await this.getLeaderInfo(); + return info?.redisUrl ?? null; + } + + /** + * Get WebDAV URL from leader info + */ + async getWebdavUrl(): Promise { + const info = await this.getLeaderInfo(); + return info?.webdavUrl ?? null; + } + + /** + * Get meta-core API URL from leader info + */ + async getApiUrl(): Promise { + const info = await this.getLeaderInfo(); + return info?.apiUrl ?? null; + } + + /** + * Call meta-core /urls API to get current URLs + * Useful for getting URLs when you don't want to read the file directly + */ + async getUrls(): Promise { + // First try using configured metaCoreUrl + let apiUrl = this.config.metaCoreUrl; + + // Fall back to reading from leader info + if (!apiUrl) { + const info = await this.getLeaderInfo(); + apiUrl = info?.apiUrl; + } + + if (!apiUrl) { + logger.error('No meta-core API URL available'); + return null; + } + + try { + const response = await fetch(`${apiUrl}/urls`, { + method: 'GET', + headers: { 'Accept': 'application/json' }, + signal: AbortSignal.timeout(5000) + }); + + if (!response.ok) { + logger.error(`Failed to get URLs: ${response.status} ${response.statusText}`); + return null; + } + + return await response.json() as URLsResponse; + } catch (error: any) { + logger.error(`Error calling /urls API: ${error.message}`); + return null; + } + } + + /** + * Wait for leader info to be available + */ + async waitForLeader(timeoutMs: number = 30000): Promise { + const startTime = Date.now(); + const pollInterval = 500; + + while (Date.now() - startTime < timeoutMs) { + const info = await this.getLeaderInfo(); + if (info) { + logger.info(`Leader found: ${info.hostname} at ${info.redisUrl}`); + return info; + } + + logger.debug('Waiting for leader...'); + await new Promise(resolve => setTimeout(resolve, pollInterval)); + } + + throw new Error(`No leader found within ${timeoutMs}ms`); + } + + /** + * Start watching for leader changes + */ + startWatching(): void { + if (this.watcher) { + return; // Already watching + } + + const lockDir = dirname(this.infoFilePath); + + try { + this.watcher = watch(lockDir, (eventType, filename) => { + if (filename === 'kv-leader.info') { + logger.debug(`Lock file ${eventType}, checking for leader...`); + this.getLeaderInfo().then(() => { + this.notifyChange(); + }).catch((err) => logger.error(`Error reloading leader info: ${err.message}`)); + } + }); + + this.watcher.on('error', (error: any) => { + logger.warn(`File watcher error: ${error.message}`); + }); + + logger.info(`Watching for leader changes in ${lockDir}`); + } catch (error: any) { + logger.error(`Failed to start watching: ${error.message}`); + } + } + + /** + * Stop watching for leader changes + */ + stopWatching(): void { + if (this.watcher) { + this.watcher.close(); + this.watcher = null; + logger.debug('Stopped watching for leader changes'); + } + } + + /** + * Register callback for leader changes + */ + onChange(callback: () => void): this { + this.onChangeCallbacks.push(callback); + return this; + } + + /** + * Notify all change callbacks + */ + private notifyChange(): void { + for (const callback of this.onChangeCallbacks) { + try { + callback(); + } catch (error: any) { + logger.error(`Error in change callback: ${error.message}`); + } + } + } + + /** + * Get cached leader info (without re-reading file) + */ + getCachedLeaderInfo(): LeaderLockInfo | null { + return this.leaderInfo; + } + + /** + * Clean up resources + */ + close(): void { + this.stopWatching(); + this.onChangeCallbacks = []; + } +} diff --git a/packages/meta-fuse-core/src/kv/LeaderDiscovery.ts b/packages/meta-fuse-core/src/kv/LeaderDiscovery.ts deleted file mode 100644 index f4e30c8..0000000 --- a/packages/meta-fuse-core/src/kv/LeaderDiscovery.ts +++ /dev/null @@ -1,276 +0,0 @@ -/** - * Leader Discovery for meta-fuse (FOLLOWER-only) - * - * This is a simplified version of leader election that only discovers - * the existing leader. meta-fuse never spawns Redis - it only connects - * to the leader's Redis instance. - * - * Architecture: - * - Leader holds flock on META_CORE_PATH/locks/kv-leader.lock - * - Leader writes info to META_CORE_PATH/locks/kv-leader.info - * - This class reads the info file to discover the leader - * - Watches for leader changes via health checks - * - Triggers reconnection callbacks on leader failure/change - */ - -import { promises as fs, watch } from 'fs'; -import { dirname } from 'path'; -import { Logger } from 'tslog'; -import type { LeaderLockInfo } from './IKVClient.js'; - -const logger = new Logger({ name: 'LeaderDiscovery' }); - -interface LeaderDiscoveryConfig { - /** Path to META_CORE_VOLUME (e.g., /meta-core) */ - metaCorePath: string; - - /** Health check interval in ms (default: 5000) */ - healthCheckInterval?: number; - - /** Max consecutive failures before marking leader as lost (default: 3) */ - maxFailures?: number; -} - -export class LeaderDiscovery { - private config: Required; - private lockFilePath: string; - private leaderInfo: LeaderLockInfo | null = null; - private healthCheckTimer: NodeJS.Timeout | null = null; - private fileWatcher: ReturnType | null = null; - private isStarted = false; - private consecutiveFailures = 0; - - // Event callbacks - private onLeaderFoundCallback?: (info: LeaderLockInfo) => void; - private onLeaderLostCallback?: () => void; - - constructor(config: LeaderDiscoveryConfig) { - this.config = { - healthCheckInterval: 5000, - maxFailures: 3, - ...config - }; - - this.lockFilePath = `${this.config.metaCorePath}/locks/kv-leader.info`; - } - - /** - * Start watching for leader - */ - async start(): Promise { - if (this.isStarted) { - logger.warn('LeaderDiscovery already started'); - return; - } - - logger.info(`Starting leader discovery, lock file: ${this.lockFilePath}`); - - // Ensure lock directory exists (for file watching) - await this.ensureLockDir(); - - // Try to read existing leader info - await this.checkForLeader(); - - // Start file watcher for immediate detection of leader changes - this.startFileWatcher(); - - // Start health check for ongoing monitoring - this.startHealthCheck(); - - this.isStarted = true; - } - - /** - * Stop watching - */ - async stop(): Promise { - if (!this.isStarted) return; - - logger.info('Stopping leader discovery'); - - if (this.healthCheckTimer) { - clearInterval(this.healthCheckTimer); - this.healthCheckTimer = null; - } - - if (this.fileWatcher) { - this.fileWatcher.close(); - this.fileWatcher = null; - } - - this.isStarted = false; - } - - /** - * Ensure lock directory exists - */ - private async ensureLockDir(): Promise { - const lockDir = dirname(this.lockFilePath); - try { - await fs.mkdir(lockDir, { recursive: true }); - } catch (error: any) { - if (error.code !== 'EEXIST') { - logger.warn(`Could not create lock directory: ${error.message}`); - } - } - } - - /** - * Check for leader by reading lock file - */ - private async checkForLeader(): Promise { - try { - const content = await fs.readFile(this.lockFilePath, 'utf-8'); - const info = JSON.parse(content) as LeaderLockInfo; - - // Validate leader info - if (info && info.api && info.http) { - const isNew = !this.leaderInfo || this.leaderInfo.api !== info.api; - this.leaderInfo = info; - this.consecutiveFailures = 0; - - if (isNew) { - logger.info(`Leader found: ${info.host} at ${info.api}`); - this.onLeaderFoundCallback?.(info); - } - } - } catch (error: any) { - if (error.code === 'ENOENT') { - // Lock file doesn't exist - no leader yet - if (this.leaderInfo) { - logger.warn('Lock file removed, leader lost'); - this.handleLeaderLost(); - } - } else { - logger.warn(`Error reading lock file: ${error.message}`); - } - } - } - - /** - * Check if current leader is still healthy - */ - private async isLeaderHealthy(): Promise { - if (!this.leaderInfo) return false; - - try { - const response = await fetch(`${this.leaderInfo.http}/health`, { - signal: AbortSignal.timeout(5000) - }); - return response.ok; - } catch { - return false; - } - } - - /** - * Handle leader being lost - */ - private handleLeaderLost(): void { - this.leaderInfo = null; - this.onLeaderLostCallback?.(); - } - - /** - * Start watching lock file for changes - */ - private startFileWatcher(): void { - try { - const lockDir = dirname(this.lockFilePath); - this.fileWatcher = watch(lockDir, async (eventType, filename) => { - if (filename === 'kv-leader.info') { - logger.debug(`Lock file ${eventType}, checking for leader...`); - await this.checkForLeader(); - } - }); - - this.fileWatcher.on('error', (error) => { - logger.warn(`File watcher error: ${error.message}`); - }); - } catch (error: any) { - logger.warn(`Could not start file watcher: ${error.message}`); - } - } - - /** - * Start health check loop - */ - private startHealthCheck(): void { - this.healthCheckTimer = setInterval(async () => { - if (!this.leaderInfo) { - // No leader known, try to find one - await this.checkForLeader(); - return; - } - - // Check if current leader is healthy - const healthy = await this.isLeaderHealthy(); - - if (healthy) { - this.consecutiveFailures = 0; - } else { - this.consecutiveFailures++; - logger.warn(`Leader health check failed (${this.consecutiveFailures}/${this.config.maxFailures})`); - - if (this.consecutiveFailures >= this.config.maxFailures) { - logger.warn('Leader appears dead, marking as lost'); - this.handleLeaderLost(); - - // Try to find new leader - await this.checkForLeader(); - } - } - }, this.config.healthCheckInterval); - } - - // ======================================================================== - // Event Registration - // ======================================================================== - - /** - * Register callback for when leader is found/changed - */ - onLeaderFound(callback: (info: LeaderLockInfo) => void): this { - this.onLeaderFoundCallback = callback; - - // If we already have leader info, call immediately - if (this.leaderInfo) { - callback(this.leaderInfo); - } - - return this; - } - - /** - * Register callback for when leader is lost - */ - onLeaderLost(callback: () => void): this { - this.onLeaderLostCallback = callback; - return this; - } - - // ======================================================================== - // Getters - // ======================================================================== - - /** - * Get current leader info - */ - getLeaderInfo(): LeaderLockInfo | null { - return this.leaderInfo; - } - - /** - * Get Redis URL of current leader - */ - getRedisUrl(): string | null { - return this.leaderInfo?.api ?? null; - } - - /** - * Check if a leader is known - */ - hasLeader(): boolean { - return this.leaderInfo !== null; - } -} diff --git a/packages/meta-fuse-core/src/kv/ServiceDiscovery.ts b/packages/meta-fuse-core/src/kv/ServiceDiscovery.ts index 211d467..7013717 100644 --- a/packages/meta-fuse-core/src/kv/ServiceDiscovery.ts +++ b/packages/meta-fuse-core/src/kv/ServiceDiscovery.ts @@ -1,11 +1,14 @@ /** * Service Discovery using shared filesystem * - * Each service registers itself in /meta-core/services/{service-name}.json - * with its API endpoint, status, and heartbeat timestamp. + * Each service registers itself in /meta-core/services/{service-name}-{hostname}.json + * with its base URL, status, and heartbeat timestamp. * * Services can discover each other by reading these JSON files. - * Stale detection: if lastHeartbeat > 60 seconds ago, service is considered dead. + * Stale detection: if lastHeartbeat > 60 seconds ago, service is considered stale. + * + * Note: meta-core (Go sidecar) handles actual service registration. + * This class is primarily used for discovering other services. */ import { promises as fs } from 'fs'; @@ -23,20 +26,8 @@ interface ServiceDiscoveryConfig { /** Service name */ serviceName: string; - /** Service version */ - version: string; - - /** HTTP API URL (can be overridden by baseUrl) */ - apiUrl?: string; - - /** Base URL for service discovery (overrides auto-detected URL) */ - baseUrl?: string; - - /** Service capabilities */ - capabilities?: string[]; - - /** Named endpoints */ - endpoints?: Record; + /** Base URL for the service */ + apiUrl: string; /** Heartbeat interval in ms (default: 30000) */ heartbeatInterval?: number; @@ -52,18 +43,19 @@ export class ServiceDiscovery { private heartbeatTimer: NodeJS.Timeout | null = null; private isShuttingDown = false; private isStarted = false; + private currentHostname: string; constructor(config: ServiceDiscoveryConfig) { this.config = { heartbeatInterval: 30000, staleThreshold: 60000, - capabilities: [], - endpoints: {}, ...config }; + this.currentHostname = hostname(); this.servicesDir = `${this.config.metaCorePath}/services`; - this.serviceFilePath = `${this.servicesDir}/${this.config.serviceName}.json`; + // Use hostname-based file naming + this.serviceFilePath = `${this.servicesDir}/${this.config.serviceName}-${this.currentHostname}.json`; } /** @@ -79,28 +71,25 @@ export class ServiceDiscovery { private buildServiceInfo(status: ServiceInfo['status']): ServiceInfo { return { name: this.config.serviceName, - version: this.config.version, - api: this.config.baseUrl || this.config.apiUrl || '', + hostname: this.currentHostname, + baseUrl: this.config.apiUrl, status, - pid: process.pid, - hostname: hostname(), - startedAt: new Date().toISOString(), - lastHeartbeat: new Date().toISOString(), - capabilities: this.config.capabilities || [], - endpoints: this.config.endpoints || {} + lastHeartbeat: new Date().toISOString() }; } /** * Register this service + * Note: In the new architecture, meta-core handles registration. + * This is kept for backwards compatibility and testing. */ async register(): Promise { await this.ensureServicesDir(); - const info = this.buildServiceInfo('starting'); + const info = this.buildServiceInfo('running'); await fs.writeFile(this.serviceFilePath, JSON.stringify(info, null, 2)); - logger.info(`Registered ${this.config.serviceName}`); + logger.info(`Registered ${this.config.serviceName}-${this.currentHostname}`); } /** @@ -135,7 +124,6 @@ export class ServiceDiscovery { // If file was deleted, re-register if (error.code === 'ENOENT') { await this.register(); - await this.updateStatus('running'); } else { logger.error('Heartbeat failed:', error); } @@ -170,39 +158,72 @@ export class ServiceDiscovery { this.stopHeartbeat(); try { - // Update status to stopped before removing - await this.updateStatus('stopped'); - - logger.info(`Unregistered ${this.config.serviceName}`); - } catch (error) { - logger.error('Failed to unregister:', error); + // Remove the service file + await fs.unlink(this.serviceFilePath); + logger.info(`Unregistered ${this.config.serviceName}-${this.currentHostname}`); + } catch (error: any) { + if (error.code !== 'ENOENT') { + logger.error('Failed to unregister:', error); + } } } + /** + * Check if a service info is stale + */ + private isStale(info: ServiceInfo): boolean { + const lastHeartbeat = new Date(info.lastHeartbeat).getTime(); + return Date.now() - lastHeartbeat > this.config.staleThreshold!; + } + // ======================================================================== // Discovery Methods // ======================================================================== /** * Discover a service by name + * Looks for files matching pattern: {name}-*.json (hostname-based naming) */ async discoverService(name: string): Promise { - const servicePath = join(this.servicesDir, `${name}.json`); - try { - const content = await fs.readFile(servicePath, 'utf-8'); - const service = JSON.parse(content) as ServiceInfo; + // First try exact match for backward compatibility + const exactPath = join(this.servicesDir, `${name}.json`); + try { + const content = await fs.readFile(exactPath, 'utf-8'); + const service = JSON.parse(content) as ServiceInfo; - // Check if service is alive (heartbeat within threshold) - const lastHeartbeat = new Date(service.lastHeartbeat).getTime(); - const isAlive = Date.now() - lastHeartbeat < this.config.staleThreshold!; + if (this.isStale(service)) { + service.status = 'stale'; + } - if (!isAlive) { - logger.warn(`Service ${name} appears stale (last heartbeat: ${service.lastHeartbeat})`); - return null; + return service.status === 'running' ? service : null; + } catch { + // Try hostname-based files } - return service; + // Search for hostname-based files: name-*.json + const files = await fs.readdir(this.servicesDir); + const matchingFiles = files.filter(f => + f.startsWith(`${name}-`) && f.endsWith('.json') + ); + + // Return first valid (non-stale) service + for (const file of matchingFiles) { + const filePath = join(this.servicesDir, file); + const content = await fs.readFile(filePath, 'utf-8'); + const service = JSON.parse(content) as ServiceInfo; + + if (this.isStale(service)) { + service.status = 'stale'; + continue; // Skip stale services + } + + if (service.status === 'running') { + return service; + } + } + + return null; } catch (error: any) { if (error.code === 'ENOENT') { return null; // Service not registered @@ -222,11 +243,18 @@ export class ServiceDiscovery { for (const file of files) { if (!file.endsWith('.json')) continue; - const name = file.replace('.json', ''); - const service = await this.discoverService(name); + const filePath = join(this.servicesDir, file); + try { + const content = await fs.readFile(filePath, 'utf-8'); + const service = JSON.parse(content) as ServiceInfo; + + if (this.isStale(service)) { + service.status = 'stale'; + } - if (service && service.status === 'running') { services.push(service); + } catch (error) { + logger.error(`Failed to read ${file}:`, error); } } @@ -247,8 +275,7 @@ export class ServiceDiscovery { if (!service) return false; // Try to ping the service's health endpoint - const healthEndpoint = service.endpoints.health || '/health'; - const url = `${service.api}${healthEndpoint}`; + const url = `${service.baseUrl}/health`; try { const response = await fetch(url, { @@ -266,16 +293,16 @@ export class ServiceDiscovery { /** * Full startup sequence + * Note: In the new architecture, meta-core handles registration. */ async start(): Promise { if (this.isStarted) return; await this.register(); - await this.updateStatus('running'); this.startHeartbeat(); this.isStarted = true; - logger.info(`Started ${this.config.serviceName} at ${this.config.baseUrl || this.config.apiUrl}`); + logger.info(`Started ${this.config.serviceName} at ${this.config.apiUrl}`); } /**