diff --git a/docker/mount-watcher.sh b/docker/mount-watcher.sh deleted file mode 100644 index 346e78a..0000000 --- a/docker/mount-watcher.sh +++ /dev/null @@ -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 <&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 "$@" diff --git a/packages/meta-sort-core/src/config/EnvConfig.ts b/packages/meta-sort-core/src/config/EnvConfig.ts index c15b9e9..7e2d2f2 100644 --- a/packages/meta-sort-core/src/config/EnvConfig.ts +++ b/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', diff --git a/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts b/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts index f7554ff..d15696a 100644 --- a/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts +++ b/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 = new Map(); + private bgTasksPerFile: Map = 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 = new Map(); + private bgBufferedTasks: Map = 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 = new Map(); + private bgCurrentPlugin: Map = 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 () => { - await this.executeTask(task); + const counterMap = task.queue === 'fast' ? this.lightTasksPerFile : this.bgTasksPerFile; + 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); + } } /** diff --git a/packages/meta-sort-core/src/container-plugins/FileSlotGate.ts b/packages/meta-sort-core/src/container-plugins/FileSlotGate.ts new file mode 100644 index 0000000..01f93d7 --- /dev/null +++ b/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 = 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 { + if (this.holders.has(fileHash)) { + return Promise.resolve(); + } + if (this.holders.size < this.capacity) { + this.holders.add(fileHash); + return Promise.resolve(); + } + return new Promise((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); + } +} diff --git a/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts b/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts index 2444e1f..bde27b2 100644 --- a/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts +++ b/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; } diff --git a/packages/meta-sort-core/src/plugin-engine/TaskScheduler.ts b/packages/meta-sort-core/src/plugin-engine/TaskScheduler.ts index 66539d2..3c993d9 100644 --- a/packages/meta-sort-core/src/plugin-engine/TaskScheduler.ts +++ b/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 | null { + return this.containerPluginScheduler?.getQueueStatus() ?? null; + } + // ========================================================================= // Shutdown // ========================================================================= diff --git a/packages/meta-sort-ui/src/pages/Monitor.tsx b/packages/meta-sort-ui/src/pages/Monitor.tsx index fd7105d..6eeb6a2 100644 --- a/packages/meta-sort-ui/src/pages/Monitor.tsx +++ b/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 ( +
+
+ {label} + + {phase.inUse}/{phase.capacity} files + {phase.waiting > 0 ? ` • ${phase.waiting} waiting for slot` : ''} + +
+
+
+
+ {phase.holders.length > 0 && ( +
+ {phase.holders.map((h) => ( +
+ {h.currentPlugin ?? '—'} + ×{h.tasksRemaining} + + {h.filePath ? getFilename(h.filePath) : h.fileHash} + +
+ ))} +
+ )} +
+ ); +} + function Monitor() { const [status, setStatus] = useState(null); const [metrics, setMetrics] = useState(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) - // Concurrency: 2x fast queue workers + // Stage 1: Validation (quick extension checks). Still task-level + // — 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) - // Concurrency: fastQueueConcurrency workers - fast: { + // Stage 2: Fast Queue — files holding a light cache slot. + // running = files in cache, pending = files waiting for a slot. + // 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) - // Concurrency: backgroundQueueConcurrency workers - background: { + // Stage 3: Background Queue — files holding a bg cache slot. + background: cacheSlots ? { + 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 && (
Discovered: Files found on disk, waiting to be processed.
-
Fast Queue: Files being processed for quick metadata extraction (hash, ffmpeg, etc.).
+
Fast Queue: 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.
Awaiting BG: Files done with fast processing, queued for slower tasks.
-
Background: Files running slow tasks like full file hashing or TMDB lookups.
+
Background: Files holding a background cache slot (full-hash, TMDB, etc.). Same slot-occupancy semantics as Fast Queue.
Complete: Files fully processed and available in the virtual filesystem.
Failed: Files that encountered errors during processing.
@@ -301,6 +358,18 @@ function Monitor() {
)} + {status?.queueStatus?.cacheSlots && ( +
+

File cache slots

+

+ 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. +

+ + +
+ )} +
{/* Metrics and Stats */}
diff --git a/packages/meta-sort-ui/src/types/index.ts b/packages/meta-sort-ui/src/types/index.ts index cd03d8d..e2265df 100644 --- a/packages/meta-sort-ui/src/types/index.ts +++ b/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 {