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

add api test and bugfix

Browse Source
This commit is contained in:
Pierre
2026-01-17 19:59:53 +01:00
parent e0fa692d73
commit 30d156494c
10 changed files with 1159 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/meta-sort-core/.mocharc.json
+5
View File
@@ -0,0 +1,5 @@
{
"spec": "dist/api/api.test.js",
"timeout": 30000,
"exit": true
}
packages/meta-sort-core/package.json
+1 -1
View File
@@ -9,7 +9,7 @@
"dev": "tsc --watch",
"start": "node dist/index.js",
"exec": "node dist/index.js",
"test": "mocha",
"test": "tsc -p tsconfig.test.json && mocha",
"lint": "eslint src --ext ts"
},
"license": "MIT",
packages/meta-sort-core/src/api/api.test.ts
+1042
View File
File diff suppressed because it is too large Load Diff
packages/meta-sort-core/src/config/EnvConfig.ts
+2 -10
View File
@@ -80,18 +80,12 @@ interface EnvConfig {
// TaskScheduler Configuration (Plugin Queue System)
// ========================================================================
/** Concurrency for fast plugin queue (plugins with <1s avg execution time). Default: 32 */
/** Concurrency for fast plugin queue. Default: 32 */
FAST_QUEUE_CONCURRENCY: number;
/** Concurrency for background plugin queue (plugins with >=1s avg execution time). Default: 8 */
/** Concurrency for background plugin queue. Default: 8 */
BACKGROUND_QUEUE_CONCURRENCY: number;
/** Threshold in ms to classify plugin as fast vs background. Default: 1000 */
FAST_THRESHOLD_MS: number;
/** Minimum samples before using measured timing for classification. Default: 10 */
MIN_SAMPLES_FOR_MEASUREMENT: number;
// ========================================================================
// Container Plugins Configuration
// All plugins run as Docker containers for isolation and flexibility
@@ -183,8 +177,6 @@ export const config: EnvConfig = {
// TaskScheduler Configuration
FAST_QUEUE_CONCURRENCY: parseInt(process.env.FAST_QUEUE_CONCURRENCY || "32", 10),
BACKGROUND_QUEUE_CONCURRENCY: parseInt(process.env.BACKGROUND_QUEUE_CONCURRENCY || "8", 10),
FAST_THRESHOLD_MS: parseInt(process.env.FAST_THRESHOLD_MS || "1000", 10),
MIN_SAMPLES_FOR_MEASUREMENT: parseInt(process.env.MIN_SAMPLES_FOR_MEASUREMENT || "10", 10),
// Container Plugins Configuration
CONTAINER_PLUGINS_CONFIG: process.env.CONTAINER_PLUGINS_CONFIG || '/app/plugins.yml',
packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts
+28
View File
@@ -90,8 +90,15 @@ export class ContainerPluginScheduler extends EventEmitter {
concurrency: options?.fastConcurrency ?? config.FAST_QUEUE_CONCURRENCY,
});
// Background queue starts paused - only runs when fast queue is idle
this.backgroundQueue = new PQueue({
concurrency: options?.backgroundConcurrency ?? config.BACKGROUND_QUEUE_CONCURRENCY,
autoStart: false,
});
// When fast queue becomes idle, start background queue
this.fastQueue.on('idle', () => {
this.onFastQueueIdle();
});
this.callbackUrl = options?.callbackUrl ?? `${config.CONTAINER_CALLBACK_URL}/api/plugins/callback`;
@@ -262,12 +269,28 @@ export class ContainerPluginScheduler extends EventEmitter {
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);
});
return true;
}
/**
* Called when fast queue becomes idle - start background queue if it has work
*/
private onFastQueueIdle(): void {
if (this.backgroundQueue.size > 0 || this.backgroundQueue.pending > 0) {
console.log('[ContainerScheduler] Fast queue idle, starting background queue');
this.backgroundQueue.start();
}
}
/**
* Execute a task (dispatch to container)
*/
@@ -709,11 +732,16 @@ export class ContainerPluginScheduler extends EventEmitter {
/**
* Resume queues
* Note: Background queue only starts if fast queue is idle (fast queue has priority)
*/
resume(): void {
this.fastQueue.start();
// Only start background if fast queue is empty
if (this.fastQueue.size === 0 && this.fastQueue.pending === 0) {
this.backgroundQueue.start();
}
// Otherwise, background will start when fast queue idle event fires
}
/**
* Dispatch tasks to all enabled plugins for a file
packages/meta-sort-core/src/index.ts
+10 -2
View File
@@ -121,8 +121,15 @@ const runProgressiveDiscovery = async () => {
// Note: Stremio updates happen automatically via incremental add/remove notifications
// during file processing (like FUSE), so no explicit refresh needed here
const triggerScan = async () => {
console.log('Triggering manual scan from API...');
console.log('[Scan] Triggering manual scan from API...');
// Reset pipeline counters and state before fresh scan
// Also publishes reset event to meta-fuse
await pipeline.reset();
console.log('[Scan] Starting fresh discovery...');
await runProgressiveDiscovery();
console.log('[Scan] Manual scan complete');
};
// Global error handlers - prevent crashes from unforeseen errors
@@ -183,10 +190,11 @@ process.on('SIGINT', async () => {
await kvManager.start();
await kvManager.waitForReady();
// Pass KV client to file processor
// Pass KV client to file processor and pipeline
const client = kvManager.getClient();
if (client) {
fileProcessor.setKVClient(client);
pipeline.setKVClient(client);
console.log(`[Startup] KV Manager ready (role: ${kvManager.isLeader() ? 'LEADER' : 'FOLLOWER'})`);
}
}
packages/meta-sort-core/src/logic/WatchedFileProcessor.ts
-4
View File
@@ -199,8 +199,6 @@ export class WatchedFileProcessor implements FileProcessorInterface {
this.taskScheduler = createTaskScheduler(pluginManager, performanceMetrics, {
fastQueueConcurrency: config.FAST_QUEUE_CONCURRENCY,
backgroundQueueConcurrency: config.BACKGROUND_QUEUE_CONCURRENCY,
fastThresholdMs: config.FAST_THRESHOLD_MS,
minSamplesForMeasurement: config.MIN_SAMPLES_FOR_MEASUREMENT,
});
// Set up event listeners for VFS updates
@@ -209,8 +207,6 @@ export class WatchedFileProcessor implements FileProcessorInterface {
console.log(`[WatchedFileProcessor] TaskScheduler initialized with:`);
console.log(` - Fast queue concurrency: ${config.FAST_QUEUE_CONCURRENCY}`);
console.log(` - Background queue concurrency: ${config.BACKGROUND_QUEUE_CONCURRENCY}`);
console.log(` - Fast threshold: ${config.FAST_THRESHOLD_MS}ms`);
console.log(` - Min samples for measurement: ${config.MIN_SAMPLES_FOR_MEASUREMENT}`);
}
/**
packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts
+59 -76
View File
@@ -1,12 +1,14 @@
import * as path from 'path';
import PQueue from 'p-queue';
import {PipelineConfig} from './PipelineConfig.js';
import {withTimeoutAndRetry, TimeoutConfig} from '../../utils/TimeoutWrapper.js';
import {performanceMetrics} from '../../metrics/PerformanceMetrics.js';
// Pub/sub channel for batched updates to meta-fuse
const UPDATE_CHANNEL = 'meta-sort:file:batch';
// Pub/sub channel for scan reset notification
const RESET_CHANNEL = 'meta-sort:scan:reset';
// Batch update interval (5 seconds)
const BATCH_INTERVAL_MS = 5000;
@@ -39,23 +41,6 @@ export class StreamingPipeline {
private batchBuffer: BatchChange[] = [];
private batchTimer: NodeJS.Timeout | null = null;
// Retry configuration
private readonly MAX_PROCESSING_RETRIES = 10;
// Fast queue timeout: Fast metadata extraction + midhash256
private readonly fastQueueTimeoutConfig: TimeoutConfig = {
baseTimeout: 120000, // 2 minutes base timeout
timeoutMultiplier: 1.5, // Increase by 1.5x each retry
maxTimeout: 600000 // Max 10 minutes
};
// Background queue timeout: Full file hashing over NAS (can be very slow)
private readonly backgroundQueueTimeoutConfig: TimeoutConfig = {
baseTimeout: 7200000, // 2 hours base timeout (for NAS)
timeoutMultiplier: 1.5, // Increase by 1.5x each retry
maxTimeout: 14400000 // Max 4 hours
};
constructor(config: PipelineConfig) {
this.config = config;
@@ -223,23 +208,14 @@ export class StreamingPipeline {
* - Compute midhash256 (< 1s)
* - Store to KV (file becomes accessible in VFS)
* - Queue for background processing
* - Includes timeout and retry logic
*/
private async processLightPhase(filePath: string): Promise<void> {
try {
// Get current retry count
const retryCount = this.config.stateManager.getRetryCount(filePath);
// Call light processing with timeout
// Call light processing
const current = this.fastProcessedCount + 1;
const queueSize = this.discoveredCount;
await withTimeoutAndRetry(
() => this.config.fileProcessor.processLightPhase(filePath, current, queueSize),
retryCount,
this.fastQueueTimeoutConfig,
`Fast queue processing file ${filePath}`
);
await this.config.fileProcessor.processLightPhase(filePath, current, queueSize);
// After light processing, add to VFS
const metadata = this.config.fileProcessor.getDatabase().get(filePath);
@@ -250,7 +226,6 @@ export class StreamingPipeline {
if (virtualPath) {
// Add to VFS immediately (file appears with permanent midhash256 ID)
this.config.virtualFileSystem.addFile(virtualPath, filePath, metadata);
//console.log(`✓ File added to VFS: ${filePath} → ${virtualPath} (midhash256: ${metadata.cid_midhash256})`);
// Queue for pub/sub notification to meta-fuse
if (metadata.cid_midhash256) {
@@ -278,30 +253,14 @@ export class StreamingPipeline {
this.fastProcessedCount++;
} catch (error: any) {
// Handle timeout or processing error
const retryCount = this.config.stateManager.getRetryCount(filePath);
if (retryCount < this.MAX_PROCESSING_RETRIES) {
// Retry: move back to pending queue with incremented retry count
const success = this.config.stateManager.retryProcessing(filePath);
if (success) {
console.warn(`[Pipeline] Fast queue processing failed for ${filePath} (attempt ${retryCount + 1}/${this.MAX_PROCESSING_RETRIES}), retrying with longer timeout. Error: ${error.message}`);
// Re-queue for fast processing
this.fastQueue.add(() => this.processLightPhase(filePath))
.catch(err => console.error(`[Pipeline] Retry failed for ${filePath}:`, err.message));
return;
}
}
// Max retries exceeded or retry failed - mark as permanently failed
console.error(`[Pipeline] Fast queue processing permanently failed for ${filePath} after ${retryCount + 1} attempts: ${error.message}`);
// Processing failed - mark as failed (manual retry available via UI)
console.error(`[Pipeline] Fast queue processing failed for ${filePath}: ${error.message}`);
// Record failed file in metrics
performanceMetrics.recordFailedFile(
filePath,
error.message || 'Unknown error',
retryCount + 1,
1,
'processing'
);
@@ -315,47 +274,25 @@ export class StreamingPipeline {
* - Compute full hashes (SHA-256, SHA-1, MD5, CRC32)
* - Update KV with additional hash metadata
* - Mark processing as complete
* - Includes timeout and retry logic
*/
private async processHashPhase(filePath: string): Promise<void> {
try {
// Get current retry count
const retryCount = this.config.stateManager.getRetryCount(filePath);
// Call hash processing with timeout
// Call hash processing
const current = this.backgroundProcessedCount + 1;
const queueSize = this.discoveredCount;
await withTimeoutAndRetry(
() => this.config.fileProcessor.processHashPhase(filePath, current, queueSize),
retryCount,
this.backgroundQueueTimeoutConfig,
`Background queue processing file ${filePath}`
);
await this.config.fileProcessor.processHashPhase(filePath, current, queueSize);
this.backgroundProcessedCount++;
} catch (error: any) {
// Handle timeout or processing error
const retryCount = this.config.stateManager.getRetryCount(filePath);
if (retryCount < this.MAX_PROCESSING_RETRIES) {
// Retry: move back to background queue with incremented retry count
console.warn(`[Pipeline] Background queue processing failed for ${filePath} (attempt ${retryCount + 1}/${this.MAX_PROCESSING_RETRIES}), retrying. Error: ${error.message}`);
// Re-queue for background processing
this.backgroundQueue.add(() => this.processHashPhase(filePath))
.catch(err => console.error(`[Pipeline] Retry failed for ${filePath}:`, err.message));
return;
}
// Max retries exceeded - mark as permanently failed
console.error(`[Pipeline] Background queue processing permanently failed for ${filePath} after ${retryCount + 1} attempts: ${error.message}`);
// Processing failed - mark as failed (manual retry available via UI)
console.error(`[Pipeline] Background queue processing failed for ${filePath}: ${error.message}`);
// Record failed file in metrics
performanceMetrics.recordFailedFile(
filePath,
error.message || 'Unknown error',
retryCount + 1,
1,
'hash'
);
@@ -432,6 +369,43 @@ export class StreamingPipeline {
};
}
/**
* Reset pipeline counters and clear state manager
* Called when triggering a fresh scan to reset statistics
* Publishes reset event to notify meta-fuse to clear its VFS
*/
async reset(): Promise<void> {
console.log('[Pipeline] Resetting counters and state...');
// Reset counters
this.discoveredCount = 0;
this.validatedCount = 0;
this.fastProcessedCount = 0;
this.backgroundProcessedCount = 0;
// Clear batch buffer
this.batchBuffer = [];
// Clear state manager (discovered, processing, done states)
this.config.stateManager.clear();
// Publish reset event to notify meta-fuse
if (this.config.kvClient && typeof (this.config.kvClient as any).publish === 'function') {
try {
const message = JSON.stringify({
timestamp: Date.now(),
action: 'reset'
});
await (this.config.kvClient as any).publish(RESET_CHANNEL, message);
console.log('[Pipeline] Reset event published to meta-fuse');
} catch (error) {
console.error('[Pipeline] Failed to publish reset event:', error);
}
}
console.log('[Pipeline] Reset complete');
}
/**
* Pause all pipeline queues
* Running tasks will complete, but no new tasks will start
@@ -520,6 +494,15 @@ export class StreamingPipeline {
console.log('[Pipeline] Container plugin scheduler connected');
}
/**
* Set the KV client (called after initialization when Redis is ready)
* This enables pub/sub notifications to meta-fuse
*/
setKVClient(kvClient: import('../../kv/IKVClient.js').IKVClient): void {
(this.config as any).kvClient = kvClient;
console.log('[Pipeline] KV client connected - pub/sub enabled');
}
/**
* Dispatch tasks to container plugins for metadata extraction
* This is fire-and-forget - container plugins write results directly to meta-core
packages/meta-sort-core/src/plugin-engine/TaskScheduler.ts
+3 -23
View File
@@ -18,7 +18,7 @@ import type {
PluginKVStore,
} from './types.js';
import type { PluginManager } from './PluginManager.js';
import type { PerformanceMetrics, PluginTimingStats } from '../metrics/PerformanceMetrics.js';
import type { PerformanceMetrics } from '../metrics/PerformanceMetrics.js';
import type { ContainerPluginScheduler } from '../container-plugins/index.js';
import { KVStore } from './PluginContext.js';
@@ -31,17 +31,11 @@ export interface TaskSchedulerConfig {
fastQueueConcurrency?: number;
/** Concurrency for background queue (default: 8) */
backgroundQueueConcurrency?: number;
/** Threshold in ms for fast classification (default: 1000) */
fastThresholdMs?: number;
/** Minimum samples before using measured timing (default: 10) */
minSamplesForMeasurement?: number;
}
const DEFAULT_CONFIG: Required<TaskSchedulerConfig> = {
fastQueueConcurrency: 32,
backgroundQueueConcurrency: 8,
fastThresholdMs: 1000,
minSamplesForMeasurement: 10,
};
// =============================================================================
@@ -379,25 +373,11 @@ export class TaskScheduler extends EventEmitter {
// =========================================================================
/**
* Classify a plugin as fast or background based on timing data and manifest
* Classify a plugin as fast or background based on manifest defaultQueue
*/
classifyPlugin(pluginId: string): TaskQueueType {
const plugins = this.pluginManager.getPlugins();
const plugin = plugins.find(p => p.id === pluginId);
const timing = this.performanceMetrics.getPluginTiming(pluginId);
// Get manifest default queue (need to access via plugin manager internals)
// For now, we'll use a workaround - check if plugin has defaultQueue in manifest
const manifest = this.getPluginManifest(pluginId);
const defaultQueue = manifest?.defaultQueue || 'fast';
// If we have enough samples, use measured timing
if (timing && timing.count >= this.config.minSamplesForMeasurement) {
return timing.average < this.config.fastThresholdMs ? 'fast' : 'background';
}
// Otherwise use manifest default
return defaultQueue;
return manifest?.defaultQueue || 'fast';
}
// =========================================================================
packages/meta-sort-core/tsconfig.test.json
+8
View File
@@ -0,0 +1,8 @@
{
"extends": "./tsconfig.json",
"compilerOptions": {
"outDir": "./dist"
},
"include": ["src/**/*"],
"exclude": ["node_modules", "src/plugin-engine/plugin.test.ts"]
}