streming pipeline improvement

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',

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<void> {
+        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 };
+        fast: { pending: number; running: number; waiting: number };
-        background: { pending: number; running: 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,
+                pending: this.fastQueue.size + fastCallbacks, // waiting + running in containers
-                running: this.fastQueue.pending - this.fastQueue.size,
+                running: fastCallbacks, // tasks running in containers
+                waiting: this.fastQueue.size,
             },
             background: {
-                pending: this.backgroundQueue.pending,
+                pending: this.backgroundQueue.size + backgroundCallbacks, // waiting + running in containers
-                running: this.backgroundQueue.pending - this.backgroundQueue.size,
+                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;

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,
                 });
 

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}`);
     }
 
     /**

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
+            // NOTE: File completion is handled by StreamingPipeline based on whether
-            if (this.unifiedStateManager) {
+            // container plugins are configured. If container plugins exist, we wait for
-                this.unifiedStateManager.completeHashProcessing(
+            // the 'file:complete' event from ContainerPluginScheduler. If not, StreamingPipeline
-                    filePath,
+            // calls completeHashProcessing() directly after this method returns.
-                    midHash256,
-                    virtualPath
-                );
-            }
         } catch (e) {
             // Handle abort errors gracefully
             if (e instanceof Error && (e.message === 'AbortError' || e.name === 'AbortError')) {

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<void> {
         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: {
+                fastQueue,
-                    pending: this.fastQueue.pending,
+                backgroundQueue
-                    size: this.fastQueue.size
-                },
-                backgroundQueue: {
-                    pending: this.backgroundQueue.pending,
-                    size: this.backgroundQueue.size
-                }
             },
             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');
     }
 

packages/meta-sort-core/src/plugin-engine/PluginManager.ts

@@ -405,12 +405,16 @@ export class PluginManager extends EventEmitter {
     }
 
     private async saveState(): Promise<void> {
-        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 || {},
             };
         });
 

packages/meta-sort-core/src/plugin-engine/TaskScheduler.ts

@@ -56,6 +56,7 @@ export class TaskScheduler extends EventEmitter {
     private completedTasks: Set<string> = new Set();
     private fileTasks: Map<string, Set<string>> = new Map(); // fileHash -> taskIds
     private fileKVStores: Map<string, PluginKVStore> = new Map(); // fileHash -> KV store
+    private filePathsMap: Map<string, string> = 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) {