improve pipeline

This commit is contained in:
Pierre
2026-03-07 21:20:08 +01:00
parent ff62c9c76e
commit 07d8331be2
5 changed files with 151 additions and 19 deletions
@@ -396,10 +396,12 @@ 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
// "Discovered" now represents unique files waiting for fast queue (validated but not yet processing)
// Use file count from container scheduler if available, otherwise fall back to task count
const filesStatus = queueStatus.files;
const uniqueFilesWaiting = filesStatus ? (filesStatus.fastWaiting + filesStatus.fastRunning) : fastQueuePending;
snapshot.totalDiscovered = uniqueFilesWaiting;
snapshot.awaitingFastQueue = uniqueFilesWaiting; // Keep for backward compatibility
// Fix awaitingBackground: subtract files currently being processed by background workers
// hashProcessing.size includes both waiting AND running files
@@ -444,6 +446,14 @@ export class UnifiedAPIServer {
const metrics = performanceMetrics.getMetrics();
snapshot.totalFailed = metrics.totalFailedFiles || 0;
// Add filtered count from pipeline stats
if (this.streamingPipeline) {
const pipelineStats = this.streamingPipeline.getStats();
snapshot.totalFiltered = pipelineStats.filtered || 0;
} else {
snapshot.totalFiltered = 0;
}
return snapshot;
});
@@ -57,6 +57,14 @@ export interface UnifiedProcessingSnapshot {
/** Files that completed fast queue but are waiting for background queue */
awaitingBackground: number;
// Filtering/duplicate metrics
/** Files filtered out due to unsupported extensions (non-media files) */
totalFiltered?: number;
/** Files with same content hash as previously processed file */
totalDuplicates?: number;
/** Number of unique content hashes seen */
uniqueHashes?: number;
// Configuration
fastQueueConcurrency?: number; // Number of fast queue workers
backgroundQueueConcurrency?: number; // Number of background queue workers
@@ -70,6 +78,10 @@ export class UnifiedProcessingStateManager {
private maxDoneHistory = 100; // Keep last 100 completed files
private totalProcessedCount = 0; // Track total number of completed files (not limited)
// Duplicate tracking
private seenHashes: Set<string> = new Set();
private duplicateCount = 0; // Files with same hash as a previously processed file
/**
* Add a file to discovered state (file discovered)
*/
@@ -126,6 +138,13 @@ export class UnifiedProcessingStateManager {
return;
}
// Track duplicates - if hash already seen, increment duplicate counter
if (this.seenHashes.has(hash)) {
this.duplicateCount++;
} else {
this.seenHashes.add(hash);
}
// Move to hash processing queue
this.hashProcessing.set(filePath, {
filePath,
@@ -273,6 +292,10 @@ export class UnifiedProcessingStateManager {
// Files that completed fast queue but are waiting for/in background queue
awaitingBackground: this.hashProcessing.size,
// Duplicate tracking (filtered count is added by API layer from pipeline stats)
totalDuplicates: this.duplicateCount,
uniqueHashes: this.seenHashes.size,
fastQueueConcurrency,
backgroundQueueConcurrency
};
@@ -296,6 +319,8 @@ export class UnifiedProcessingStateManager {
this.hashProcessing.clear();
this.done = [];
this.totalProcessedCount = 0;
this.seenHashes.clear();
this.duplicateCount = 0;
}
/**
@@ -366,4 +391,14 @@ export class UnifiedProcessingStateManager {
getHashProcessingFiles(): Map<string, UnifiedFileState> {
return this.hashProcessing;
}
/**
* Get duplicate tracking statistics
*/
getDuplicateStats(): { duplicateCount: number; uniqueHashes: number } {
return {
duplicateCount: this.duplicateCount,
uniqueHashes: this.seenHashes.size
};
}
}
@@ -20,6 +20,7 @@ export class StreamingPipeline {
private discoveredCount = 0;
private validatedCount = 0;
private filteredCount = 0; // Files filtered out (unsupported extensions)
private fastProcessedCount = 0;
private backgroundProcessedCount = 0;
@@ -106,6 +107,7 @@ export class StreamingPipeline {
const ext = path.extname(filePath).toLowerCase();
if (!this.config.supportedExtensions.has(ext)) {
// Skip unsupported file types - remove from discovered
this.filteredCount++;
this.config.stateManager.removeFile(filePath);
return;
}
@@ -353,6 +355,7 @@ export class StreamingPipeline {
return {
discovered: this.discoveredCount,
validated: this.validatedCount,
filtered: this.filteredCount,
fastProcessed: this.fastProcessedCount,
backgroundProcessed: this.backgroundProcessedCount,
queues: {
@@ -396,6 +399,7 @@ export class StreamingPipeline {
// Step 2: Reset counters
this.discoveredCount = 0;
this.validatedCount = 0;
this.filteredCount = 0;
this.fastProcessedCount = 0;
this.backgroundProcessedCount = 0;