fix metasort pipeline

docker/mount-watcher.sh

@@ -1,386 +0,0 @@
-#!/bin/bash
-# =============================================================================
-# Mount Watcher Script
-#
-# Monitors /meta-core/mounts/mounts.json and manages NFS, SMB, and rclone mounts.
-# Runs as a supervisord program, polling every 5 seconds.
-#
-# Uses findmnt to check actual mount status (source of truth).
-# Writes errors to /meta-core/mounts/errors/{id}.error
-# =============================================================================
-
-set -u
-
-CONFIG_DIR="/meta-core/mounts"
-CONFIG_FILE="$CONFIG_DIR/mounts.json"
-ERROR_DIR="$CONFIG_DIR/errors"
-FILES_PATH="${FILES_PATH:-/files}"
-POLL_INTERVAL="${MOUNT_POLL_INTERVAL:-5}"
-
-# ANSI colors
-RED='\033[0;31m'
-GREEN='\033[0;32m'
-YELLOW='\033[1;33m'
-CYAN='\033[0;36m'
-NC='\033[0m' # No Color
-
-log() {
-    echo -e "[$(date -Iseconds)] [${CYAN}mount-watcher${NC}] $1"
-}
-
-log_success() {
-    echo -e "[$(date -Iseconds)] [${CYAN}mount-watcher${NC}] ${GREEN}$1${NC}"
-}
-
-log_warn() {
-    echo -e "[$(date -Iseconds)] [${CYAN}mount-watcher${NC}] ${YELLOW}$1${NC}"
-}
-
-log_error() {
-    echo -e "[$(date -Iseconds)] [${CYAN}mount-watcher${NC}] ${RED}$1${NC}"
-}
-
-# Initialize directories
-init_dirs() {
-    mkdir -p "$CONFIG_DIR" "$ERROR_DIR"
-
-    # Initialize empty config if not exists
-    if [ ! -f "$CONFIG_FILE" ]; then
-        echo '{"version":1,"mounts":[]}' > "$CONFIG_FILE"
-        log "Initialized empty mounts config"
-    fi
-}
-
-# Check if path is mounted
-is_mounted() {
-    local mount_path="$1"
-    findmnt -n "$mount_path" > /dev/null 2>&1
-}
-
-# Clear error file for mount
-clear_error() {
-    local id="$1"
-    rm -f "$ERROR_DIR/$id.error"
-}
-
-# Write error to file
-write_error() {
-    local id="$1"
-    local error="$2"
-    echo "$(date -Iseconds)" > "$ERROR_DIR/$id.error"
-    echo "$error" >> "$ERROR_DIR/$id.error"
-}
-
-# Mount NFS share (always read-only)
-do_mount_nfs() {
-    local id="$1"
-    local server="$2"
-    local path="$3"
-    local mount_path="$4"
-    local options="${5:-}"
-
-    mkdir -p "$mount_path"
-
-    # Always mount read-only for safety
-    local mount_opts="ro"
-    [ -n "$options" ] && mount_opts="${mount_opts},${options}"
-
-    log "Mounting NFS (read-only): ${server}:${path} -> ${mount_path}"
-    output=$(mount -t nfs -o "${mount_opts}" "${server}:${path}" "${mount_path}" 2>&1)
-    local rc=$?
-
-    if [ $rc -eq 0 ]; then
-        clear_error "$id"
-        log_success "NFS mount successful: $mount_path"
-    else
-        write_error "$id" "$output"
-        log_error "NFS mount failed: $output"
-    fi
-    return $rc
-}
-
-# Mount SMB/CIFS share (always read-only)
-do_mount_smb() {
-    local id="$1"
-    local server="$2"
-    local share="$3"
-    local mount_path="$4"
-    local username="${5:-}"
-    local password_obscured="${6:-}"
-    local domain="${7:-}"
-    local options="${8:-}"
-
-    mkdir -p "$mount_path"
-
-    # Always mount read-only for safety
-    local mount_opts="ro"
-
-    # If we have credentials, reveal password and build options
-    if [ -n "$username" ]; then
-        mount_opts="${mount_opts},username=${username}"
-
-        if [ -n "$password_obscured" ]; then
-            # Reveal password using rclone (only exists in memory briefly)
-            local password
-            password=$(rclone reveal "$password_obscured" 2>/dev/null)
-            if [ -n "$password" ]; then
-                mount_opts="${mount_opts},password=${password}"
-            fi
-        fi
-
-        if [ -n "$domain" ]; then
-            mount_opts="${mount_opts},domain=${domain}"
-        fi
-    fi
-
-    # Add any additional options
-    if [ -n "$options" ]; then
-        if [ -n "$mount_opts" ]; then
-            mount_opts="${mount_opts},${options}"
-        else
-            mount_opts="$options"
-        fi
-    fi
-
-    log "Mounting SMB (read-only): //${server}/${share} -> ${mount_path}"
-
-    # Execute mount directly (not via eval) to handle special chars in password
-    local output
-    local rc
-    if [ -n "$mount_opts" ]; then
-        output=$(mount -t cifs "//${server}/${share}" "${mount_path}" -o "${mount_opts}" 2>&1)
-        rc=$?
-    else
-        output=$(mount -t cifs "//${server}/${share}" "${mount_path}" 2>&1)
-        rc=$?
-    fi
-
-    # Clear password from any variables (best effort)
-    password=""
-    mount_opts=""
-
-    if [ $rc -eq 0 ]; then
-        clear_error "$id"
-        log_success "SMB mount successful: $mount_path"
-    else
-        write_error "$id" "$output"
-        log_error "SMB mount failed: $output"
-    fi
-    return $rc
-}
-
-# Mount rclone remote using RC API (always read-only)
-do_mount_rclone() {
-    local id="$1"
-    local remote="$2"
-    local remote_path="$3"
-    local mount_path="$4"
-    local options="${5:-}"
-
-    mkdir -p "$mount_path"
-
-    # Build the remote source string
-    local fs="${remote}:${remote_path}"
-
-    # Use rclone RC API to mount (always read-only for safety)
-    log "Mounting rclone (read-only): ${fs} -> ${mount_path}"
-
-    # Build JSON body for rclone mount with read-only option
-    local json_body
-    json_body=$(cat <<EOF
-{
-    "fs": "${fs}",
-    "mountPoint": "${mount_path}",
-    "mountOpt": {
-        "AllowOther": true,
-        "ReadOnly": true
-    },
-    "vfsOpt": {
-        "CacheMode": 2,
-        "ReadOnly": true
-    }
-}
-EOF
-)
-
-    output=$(curl -s -X POST \
-        -H "Content-Type: application/json" \
-        -u admin:admin \
-        -d "$json_body" \
-        "http://127.0.0.1:5572/mount/mount" 2>&1)
-
-    # Wait a moment for mount to establish
-    sleep 2
-
-    if is_mounted "$mount_path"; then
-        clear_error "$id"
-        log_success "rclone mount successful: $mount_path"
-        return 0
-    else
-        local error_msg="rclone mount failed"
-        # Try to extract error from response
-        if echo "$output" | grep -q "error"; then
-            error_msg="$output"
-        fi
-        write_error "$id" "$error_msg"
-        log_error "rclone mount failed: $error_msg"
-        return 1
-    fi
-}
-
-# Unmount a path
-do_unmount() {
-    local id="$1"
-    local mount_path="$2"
-    local type="$3"
-
-    if ! is_mounted "$mount_path"; then
-        log "Already unmounted: $mount_path"
-        return 0
-    fi
-
-    if [ "$type" = "rclone" ]; then
-        # Use rclone RC API to unmount
-        log "Unmounting rclone: $mount_path"
-        curl -s -X POST \
-            -H "Content-Type: application/json" \
-            -u admin:admin \
-            -d "{\"mountPoint\":\"${mount_path}\"}" \
-            "http://127.0.0.1:5572/mount/unmount" > /dev/null 2>&1
-    else
-        log "Unmounting: $mount_path"
-        umount "$mount_path" 2>&1
-    fi
-
-    # Wait a moment
-    sleep 1
-
-    # Check if still mounted
-    if is_mounted "$mount_path"; then
-        # Force unmount with lazy option
-        log_warn "Force unmounting (lazy): $mount_path"
-        umount -l "$mount_path" 2>&1
-        sleep 1
-    fi
-
-    if is_mounted "$mount_path"; then
-        log_error "Failed to unmount: $mount_path"
-        return 1
-    fi
-
-    log_success "Unmounted: $mount_path"
-    return 0
-}
-
-# Process all mounts from config
-process_mounts() {
-    # Check if config file exists
-    if [ ! -f "$CONFIG_FILE" ]; then
-        return
-    fi
-
-    # Parse JSON using python3
-    local mounts
-    mounts=$(python3 -c '
-import sys, json
-
-try:
-    with open("'"$CONFIG_FILE"'", "r") as f:
-        data = json.load(f)
-except:
-    sys.exit(0)
-
-for m in data.get("mounts", []):
-    # Output pipe-separated values
-    fields = [
-        m.get("id", ""),
-        m.get("type", ""),
-        str(m.get("enabled", True)),
-        str(m.get("desiredMounted", False)),
-        m.get("mountPath", ""),
-        m.get("nfsServer", ""),
-        m.get("nfsPath", ""),
-        m.get("smbServer", ""),
-        m.get("smbShare", ""),
-        m.get("smbUsername", ""),
-        m.get("smbPasswordObscured", ""),
-        m.get("smbDomain", ""),
-        m.get("rcloneRemote", ""),
-        m.get("rclonePath", ""),
-        m.get("options", "")
-    ]
-    print("|".join(fields))
-' 2>/dev/null)
-
-    if [ -z "$mounts" ]; then
-        return
-    fi
-
-    while IFS='|' read -r id type enabled desired mount_path nfs_server nfs_path smb_server smb_share smb_username smb_password_obscured smb_domain rclone_remote rclone_path options; do
-        # Skip empty lines
-        [ -z "$id" ] && continue
-
-        # Check actual mount status
-        local is_mounted_now=false
-        is_mounted "$mount_path" && is_mounted_now=true
-
-        # Handle disabled mounts - unmount if currently mounted
-        if [ "$enabled" != "True" ] && [ "$enabled" != "true" ]; then
-            if [ "$is_mounted_now" = true ]; then
-                log "Mount $id ($mount_path) disabled, unmounting..."
-                do_unmount "$id" "$mount_path" "$type"
-            fi
-            continue
-        fi
-
-        # Handle desired state
-        if [ "$desired" = "True" ] || [ "$desired" = "true" ]; then
-            # Should be mounted
-            if [ "$is_mounted_now" = false ]; then
-                log "Mount $id ($mount_path) desired but not mounted, mounting..."
-                case "$type" in
-                    nfs)
-                        do_mount_nfs "$id" "$nfs_server" "$nfs_path" "$mount_path" "$options"
-                        ;;
-                    smb)
-                        do_mount_smb "$id" "$smb_server" "$smb_share" "$mount_path" "$smb_username" "$smb_password_obscured" "$smb_domain" "$options"
-                        ;;
-                    rclone)
-                        do_mount_rclone "$id" "$rclone_remote" "$rclone_path" "$mount_path" "$options"
-                        ;;
-                    *)
-                        log_error "Unknown mount type: $type"
-                        ;;
-                esac
-            fi
-        else
-            # Should be unmounted
-            if [ "$is_mounted_now" = true ]; then
-                log "Mount $id ($mount_path) not desired, unmounting..."
-                do_unmount "$id" "$mount_path" "$type"
-            fi
-        fi
-    done <<< "$mounts"
-}
-
-# Main loop
-main() {
-    log "Starting mount watcher (poll interval: ${POLL_INTERVAL}s)"
-
-    init_dirs
-
-    while true; do
-        process_mounts
-        sleep "$POLL_INTERVAL"
-    done
-}
-
-# Handle signals for graceful shutdown
-cleanup() {
-    log "Shutting down mount watcher..."
-    exit 0
-}
-
-trap cleanup SIGTERM SIGINT
-
-main "$@"

packages/meta-sort-core/src/config/EnvConfig.ts

@@ -69,6 +69,18 @@ interface EnvConfig {
     /** Concurrency for fast plugin queue. Default: 32 */
     FAST_QUEUE_CONCURRENCY: number;
 
+    /**
+     * Maximum number of files concurrently in the light (fast-queue) phase.
+     * Caps fan-out so the underlying file cache (WebDAV/nginx) only has to
+     * hold N files at a time. Default: 16.
+     */
+    FILE_LIGHT_SLOTS: number;
+
+    /**
+     * Maximum number of files concurrently in the background phase. Default: 16.
+     */
+    FILE_BG_SLOTS: number;
+
     // ========================================================================
     // Container Plugins Configuration
     // All plugins run as Docker containers for isolation and flexibility
@@ -139,6 +151,8 @@ export const config: EnvConfig = {
 
     // TaskScheduler Configuration
     FAST_QUEUE_CONCURRENCY: parseInt(process.env.FAST_QUEUE_CONCURRENCY || "32", 10),
+    FILE_LIGHT_SLOTS: parseInt(process.env.FILE_LIGHT_SLOTS || "16", 10),
+    FILE_BG_SLOTS: parseInt(process.env.FILE_BG_SLOTS || "16", 10),
 
     // Container Plugins Configuration
     CONTAINER_PLUGINS_CONFIG: process.env.CONTAINER_PLUGINS_CONFIG || '/app/plugins.yml',

packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts

@@ -21,6 +21,7 @@ import type { TaskQueueType } from '../plugin-engine/types.js';
 import type { IKVClient } from '../kv/IKVClient.js';
 import { type ExtendedContainerTask } from '../types/ExtendedInterfaces.js';
 import { config } from '../config/EnvConfig.js';
+import { FileSlotGate } from './FileSlotGate.js';
 
 /**
  * Scheduler events
@@ -55,6 +56,18 @@ export interface GateStatus {
     backgroundRunning: number;
 }
 
+/**
+ * Per-file holder snapshot exposed via getQueueStatus for the dashboard.
+ * Lets operators see which 16 files are tying up cache slots and what plugin
+ * each file is currently waiting on (the long-pole per file).
+ */
+export interface CacheSlotHolder {
+    fileHash: string;
+    filePath?: string;
+    currentPlugin?: string;
+    tasksRemaining: number;
+}
+
 /**
  * Container Plugin Scheduler
  */
@@ -73,11 +86,30 @@ export class ContainerPluginScheduler extends EventEmitter {
     private kvClient: IKVClient | null = null;
     private isAcceptingNewTasks: boolean = true; // Gate control for safe mount/unmount
 
+    // File-level cache slots — see FileSlotGate.ts. Caps the number of files
+    // whose bytes are concurrently being touched in each phase, so the WebDAV
+    // cache only has to hold N files at a time regardless of plugin fan-out.
+    private lightFileGate: FileSlotGate;
+    private bgFileGate: FileSlotGate;
+    // Per-(file, phase) outstanding task counter — drives slot release.
+    private lightTasksPerFile: Map<string, number> = new Map();
+    private bgTasksPerFile: Map<string, number> = new Map();
+    // Tasks staged while waiting for a slot. Once the gate grants the file a
+    // slot we flush the buffered tasks into the underlying PQueue.
+    private lightBufferedTasks: Map<string, ContainerTask[]> = new Map();
+    private bgBufferedTasks: Map<string, ContainerTask[]> = new Map();
+    // Currently-dispatched plugin per file in each phase — used by the
+    // dashboard to show "this file is currently waiting on plugin X".
+    private lightCurrentPlugin: Map<string, string> = new Map();
+    private bgCurrentPlugin: Map<string, string> = new Map();
+
     constructor(
         containerManager: ContainerManager,
         options?: {
             fastConcurrency?: number;
             backgroundConcurrency?: number;
+            lightFileSlots?: number;
+            bgFileSlots?: number;
             callbackUrl?: string;
             metaCoreUrl?: string;
             defaultTimeout?: number;
@@ -98,6 +130,9 @@ export class ContainerPluginScheduler extends EventEmitter {
             autoStart: false,
         });
 
+        this.lightFileGate = new FileSlotGate(options?.lightFileSlots ?? config.FILE_LIGHT_SLOTS);
+        this.bgFileGate = new FileSlotGate(options?.bgFileSlots ?? config.FILE_BG_SLOTS);
+
         // When fast queue becomes idle, start background queue
         this.fastQueue.on('idle', () => {
             this.onFastQueueIdle();
@@ -246,6 +281,17 @@ export class ContainerPluginScheduler extends EventEmitter {
     /**
      * Enqueue a task for execution
      * Returns false if the gate is closed and the task was rejected
+     *
+     * Tasks pass through the per-phase FileSlotGate before reaching PQueue:
+     *   - If the file already holds a slot in this phase, the task goes
+     *     straight to PQueue.
+     *   - Otherwise the task is buffered, and the file's first task triggers
+     *     a slot acquire. When the slot is granted, all buffered tasks for
+     *     that file are flushed into PQueue.
+     *
+     * This caps concurrent files per phase (and therefore concurrent file
+     * fetches against CORN/SMB) at the gate's capacity, regardless of how
+     * many plugin tasks each file fans out to.
      */
     enqueueTask(task: ContainerTask): boolean {
         if (!this.isAcceptingNewTasks) {
@@ -262,20 +308,90 @@ export class ContainerPluginScheduler extends EventEmitter {
             return false;
         }
 
-        const queue = task.queue === 'background' ? this.backgroundQueue : this.fastQueue;
-
         // If adding to fast queue, pause background queue (fast takes priority)
         if (task.queue === 'fast' && !this.backgroundQueue.isPaused) {
             console.log('[ContainerScheduler] Fast task added, pausing background queue');
             this.backgroundQueue.pause();
         }
 
-        queue.add(async () => {
+        const counterMap = task.queue === 'fast' ? this.lightTasksPerFile : this.bgTasksPerFile;
-            await this.executeTask(task);
+        const bufferMap = task.queue === 'fast' ? this.lightBufferedTasks : this.bgBufferedTasks;
+        const gate = task.queue === 'fast' ? this.lightFileGate : this.bgFileGate;
+
+        // Increment the per-file counter for this phase eagerly. Decrement
+        // happens in the executeTask finally clause.
+        counterMap.set(task.fileHash, (counterMap.get(task.fileHash) ?? 0) + 1);
+
+        if (gate.holds(task.fileHash)) {
+            // File already has a slot — dispatch immediately.
+            this.dispatchToPQueue(task);
+            return true;
+        }
+
+        // File doesn't hold a slot. Buffer the task. If this is the first
+        // buffered task for the file, trigger gate.acquire and flush on grant.
+        const existingBuffer = bufferMap.get(task.fileHash);
+        if (existingBuffer) {
+            existingBuffer.push(task);
+            return true;
+        }
+
+        bufferMap.set(task.fileHash, [task]);
+        gate.acquire(task.fileHash).then(() => {
+            const buffered = bufferMap.get(task.fileHash) ?? [];
+            bufferMap.delete(task.fileHash);
+            for (const t of buffered) {
+                this.dispatchToPQueue(t);
+            }
+            // Edge case: if every buffered task got cancelled by clear() before
+            // the slot was granted, release immediately so the slot doesn't
+            // leak.
+            if (buffered.length === 0 && (counterMap.get(task.fileHash) ?? 0) === 0) {
+                gate.release(task.fileHash);
+            }
         });
         return true;
     }
 
+    /**
+     * Dispatch a task into the underlying PQueue and release its file's gate
+     * slot when the file's last task in that phase finishes.
+     */
+    private dispatchToPQueue(task: ContainerTask): void {
+        const queue = task.queue === 'background' ? this.backgroundQueue : this.fastQueue;
+        queue.add(async () => {
+            try {
+                await this.executeTask(task);
+            } finally {
+                this.onTaskFinished(task);
+            }
+        });
+    }
+
+    /**
+     * Bookkeeping after a task leaves PQueue (success, failure, or thrown).
+     * Decrements the per-(file, phase) task counter and releases the gate
+     * slot when the counter hits zero.
+     */
+    private onTaskFinished(task: ContainerTask): void {
+        const counterMap = task.queue === 'fast' ? this.lightTasksPerFile : this.bgTasksPerFile;
+        const currentMap = task.queue === 'fast' ? this.lightCurrentPlugin : this.bgCurrentPlugin;
+        const gate = task.queue === 'fast' ? this.lightFileGate : this.bgFileGate;
+
+        const remaining = (counterMap.get(task.fileHash) ?? 1) - 1;
+        if (remaining <= 0) {
+            counterMap.delete(task.fileHash);
+            currentMap.delete(task.fileHash);
+            gate.release(task.fileHash);
+        } else {
+            counterMap.set(task.fileHash, remaining);
+            // Clear current-plugin only if it points at this completed task.
+            if (currentMap.get(task.fileHash) === task.pluginId) {
+                currentMap.delete(task.fileHash);
+            }
+        }
+    }
+
     /**
      * Called when fast queue becomes idle - start background queue if it has work
      */
@@ -314,6 +430,11 @@ export class ContainerPluginScheduler extends EventEmitter {
         task.dispatchedAt = Date.now();
         task.instanceName = instance.containerName;
 
+        // Record which plugin is currently running for this file (per phase)
+        // so the dashboard can show the long-pole plugin per held cache slot.
+        const currentMap = task.queue === 'fast' ? this.lightCurrentPlugin : this.bgCurrentPlugin;
+        currentMap.set(task.fileHash, task.pluginId);
+
         this.emit('task:dispatched', { task });
 
         try {
@@ -581,6 +702,16 @@ export class ContainerPluginScheduler extends EventEmitter {
             backgroundRunning: number; // unique files with background tasks running
             backgroundWaiting: number; // unique files with background tasks waiting
         };
+        // Per-phase cache-slot view — bounded by FileSlotGate. This is the
+        // metric operators actually care about: how many distinct files are
+        // tying up the WebDAV cache right now, and which plugin is the
+        // long-pole on each. `paused` reflects the underlying PQueue state so
+        // the UI can distinguish "16 files reserved but bg PQueue paused"
+        // from "16 files actively dispatching".
+        cacheSlots: {
+            fast: { capacity: number; inUse: number; waiting: number; paused: boolean; holders: CacheSlotHolder[] };
+            background: { capacity: number; inUse: number; waiting: number; paused: boolean; holders: CacheSlotHolder[] };
+        };
     } {
         // Count pending callbacks by queue type
         // These are tasks dispatched to containers awaiting completion - the real "running" count
@@ -639,6 +770,56 @@ export class ContainerPluginScheduler extends EventEmitter {
                 backgroundRunning: backgroundRunningFiles.size,
                 backgroundWaiting: backgroundWaitingFiles.size,
             },
+            cacheSlots: {
+                fast: this.snapshotCacheSlot('fast'),
+                background: this.snapshotCacheSlot('background'),
+            },
+        };
+    }
+
+    /**
+     * Build the per-phase cache-slot snapshot for the queue status payload.
+     * `currentPlugin` is the plugin currently dispatched against the file
+     * (the long-pole) — falls back to undefined if no task is mid-dispatch.
+     * `tasksRemaining` is the count of light/bg tasks still outstanding for
+     * the file in this phase, which is what holds the slot open.
+     */
+    private snapshotCacheSlot(
+        phase: TaskQueueType
+    ): { capacity: number; inUse: number; waiting: number; paused: boolean; holders: CacheSlotHolder[] } {
+        const gate = phase === 'fast' ? this.lightFileGate : this.bgFileGate;
+        const counterMap = phase === 'fast' ? this.lightTasksPerFile : this.bgTasksPerFile;
+        const currentMap = phase === 'fast' ? this.lightCurrentPlugin : this.bgCurrentPlugin;
+        const queue = phase === 'fast' ? this.fastQueue : this.backgroundQueue;
+
+        const holders: CacheSlotHolder[] = gate.getHolders().map((fileHash) => {
+            // Find any task for this file in this phase to recover filePath
+            // (one filePath is enough for the UI; duplicates collapse).
+            let filePath: string | undefined;
+            const taskIds = this.fileTasks.get(fileHash);
+            if (taskIds) {
+                for (const tid of taskIds) {
+                    const t = this.pendingTasks.get(tid);
+                    if (t && t.queue === phase) {
+                        filePath = t.filePath;
+                        break;
+                    }
+                }
+            }
+            return {
+                fileHash,
+                filePath,
+                currentPlugin: currentMap.get(fileHash),
+                tasksRemaining: counterMap.get(fileHash) ?? 0,
+            };
+        });
+
+        return {
+            capacity: gate.capacity,
+            inUse: gate.inUse(),
+            waiting: gate.waiting(),
+            paused: queue.isPaused,
+            holders,
         };
     }
 
@@ -751,6 +932,22 @@ export class ContainerPluginScheduler extends EventEmitter {
         this.fileTasks.clear();
         this.fileCompletedPlugins.clear();
         this.dependencyWaiters.clear();
+
+        // Drop file-slot-gate state. Tasks that were buffered or holding
+        // slots are no longer interesting once we've cleared the queues; a
+        // fresh scheduler state implies fresh gate state.
+        this.lightTasksPerFile.clear();
+        this.bgTasksPerFile.clear();
+        this.lightBufferedTasks.clear();
+        this.bgBufferedTasks.clear();
+        this.lightCurrentPlugin.clear();
+        this.bgCurrentPlugin.clear();
+        for (const fileHash of this.lightFileGate.getHolders()) {
+            this.lightFileGate.release(fileHash);
+        }
+        for (const fileHash of this.bgFileGate.getHolders()) {
+            this.bgFileGate.release(fileHash);
+        }
     }
 
     /**

packages/meta-sort-core/src/container-plugins/FileSlotGate.ts

@@ -0,0 +1,81 @@
+/**
+ * FileSlotGate — bounds the number of files concurrently in a processing phase.
+ *
+ * Motivation: every plugin that runs against a file pulls bytes from CORN/SMB
+ * via WebDAV. Without a per-file gate, the task queue happily fans 8 plugins
+ * across hundreds of files at once, multiplying the WebDAV fetch cost by the
+ * plugin count and overwhelming whatever local cache exists. The gate caps the
+ * number of files that can be "in flight" in a given phase so the cache only
+ * has to hold N files at a time.
+ *
+ * Semantics:
+ *   - Keyed by fileHash. A file holds at most one slot in a given phase.
+ *   - acquire(fileHash) resolves immediately if the file already holds a slot
+ *     or there is free capacity; otherwise it queues FIFO.
+ *   - release(fileHash) is idempotent for non-holders. When called on a real
+ *     holder it frees the slot and wakes the next waiter.
+ *
+ * The gate intentionally knows nothing about tasks or plugins — it is a thin
+ * file-level semaphore. Counting tasks per (file, phase) and deciding when to
+ * release is the caller's responsibility.
+ */
+export class FileSlotGate {
+    private readonly holders: Set<string> = new Set();
+    private readonly waiters: Array<{ fileHash: string; resolve: () => void }> = [];
+
+    constructor(public readonly capacity: number) {
+        if (!Number.isFinite(capacity) || capacity < 1) {
+            throw new Error(`FileSlotGate capacity must be a positive integer, got ${capacity}`);
+        }
+    }
+
+    /**
+     * Acquire a slot for fileHash. If the file already holds a slot, resolves
+     * immediately (re-entrant for the same file). Otherwise queues until a
+     * slot is released.
+     */
+    acquire(fileHash: string): Promise<void> {
+        if (this.holders.has(fileHash)) {
+            return Promise.resolve();
+        }
+        if (this.holders.size < this.capacity) {
+            this.holders.add(fileHash);
+            return Promise.resolve();
+        }
+        return new Promise<void>((resolve) => {
+            this.waiters.push({ fileHash, resolve });
+        });
+    }
+
+    /**
+     * Release the slot held by fileHash. No-op if the file doesn't hold one.
+     * Wakes the next FIFO waiter, if any, and grants it a slot.
+     */
+    release(fileHash: string): void {
+        if (!this.holders.delete(fileHash)) {
+            return;
+        }
+        const next = this.waiters.shift();
+        if (!next) {
+            return;
+        }
+        this.holders.add(next.fileHash);
+        next.resolve();
+    }
+
+    holds(fileHash: string): boolean {
+        return this.holders.has(fileHash);
+    }
+
+    inUse(): number {
+        return this.holders.size;
+    }
+
+    waiting(): number {
+        return this.waiters.length;
+    }
+
+    getHolders(): string[] {
+        return Array.from(this.holders);
+    }
+}

packages/meta-sort-core/src/logic/WatchedFileProcessor.ts

@@ -106,7 +106,7 @@ export class WatchedFileProcessor implements FileProcessorInterface {
             // Map StreamingPipeline queues to the 3-queue architecture UI expects
             // PQueue: .pending = currently running, .size = waiting in queue
             // API expects: .pending = running, .size = waiting
-            const result = {
+            const result: any = {
                 preProcessQueue: {
                     // validation = pre-process (quick extension checks)
                     size: queues.validation.size, // waiting in queue
@@ -130,6 +130,15 @@ export class WatchedFileProcessor implements FileProcessorInterface {
                 // File-level counts (what users care about)
                 files: queues.files
             };
+
+            // Layer in the per-file cache-slot view from the ContainerPluginScheduler
+            // when available — this is the metric that reflects WebDAV/cache load
+            // (number of distinct files concurrently being touched), not just task
+            // throughput. Older deployments without container plugins simply omit it.
+            const containerStatus = this.taskScheduler?.getContainerQueueStatus?.() ?? null;
+            if (containerStatus?.cacheSlots) {
+                result.cacheSlots = containerStatus.cacheSlots;
+            }
             return result;
         }
 

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

@@ -584,6 +584,16 @@ export class TaskScheduler extends EventEmitter {
         return [];
     }
 
+    /**
+     * Pass-through to ContainerPluginScheduler.getQueueStatus so callers like
+     * WatchedFileProcessor can surface the file-cache-slot view (capacity,
+     * holders, current plugin per file) without reaching past the scheduler
+     * boundary themselves.
+     */
+    getContainerQueueStatus(): ReturnType<ContainerPluginScheduler['getQueueStatus']> | null {
+        return this.containerPluginScheduler?.getQueueStatus() ?? null;
+    }
+
     // =========================================================================
     // Shutdown
     // =========================================================================

packages/meta-sort-ui/src/pages/Monitor.tsx

@@ -5,7 +5,8 @@ import {
     RedisStats,
     QueueItem,
     HashTimingStats,
-    FailedFile
+    FailedFile,
+    CacheSlotPhase
 } from '../types';
 import {
     formatMs,
@@ -14,6 +15,43 @@ import {
     getFilename
 } from '../utils/format';
 
+/**
+ * Renders a single phase of the file-cache-slot view: capacity bar, count of
+ * waiting files, and a list of currently-held files showing which plugin
+ * each is currently waiting on. The plugin column is the long-pole — it tells
+ * operators which plugin is the per-file bottleneck.
+ */
+function CacheSlotPanel({ label, phase }: { label: string; phase: CacheSlotPhase }) {
+    const fillPct = Math.min(100, (phase.inUse / Math.max(1, phase.capacity)) * 100);
+    return (
+        <div style={{ marginTop: '0.75rem' }}>
+            <div style={{ display: 'flex', justifyContent: 'space-between', alignItems: 'baseline' }}>
+                <strong>{label}</strong>
+                <span style={{ color: '#9ca3af', fontSize: '0.85rem' }}>
+                    {phase.inUse}/{phase.capacity} files
+                    {phase.waiting > 0 ? ` • ${phase.waiting} waiting for slot` : ''}
+                </span>
+            </div>
+            <div style={{ height: '6px', background: '#0f3460', borderRadius: '3px', overflow: 'hidden', margin: '0.25rem 0 0.5rem' }}>
+                <div style={{ width: `${fillPct}%`, height: '100%', background: '#4ade80', transition: 'width 0.5s linear' }} />
+            </div>
+            {phase.holders.length > 0 && (
+                <div style={{ fontSize: '0.78rem', color: '#cbd5e1', maxHeight: '160px', overflow: 'auto' }}>
+                    {phase.holders.map((h) => (
+                        <div key={h.fileHash} style={{ display: 'flex', gap: '0.5rem', padding: '2px 0' }}>
+                            <span style={{ color: '#facc15', minWidth: '8rem' }}>{h.currentPlugin ?? '—'}</span>
+                            <span style={{ color: '#94a3b8', minWidth: '4rem' }}>×{h.tasksRemaining}</span>
+                            <span style={{ flex: 1, overflow: 'hidden', textOverflow: 'ellipsis', whiteSpace: 'nowrap' }} title={h.filePath ?? h.fileHash}>
+                                {h.filePath ? getFilename(h.filePath) : h.fileHash}
+                            </span>
+                        </div>
+                    ))}
+                </div>
+            )}
+        </div>
+    );
+}
+
 function Monitor() {
     const [status, setStatus] = useState<ProcessingStatus | null>(null);
     const [metrics, setMetrics] = useState<Metrics | null>(null);
@@ -107,31 +145,50 @@ function Monitor() {
     // Get queue statistics for display
     // StreamingPipeline stages: validation → lightProcessing → hashProcessing
     // UI mapping: Validation → Fast Queue → Background Queue
+    //
+    // Fast/Background "Running/Max" represents file-cache slot occupancy
+    // (cacheSlots.inUse / capacity), not raw task throughput. This matches the
+    // operator mental model: "how many files are currently tying up the WebDAV
+    // cache?" The task-callback count was misleading because dependency-wait
+    // and per-plugin HTTP queueing made it under-report by ~5×.
     const getQueueStats = () => {
         const computed = status?.computed;
+        const cacheSlots = status?.queueStatus?.cacheSlots;
         const fastQueueConcurrency = status?.fastQueueConcurrency || 16;
         const backgroundQueueConcurrency = status?.backgroundQueueConcurrency || 8;
 
         return {
-            // Stage 1: Validation (quick extension checks)
+            // Stage 1: Validation (quick extension checks). Still task-level
-            // Concurrency: 2x fast queue workers
+            // — pre-process is a lightweight midhash256 scan, no cache impact.
             preProcess: {
                 running: computed?.preProcessRunning || 0,
                 pending: computed?.preProcessPending || 0,
                 max: fastQueueConcurrency * 2,
                 paused: computed?.preProcessPaused || false
             },
-            // Stage 2: Fast Queue (midhash256 + FFmpeg + plugins)
+            // Stage 2: Fast Queue — files holding a light cache slot.
-            // Concurrency: fastQueueConcurrency workers
+            // running = files in cache, pending = files waiting for a slot.
-            fast: {
+            // paused comes from the cacheSlots payload (true ContainerPluginScheduler
+            // PQueue state) — the legacy `computed.fastQueuePaused` field is sourced
+            // from a different StreamingPipeline queue and is unreliable here.
+            fast: cacheSlots ? {
+                running: cacheSlots.fast.inUse,
+                pending: cacheSlots.fast.waiting,
+                max: cacheSlots.fast.capacity,
+                paused: cacheSlots.fast.paused
+            } : {
                 running: computed?.fastQueueRunning || 0,
                 pending: computed?.fastQueuePending || 0,
                 max: fastQueueConcurrency,
                 paused: computed?.fastQueuePaused || false
             },
-            // Stage 3: Background Queue (SHA-256 full file hash)
+            // Stage 3: Background Queue — files holding a bg cache slot.
-            // Concurrency: backgroundQueueConcurrency workers
+            background: cacheSlots ? {
-            background: {
+                running: cacheSlots.background.inUse,
+                pending: cacheSlots.background.waiting,
+                max: cacheSlots.background.capacity,
+                paused: cacheSlots.background.paused
+            } : {
                 running: computed?.backgroundQueueRunning || 0,
                 pending: computed?.backgroundQueuePending || 0,
                 max: backgroundQueueConcurrency,
@@ -165,9 +222,9 @@ function Monitor() {
                     {showPipelineHelp && (
                         <div className="pipeline-help">
                             <div className="help-item"><span className="help-dot discovered"></span><strong>Discovered:</strong> Files found on disk, waiting to be processed.</div>
-                            <div className="help-item"><span className="help-dot fast"></span><strong>Fast Queue:</strong> Files being processed for quick metadata extraction (hash, ffmpeg, etc.).</div>
+                            <div className="help-item"><span className="help-dot fast"></span><strong>Fast Queue:</strong> Files holding a light cache slot. The badge shows slots-in-use over the gate capacity (e.g. 16/16). Pending = files waiting for a slot to free.</div>
                             <div className="help-item"><span className="help-dot awaiting-background"></span><strong>Awaiting BG:</strong> Files done with fast processing, queued for slower tasks.</div>
-                            <div className="help-item"><span className="help-dot background"></span><strong>Background:</strong> Files running slow tasks like full file hashing or TMDB lookups.</div>
+                            <div className="help-item"><span className="help-dot background"></span><strong>Background:</strong> Files holding a background cache slot (full-hash, TMDB, etc.). Same slot-occupancy semantics as Fast Queue.</div>
                             <div className="help-item"><span className="help-dot done"></span><strong>Complete:</strong> Files fully processed and available in the virtual filesystem.</div>
                             <div className="help-item"><span className="help-dot failed"></span><strong>Failed:</strong> Files that encountered errors during processing.</div>
                         </div>
@@ -301,6 +358,18 @@ function Monitor() {
                 </div>
             )}
 
+            {status?.queueStatus?.cacheSlots && (
+                <div className="card cache-slot-section" style={{ marginTop: '1rem' }}>
+                    <h2>File cache slots</h2>
+                    <p style={{ color: '#888', fontSize: '0.85rem', marginTop: '-0.5rem' }}>
+                        Number of distinct files concurrently held by the WebDAV cache per phase. Caps how
+                        many file fetches CORN/SMB sees at once, regardless of plugin fan-out.
+                    </p>
+                    <CacheSlotPanel label="Light (fast queue)" phase={status.queueStatus.cacheSlots.fast} />
+                    <CacheSlotPanel label="Background" phase={status.queueStatus.cacheSlots.background} />
+                </div>
+            )}
+
             <div className="grid grid-2">
                 {/* Metrics and Stats */}
                 <div className="card">

packages/meta-sort-ui/src/types/index.ts

@@ -75,6 +75,29 @@ export interface TaskSchedulerQueueStatus {
     preProcessQueue?: TaskQueueStatus;
     fastQueue?: TaskQueueStatus;
     backgroundQueue?: TaskQueueStatus;
+    cacheSlots?: CacheSlotsStatus;
+}
+
+export interface CacheSlotHolder {
+    fileHash: string;
+    filePath?: string;
+    currentPlugin?: string;
+    tasksRemaining: number;
+}
+
+export interface CacheSlotPhase {
+    capacity: number;
+    inUse: number;
+    waiting: number;
+    /** True when the underlying PQueue is paused — bg phase is paused while
+     * fast has work, so its slots can be "reserved" without active dispatch. */
+    paused: boolean;
+    holders: CacheSlotHolder[];
+}
+
+export interface CacheSlotsStatus {
+    fast: CacheSlotPhase;
+    background: CacheSlotPhase;
 }
 
 export interface ProcessingStatus {