diff --git a/packages/meta-sort-core/src/kv/RedisClient.ts b/packages/meta-sort-core/src/kv/RedisClient.ts index 2983d9b..85c253c 100644 --- a/packages/meta-sort-core/src/kv/RedisClient.ts +++ b/packages/meta-sort-core/src/kv/RedisClient.ts @@ -595,6 +595,19 @@ export class RedisKVClient implements IKVClient { break; } console.error('[Redis] Stream consumer error:', error.message); + + // Handle NOGROUP error - consumer group was deleted (e.g., stream cleared) + // Recreate the consumer group to recover + if (error.message?.includes('NOGROUP')) { + console.log('[Redis] Consumer group was deleted, recreating...'); + try { + await this.initStreamConsumer(stream, group); + console.log('[Redis] Consumer group recreated, resuming consumption'); + } catch (initError: any) { + console.error('[Redis] Failed to recreate consumer group:', initError.message); + } + } + // Brief pause before retrying await new Promise(resolve => setTimeout(resolve, 1000)); } diff --git a/packages/meta-sort-core/src/logic/fileProcessor/FileProcessorPiscina.ts b/packages/meta-sort-core/src/logic/fileProcessor/FileProcessorPiscina.ts index 4012d55..97d36af 100644 --- a/packages/meta-sort-core/src/logic/fileProcessor/FileProcessorPiscina.ts +++ b/packages/meta-sort-core/src/logic/fileProcessor/FileProcessorPiscina.ts @@ -565,16 +565,8 @@ export class FileProcessorPiscina implements FileAnalyzerInterface{ } } else { console.error(`Error in hash processing for ${filePath}:`, e); - - // Mark hash processing as failed for real errors - if (this.unifiedStateManager) { - this.unifiedStateManager.completeHashProcessing( - filePath, - undefined, - undefined, - e instanceof Error ? e.message : String(e) - ); - } + // Note: StreamingPipeline handles error completion via its catch block + // to avoid double-counting. Do not call completeHashProcessing here. } throw e; } finally { diff --git a/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts b/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts index 773bde9..dafb590 100644 --- a/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts +++ b/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts @@ -23,6 +23,12 @@ export class StreamingPipeline { private fastProcessedCount = 0; private backgroundProcessedCount = 0; + // Reset generation counter - incremented on each reset + // Used to ignore callbacks from tasks started before the reset + private resetGeneration = 0; + // Track which generation each file was started in + private fileGenerations: Map = new Map(); + constructor(config: PipelineConfig) { this.config = config; @@ -108,6 +114,13 @@ export class StreamingPipeline { // Note: Skipped for performance - extension check is sufficient // If needed, could add MIME validation via external library + // Check generation before incrementing counter + // Reset may have happened during validation + if (this.fileGenerations.get(filePath) !== this.resetGeneration) { + this.config.stateManager.removeFile(filePath); + return; + } + this.validatedCount++; // Remove from discovered - file will be tracked by PQueue pending @@ -135,6 +148,13 @@ export class StreamingPipeline { * @param midhash256 - Pre-computed midhash256 from meta-core (if available) */ private async processLightPhase(filePath: string, midhash256?: string): Promise { + // Check if this file is from the current reset generation + const fileGeneration = this.fileGenerations.get(filePath); + if (fileGeneration !== this.resetGeneration) { + console.log(`[Pipeline] Ignoring stale light phase for ${filePath}`); + return; + } + try { // Call light processing with midhash256 from meta-core const current = this.fastProcessedCount + 1; @@ -171,8 +191,22 @@ export class StreamingPipeline { .catch(err => console.error(`[Pipeline] Background queue error for ${filePath}:`, err.message)); } + // Check generation again before incrementing counter + // Reset may have happened during processing + if (this.fileGenerations.get(filePath) !== this.resetGeneration) { + console.log(`[Pipeline] Ignoring stale fast queue completion for ${filePath}`); + return; + } + this.fastProcessedCount++; } catch (error: any) { + // Check if this file is from the current reset generation + const fileGeneration = this.fileGenerations.get(filePath); + if (fileGeneration !== this.resetGeneration) { + console.log(`[Pipeline] Ignoring stale fast queue failure for ${filePath}`); + return; + } + // Processing failed - mark as failed (manual retry available via UI) console.error(`[Pipeline] Fast queue processing failed for ${filePath}: ${error.message}`); @@ -196,6 +230,13 @@ export class StreamingPipeline { * - Mark processing as complete (if no container plugins) */ private async processHashPhase(filePath: string): Promise { + // Check if this file is from the current reset generation + const fileGeneration = this.fileGenerations.get(filePath); + if (fileGeneration !== this.resetGeneration) { + console.log(`[Pipeline] Ignoring stale hash phase for ${filePath}`); + return; + } + try { // Call hash processing const current = this.backgroundProcessedCount + 1; @@ -203,6 +244,13 @@ export class StreamingPipeline { await this.config.fileProcessor.processHashPhase(filePath, current, queueSize); + // Check generation again before incrementing counter + // Reset may have happened during processing + if (this.fileGenerations.get(filePath) !== this.resetGeneration) { + console.log(`[Pipeline] Ignoring stale hash phase completion for ${filePath}`); + return; + } + this.backgroundProcessedCount++; // If container plugins are NOT configured, mark file as complete immediately. @@ -216,9 +264,17 @@ export class StreamingPipeline { } catch { // Renaming rule may fail for incomplete metadata } + // Clean up generation tracking + this.fileGenerations.delete(filePath); this.config.stateManager.completeHashProcessing(filePath, hashId, virtualPath); } } catch (error: any) { + // Check generation again after processing (in case reset happened during processing) + if (this.fileGenerations.get(filePath) !== this.resetGeneration) { + console.log(`[Pipeline] Ignoring stale hash phase failure for ${filePath}`); + return; + } + // Processing failed - mark as failed (manual retry available via UI) console.error(`[Pipeline] Background queue processing failed for ${filePath}: ${error.message}`); @@ -230,6 +286,9 @@ export class StreamingPipeline { 'hash' ); + // Clean up generation tracking + this.fileGenerations.delete(filePath); + // Complete with error this.config.stateManager.completeHashProcessing(filePath, undefined, undefined, error.message); } @@ -241,6 +300,8 @@ export class StreamingPipeline { * @param midhash256 - Pre-computed midhash256 from meta-core (if available) */ async handleFileAdded(filePath: string, midhash256?: string): Promise { + // Track which generation this file was started in + this.fileGenerations.set(filePath, this.resetGeneration); // Process through validation stage, passing midhash256 this.validationQueue.add(() => this.validateFile(filePath, midhash256)) @@ -253,6 +314,8 @@ export class StreamingPipeline { * @param midhash256 - Pre-computed midhash256 from meta-core (if available) */ async handleFileChanged(filePath: string, midhash256?: string): Promise { + // Track which generation this file was started in + this.fileGenerations.set(filePath, this.resetGeneration); // Reprocess through validation stage this.config.stateManager.removeFile(filePath); @@ -325,9 +388,25 @@ export class StreamingPipeline { * Note: meta-fuse listens to file:events stream from meta-core for updates */ async reset(): Promise { - console.log('[Pipeline] Resetting counters and state...'); + // Increment reset generation FIRST - this invalidates all in-flight tasks + this.resetGeneration++; + console.log(`[Pipeline] Resetting counters and state (generation ${this.resetGeneration})...`); - // Reset counters + // Step 1: Clear all queues to cancel pending tasks + // This prevents in-flight tasks from incrementing counters after reset + this.validationQueue.clear(); + this.fastQueue.clear(); + this.backgroundQueue.clear(); + + // Clear container plugin scheduler if available + if (this.config.containerPluginScheduler) { + this.config.containerPluginScheduler.clear(); + } + + // Clear file generation tracking + this.fileGenerations.clear(); + + // Step 2: Reset counters this.discoveredCount = 0; this.validatedCount = 0; this.fastProcessedCount = 0; @@ -336,6 +415,9 @@ export class StreamingPipeline { // Clear state manager (discovered, processing, done states) this.config.stateManager.clear(); + // Reset performance metrics (including failed files counter) + performanceMetrics.reset(); + console.log('[Pipeline] Reset complete'); } @@ -427,8 +509,37 @@ export class StreamingPipeline { // Listen for file:complete events to transition files to done state scheduler.on('file:complete', async ({ fileHash, filePath }) => { + // Check if this file was started in the current reset generation + // If not, ignore this callback - the file was from a previous reset cycle + const fileGeneration = this.fileGenerations.get(filePath); + if (fileGeneration !== this.resetGeneration) { + console.log(`[Pipeline] Ignoring stale callback for ${filePath} (gen ${fileGeneration} vs current ${this.resetGeneration})`); + return; + } + + // Additional check: verify file is actually in hashProcessing state + // This handles the race condition where: + // 1. Old task starts with gen 1, enters hashProcessing + // 2. Reset clears state and increments to gen 2 + // 3. New task starts with same filePath, sets fileGenerations[filePath] = 2 + // 4. Old task completes, passes generation check (sees gen 2), but shouldn't complete + // The file's hash in hashProcessing state should match the callback's fileHash + const hashProcessingState = this.config.stateManager.getHashProcessingFiles().get(filePath); + if (!hashProcessingState) { + console.log(`[Pipeline] Ignoring orphan callback for ${filePath} (not in hashProcessing state)`); + return; + } + // Verify hash matches - if different, this is a stale callback for an old task + if (hashProcessingState.hash && hashProcessingState.hash !== fileHash) { + console.log(`[Pipeline] Ignoring stale callback for ${filePath} (hash mismatch: ${hashProcessingState.hash} vs ${fileHash})`); + return; + } + console.log(`[Pipeline] Container plugins complete for ${filePath} (${fileHash})`); + // Clean up generation tracking + this.fileGenerations.delete(filePath); + // Fetch metadata from Redis (container plugins write directly to Redis via /meta API) let virtualPath: string | undefined; let metadata: Record | undefined;