diff --git a/packages/meta-fuse-core/src/index.ts b/packages/meta-fuse-core/src/index.ts index 9e2eabf..4ed730b 100644 --- a/packages/meta-fuse-core/src/index.ts +++ b/packages/meta-fuse-core/src/index.ts @@ -11,6 +11,7 @@ import 'dotenv/config'; import { Logger } from 'tslog'; import { KVManager } from './kv/KVManager.js'; import { RedisClient, StreamMessage } from './kv/RedisClient.js'; +import { SSEEventClient, type SSEEvent } from './kv/SSEEventClient.js'; import { VirtualFileSystem } from './vfs/VirtualFileSystem.js'; import { StreamingStateBuilder } from './vfs/StreamingStateBuilder.js'; import { APIServer } from './api/APIServer.js'; @@ -76,13 +77,16 @@ async function main(): Promise { let vfs: VirtualFileSystem | null = null; let stateBuilder: StreamingStateBuilder | null = null; let apiServer: APIServer | null = null; + let sseClient: SSEEventClient | null = null; // Graceful shutdown const shutdown = async (signal: string): Promise => { logger.info(`Received ${signal}, shutting down...`); try { - // Stop stream consumer first + // Stop SSE consumer first + if (sseClient) await sseClient.stop(); + // Stop legacy stream consumer if it ever ran if (redisClient) { redisClient.stopStreamConsumer(); } @@ -155,35 +159,52 @@ async function main(): Promise { // This enables vfs.refresh() to replay the event stream vfs.setStreamingStateBuilder(stateBuilder); - // Bootstrap: Replay meta:events stream from position 0 - // This rebuilds state from all historical events - logger.info(`Replaying ${EVENTS_STREAM} stream from position 0 (streaming bootstrap)...`); - const bootstrapStart = Date.now(); - - const lastId = await redisClient.replayStream( - EVENTS_STREAM, - '0', - async (message: StreamMessage) => { + // Bootstrap + live stream via SSE on /api/events/meta. + // + // We persist the cursor; on a clean restart we resume from where + // we left off. On a `gap` event (cursor trimmed out of retention) + // we fall back to replaying from the oldest available entry, + // which is good enough to rebuild VFS state because every property + // change has been republished into the stream by meta-core. + const apiUrl = leaderInfo?.apiUrl; + if (!apiUrl) { + throw new Error('meta-core apiUrl not available from leader info; cannot start SSE consumer'); + } + logger.info(`Starting SSE consumer for /api/events/meta (apiUrl=${apiUrl})...`); + sseClient = new SSEEventClient({ + url: `${apiUrl.replace(/\/+$/, '')}/api/events/meta`, + // Deliberately NOT persisting the cursor. meta-fuse holds + // VFS state purely in memory, so each restart needs to + // rebuild from a full event replay. Setting cursorPath to + // null + initialCursor to "0-0" forces the gap-handling + // path which resumes from the oldest available stream + // entry — i.e. effectively a full bootstrap. + cursorPath: null, + logTag: '[meta-fuse SSE]', + initialCursor: '0-0', + onEvent: async (e: SSEEvent) => { + // Translate SSE event back into the StreamMessage shape + // StreamingStateBuilder expects. + const message: StreamMessage = { + id: e.id, + type: e.event as any, + key: e.data?.key, + path: e.data?.path, + size: e.data?.size, + midhash256: e.data?.midhash256, + oldPath: e.data?.oldPath, + watcherId: e.data?.watcherId, + timestamp: e.data?.timestamp ?? e.data?.ts, + } as StreamMessage; await handleStreamingEvent(message, stateBuilder!); }, - 100 // Batch size - ); - - const bootstrapTime = Date.now() - bootstrapStart; - const stats = vfs.getStats(); - logger.info(`Streaming bootstrap complete: ${stats.fileCount} files in ${bootstrapTime}ms`); - - // Start live stream consumer from where replay left off - logger.info(`Starting live stream consumer from ${lastId}...`); - redisClient.startSimpleStreamConsumer( - EVENTS_STREAM, - lastId, - async (message: StreamMessage) => { - await handleStreamingEvent(message, stateBuilder!); + onGap: (payload) => { + logger.warn(`SSE cursor ${payload.requested} trimmed; resuming from ${payload.resumeFrom}`); }, - 5000 // 5 second block timeout - ).catch(error => { - logger.error('Stream consumer error:', error); + }); + // Fire-and-forget; start() returns when stop() fires. + sseClient.start().catch((error) => { + logger.error('SSE consumer terminated:', error); }); // Initialize API server (with kvManager for service discovery) diff --git a/packages/meta-fuse-core/src/kv/KVManager.ts b/packages/meta-fuse-core/src/kv/KVManager.ts index 58c0078..8b1229b 100644 --- a/packages/meta-fuse-core/src/kv/KVManager.ts +++ b/packages/meta-fuse-core/src/kv/KVManager.ts @@ -113,7 +113,7 @@ export class KVManager { try { const leaderInfo = await this.leaderClient.waitForLeader(30000); logger.info(`Connecting to Redis at ${leaderInfo.redisUrl}...`); - await this.connectToRedis(leaderInfo.redisUrl); + await this.connectToRedis(leaderInfo.redisUrl, leaderInfo.apiUrl); } catch (error: any) { logger.error(`Failed to connect to leader: ${error.message}`); throw error; @@ -141,7 +141,7 @@ export class KVManager { try { const leaderInfo = await this.leaderClient.waitForLeader(30000); logger.info(`Reconnecting to Redis at ${leaderInfo.redisUrl}...`); - await this.connectToRedis(leaderInfo.redisUrl); + await this.connectToRedis(leaderInfo.redisUrl, leaderInfo.apiUrl); } catch (error: any) { logger.error(`Failed to reconnect: ${error.message}`); } @@ -178,20 +178,28 @@ export class KVManager { /** * Connect to Redis */ - private async connectToRedis(url: string): Promise { + private async connectToRedis(url: string, apiUrl?: string): Promise { try { // Disconnect existing client if any await this.disconnectRedis(); - // Create new client + // Create new client. When apiUrl is supplied, reads route + // through meta-core HTTP (api-mediated-access PR C). + // When url is also empty, the client is HTTP-only and never + // opens a Redis socket (post-PR D state). this.redisClient = new RedisClient({ - url, + url, // may be '' in PR D world prefix: this.config.redisPrefix, - filesVolume: this.config.filesPath + filesVolume: this.config.filesPath, + metaCoreApiUrl: apiUrl, }); await this.redisClient.connect(); - logger.info('Connected to Redis'); + if (url) { + logger.info('Connected to Redis'); + } else { + logger.info('Storage client ready (HTTP-only mode, no Redis)'); + } // Notify ready this.notifyReady(); diff --git a/packages/meta-fuse-core/src/kv/MetaCoreApiClient.ts b/packages/meta-fuse-core/src/kv/MetaCoreApiClient.ts new file mode 100644 index 0000000..c38e33c --- /dev/null +++ b/packages/meta-fuse-core/src/kv/MetaCoreApiClient.ts @@ -0,0 +1,121 @@ +/** + * MetaCoreApiClient — HTTP read client against meta-core for meta-fuse. + * + * meta-fuse needs to read file metadata (property GETs, full-record fetch, + * hashId enumeration) as it builds the VFS. After the api-mediated-access + * lockdown lands these go through meta-core's /meta/{hash} family instead + * of direct Redis reads. + * + * Only the methods meta-fuse actually calls are implemented. See + * meta-core docs/api-mediated-access.md for the full surface. + */ + +export interface MetaCoreApiClientConfig { + apiUrl: string; + timeoutMs?: number; +} + +export class MetaCoreApiClient { + private apiUrl: string; + private timeoutMs: number; + + constructor(config: MetaCoreApiClientConfig) { + this.apiUrl = config.apiUrl.replace(/\/+$/, ''); + this.timeoutMs = config.timeoutMs ?? 30000; + } + + /** + * GET /meta/{hash}/{prop}. Returns the raw string value or null on 404. + * meta-core sends text/plain for property reads. + */ + async getProperty(hashId: string, property: string): Promise { + const path = `/meta/${encodeURIComponent(hashId)}/${this.encodePropertyPath(property)}`; + const url = `${this.apiUrl}${path}`; + const response = await fetch(url, { + method: 'GET', + signal: AbortSignal.timeout(this.timeoutMs), + headers: { Accept: 'text/plain' }, + }); + if (response.status === 404) return null; + if (!response.ok) { + throw new Error(`[MetaCoreApiClient] GET ${path} → ${response.status} ${response.statusText}`); + } + const text = await response.text(); + return text === '' ? null : text; + } + + /** + * GET /meta/{hash}. Returns the flat property map or null on 404. + */ + async getMetadataFlat(hashId: string): Promise | null> { + const path = `/meta/${encodeURIComponent(hashId)}`; + const body = await this.fetchJsonAllowing404('GET', path); + if (body === null) return null; + const flat = (body as any)?.metadata; + if (!flat || typeof flat !== 'object') return null; + return flat as Record; + } + + /** + * GET /meta. Enumerates every known hashId. + */ + async getAllHashIds(): Promise { + const body = await this.fetchJson('GET', '/meta'); + const ids = (body as any)?.hashIds; + return Array.isArray(ids) ? ids : []; + } + + /** + * GET /health on meta-core. Used as the connectivity probe. + */ + async health(): Promise { + try { + const response = await fetch(`${this.apiUrl}/health`, { + method: 'GET', + signal: AbortSignal.timeout(Math.min(this.timeoutMs, 5000)), + headers: { Accept: 'application/json' }, + }); + return response.ok; + } catch { + return false; + } + } + + private encodePropertyPath(property: string): string { + return property + .split('/') + .map(segment => encodeURIComponent(segment)) + .join('/'); + } + + private async fetchJson(method: string, path: string): Promise { + const url = `${this.apiUrl}${path}`; + const response = await fetch(url, { + method, + signal: AbortSignal.timeout(this.timeoutMs), + headers: { Accept: 'application/json' }, + }); + if (!response.ok) { + throw new Error(`[MetaCoreApiClient] ${method} ${path} → ${response.status} ${response.statusText}`); + } + const text = await response.text(); + if (!text) return null; + try { return JSON.parse(text); } catch { return text; } + } + + private async fetchJsonAllowing404(method: string, path: string): Promise { + const url = `${this.apiUrl}${path}`; + const response = await fetch(url, { + method, + signal: AbortSignal.timeout(this.timeoutMs), + headers: { Accept: 'application/json' }, + }); + if (response.status === 404) return null; + if (!response.ok) { + throw new Error(`[MetaCoreApiClient] ${method} ${path} → ${response.status} ${response.statusText}`); + } + const text = await response.text(); + if (!text) return null; + try { return JSON.parse(text); } catch { return text; } + } +} diff --git a/packages/meta-fuse-core/src/kv/RedisClient.ts b/packages/meta-fuse-core/src/kv/RedisClient.ts index 94c7082..5f80b26 100644 --- a/packages/meta-fuse-core/src/kv/RedisClient.ts +++ b/packages/meta-fuse-core/src/kv/RedisClient.ts @@ -12,6 +12,7 @@ import { Redis } from 'ioredis'; import { Logger } from 'tslog'; import * as os from 'os'; import type { IKVClient, KeyValuePair } from './IKVClient.js'; +import { MetaCoreApiClient } from './MetaCoreApiClient.js'; const logger = new Logger({ name: 'RedisClient' }); @@ -104,15 +105,27 @@ export interface RedisClientConfig { prefix?: string; filesVolume?: string; reconnectInterval?: number; + + /** + * meta-core HTTP API base URL. When set, read methods (getProperty, + * getProperties, getMetadataFlat, getAllHashIds, health) route through + * the /meta/{hash} HTTP family instead of touching Redis directly. + * + * This is the meta-fuse half of the api-mediated-access lockdown (see + * meta-core docs/api-mediated-access.md). Reads stay on Redis when + * left unset so tests / one-off scripts still work. + */ + metaCoreApiUrl?: string; } export class RedisClient implements Partial { private client: Redis | null = null; private subscriber: Redis | null = null; - private config: Required; + private config: Required> & Pick; private isConnected = false; private reconnectTimer: NodeJS.Timeout | null = null; private subscriptions: Map void> = new Map(); + private apiClient: MetaCoreApiClient | null = null; // Stream consumer state private streamConsumerRunning = false; @@ -125,20 +138,33 @@ export class RedisClient implements Partial { prefix: config.prefix ?? process.env.REDIS_PREFIX ?? '', filesVolume: config.filesVolume ?? process.env.FILES_VOLUME ?? '/files', reconnectInterval: config.reconnectInterval ?? 5000, + metaCoreApiUrl: config.metaCoreApiUrl, }; + if (this.config.metaCoreApiUrl) { + this.apiClient = new MetaCoreApiClient({ apiUrl: this.config.metaCoreApiUrl }); + logger.info(`Reads routed through meta-core API at ${this.config.metaCoreApiUrl}`); + } + // Unique consumer name for this instance this.consumerName = `${os.hostname()}-${process.pid}`; } /** - * Connect to Redis + * Connect to Redis. In HTTP-only mode (no URL + apiClient set) this is + * a no-op — reads route through MetaCoreApiClient and stream + * consumption is owned by SSEEventClient at the index.ts layer. */ async connect(): Promise { if (this.client && this.isConnected) { return; } + if (this.apiClient && !this.config.url) { + logger.info('HTTP-only mode — no direct Redis connection'); + return; + } + try { this.client = new Redis(this.config.url, { retryStrategy: (times: number) => { @@ -236,9 +262,18 @@ export class RedisClient implements Partial { } /** - * Check if connected + * Check if the metadata backend is reachable. + * + * In HTTP-only mode (api-mediated-access PR D) there's no Redis socket; + * reachability is signalled by having an apiClient configured. The + * health badge is keyed off this; reporting `false` there would make + * the UI permanently show "Disconnected" even though the service is + * working fine over HTTP. */ get connected(): boolean { + if (this.apiClient && !this.config.url) { + return true; + } return this.isConnected; } @@ -255,6 +290,15 @@ export class RedisClient implements Partial { * @returns The property value or null if not found */ async getProperty(hashId: string, property: string): Promise { + if (this.apiClient) { + try { + return await this.apiClient.getProperty(hashId, property); + } catch (error: any) { + logger.error(`HTTP getProperty failed for ${hashId}/${property}:`, error.message); + return null; + } + } + if (!this.client || !this.isConnected) { return null; } diff --git a/packages/meta-fuse-core/src/kv/SSEEventClient.ts b/packages/meta-fuse-core/src/kv/SSEEventClient.ts new file mode 100644 index 0000000..aa64589 --- /dev/null +++ b/packages/meta-fuse-core/src/kv/SSEEventClient.ts @@ -0,0 +1,229 @@ +/** + * SSEEventClient — HTTP Server-Sent Events client for meta-core event streams. + * + * Replaces direct Redis Streams consumption. See the same-named module in + * meta-sort for the design rationale (meta-core docs/api-mediated-access.md). + */ +import { promises as fs } from 'fs'; +import { dirname } from 'path'; + +export interface SSEEvent { + id: string; + event: string; + data: any; +} + +export interface SSEClientOptions { + url: string; + cursorPath: string | null; + onEvent: (e: SSEEvent) => Promise | void; + onGap?: (payload: { requested: string; resumeFrom: string; reason: string }) => Promise | void; + logTag?: string; + reconnectBaseMs?: number; + reconnectMaxMs?: number; + flushIntervalMs?: number; + /** When set, sent as Last-Event-ID on the *first* connect (e.g. to + * force replay from a specific point). After the first event the + * persisted cursor takes over. */ + initialCursor?: string; +} + +export class SSEEventClient { + private readonly opts: Required> & Pick; + private cursor: string = ''; + private cursorDirty = false; + private abort: AbortController | null = null; + private flushTimer: NodeJS.Timeout | null = null; + private stopped = false; + private backoff = 0; + private firstConnect = true; + + constructor(options: SSEClientOptions) { + this.opts = { + url: options.url, + cursorPath: options.cursorPath, + onEvent: options.onEvent, + onGap: options.onGap, + logTag: options.logTag ?? '[SSE]', + reconnectBaseMs: options.reconnectBaseMs ?? 500, + reconnectMaxMs: options.reconnectMaxMs ?? 30_000, + flushIntervalMs: options.flushIntervalMs ?? 2000, + initialCursor: options.initialCursor, + }; + this.backoff = this.opts.reconnectBaseMs; + } + + async start(): Promise { + await this.loadCursor(); + if (!this.cursor && this.opts.initialCursor) { + this.cursor = this.opts.initialCursor; + } + this.startFlushTimer(); + await this.runLoop(); + } + + async stop(): Promise { + this.stopped = true; + if (this.abort) { + this.abort.abort(); + this.abort = null; + } + if (this.flushTimer) { + clearInterval(this.flushTimer); + this.flushTimer = null; + } + await this.flushCursor(); + } + + getCursor(): string { + return this.cursor; + } + + private async loadCursor(): Promise { + if (!this.opts.cursorPath) return; + try { + const data = await fs.readFile(this.opts.cursorPath, 'utf-8'); + const trimmed = data.trim(); + if (trimmed) { + this.cursor = trimmed; + console.log(`${this.opts.logTag} Resuming from cursor ${this.cursor}`); + } + } catch (err: any) { + if (err.code !== 'ENOENT') { + console.warn(`${this.opts.logTag} Could not read cursor file ${this.opts.cursorPath}: ${err.message}`); + } + } + } + + private startFlushTimer(): void { + this.flushTimer = setInterval(() => { + if (this.cursorDirty) { + this.flushCursor().catch(err => + console.warn(`${this.opts.logTag} Cursor flush failed: ${err.message}`) + ); + } + }, this.opts.flushIntervalMs); + } + + private async flushCursor(): Promise { + if (!this.opts.cursorPath || !this.cursorDirty) return; + this.cursorDirty = false; + try { + await fs.mkdir(dirname(this.opts.cursorPath), { recursive: true }); + await fs.writeFile(this.opts.cursorPath, this.cursor, 'utf-8'); + } catch (err: any) { + this.cursorDirty = true; + throw err; + } + } + + private async runLoop(): Promise { + while (!this.stopped) { + try { + await this.connect(); + this.backoff = this.opts.reconnectBaseMs; + } catch (err: any) { + if (this.stopped) return; + console.warn(`${this.opts.logTag} Stream error, reconnecting in ${this.backoff}ms: ${err?.message ?? err}`); + await this.sleep(this.backoff); + this.backoff = Math.min(this.backoff * 2, this.opts.reconnectMaxMs); + } + } + } + + private async connect(): Promise { + this.abort = new AbortController(); + const headers: Record = { + Accept: 'text/event-stream', + 'Cache-Control': 'no-cache', + }; + if (this.cursor) { + headers['Last-Event-ID'] = this.cursor; + } + this.firstConnect = false; + + const response = await fetch(this.opts.url, { + method: 'GET', + headers, + signal: this.abort.signal, + }); + if (!response.ok) throw new Error(`HTTP ${response.status} ${response.statusText}`); + if (!response.body) throw new Error('Response has no body'); + + const reader = response.body.getReader(); + const decoder = new TextDecoder('utf-8'); + let buffer = ''; + + try { + while (!this.stopped) { + const { value, done } = await reader.read(); + if (done) return; + buffer += decoder.decode(value, { stream: true }); + let idx: number; + while ((idx = buffer.indexOf('\n\n')) !== -1) { + const rawChunk = buffer.slice(0, idx); + buffer = buffer.slice(idx + 2); + await this.handleChunk(rawChunk); + } + } + } finally { + try { reader.releaseLock(); } catch { /* nothing useful */ } + } + } + + private async handleChunk(chunk: string): Promise { + let id = ''; + let event = ''; + const dataLines: string[] = []; + for (const rawLine of chunk.split('\n')) { + if (!rawLine || rawLine.startsWith(':')) continue; + const colon = rawLine.indexOf(':'); + if (colon === -1) continue; + const field = rawLine.slice(0, colon); + let value = rawLine.slice(colon + 1); + if (value.startsWith(' ')) value = value.slice(1); + switch (field) { + case 'id': id = value; break; + case 'event': event = value; break; + case 'data': dataLines.push(value); break; + } + } + if (dataLines.length === 0) return; + + let data: any; + try { + data = JSON.parse(dataLines.join('\n')); + } catch (err: any) { + console.warn(`${this.opts.logTag} Skipping unparseable event ${id || '(no id)'}: ${err.message}`); + return; + } + + const ev: SSEEvent = { id, event: event || 'message', data }; + if (ev.event === 'gap') { + if (this.opts.onGap) { + await this.opts.onGap(ev.data); + } else { + console.warn(`${this.opts.logTag} Gap event (cursor ${ev.data?.requested} trimmed): resumed from ${ev.data?.resumeFrom}`); + } + if (typeof ev.data?.resumeFrom === 'string' && ev.data.resumeFrom) { + this.cursor = ev.data.resumeFrom; + this.cursorDirty = true; + } + return; + } + + try { + await this.opts.onEvent(ev); + } catch (err: any) { + console.error(`${this.opts.logTag} onEvent handler failed for ${ev.id}: ${err?.stack ?? err}`); + } + if (ev.id) { + this.cursor = ev.id; + this.cursorDirty = true; + } + } + + private sleep(ms: number): Promise { + return new Promise(resolve => setTimeout(resolve, ms)); + } +}