From e0fa692d7372cdfa0c87d30e869b06e948dbc046 Mon Sep 17 00:00:00 2001 From: Pierre Date: Sat, 17 Jan 2026 14:50:48 +0100 Subject: [PATCH] bug fix and improvement --- README.md | 84 ++++--- .../src/api/UnifiedAPIServer.ts | 228 ++++++++++++++++-- .../meta-sort-core/src/config/EnvConfig.ts | 22 +- .../src/container-plugins/ContainerManager.ts | 147 +++-------- .../ContainerPluginScheduler.ts | 90 ++++++- .../src/container-plugins/DockerClient.ts | 10 +- .../src/container-plugins/index.ts | 1 + .../src/container-plugins/types.ts | 2 +- packages/meta-sort-core/src/index.ts | 4 + .../src/logic/pipeline/StreamingPipeline.ts | 30 +++ packages/meta-sort-ui/src/pages/Monitor.tsx | 58 ++++- packages/meta-sort-ui/src/types/index.ts | 3 + 12 files changed, 496 insertions(+), 183 deletions(-) diff --git a/README.md b/README.md index c643acd..41b4c86 100644 --- a/README.md +++ b/README.md @@ -64,29 +64,37 @@ The **FILES_VOLUME** (canonical name: `/files`) is the shared drive for all medi ``` FILES_VOLUME (/files) -├── media1/ # Local folder mounted from host +├── watch/ # Host media folder (read-only bind mount) │ ├── Movies/ │ │ └── Inception (2010)/ │ │ └── Inception.mkv │ └── TV Shows/ │ └── Breaking Bad/ -│ └── Season 01/ │ -├── media2/ # Another local media folder -│ └── Anime/ -│ └── Steins Gate/ +├── test/ # Test media folder (development) │ -├── smb/ # SMB/CIFS remote mount -│ └── NAS-Media/ -│ └── Movies/ +├── plugin/ # Plugin output files (read-write) +│ └── {plugin-id}/ # Per-plugin output directory │ -├── nfs/ # NFS remote mount -│ └── server-media/ -│ -└── gdrive/ # rclone remote mount (Google Drive) - └── Cloud-Media/ +└── corn/ # SMB/rclone mounts (mounted inside container) + ├── nas-share/ # SMB mount via rclone + └── gdrive/ # Google Drive mount via rclone ``` +### WebDAV File Access + +meta-sort exposes `/files` via **WebDAV** at the `/webdav/` endpoint. This allows: +- **Container plugins** to access files without direct volume mounts +- **meta-fuse** and **meta-stremio** to read files from meta-sort +- Access to dynamic mounts (SMB, rclone) that can't be pre-mounted in other containers + +``` +Container Plugin ──► WebDAV (http://meta-sort/webdav/) ──► /files/ + GET, PUT, DELETE, MKCOL supported +``` + +Environment variable: `PLUGIN_WEBDAV_URL=http://meta-sort-dev/webdav` + ### File Path References All file paths stored in the KV database are **relative to FILES_VOLUME**: @@ -501,8 +509,20 @@ services: | `META_CORE_PATH` | `/meta-core` | Path to META_CORE_VOLUME (infrastructure) | | `FILES_PATH` | `/files` | Path to FILES_VOLUME (shared media) | | `REDIS_URL` | Auto-discover | Redis connection (from leader lock file) | -| `WATCH_PATHS` | `$FILES_PATH` | Folders to watch for media | +| `WATCH_FOLDER_LIST` | `$FILES_PATH` | Comma-separated folders to watch for media | | `SERVICE_NAME` | Container hostname | Unique service identifier | +| `BASE_URL` | `http://localhost:8180` | External URL for this service | + +**Container Plugins:** + +| Variable | Default | Description | +|----------|---------|-------------| +| `CONTAINER_PLUGINS_CONFIG` | `/app/plugins.yml` | Path to plugins configuration | +| `DOCKER_SOCKET_PATH` | `/var/run/docker.sock` | Docker socket for container management | +| `CONTAINER_CALLBACK_URL` | `http://meta-sort:80` | URL plugins use to call back | +| `CONTAINER_NETWORK` | `meta-network` | Docker network for plugin containers | +| `PLUGIN_WEBDAV_URL` | - | WebDAV URL for plugin file access (e.g., `http://meta-sort-dev/webdav`) | +| `PLUGIN_STACK_NAME` | - | Docker Compose project name for grouping containers | --- @@ -538,20 +558,30 @@ Plugins are the core processing units that extract and enrich metadata. Each plu - Runs in either **fast** (high concurrency) or **background** (low concurrency) queue - Can depend on other plugins to ensure correct execution order -### Built-in Plugins +### Container Plugins -| Plugin | Queue | Priority | Description | -|--------|-------|----------|-------------| -| **file-info** | fast | 10 | Basic file info (type, MIME, size, midhash256) | -| **ffmpeg** | fast | 15 | Video/audio/subtitle stream metadata via FFprobe | -| **filename-parser** | fast | 20 | Parse title, season, episode, year, quality from filename | -| **jellyfin-nfo** | fast | 25 | Parse Jellyfin/Kodi NFO metadata files | -| **tmdb** | background | 30 | Fetch metadata from The Movie Database API | -| **anime-detector** | fast | 35 | Detect anime content from keywords and audio tracks | -| **language** | fast | 40 | Aggregate languages from all streams (ISO 639-3) | -| **subtitle** | fast | 45 | Find sibling subtitle files and detect language | -| **torrent** | fast | 50 | Parse .torrent file metadata | -| **full-hash** | background | 100 | Compute SHA-256 hash of entire file | +All plugins run as **Docker containers** for isolation and flexibility. Plugins are configured in `plugins.yml` and communicate via HTTP API. + +| Plugin | Language | Queue | Priority | Description | +|--------|----------|-------|----------|-------------| +| **file-info** | TypeScript | fast | 10 | Basic file info (type, MIME, size, midhash256) | +| **ffmpeg** | TypeScript | fast | 15 | Video/audio/subtitle stream metadata via FFprobe | +| **filename-parser** | TypeScript | fast | 20 | Parse title, season, episode, year, quality from filename | +| **jellyfin-nfo** | TypeScript | fast | 25 | Parse Jellyfin/Kodi NFO metadata files | +| **tmdb** | TypeScript | background | 30 | Fetch metadata from The Movie Database API | +| **anime-detector** | TypeScript | fast | 35 | Detect anime content from keywords and audio tracks | +| **language** | TypeScript | fast | 40 | Aggregate languages from all streams (ISO 639-3) | +| **subtitle** | TypeScript | fast | 45 | Find sibling subtitle files and detect language | +| **torrent** | TypeScript | fast | 50 | Parse .torrent file metadata | +| **full-hash** | TypeScript | background | 100 | Compute SHA-256/SHA-1/MD5 hashes (JS implementation) | +| **fast-full-hash** | **Rust** | background | 100 | High-performance multi-algorithm hashing (Rust/Axum) | + +### Plugin File Access + +Plugins access files via **WebDAV** served by meta-sort nginx. This allows plugins to: +- Access files without requiring direct volume mounts +- See SMB/rclone mounts that are mounted dynamically inside meta-sort +- Read and write to the plugin output directory at `/files/plugin/` ### Plugin Manifest Example diff --git a/packages/meta-sort-core/src/api/UnifiedAPIServer.ts b/packages/meta-sort-core/src/api/UnifiedAPIServer.ts index 4324e8f..a24f1b6 100644 --- a/packages/meta-sort-core/src/api/UnifiedAPIServer.ts +++ b/packages/meta-sort-core/src/api/UnifiedAPIServer.ts @@ -17,7 +17,8 @@ import type { KVManager } from '../kv/KVManager.js'; import { config } from '../config/EnvConfig.js'; import type { PluginManager } from '../plugin-engine/PluginManager.js'; import type { TaskScheduler } from '../plugin-engine/TaskScheduler.js'; -import type { ContainerManager, ContainerPluginScheduler, PluginCallbackPayload } from '../container-plugins/index.js'; +import type { ContainerManager, ContainerPluginScheduler, PluginCallbackPayload, GateStatus } from '../container-plugins/index.js'; +import type { StreamingPipeline } from '../logic/pipeline/StreamingPipeline.js'; export interface UnifiedAPIServerConfig { /** Port to listen on (default: 3000) */ @@ -44,6 +45,7 @@ export class UnifiedAPIServer { private getTaskScheduler: (() => TaskScheduler | null) | null = null; private containerManager: ContainerManager | null = null; private containerPluginScheduler: ContainerPluginScheduler | null = null; + private streamingPipeline: StreamingPipeline | null = null; // Cache for total file size (avoid running du on every request) private totalSizeCache: { value: number; timestamp: number } | null = null; @@ -104,34 +106,43 @@ export class UnifiedAPIServer { } /** - * Get total size of watch folders using du command (cached) + * Get total size of processed files from Redis sizeByte field (cached) + * This is "eventually correct" - accumulates as files are processed by the file-info plugin */ - private getTotalWatchFolderSize(): number { + private async getTotalFileSizeFromRedis(): Promise { // Check cache const now = Date.now(); if (this.totalSizeCache && (now - this.totalSizeCache.timestamp) < UnifiedAPIServer.TOTAL_SIZE_CACHE_TTL_MS) { return this.totalSizeCache.value; } - // Calculate total size using du -sb on each watch folder + // Sum sizeByte from all file entries in Redis let totalSize = 0; - const watchFolders = config.WATCH_FOLDER_LIST?.split(',').map(f => f.trim()).filter(f => f) || []; - for (const folder of watchFolders) { - try { - // du -sb gives total bytes for directory - const output = execSync(`du -sb "${folder}" 2>/dev/null || echo "0"`, { - encoding: 'utf8', - timeout: 10000 // 10s timeout per folder - }); - const bytes = parseInt(output.split('\t')[0], 10); - if (!isNaN(bytes)) { - totalSize += bytes; + if (!this.kvClient) { + return 0; + } + + try { + // Get all file hash IDs + const hashIds = await this.kvClient.getAllHashIds(); + + // Sum up sizeByte values (batch process for efficiency) + for (const hashId of hashIds) { + try { + const sizeByte = await this.kvClient.getMetadata(hashId, 'sizeByte'); + if (sizeByte !== null && sizeByte !== undefined) { + const size = typeof sizeByte === 'number' ? sizeByte : parseInt(String(sizeByte), 10); + if (!isNaN(size) && size > 0) { + totalSize += size; + } + } + } catch { + // Skip entries without sizeByte } - } catch (error) { - // Skip folders that fail (permission issues, doesn't exist, etc.) - console.warn(`Failed to get size for folder ${folder}:`, error); } + } catch (error) { + console.warn('[API] Failed to calculate total size from Redis:', error); } // Update cache @@ -566,6 +577,89 @@ export class UnifiedAPIServer { } }); + // POST /api/mounts/:id/safe-unmount - Safe unmount with queue drain + // Closes gate, waits for queues to empty, then unmounts + this.app.post<{ + Params: { id: string }; + Querystring: { timeout?: string }; + }>('/api/mounts/:id/safe-unmount', async (request, reply) => { + const timeoutMs = parseInt(request.query.timeout || '60000', 10); + + try { + const mountsConfig = await readMountsConfig(); + const mount = mountsConfig.mounts.find((m: any) => m.id === request.params.id); + + if (!mount) { + return reply.status(404).send({ error: 'Mount not found' }); + } + + // Step 1: Close the gate to stop new jobs + if (this.containerPluginScheduler) { + console.log(`[Mounts] Safe unmount: closing gate for ${mount.name}`); + this.containerPluginScheduler.setGate(false); + + // Step 2: Wait for queues to drain + console.log(`[Mounts] Safe unmount: waiting for queues to drain (timeout: ${timeoutMs}ms)`); + const drained = await this.containerPluginScheduler.waitForEmpty(timeoutMs); + + if (!drained) { + // Re-open gate and abort + this.containerPluginScheduler.setGate(true); + const gateStatus = this.containerPluginScheduler.getGateStatus(); + return reply.status(408).send({ + status: 'timeout', + message: `Queue did not drain within ${timeoutMs}ms. Gate re-opened.`, + gateStatus + }); + } + } + + // Step 3: Request unmount + mount.desiredMounted = false; + await writeMountsConfig(mountsConfig); + + // Step 4: Wait for unmount to complete + console.log(`[Mounts] Safe unmount: waiting for mount to detach: ${mount.name}`); + let unmountSuccess = false; + for (let i = 0; i < 30; i++) { // 15 seconds max wait + if (!isMounted(mount.mountPath)) { + unmountSuccess = true; + break; + } + await new Promise(r => setTimeout(r, 500)); + } + + // Step 5: Re-open the gate + if (this.containerPluginScheduler) { + console.log(`[Mounts] Safe unmount: re-opening gate`); + this.containerPluginScheduler.setGate(true); + } + + if (!unmountSuccess) { + console.warn(`[Mounts] Safe unmount: mount still attached after timeout: ${mount.name}`); + return { + status: 'warning', + message: 'Unmount requested but mount may still be attached', + gateStatus: this.containerPluginScheduler?.getGateStatus() ?? null + }; + } + + console.log(`[Mounts] Safe unmount completed: ${mount.name}`); + return { + status: 'ok', + message: 'Safe unmount completed', + gateStatus: this.containerPluginScheduler?.getGateStatus() ?? null + }; + } catch (error: any) { + // Re-open gate on error + if (this.containerPluginScheduler) { + this.containerPluginScheduler.setGate(true); + } + console.error(`[Mounts] Safe unmount error: ${error.message}`); + return reply.status(500).send({ error: error.message }); + } + }); + // DELETE /api/mounts/:id - Remove mount config this.app.delete<{ Params: { id: string } }>('/api/mounts/:id', async (request, reply) => { const fs = await import('fs/promises'); @@ -1341,8 +1435,8 @@ export class UnifiedAPIServer { } } - // Get total size of watch folders (cached, uses du command) - const totalSize = this.getTotalWatchFolderSize(); + // Get total size from Redis sizeByte (eventually correct as files are processed) + const totalSize = await this.getTotalFileSizeFromRedis(); return { prefix: 'file:', @@ -1519,6 +1613,9 @@ export class UnifiedAPIServer { // hashProcessing.size includes both waiting AND running files snapshot.awaitingBackground = Math.max(0, snapshot.awaitingBackground - backgroundQueueRunning); + // Get gate status if container scheduler is available + const gateStatus = this.containerPluginScheduler?.getGateStatus(); + (snapshot as any).computed = { // Pre-process queue (validation - quick extension checks) preProcessRunning, @@ -1537,6 +1634,11 @@ export class UnifiedAPIServer { backgroundQueueTotal: backgroundQueueRunning + backgroundQueuePending + (queueStatus.backgroundQueue?.size || 0), backgroundQueuePaused: queueStatus.backgroundQueue?.isPaused || false, + // Gate control status (for safe mount/unmount operations) + gateOpen: gateStatus?.isOpen ?? true, + gateStatus: gateStatus ?? null, + pipelinePaused: this.streamingPipeline?.isPaused() ?? false, + // Legacy compatibility fields (UI may still reference these) actualRunningHashWorkers: backgroundQueueRunning, actualRunningLightWorkers: preProcessRunning + fastQueueRunning, @@ -1604,6 +1706,80 @@ export class UnifiedAPIServer { return { status: 'ok', message: `${queued} files queued for retry` }; }); + + // POST /api/processing/stop - Stop all processing + // Pauses pipeline queues and closes container plugin gate + // Running tasks will complete, but no new tasks will start + this.app.post('/api/processing/stop', async (request, reply) => { + // Pause the streaming pipeline (stops new files from being processed) + if (this.streamingPipeline) { + this.streamingPipeline.pause(); + } + + // Close the container plugin gate (stops new plugin tasks) + if (this.containerPluginScheduler) { + this.containerPluginScheduler.setGate(false); + } + + const gateStatus = this.containerPluginScheduler?.getGateStatus() ?? null; + const pipelinePaused = this.streamingPipeline?.isPaused() ?? false; + + console.log('[API] Processing stopped - pipeline paused, gate closed'); + + return { + status: 'ok', + message: 'Processing stopped - pipeline paused, gate closed', + gateStatus, + pipelinePaused + }; + }); + + // POST /api/processing/start - Resume all processing + // Resumes pipeline queues and opens container plugin gate + this.app.post('/api/processing/start', async (request, reply) => { + // Resume the streaming pipeline + if (this.streamingPipeline) { + this.streamingPipeline.resume(); + } + + // Open the container plugin gate + if (this.containerPluginScheduler) { + this.containerPluginScheduler.setGate(true); + } + + const gateStatus = this.containerPluginScheduler?.getGateStatus() ?? null; + const pipelinePaused = this.streamingPipeline?.isPaused() ?? false; + + console.log('[API] Processing started - pipeline resumed, gate opened'); + + return { + status: 'ok', + message: 'Processing started - pipeline resumed, gate opened', + gateStatus, + pipelinePaused + }; + }); + + // POST /api/processing/wait-empty - Wait for queues to drain + // Optional query param: timeout (ms, default 60000) + this.app.post<{ Querystring: { timeout?: string } }>('/api/processing/wait-empty', async (request, reply) => { + if (!this.containerPluginScheduler) { + return reply.status(503).send({ + status: 'error', + message: 'Container plugin scheduler not available' + }); + } + + const timeoutMs = parseInt(request.query.timeout || '60000', 10); + const success = await this.containerPluginScheduler.waitForEmpty(timeoutMs); + const gateStatus = this.containerPluginScheduler.getGateStatus(); + + return { + status: success ? 'ok' : 'timeout', + message: success ? 'Queues are empty' : `Timeout after ${timeoutMs}ms`, + gateStatus + }; + }); } @@ -1639,8 +1815,8 @@ export class UnifiedAPIServer { } } - // Get total size of watch folders (cached, uses du command) - const totalSize = this.getTotalWatchFolderSize(); + // Get total size from Redis sizeByte (eventually correct as files are processed) + const totalSize = await this.getTotalFileSizeFromRedis(); return { fileCount: hashIds.length, @@ -2483,6 +2659,14 @@ export class UnifiedAPIServer { console.log('[UnifiedAPIServer] Container plugin manager and scheduler set'); } + /** + * Set streaming pipeline (for pause/resume control) + */ + public setStreamingPipeline(pipeline: StreamingPipeline): void { + this.streamingPipeline = pipeline; + console.log('[UnifiedAPIServer] Streaming pipeline set'); + } + /** * Start the HTTP server */ diff --git a/packages/meta-sort-core/src/config/EnvConfig.ts b/packages/meta-sort-core/src/config/EnvConfig.ts index 753494a..4b6c5cc 100644 --- a/packages/meta-sort-core/src/config/EnvConfig.ts +++ b/packages/meta-sort-core/src/config/EnvConfig.ts @@ -113,21 +113,11 @@ interface EnvConfig { CONTAINER_NETWORK: string; /** - * Plugin file mounts - maps subpaths to host paths for /files in plugin containers. - * Format: "subpath:host_path,subpath2:host_path2" - * Example: "watch:/data/media,test:/app/test/media" - * Results in /files/watch and /files/test mounted in plugin containers (read-only). + * WebDAV URL for plugin file access. + * Plugins access /files via this WebDAV endpoint served by meta-sort nginx. + * Example: "http://meta-sort-dev/webdav" */ - PLUGIN_FILE_MOUNTS?: string; - - /** Host path for plugin cache folder - each plugin gets its own subdirectory at /cache (optional, dev only) */ - PLUGIN_CACHE_FOLDER?: string; - - /** - * Host path for plugin output folders - each plugin gets /files/plugin/ mount (READ-WRITE) - * Plugins decide whether to use this folder - we always mount it if this env is set - */ - PLUGIN_OUTPUT_FOLDER?: string; + PLUGIN_WEBDAV_URL?: string; /** Docker Compose project name for grouping containers in Docker Desktop (optional). When set, plugin containers appear grouped as a stack. */ PLUGIN_STACK_NAME?: string; @@ -202,9 +192,7 @@ export const config: EnvConfig = { CONTAINER_CALLBACK_URL: process.env.CONTAINER_CALLBACK_URL || 'http://meta-sort:8180', CONTAINER_META_CORE_URL: process.env.CONTAINER_META_CORE_URL || 'http://meta-sort:9000', CONTAINER_NETWORK: process.env.CONTAINER_NETWORK || 'meta-network', - PLUGIN_FILE_MOUNTS: process.env.PLUGIN_FILE_MOUNTS, - PLUGIN_CACHE_FOLDER: process.env.PLUGIN_CACHE_FOLDER, - PLUGIN_OUTPUT_FOLDER: process.env.PLUGIN_OUTPUT_FOLDER, + PLUGIN_WEBDAV_URL: process.env.PLUGIN_WEBDAV_URL, PLUGIN_STACK_NAME: process.env.PLUGIN_STACK_NAME, }; diff --git a/packages/meta-sort-core/src/container-plugins/ContainerManager.ts b/packages/meta-sort-core/src/container-plugins/ContainerManager.ts index 6886d75..f093f47 100644 --- a/packages/meta-sort-core/src/container-plugins/ContainerManager.ts +++ b/packages/meta-sort-core/src/container-plugins/ContainerManager.ts @@ -49,62 +49,29 @@ function getContainerName(pluginId: string, instanceIndex: number): string { */ /** * ============================================================================ - * PLUGIN MOUNT ARCHITECTURE - HARDCODED PATHS + * PLUGIN FILE ACCESS ARCHITECTURE - WebDAV * ============================================================================ * - * Each plugin container has exactly 3 mount points (hardcoded): + * Plugins access the /files directory via WebDAV served by meta-sort nginx: * - * /files (READ-ONLY) - Files to process, same structure as meta-sort - * /output (READ-WRITE) - Plugin-generated files - * /cache (READ-WRITE) - Plugin-specific cache + * /webdav (WebDAV endpoint) - Exposes /files with caching + * /files/watch - Watch folder (media files) + * /files/test - Test media folder + * /files/plugin/ - Plugin output (each plugin writes to /files/plugin//) + * /files/corn - SMB mounts (mounted by rclone inside meta-sort) * - * Host paths are configured via environment variables: - * - PLUGIN_FILE_MOUNTS: Maps subpaths under /files to host paths - * Format: "subpath:host_path,subpath2:host_path2" - * Example: "watch:/data/media,test:/app/test/media" - * Results in: /files/watch, /files/test mounted in container + * WebDAV configuration: + * - PLUGIN_WEBDAV_URL: WebDAV endpoint URL (e.g., "http://meta-sort-dev/webdav") * - * - PLUGIN_CACHE_FOLDER: Base path for /cache mounts - * - PLUGIN_OUTPUT_FOLDER: Base path for /output mounts - * - * SECURITY: Plugins have read-only access to /files. - * Only /cache and /output are writable. + * Benefits: + * - Works with Docker's overlay filesystem (no NFS file handle issues) + * - nginx caching for multiple plugins accessing same file + * - Range requests for streaming large files + * - Bidirectional (read/write) via standard HTTP methods * * ============================================================================ */ -/** - * File mount configuration - maps container path to host path - */ -interface FileMountConfig { - containerPath: string; - hostPath: string; -} - -/** - * Parse PLUGIN_FILE_MOUNTS env var into mount configs - * Format: "container_path:host_path,container_path2:host_path2" - */ -function parseFileMounts(mountsStr?: string): FileMountConfig[] { - if (!mountsStr) { - return []; - } - - const mounts: FileMountConfig[] = []; - const pairs = mountsStr.split(',').map(s => s.trim()).filter(s => s); - - for (const pair of pairs) { - const [containerPath, hostPath] = pair.split(':').map(s => s.trim()); - if (containerPath && hostPath) { - mounts.push({ containerPath, hostPath }); - } else { - console.warn(`[ContainerManager] Invalid mount format: "${pair}" - expected "container_path:host_path"`); - } - } - - return mounts; -} - export class ContainerManager extends EventEmitter { private dockerClient: DockerClient; private pluginsConfig: ContainerPluginsConfig | null = null; @@ -113,9 +80,7 @@ export class ContainerManager extends EventEmitter { private roundRobinCounters: Map = new Map(); private initialized: boolean = false; private shutdownInProgress: boolean = false; - private fileMounts: FileMountConfig[] = []; - private pluginCacheFolder?: string; - private pluginOutputFolder?: string; + private webdavUrl?: string; private stackName?: string; constructor( @@ -129,14 +94,8 @@ export class ContainerManager extends EventEmitter { super(); this.dockerClient = dockerClient || new DockerClient(config.DOCKER_SOCKET_PATH); - // Parse file mounts from PLUGIN_FILE_MOUNTS env var - this.fileMounts = parseFileMounts(config.PLUGIN_FILE_MOUNTS); - - // Configure plugin cache folder (READ-WRITE) - this.pluginCacheFolder = config.PLUGIN_CACHE_FOLDER; - - // Configure plugin output folder (READ-WRITE for /plugin-files) - this.pluginOutputFolder = config.PLUGIN_OUTPUT_FOLDER; + // WebDAV configuration for plugin file access + this.webdavUrl = config.PLUGIN_WEBDAV_URL; // Configure stack name for Docker Desktop grouping this.stackName = config.PLUGIN_STACK_NAME; @@ -152,21 +111,18 @@ export class ContainerManager extends EventEmitter { console.log('[ContainerManager] Initializing...'); - // Log configured host mounts - console.log('[ContainerManager] Plugin mount configuration (hardcoded paths):'); - console.log(' /files (RO) - Files to process:'); - if (this.fileMounts.length > 0) { - for (const mount of this.fileMounts) { - const targetPath = mount.containerPath.startsWith('/files') - ? mount.containerPath - : `/files/${mount.containerPath}`; - console.log(` ${targetPath} -> ${mount.hostPath}`); - } + // Log WebDAV configuration + console.log('[ContainerManager] Plugin file access configuration (WebDAV):'); + if (this.webdavUrl) { + console.log(` WebDAV URL: ${this.webdavUrl}`); + console.log(' /files/watch - Watch folder (media files)'); + console.log(' /files/test - Test media folder'); + console.log(' /files/plugin/ - Plugin output (per-plugin subdirs)'); + console.log(' /files/corn - SMB mounts (visible via WebDAV)'); } else { - console.warn(' (not configured) - Set PLUGIN_FILE_MOUNTS'); + console.warn(' WebDAV not configured. Set PLUGIN_WEBDAV_URL.'); + console.warn(' Plugins will not have access to files until WebDAV is configured.'); } - console.log(` /cache (RW) - Plugin cache: ${this.pluginCacheFolder || '(not configured)'}`); - console.log(` /output (RW) - Plugin output: ${this.pluginOutputFolder || '(not configured)'}`) // Initialize Docker client await this.dockerClient.initialize(); @@ -336,47 +292,18 @@ export class ContainerManager extends EventEmitter { } /** - * PLUGIN MOUNT ARCHITECTURE - HARDCODED PATHS: - * /files (READ-ONLY) - Files to process - * /output (READ-WRITE) - Plugin-generated files - * /cache (READ-WRITE) - Plugin-specific cache + * PLUGIN FILE ACCESS ARCHITECTURE - WebDAV: + * Plugins access files via WebDAV URL (no container mounts needed) + * Plugin output goes to /files/plugin// via WebDAV PUT + * Sees all mounts including SMB via rclone + * + * Note: No volume mounts for /files - plugins use HTTP/WebDAV client */ const mounts: Array<{ source: string; target: string; readonly: boolean }> = []; - // /files/* (READ-ONLY) - Mount each configured subpath under /files - for (const fileMount of this.fileMounts) { - // fileMount.containerPath is the subpath (e.g., "watch", "test") - // Mount at /files/{subpath} - const targetPath = fileMount.containerPath.startsWith('/files') - ? fileMount.containerPath - : `/files/${fileMount.containerPath}`; - mounts.push({ - source: fileMount.hostPath, - target: targetPath, - readonly: true, - }); - } - - if (this.fileMounts.length === 0) { - console.warn(`[ContainerManager] WARNING: No file mounts configured. Plugin '${pluginId}' won't have access to files.`); - } - - // /cache (READ-WRITE) - Plugin-specific cache - if (this.pluginCacheFolder) { - mounts.push({ - source: `${this.pluginCacheFolder}/${pluginId}`, - target: '/cache', - readonly: false, - }); - } - - // /output (READ-WRITE) - Plugin-generated files - if (this.pluginOutputFolder) { - mounts.push({ - source: `${this.pluginOutputFolder}/${pluginId}`, - target: '/output', - readonly: false, - }); + // No file mounts needed - plugins access files via WebDAV + if (!this.webdavUrl) { + console.warn(`[ContainerManager] WARNING: WebDAV not configured. Plugin '${pluginId}' won't have access to files. Set PLUGIN_WEBDAV_URL.`); } // Create container @@ -395,6 +322,8 @@ export class ContainerManager extends EventEmitter { PLUGIN_ID: pluginId, META_CORE_URL: this.metaCoreUrl, CALLBACK_URL: this.callbackUrl, + WEBDAV_URL: this.webdavUrl || '', + FILES_PATH: '/files', // Virtual path - files accessed via WebDAV }, // Docker Desktop grouping stackName: this.stackName, diff --git a/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts b/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts index 03b8f59..8070840 100644 --- a/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts +++ b/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts @@ -43,6 +43,17 @@ interface PendingCallback { timeoutHandle: NodeJS.Timeout; } +/** + * Gate status for queue control + */ +export interface GateStatus { + isOpen: boolean; + fastPending: number; + fastRunning: number; + backgroundPending: number; + backgroundRunning: number; +} + /** * Container Plugin Scheduler */ @@ -59,6 +70,7 @@ export class ContainerPluginScheduler extends EventEmitter { private metaCoreUrl: string; private defaultTimeout: number = 60000; // 60 seconds private kvClient: IKVClient | null = null; + private isAcceptingNewTasks: boolean = true; // Gate control for safe mount/unmount constructor( containerManager: ContainerManager, @@ -231,13 +243,29 @@ export class ContainerPluginScheduler extends EventEmitter { /** * Enqueue a task for execution + * Returns false if the gate is closed and the task was rejected */ - enqueueTask(task: ContainerTask): void { + enqueueTask(task: ContainerTask): boolean { + if (!this.isAcceptingNewTasks) { + console.log(`[Gate] Task ${task.id} rejected - gate closed (plugin: ${task.pluginId}, file: ${task.filePath})`); + // Remove from tracking since it won't be executed + this.pendingTasks.delete(task.id); + const fileTasks = this.fileTasks.get(task.fileHash); + if (fileTasks) { + fileTasks.delete(task.id); + if (fileTasks.size === 0) { + this.fileTasks.delete(task.fileHash); + } + } + return false; + } + const queue = task.queue === 'background' ? this.backgroundQueue : this.fastQueue; queue.add(async () => { await this.executeTask(task); }); + return true; } /** @@ -523,6 +551,66 @@ export class ContainerPluginScheduler extends EventEmitter { }; } + /** + * Set the gate state (open/closed) + * When closed, new tasks will be rejected but existing tasks will complete + */ + setGate(accepting: boolean): void { + const wasAccepting = this.isAcceptingNewTasks; + this.isAcceptingNewTasks = accepting; + if (wasAccepting !== accepting) { + console.log(`[Gate] Gate ${accepting ? 'opened' : 'closed'} - ${accepting ? 'accepting' : 'rejecting'} new tasks`); + } + } + + /** + * Check if the gate is open (accepting new tasks) + */ + isGateOpen(): boolean { + return this.isAcceptingNewTasks; + } + + /** + * Get detailed gate status including queue counts + */ + getGateStatus(): GateStatus { + const queueStatus = this.getQueueStatus(); + return { + isOpen: this.isAcceptingNewTasks, + fastPending: queueStatus.fast.pending, + fastRunning: queueStatus.fast.running, + backgroundPending: queueStatus.background.pending, + backgroundRunning: queueStatus.background.running, + }; + } + + /** + * Wait for both queues to drain (all tasks completed) + * Useful for safe unmount operations + * @param timeoutMs Maximum time to wait (default: 60000ms) + * @returns true if queues are empty, false if timeout + */ + async waitForEmpty(timeoutMs: number = 60000): Promise { + const start = Date.now(); + const pollInterval = 500; + + while (Date.now() - start < timeoutMs) { + const status = this.getGateStatus(); + const totalActive = status.fastPending + status.fastRunning + + status.backgroundPending + status.backgroundRunning; + + if (totalActive === 0) { + console.log(`[Gate] Queues drained after ${Date.now() - start}ms`); + return true; + } + + await new Promise(r => setTimeout(r, pollInterval)); + } + + console.log(`[Gate] Timeout waiting for queues to drain (${timeoutMs}ms)`); + return false; + } + /** * Get pending tasks for a file */ diff --git a/packages/meta-sort-core/src/container-plugins/DockerClient.ts b/packages/meta-sort-core/src/container-plugins/DockerClient.ts index 23146b5..b82d8c3 100644 --- a/packages/meta-sort-core/src/container-plugins/DockerClient.ts +++ b/packages/meta-sort-core/src/container-plugins/DockerClient.ts @@ -103,6 +103,12 @@ export class DockerClient { } } + // Build Binds array for bind mounts + const binds = options.mounts?.map(m => { + const opts = [m.readonly ? 'ro' : 'rw']; + return `${m.source}:${m.target}:${opts.join(',')}`; + }) || []; + const createOptions: ContainerCreateOptions = { Image: options.image, name: options.name, @@ -111,9 +117,7 @@ export class DockerClient { : undefined, Labels: labels, HostConfig: { - Binds: options.mounts?.map(m => - `${m.source}:${m.target}:${m.readonly ? 'ro' : 'rw'}` - ), + Binds: binds.length > 0 ? binds : undefined, NetworkMode: options.network, Memory: options.resources?.memory ? parseMemoryLimit(options.resources.memory) diff --git a/packages/meta-sort-core/src/container-plugins/index.ts b/packages/meta-sort-core/src/container-plugins/index.ts index f78167d..8b93b04 100644 --- a/packages/meta-sort-core/src/container-plugins/index.ts +++ b/packages/meta-sort-core/src/container-plugins/index.ts @@ -31,4 +31,5 @@ export { ContainerPluginScheduler, createContainerPluginScheduler, type ContainerSchedulerEvents, + type GateStatus, } from './ContainerPluginScheduler.js'; diff --git a/packages/meta-sort-core/src/container-plugins/types.ts b/packages/meta-sort-core/src/container-plugins/types.ts index 649ee1a..4fccd88 100644 --- a/packages/meta-sort-core/src/container-plugins/types.ts +++ b/packages/meta-sort-core/src/container-plugins/types.ts @@ -283,7 +283,7 @@ export interface ContainerSpawnOptions { env?: Record; /** Resource limits */ resources?: ContainerResourceLimits; - /** Volume mounts */ + /** Volume mounts (bind mounts only - files accessed via WebDAV) */ mounts?: Array<{ source: string; target: string; diff --git a/packages/meta-sort-core/src/index.ts b/packages/meta-sort-core/src/index.ts index 991fd11..0c324b6 100644 --- a/packages/meta-sort-core/src/index.ts +++ b/packages/meta-sort-core/src/index.ts @@ -258,6 +258,10 @@ process.on('SIGINT', async () => { console.log('[Startup] ContainerPluginScheduler connected to StreamingPipeline'); } + // Connect StreamingPipeline to API server (for pause/resume control) + apiServer.setStreamingPipeline(pipeline); + console.log('[Startup] StreamingPipeline connected to API server'); + // First, rebuild VFS from KV to restore state from previous runs console.log('[Startup] Rebuilding VirtualFileSystem from KV...'); await fileProcessor.rebuildVFSFromKV(); diff --git a/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts b/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts index ba811c6..80f3995 100644 --- a/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts +++ b/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts @@ -432,6 +432,36 @@ export class StreamingPipeline { }; } + /** + * Pause all pipeline queues + * Running tasks will complete, but no new tasks will start + */ + pause(): void { + this.validationQueue.pause(); + this.fastQueue.pause(); + this.backgroundQueue.pause(); + console.log('[Pipeline] All queues paused'); + } + + /** + * Resume all pipeline queues + */ + resume(): void { + this.validationQueue.start(); + this.fastQueue.start(); + this.backgroundQueue.start(); + console.log('[Pipeline] All queues resumed'); + } + + /** + * Check if pipeline is paused + */ + isPaused(): boolean { + return this.validationQueue.isPaused || + this.fastQueue.isPaused || + this.backgroundQueue.isPaused; + } + /** * Log current pipeline progress */ diff --git a/packages/meta-sort-ui/src/pages/Monitor.tsx b/packages/meta-sort-ui/src/pages/Monitor.tsx index 95db125..da77843 100644 --- a/packages/meta-sort-ui/src/pages/Monitor.tsx +++ b/packages/meta-sort-ui/src/pages/Monitor.tsx @@ -63,6 +63,24 @@ function Monitor() { } }; + const stopProcessing = async () => { + try { + await fetch('/api/processing/stop', { method: 'POST' }); + fetchData(); + } catch (err) { + console.error('Failed to stop processing:', err); + } + }; + + const startProcessing = async () => { + try { + await fetch('/api/processing/start', { method: 'POST' }); + fetchData(); + } catch (err) { + console.error('Failed to start processing:', err); + } + }; + const retryFile = async (filePath: string) => { try { await fetch('/api/processing/retry', { @@ -156,9 +174,20 @@ function Monitor() {

Processing Monitor

- +
+ {status?.computed?.gateOpen === false ? ( + + ) : ( + + )} + +
{/* Pipeline Progress Bar - 6 Stage (including failed) */} @@ -523,6 +552,29 @@ function Monitor() { font-size: 1.8rem; } + .header-actions { + display: flex; + gap: 12px; + } + + .btn-success { + background: linear-gradient(135deg, #10b981, #059669); + color: white; + } + + .btn-success:hover { + background: linear-gradient(135deg, #059669, #047857); + } + + .btn-danger { + background: linear-gradient(135deg, #ef4444, #dc2626); + color: white; + } + + .btn-danger:hover { + background: linear-gradient(135deg, #dc2626, #b91c1c); + } + /* Pipeline Progress Bar */ .pipeline-section { diff --git a/packages/meta-sort-ui/src/types/index.ts b/packages/meta-sort-ui/src/types/index.ts index e6e8ce7..b5130a0 100644 --- a/packages/meta-sort-ui/src/types/index.ts +++ b/packages/meta-sort-ui/src/types/index.ts @@ -47,6 +47,9 @@ export interface ComputedQueueStats { backgroundQueueTotal: number; backgroundQueuePaused: boolean; + // Gate control status + gateOpen: boolean; + // Legacy fields for backwards compatibility actualRunningLightWorkers: number; actualRunningHashWorkers: number;