meta-sort/docs/streaming-pipeline-architecture.md

Streaming Pipeline Architecture

The meta-sort streaming pipeline processes media files as they arrive from meta-core (the file watcher), classifies them by extension, extracts light metadata + a permanent identifier, and dispatches heavier work to container plugins. It is implemented in packages/meta-sort/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts.

---

Contents

1. Overview
2. Event source: meta-core SSE
3. Pipeline stages
4. State manager
5. Concurrency model
6. Reset semantics (generation counter)
7. Stats and observability
8. Key code locations

---

Overview

                  meta-core /api/events/files (SSE)
                              │
                              ▼
                   FileEventConsumer
                              │   add / change / delete / rename / reset
                              ▼
   handleFileAdded ─┐
   handleFileChanged┤───────► validationQueue (PQueue)
   handleFileDeleted┘                │   extension filter
                                     ▼
                              fastQueue (PQueue)
                          processLightPhase
                          - filename parse, FFprobe-light, stats
                          - midhash256 (permanent ID) computed or
                            adopted from meta-core's pre-computed value
                          - KV write → file appears in VFS
                          - dispatch all enabled container plugins
                                     │
                                     ▼
                              backgroundQueue (PQueue)
                          processHashPhase
                          - SHA-256, additional heavy hashing
                                     │
                                     ▼
                  await ContainerPluginScheduler `file:complete`
                  (or complete immediately if no container plugins)
                                     │
                                     ▼
                              UnifiedProcessingStateManager → DONE

Two things are worth highlighting because they differ from older documentation:

1. midhash256 is the permanent file ID. It is computed during the fast queue (or supplied pre-computed by meta-core in the SSE payload) and the file becomes accessible in the VFS as soon as the fast queue commits its KV write. There is no tempId and no rename when SHA-256 finishes.
2. There is no in-process plugin pipeline. Filename parsing, ffprobe, TMDB, etc. are container plugins running as separate Docker containers; StreamingPipeline dispatches to them via ContainerPluginScheduler.dispatchAllPlugins and waits for the scheduler's file:complete event before marking the file DONE.

---

Event source: meta-core SSE

meta-sort no longer reads the filesystem directly. File discovery and change watching live in meta-core, which streams events over HTTP SSE.

• Subscriber: src/events/FileEventConsumer.ts — converts each event to an absolute path (filesPath + relative path from meta-core) and forwards it to the pipeline.
• Transport: src/events/SSEEventClient.ts — long-lived HTTP connection to ${metaCoreApiUrl}/api/events/files, with cursor persistence at /meta-core/cursors/meta-sort-files.cursor, automatic reconnect, and Last-Event-ID resume semantics.
• Cursor gap (gap event): the cursor was trimmed out of retention. The default response is to log and wait for the next reset event from meta-core; the pipeline rebuilds from there.

Event types and their effect on the pipeline:

SSE event	Pipeline call	Notes
add	handleFileAdded(path, midhash256?)	Standard new-file path.
change	handleFileChanged(path, midhash256?)	Removes any prior state, re-validates.
delete	handleFileDeleted(path)	Drops state + calls fileProcessor.deleteFile.
rename	delete(oldPath) + add(newPath)	Old path is removed; new path is re-ingested.
reset	pipeline.reset()	Increments the reset generation, clears queues.

---

Pipeline stages

Stage 1: Validation (validateFile)

StreamingPipeline.validateFile(filePath, midhash256?)

• Extension check against PipelineConfig.supportedExtensions (src/config/SupportedFileTypes.ts). No I/O.
• Files with unsupported extensions are removed from the state manager and counted in filteredCount. They do not progress.
• Strict MIME validation is intentionally not implemented — the comment in code reserves the option but it is off by default and there is no external library wired in.
• A generation check protects against in-flight tasks completing after a reset() (see Reset semantics).
• Valid files are enqueued onto the fast queue via this.fastQueue.add(() => this.processLightPhase(filePath, midhash256)).

Stage 2: Fast queue (processLightPhase)

StreamingPipeline.processLightPhase(filePath, midhash256?) calls fileProcessor.processLightPhase(...) which:

1. Computes or adopts midhash256 (the permanent file ID, written as cid_midhash256 on the metadata record).
2. Extracts light metadata (filename parsing via @metazla/filename-tools, basic stat/ffprobe info — see src/logic/fileProcessor/ for the adapter, and the file-info / filename-parser / ffmpeg container plugins for the actual work).
3. Writes metadata to Redis through the KV client. The file is now visible to meta-fuse, meta-stremio, etc.

After processLightPhase returns, StreamingPipeline:

• Sends a fire-and-forget /add POST to the Stremio addon on http://localhost:7000 (silent fail if it isn't running).
• Calls dispatchContainerPluginTasks(filePath, hashId, existingMeta) — this iterates every enabled container plugin via ContainerPluginScheduler.dispatchAllPlugins. Tasks are routed to the scheduler's own fast / background queues based on each plugin manifest's defaultQueue. Plugins write results back through meta-core's /meta/:hash API; StreamingPipeline does not aggregate them itself.
• Enqueues the file on the background queue (processHashPhase) for the in-process hashing pass.

Stage 3: Background queue (processHashPhase)

StreamingPipeline.processHashPhase(filePath) calls fileProcessor.processHashPhase(...) for the slow in-process hashing (SHA-256 and any other heavy hashes the file processor computes).

Completion path depends on whether container plugins are configured:

• No container plugins: the file is moved to DONE immediately via stateManager.completeHashProcessing(filePath, hashId, virtualPath).
• Container plugins configured: the background queue completes, but the file stays in hashProcessing until the ContainerPluginScheduler emits file:complete for that (fileHash, filePath) pair. The scheduler-side listener (setContainerPluginScheduler in StreamingPipeline.ts) re-reads metadata from Redis, computes the virtual path via MetaDataToFolderStruct.renamingRule(...), and then transitions the file to DONE.

The scheduler listener also defends against three races: a different reset generation, a missing hashProcessing entry, and a hash-mismatch where the callback belongs to a previous attempt for the same path.

---

State manager

src/logic/UnifiedProcessingStateManager.ts

States: 'discovered' | 'lightProcessing' | 'hashProcessing' | 'done'.

discovered ── startLightProcessing ──► lightProcessing
                                              │
                                              │ completeLightProcessing
                                              ▼
                                       hashProcessing ── completeHashProcessing ──► done

Notable details:

• done is kept as a bounded array (maxDoneHistory = 100); a running totalProcessedCount survives that truncation.
• seenHashes: Set<string> + duplicateCount track per-content duplicate ingestion.
• UnifiedFileState tracks per-stage *StartedAt / *CompletedAt timestamps and derived *ProcessingTime values; this is what the /api/processing/status and /api/processing/queue endpoints expose.

---

Concurrency model

Three independent p-queue instances inside StreamingPipeline:

Queue	Field	Configured by
Validation	validationQueue	PipelineConfig.validationConcurrency
Fast (light phase)	fastQueue	PipelineConfig.fastQueueConcurrency
Background (hash phase)	backgroundQueue	PipelineConfig.backgroundQueueConcurrency

Defaults are derived in src/index.ts:

const cpuCount = os.cpus().length;
const defaultBackgroundWorkers = Math.max(1, Math.floor(cpuCount / 2));
const validationConcurrency = config.MAX_WORKER_THREADS
    ? config.MAX_WORKER_THREADS * 2 : cpuCount * 2;
const fastQueueConcurrency = config.MAX_WORKER_THREADS
    ?? cpuCount;
const backgroundQueueConcurrency = config.MAX_WORKER_THREADS
    ?? defaultBackgroundWorkers;

i.e. on a host with os.cpus().length = N:

• validationConcurrency = N * 2
• fastQueueConcurrency = N
• backgroundQueueConcurrency = floor(N / 2) (minimum 1)

Setting MAX_WORKER_THREADS=M overrides all three to (M * 2, M, M).

Note: container-plugin tasks do not share these queues — they are scheduled inside ContainerPluginScheduler (src/container-plugins/), whose own concurrencies come from FAST_QUEUE_CONCURRENCY (default 32) and the same backgroundQueueConcurrency value as above. See plugin-task-queue-architecture.md.

Pause / resume

pipeline.pause() / pipeline.resume() pause all three internal queues. The API server wires these to operator controls (see UnifiedAPIServer.setStreamingPipeline). In-flight tasks finish; no new ones start until resume.

---

Reset semantics (generation counter)

StreamingPipeline keeps a resetGeneration counter and a per-file fileGenerations: Map<string, number>. Every time reset() is called (triggered by a reset SSE event from meta-core, or by an operator action that fans out to one), the counter increments and the queues are cleared.

Each long-running phase checks fileGenerations.get(filePath) === resetGeneration at every yield point and bails out silently if the file belongs to an older generation. This is the primary defense against:

• Old tasks completing into a cleared state manager.
• Container plugin callbacks arriving for files that were since rescheduled (callback handler additionally verifies the hash matches the current hashProcessing entry).

The reset also resets discoveredCount, validatedCount, filteredCount, fastProcessedCount, backgroundProcessedCount, clears the ContainerPluginScheduler, and resets performanceMetrics.

---

Stats and observability

StreamingPipeline.getStats() returns:

• counters: discovered, validated, filtered, fastProcessed, backgroundProcessed
• per-queue { pending, size } for validation, fast, background — if a ContainerPluginScheduler is attached, the fast / background numbers come from the scheduler's queues (those are the user-meaningful numbers because that's where the plugin work actually runs)
• state: the full UnifiedProcessingStateManager.getSnapshot()

These are exposed via the API endpoints registered in src/api/UnifiedAPIServer.ts:

• GET /api/processing/status — counts + state snapshot
• GET /api/processing/queue — per-queue depths
• GET /api/processing/failed — failed-file roster (also recorded into performanceMetrics)
• POST /api/processing/retry and POST /api/processing/retry-all
• POST /api/processing/wait-empty — pauseless wait for the queues to drain (useful in tests)

---

Key code locations

Component	File
Pipeline orchestrator	src/logic/pipeline/StreamingPipeline.ts
Pipeline config	src/logic/pipeline/PipelineConfig.ts
State manager	src/logic/UnifiedProcessingStateManager.ts
File processor (light/hash phases)	src/logic/WatchedFileProcessor.ts, src/logic/fileProcessor/
SSE subscriber	src/events/FileEventConsumer.ts
SSE transport + cursor	src/events/SSEEventClient.ts
Container plugin dispatch	src/container-plugins/ContainerPluginScheduler.ts
Extension allow-list	src/config/SupportedFileTypes.ts
Concurrency defaults / startup wiring	src/index.ts
API endpoints	src/api/UnifiedAPIServer.ts