metastar/meta-sort
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

bugfix queue

Browse Source
This commit is contained in:
Pierre
2026-01-16 21:53:02 +01:00
parent 818f043e70
commit 4642add452
10 changed files with 1196 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/meta-sort-core/src/api/UnifiedAPIServer.ts
+438 -31
View File
@@ -222,6 +222,9 @@ export class UnifiedAPIServer {
// File download routes (under /api/file)
this.setupFileRoutes();
// Mount management routes (under /api/mounts)
this.setupMountRoutes();
}
/**
@@ -278,6 +281,338 @@ export class UnifiedAPIServer {
});
}
/**
* Setup Mount Management API routes
* Manages NFS, SMB, and rclone remote storage mounts
*/
private setupMountRoutes(): void {
const MOUNTS_DIR = `${config.META_CORE_PATH}/mounts`;
const MOUNTS_FILE = `${MOUNTS_DIR}/mounts.json`;
const ERRORS_DIR = `${MOUNTS_DIR}/errors`;
const FILES_PATH = config.FILES_PATH;
// Helper: ensure directories exist
const ensureDirs = async () => {
const fs = await import('fs/promises');
await fs.mkdir(MOUNTS_DIR, { recursive: true });
await fs.mkdir(ERRORS_DIR, { recursive: true });
};
// Helper: sanitize name for filesystem
const sanitizeName = (name: string): string => {
return name
.toLowerCase()
.replace(/[^a-z0-9-_]/g, '-')
.replace(/-+/g, '-')
.replace(/^-|-$/g, '')
.slice(0, 64);
};
// Helper: read mounts config
const readMountsConfig = async (): Promise<{ version: number; mounts: any[] }> => {
const fs = await import('fs/promises');
try {
const data = await fs.readFile(MOUNTS_FILE, 'utf-8');
return JSON.parse(data);
} catch (e: any) {
if (e.code === 'ENOENT') {
return { version: 1, mounts: [] };
}
throw e;
}
};
// Helper: write mounts config
const writeMountsConfig = async (mountsConfig: { version: number; mounts: any[] }) => {
const fs = await import('fs/promises');
await ensureDirs();
await fs.writeFile(MOUNTS_FILE, JSON.stringify(mountsConfig, null, 2));
};
// Helper: check if path is mounted
const isMounted = (mountPath: string): boolean => {
try {
const output = execSync(`findmnt -n "${mountPath}" 2>/dev/null || true`, {
encoding: 'utf-8',
timeout: 5000 // 5 second timeout to prevent hanging on slow mounts
});
return output.trim().length > 0;
} catch {
return false;
}
};
// Helper: read error for mount
const readError = async (id: string): Promise<string | undefined> => {
const fs = await import('fs/promises');
try {
const errorFile = `${ERRORS_DIR}/${id}.error`;
const content = await fs.readFile(errorFile, 'utf-8');
const lines = content.trim().split('\n');
return lines.slice(1).join('\n'); // Skip timestamp line
} catch {
return undefined;
}
};
// GET /api/mounts - List all mounts with status
this.app.get('/api/mounts', async (request, reply) => {
try {
await ensureDirs();
const mountsConfig = await readMountsConfig();
const mounts = await Promise.all(mountsConfig.mounts.map(async (mount: any) => {
const mounted = isMounted(mount.mountPath);
const error = await readError(mount.id);
return {
...mount,
mounted,
error,
lastChecked: Date.now()
};
}));
return { mounts };
} catch (error: any) {
return reply.status(500).send({ error: error.message });
}
});
// GET /api/mounts/rclone/remotes - List available rclone remotes
this.app.get('/api/mounts/rclone/remotes', async (request, reply) => {
try {
const response = await fetch('http://127.0.0.1:5572/config/listremotes', {
headers: {
'Authorization': 'Basic ' + Buffer.from('admin:admin').toString('base64')
}
});
if (!response.ok) {
return { remotes: [] };
}
const data = await response.json() as { remotes?: string[] };
const remoteNames = data.remotes || [];
// Get type for each remote
const remotes = await Promise.all(remoteNames.map(async (name: string) => {
try {
const typeRes = await fetch('http://127.0.0.1:5572/config/get', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': 'Basic ' + Buffer.from('admin:admin').toString('base64')
},
body: JSON.stringify({ name: name.replace(/:$/, '') })
});
const typeData = await typeRes.json() as { type?: string };
return {
name: name.replace(/:$/, ''),
type: typeData.type || 'unknown'
};
} catch {
return { name: name.replace(/:$/, ''), type: 'unknown' };
}
}));
return { remotes };
} catch (error: any) {
return { remotes: [] };
}
});
// POST /api/mounts - Create new mount
interface CreateMountBody {
name?: string;
type?: 'nfs' | 'smb' | 'rclone';
enabled?: boolean;
nfsServer?: string;
nfsPath?: string;
smbServer?: string;
smbShare?: string;
smbUsername?: string;
smbPassword?: string;
smbDomain?: string;
rcloneRemote?: string;
rclonePath?: string;
}
this.app.post<{ Body: CreateMountBody }>('/api/mounts', async (request, reply) => {
const { randomUUID } = await import('crypto');
const fs = await import('fs/promises');
try {
await ensureDirs();
const body = request.body;
if (!body.name) {
return reply.status(400).send({ error: 'Mount name is required' });
}
if (!body.type || !['nfs', 'smb', 'rclone'].includes(body.type)) {
return reply.status(400).send({ error: 'Valid mount type (nfs, smb, rclone) is required' });
}
const id = randomUUID();
const safeName = sanitizeName(body.name);
const mountPath = `${FILES_PATH}/${safeName}`;
// Check if mount path already exists in config
const existingConfig = await readMountsConfig();
const pathExists = existingConfig.mounts.some((m: any) => m.mountPath === mountPath);
if (pathExists) {
return reply.status(400).send({ error: `Mount path ${mountPath} already configured` });
}
const mount: Record<string, any> = {
id,
name: body.name,
type: body.type,
enabled: body.enabled !== false,
desiredMounted: body.enabled !== false, // Auto-mount if enabled
mountPath
};
// Type-specific fields
if (body.type === 'nfs') {
if (!body.nfsServer || !body.nfsPath) {
return reply.status(400).send({ error: 'NFS server and path are required' });
}
mount.nfsServer = body.nfsServer;
mount.nfsPath = body.nfsPath;
} else if (body.type === 'smb') {
if (!body.smbServer || !body.smbShare) {
return reply.status(400).send({ error: 'SMB server and share are required' });
}
mount.smbServer = body.smbServer;
mount.smbShare = body.smbShare;
// Store credentials - password is obscured using rclone
if (body.smbUsername) {
mount.smbUsername = body.smbUsername;
}
if (body.smbPassword) {
try {
// Obscure password using rclone (prevents plaintext storage)
const obscured = execSync(`rclone obscure "${body.smbPassword.replace(/"/g, '\\"')}"`, {
encoding: 'utf-8',
timeout: 5000
}).trim();
mount.smbPasswordObscured = obscured;
} catch (err) {
console.error('[Mounts] Failed to obscure password:', err);
return reply.status(500).send({ error: 'Failed to secure password' });
}
}
if (body.smbDomain) {
mount.smbDomain = body.smbDomain;
}
} else if (body.type === 'rclone') {
if (!body.rcloneRemote) {
return reply.status(400).send({ error: 'rclone remote is required' });
}
mount.rcloneRemote = body.rcloneRemote;
mount.rclonePath = body.rclonePath || '';
}
existingConfig.mounts.push(mount);
await writeMountsConfig(existingConfig);
console.log(`[Mounts] Created mount config: ${mount.name} (${mount.type}) -> ${mount.mountPath}`);
return { mount };
} catch (error: any) {
console.error('[Mounts] Error creating mount:', error);
return reply.status(500).send({ error: error.message });
}
});
// POST /api/mounts/:id/mount - Request mount
this.app.post<{ Params: { id: string } }>('/api/mounts/:id/mount', async (request, reply) => {
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' });
}
mount.desiredMounted = true;
await writeMountsConfig(mountsConfig);
console.log(`[Mounts] Mount requested: ${mount.name}`);
return { status: 'ok', message: 'Mount requested' };
} catch (error: any) {
return reply.status(500).send({ error: error.message });
}
});
// POST /api/mounts/:id/unmount - Request unmount
this.app.post<{ Params: { id: string } }>('/api/mounts/:id/unmount', async (request, reply) => {
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' });
}
mount.desiredMounted = false;
await writeMountsConfig(mountsConfig);
console.log(`[Mounts] Unmount requested: ${mount.name}`);
return { status: 'ok', message: 'Unmount requested' };
} catch (error: any) {
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');
try {
const mountsConfig = await readMountsConfig();
const index = mountsConfig.mounts.findIndex((m: any) => m.id === request.params.id);
if (index === -1) {
return reply.status(404).send({ error: 'Mount not found' });
}
const mount = mountsConfig.mounts[index];
// Request unmount first
mount.desiredMounted = false;
await writeMountsConfig(mountsConfig);
// Wait for unmount (poll for up to 15 seconds)
for (let i = 0; i < 30; i++) {
if (!isMounted(mount.mountPath)) break;
await new Promise(r => setTimeout(r, 500));
}
// Remove from config
mountsConfig.mounts.splice(index, 1);
await writeMountsConfig(mountsConfig);
// Clean up files
try {
await fs.unlink(`${ERRORS_DIR}/${mount.id}.error`);
} catch { /* ignore */ }
try {
await fs.rmdir(mount.mountPath);
} catch { /* ignore */ }
console.log(`[Mounts] Deleted mount: ${mount.name}`);
return { status: 'ok' };
} catch (error: any) {
console.error('[Mounts] Error deleting mount:', error);
return reply.status(500).send({ error: error.message });
}
});
console.log('[UnifiedAPIServer] Mount management routes registered at /api/mounts/*');
}
/**
* Setup FUSE API routes
*/
@@ -1108,26 +1443,47 @@ export class UnifiedAPIServer {
// Get active processing queue (files currently being processed)
this.app.get('/api/processing/queue', async (request, reply) => {
const items: Array<{ path: string; phase: 'light' | 'hash'; startTime?: number }> = [];
const items: Array<{ path: string; phase: 'light' | 'hash'; startTime?: number; plugin?: string }> = [];
// Get files in light processing
const lightProcessing = this.unifiedStateManager!.getLightProcessingFiles();
for (const [filePath, info] of lightProcessing) {
// Get running tasks from TaskScheduler (container plugin tasks)
const taskScheduler = this.getTaskScheduler?.();
const runningTasks = taskScheduler?.getRunningTasks() || [];
// Add running plugin tasks - map queue type to phase
// 'fast' queue = 'light' phase, 'background' queue = 'hash' phase
for (const task of runningTasks) {
items.push({
path: filePath,
phase: 'light',
startTime: info.lightProcessingStartedAt
path: task.filePath,
phase: task.queue === 'background' ? 'hash' : 'light',
startTime: task.startTime,
plugin: task.pluginId
});
}
// Get files in hash processing
// Get files in light processing (midhash256 computation - no plugin yet)
const lightProcessing = this.unifiedStateManager!.getLightProcessingFiles();
for (const [filePath, info] of lightProcessing) {
// Only add if not already in plugin tasks
if (!runningTasks.some(t => t.filePath === filePath)) {
items.push({
path: filePath,
phase: 'light',
startTime: info.lightProcessingStartedAt
});
}
}
// Get files in hash processing (waiting for or running background tasks)
const hashProcessing = this.unifiedStateManager!.getHashProcessingFiles();
for (const [filePath, info] of hashProcessing) {
items.push({
path: filePath,
phase: 'hash',
startTime: info.hashProcessingStartedAt
});
// Only add if not already in plugin tasks
if (!runningTasks.some(t => t.filePath === filePath)) {
items.push({
path: filePath,
phase: 'hash',
startTime: info.hashProcessingStartedAt
});
}
}
return { items };
@@ -1135,24 +1491,10 @@ export class UnifiedAPIServer {
// Get unified processing status snapshot (4 tabs: pending, lightProcessing, hashProcessing, done)
this.app.get('/api/processing/status', async (request, reply) => {
// Get ALL file paths currently in VFS (most accurate count)
const allVfsFiles = this.vfs.getAllFiles();
// Count only actual media files (not virtual metadata files like .meta, .nfo)
const mediaFiles = allVfsFiles.filter(f => {
return !f.endsWith('.meta') && !f.endsWith('.nfo') && !f.endsWith('.xml');
});
const actualMediaFileCount = mediaFiles.length;
// Count files with completed hash computation
const filesWithHash = this.vfs.countFilesWithHash();
// Get snapshot with actual VFS counts
const snapshot = this.unifiedStateManager!.getSnapshot(actualMediaFileCount, this.fastQueueConcurrency, this.backgroundQueueConcurrency);
// Override totalDone with actual count of files with hashes in VFS
// (not just current session, but ALL files in VFS with completed hash)
snapshot.totalDone = filesWithHash;
// Get snapshot from state manager
// Note: VFS-related features have been moved to meta-fuse
// totalDone now uses state manager's count of fully processed files
const snapshot = this.unifiedStateManager!.getSnapshot(0, this.fastQueueConcurrency, this.backgroundQueueConcurrency);
// Add queue status if available
if (this.getQueueStatus) {
@@ -1168,6 +1510,15 @@ export class UnifiedAPIServer {
const backgroundQueueRunning = queueStatus.backgroundQueue?.running || 0;
const backgroundQueuePending = queueStatus.backgroundQueue?.pending || 0;
// "Discovered" now represents files waiting for fast queue (validated but not yet processing)
// This merges the old "discovered" and "awaitingFastQueue" concepts
snapshot.totalDiscovered = fastQueuePending;
snapshot.awaitingFastQueue = fastQueuePending; // Keep for backward compatibility
// Fix awaitingBackground: subtract files currently being processed by background workers
// hashProcessing.size includes both waiting AND running files
snapshot.awaitingBackground = Math.max(0, snapshot.awaitingBackground - backgroundQueueRunning);
(snapshot as any).computed = {
// Pre-process queue (validation - quick extension checks)
preProcessRunning,
@@ -1195,8 +1546,64 @@ export class UnifiedAPIServer {
};
}
// Add failed files count from metrics
const metrics = performanceMetrics.getMetrics();
(snapshot as any).totalFailed = metrics.totalFailedFiles || 0;
return snapshot;
});
// Get failed files list
this.app.get('/api/processing/failed', async (request, reply) => {
const metrics = performanceMetrics.getMetrics();
return {
failedFiles: metrics.failedFiles || [],
totalFailed: metrics.totalFailedFiles || 0
};
});
// Retry a failed file
this.app.post<{ Body: { filePath: string } }>('/api/processing/retry', async (request, reply) => {
const { filePath } = request.body;
if (!filePath) {
return reply.status(400).send({ error: 'filePath is required' });
}
try {
// Add file back to discovered state for reprocessing
this.unifiedStateManager!.addDiscovered(filePath);
// Trigger scan callback if available to pick up the file
if (this.triggerScanCallback) {
// Don't await - just trigger async
this.triggerScanCallback().catch(() => {});
}
return { status: 'ok', message: `File ${filePath} queued for retry` };
} catch (error: any) {
return reply.status(500).send({ error: error.message });
}
});
// Retry all failed files
this.app.post('/api/processing/retry-all', async (request, reply) => {
const metrics = performanceMetrics.getMetrics();
const failedFiles = metrics.failedFiles || [];
let queued = 0;
for (const file of failedFiles) {
this.unifiedStateManager!.addDiscovered(file.filePath);
queued++;
}
// Trigger scan callback if available
if (this.triggerScanCallback) {
this.triggerScanCallback().catch(() => {});
}
return { status: 'ok', message: `${queued} files queued for retry` };
});
}
packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts
+37
View File
@@ -577,6 +577,43 @@ export class ContainerPluginScheduler extends EventEmitter {
/**
* Pause queues
*/
/**
* Get currently running tasks (status === 'dispatched')
* Returns array of { taskId, pluginId, fileHash, filePath, queue, startTime }
*/
getRunningTasks(): Array<{
taskId: string;
pluginId: string;
fileHash: string;
filePath: string;
queue: TaskQueueType;
startTime?: number;
}> {
const running: Array<{
taskId: string;
pluginId: string;
fileHash: string;
filePath: string;
queue: TaskQueueType;
startTime?: number;
}> = [];
for (const [taskId, task] of this.pendingTasks) {
if (task.status === 'dispatched') {
running.push({
taskId,
pluginId: task.pluginId,
fileHash: task.fileHash,
filePath: task.filePath,
queue: task.queue,
startTime: task.dispatchedAt,
});
}
}
return running;
}
pause(): void {
this.fastQueue.pause();
this.backgroundQueue.pause();
packages/meta-sort-core/src/logic/UnifiedProcessingStateManager.ts
+51 -26
View File
@@ -2,7 +2,7 @@
* UnifiedProcessingStateManager
*
* Tracks files through the complete processing pipeline with 4 states:
* 1. Pending: Files discovered in filesystem (not yet processed)
* 1. Discovered: Files discovered in filesystem (not yet processed)
* 2. Light Processing: Quick metadata extraction + midhash256 computation (~20ms)
* 3. Hash Processing: SHA-256 computation for compatibility (slow, background)
* 4. Done: Fully processed files
@@ -15,7 +15,7 @@
import { performanceMetrics } from '../metrics/PerformanceMetrics.js';
export type FileState = 'pending' | 'lightProcessing' | 'hashProcessing' | 'done';
export type FileState = 'discovered' | 'lightProcessing' | 'hashProcessing' | 'done';
export interface UnifiedFileState {
filePath: string;
@@ -42,24 +42,29 @@ export interface UnifiedFileState {
}
export interface UnifiedProcessingSnapshot {
pending: UnifiedFileState[];
discovered: UnifiedFileState[];
lightProcessing: UnifiedFileState[];
hashProcessing: UnifiedFileState[];
done: UnifiedFileState[];
totalPending: number;
totalDiscovered: number;
totalLightProcessing: number;
totalHashProcessing: number;
totalDone: number;
totalFilesInVFS: number; // Total number of files accessible in the Virtual File System
/** Files that completed validation but are waiting for fast queue */
awaitingFastQueue: number;
/** Files that completed fast queue but are waiting for background queue */
awaitingBackground: number;
// Configuration
fastQueueConcurrency?: number; // Number of fast queue workers
backgroundQueueConcurrency?: number; // Number of background queue workers
}
export class UnifiedProcessingStateManager {
private pending: Map<string, UnifiedFileState> = new Map();
private discovered: Map<string, UnifiedFileState> = new Map();
private lightProcessing: Map<string, UnifiedFileState> = new Map();
private hashProcessing: Map<string, UnifiedFileState> = new Map();
private done: UnifiedFileState[] = [];
@@ -67,28 +72,35 @@ export class UnifiedProcessingStateManager {
private totalProcessedCount = 0; // Track total number of completed files (not limited)
/**
* Add a file to pending state (file discovered)
* Add a file to discovered state (file discovered)
*/
addPending(filePath: string): void {
this.pending.set(filePath, {
addDiscovered(filePath: string): void {
this.discovered.set(filePath, {
filePath,
state: 'pending',
state: 'discovered',
discoveredAt: Date.now()
});
}
/**
* Move file from pending to light processing
* @deprecated Use addDiscovered instead
*/
addPending(filePath: string): void {
this.addDiscovered(filePath);
}
/**
* Move file from discovered to light processing
*/
startLightProcessing(filePath: string): void {
const pendingState = this.pending.get(filePath);
const discoveredState = this.discovered.get(filePath);
const existingState = this.lightProcessing.get(filePath);
this.pending.delete(filePath);
this.discovered.delete(filePath);
this.lightProcessing.set(filePath, {
filePath,
state: 'lightProcessing',
discoveredAt: pendingState?.discoveredAt || existingState?.discoveredAt,
discoveredAt: discoveredState?.discoveredAt || existingState?.discoveredAt,
lightProcessingStartedAt: Date.now(),
retryCount: existingState?.retryCount || 0
});
@@ -204,7 +216,7 @@ export class UnifiedProcessingStateManager {
* Remove a file from all states (e.g., when deleted)
*/
removeFile(filePath: string): void {
this.pending.delete(filePath);
this.discovered.delete(filePath);
this.lightProcessing.delete(filePath);
this.hashProcessing.delete(filePath);
this.done = this.done.filter(f => f.filePath !== filePath);
@@ -221,10 +233,10 @@ export class UnifiedProcessingStateManager {
const MAX_ITEMS_PER_STATE = 100;
// Efficiently extract only the needed items without creating full arrays
const getPendingSample = (): UnifiedFileState[] => {
const getDiscoveredSample = (): UnifiedFileState[] => {
const result: UnifiedFileState[] = [];
let count = 0;
for (const item of this.pending.values()) {
for (const item of this.discovered.values()) {
if (count >= MAX_ITEMS_PER_STATE) break;
result.push(item);
count++;
@@ -255,17 +267,22 @@ export class UnifiedProcessingStateManager {
};
return {
pending: getPendingSample(),
discovered: getDiscoveredSample(),
lightProcessing: getLightProcessingSample(),
hashProcessing: getHashProcessingSample(),
done: this.done.slice(0, 50), // Return only last 50 for API
totalPending: this.pending.size,
totalDiscovered: this.discovered.size,
totalLightProcessing: this.lightProcessing.size,
totalHashProcessing: this.hashProcessing.size,
totalDone: this.totalProcessedCount, // Total count of all processed files (not limited to history)
totalFilesInVFS: vfsFileCount, // Total number of files accessible in the Virtual File System
// Files that completed validation but are waiting for fast queue (set by API layer from queue status)
awaitingFastQueue: 0,
// Files that completed fast queue but are waiting for/in background queue
awaitingBackground: this.hashProcessing.size,
fastQueueConcurrency,
backgroundQueueConcurrency
};
@@ -275,7 +292,7 @@ export class UnifiedProcessingStateManager {
* Get the state of a file in any processing queue
*/
getFileState(filePath: string): UnifiedFileState | undefined {
return this.pending.get(filePath) ||
return this.discovered.get(filePath) ||
this.lightProcessing.get(filePath) ||
this.hashProcessing.get(filePath);
}
@@ -284,7 +301,7 @@ export class UnifiedProcessingStateManager {
* Clear all states (useful for testing or reset)
*/
clear(): void {
this.pending.clear();
this.discovered.clear();
this.lightProcessing.clear();
this.hashProcessing.clear();
this.done = [];
@@ -295,11 +312,11 @@ export class UnifiedProcessingStateManager {
* Get the total number of files in all states
*/
getTotalFileCount(): number {
return this.pending.size + this.lightProcessing.size + this.hashProcessing.size + this.done.length;
return this.discovered.size + this.lightProcessing.size + this.hashProcessing.size + this.done.length;
}
/**
* Retry processing for a file (move back to pending with incremented retry count)
* Retry processing for a file (move back to discovered with incremented retry count)
*/
retryProcessing(filePath: string): boolean {
const lightState = this.lightProcessing.get(filePath);
@@ -315,9 +332,9 @@ export class UnifiedProcessingStateManager {
this.lightProcessing.delete(filePath);
this.hashProcessing.delete(filePath);
this.pending.set(filePath, {
this.discovered.set(filePath, {
filePath,
state: 'pending',
state: 'discovered',
discoveredAt: state.discoveredAt,
retryCount,
lastRetryAt: Date.now()
@@ -332,8 +349,16 @@ export class UnifiedProcessingStateManager {
getRetryCount(filePath: string): number {
const lightState = this.lightProcessing.get(filePath);
const hashState = this.hashProcessing.get(filePath);
const pendingState = this.pending.get(filePath);
return lightState?.retryCount || hashState?.retryCount || pendingState?.retryCount || 0;
const discoveredState = this.discovered.get(filePath);
return lightState?.retryCount || hashState?.retryCount || discoveredState?.retryCount || 0;
}
/**
* Get all files currently discovered (not yet processing)
* Returns the internal Map for direct iteration
*/
getDiscoveredFiles(): Map<string, UnifiedFileState> {
return this.discovered;
}
/**
packages/meta-sort-core/src/logic/WatchedFileProcessor.ts
+9 -2
View File
@@ -285,9 +285,16 @@ export class WatchedFileProcessor implements FileProcessorInterface {
}
}
markDiscovered(filePath: string): void {
// STATE 1: Mark as discovered when file is found (before queue)
this.unifiedStateManager.addDiscovered(filePath);
}
/**
* @deprecated Use markDiscovered instead
*/
markPending(filePath: string): void {
// STATE 1: Mark as pending when file is discovered (before queue)
this.unifiedStateManager.addPending(filePath);
this.markDiscovered(filePath);
}
async queueFile(filePath: string): Promise<void> {
packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts
+13 -4
View File
@@ -157,6 +157,9 @@ export class StreamingPipeline {
for await (const filePath of discoveryStream) {
this.discoveredCount++;
// Mark as discovered immediately when file is found
this.config.stateManager.addDiscovered(filePath);
// Stage 1: Validation (fire and forget)
this.validationQueue.add(() => this.validateFile(filePath))
.catch(err => console.error(`[Pipeline] Validation error for ${filePath}:`, err.message));
@@ -181,14 +184,16 @@ export class StreamingPipeline {
* Stage 1: Validation
* - Quick extension check (no file I/O)
* - Optional MIME validation (reads file header)
* - Mark file as pending
* - Files already marked as discovered in start()
* - After validation passes, file is removed from discovered (tracked by PQueue pending)
*/
private async validateFile(filePath: string): Promise<void> {
try {
// Extension check (instant, no I/O)
const ext = path.extname(filePath).toLowerCase();
if (!this.config.supportedExtensions.has(ext)) {
// Skip unsupported file types
// Skip unsupported file types - remove from discovered
this.config.stateManager.removeFile(filePath);
return;
}
@@ -196,15 +201,19 @@ export class StreamingPipeline {
// Note: Skipped for performance - extension check is sufficient
// If needed, could add MIME validation via external library
// Mark as pending
this.config.stateManager.addPending(filePath);
this.validatedCount++;
// Remove from discovered - file will be tracked by PQueue pending
// until a fast queue worker picks it up and calls startLightProcessing()
this.config.stateManager.removeFile(filePath);
// Stage 2: Fast Queue - metadata + midhash256 (fire and forget)
this.fastQueue.add(() => this.processLightPhase(filePath))
.catch(err => console.error(`[Pipeline] Fast queue error for ${filePath}:`, err.message));
} catch (error: any) {
console.error(`[Pipeline] Validation failed for ${filePath}:`, error.message);
// Remove failed file from discovered
this.config.stateManager.removeFile(filePath);
}
}
packages/meta-sort-core/src/plugin-engine/TaskScheduler.ts
+18
View File
@@ -574,6 +574,24 @@ export class TaskScheduler extends EventEmitter {
};
}
/**
* Get currently running tasks with plugin info
* Delegates to container plugin scheduler
*/
getRunningTasks(): Array<{
taskId: string;
pluginId: string;
fileHash: string;
filePath: string;
queue: TaskQueueType;
startTime?: number;
}> {
if (this.containerPluginScheduler) {
return this.containerPluginScheduler.getRunningTasks();
}
return [];
}
// =========================================================================
// Shutdown
// =========================================================================
packages/meta-sort-ui/src/pages/Monitor.tsx
+175 -75
View File
@@ -4,7 +4,8 @@ import {
Metrics,
RedisStats,
QueueItem,
HashTimingStats
HashTimingStats,
FailedFile
} from '../types';
import {
formatMs,
@@ -19,14 +20,16 @@ function Monitor() {
const [metrics, setMetrics] = useState<Metrics | null>(null);
const [redisStats, setRedisStats] = useState<RedisStats | null>(null);
const [queue, setQueue] = useState<QueueItem[]>([]);
const [failedFiles, setFailedFiles] = useState<FailedFile[]>([]);
const fetchData = useCallback(async () => {
try {
const [statusRes, metricsRes, queueRes, statsRes] = await Promise.all([
const [statusRes, metricsRes, queueRes, statsRes, failedRes] = await Promise.all([
fetch('/api/processing/status'),
fetch('/api/metrics'),
fetch('/api/processing/queue'),
fetch('/api/stats')
fetch('/api/stats'),
fetch('/api/processing/failed')
]);
if (statusRes.ok) setStatus(await statusRes.json());
@@ -36,6 +39,10 @@ function Monitor() {
setQueue(queueData.items || []);
}
if (statsRes.ok) setRedisStats(await statsRes.json());
if (failedRes.ok) {
const failedData = await failedRes.json();
setFailedFiles(failedData.failedFiles || []);
}
} catch (err) {
console.error('Failed to fetch data:', err);
}
@@ -56,26 +63,52 @@ function Monitor() {
}
};
// Calculate progress for 5-stage pipeline
const retryFile = async (filePath: string) => {
try {
await fetch('/api/processing/retry', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ filePath })
});
fetchData();
} catch (err) {
console.error('Failed to retry file:', err);
}
};
const retryAllFailed = async () => {
try {
await fetch('/api/processing/retry-all', { method: 'POST' });
fetchData();
} catch (err) {
console.error('Failed to retry all files:', err);
}
};
// Calculate progress for 6-stage pipeline (including failed)
// Discovered = files waiting for fast queue (validated but not yet processing)
const getProgress = () => {
if (!status) return { pending: 0, preProcess: 0, fast: 0, background: 0, done: 0, total: 0 };
if (!status) return { discovered: 0, fast: 0, awaitingBackground: 0, background: 0, done: 0, failed: 0, total: 0 };
const computed = status.computed;
const preProcessCount = computed?.preProcessRunning || 0;
const fastCount = (computed?.fastQueueRunning || 0) + (computed?.fastQueuePending || 0);
const backgroundCount = (computed?.backgroundQueueRunning || 0) + (computed?.backgroundQueuePending || 0);
const discoveredCount = status.totalDiscovered || 0; // Files waiting for fast queue
const fastRunningCount = computed?.fastQueueRunning || 0;
const awaitingBackgroundCount = status.awaitingBackground || 0;
const backgroundRunningCount = computed?.backgroundQueueRunning || 0;
const failedCount = status.totalFailed || 0;
// Total = pending + all processing stages + done
const total = status.totalPending + preProcessCount + fastCount + backgroundCount + status.totalDone;
// Total = discovered + fast running + awaiting bg + bg running + done + failed
const total = discoveredCount + fastRunningCount + awaitingBackgroundCount + backgroundRunningCount + status.totalDone + failedCount;
if (total === 0) return { pending: 0, preProcess: 0, fast: 0, background: 0, done: 100, total: 0 };
if (total === 0) return { discovered: 0, fast: 0, awaitingBackground: 0, background: 0, done: 100, failed: 0, total: 0 };
return {
pending: (status.totalPending / total) * 100,
preProcess: (preProcessCount / total) * 100,
fast: (fastCount / total) * 100,
background: (backgroundCount / total) * 100,
discovered: (discoveredCount / total) * 100,
fast: (fastRunningCount / total) * 100,
awaitingBackground: (awaitingBackgroundCount / total) * 100,
background: (backgroundRunningCount / total) * 100,
done: (status.totalDone / total) * 100,
failed: (failedCount / total) * 100,
total
};
};
@@ -128,32 +161,25 @@ function Monitor() {
</button>
</div>
{/* Pipeline Progress Bar - 5 Stage */}
{/* Pipeline Progress Bar - 6 Stage (including failed) */}
{status && (
<div className="card pipeline-section">
<h2>Processing Pipeline</h2>
<div className="pipeline-bar-container">
<div className="pipeline-bar">
<div className="bar-segment pending" style={{ width: `${progress.pending}%` }} />
<div className="bar-segment preprocess" style={{ width: `${progress.preProcess}%` }} />
<div className="bar-segment discovered" style={{ width: `${progress.discovered}%` }} />
<div className="bar-segment fast" style={{ width: `${progress.fast}%` }} />
<div className="bar-segment awaiting-background" style={{ width: `${progress.awaitingBackground}%` }} />
<div className="bar-segment background" style={{ width: `${progress.background}%` }} />
<div className="bar-segment done" style={{ width: `${progress.done}%` }} />
<div className="bar-segment failed" style={{ width: `${progress.failed}%` }} />
</div>
</div>
<div className="pipeline-stats">
<div className="pipeline-stat pending">
<div className="pipeline-stat discovered">
<span className="stat-dot"></span>
<span className="stat-label">Pending</span>
<span className="stat-value">{formatNumber(status.totalPending)}</span>
</div>
<div className="pipeline-stat preprocess">
<span className="stat-dot"></span>
<span className="stat-label">Validation</span>
<span className="stat-value">
{queueStats.preProcess.running}/{queueStats.preProcess.max}
</span>
{queueStats.preProcess.paused && <span className="paused-indicator">paused</span>}
<span className="stat-label">Discovered</span>
<span className="stat-value">{formatNumber(status.totalDiscovered)}</span>
</div>
<div className="pipeline-stat fast">
<span className="stat-dot"></span>
@@ -163,9 +189,14 @@ function Monitor() {
</span>
{queueStats.fast.paused && <span className="paused-indicator">paused</span>}
</div>
<div className="pipeline-stat awaiting-background">
<span className="stat-dot"></span>
<span className="stat-label">Awaiting BG</span>
<span className="stat-value">{formatNumber(status.awaitingBackground || 0)}</span>
</div>
<div className="pipeline-stat background">
<span className="stat-dot"></span>
<span className="stat-label">Background Queue</span>
<span className="stat-label">Background</span>
<span className="stat-value">
{queueStats.background.running}/{queueStats.background.max}
</span>
@@ -176,11 +207,19 @@ function Monitor() {
<span className="stat-label">Complete</span>
<span className="stat-value">{formatNumber(status.totalDone)}</span>
</div>
{(status.totalFailed || 0) > 0 && (
<div className="pipeline-stat failed">
<span className="stat-dot"></span>
<span className="stat-label">Failed</span>
<span className="stat-value">{formatNumber(status.totalFailed || 0)}</span>
</div>
)}
</div>
{progress.total > 0 && (
<div className="pipeline-summary">
{formatNumber(progress.total)} total files |{' '}
{((status.totalDone / progress.total) * 100).toFixed(1)}% complete
{(status.totalFailed || 0) > 0 && ` | ${status.totalFailed} failed`}
</div>
)}
</div>
@@ -189,41 +228,6 @@ function Monitor() {
{/* Queue Status Cards - StreamingPipeline stages */}
{status && (
<div className="queue-cards">
<div className="queue-card preprocess">
<div className="queue-card-header">
<span className="queue-icon">✓</span>
<span className="queue-name">Validation</span>
</div>
<div className="queue-card-body">
<div className="queue-stat">
<span className="queue-stat-label">Running</span>
<span className="queue-stat-value">{queueStats.preProcess.running}/{queueStats.preProcess.max}</span>
</div>
<div className="queue-stat">
<span className="queue-stat-label">Pending</span>
<span className="queue-stat-value">{queueStats.preProcess.pending}</span>
</div>
<div className="queue-progress-bar">
<div
className="queue-progress-fill"
style={{ width: `${(queueStats.preProcess.running / queueStats.preProcess.max) * 100}%` }}
/>
</div>
<div className="queue-status">
{queueStats.preProcess.paused ? (
<span className="status-paused">⏸ Paused</span>
) : queueStats.preProcess.running > 0 ? (
<span className="status-running">▶ Running</span>
) : (
<span className="status-idle">● Idle</span>
)}
</div>
</div>
<div className="queue-card-footer">
<span className="queue-description">Extension checks</span>
</div>
</div>
<div className="queue-card fast">
<div className="queue-card-header">
<span className="queue-icon">⚡</span>
@@ -254,9 +258,6 @@ function Monitor() {
)}
</div>
</div>
<div className="queue-card-footer">
<span className="queue-description">MidHash + FFmpeg + Plugins</span>
</div>
</div>
<div className="queue-card background">
@@ -289,9 +290,6 @@ function Monitor() {
)}
</div>
</div>
<div className="queue-card-footer">
<span className="queue-description">SHA-256 full file hash</span>
</div>
</div>
</div>
)}
@@ -457,6 +455,47 @@ function Monitor() {
)}
</div>
{/* Failed Files */}
{failedFiles.length > 0 && (
<div className="card failed-files-section">
<div className="card-header-with-action">
<h2>Failed Files ({failedFiles.length})</h2>
<button className="btn btn-small btn-warning" onClick={retryAllFailed}>
Retry All
</button>
</div>
<table>
<thead>
<tr>
<th>File</th>
<th>Reason</th>
<th>Retries</th>
<th>Action</th>
</tr>
</thead>
<tbody>
{failedFiles.slice(0, 100).map((file, i) => (
<tr key={i}>
<td className="file-path" title={file.filePath}>{getFilename(file.filePath)}</td>
<td className="error-reason" title={file.reason}>
{file.reason.length > 50 ? file.reason.substring(0, 50) + '...' : file.reason}
</td>
<td>{file.retryCount}</td>
<td>
<button
className="btn btn-small btn-retry"
onClick={() => retryFile(file.filePath)}
>
Retry
</button>
</td>
</tr>
))}
</tbody>
</table>
</div>
)}
{/* Watched Folders */}
{status && status.watchedFolders && (
<div className="card">
@@ -510,7 +549,7 @@ function Monitor() {
transition: width 0.3s ease;
}
.bar-segment.pending {
.bar-segment.discovered {
background: linear-gradient(90deg, #6b7280, #9ca3af);
}
@@ -518,10 +557,18 @@ function Monitor() {
background: linear-gradient(90deg, #3b82f6, #60a5fa);
}
.bar-segment.awaiting-fast {
background: linear-gradient(90deg, #06b6d4, #22d3ee);
}
.bar-segment.fast {
background: linear-gradient(90deg, #f59e0b, #fbbf24);
}
.bar-segment.awaiting-background {
background: linear-gradient(90deg, #ec4899, #f472b6);
}
.bar-segment.background {
background: linear-gradient(90deg, #8b5cf6, #a78bfa);
}
@@ -530,6 +577,10 @@ function Monitor() {
background: linear-gradient(90deg, #10b981, #34d399);
}
.bar-segment.failed {
background: linear-gradient(90deg, #ef4444, #f87171);
}
.pipeline-stats {
display: flex;
justify-content: space-around;
@@ -549,11 +600,14 @@ function Monitor() {
border-radius: 50%;
}
.pipeline-stat.pending .stat-dot { background: #9ca3af; }
.pipeline-stat.discovered .stat-dot { background: #9ca3af; }
.pipeline-stat.preprocess .stat-dot { background: #60a5fa; }
.pipeline-stat.awaiting-fast .stat-dot { background: #22d3ee; }
.pipeline-stat.fast .stat-dot { background: #fbbf24; }
.pipeline-stat.awaiting-background .stat-dot { background: #f472b6; }
.pipeline-stat.background .stat-dot { background: #a78bfa; }
.pipeline-stat.done .stat-dot { background: #34d399; }
.pipeline-stat.failed .stat-dot { background: #f87171; }
.pipeline-stat .stat-label {
color: var(--text-secondary);
@@ -565,11 +619,14 @@ function Monitor() {
font-size: 1.1rem;
}
.pipeline-stat.pending .stat-value { color: #9ca3af; }
.pipeline-stat.discovered .stat-value { color: #9ca3af; }
.pipeline-stat.preprocess .stat-value { color: #60a5fa; }
.pipeline-stat.awaiting-fast .stat-value { color: #22d3ee; }
.pipeline-stat.fast .stat-value { color: #fbbf24; }
.pipeline-stat.awaiting-background .stat-value { color: #f472b6; }
.pipeline-stat.background .stat-value { color: #a78bfa; }
.pipeline-stat.done .stat-value { color: #34d399; }
.pipeline-stat.failed .stat-value { color: #f87171; }
.paused-indicator {
font-size: 0.75rem;
@@ -589,7 +646,7 @@ function Monitor() {
/* Queue Status Cards */
.queue-cards {
display: grid;
grid-template-columns: repeat(3, 1fr);
grid-template-columns: repeat(2, 1fr);
gap: 16px;
margin-bottom: 24px;
}
@@ -941,6 +998,11 @@ function Monitor() {
color: #a78bfa;
}
.stage-plugin {
background: rgba(6, 182, 212, 0.15);
color: #22d3ee;
}
.plugin-badge {
display: inline-block;
margin-left: 8px;
@@ -972,6 +1034,44 @@ function Monitor() {
margin-bottom: 8px;
font-family: monospace;
}
/* Failed Files Section */
.failed-files-section {
border-left: 4px solid #ef4444;
}
.failed-files-section h2 {
color: #f87171;
}
.error-reason {
color: #f87171;
font-size: 0.85rem;
max-width: 300px;
overflow: hidden;
text-overflow: ellipsis;
white-space: nowrap;
}
.btn-warning {
background: linear-gradient(135deg, #f59e0b, #d97706);
color: white;
}
.btn-warning:hover {
background: linear-gradient(135deg, #d97706, #b45309);
}
.btn-retry {
background: linear-gradient(135deg, #3b82f6, #2563eb);
color: white;
padding: 4px 8px;
font-size: 0.75rem;
}
.btn-retry:hover {
background: linear-gradient(135deg, #2563eb, #1d4ed8);
}
`}</style>
</div>
);
packages/meta-sort-ui/src/pages/Mounts.tsx
+46 -10
View File
@@ -11,6 +11,9 @@ interface MountConfig {
nfsPath?: string;
smbServer?: string;
smbShare?: string;
smbUsername?: string;
smbPassword?: string;
smbDomain?: string;
rcloneRemote?: string;
rclonePath?: string;
}
@@ -266,15 +269,7 @@ function Mounts() {
</select>
</div>
<div className="form-group">
<label>Mount Path</label>
<input
type="text"
value={newMount.mountPath || ''}
onChange={(e) => setNewMount({ ...newMount, mountPath: e.target.value })}
placeholder="/mnt/remote"
/>
</div>
<p className="form-hint">Mount path will be: /files/{newMount.name ? newMount.name.toLowerCase().replace(/[^a-z0-9-_]/g, '-').slice(0, 64) : '{name}'}</p>
{newMount.type === 'rclone' && (
<>
@@ -345,6 +340,33 @@ function Mounts() {
placeholder="media"
/>
</div>
<div className="form-group">
<label>Username (optional)</label>
<input
type="text"
value={newMount.smbUsername || ''}
onChange={(e) => setNewMount({ ...newMount, smbUsername: e.target.value })}
placeholder="user"
/>
</div>
<div className="form-group">
<label>Password (optional)</label>
<input
type="password"
value={newMount.smbPassword || ''}
onChange={(e) => setNewMount({ ...newMount, smbPassword: e.target.value })}
placeholder="********"
/>
</div>
<div className="form-group">
<label>Domain (optional)</label>
<input
type="text"
value={newMount.smbDomain || ''}
onChange={(e) => setNewMount({ ...newMount, smbDomain: e.target.value })}
placeholder="WORKGROUP"
/>
</div>
</>
)}
@@ -522,9 +544,11 @@ function Mounts() {
bottom: 0;
background: rgba(0, 0, 0, 0.7);
display: flex;
align-items: center;
align-items: flex-start;
justify-content: center;
z-index: 1000;
overflow-y: auto;
padding: 40px 20px;
}
.modal {
@@ -533,6 +557,8 @@ function Mounts() {
border-radius: 12px;
width: 90%;
max-width: 500px;
max-height: none;
margin: auto;
}
.modal-header {
@@ -593,6 +619,16 @@ function Mounts() {
padding-top: 20px;
border-top: 1px solid var(--border-color);
}
.form-hint {
font-size: 0.85rem;
color: var(--text-secondary);
margin-bottom: 20px;
padding: 8px 12px;
background: var(--bg-tertiary);
border-radius: 6px;
font-family: monospace;
}
`}</style>
</div>
);
packages/meta-sort-ui/src/types/index.ts
+23 -3
View File
@@ -2,7 +2,7 @@
* Shared type definitions for meta-sort-ui
*/
export type FileState = 'pending' | 'lightProcessing' | 'hashProcessing' | 'done';
export type FileState = 'discovered' | 'lightProcessing' | 'hashProcessing' | 'done';
export interface FileStateInfo {
filePath: string;
@@ -75,14 +75,21 @@ export interface TaskSchedulerQueueStatus {
}
export interface ProcessingStatus {
pending: FileStateInfo[];
discovered: FileStateInfo[];
lightProcessing: FileStateInfo[];
hashProcessing: FileStateInfo[];
done: FileStateInfo[];
totalPending: number;
/** Files waiting for fast queue (validated but not yet processing) */
totalDiscovered: number;
totalLightProcessing: number;
totalHashProcessing: number;
totalDone: number;
/** Total number of failed files */
totalFailed?: number;
/** @deprecated Use totalDiscovered instead. Kept for backward compatibility. */
awaitingFastQueue: number;
/** Files that completed fast queue but are waiting for background queue */
awaitingBackground: number;
watchedFolders: string[];
fastQueueConcurrency?: number;
backgroundQueueConcurrency?: number;
@@ -93,6 +100,19 @@ export interface ProcessingStatus {
preProcessConcurrency?: number;
}
export interface FailedFile {
filePath: string;
reason: string;
timestamp: number;
retryCount: number;
stage: 'metadata' | 'hash' | 'processing';
}
export interface FailedFilesResponse {
failedFiles: FailedFile[];
totalFailed: number;
}
export interface HashTimingStats {
total: number;
count: number;