diff --git a/packages/meta-sort-core/src/api/UnifiedAPIServer.ts b/packages/meta-sort-core/src/api/UnifiedAPIServer.ts index 579f233..a738814 100644 --- a/packages/meta-sort-core/src/api/UnifiedAPIServer.ts +++ b/packages/meta-sort-core/src/api/UnifiedAPIServer.ts @@ -988,6 +988,7 @@ export class UnifiedAPIServer { url: string; api: string; status: string; + role?: string; }> = []; try { @@ -1001,6 +1002,7 @@ export class UnifiedAPIServer { url: svc.baseUrl || '', api: svc.baseUrl || '', status: svc.status || 'unknown', + role: svc.role, }); } } diff --git a/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts b/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts index e0f95c6..787fed1 100644 --- a/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts +++ b/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts @@ -19,9 +19,13 @@ import type { } from './types.js'; import type { TaskQueueType } from '../plugin-engine/types.js'; import type { IKVClient } from '../kv/IKVClient.js'; -import { hasPublish, type ExtendedContainerTask } from '../types/ExtendedInterfaces.js'; +import { hasStreamSupport, type ExtendedContainerTask } from '../types/ExtendedInterfaces.js'; import { config } from '../config/EnvConfig.js'; +// Redis Streams for reliable event delivery +const EVENTS_STREAM = 'meta-sort:events'; +const STREAM_MAXLEN = 10000; + /** * Scheduler events */ @@ -212,27 +216,30 @@ export class ContainerPluginScheduler extends EventEmitter { } /** - * Publish plugin completion event to Redis - * Channel: meta-sort:plugin:complete + * Publish plugin completion event to Redis Streams * - * External services (like meta-stremio) can subscribe to this channel + * External services (like meta-fuse, meta-stremio) consume this stream * to be notified when specific plugins complete for a file. */ private async publishPluginComplete(fileHash: string, pluginId: string, filePath: string): Promise { - if (!this.kvClient || !hasPublish(this.kvClient)) { + if (!this.kvClient || !hasStreamSupport(this.kvClient)) { return; } try { - const message = JSON.stringify({ + const payload = JSON.stringify({ fileHash, pluginId, filePath, timestamp: Date.now() }); - await this.kvClient.publish('meta-sort:plugin:complete', message); + await this.kvClient.xadd(EVENTS_STREAM, STREAM_MAXLEN, { + type: 'plugin:complete', + payload, + timestamp: Date.now().toString() + }); } catch (error) { - // Silent fail - pub/sub is optional + // Silent fail - stream publish is optional } } diff --git a/packages/meta-sort-core/src/kv/IKVClient.ts b/packages/meta-sort-core/src/kv/IKVClient.ts index a05953d..d5dbc54 100644 --- a/packages/meta-sort-core/src/kv/IKVClient.ts +++ b/packages/meta-sort-core/src/kv/IKVClient.ts @@ -184,4 +184,7 @@ export interface ServiceInfo { /** Last heartbeat timestamp (ISO string) */ lastHeartbeat: string; + + /** Role for meta-core instances: 'leader', 'follower', or undefined for other services */ + role?: string; } diff --git a/packages/meta-sort-core/src/kv/KVManager.ts b/packages/meta-sort-core/src/kv/KVManager.ts index 28ade05..0df09b3 100644 --- a/packages/meta-sort-core/src/kv/KVManager.ts +++ b/packages/meta-sort-core/src/kv/KVManager.ts @@ -130,8 +130,7 @@ export class KVManager { // Start watching for leader changes this.leaderClient.startWatching(); - // Initialize service discovery (meta-core handles actual registration) - // We keep this for service discovery (reading other services) + // Initialize and start service discovery for registration and heartbeat const apiUrl = this.config.baseUrl || `http://${hostname()}:${this.config.apiPort}`; this.serviceDiscovery = new ServiceDiscovery({ metaCorePath: this.config.metaCorePath, @@ -139,7 +138,7 @@ export class KVManager { version: '1.0.0', apiUrl }); - // Note: We don't start ServiceDiscovery - meta-core handles registration + await this.serviceDiscovery.start(); this.isStarted = true; this.notifyReady(); diff --git a/packages/meta-sort-core/src/kv/LeaderClient.ts b/packages/meta-sort-core/src/kv/LeaderClient.ts index cb4a637..2e3372d 100644 --- a/packages/meta-sort-core/src/kv/LeaderClient.ts +++ b/packages/meta-sort-core/src/kv/LeaderClient.ts @@ -39,18 +39,87 @@ export class LeaderClient { private watcher: FSWatcher | null = null; private onChangeCallbacks: (() => void)[] = []; + // URL caching + private cachedUrls: URLsResponse | null = null; + private urlsCacheTime: number = 0; + private readonly urlsCacheTTL: number = 5000; // 5 seconds + constructor(config: LeaderClientConfig) { this.config = config; this.infoFilePath = `${config.metaCorePath}/locks/kv-leader.info`; } /** - * Read leader info from the info file + * Read API URL from file (plain text format) + */ + private async getApiUrlFromFile(): Promise { + try { + const content = await fs.readFile(this.infoFilePath, 'utf-8'); + return content.trim() || null; + } catch (error) { + console.error('[LeaderClient] Failed to read API URL from file:', error); + return null; + } + } + + /** + * Fetch URLs from meta-core /urls API with caching + */ + private async fetchUrls(apiUrl: string): Promise { + // Check cache + const now = Date.now(); + if (this.cachedUrls && (now - this.urlsCacheTime) < this.urlsCacheTTL) { + return this.cachedUrls; + } + + try { + const response = await fetch(`${apiUrl}/urls`, { + method: 'GET', + headers: { 'Accept': 'application/json' }, + signal: AbortSignal.timeout(5000) + }); + + if (!response.ok) { + console.error('[LeaderClient] Failed to fetch URLs:', response.status, response.statusText); + return null; + } + + this.cachedUrls = await response.json() as URLsResponse; + this.urlsCacheTime = now; + return this.cachedUrls; + } catch (error) { + console.error('[LeaderClient] Error calling /urls API:', error); + return null; + } + } + + /** + * Read leader info from file and /urls API */ async getLeaderInfo(): Promise { try { - const content = await fs.readFile(this.infoFilePath, 'utf-8'); - this.leaderInfo = JSON.parse(content) as LeaderLockInfo; + // Read API URL from file (plain text) + const apiUrl = await this.getApiUrlFromFile(); + if (!apiUrl) { + return null; + } + + // Fetch full info from /urls API + const urls = await this.fetchUrls(apiUrl); + if (!urls) { + return null; + } + + // Convert URLsResponse to LeaderLockInfo + this.leaderInfo = { + hostname: urls.hostname, + baseUrl: urls.baseUrl, + apiUrl: urls.apiUrl, + redisUrl: urls.redisUrl, + webdavUrl: urls.webdavUrl, + timestamp: Date.now(), + pid: 0 // Unknown for remote leader + }; return this.leaderInfo; } catch (error) { console.error('[LeaderClient] Failed to read leader info:', error); @@ -90,10 +159,9 @@ export class LeaderClient { // First try using configured metaCoreUrl let apiUrl = this.config.metaCoreUrl; - // Fall back to reading from leader info + // Fall back to reading from file if (!apiUrl) { - const info = await this.getLeaderInfo(); - apiUrl = info?.apiUrl; + apiUrl = await this.getApiUrlFromFile(); } if (!apiUrl) { @@ -101,23 +169,7 @@ export class LeaderClient { return null; } - try { - const response = await fetch(`${apiUrl}/urls`, { - method: 'GET', - headers: { 'Accept': 'application/json' }, - signal: AbortSignal.timeout(5000) - }); - - if (!response.ok) { - console.error('[LeaderClient] Failed to get URLs:', response.status, response.statusText); - return null; - } - - return await response.json() as URLsResponse; - } catch (error) { - console.error('[LeaderClient] Error calling /urls API:', error); - return null; - } + return this.fetchUrls(apiUrl); } /** @@ -154,7 +206,11 @@ export class LeaderClient { try { this.watcher = watch(lockDir, (eventType, filename) => { if (filename === 'kv-leader.info') { - console.log('[LeaderClient] Leader info changed, reloading...'); + console.log('[LeaderClient] Leader info changed, invalidating cache...'); + // Invalidate cache to force fresh API call + this.cachedUrls = null; + this.urlsCacheTime = 0; + this.getLeaderInfo().then(() => { this.notifyChange(); }).catch(console.error); diff --git a/packages/meta-sort-core/src/kv/RedisClient.ts b/packages/meta-sort-core/src/kv/RedisClient.ts index cf01140..b43be6f 100644 --- a/packages/meta-sort-core/src/kv/RedisClient.ts +++ b/packages/meta-sort-core/src/kv/RedisClient.ts @@ -338,6 +338,39 @@ export class RedisKVClient implements IKVClient { return await this.redis.publish(channel, message); } + // ======================================================================== + // Redis Streams Operations + // ======================================================================== + + /** + * Add an entry to a Redis stream + * Used for reliable event delivery between services (replaces pub/sub) + * + * @param stream - Stream name (e.g., 'meta-sort:events') + * @param maxlen - Maximum stream length (approximate, uses ~ for efficiency) + * @param fields - Object containing field-value pairs to add + * @returns The ID of the added entry + */ + async xadd(stream: string, maxlen: number, fields: Record): Promise { + // Convert fields object to flat array for ioredis + const fieldArray: string[] = []; + for (const [key, value] of Object.entries(fields)) { + fieldArray.push(key, value); + } + + // Use XADD with MAXLEN ~ for approximate trimming (more efficient) + const result = await this.redis.xadd( + stream, + 'MAXLEN', + '~', + maxlen.toString(), + '*', + ...fieldArray + ); + + return result as string; + } + // ======================================================================== // Lifecycle // ======================================================================== diff --git a/packages/meta-sort-core/src/kv/ServiceDiscovery.ts b/packages/meta-sort-core/src/kv/ServiceDiscovery.ts index c11d59b..89ce165 100644 --- a/packages/meta-sort-core/src/kv/ServiceDiscovery.ts +++ b/packages/meta-sort-core/src/kv/ServiceDiscovery.ts @@ -7,8 +7,8 @@ * Services can discover each other by reading these JSON files. * 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. + * Each service must call start() to register itself and begin heartbeat updates. + * meta-core runs a cleanup process that removes services with stale heartbeats. */ import { promises as fs } from 'fs'; @@ -79,9 +79,8 @@ export class ServiceDiscovery { } /** - * Register this service - * Note: In the new architecture, meta-core handles registration. - * This is kept for backwards compatibility and testing. + * Register this service by writing service info to the shared filesystem. + * meta-core's cleanup process monitors these files and removes stale entries. */ async register(): Promise { await this.ensureServicesDir(); @@ -314,8 +313,7 @@ export class ServiceDiscovery { // ======================================================================== /** - * Full startup sequence - * Note: In the new architecture, meta-core handles registration. + * Full startup sequence: register service and start heartbeat loop. */ async start(): Promise { await this.register(); diff --git a/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts b/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts index 8a098a1..f099118 100644 --- a/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts +++ b/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts @@ -15,7 +15,11 @@ import {VirtualFileSystem} from "../api/VirtualFileSystem.js"; import {performanceMetrics} from "../metrics/PerformanceMetrics.js"; import {UnifiedProcessingStateManager} from "./UnifiedProcessingStateManager.js"; import type {IKVClient} from "../kv/IKVClient.js"; -import { hasPublish, type ExtendedFileAnalyzer, type ExtendedHashMeta } from "../types/ExtendedInterfaces.js"; +import { hasStreamSupport, type ExtendedFileAnalyzer, type ExtendedHashMeta } from "../types/ExtendedInterfaces.js"; + +// Redis Streams for reliable event delivery +const EVENTS_STREAM = 'meta-sort:events'; +const STREAM_MAXLEN = 10000; import PQueue from "p-queue"; import os from 'os'; import { TaskScheduler, createTaskScheduler, MetadataNodeKVStore } from "../plugin-engine/index.js"; @@ -244,25 +248,28 @@ export class WatchedFileProcessor implements FileProcessorInterface { } /** - * Publish plugin completion event to Redis - * Channel: meta-sort:plugin:complete + * Publish plugin completion event to Redis Streams */ private async publishPluginComplete(fileHash: string, pluginId: string, filePath: string): Promise { const kvClient = this.getKVClient(); - if (!kvClient || !hasPublish(kvClient)) { + if (!kvClient || !hasStreamSupport(kvClient)) { return; } try { - const message = JSON.stringify({ + const payload = JSON.stringify({ fileHash, pluginId, filePath, timestamp: Date.now() }); - await kvClient.publish('meta-sort:plugin:complete', message); + await kvClient.xadd(EVENTS_STREAM, STREAM_MAXLEN, { + type: 'plugin:complete', + payload, + timestamp: Date.now().toString() + }); } catch (error) { - // Silent fail - pub/sub is optional + // Silent fail - stream publish is optional } } diff --git a/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts b/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts index 54feea0..05cbbc4 100644 --- a/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts +++ b/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts @@ -2,12 +2,14 @@ import * as path from 'path'; import PQueue from 'p-queue'; import {PipelineConfig} from './PipelineConfig.js'; import {performanceMetrics} from '../../metrics/PerformanceMetrics.js'; +import { hasStreamSupport } from '../../types/ExtendedInterfaces.js'; -// Pub/sub channel for batched updates to meta-fuse -const UPDATE_CHANNEL = 'meta-sort:file:batch'; +// Redis Streams for reliable event delivery to meta-fuse +// Single stream with type field for all event types (batch, reset, plugin:complete) +const EVENTS_STREAM = 'meta-sort:events'; -// Pub/sub channel for scan reset notification -const RESET_CHANNEL = 'meta-sort:scan:reset'; +// Maximum stream length (approximate, ~10000 entries provides hours of history) +const STREAM_MAXLEN = 10000; // Batch update interval (5 seconds) const BATCH_INTERVAL_MS = 5000; @@ -24,7 +26,7 @@ interface BatchChange { * * With midhash256 being instant (< 1s), files appear in VFS immediately * - * Publishes batched updates to meta-fuse via Redis pub/sub every 5 seconds. + * Publishes batched updates to meta-fuse via Redis Streams every 5 seconds. */ export class StreamingPipeline { private validationQueue: PQueue; @@ -37,7 +39,7 @@ export class StreamingPipeline { private fastProcessedCount = 0; private backgroundProcessedCount = 0; - // Batch buffer for pub/sub notifications + // Batch buffer for stream notifications private batchBuffer: BatchChange[] = []; private batchTimer: NodeJS.Timeout | null = null; @@ -100,7 +102,7 @@ export class StreamingPipeline { } /** - * Flush the batch buffer and publish to Redis + * Flush the batch buffer and publish to Redis Streams */ private async flushBatch(): Promise { if (this.batchBuffer.length === 0) { @@ -112,13 +114,17 @@ export class StreamingPipeline { try { const kvClient = this.config.kvClient; - if (kvClient && typeof (kvClient as any).publish === 'function') { - const message = JSON.stringify({ + if (kvClient && hasStreamSupport(kvClient)) { + const payload = JSON.stringify({ timestamp: Date.now(), changes }); - await (kvClient as any).publish(UPDATE_CHANNEL, message); - console.log(`[Pipeline] Published batch update: ${changes.length} changes`); + await kvClient.xadd(EVENTS_STREAM, STREAM_MAXLEN, { + type: 'batch', + payload, + timestamp: Date.now().toString() + }); + console.log(`[Pipeline] Published batch update to stream: ${changes.length} changes`); } } catch (error: any) { // Put changes back in buffer on failure @@ -423,15 +429,19 @@ export class StreamingPipeline { // Clear state manager (discovered, processing, done states) this.config.stateManager.clear(); - // Publish reset event to notify meta-fuse - if (this.config.kvClient && typeof (this.config.kvClient as any).publish === 'function') { + // Publish reset event to Redis Streams to notify meta-fuse + if (this.config.kvClient && hasStreamSupport(this.config.kvClient)) { try { - const message = JSON.stringify({ + const payload = JSON.stringify({ timestamp: Date.now(), action: 'reset' }); - await (this.config.kvClient as any).publish(RESET_CHANNEL, message); - console.log('[Pipeline] Reset event published to meta-fuse'); + await this.config.kvClient.xadd(EVENTS_STREAM, STREAM_MAXLEN, { + type: 'reset', + payload, + timestamp: Date.now().toString() + }); + console.log('[Pipeline] Reset event published to stream'); } catch (error) { console.error('[Pipeline] Failed to publish reset event:', error); } diff --git a/packages/meta-sort-core/src/types/ExtendedInterfaces.ts b/packages/meta-sort-core/src/types/ExtendedInterfaces.ts index 5126c5a..cb3bd23 100644 --- a/packages/meta-sort-core/src/types/ExtendedInterfaces.ts +++ b/packages/meta-sort-core/src/types/ExtendedInterfaces.ts @@ -41,6 +41,37 @@ export function hasPublish(client: IKVClient): client is IKVClientWithPubSub { return typeof (client as IKVClientWithPubSub).publish === 'function'; } +/** + * IKVClient with Redis Streams capabilities + * + * Redis Streams provide reliable event delivery with persistence, + * consumer groups, and replay capabilities. Preferred over pub/sub + * for service-to-service communication. + */ +export interface IKVClientWithStreams extends IKVClient { + /** + * Add an entry to a Redis stream + * + * @param stream - Stream name (e.g., 'meta-sort:events') + * @param maxlen - Maximum stream length (approximate, uses ~ for efficiency) + * @param fields - Object containing field-value pairs to add + * @returns The ID of the added entry + */ + xadd(stream: string, maxlen: number, fields: Record): Promise; + + /** + * Get the underlying Redis client (implementation-specific) + */ + getRedisClient?(): unknown; +} + +/** + * Type guard to check if a KVClient supports Redis Streams + */ +export function hasStreamSupport(client: IKVClient): client is IKVClientWithStreams { + return typeof (client as IKVClientWithStreams).xadd === 'function'; +} + // ============================================================================= // HashMeta Extensions // ============================================================================= diff --git a/packages/meta-sort-ui/src/components/ServiceNav.tsx b/packages/meta-sort-ui/src/components/ServiceNav.tsx index 5032fa9..db407b0 100644 --- a/packages/meta-sort-ui/src/components/ServiceNav.tsx +++ b/packages/meta-sort-ui/src/components/ServiceNav.tsx @@ -7,6 +7,7 @@ interface ServiceInfo { status: string; capabilities: string[]; version: string; + role?: string; // "leader", "follower", or undefined (for non-meta-core services) } interface ServicesResponse { @@ -55,7 +56,11 @@ function ServiceNav() { if (loading) return null; - const sortedServices = [...services].sort((a, b) => a.name.localeCompare(b.name)); + // Filter out follower meta-core instances (only show leader) + const filteredServices = services.filter( + s => s.name !== 'meta-core' || s.role === 'leader' + ); + const sortedServices = [...filteredServices].sort((a, b) => a.name.localeCompare(b.name)); if (sortedServices.length === 0) return null;