metastar/meta-fuse
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

bugfix event

Browse Source
This commit is contained in:
Pierre
2026-03-07 14:44:27 +01:00
parent c2420aa1fa
commit e384d4b1b5
6 changed files with 74 additions and 219 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/meta-fuse-core/src/api/APIServer.ts
+3 -64
View File
@@ -85,10 +85,7 @@ export class APIServer {
this.app.post('/api/fuse/rules/validate', this.handleValidateRule.bind(this));
this.app.get('/api/fuse/rules/variables', this.handleGetVariables.bind(this));
// Service discovery (for inter-service navigation)
if (this.kvManager) {
this.app.get('/api/services', this.handleServices.bind(this));
}
// Note: /api/services is handled by nginx proxy to meta-core (centralized service discovery)
}
/**
@@ -162,7 +159,7 @@ export class APIServer {
return;
}
const attrs = this.vfs.getattr(path);
const attrs = await this.vfs.getattr(path);
if (attrs === null) {
reply.status(404).send({ error: 'Path not found' });
@@ -198,7 +195,7 @@ export class APIServer {
return;
}
const result = this.vfs.read(path);
const result = await this.vfs.read(path);
if (result === null) {
reply.status(404).send({ error: 'File not found' });
@@ -261,64 +258,6 @@ export class APIServer {
reply.send({ status: 'ok' });
}
private async handleServices(_req: FastifyRequest, reply: FastifyReply): Promise<void> {
interface ServiceResponse {
name: string;
url: string;
api: string;
status: string;
role?: string;
}
const services: ServiceResponse[] = [];
let leaderInfo: {
hostname: string;
baseUrl: string;
apiUrl: string;
redisUrl: string;
webdavUrl: string;
} | null = null;
try {
const serviceDiscovery = this.kvManager?.getServiceDiscovery();
if (serviceDiscovery) {
const allServices = await serviceDiscovery.discoverAllServices();
for (const svc of allServices) {
services.push({
name: svc.name || 'Unknown',
url: svc.baseUrl || '',
api: svc.baseUrl || '',
status: svc.status || 'unknown',
role: svc.role,
});
}
}
} catch (error) {
logger.error('Error discovering services:', error);
}
// Get leader info
if (this.kvManager) {
const info = this.kvManager.getLeaderInfo();
if (info) {
leaderInfo = {
hostname: info.hostname,
baseUrl: info.baseUrl,
apiUrl: info.apiUrl,
redisUrl: info.redisUrl,
webdavUrl: info.webdavUrl,
};
}
}
reply.send({
services,
current: 'meta-fuse',
leader: leaderInfo,
});
}
// ============================================
// Renaming Rules API Handlers
// ============================================
packages/meta-fuse-core/src/kv/KVManager.ts
+12 -12
View File
@@ -13,7 +13,7 @@
import { hostname } from 'os';
import { Logger } from 'tslog';
import { LeaderClient } from './LeaderClient.js';
import { ServiceDiscovery } from './ServiceDiscovery.js';
import { ServiceRegistration } from './ServiceRegistration.js';
import { RedisClient } from './RedisClient.js';
import type { LeaderLockInfo } from './IKVClient.js';
@@ -48,7 +48,7 @@ interface KVManagerConfig {
export class KVManager {
private config: Required<KVManagerConfig>;
private leaderClient: LeaderClient | null = null;
private serviceDiscovery: ServiceDiscovery | null = null;
private serviceRegistration: ServiceRegistration | null = null;
private redisClient: RedisClient | null = null;
private isStarted = false;
private isShuttingDown = false;
@@ -80,14 +80,14 @@ export class KVManager {
logger.info(`Starting KVManager for ${this.config.serviceName}...`);
// Initialize and start service discovery for registration and heartbeat
// Initialize and start service registration for heartbeat
const apiUrl = this.config.baseUrl || `http://${hostname()}:${this.config.apiPort}`;
this.serviceDiscovery = new ServiceDiscovery({
this.serviceRegistration = new ServiceRegistration({
metaCorePath: this.config.metaCorePath,
serviceName: this.config.serviceName,
apiUrl
});
await this.serviceDiscovery.start();
await this.serviceRegistration.start();
// If direct Redis URL provided, skip leader discovery
if (this.config.redisUrl) {
@@ -162,10 +162,10 @@ export class KVManager {
this.leaderClient = null;
}
// Stop service discovery (if it was started)
if (this.serviceDiscovery) {
await this.serviceDiscovery.stop();
this.serviceDiscovery = null;
// Stop service registration (if it was started)
if (this.serviceRegistration) {
await this.serviceRegistration.stop();
this.serviceRegistration = null;
}
// Disconnect Redis
@@ -341,10 +341,10 @@ export class KVManager {
}
/**
* Get service discovery (for inter-service navigation)
* Get service registration instance
*/
getServiceDiscovery(): ServiceDiscovery | null {
return this.serviceDiscovery;
getServiceRegistration(): ServiceRegistration | null {
return this.serviceRegistration;
}
/**
packages/meta-fuse-core/src/kv/ServiceDiscovery.ts → packages/meta-fuse-core/src/kv/ServiceRegistration.ts
+8 -128
View File
@@ -1,11 +1,11 @@
/**
* Service Discovery using shared filesystem
* Service Registration using shared filesystem
*
* 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 stale.
* Service discovery is centralized in meta-core - this class only handles registration.
* For service discovery, use the /api/services endpoint (proxied to meta-core via nginx).
*
* Each service must call start() to register itself and begin heartbeat updates.
* meta-core runs a cleanup process that removes services with stale heartbeats.
@@ -13,13 +13,12 @@
import { promises as fs } from 'fs';
import { hostname } from 'os';
import { join } from 'path';
import { Logger } from 'tslog';
import type { ServiceInfo } from './IKVClient.js';
const logger = new Logger({ name: 'ServiceDiscovery' });
const logger = new Logger({ name: 'ServiceRegistration' });
interface ServiceDiscoveryConfig {
interface ServiceRegistrationConfig {
/** Path to META_CORE_VOLUME (e.g., /meta-core) */
metaCorePath: string;
@@ -36,8 +35,8 @@ interface ServiceDiscoveryConfig {
staleThreshold?: number;
}
export class ServiceDiscovery {
private config: ServiceDiscoveryConfig;
export class ServiceRegistration {
private config: ServiceRegistrationConfig;
private servicesDir: string;
private serviceFilePath: string;
private heartbeatTimer: NodeJS.Timeout | null = null;
@@ -45,7 +44,7 @@ export class ServiceDiscovery {
private isStarted = false;
private currentHostname: string;
constructor(config: ServiceDiscoveryConfig) {
constructor(config: ServiceRegistrationConfig) {
this.config = {
heartbeatInterval: 30000,
staleThreshold: 60000,
@@ -167,125 +166,6 @@ export class ServiceDiscovery {
}
}
/**
* 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<ServiceInfo | null> {
try {
// 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;
if (this.isStale(service)) {
service.status = 'stale';
}
return service.status === 'running' ? service : null;
} catch {
// Try hostname-based files
}
// 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
}
throw error;
}
}
/**
* Discover all registered services
*/
async discoverAllServices(): Promise<ServiceInfo[]> {
try {
const files = await fs.readdir(this.servicesDir);
const services: ServiceInfo[] = [];
for (const file of files) {
if (!file.endsWith('.json')) continue;
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';
}
services.push(service);
} catch (error) {
logger.error(`Failed to read ${file}:`, error);
}
}
return services;
} catch (error: any) {
if (error.code === 'ENOENT') {
return []; // Services directory doesn't exist yet
}
throw error;
}
}
/**
* Check if a service is healthy
*/
async isServiceHealthy(name: string): Promise<boolean> {
const service = await this.discoverService(name);
if (!service) return false;
// Try to ping the service's health endpoint
const url = `${service.baseUrl}/health`;
try {
const response = await fetch(url, {
signal: AbortSignal.timeout(5000)
});
return response.ok;
} catch {
return false;
}
}
// ========================================================================
// Lifecycle
// ========================================================================
packages/meta-fuse-core/src/vfs/RulesPropertyExtractor.ts
+1
View File
@@ -30,6 +30,7 @@ const CORE_PROPERTIES = new Set([
'filePath', // Source file location (must exist for file to appear in VFS)
'size', // File size attribute
'fileSize', // Alternative name for size
'sizeByte', // Alternative name for size (used by meta-sort plugins)
'mtime', // Modification time
'ctime', // Creation time
'fileName', // Filename for fallback naming
packages/meta-fuse-core/src/vfs/VirtualFileSystem.ts
+42 -2
View File
@@ -667,6 +667,38 @@ export class VirtualFileSystem implements VFSUpdateCallback {
this.cachedStats.totalSize += size;
}
/**
* Ensure file has a valid size, fetching via WebDAV if needed
* This is called lazily when getattr() or read() is invoked
*/
private async ensureFileSize(node: VFSNode): Promise<void> {
// Only fetch if size is 0/missing and we have WebDAV configured
if (node.size && node.size > 0) {
return;
}
if (!node.sourcePath || !this.config.webdavBaseUrl) {
return;
}
logger.debug(`Fetching file size via WebDAV for: ${node.sourcePath}`);
const stats = await this.statViaWebDAV(node.sourcePath);
if (stats && stats.size > 0) {
// Update the node with the fetched size
const oldSize = node.size ?? 0;
node.size = stats.size;
if (stats.mtime) {
node.mtime = stats.mtime;
}
// Update total size stats
this.cachedStats.totalSize = this.cachedStats.totalSize - oldSize + stats.size;
logger.debug(`WebDAV size fetch successful: ${node.sourcePath} = ${stats.size} bytes`);
} else {
logger.warn(`WebDAV size fetch failed for: ${node.sourcePath}`);
}
}
/**
* Get file stats via WebDAV HEAD request
*/
@@ -841,8 +873,9 @@ export class VirtualFileSystem implements VFSUpdateCallback {
/**
* Get file attributes
* Lazily fetches file size via WebDAV if missing
*/
getattr(filepath: string): FileAttributes | null {
async getattr(filepath: string): Promise<FileAttributes | null> {
filepath = this.normalizePath(filepath);
const node = this.nodes.get(filepath);
@@ -862,6 +895,9 @@ export class VirtualFileSystem implements VFSUpdateCallback {
gid: this.config.gid,
};
} else {
// Lazily fetch size via WebDAV if missing
await this.ensureFileSize(node);
return {
size: node.size!,
mode: 0o100000 | this.config.fileMode,
@@ -885,8 +921,9 @@ export class VirtualFileSystem implements VFSUpdateCallback {
/**
* Read file (returns source path and/or WebDAV URL for actual reading)
* Lazily fetches file size via WebDAV if missing
*/
read(filepath: string): ReadResult | null {
async read(filepath: string): Promise<ReadResult | null> {
filepath = this.normalizePath(filepath);
const node = this.nodes.get(filepath);
@@ -894,6 +931,9 @@ export class VirtualFileSystem implements VFSUpdateCallback {
return null;
}
// Lazily fetch size via WebDAV if missing
await this.ensureFileSize(node);
// Compute WebDAV URL if configured
let webdavUrl: string | null = null;
if (this.config.webdavBaseUrl && node.sourcePath) {