From 133e6bc61cbb5dae1a0a5e9b58453ad9299b4669 Mon Sep 17 00:00:00 2001 From: Pierre Date: Tue, 20 Jan 2026 22:39:13 +0100 Subject: [PATCH] streming pipeline improvement --- .../meta-sort-core/src/config/EnvConfig.ts | 4 -- .../ContainerPluginScheduler.ts | 50 ++++++++++--- packages/meta-sort-core/src/index.ts | 3 +- .../src/logic/WatchedFileProcessor.ts | 2 - .../fileProcessor/FileProcessorPiscina.ts | 12 ++-- .../src/logic/pipeline/StreamingPipeline.ts | 70 ++++++++++++++++--- .../src/plugin-engine/PluginManager.ts | 8 ++- .../src/plugin-engine/TaskScheduler.ts | 16 ++++- 8 files changed, 125 insertions(+), 40 deletions(-) diff --git a/packages/meta-sort-core/src/config/EnvConfig.ts b/packages/meta-sort-core/src/config/EnvConfig.ts index 0b3ca62..423c82a 100644 --- a/packages/meta-sort-core/src/config/EnvConfig.ts +++ b/packages/meta-sort-core/src/config/EnvConfig.ts @@ -83,9 +83,6 @@ interface EnvConfig { /** Concurrency for fast plugin queue. Default: 32 */ FAST_QUEUE_CONCURRENCY: number; - /** Concurrency for background plugin queue. Default: 8 */ - BACKGROUND_QUEUE_CONCURRENCY: number; - // ======================================================================== // Container Plugins Configuration // All plugins run as Docker containers for isolation and flexibility @@ -187,7 +184,6 @@ export const config: EnvConfig = { // TaskScheduler Configuration FAST_QUEUE_CONCURRENCY: parseInt(process.env.FAST_QUEUE_CONCURRENCY || "32", 10), - BACKGROUND_QUEUE_CONCURRENCY: parseInt(process.env.BACKGROUND_QUEUE_CONCURRENCY || "8", 10), // Container Plugins Configuration CONTAINER_PLUGINS_CONFIG: process.env.CONTAINER_PLUGINS_CONFIG || '/app/plugins.yml', diff --git a/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts b/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts index 8391522..69ed4fc 100644 --- a/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts +++ b/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts @@ -76,7 +76,6 @@ export class ContainerPluginScheduler extends EventEmitter { containerManager: ContainerManager, options?: { fastConcurrency?: number; - backgroundConcurrency?: number; callbackUrl?: string; metaCoreUrl?: string; defaultTimeout?: number; @@ -91,8 +90,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 this.backgroundQueue = new PQueue({ - concurrency: options?.backgroundConcurrency ?? config.BACKGROUND_QUEUE_CONCURRENCY, + concurrency: Infinity, autoStart: false, }); @@ -285,6 +285,7 @@ export class ContainerPluginScheduler extends EventEmitter { * Called when fast queue becomes idle - start background queue if it has work */ private onFastQueueIdle(): void { + console.log(`[ContainerScheduler] onFastQueueIdle: bgQueue.size=${this.backgroundQueue.size}, bgQueue.pending=${this.backgroundQueue.pending}, bgQueue.isPaused=${this.backgroundQueue.isPaused}`); if (this.backgroundQueue.size > 0 || this.backgroundQueue.pending > 0) { console.log('[ContainerScheduler] Fast queue idle, starting background queue'); this.backgroundQueue.start(); @@ -295,6 +296,8 @@ export class ContainerPluginScheduler extends EventEmitter { * Execute a task (dispatch to container) */ private async executeTask(task: ContainerTask): Promise { + console.log(`[ContainerScheduler] executeTask: ${task.pluginId} for ${task.fileHash} (queue: ${task.queue})`); + // Wait for dependencies to complete first if (task.dependencies && task.dependencies.length > 0) { task.status = 'waiting'; @@ -302,6 +305,7 @@ export class ContainerPluginScheduler extends EventEmitter { } const instance = this.containerManager.getHealthyInstance(task.pluginId); + console.log(`[ContainerScheduler] getHealthyInstance(${task.pluginId}): ${instance ? instance.containerName : 'null'}`); if (!instance) { task.status = 'failed'; @@ -320,7 +324,9 @@ export class ContainerPluginScheduler extends EventEmitter { try { // Fetch fresh metadata after dependencies complete // This ensures we get metadata written by dependency plugins + console.log(`[ContainerScheduler] Fetching metadata for ${task.fileHash}...`); const existingMeta = await this.fetchExistingMetadata(task.fileHash); + console.log(`[ContainerScheduler] Dispatching to ${instance.baseUrl}/process for ${task.pluginId}`); // Build process request const request: PluginProcessRequest = { @@ -334,6 +340,7 @@ export class ContainerPluginScheduler extends EventEmitter { // Dispatch to container const response = await this.dispatchToContainer(instance.baseUrl, request); + console.log(`[ContainerScheduler] dispatchToContainer result for ${task.pluginId}: ${response.status}`); if (response.status !== 'accepted') { throw new Error(response.error || 'Task rejected by plugin'); @@ -346,6 +353,7 @@ export class ContainerPluginScheduler extends EventEmitter { await this.waitForCallback(task, timeout); } catch (error) { + console.error(`[ContainerScheduler] Task ${task.pluginId} failed:`, error); task.status = 'failed'; task.error = error instanceof Error ? error.message : String(error); task.completedAt = Date.now(); @@ -489,14 +497,18 @@ export class ContainerPluginScheduler extends EventEmitter { */ private checkFileCompletion(fileHash: string): void { const taskIds = this.fileTasks.get(fileHash); - if (!taskIds) return; + if (!taskIds) { + return; + } let allComplete = true; let filePath: string | undefined; for (const taskId of taskIds) { const task = this.pendingTasks.get(taskId); - if (!task) continue; + if (!task) { + continue; + } filePath = task.filePath; @@ -553,21 +565,38 @@ export class ContainerPluginScheduler extends EventEmitter { /** * Get queue status + * PQueue: .pending = running + waiting, .size = waiting only */ getQueueStatus(): { - fast: { pending: number; running: number }; - background: { pending: number; running: number }; + fast: { pending: number; running: number; waiting: number }; + background: { pending: number; running: number; waiting: number }; pendingTasks: number; pendingCallbacks: number; } { + // Count pending callbacks by queue type + // These are tasks dispatched to containers awaiting completion - the real "running" count + let fastCallbacks = 0; + let backgroundCallbacks = 0; + for (const callback of this.pendingCallbacks.values()) { + if (callback.task.queue === 'fast') { + fastCallbacks++; + } else { + backgroundCallbacks++; + } + } + + // PQueue .size = waiting in queue, .pending = running + waiting + // But PQueue "running" is just dispatching HTTP requests - actual running is callbacks return { fast: { - pending: this.fastQueue.pending, - running: this.fastQueue.pending - this.fastQueue.size, + pending: this.fastQueue.size + fastCallbacks, // waiting + running in containers + running: fastCallbacks, // tasks running in containers + waiting: this.fastQueue.size, }, background: { - pending: this.backgroundQueue.pending, - running: this.backgroundQueue.pending - this.backgroundQueue.size, + pending: this.backgroundQueue.size + backgroundCallbacks, // waiting + running in containers + running: backgroundCallbacks, // tasks running in containers + waiting: this.backgroundQueue.size, }, pendingTasks: this.pendingTasks.size, pendingCallbacks: this.pendingCallbacks.size, @@ -794,7 +823,6 @@ 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 4e02ef7..96d81b6 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, backgroundQueueConcurrency, fastQueueConcurrency, () => fileProcessor.getQueueStatus(), kvManager || undefined, getPluginManagerInstance, () => fileProcessor.getTaskScheduler()); + }, unifiedStateManager, getDuplicateResult, kvClient || undefined, triggerScan, undefined, fastQueueConcurrency, () => fileProcessor.getQueueStatus(), kvManager || undefined, getPluginManagerInstance, () => fileProcessor.getTaskScheduler()); await apiServer.start(); console.log('Unified API Server started successfully'); @@ -229,7 +229,6 @@ process.on('SIGINT', async () => { // Create container plugin scheduler containerPluginScheduler = new ContainerPluginScheduler(containerManager, { fastConcurrency: config.FAST_QUEUE_CONCURRENCY, - backgroundConcurrency: config.BACKGROUND_QUEUE_CONCURRENCY, kvClient: kvClient || undefined, }); diff --git a/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts b/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts index 871bc03..23e1bb7 100644 --- a/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts +++ b/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts @@ -198,7 +198,6 @@ export class WatchedFileProcessor implements FileProcessorInterface { const pluginManager = await initializePluginManager(); this.taskScheduler = createTaskScheduler(pluginManager, performanceMetrics, { fastQueueConcurrency: config.FAST_QUEUE_CONCURRENCY, - backgroundQueueConcurrency: config.BACKGROUND_QUEUE_CONCURRENCY, }); // Set up event listeners for VFS updates @@ -206,7 +205,6 @@ export class WatchedFileProcessor implements FileProcessorInterface { console.log(`[WatchedFileProcessor] TaskScheduler initialized with:`); console.log(` - Fast queue concurrency: ${config.FAST_QUEUE_CONCURRENCY}`); - console.log(` - Background queue concurrency: ${config.BACKGROUND_QUEUE_CONCURRENCY}`); } /** diff --git a/packages/meta-sort-core/src/logic/fileProcessor/FileProcessorPiscina.ts b/packages/meta-sort-core/src/logic/fileProcessor/FileProcessorPiscina.ts index ff640d7..c8d2c63 100644 --- a/packages/meta-sort-core/src/logic/fileProcessor/FileProcessorPiscina.ts +++ b/packages/meta-sort-core/src/logic/fileProcessor/FileProcessorPiscina.ts @@ -521,14 +521,10 @@ export class FileProcessorPiscina implements FileAnalyzerInterface{ } } - // STATE: Hash processing complete, move to done - if (this.unifiedStateManager) { - this.unifiedStateManager.completeHashProcessing( - filePath, - midHash256, - virtualPath - ); - } + // NOTE: File completion is handled by StreamingPipeline based on whether + // container plugins are configured. If container plugins exist, we wait for + // the 'file:complete' event from ContainerPluginScheduler. If not, StreamingPipeline + // calls completeHashProcessing() directly after this method returns. } catch (e) { // Handle abort errors gracefully if (e instanceof Error && (e.message === 'AbortError' || e.name === 'AbortError')) { diff --git a/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts b/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts index 378d28f..131f7bd 100644 --- a/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts +++ b/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts @@ -273,7 +273,7 @@ export class StreamingPipeline { * Stage 3: Background Queue Phase * - Compute full hashes (SHA-256, SHA-1, MD5, CRC32) * - Update KV with additional hash metadata - * - Mark processing as complete + * - Mark processing as complete (if no container plugins) */ private async processHashPhase(filePath: string): Promise { try { @@ -284,6 +284,20 @@ export class StreamingPipeline { await this.config.fileProcessor.processHashPhase(filePath, current, queueSize); this.backgroundProcessedCount++; + + // If container plugins are NOT configured, mark file as complete immediately. + // Otherwise, wait for 'file:complete' event from ContainerPluginScheduler. + if (!this.config.containerPluginScheduler) { + const metadata = this.config.fileProcessor.getDatabase().get(filePath); + const hashId = metadata?.cid_midhash256; + let virtualPath: string | undefined; + try { + virtualPath = metadata ? this.config.metaDataToFolderStruct.renamingRule(metadata as any, filePath) : undefined; + } catch { + // Renaming rule may fail for incomplete metadata + } + this.config.stateManager.completeHashProcessing(filePath, hashId, virtualPath); + } } catch (error: any) { // Processing failed - mark as failed (manual retry available via UI) console.error(`[Pipeline] Background queue processing failed for ${filePath}: ${error.message}`); @@ -346,6 +360,30 @@ export class StreamingPipeline { * Get pipeline statistics */ getStats() { + // Get container scheduler queue status if available + const containerScheduler = this.config.containerPluginScheduler; + const containerQueueStatus = containerScheduler?.getQueueStatus(); + + // Use container scheduler queues for fast/background if available + // The container scheduler manages the actual plugin task execution + // PQueue: .pending = running + waiting, .size = waiting only + // So running = pending - size (but ensure non-negative) + const fastQueue = containerQueueStatus ? { + pending: Math.max(0, containerQueueStatus.fast.running), // running jobs + size: Math.max(0, containerQueueStatus.fast.pending) // total pending (running + waiting) + } : { + pending: this.fastQueue.pending, + size: this.fastQueue.size + }; + + const backgroundQueue = containerQueueStatus ? { + pending: Math.max(0, containerQueueStatus.background.running), // running jobs + size: Math.max(0, containerQueueStatus.background.pending) // total pending + } : { + pending: this.backgroundQueue.pending, + size: this.backgroundQueue.size + }; + return { discovered: this.discoveredCount, validated: this.validatedCount, @@ -356,14 +394,8 @@ export class StreamingPipeline { pending: this.validationQueue.pending, size: this.validationQueue.size }, - fastQueue: { - pending: this.fastQueue.pending, - size: this.fastQueue.size - }, - backgroundQueue: { - pending: this.backgroundQueue.pending, - size: this.backgroundQueue.size - } + fastQueue, + backgroundQueue }, state: this.config.stateManager.getSnapshot() }; @@ -491,6 +523,26 @@ export class StreamingPipeline { */ setContainerPluginScheduler(scheduler: import('../../container-plugins/ContainerPluginScheduler.js').ContainerPluginScheduler): void { (this.config as any).containerPluginScheduler = scheduler; + + // Listen for file:complete events to transition files to done state + scheduler.on('file:complete', ({ fileHash, filePath }) => { + console.log(`[Pipeline] Container plugins complete for ${filePath} (${fileHash})`); + + // Get virtual path from metadata if available + let virtualPath: string | undefined; + const metadata = this.config.fileProcessor?.getDatabase().get(filePath); + if (metadata) { + try { + virtualPath = this.config.metaDataToFolderStruct.renamingRule(metadata as any, filePath); + } catch { + // Renaming rule may fail for incomplete metadata + } + } + + // Transition file to done state + this.config.stateManager.completeHashProcessing(filePath, fileHash, virtualPath); + }); + console.log('[Pipeline] Container plugin scheduler connected'); } diff --git a/packages/meta-sort-core/src/plugin-engine/PluginManager.ts b/packages/meta-sort-core/src/plugin-engine/PluginManager.ts index 3efc576..4c23998 100644 --- a/packages/meta-sort-core/src/plugin-engine/PluginManager.ts +++ b/packages/meta-sort-core/src/plugin-engine/PluginManager.ts @@ -405,12 +405,16 @@ export class PluginManager extends EventEmitter { } private async saveState(): Promise { - const state: PluginState = { plugins: {} }; + // Load existing state to preserve config for plugins that aren't currently loaded + // This prevents config loss when plugins fail to initialize + const existingState = await this.loadState(); + const state: PluginState = { plugins: { ...existingState.plugins } }; + // Update state for currently loaded plugins this.manifests.forEach((_, pluginId) => { state.plugins[pluginId] = { active: this.activePluginIds.has(pluginId), - config: this.configs.get(pluginId) || {}, + config: this.configs.get(pluginId) || state.plugins[pluginId]?.config || {}, }; }); diff --git a/packages/meta-sort-core/src/plugin-engine/TaskScheduler.ts b/packages/meta-sort-core/src/plugin-engine/TaskScheduler.ts index 55716fd..66539d2 100644 --- a/packages/meta-sort-core/src/plugin-engine/TaskScheduler.ts +++ b/packages/meta-sort-core/src/plugin-engine/TaskScheduler.ts @@ -56,6 +56,7 @@ export class TaskScheduler extends EventEmitter { private completedTasks: Set = new Set(); private fileTasks: Map> = new Map(); // fileHash -> taskIds private fileKVStores: Map = new Map(); // fileHash -> KV store + private filePathsMap: Map = new Map(); // fileHash -> filePath (for completion tracking) // Container plugin support private containerPluginScheduler: ContainerPluginScheduler | null = null; @@ -157,6 +158,9 @@ export class TaskScheduler extends EventEmitter { } const fileTaskIds = this.fileTasks.get(fileHash)!; + // Track filePath for completion events + this.filePathsMap.set(fileHash, filePath); + // Store KV for this file (create new if not provided) if (!this.fileKVStores.has(fileHash)) { this.fileKVStores.set(fileHash, kv || new KVStore()); @@ -391,7 +395,9 @@ export class TaskScheduler extends EventEmitter { private checkFileComplete(fileHash: string): void { const fileTaskIds = this.fileTasks.get(fileHash); - if (!fileTaskIds) return; + if (!fileTaskIds) { + return; + } // Check if all tasks for this file are completed let filePath = ''; @@ -399,13 +405,18 @@ export class TaskScheduler extends EventEmitter { if (!this.completedTasks.has(taskId)) { return; // Still have pending/running tasks } - // Try to get filePath from any task + // Try to get filePath from any task (might be in pendingTasks if not yet removed) if (!filePath) { const task = this.pendingTasks.get(taskId); if (task) filePath = task.filePath; } } + // Try to get filePath from file paths map if not found + if (!filePath) { + filePath = this.filePathsMap.get(fileHash) || ''; + } + // All tasks complete this.emitEvent({ type: 'file:complete', @@ -416,6 +427,7 @@ export class TaskScheduler extends EventEmitter { // Clean up file tracking this.fileTasks.delete(fileHash); this.fileKVStores.delete(fileHash); + this.filePathsMap.delete(fileHash); // Clean up completed task IDs for this file to prevent memory leak for (const taskId of fileTaskIds) {