From 53c4463084a3359f0fa55edfb4c6e5a470daabdb Mon Sep 17 00:00:00 2001 From: Pierre Date: Sat, 24 Jan 2026 16:20:58 +0100 Subject: [PATCH] background queue fix --- .../src/container-plugins/ContainerManager.ts | 97 ++++++++++++++++++- .../ContainerPluginScheduler.ts | 63 +++++++++++- packages/meta-sort-core/src/index.ts | 6 +- packages/meta-sort-core/src/kv/KVManager.ts | 8 +- .../src/logic/WatchedFileProcessor.ts | 4 +- .../src/logic/pipeline/StreamingPipeline.ts | 4 +- 6 files changed, 170 insertions(+), 12 deletions(-) diff --git a/packages/meta-sort-core/src/container-plugins/ContainerManager.ts b/packages/meta-sort-core/src/container-plugins/ContainerManager.ts index 7f56f41..d782abb 100644 --- a/packages/meta-sort-core/src/container-plugins/ContainerManager.ts +++ b/packages/meta-sort-core/src/container-plugins/ContainerManager.ts @@ -7,6 +7,7 @@ import { EventEmitter } from 'events'; import { promises as fs } from 'fs'; +import { join } from 'path'; import { DockerClient } from './DockerClient.js'; import { loadConfig, @@ -25,6 +26,15 @@ import type { } from './types.js'; import { config } from '../config/EnvConfig.js'; +/** + * Service info structure from meta-core service discovery + */ +interface ServiceInfo { + name: string; + api: string; + endpoints: Record; +} + /** * Container manager events */ @@ -84,25 +94,78 @@ export class ContainerManager extends EventEmitter { private shutdownInProgress: boolean = false; private webdavUrl?: string; private stackName?: string; + private callbackUrl: string; + private metaCoreUrl: string; constructor( private configPath: string = config.CONTAINER_PLUGINS_CONFIG, private network: string = config.CONTAINER_NETWORK, private filesPath: string = config.FILES_PATH, - private callbackUrl: string = config.CONTAINER_CALLBACK_URL, - private metaCoreUrl: string = config.CONTAINER_META_CORE_URL, + callbackUrl?: string, + metaCoreUrl?: string, dockerClient?: DockerClient ) { super(); this.dockerClient = dockerClient || new DockerClient(config.DOCKER_SOCKET_PATH); - // WebDAV configuration for plugin file access + // Initialize with env vars as fallback (will be overridden by service discovery) + this.callbackUrl = callbackUrl ?? config.CONTAINER_CALLBACK_URL; + this.metaCoreUrl = metaCoreUrl ?? config.CONTAINER_META_CORE_URL; this.webdavUrl = config.PLUGIN_WEBDAV_URL; // Configure stack name for Docker Desktop grouping this.stackName = config.PLUGIN_STACK_NAME; } + /** + * Discover service URLs from meta-core service discovery file + * Falls back to environment variables if discovery fails + */ + private async discoverServiceUrls(): Promise { + const serviceFile = join(config.META_CORE_PATH, 'services', 'meta-sort.json'); + + try { + const content = await fs.readFile(serviceFile, 'utf-8'); + const serviceInfo: ServiceInfo = JSON.parse(content); + + // Extract URLs from service info + const api = serviceInfo.api; + const endpoints = serviceInfo.endpoints || {}; + + // Update URLs from service discovery + if (endpoints.callback) { + this.callbackUrl = endpoints.callback; + console.log(`[ContainerManager] Discovered callback URL: ${this.callbackUrl}`); + } else if (api) { + // Fallback: construct from api base + this.callbackUrl = `${api}/api/plugins/callback`; + console.log(`[ContainerManager] Constructed callback URL: ${this.callbackUrl}`); + } + + if (endpoints.webdav) { + this.webdavUrl = endpoints.webdav; + console.log(`[ContainerManager] Discovered WebDAV URL: ${this.webdavUrl}`); + } else if (api) { + // Fallback: construct from api base + this.webdavUrl = `${api}/webdav`; + console.log(`[ContainerManager] Constructed WebDAV URL: ${this.webdavUrl}`); + } + + if (endpoints.health) { + // Extract meta-core URL from health endpoint + const healthUrl = new URL(endpoints.health); + this.metaCoreUrl = `${healthUrl.protocol}//${healthUrl.host}`; + console.log(`[ContainerManager] Discovered meta-core URL: ${this.metaCoreUrl}`); + } + + console.log('[ContainerManager] Service discovery successful'); + } catch (error) { + console.warn(`[ContainerManager] Service discovery failed, using env vars: ${error}`); + console.log(`[ContainerManager] Using fallback callback URL: ${this.callbackUrl}`); + console.log(`[ContainerManager] Using fallback WebDAV URL: ${this.webdavUrl || 'not set'}`); + } + } + /** * Initialize the container manager */ @@ -113,6 +176,9 @@ export class ContainerManager extends EventEmitter { console.log('[ContainerManager] Initializing...'); + // Discover service URLs from meta-core service discovery + await this.discoverServiceUrls(); + // Log WebDAV configuration console.log('[ContainerManager] Plugin file access configuration (WebDAV):'); if (this.webdavUrl) { @@ -122,8 +188,8 @@ export class ContainerManager extends EventEmitter { console.log(' /files/plugin/ - Plugin output (per-plugin subdirs)'); console.log(' /files/corn - SMB mounts (visible via WebDAV)'); } else { - console.warn(' WebDAV not configured. Set PLUGIN_WEBDAV_URL.'); - console.warn(' Plugins will not have access to files until WebDAV is configured.'); + console.warn(' WebDAV not configured.'); + console.warn(' Plugins will not have access to files.'); } // Initialize Docker client @@ -914,6 +980,27 @@ export class ContainerManager extends EventEmitter { getPluginsConfig(): ContainerPluginsConfig | null { return this.pluginsConfig; } + + /** + * Get the discovered callback URL for plugins + */ + getCallbackUrl(): string { + return this.callbackUrl; + } + + /** + * Get the discovered meta-core URL + */ + getMetaCoreUrl(): string { + return this.metaCoreUrl; + } + + /** + * Get the discovered WebDAV URL for file access + */ + getWebdavUrl(): string | undefined { + return this.webdavUrl; + } } /** diff --git a/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts b/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts index 69ed4fc..23a29b4 100644 --- a/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts +++ b/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts @@ -76,6 +76,7 @@ export class ContainerPluginScheduler extends EventEmitter { containerManager: ContainerManager, options?: { fastConcurrency?: number; + backgroundConcurrency?: number; callbackUrl?: string; metaCoreUrl?: string; defaultTimeout?: number; @@ -90,9 +91,9 @@ export class ContainerPluginScheduler extends EventEmitter { }); // Background queue starts paused - only runs when fast queue is idle - // Concurrency is unlimited - actual limit is determined by plugin instances + // Default to same concurrency as fast queue if not specified this.backgroundQueue = new PQueue({ - concurrency: Infinity, + concurrency: options?.backgroundConcurrency ?? config.FAST_QUEUE_CONCURRENCY, autoStart: false, }); @@ -101,12 +102,32 @@ export class ContainerPluginScheduler extends EventEmitter { this.onFastQueueIdle(); }); + // Initialize with env vars as fallback (will be updated from ContainerManager after discovery) this.callbackUrl = options?.callbackUrl ?? `${config.CONTAINER_CALLBACK_URL}/api/plugins/callback`; this.metaCoreUrl = options?.metaCoreUrl ?? config.CONTAINER_META_CORE_URL; this.defaultTimeout = options?.defaultTimeout ?? 60000; this.kvClient = options?.kvClient ?? null; } + /** + * Update URLs from ContainerManager after service discovery + * Call this after ContainerManager.initialize() completes + */ + updateUrlsFromManager(): void { + const managerCallbackUrl = this.containerManager.getCallbackUrl(); + const managerMetaCoreUrl = this.containerManager.getMetaCoreUrl(); + + if (managerCallbackUrl) { + this.callbackUrl = managerCallbackUrl; + console.log(`[ContainerPluginScheduler] Updated callback URL: ${this.callbackUrl}`); + } + + if (managerMetaCoreUrl) { + this.metaCoreUrl = managerMetaCoreUrl; + console.log(`[ContainerPluginScheduler] Updated meta-core URL: ${this.metaCoreUrl}`); + } + } + /** * Wait for dependency plugins to complete for a file */ @@ -572,16 +593,47 @@ export class ContainerPluginScheduler extends EventEmitter { background: { pending: number; running: number; waiting: number }; pendingTasks: number; pendingCallbacks: number; + // File-level counts (what users care about) + files: { + fastRunning: number; // unique files with fast tasks running + fastWaiting: number; // unique files with fast tasks waiting + backgroundRunning: number; // unique files with background tasks running + backgroundWaiting: number; // unique files with background tasks waiting + }; } { // Count pending callbacks by queue type // These are tasks dispatched to containers awaiting completion - the real "running" count let fastCallbacks = 0; let backgroundCallbacks = 0; + const fastRunningFiles = new Set(); + const backgroundRunningFiles = new Set(); + for (const callback of this.pendingCallbacks.values()) { if (callback.task.queue === 'fast') { fastCallbacks++; + fastRunningFiles.add(callback.task.fileHash); } else { backgroundCallbacks++; + backgroundRunningFiles.add(callback.task.fileHash); + } + } + + // Count unique files with waiting tasks + const fastWaitingFiles = new Set(); + const backgroundWaitingFiles = new Set(); + + for (const task of this.pendingTasks.values()) { + if (task.status === 'pending') { + if (task.queue === 'fast') { + // Only count as waiting if not already running + if (!fastRunningFiles.has(task.fileHash)) { + fastWaitingFiles.add(task.fileHash); + } + } else { + if (!backgroundRunningFiles.has(task.fileHash)) { + backgroundWaitingFiles.add(task.fileHash); + } + } } } @@ -600,6 +652,12 @@ export class ContainerPluginScheduler extends EventEmitter { }, pendingTasks: this.pendingTasks.size, pendingCallbacks: this.pendingCallbacks.size, + files: { + fastRunning: fastRunningFiles.size, + fastWaiting: fastWaitingFiles.size, + backgroundRunning: backgroundRunningFiles.size, + backgroundWaiting: backgroundWaitingFiles.size, + }, }; } @@ -823,6 +881,7 @@ export function createContainerPluginScheduler( containerManager: ContainerManager, options?: { fastConcurrency?: number; + backgroundConcurrency?: number; callbackUrl?: string; metaCoreUrl?: string; kvClient?: IKVClient; diff --git a/packages/meta-sort-core/src/index.ts b/packages/meta-sort-core/src/index.ts index 96d81b6..b32fa99 100644 --- a/packages/meta-sort-core/src/index.ts +++ b/packages/meta-sort-core/src/index.ts @@ -205,7 +205,7 @@ process.on('SIGINT', async () => { port: config.FUSE_API_PORT, host: config.FUSE_API_HOST, enableCors: true, - }, unifiedStateManager, getDuplicateResult, kvClient || undefined, triggerScan, undefined, fastQueueConcurrency, () => fileProcessor.getQueueStatus(), kvManager || undefined, getPluginManagerInstance, () => fileProcessor.getTaskScheduler()); + }, unifiedStateManager, getDuplicateResult, kvClient || undefined, triggerScan, backgroundQueueConcurrency, fastQueueConcurrency, () => fileProcessor.getQueueStatus(), kvManager || undefined, getPluginManagerInstance, () => fileProcessor.getTaskScheduler()); await apiServer.start(); console.log('Unified API Server started successfully'); @@ -229,9 +229,13 @@ process.on('SIGINT', async () => { // Create container plugin scheduler containerPluginScheduler = new ContainerPluginScheduler(containerManager, { fastConcurrency: config.FAST_QUEUE_CONCURRENCY, + backgroundConcurrency: backgroundQueueConcurrency, kvClient: kvClient || undefined, }); + // Update scheduler URLs from service discovery + containerPluginScheduler.updateUrlsFromManager(); + // Connect to API server (for callback handling) apiServer.setContainerPluginManager(containerManager, containerPluginScheduler); console.log('[Startup] ContainerPluginScheduler connected to API server'); diff --git a/packages/meta-sort-core/src/kv/KVManager.ts b/packages/meta-sort-core/src/kv/KVManager.ts index cf8aa55..94f8656 100644 --- a/packages/meta-sort-core/src/kv/KVManager.ts +++ b/packages/meta-sort-core/src/kv/KVManager.ts @@ -135,7 +135,10 @@ export class KVManager { console.log(`[KVManager] Using direct Redis URL: ${this.config.redisUrl}`); this.kvClient = await this.createRedisClient(this.config.redisUrl); - // Initialize service discovery even with direct Redis URL + // NOTE: Service discovery is handled by Go meta-core sidecar + // The Go binary writes to /meta-core/services/{serviceName}.json with correct internal IPs + // TypeScript ServiceDiscovery would overwrite with external BASE_URL, breaking internal discovery + // Keep ServiceDiscovery instance for reading/discovery only, but don't call start() const apiUrl = this.config.baseUrl || `http://localhost:${this.config.apiPort}`; this.serviceDiscovery = new ServiceDiscovery({ metaCorePath: this.config.metaCorePath, @@ -151,7 +154,8 @@ export class KVManager { processing: '/api/processing/status' } }); - await this.serviceDiscovery.start(); + // Don't call start() - Go meta-core handles registration + // await this.serviceDiscovery.start(); this.isStarted = true; this.notifyReady(); diff --git a/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts b/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts index 23e1bb7..274b8dd 100644 --- a/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts +++ b/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts @@ -147,7 +147,9 @@ export class WatchedFileProcessor implements FileProcessorInterface { pending: queues.backgroundQueue.size, // waiting in queue size: queues.backgroundQueue.size, isPaused: false - } + }, + // File-level counts (what users care about) + files: queues.files }; return result; } diff --git a/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts b/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts index 131f7bd..e5998ba 100644 --- a/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts +++ b/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts @@ -395,7 +395,9 @@ export class StreamingPipeline { size: this.validationQueue.size }, fastQueue, - backgroundQueue + backgroundQueue, + // File-level counts from container scheduler (what users care about) + files: containerQueueStatus?.files }, state: this.config.stateManager.getSnapshot() };