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

leader discovery fix and code cleaning

Browse Source
This commit is contained in:
Pierre
2026-02-01 13:39:42 +01:00
parent f220972cb2
commit 34ff1375c6
13 changed files with 4961 additions and 878 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/containerized-plugin-architecture.md
+1159
View File
File diff suppressed because it is too large Load Diff
docs/fast-hash-global-id.md
+781
View File
@@ -0,0 +1,781 @@
# MidHash256 Global ID Architecture
## Document Purpose
This document describes the architectural design of the midhash256 global identifier system for Meta-Mesh file metadata. MidHash256 is a fast-hash algorithm that computes SHA-256 of the middle 1MB of a file plus its size. This document explains the algorithm, rationale, performance characteristics, uniqueness guarantees, and trade-offs compared to full-file SHA-256.
---
## System Overview
Meta-Mesh uses **content-addressable storage** where each unique file is identified by a cryptographic hash of its content. This global identifier enables deduplication, distributed synchronization, and metadata consistency across the system.
### Current Implementation
The existing system uses **full-file SHA-256** as the global identifier:
```typescript
// Current: Hash entire file
const hash = await computeSHA256(filePath); // Reads 100% of file
// Result: "sha256:a3f2bc9..."
```
**Properties:**
- **Uniqueness**: Cryptographically strong (collision probability ≈ 2^-256)
- **Consistency**: Identical content always produces identical hash
- **Performance**: Slow for large files (reads entire file, CPU-intensive)
### Problem Statement
For large media files (10GB-50GB video files), computing full-file SHA-256 creates significant bottlenecks:
1. **Processing Time**: Reading and hashing 50GB takes 15-20 minutes per file
2. **Resource Intensive**: CPU-bound SHA-256 computation at full file size
3. **Pipeline Bottleneck**: Hash computation is the slowest stage in the processing pipeline
4. **Delayed Availability**: Files cannot be fully indexed until hash completes
5. **ID Rotation Complexity**: Temporary IDs (tempId) are used during processing, requiring complex ID rotation logic when hash completes
### Core Design Principle
**A global identifier should be fast to compute while maintaining practical uniqueness guarantees for the target domain (personal media libraries).**
This design prioritizes:
- **Speed**: Sub-second hash computation for files of any size
- **Practicality**: Sufficient uniqueness for real-world media collections (millions of files)
- **Simplicity**: Single-phase processing eliminates temporary ID rotation
- **Determinism**: Identical content produces identical hash
---
## Architecture Design
### Fast-Hash Algorithm
The fast-hash algorithm computes a SHA-256 hash of a **sampled portion** of the file rather than the entire content. This provides a fast, deterministic, and practically unique identifier.
#### Algorithm Specification
```
computeMidHash256(filePath) -> CID:
1. fileSize = getFileSize(filePath)
2. sampleData = extractSample(filePath, fileSize)
3. hashInput = concat(uint64_BE(fileSize), sampleData)
4. hashBuffer = SHA256(hashInput)
5. digest = createDigest(0x1000, hashBuffer)
6. return CID.createV1(0x1000, digest)
extractSample(filePath, fileSize) -> bytes:
if fileSize <= 1MB:
return readEntireFile(filePath)
else:
middleOffset = (fileSize - 1MB) / 2
return readBytes(filePath, offset=middleOffset, length=1MB)
```
#### Detailed Steps
**Step 1: File Size Measurement**
```typescript
const fileSize = (await fs.stat(filePath)).size;
```
- Use file system metadata (instant, no file read required)
- Size is measured in bytes (64-bit integer)
**Step 2: Sample Extraction**
For files **≤ 1MB**:
```typescript
const sampleData = await fs.readFile(filePath);
// Read entire file (small enough to process quickly)
```
For files **> 1MB**:
```typescript
const middleOffset = Math.floor((fileSize - SAMPLE_SIZE) / 2);
const SAMPLE_SIZE = 1024 * 1024; // 1MB
const fd = await fs.open(filePath, 'r');
const buffer = Buffer.allocUnsafe(SAMPLE_SIZE);
await fd.read(buffer, 0, SAMPLE_SIZE, middleOffset);
await fd.close();
```
- Extract exactly 1MB from the middle of the file
- **Middle sampling rationale**: Avoids headers (start) and footers (end) which may be identical across files in the same format
**Step 3: Hash Input Construction**
```typescript
// Prepend file size as 64-bit big-endian integer
const sizeBuffer = Buffer.allocUnsafe(8);
sizeBuffer.writeBigUInt64BE(BigInt(fileSize), 0);
// Concatenate: [fileSize (8 bytes)] + [sampleData (≤ 1MB)]
const hashInput = Buffer.concat([sizeBuffer, sampleData]);
```
**Why include file size:**
- **Uniqueness boost**: Files with identical middle 1MB but different sizes produce different hashes
- **Size change detection**: Modifying file size (even without changing middle content) changes hash
- **Minimal overhead**: 8 bytes added to hash input (negligible)
**Step 4: SHA-256 Computation**
```typescript
import { createHash } from 'crypto';
const hashBuffer = createHash('sha256')
.update(hashInput)
.digest();
const digest = create(0x1000, hashBuffer);
const midhash256 = CID.createV1(0x1000, digest).toString();
```
- SHA-256 is computed on ≤ 1MB + 8 bytes (maximum 1,048,584 bytes)
- Result is wrapped in CID v1 format with custom multicodec `0x1000`
- Base32 CID encoding produces strings like "bafkr4ih5kapbjzqvmj7jxr..."
- This format is homogeneous with other CID types (sha256, sha1, etc.)
#### Example Hash Computation
**Small File (500KB)**
```
File: documentary.mp4 (512,000 bytes)
1. fileSize = 512000
2. sampleData = [entire file, 512KB]
3. hashInput = [0x0000000000007D00] + [512KB of data]
4. hashBuffer = SHA256(hashInput)
5. digest = createDigest(0x1000, hashBuffer)
6. result = CID.createV1(0x1000, digest) = "bafkr4ih5kapbjzqvmj7jxr..."
```
**Large File (50GB)**
```
File: movie-4k.mkv (53,687,091,200 bytes)
1. fileSize = 53687091200
2. middleOffset = (53687091200 - 1048576) / 2 = 26,843,021,312
3. sampleData = read 1MB at offset 26,843,021,312
4. hashInput = [0x0000000C80000000] + [1MB from middle]
5. hashBuffer = SHA256(hashInput)
6. digest = createDigest(0x1000, hashBuffer)
7. result = CID.createV1(0x1000, digest) = "bafkr4if3dqvyqvuqfyfg..."
```
---
## Uniqueness Analysis
### Uniqueness Factors
The fast-hash provides uniqueness through multiple independent factors:
1. **File Size (64-bit)**: Exact byte count must match
2. **Middle Content (1MB)**: 1,048,576 bytes of content must match
3. **Content Position**: Content must be at the exact middle position
### Collision Scenarios
A collision occurs when two different files produce the same fast-hash. This requires:
```
FileA ≠ FileB
AND
size(FileA) == size(FileB)
AND
middle1MB(FileA) == middle1MB(FileB)
```
#### Scenario 1: Different Video Files (Same Source)
**Low Risk - Extremely Unlikely**
Two different episodes of a TV show encoded with the same settings:
- **File Size**: Different (episode durations vary, file sizes differ by seconds of content)
- **Middle Content**: Different (different scenes, audio, subtitles)
- **Collision Probability**: ≈ 0% (file sizes will differ)
#### Scenario 2: Re-encoded Video (Same Source Material)
**Low Risk - Practically Impossible**
Same video encoded twice with different encoders or settings:
- **File Size**: Different (encoding settings, bitrate, codecs affect size)
- **Middle Content**: Different (encoding algorithms produce different compressed data)
- **Collision Probability**: ≈ 0% (encoding is deterministic per encoder but differs between encoders)
#### Scenario 3: Identical Container, Different Streams
**Medium Risk - Possible but Detectable**
Same video with different audio tracks or subtitles:
- **File Size**: Different (additional audio/subtitle streams change size)
- **Middle Content**: May differ (if audio/subtitle data interleaved)
- **Collision Probability**: Very low (size typically differs; if not, middle content likely differs)
#### Scenario 4: Truncated or Extended Files
**No Risk - Guaranteed Different**
File that is truncated or extended (padding added):
- **File Size**: Different (guaranteed)
- **Collision Probability**: 0% (size difference guarantees unique hash)
#### Scenario 5: Files with Identical Middles (Crafted Attack)
**Theoretical Risk - Requires Intentional Manipulation**
Maliciously crafted files with identical size and middle content:
- **File Size**: Identical (by design)
- **Middle Content**: Identical (by design)
- **Start/End Content**: Different (not included in hash)
- **Collision Probability**: 100% if crafted (but not a concern for media files)
**Mitigation**: Media files are organically created content, not crafted adversarially. This scenario does not occur in practice for video/audio files.
---
## Performance Characteristics
### Computation Time
| File Size | Full SHA-256 | Fast-Hash | Speedup | Time Saved |
|-----------|--------------|-----------|---------|------------|
| 100 MB | 2 sec | 0.02 sec | 100x | 1.98 sec |
| 1 GB | 20 sec | 0.02 sec | 1000x | 19.98 sec |
| 10 GB | 200 sec | 0.02 sec | 10,000x | 199.98 sec |
| 50 GB | 1000 sec | 0.02 sec | 50,000x | 999.98 sec |
**Assumptions:**
- Disk read speed: 500 MB/s (SSD)
- SHA-256 computation: 500 MB/s (CPU-bound)
- Fast-hash: 1MB sample + 8 bytes = 1.048 MB to hash
**Result**: Fast-hash is **O(1)** regardless of file size (always hashes ≈ 1MB).
### I/O Characteristics
**Full SHA-256 I/O:**
```
Total I/O = fileSize (sequential read)
For 50GB file: 50GB read
```
**Fast-Hash I/O:**
```
Total I/O = 1MB (single seek + read)
For 50GB file: 1MB read (99.998% reduction)
```
**Disk Operations:**
- **Full SHA-256**: Sequential scan (no seeks, but reads everything)
- **Fast-Hash**: Single seek to middle + 1MB read (minimal I/O)
### CPU Characteristics
**SHA-256 Computation:**
- **Full SHA-256**: SHA-256 state updated for every block of file (many GB)
- **Fast-Hash**: SHA-256 state updated for ≤ 1MB (fixed)
**CPU Time:**
- **Full SHA-256**: O(n) where n = file size
- **Fast-Hash**: O(1) regardless of file size
### Memory Characteristics
Both approaches use streaming (constant memory):
- **Full SHA-256**: Reads chunks (e.g., 64KB), updates hash, repeats
- **Fast-Hash**: Reads 1MB sample into memory, computes hash
**Memory Usage:**
- **Full SHA-256**: ~64KB buffer + SHA-256 state (minimal)
- **Fast-Hash**: ~1MB buffer + SHA-256 state (acceptable)
---
## Trade-offs and Design Decisions
### Advantages
#### 1. Dramatic Performance Improvement
Fast-hash reduces hash computation time by **100x to 50,000x** depending on file size. For a 50GB file:
- **Full SHA-256**: 16.7 minutes
- **Fast-Hash**: 0.02 seconds
This eliminates the hash computation bottleneck in the processing pipeline.
#### 2. Simplified Processing Pipeline
**Current System (with full SHA-256):**
```
Discovery → Validation → Metadata (tempId) → Hash → Rotate ID (tempId → hash)
```
- Temporary ID assigned during metadata extraction
- Final hash computed later (slow)
- ID rotation required (update all references from tempId to hash)
**New System (with fast-hash):**
```
Discovery → Validation → Metadata + Fast-Hash → Done
```
- Fast-hash computed immediately during metadata extraction (instant)
- No temporary ID needed
- No ID rotation required
- Simpler state management
#### 3. Immediate File Availability
Files can be fully indexed and accessible within seconds of discovery:
- **Current**: 15-20 minutes for large files (waiting for hash)
- **Fast-Hash**: < 1 second (hash computed instantly)
Virtual filesystem and search become immediately available.
#### 4. Constant-Time Complexity
Hash computation time is **O(1)** regardless of file size. This provides:
- **Predictable Performance**: Always ~20ms per file
- **Scalability**: 100GB files hash as fast as 1GB files
- **Resource Efficiency**: CPU and disk I/O are minimized
#### 5. Sufficient Uniqueness for Target Domain
For personal media libraries (even large ones with millions of files):
- **Practical Collision Risk**: Effectively zero for organically created media files
- **Theoretical Collision Risk**: Requires identical file size AND middle 1MB (extremely rare)
#### 6. Deterministic and Reproducible
Same file always produces same hash:
- **Content Addressable**: Hash uniquely identifies file content
- **Synchronization**: Distributed nodes agree on file identity
- **Deduplication**: Duplicate files detected reliably
---
### Disadvantages
#### 1. Weaker Cryptographic Guarantees
**Full SHA-256:**
- Collision probability: 2^-256 (effectively impossible)
- Security: Cryptographically secure (cannot be forged)
**Fast-Hash:**
- Collision probability: Higher (depends on middle 1MB, not full content)
- Security: Not cryptographically secure for authentication (only samples content)
**Impact:** Fast-hash should not be used for security-critical applications (e.g., verifying file integrity for software distribution). For media library management, this is acceptable.
#### 2. Potential Collisions for Specific File Types
**High-Risk File Types** (not applicable to Meta-Mesh):
- **Sparse Files**: Files with large regions of zeros (e.g., disk images)
- **Generated Files**: Programmatically generated with patterns
- **Container Files with Metadata-Only Changes**: Same content, different headers/footers
**Low-Risk File Types** (Meta-Mesh target):
- **Video Files**: Compressed content varies throughout file
- **Audio Files**: Waveform data varies throughout file
- **Subtitle Files**: Text content varies throughout file
**Mitigation:** Meta-Mesh targets video/audio files, where collision risk is minimal.
#### 3. Partial Content Modifications Undetected
If only the start or end of a file changes (and middle 1MB is unchanged):
- **Full SHA-256**: Detects change (entire file is hashed)
- **Fast-Hash**: Does not detect change (middle is unchanged)
**Example Scenario:**
```
Original: [intro][middle content][outro]
Modified: [new-intro][middle content][outro]
Fast-Hash: SAME (middle unchanged)
Full SHA-256: DIFFERENT (intro changed)
```
**Impact for Meta-Mesh:** Video/audio files are typically modified by re-encoding (entire file changes) rather than splicing (start/end modification). Subtitle files are small (< 1MB) and fully hashed. This scenario is rare in practice.
#### 4. Not Suitable for Integrity Verification
Fast-hash cannot verify file integrity:
- **Full SHA-256**: Detects any single-bit change anywhere in file
- **Fast-Hash**: Only detects changes in middle 1MB or file size
**Impact:** Fast-hash is for **identification** (deduplication, indexing), not **verification** (ensuring file is uncorrupted). If integrity verification is needed, full SHA-256 should be computed separately.
#### 5. Incompatible with Existing Hash Standards
Fast-hash is not compatible with:
- **Torrent Hashes**: Use full-file SHA-1 or info-hash
- **IPFS CIDs**: Use full-file multihash
- **Checksums**: Use CRC32, MD5, or full SHA-256
**Impact:** Fast-hash cannot be used for interoperability with external systems that require standard hash formats. Meta-Mesh can compute multiple hashes (fast-hash for internal use, full SHA-256 for external interoperability).
---
## Collision Risk Analysis
### Mathematical Model
For a collection of **n** files, the probability of at least one collision is approximated by the birthday paradox:
```
P(collision) ≈ 1 - e^(-n² / (2 × H))
Where:
H = hash space size
```
**Hash Space Size:**
- **Full SHA-256**: H = 2^256 (enormous, collision effectively impossible)
- **Fast-Hash**: H depends on middle content diversity
**Effective Hash Space for Fast-Hash:**
The middle 1MB provides:
```
Theoretical space: 2^(8 × 1,048,576) = 2^8,388,608 (astronomical)
Practical space: Much smaller due to:
- File size constraint (limits possible values)
- Content structure (video codecs produce non-random data)
```
**Practical Estimate:**
For video files, the middle 1MB contains:
- Compressed video frames (varies by scene, motion, color)
- Interleaved audio data (varies by sound, music, dialogue)
- Possibly subtitle data (varies by text)
**Estimated effective hash space**: 2^64 to 2^80 (considering content diversity)
**Collision Probability for Media Collections:**
| Collection Size (n) | Hash Space (H) | Collision Probability |
|---------------------|----------------|-----------------------|
| 1,000 files | 2^64 | ≈ 0.000027% |
| 10,000 files | 2^64 | ≈ 0.27% |
| 100,000 files | 2^64 | ≈ 27% |
| 1,000,000 files | 2^64 | ≈ 99.97% |
| 10,000 files | 2^80 | ≈ 0.0000000041% |
| 100,000 files | 2^80 | ≈ 0.000041% |
| 1,000,000 files | 2^80 | ≈ 0.041% |
**Interpretation:**
- For **typical personal media libraries** (1,000 - 10,000 files): Collision risk is negligible
- For **large media libraries** (100,000+ files): Collision risk becomes measurable but remains low if hash space is large (2^80)
- For **massive institutional collections** (1,000,000+ files): Collision detection and resolution may be necessary
**Mitigation Strategy:**
If a collision is detected (two different files with same fast-hash), the system can:
1. **Log the collision** for manual review
2. **Fall back to full SHA-256** for the conflicting files
3. **Use a secondary identifier** (full path, full SHA-256) to disambiguate
---
## Integration with Existing Components
### etcd Storage Structure
**Current Structure:**
```
/file/hash/{fullSHA256}/property/path → value
/file/tmp/{tempId}/property/path → value (temporary)
/tempid/{tempId} → {fullSHA256} (rotation mapping)
```
**New Structure with Fast-Hash:**
```
/file/hash/{midhash256}/property/path → value
```
**Changes:**
- Replace `fullSHA256` with `midhash256` as the primary key
- Eliminate `/file/tmp/{tempId}` keys (no temporary storage needed)
- Eliminate `/tempid/{tempId}` reverse lookup (no ID rotation)
**Simplified Key Structure:**
```
/file/bafkr4ih5kapbjzqvmj7jxr.../title → "Inception"
/file/bafkr4ih5kapbjzqvmj7jxr.../year → "2010"
/file/bafkr4ih5kapbjzqvmj7jxr.../video/codec → "h265"
```
(Using CID v1 format with custom multicodec 0x1000)
### Processing Pipeline
**Current Pipeline (Two-Phase):**
```typescript
// Phase 1: Light Processing (quick metadata)
const tempId = generateTempId();
await extractMetadata(filePath, tempId);
await etcd.put(`/file/tmp/${tempId}/title`, title);
// Phase 2: Hash Processing (slow full SHA-256)
const hash = await computeFullSHA256(filePath); // 15+ minutes
await rotateId(tempId, hash); // Move /tmp/{tempId} → /file/hash/{hash}
```
**New Pipeline (Single-Phase):**
```typescript
// Single Phase: Metadata + Fast-Hash
const midhash256 = await computecomputeMidHash256(filePath); // < 1 second
await extractMetadata(filePath, midhash256);
await etcd.put(`/file/hash/${midhash256}/title`, title);
```
**Benefits:**
- **Simpler Code**: No ID rotation logic
- **Faster Processing**: No waiting for slow hash computation
- **Fewer etcd Operations**: No temporary key creation/deletion
### Virtual Filesystem
**Impact:** Minimal changes required.
**Current:**
```typescript
// Retrieve file by full SHA-256
const metadata = await etcd.get(`/file/hash/${fullSHA256}/...`);
```
**New:**
```typescript
// Retrieve file by fast-hash
const metadata = await etcd.get(`/file/hash/${midhash256}/...`);
```
The virtual filesystem organizes files by metadata properties (title, year, season, episode), not by hash. The hash is used only as the unique key for metadata storage. Changing from full SHA-256 to fast-hash is transparent to the virtual filesystem.
### Deduplication
**Current Deduplication:**
```typescript
// Check if file already exists
const existingFile = await etcd.get(`/file/hash/${fullSHA256}`);
if (existingFile) {
// Duplicate detected (same content hash)
// Add new path to paths array
}
```
**New Deduplication:**
```typescript
// Check if file already exists
const existingFile = await etcd.get(`/file/hash/${midhash256}`);
if (existingFile) {
// Duplicate detected (same fast-hash)
// Verify paths are different (not the same file)
// Add new path to paths array
}
```
Deduplication logic remains identical. Fast-hash serves the same purpose as full SHA-256 (unique file identifier).
### Stremio Addon
**Impact:** None (uses metadata, not hash directly).
The Stremio addon retrieves files by metadata queries (title, season, episode), not by hash. The hash is used internally for metadata storage. Changing to fast-hash is transparent to the addon.
### Meta-Orbit (Distributed Synchronization)
**Impact:** Hash serves as content identifier for P2P synchronization.
**Current Synchronization:**
```typescript
// Node A: "I have file with hash sha256:abc123"
// Node B: "I also have sha256:abc123" → Same file, sync metadata
```
**New Synchronization:**
```typescript
// Node A: "I have file with CID bafkr4ih5kapbjzqvmj7jxr..."
// Node B: "I also have bafkr4ih5kapbjzqvmj7jxr..." → Same file (likely), sync metadata
```
**Consideration:** Fast-hash collisions are more likely than full SHA-256 collisions. If two nodes have files with the same fast-hash but different content (collision), synchronization may treat them as duplicates incorrectly.
**Mitigation:**
- **Collision Detection**: Compare additional metadata (file size, title, resolution) to detect false positives
- **Manual Resolution**: Allow users to manually resolve ambiguous cases
- **Fallback to Full Hash**: If collision suspected, compute full SHA-256 for confirmation
---
## Comparison with Alternative Approaches
### Fast-Hash vs. Full SHA-256
| Aspect | Full SHA-256 | Fast-Hash |
|--------|--------------|-----------|
| **Computation Time** | O(n), 15+ min for 50GB | O(1), < 1 sec for any size |
| **Uniqueness** | Cryptographic (2^-256) | Practical (depends on middle content) |
| **I/O Required** | Full file read (50GB) | 1MB read (fixed) |
| **Memory Usage** | Streaming (64KB buffer) | 1MB buffer |
| **Collision Risk** | Effectively zero | Very low (0.041% for 1M files) |
| **Security** | Cryptographically secure | Not secure (samples only) |
| **Suitable For** | Integrity verification, security | Deduplication, indexing |
**Decision Rationale:** Fast-hash prioritizes speed and practicality for media library management over cryptographic strength.
---
### Fast-Hash vs. Partial SHA-256 (First/Last 1MB)
**Alternative Approach:**
```
Hash first 1MB + last 1MB of file (instead of middle 1MB)
```
**Comparison:**
| Aspect | First/Last 1MB | Middle 1MB (Fast-Hash) |
|--------|----------------|------------------------|
| **Computation Time** | Same (2MB to hash) | Faster (1MB to hash) |
| **Uniqueness** | Better (captures headers + footers) | Good (captures content body) |
| **Collision Risk** | Lower (2MB of data) | Higher (1MB of data) |
| **Header Sensitivity** | High (format headers differ) | Low (content body may differ more) |
**Decision Rationale:** Middle 1MB is preferred because:
- **Faster**: 1MB vs. 2MB (50% less data to hash)
- **Simpler**: Single seek vs. two seeks
- **Content-Focused**: Middle contains unique content (not generic headers/footers)
---
### Fast-Hash vs. Random Sampling
**Alternative Approach:**
```
Sample random chunks throughout the file (e.g., 10 × 100KB chunks)
```
**Comparison:**
| Aspect | Random Sampling | Middle 1MB (Fast-Hash) |
|--------|----------------|------------------------|
| **Determinism** | Requires fixed seed (complex) | Deterministic (middle offset) |
| **Disk I/O** | Multiple seeks (slower) | Single seek (faster) |
| **Uniqueness** | Better (samples distributed) | Good (single large sample) |
| **Complexity** | High (seed management, chunk distribution) | Low (simple offset calculation) |
**Decision Rationale:** Middle 1MB is preferred because:
- **Simpler**: Single offset calculation, no seed management
- **Faster**: Single disk seek, not multiple
- **Deterministic**: Always samples exact middle (no randomness)
---
### Fast-Hash vs. Metadata-Based ID
**Alternative Approach:**
```
Use metadata (title + year + duration) as identifier instead of content hash
```
**Comparison:**
| Aspect | Metadata-Based ID | Fast-Hash |
|--------|-------------------|-----------|
| **Uniqueness** | Low (many files share same title/year) | High (content-based) |
| **Collision Risk** | High (identical metadata common) | Low (content must match) |
| **Deduplication** | Unreliable (different files, same metadata) | Reliable (same content → same hash) |
| **Content Change** | ID unchanged (metadata same) | ID changed (content changed) |
**Decision Rationale:** Content-based hash (fast-hash) is superior for deduplication and identity because:
- **Unique**: Content determines ID, not metadata
- **Reliable**: Different content → different ID (guaranteed)
- **Deduplication**: Identical content → same ID (accurate duplicate detection)
---
## Design Intent: Supplement, Not Replacement
**IMPORTANT**: midhash256 is a **supplement** to full-file hashes, **not a replacement**.
### Two-Phase Processing Model
MetaMesh implements a two-phase processing pipeline:
**Phase 1: Light Processing (midhash256)**
- Compute midhash256 hash (< 1 second)
- Extract metadata (filename parsing, FFmpeg analysis)
- Store in etcd with midhash256 as primary key
- **File becomes immediately accessible in Virtual Filesystem**
**Phase 2: Hash Processing (Full Hashes - Background)**
- Compute full-file hashes: SHA-256, SHA-1, MD5, CRC32 (slow, minutes)
- Update etcd with additional hash metadata
- Required for Meta-Orbit distributed synchronization
### Why Both Are Needed
**midhash256 Purpose:**
- **Immediate file availability**: Users can browse/stream files within seconds of discovery
- **Simplified ID management**: No temporary IDs or complex ID rotation
- **Fast deduplication**: Quickly identify potential duplicates
**Full Hash Purpose:**
- **Meta-Orbit integration**: Distributed P2P synchronization requires cryptographically secure hashes
- **External system compatibility**: Torrent verification, IPFS CIDs
- **Collision resolution**: If midhash256 collision detected, full SHA-256 provides definitive answer
### Processing Flow
```
File Discovery
↓
Light Processing (32 workers, parallel)
- Compute midhash256 (~20ms)
- Extract metadata
- Store in etcd: /file/{midhash256}/...
- Add to VFS (file accessible immediately) ✓
↓
Hash Processing (8 workers, background)
- Compute SHA-256, SHA-1, MD5, CRC32 (15+ minutes for 50GB)
- Update etcd with full hashes
- Enable Meta-Orbit synchronization ✓
```
### Collision Detection and Handling
When a file is processed, the system checks if another file with the same midhash256 already exists:
**If collision detected:**
1. **Log warning**: Alert administrator of potential collision
2. **Skip VFS addition**: Do not add colliding file to Virtual Filesystem
3. **Force full hash**: Compute SHA-256 immediately (bypass light processing)
4. **Use SHA-256 as key**: Store in etcd using full SHA-256 instead of midhash256
This ensures:
- **No data corruption**: Different files never share the same ID
- **No silent failures**: Collisions are logged and handled explicitly
- **Fallback path**: System continues working even if collision occurs
---
## Conclusion
The midhash256 global ID architecture provides a **practical, high-performance supplement** to full-file SHA-256 for media library management. By sampling 1MB from the middle of each file and prepending the file size, the algorithm achieves:
- **100x to 50,000x speedup** over full-file hashing (for Phase 1 processing)
- **O(1) constant-time complexity** regardless of file size
- **Sufficient uniqueness** for personal media libraries (millions of files)
- **Immediate file availability** (accessible in VFS within seconds)
- **Two-phase processing** (fast accessibility + background full hashing)
The trade-offs—weaker cryptographic guarantees, potential collisions for adversarial cases, and unsuitability for integrity verification—are acceptable for Meta-Mesh's target domain (personal media libraries where files are organically created, not adversarially crafted).
This architecture aligns with Meta-Mesh's design philosophy: **optimize for the common case** (large media files, organic content, personal libraries) while maintaining **practicality and performance** over theoretical perfection.
Meta-Mesh leverages midhash256 for **immediate file availability** while computing full SHA-256 hashes **in the background** for Meta-Orbit distributed synchronization and external system compatibility. This dual-hash approach provides both speed and security.
docs/plugin-system.md
+762
View File
@@ -0,0 +1,762 @@
# Meta-Sort Plugin System Architecture
This document describes the plugin system architecture for meta-sort, enabling extensible metadata extraction through dynamically loaded plugins.
## Overview
The plugin system provides a modular architecture for metadata extraction. Plugins are self-contained units that:
- Are discovered and loaded from a plugins directory at startup
- Can be activated/deactivated at runtime without restart
- Process files sequentially in dependency order
- Store metadata in a flat key-value format
- Have access to their own persistent cache folder
## Directory Structure
```
packages/meta-sort/packages/meta-sort-core/
├── plugins/ # All plugins (shipped + custom)
│ ├── file-info/
│ │ ├── manifest.yml # Plugin metadata & config schema
│ │ └── index.ts # Plugin implementation
│ ├── ffmpeg/
│ ├── filename-parser/
│ ├── anime-detector/
│ ├── jellyfin-nfo/
│ ├── subtitle/
│ ├── torrent/
│ ├── language/
│ ├── tmdb/
│ └── debug/
├── plugins.json # Runtime state (active/inactive + configs)
└── src/
└── plugin/
├── types.ts # Type definitions
├── PluginLoader.ts # Manifest parsing, dynamic import
├── PluginContext.ts # Context factory
├── PluginManager.ts # Lifecycle management
└── PluginExecutor.ts # Dependency-ordered execution
```
Cache folders are stored separately under `CACHE_FOLDER_PATH` (default `/data/cache/plugins/`):
```
/data/cache/plugins/
├── ffmpeg/
│ ├── {cid1}.json
│ └── {cid2}.json
├── tmdb/
│ ├── {cid1}.json
│ └── {cid2}.json
└── ...
```
## Plugin Manifest
Each plugin must have a `manifest.yml` file defining its metadata, dependencies, and configuration schema.
```yaml
# manifest.yml
id: tmdb # Unique identifier (matches folder name)
name: TMDB Enricher # Human-readable name
version: 1.0.0 # Semantic version
description: Enriches metadata from The Movie Database API
# Plugins that must execute BEFORE this one
dependencies:
- filename-parser # Needs parsed title/year
# Configuration schema (rendered in dashboard)
config:
apiKey:
type: string
label: API Key
required: true
secret: true # Masked in UI
language:
type: string
label: Language
default: en-US
includeAdult:
type: boolean
label: Include Adult Content
default: false
# Metadata schema for editor UI (optional)
# Describes keys this plugin writes
schema:
tmdb/id:
label: TMDB ID
type: string
readonly: true
tmdb/title:
label: Title
type: string
tmdb/overview:
label: Overview
type: text
tmdb/poster:
label: Poster URL
type: string
format: url
tmdb/genre/{n}:
label: Genre
type: string
indexed: true # Indicates indexed keys (0, 1, 2...)
```
### Manifest Fields
| Field | Required | Description |
|-------|----------|-------------|
| `id` | Yes | Unique plugin identifier, must match folder name |
| `name` | Yes | Display name for dashboard |
| `version` | Yes | Semantic version string |
| `description` | No | Brief description |
| `dependencies` | No | Array of plugin IDs that must run before this one |
| `config` | No | Configuration schema for dashboard UI |
| `schema` | No | Metadata keys written by this plugin (for editor) |
### Config Field Types
```yaml
config:
fieldName:
type: string | number | boolean | select
label: Display Label
required: true | false # Default: false
default: <value> # Default value
secret: true | false # Mask in UI (default: false)
options: # For type: select
- value: opt1
label: Option 1
- value: opt2
label: Option 2
```
## Plugin Interface
Plugins export a default object implementing the `Plugin` interface.
```typescript
// types.ts
export interface Plugin {
/**
* Called once when the plugin is loaded at startup.
* Use for one-time initialization.
*/
onLoad?(ctx: PluginLifecycleContext): Promise<void>;
/**
* Called when the plugin is activated.
* Triggered at startup (if previously active) or from dashboard.
*/
onActivate?(ctx: PluginLifecycleContext): Promise<void>;
/**
* Called when the plugin is deactivated from the dashboard.
* Use for cleanup, releasing resources.
*/
onDeactivate?(ctx: PluginLifecycleContext): Promise<void>;
/**
* Called when the plugin is unloaded at shutdown.
*/
onUnload?(ctx: PluginLifecycleContext): Promise<void>;
/**
* Process a single file. Called for each file in the library.
* Must be implemented by all plugins.
*/
process(ctx: PluginContext): Promise<void>;
}
export interface PluginLifecycleContext {
/** Plugin configuration from plugins.json */
config: Record<string, unknown>;
/** Absolute path to plugin's cache directory */
cacheDir: string;
/** Scoped logger */
log: PluginLogger;
}
```
## Plugin Context
The `PluginContext` is injected into the `process()` function, providing access to file metadata and utilities.
```typescript
export interface PluginContext {
/**
* Key-value storage for the current file's metadata.
* Keys use path notation: "video/codec", "audio/0/language"
*/
kv: PluginKVStore;
/** Absolute path to the file being processed */
filePath: string;
/** Plugin configuration values */
config: Record<string, unknown>;
/** Absolute path to plugin's cache directory */
cacheDir: string;
/** Cache utilities for persistent storage */
cache: PluginCache;
/** Scoped logger (prefixed with plugin name) */
log: PluginLogger;
}
```
### KV Store Interface
```typescript
export interface PluginKVStore {
/**
* Get a value by key.
* @param key Path-style key, e.g., "video/codec" or "audio/0/language"
*/
get(key: string): string | undefined;
/**
* Set a value. Only string values are supported.
* @param key Path-style key
* @param value String value
*/
set(key: string, value: string): void;
/**
* Delete a key.
*/
delete(key: string): void;
/**
* Get all keys, optionally filtered by prefix.
* @param prefix Optional prefix filter, e.g., "audio/" returns ["audio/0/codec", "audio/0/language", ...]
*/
keys(prefix?: string): string[];
/**
* Get all key-value pairs, optionally filtered by prefix.
*/
entries(prefix?: string): Array<[string, string]>;
}
```
### Cache Interface
```typescript
export interface PluginCache {
/**
* Get full path for a cache file.
* @param filename Relative filename within cache directory
*/
getPath(filename: string): string;
/**
* Read and parse a JSON cache file.
* @returns Parsed data or null if not found
*/
readJson<T>(filename: string): Promise<T | null>;
/**
* Write data as JSON to cache file.
*/
writeJson(filename: string, data: unknown): Promise<void>;
/**
* Check if a cache file exists.
*/
exists(filename: string): Promise<boolean>;
/**
* Delete a cache file.
*/
delete(filename: string): Promise<void>;
/**
* Clear all files in the plugin's cache directory.
*/
clear(): Promise<void>;
}
```
### Logger Interface
```typescript
export interface PluginLogger {
debug(message: string, ...args: unknown[]): void;
info(message: string, ...args: unknown[]): void;
warn(message: string, ...args: unknown[]): void;
error(message: string, ...args: unknown[]): void;
}
```
## Plugin Implementation Example
```typescript
// plugins/tmdb/index.ts
import type { Plugin, PluginContext, PluginLifecycleContext } from '../../src/plugin/types';
interface TmdbConfig {
apiKey: string;
language: string;
includeAdult: boolean;
}
const plugin: Plugin = {
async onLoad(ctx: PluginLifecycleContext) {
ctx.log.info('TMDB plugin loaded');
},
async onActivate(ctx: PluginLifecycleContext) {
const config = ctx.config as TmdbConfig;
if (!config.apiKey) {
ctx.log.warn('No API key configured, plugin will skip processing');
}
},
async onDeactivate(ctx: PluginLifecycleContext) {
ctx.log.info('TMDB plugin deactivated');
},
async process(ctx: PluginContext) {
const config = ctx.config as TmdbConfig;
// Skip if not configured
if (!config.apiKey) return;
// Get data from previous plugins
const title = ctx.kv.get('filename/title');
const year = ctx.kv.get('filename/year');
const videoType = ctx.kv.get('video/type');
if (!title) return;
// Check cache first (keyed by title+year)
const cacheKey = `${sanitize(title)}-${year || 'unknown'}.json`;
const cached = await ctx.cache.readJson<TmdbResult>(cacheKey);
if (cached) {
applyResult(ctx.kv, cached);
return;
}
// Fetch from TMDB API
try {
const result = await fetchTmdb(title, year, videoType, config);
if (result) {
await ctx.cache.writeJson(cacheKey, result);
applyResult(ctx.kv, result);
}
} catch (error) {
ctx.log.error('TMDB fetch failed', error);
}
}
};
function applyResult(kv: PluginContext['kv'], result: TmdbResult) {
kv.set('tmdb/id', String(result.id));
kv.set('tmdb/title', result.title);
kv.set('tmdb/overview', result.overview);
kv.set('tmdb/poster', result.poster_path);
kv.set('tmdb/backdrop', result.backdrop_path);
kv.set('tmdb/rating', String(result.vote_average));
kv.set('tmdb/year', result.release_date?.slice(0, 4) || '');
// Genres as indexed keys
result.genres?.forEach((genre, i) => {
kv.set(`tmdb/genre/${i}`, genre.name);
});
}
export default plugin;
```
## Metadata Key Format
All metadata is stored as flat key-value pairs with string values only. Keys use path notation with `/` as separator.
### Key Naming Convention
```
{plugin-id}/{field} # Simple field
{plugin-id}/{group}/{field} # Grouped field
{plugin-id}/{group}/{index}/{field} # Indexed field (for lists)
```
### Examples
```
# File info plugin
file/name = "Movie.2024.1080p.mkv"
file/size = "4294967296"
file/type = "video"
file/mime = "video/x-matroska"
# FFmpeg plugin
video/codec = "h264"
video/width = "1920"
video/height = "1080"
video/duration = "7200.5"
video/bitrate = "8000000"
audio/0/codec = "aac"
audio/0/language = "eng"
audio/0/channels = "6"
audio/1/codec = "dts"
audio/1/language = "fra"
audio/1/channels = "6"
subtitle/0/language = "eng"
subtitle/0/format = "subrip"
subtitle/1/language = "fra"
subtitle/1/format = "subrip"
# Filename parser plugin
filename/title = "Movie Title"
filename/year = "2024"
filename/quality = "1080p"
filename/source = "bluray"
# TMDB plugin
tmdb/id = "12345"
tmdb/title = "Movie Title"
tmdb/overview = "A movie about..."
tmdb/poster = "/path/to/poster.jpg"
tmdb/genre/0 = "Action"
tmdb/genre/1 = "Adventure"
```
### Key Constraints
- Keys must be non-empty strings
- Keys use `/` as path separator
- Values must be strings (numbers converted via `String()`)
- No nested objects or arrays (use indexed keys instead)
- Keys are case-sensitive
- Recommended: use lowercase with hyphens for multi-word keys
## Plugin Manager
The `PluginManager` handles plugin lifecycle and orchestration.
```typescript
export class PluginManager extends EventEmitter {
/**
* Scan plugins directory and load manifests.
* Called once at startup.
*/
async scanPlugins(): Promise<PluginInfo[]>;
/**
* Load all plugins and activate those marked active in state.
* Called once at startup after scanning.
*/
async loadAll(): Promise<void>;
/**
* Activate a plugin (hot, no restart required).
* Calls plugin.onActivate(), rebuilds execution order, saves state.
* Emits 'plugin:activated' event.
*/
async activate(pluginId: string): Promise<void>;
/**
* Deactivate a plugin (hot, no restart required).
* Calls plugin.onDeactivate(), rebuilds execution order, saves state.
* Emits 'plugin:deactivated' event.
*/
async deactivate(pluginId: string): Promise<void>;
/**
* Update plugin configuration.
* Saves to plugins.json.
* Emits 'plugin:config-changed' event.
*/
async updateConfig(pluginId: string, config: Record<string, unknown>): Promise<void>;
/**
* Clear a plugin's cache directory.
*/
async clearCache(pluginId: string): Promise<void>;
/**
* Get all discovered plugins.
*/
getPlugins(): PluginInfo[];
/**
* Get only active plugins.
*/
getActivePlugins(): PluginInfo[];
/**
* Get execution order (dependency-sorted active plugin IDs).
*/
getExecutionOrder(): string[];
/**
* Process a file through all active plugins.
*/
async processFile(filePath: string, kv: KVStore): Promise<ProcessingResult>;
/**
* Shutdown: deactivate and unload all plugins.
*/
async shutdown(): Promise<void>;
}
```
### Events
| Event | Payload | Description |
|-------|---------|-------------|
| `plugin:activated` | `{ pluginId: string }` | Plugin was activated |
| `plugin:deactivated` | `{ pluginId: string }` | Plugin was deactivated |
| `plugin:config-changed` | `{ pluginId: string, config: object }` | Plugin config updated |
| `plugin:error` | `{ pluginId: string, error: Error }` | Plugin error occurred |
### Plugin Info
```typescript
export interface PluginInfo {
id: string;
name: string;
version: string;
description?: string;
dependencies: string[];
active: boolean;
config: Record<string, unknown>;
configSchema?: ConfigSchema;
metadataSchema?: MetadataSchema;
status: 'loaded' | 'error' | 'unloaded';
error?: string;
}
```
## State Management
Plugin state is persisted in `plugins.json` located in the meta-sort-core package root.
```json
{
"plugins": {
"file-info": {
"active": true,
"config": {}
},
"ffmpeg": {
"active": true,
"config": {}
},
"tmdb": {
"active": true,
"config": {
"apiKey": "your-api-key",
"language": "en-US",
"includeAdult": false
}
},
"custom-plugin": {
"active": false,
"config": {}
}
}
}
```
### State Behavior
- **New plugins**: Discovered plugins not in state file are added with `active: false`
- **Removed plugins**: Plugins in state but not on disk are kept (preserves config for reinstall)
- **Config merge**: Default values from manifest are merged with saved config
- **Persistence**: State is saved immediately on any change
## Execution Pipeline
### Startup Sequence
```
1. PluginManager.scanPlugins()
└── Scan plugins/ directory
└── Parse each manifest.yml
└── Validate manifest structure
2. PluginManager.loadAll()
└── Load plugins.json state
└── For each discovered plugin:
├── Dynamic import(plugin/index.js)
├── Validate Plugin interface
├── Call plugin.onLoad()
└── If active in state:
└── Call plugin.onActivate()
└── Resolve execution order (topological sort)
```
### File Processing Sequence
```
1. Create KVStore for file
2. For each active plugin (in dependency order):
├── Create PluginContext
├── Start timing
├── try:
│ └── await plugin.process(ctx)
└── catch:
└── Log error, continue to next plugin
3. Persist KV to Redis
```
### Activation Sequence (Hot)
```
1. PluginManager.activate(pluginId)
├── Validate plugin exists and is loaded
├── Check dependencies are active
├── Call plugin.onActivate()
├── Mark as active in memory
├── Rebuild execution order
├── Save plugins.json
└── Emit 'plugin:activated' event
2. Caller receives event
└── Trigger library rescan
```
### Deactivation Sequence (Hot)
```
1. PluginManager.deactivate(pluginId)
├── Check no active plugins depend on this one
├── Call plugin.onDeactivate()
├── Mark as inactive in memory
├── Rebuild execution order
├── Save plugins.json
└── Emit 'plugin:deactivated' event
2. Caller receives event
└── Trigger library rescan
Note: Metadata written by the plugin remains in Redis.
A separate "clean metadata" feature can remove orphaned keys.
```
## Dependency Resolution
Plugins declare dependencies in their manifest. The system performs topological sort to determine execution order.
### Rules
1. A plugin cannot be activated if any of its dependencies are inactive
2. A plugin cannot be deactivated if any active plugin depends on it
3. Circular dependencies are detected and rejected at load time
4. Missing dependencies are logged as warnings; plugin can still load but may fail
### Example
```yaml
# filename-parser depends on file-info
# tmdb depends on filename-parser
# anime-detector depends on filename-parser
Resolved order: file-info → filename-parser → [tmdb, anime-detector]
```
Plugins at the same dependency level may execute in any order (or parallel in future).
## REST API
The plugin system exposes REST endpoints for dashboard integration.
### Endpoints
```
GET /api/plugins # List all plugins
GET /api/plugins/:id # Get plugin details
POST /api/plugins/:id/activate # Activate plugin
POST /api/plugins/:id/deactivate # Deactivate plugin
PUT /api/plugins/:id/config # Update plugin config
DELETE /api/plugins/:id/cache # Clear plugin cache
GET /api/plugins/:id/schema # Get metadata schema (for editor)
```
### Response Format
```typescript
// GET /api/plugins
interface PluginsResponse {
plugins: PluginInfo[];
executionOrder: string[];
}
// POST /api/plugins/:id/activate
interface ActivateResponse {
success: boolean;
error?: string; # e.g., "Dependency 'x' is not active"
}
// PUT /api/plugins/:id/config
interface ConfigUpdateRequest {
config: Record<string, unknown>;
}
```
## Error Handling
### Plugin Load Errors
- Invalid manifest: Plugin marked as `status: 'error'`, logged, skipped
- Import failure: Plugin marked as `status: 'error'`, logged, skipped
- Missing dependencies: Warning logged, plugin can load but activation may fail
### Plugin Execution Errors
- Errors in `process()` are caught, logged, and execution continues to next plugin
- The file's metadata from previous plugins is preserved
- Error count is tracked per plugin for monitoring
### Lifecycle Hook Errors
- Errors in `onActivate()` prevent activation, error returned to caller
- Errors in `onDeactivate()` are logged but deactivation proceeds
- Errors in `onLoad()`/`onUnload()` are logged, plugin marked as error state
## Future Considerations
### Metadata Cleanup
A future feature will allow cleaning metadata keys written by a specific plugin:
```typescript
// Future API
await pluginManager.cleanMetadata(pluginId);
// Removes all keys matching "{pluginId}/*" from Redis
```
### Parallel Execution
Plugins at the same dependency level could execute in parallel:
```typescript
// Future: parallel execution of independent plugins
const level1 = ['file-info'];
const level2 = ['ffmpeg', 'filename-parser']; // parallel
const level3 = ['tmdb', 'anime-detector']; // parallel
```
### Plugin Hot Reload
Reloading a plugin's code without full restart (for development):
```typescript
// Future API
await pluginManager.reload(pluginId);
// Re-imports plugin code, preserves state
```
docs/plugin-task-queue-architecture.md
+962
View File
@@ -0,0 +1,962 @@
# Plugin Task Queue Architecture
**Author**: Meta Mesh Development Team
**Last Updated**: 2026-01-07
**Status**: Current Implementation
---
## Table of Contents
1. [Overview](#overview)
2. [Clarifications](#clarifications)
3. [Problem Statement](#problem-statement)
4. [Proposed Architecture](#proposed-architecture)
5. [Design Decisions](#design-decisions)
6. [Plugin Task Model](#plugin-task-model)
7. [Queue Classification](#queue-classification)
8. [Dependency and Priority Handling](#dependency-and-priority-handling)
9. [Data Flow](#data-flow)
10. [VFS Event System](#vfs-event-system)
11. [Performance Characteristics](#performance-characteristics)
12. [Migration Path](#migration-path)
---
## Overview
This document describes the **two-queue plugin task architecture** for meta-sort's processing pipeline. The architecture separates plugin execution into discrete, queueable tasks and routes them to specialized queues based on execution characteristics.
### Key Characteristics
- **Task Granularity**: Each plugin execution becomes a separate queueable task
- **Two-Queue Model**: Fast queue (high concurrency: 32) and Background queue (low concurrency: 8)
- **Sequential Queue Execution**: Background queue is paused while Fast queue has pending work
- **Manifest-Based Classification**: Plugins declare their queue via `defaultQueue` in manifest, with measurement-based override after sufficient samples
- **Dependency Preservation**: Plugin dependencies respected across queue boundaries (including cross-queue)
- **Priority Ordering**: Tasks sorted by execution time within each queue (fastest first)
- **Sequential Per-File Execution**: Plugins execute sequentially per file; concurrency is between files
### Goals
| Goal | Description |
|------|-------------|
| **Throughput** | Fast plugins don't wait behind slow plugins |
| **Resource Efficiency** | Appropriate concurrency for each workload type |
| **Simplicity** | Two queues are easier to reason about than N queues |
| **Predictability** | Classification via manifest with measurement-based refinement |
| **Observability** | Clear separation enables better monitoring |
---
## Clarifications
This section documents design clarifications made during architecture review.
### C1: `manifest.defaultQueue` Field
The `defaultQueue` field is implemented in the plugin manifest schema. See `packages/meta-sort/packages/meta-sort-core/src/plugin-engine/types.ts` for the definition.
Plugins specify their default queue in the manifest as either 'fast' or 'background'.
### C2: `full-hash` Plugin Implementation
The `full-hash` plugin is a wrapper around the existing hash computation logic, not a rewrite. It imports and calls the existing hash computation function, is classified as a background queue plugin, and allows future hash algorithms to be added as separate plugins.
### C3: Queue Replacement Strategy
The new two-queue model replaces and improves the current queue structure:
| Current Queue | New Queue | Notes |
|---------------|-----------|-------|
| Light Processing Queue | **Fast Queue** | Same purpose, now handles only fast plugins |
| Hash Processing Queue | **Background Queue** | Now handles slow plugins including `full-hash` |
| Validation Queue | (unchanged) | Remains for extension filtering |
The current `lightQueue` + `hashQueue` in `WatchedFileProcessor` will be replaced by `fastQueue` + `backgroundQueue` with the new task-based scheduling.
### C4: midhash256 "Inline" Meaning
"Inline" means midhash256 computation happens within the light processing queue worker, before any plugin tasks are created. It is NOT a plugin because:
1. It provides the file identity (`fileHash`) required to create task IDs
2. It must complete before task creation (tasks are identified as `${fileHash}:${pluginId}`)
3. It requires file access, so it runs in the queue worker context
The light processing queue worker first computes midhash256 (approximately 150ms), which produces the fileHash. Then it creates PluginTask entries for all plugins using that fileHash, and finally enqueues tasks to the Fast and Background queues.
### C5: VFS Population via Event Listener System
Files should appear in the VFS as soon as sufficient metadata is available, not after all fast plugins complete. This is achieved via an event listener system:
- **Event**: `plugin:complete` emitted when any plugin finishes for a file
- **Listener**: meta-fuse subscribes and checks if required metadata fields are present
- **VFS Update**: File added to VFS when minimum required fields exist (e.g., after `filename-parser` completes)
This replaces the current "wait for all plugins" approach and provides faster VFS visibility.
**Current Infrastructure**: Redis pub/sub channel `meta-sort:file:batch` already exists and batches updates every 5 seconds. This can be extended or a new real-time channel added for plugin completion events.
**Note**: The `renamingRule` function is NOT deprecated - it's actively used in both meta-sort and meta-fuse to generate virtual paths from metadata.
---
## Problem Statement
### Current Architecture
The current light processing phase executes all plugins sequentially for each file:
```
Light Processing Queue (16 workers)
│
└── Worker picks up file
│
├── midhash256 computation (~150ms) ◄── Inline, not a plugin
│
└── executePlugins(file)
│
├── file-info (~2ms)
├── filename-parser (~8ms)
├── ffmpeg (~300ms)
├── tmdb (~2000ms) ◄── Network call
├── language (~50ms)
└── subtitle (~30ms)
│
└── Total: ~2400ms per file
```
### Problems Identified
1. **Slow Plugin Blocking**
- A 2-second tmdb API call blocks the worker for the entire duration
- Fast plugins (file-info, filename-parser) complete in under 10ms but wait in the same queue
- With 16 workers, only 16 tmdb calls can run concurrently
2. **Suboptimal Resource Utilization**
- Network-bound plugins (tmdb) could run with higher concurrency (rate limit permitting)
- CPU-bound plugins share workers with I/O-bound plugins
- No differentiation between instant operations and heavy operations
3. **Throughput Bottleneck**
- For 1000 files: 1000 × 2400ms = 2,400,000ms total work
- With 16 workers: 2,400,000 / 16 = 150,000ms = 2.5 minutes minimum
- But instant plugins could process all 1000 files in seconds
4. **Inflexible Concurrency**
- All plugins share the same concurrency limit
- Cannot tune concurrency per-plugin or per-resource-type
---
## Proposed Architecture
### Two-Queue Model
Split plugin execution into two queues based on execution time characteristics:
```
File Discovery
│
▼
┌─────────────────────────────────────────────────────────────────────────┐
│ midhash256 computation (inline, ~150ms) │
│ ════════════════════════════════════════════════════════════════════ │
│ Required before any plugin processing - provides file identity │
└──────────────────────────────────┬──────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────┐
│ FAST QUEUE │
│ (High Concurrency: 32) │
│ │
│ Sorted by average execution time (fastest first) │
│ Plugins execute sequentially per file, concurrency is between files │
│ │
│ ┌──────────┐ ┌─────────────────┐ ┌──────────┐ ┌──────────┐ │
│ │file-info │─▶│ filename-parser │─▶│ ffmpeg │─▶│ language │ │
│ │ ~2ms │ │ ~8ms │ │ ~300ms │ │ ~50ms │ │
│ └──────────┘ └─────────────────┘ └──────────┘ └──────────┘ │
│ │
│ Classification: avgTime < 1000ms OR manifest.defaultQueue = 'fast' │
│ ════════════════════════════════════════════════════════════════════ │
│ Files appear in VFS after fast processing completes │
└─────────────────────────────────────────────────────────────────────────┘
│
│ Fast queue IDLE (all fast tasks complete)
│ Background queue STARTS (in-progress tasks allowed to complete)
▼
┌─────────────────────────────────────────────────────────────────────────┐
│ BACKGROUND QUEUE │
│ (Low Concurrency: 8) │
│ ⚠ PAUSED while Fast queue has work │
│ │
│ Sorted by average execution time (fastest first) │
│ Plugins execute sequentially per file, concurrency is between files │
│ │
│ ┌──────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ tmdb │ │ full-hash │ │ transcode │ │
│ │ ~2000ms │ │ ~5000ms │ │ ~30000ms │ │
│ └──────────┘ └─────────────┘ └─────────────┘ │
│ │
│ Classification: avgTime >= 1000ms OR manifest.defaultQueue = 'background'│
└─────────────────────────────────────────────────────────────────────────┘
│
▼
File Complete (all plugin tasks finished)
```
**Critical Constraint**: The Background queue does NOT process while the Fast queue has pending work. This ensures all files become visible in the VFS as quickly as possible before background enrichment begins.
### Why Two Queues?
We evaluated several queue architectures:
| Architecture | Queues | Complexity | Chosen |
|-------------|--------|------------|--------|
| Per-plugin queues | N (one per plugin) | High | No |
| Resource-typed queues | 4 (CPU/IO/Network/Heavy) | Medium | No |
| Time-based queues | 2 (Fast/Background) | Low | **Yes** |
| Single priority queue | 1 | Low | No |
**Two queues provide the optimal balance:**
- **Simpler than N queues**: Easier to reason about, less coordination overhead
- **Better than single queue**: Slow tasks don't block fast tasks
- **Adaptive**: Classification based on actual measurements, not static assumptions
- **Resource-aware**: Background queue naturally limits heavy operations
---
## Design Decisions
### Decision 1: Task Granularity per Plugin
**Problem**: How should work be divided across queues?
**Options Considered**:
1. **Per-file tasks**: Current model (one task = all plugins for one file)
2. **Per-plugin tasks**: Each plugin execution is a separate task
3. **Per-phase tasks**: Group plugins by processing phase
**Decision**: Per-plugin tasks (Option 2)
**Rationale**:
- Enables fine-grained scheduling and prioritization
- Allows slow plugins to be isolated from fast plugins
- Enables per-plugin concurrency tuning
- Better resource utilization (fast plugins don't wait for slow ones)
- Natural fit for dependency-based execution order
---
### Decision 2: Two Queues (Fast/Background)
**Problem**: How many queues should exist, and how should they be differentiated?
**Options Considered**:
1. **Per-plugin queues**: Maximum isolation, maximum complexity
2. **Four resource-typed queues**: CPU, I/O, Network, Heavy (mixed)
3. **Two time-based queues**: Fast (under 1s) and Background (1s or more)
4. **Single priority queue**: Simplest, but slow tasks still block workers
**Decision**: Two time-based queues (Option 3)
**Rationale**:
The key insight is that execution time is what matters for user experience, not resource type:
- Users want to see results quickly
- Fast plugins should complete for all files before slow plugins start
- The distinction between "fast enough to wait for" and "do in background" is the meaningful boundary
Resource-typing (Option 2) was rejected because:
- Hash computation is both I/O and CPU intensive - which queue?
- FFmpeg is I/O + CPU + process spawning - hard to categorize
- Resource types don't map cleanly to user-perceived latency
Per-plugin queues (Option 1) were rejected because:
- Plugin dependencies require coordination across queues anyway
- Tracking "file complete" across N queues is complex
- Diminishing returns beyond 2 queues for this workload
---
### Decision 3: Time-Based Classification
**Problem**: How should plugins be assigned to Fast vs Background queue?
**Options Considered**:
1. **Static configuration**: Plugin manifest declares queue type
2. **Dynamic measurement**: Classify based on observed execution times
3. **Hybrid**: Static default with dynamic override
**Decision**: Dynamic measurement (Option 2)
**Rationale**:
- Execution time varies by environment (fast SSD vs slow NAS)
- Plugin behavior may change over time (API latency varies)
- Removes need for manual configuration
- Self-healing: if a plugin becomes faster/slower, classification adapts
**Implementation Notes**:
- Uses existing performanceMetrics.pluginTimings to calculate average time
- Default threshold is 1000ms (1 second)
- New plugins with no measurements start in Fast queue, then migrate to Background if their average exceeds threshold
---
### Decision 4: Priority Within Queues
**Problem**: How should tasks be ordered within each queue?
**Options Considered**:
1. **FIFO**: Process in arrival order
2. **Time-based priority**: Fastest tasks first
3. **Dependency-based priority**: Ready tasks with satisfied dependencies first
**Decision**: Time-based priority with dependency gating (Options 2 + 3)
**Rationale**:
- Fastest-first maximizes throughput (more completions per time unit)
- Dependencies must be satisfied regardless of priority
- Combines both: tasks are eligible when dependencies met, then sorted by time
Tasks with satisfied dependencies get priority based on inverse average execution time - faster tasks run first.
---
### Decision 5: Concurrency Levels
**Problem**: What concurrency should each queue have?
**Options Considered**:
1. **Equal concurrency**: Both queues get 16 workers
2. **Inverse concurrency**: Fast gets more, Background gets less
3. **Adaptive concurrency**: Adjust based on queue depth
**Decision**: Inverse concurrency (Option 2)
**Rationale**:
| Queue | Concurrency | Rationale |
|-------|-------------|-----------|
| Fast | 32 | Fast tasks complete quickly, high concurrency reduces latency |
| Background | 8 | Slow tasks are resource-intensive, limit to prevent thrashing |
- Fast queue: Higher concurrency is safe because tasks complete quickly
- Background queue: Lower concurrency prevents resource exhaustion (network, I/O, memory)
- Background queue concurrency also enables natural rate limiting for API-based plugins
---
### Decision 6: Hash Computation Placement
**Problem**: Hash computation is both I/O-intensive (reading file) and CPU-intensive (computing hash). How should it be handled?
**Analysis**:
- **midhash256**: Reads 256KB chunks, fast (approximately 150ms average) → Inline computation (not a plugin)
- **Full hash (SHA-256)**: Reads entire file, slow (5-30 seconds) → Background queue plugin
**Decision**:
- midhash256 is computed inline before any plugin processing (provides file identity)
- Full hash becomes a background queue plugin (`full-hash` plugin)
**Rationale**:
- midhash256 is required to identify the file before any plugin can run - it cannot be a plugin itself
- Full hash is slow and non-blocking for VFS visibility, making it ideal for background processing
- This absorbs the previous "Hash Processing" stage into the Background queue
- Hash algorithms become plugins, allowing new hash types to be added without core changes
---
### Decision 7: Sequential Queue Execution (Fast-First)
**Problem**: Should the Fast and Background queues run concurrently, or should Background wait for Fast to complete?
**Options Considered**:
1. **Concurrent execution**: Both queues process tasks simultaneously
2. **Sequential execution**: Background queue paused while Fast queue has work
3. **Hybrid**: Background runs at reduced concurrency while Fast is active
**Decision**: Sequential execution (Option 2) - Background queue does not start new tasks while Fast queue has work
**Important**: When new fast work arrives while background is processing, in-progress background tasks are allowed to complete - only new background task starts are paused.
**Rationale**:
This is a critical architectural constraint that preserves the current system behavior and provides several benefits:
1. **Predictable Resource Usage**
- Fast queue can use full system resources without competition
- No resource contention between fast and slow operations
- Prevents slow operations (network, heavy I/O) from impacting fast processing
2. **User Experience Priority**
- Files appear in VFS as quickly as possible
- All files become visible before enrichment begins
- Users see a complete library structure immediately, then metadata enriches over time
3. **Simpler Mental Model**
- Clear phases: "discovery/fast processing" then "background enrichment"
- Easier to monitor and debug
- Progress is linear and predictable
4. **Batch Efficiency**
- Background tasks (like tmdb API calls) can be batched more effectively
- Rate limiting is easier to manage when all network calls happen together
- Reduces API call overhead through potential batching
**Visual Timeline**:
```
Time ──────────────────────────────────────────────────────────────────────▶
┌─────────────────── FAST QUEUE ACTIVE ───────────────────┐
│ │
│ File A: [file-info][filename-parser][ffmpeg] │
│ File B: [file-info][filename-parser][ffmpeg] │
│ File C: [file-info][filename-parser][ffmpeg] │
│ ... │
│ File N: [file-info][filename-parser][ffmpeg] │
│ │
└──────────────────────────────────────────────────────────┘
│
│ Fast queue idle
▼
┌─────────────────── BACKGROUND QUEUE ACTIVE ─────────────────────────▶
│
│ File A: [tmdb][full-hash]
│ File B: [tmdb][full-hash]
│ File C: [tmdb][full-hash]
│ ...
│
└─────────────────────────────────────────────────────────────────────▶
All files visible in VFS ──┘
│
Enrichment happens in background
```
---
### Decision 8: Sequential Plugin Execution Per File
**Problem**: Should multiple plugins for the same file execute concurrently, or sequentially?
**Options Considered**:
1. **Concurrent per-file**: Multiple plugins process the same file simultaneously
2. **Sequential per-file**: Plugins execute one at a time per file, concurrency between files
**Decision**: Sequential per-file execution (Option 2)
**Rationale**:
- Plugins share a KV store per file - concurrent writes would cause race conditions
- Dependencies require sequential execution anyway (tmdb needs filename-parser results)
- Simplifies plugin development - no need to handle concurrent access
- Concurrency between files (32 fast, 8 background) provides sufficient throughput
- Easier to debug and reason about plugin execution order
---
### Decision 9: Shared KV Store Across Tasks
**Problem**: How should plugin results be shared between tasks for the same file?
**Decision**: Single shared KV store per file, persisted across all plugin tasks
**Behavior**:
- KV store is created when file enters processing (after midhash256)
- All plugins for a file read/write to the same KV store
- Once a plugin sets a value, subsequent plugins can read it
- KV store persists to Redis after each plugin completes
- Cross-queue access works transparently (fast plugin results available to background plugins)
For example, the filename-parser plugin (fast queue) sets parsed title and year data. Later, the tmdb plugin (background queue) reads that parsed data from the same KV store to make its API request.
---
### Decision 10: Cross-Queue Dependencies
**Problem**: What happens when a fast plugin depends on a background plugin?
**Decision**: Dependencies are always respected, regardless of queue classification
**Behavior**:
- If a fast plugin declares a dependency on a background plugin, it waits
- This effectively moves the dependent fast plugin to execute after background
- The task remains in the fast queue but cannot start until dependency completes
**Recommendation**: Avoid cross-queue dependencies where possible. If a fast plugin needs background data, consider whether it should be a background plugin instead.
---
## Plugin Task Model
### Task Definition
Each plugin execution becomes a discrete task with the following properties:
- **id**: Unique task ID in format `${fileHash}:${pluginId}`
- **fileHash**: File identity (midhash256, computed inline before tasks)
- **filePath**: File path for processing
- **pluginId**: Plugin to execute
- **dependencies**: Plugin IDs this task depends on
- **queue**: Target queue ('fast' or 'background') from classification
- **priority**: Based on average execution time (lower = higher priority)
- **estimatedTimeMs**: From performance metrics
- **status**: One of 'pending', 'ready', 'running', 'completed', or 'failed'
- **createdAt**, **startedAt**, **completedAt**: Timestamps
- **result**: Plugin result (on completion)
- **error**: Error message (on failure)
**Note**: The `fileHash` (midhash256) is computed inline before task creation, so it's always available when tasks are created.
### Task Lifecycle
```
┌─────────┐
│ PENDING │ ◄── Created when file enters light processing
└────┬────┘
│
│ All dependencies completed
▼
┌─────────┐
│ READY │ ◄── Eligible for queue placement
└────┬────┘
│
│ Worker picks up task
▼
┌─────────┐
│ RUNNING │ ◄── Plugin.process() executing
└────┬────┘
│
┌─────────────┴─────────────┐
│ │
▼ ▼
┌───────────┐ ┌──────────┐
│ COMPLETED │ │ FAILED │
└───────────┘ └──────────┘
```
### Task Creation
When a file enters light processing, tasks are created for all active plugins. The plugin manager provides the execution order via topological sort, and performance metrics provide estimated timing for each plugin. Each task is assigned dependencies based on the plugin manifest.
---
## Queue Classification
### Classification Algorithm
Classification uses the following logic:
1. Check if plugin has sufficient measurement data (at least 10 samples)
2. If not enough data, use the manifest's `defaultQueue` value (defaults to 'fast')
3. If enough data exists, use measured average time against a 1000ms threshold
Classification is determined at task creation time and does not change for existing tasks. This keeps the system simple and predictable.
### Expected Classification
Based on current measurements and recommended `defaultQueue` values:
| Plugin | Avg Time | defaultQueue | Classification |
|--------|----------|--------------|----------------|
| file-info | ~2ms | fast | Fast |
| filename-parser | ~8ms | fast | Fast |
| ffmpeg | ~300ms | fast | Fast |
| language | ~50ms | fast | Fast |
| subtitle | ~30ms | fast | Fast |
| jellyfin-nfo | ~20ms | fast | Fast |
| tmdb | ~2000ms | background | Background |
| full-hash | ~5000ms+ | background | Background |
**Note**: midhash256 is not listed as it's computed inline, not as a plugin.
### Manifest Schema Extension
Plugins can declare their default queue in the manifest with a `defaultQueue` field:
- `fast` (default): Plugin is assumed to complete quickly (under 1s)
- `background`: Plugin is assumed to be slow (1s or more), e.g., API calls, full file hashing
This is used for classification until enough measurements are collected. After 10+ executions, measured average time takes precedence.
---
## Dependency and Priority Handling
### Dependency Resolution
Plugin dependencies are declared in manifests and respected across queues. For example, the tmdb plugin depends on file-info, filename-parser, and jellyfin-nfo.
**Cross-Queue Dependencies**:
- tmdb (Background) depends on filename-parser (Fast)
- tmdb task remains 'pending' until filename-parser task is 'completed'
- Once dependencies satisfied, tmdb task becomes 'ready' and enters Background queue
### Dependency Graph Example
```
file-info (Fast)
/ \
/ \
filename-parser (Fast) ffmpeg (Fast)
| / \
| / \
jellyfin-nfo (Fast) language subtitle
| (Fast) (Fast)
|
▼
tmdb (Background) ◄── Waits for filename-parser & jellyfin-nfo
|
▼
full-hash (Background) ◄── No deps, but queued after fast completes
```
**Note**: `full-hash` has no plugin dependencies but runs in Background queue. It starts only after all fast tasks complete due to sequential queue execution.
### Task Scheduling
The TaskScheduler maintains pending and completed task sets. When a task is enqueued, the scheduler checks if all dependencies are satisfied. If ready, it adds the task to the appropriate queue with priority based on estimated execution time.
After task completion, timing is recorded for future classification, and dependent tasks are checked for readiness.
---
## Data Flow
### Complete Processing Flow
```
┌─────────────────────────────────────────────────────────────────────────────┐
│ 1. File Discovery │
│ FolderWatcher yields file path │
└──────────────────────────────────┬──────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ 2. Validation Queue (unchanged) │
│ Extension check, state initialization │
└──────────────────────────────────┬──────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ 3. midhash256 Computation (inline) │
│ Compute file identity hash before any plugin processing │
│ This is NOT a plugin - required for task identification │
└──────────────────────────────────┬──────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ 4. Task Creation │
│ Create PluginTask for each active plugin (fileHash now available) │
│ │
│ file-info task (pending, no deps, fast) → immediately ready │
│ filename-parser task (pending, deps: file-info, fast) │
│ ffmpeg task (pending, deps: file-info, fast) │
│ tmdb task (pending, deps: filename-parser, background) │
│ full-hash task (pending, no deps, background) │
│ ... │
└──────────────────────────────────┬──────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ 5. Fast Queue Processing │
│ Concurrency: 32 (between files, sequential per file) │
│ │
│ ┌────────────┐ ┌─────────────────┐ ┌──────────┐ ┌────────┐ │
│ │ file-info │───▶│ filename-parser │───▶│ ffmpeg │───▶│language│ │
│ │ (ready) │ │ (ready after │ │ (ready) │ │(ready) │ │
│ └────────────┘ │ file-info) │ └──────────┘ └────────┘ │
│ └────────┬────────┘ │
│ │ │
│ ▼ │
│ 🎯 VFS EVENT EMITTED │
│ File visible in VFS after filename-parser │
│ (minimum metadata available for path generation) │
│ │
│ ════════════════════════════════════════════════════════════════════════ │
│ Background queue tasks are queued but PAUSED during this phase │
└──────────────────────────────────┬──────────────────────────────────────────┘
│
│ ⚠ Fast queue becomes IDLE (no pending fast tasks)
│ ⚠ Background queue STARTS (in-progress tasks finish)
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ 6. Background Queue Processing │
│ Concurrency: 8 (between files, sequential per file) │
│ ⚠ ONLY starts new tasks when Fast queue is idle │
│ ⚠ In-progress background tasks complete even if new fast work arrives │
│ │
│ ┌──────────┐ ┌─────────────┐ │
│ │ tmdb │ │ full-hash │ │
│ │ (ready) │ │ (ready) │ │
│ └──────────┘ └─────────────┘ │
│ │
│ Slow plugins complete, metadata enriched, full hashes computed │
└──────────────────────────────────┬──────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ 7. File Complete │
│ All plugin tasks for file completed (last task finishes) │
│ State: done │
└─────────────────────────────────────────────────────────────────────────────┘
```
**Key Changes from Previous Architecture**:
- Hash Processing stage is eliminated - `full-hash` is now a Background queue plugin
- midhash256 is computed inline (not a plugin) as it's required for file identity
- Pipeline stages: Validation → midhash → Fast Queue → Background Queue → Done
- VFS visibility decoupled via event system - files appear after `filename-parser` completes (approximately 160ms)
### Multi-File Sequential Queue Processing
With sequential queue execution, ALL fast tasks complete before ANY background tasks start. However, VFS visibility is decoupled via the event system:
```
Time ──────────────────────────────────────────────────────────────────────▶
┌──────────────── FAST QUEUE PHASE ────────────────┐ ┌─── BACKGROUND PHASE ───▶
│ │ │
File A: [file-info][filename-parser]▼[ffmpeg] │ │ [tmdb][full-hash]
File B: .....[file-info][filename-parser]▼[ffmpeg] │ │ [tmdb][full-hash]
File C: ..........[file-info][filename-parser]▼[ffmpeg] │ │ [tmdb][full-hash]
File D: ...............[file-info][filename-parser]▼[ffmpeg] [tmdb][full-hash]
│ │ │ │
│ │ │ │
│ ▼ │ │
│ VFS EVENT: file visible │ │
│ (after filename-parser) │ │
│ │ │
└───────────────────────────────────────────────────┘ └─────────────────────────▶
│
│
Fast queue complete
Background enrichment begins
```
**Key behavior**:
- Fast queue processes all files with high concurrency (32)
- VFS visibility is IMMEDIATE after `filename-parser` completes (via event system)
- Files appear in VFS approximately 160ms after discovery (midhash + file-info + filename-parser)
- Background queue is PAUSED until fast queue becomes idle
- Once fast queue is empty, background queue starts (8 workers)
- If new files are discovered, fast queue resumes and background pauses again
- TMDB metadata enriches already-visible files when background processing completes
---
## VFS Event System
### Overview
The VFS (Virtual File System) population is decoupled from the task queue via an event-driven architecture. Files appear in the VFS as soon as sufficient metadata is available, rather than waiting for all plugins to complete.
### Event Architecture
```
┌─────────────────────────────────────────────────────────────────────────────┐
│ TASK SCHEDULER (meta-sort) │
│ │
│ Plugin executes → Result stored in KV → Event emitted │
│ │
│ Event contains: fileHash, pluginId, filePath, timestamp │
└──────────────────────────────────┬──────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ REDIS PUB/SUB (event transport) │
│ │
│ Channel: 'meta-sort:plugin:complete' │
│ Message: { fileHash, pluginId, filePath, timestamp } │
│ │
│ (Existing 'meta-sort:file:batch' channel remains for batch updates) │
└──────────────────────────────────┬──────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ VFS LISTENER (meta-fuse) │
│ │
│ On 'plugin:complete': │
│ 1. Fetch metadata from Redis │
│ 2. Check if minimum required fields present: │
│ - For TV: title, season, episode │
│ - For Movies: title, year │
│ 3. If sufficient: compute virtualPath via renamingRule() │
│ 4. Add file to VFS │
│ 5. File appears in FUSE/WebDAV │
│ │
│ On subsequent 'plugin:complete' for same file: │
│ - Update metadata (enrichment from tmdb, etc.) │
│ - File already visible, just metadata improves │
└─────────────────────────────────────────────────────────────────────────────┘
```
### Minimum Metadata Requirements
Files appear in VFS when these minimum fields are available (typically after `filename-parser` plugin):
| Content Type | Required Fields | Typical Provider Plugin |
|--------------|-----------------|------------------------|
| TV Episode | `title`, `season`, `episode` | filename-parser |
| Movie | `title`, `year` | filename-parser |
| Unknown | `filename` (fallback to Unsorted/) | file-info |
### Event vs Batch Trade-offs
| Approach | Latency | Overhead | Use Case |
|----------|---------|----------|----------|
| Per-plugin events | ~10ms | High (many messages) | Real-time VFS updates |
| Batch updates (5s) | ~5000ms | Low | Bulk metadata sync |
| Hybrid | ~10ms for VFS, batched for full sync | Medium | **Recommended** |
**Recommended Hybrid Approach**:
- Emit per-plugin events for VFS population (instant visibility)
- Keep batch updates for full metadata sync and reconnection recovery
- meta-fuse uses events for real-time, polling (30s) as safety net
### Integration with Current Infrastructure
The current Redis pub/sub infrastructure (`meta-sort:file:batch`) is preserved. The new event system adds:
1. **New channel**: `meta-sort:plugin:complete` for per-plugin events
2. **Event structure**: fileHash, pluginId, filePath, timestamp
3. **Backward compatibility**: Batch channel continues to work for existing consumers
---
## Performance Characteristics
### Expected Improvements
| Metric | Current | Proposed | Improvement |
|--------|---------|----------|-------------|
| Time to VFS visibility | ~2400ms | ~160ms (after filename-parser) | 15x faster |
| Fast plugin throughput | 16 concurrent | 32 concurrent | 2x |
| Background isolation | None | Full | Slow plugins don't block fast |
| Resource utilization | Uniform | Optimized | Better CPU/IO/network balance |
### Throughput Analysis
**Current Model** (1000 files):
- Total work: 1000 files × 2400ms = 2,400,000ms
- Workers: 16
- Time: 2,400,000ms / 16 = 150,000ms = 2.5 minutes
**Proposed Model** (1000 files):
- Inline midhash: 1000 × 150ms = 150,000ms (part of validation/task creation)
- Fast queue work: 1000 × (2 + 8 + 300 + 50 + 30)ms = 390,000ms
- Fast workers: 32
- Fast time: 390,000ms / 32 = 12,187ms ≈ 12 seconds
- Background queue work: 1000 × (2000 + 5000)ms = 7,000,000ms (tmdb + full-hash)
- Background workers: 8
- Background time: 7,000,000ms / 8 = 875,000ms ≈ 14.5 minutes
- Total time: ~15 minutes (background dominates, but includes full hashing)
- Time to VFS: ~5 seconds (after filename-parser, via event system)
**Key Insights**:
1. **VFS Visibility**: With the event system, files appear in VFS after `filename-parser` completes (approximately 160ms = midhash + file-info + filename-parser). For 1000 files with 32 workers: 160ms × 1000 / 32 = 5 seconds to full VFS visibility.
2. **Background Processing**: Total time is longer because full hashing is now included in background queue, but this doesn't affect user-perceived latency.
3. **Decoupled Enrichment**: TMDB metadata arrives later but files are already browsable.
### Memory Characteristics
Task tracking overhead per file:
- Tasks per file: ~8 plugins
- Task object size: ~200 bytes
- Per-file overhead: 1600 bytes
For 100,000 files:
- In-flight tasks (max): ~10,000 (based on queue depths)
- Memory overhead: 10,000 × 1600 bytes = 16MB
Acceptable memory overhead for significant throughput improvement.
---
## Migration Path
### Phase 1: Task Infrastructure
1. Define PluginTask interface and types
2. Implement TaskScheduler with single queue (current behavior)
3. Add task creation during light processing (after inline midhash)
4. Validate dependency resolution works correctly
5. Add `defaultQueue` field to manifest schema
### Phase 2: Two-Queue Split
1. Add fastQueue and backgroundQueue to TaskScheduler
2. Implement classification logic (manifest.defaultQueue + 1000ms threshold)
3. Route tasks to appropriate queues based on classification
4. Implement sequential queue execution (background pauses while fast has work)
5. Convert `full-hash` computation to a background queue plugin
### Phase 3: Polish
1. Implement priority ordering within queues (fastest first)
2. Expose queue metrics via API (queue depths, throughput, etc.)
3. Add configuration for threshold and concurrency levels
4. Update existing plugins with appropriate `defaultQueue` values
### Backward Compatibility
The new architecture is internal to light processing. External interfaces remain unchanged:
- Same file state transitions (pending → lightProcessing → hashProcessing → done)
- Same VFS update timing (after light processing completes)
- Same API endpoints and responses
- Same plugin interface (plugins unaware of queue routing)
---
## Appendix: Alternative Architectures Considered
### Per-Plugin Queues
Separate queue per plugin (file-info queue, ffmpeg queue, tmdb queue, etc.) with individual concurrency settings.
**Rejected because**:
- N queues require N times coordination logic
- Tracking file completion across N queues is complex
- Dependency chains span multiple queues
- Diminishing returns beyond 2-3 queues
### Four Resource-Typed Queues
Queues for CPU (16 workers), I/O (8 workers), Network (4 workers), and Heavy (4 workers).
**Rejected because**:
- Classification is ambiguous (hash is both I/O and CPU)
- FFmpeg is I/O + CPU + process spawning
- Resource type doesn't directly correlate with user-perceived latency
- More complex than necessary for the problem being solved
### Single Priority Queue
Single priority queue sorted by average time with 16 workers.
**Rejected because**:
- Slow tasks still consume workers
- No isolation between fast and slow workloads
- Cannot tune concurrency per workload type
- Fast tasks can be blocked behind slow tasks already in progress
---
## Glossary
- **Task**: A discrete unit of work (one plugin execution for one file)
- **Fast Queue**: High-concurrency (32) queue for plugins completing in under 1 second
- **Background Queue**: Low-concurrency (8) queue for plugins completing in 1 second or more
- **Classification**: Determining which queue a plugin's tasks should use (based on `defaultQueue` manifest field or measured average time)
- **Dependency**: A plugin that must complete before another plugin can start
- **Priority**: Ordering within a queue based on expected execution time (fastest first)
- **defaultQueue**: Manifest field specifying the default queue for a plugin before measurements are available
- **midhash256**: Inline hash computation (not a plugin) that provides file identity before task creation
- **Sequential Queue Execution**: Background queue only processes when Fast queue is idle
---
## References
- [Streaming Pipeline Architecture](./streaming-pipeline-architecture.md)
- [p-queue Priority Queues](https://github.com/sindresorhus/p-queue#priority)
- Plugin System Types: `packages/meta-sort/packages/meta-sort-core/src/plugin-engine/types.ts`
- PerformanceMetrics: `packages/meta-sort/packages/meta-sort-core/src/metrics/PerformanceMetrics.ts`
docs/streaming-pipeline-architecture.md
+863
View File
@@ -0,0 +1,863 @@
# Streaming Pipeline Architecture
**Author**: Meta Mesh Development Team
**Last Updated**: 2025-11-03
**Status**: Current Implementation
---
## Table of Contents
1. [Overview](#overview)
2. [Architecture Principles](#architecture-principles)
3. [Pipeline Stages](#pipeline-stages)
4. [Component Design](#component-design)
5. [Data Flow](#data-flow)
6. [State Management](#state-management)
7. [Concurrency Model](#concurrency-model)
8. [Performance Characteristics](#performance-characteristics)
9. [Architecture Decisions](#architecture-decisions)
10. [Future Considerations](#future-considerations)
---
## Overview
The Meta Mesh streaming pipeline is a **multi-stage asynchronous processing system** designed to efficiently process hundreds of thousands of media files with minimal memory overhead and optimal CPU utilization.
### Key Characteristics
- **Streaming Architecture**: Files are processed as they're discovered, not batched
- **Multi-Stage Pipeline**: Four independent stages with different concurrency levels
- **Non-Blocking**: Uses async generators to prevent event loop starvation
- **Scalable**: Handles 900k+ files and 100TB+ datasets
- **Separation of Concerns**: Generic file discovery separated from business logic
### Performance Goals
| Metric | Target | Current |
|--------|--------|---------|
| Discovery Speed | Immediate streaming | ✅ Achieved |
| Time to First Result | < 5 seconds | ✅ Achieved |
| Event Loop Blocking | Zero | ✅ Achieved |
| Memory Overhead | O(n) for queues only | ✅ Achieved |
| CPU Utilization | 100% during processing | ✅ Achieved |
---
## Architecture Principles
### 1. Streaming Over Batching
**Problem**: Original implementation waited for complete file discovery before processing, causing 15+ minute delays.
**Solution**: Stream files through the pipeline as they're discovered.
```typescript
// ❌ Batch Processing (Old)
const allFiles = await discoverAllFiles(); // Wait for all files
await processFiles(allFiles); // Then process
// ✅ Streaming Processing (New)
for await (const file of discoverFiles()) { // Process as discovered
pipeline.validate(file); // Immediate processing
}
```
### 2. Separation of Concerns
**Problem**: Business logic (extension filtering) was mixed with generic utilities (file discovery).
**Solution**: Clean separation between packages:
- **meta-hash** (Generic Library): Platform-agnostic file discovery
- **meta-mesh** (Application): Business-specific processing logic
This allows `meta-hash` to be reused in other contexts without Meta Mesh-specific dependencies.
### 3. Event Loop Yielding
**Problem**: Tight synchronous loops starved the Node.js event loop, preventing PQueue from dispatching work.
**Solution**: Use async generators that naturally yield to the event loop:
```typescript
async *walkDirectory(directory: string): AsyncGenerator<string> {
const entries = await readdir(directory); // Async I/O - yields to event loop
for (const entry of entries) {
yield fullPath; // Generator yield - allows event loop to process other tasks
}
}
```
### 4. Independent Stage Queues
**Problem**: Single queue with uniform concurrency doesn't match workload characteristics.
**Solution**: Each stage has its own queue with optimized concurrency:
- **Validation**: High concurrency (fast I/O checks)
- **Metadata**: Medium concurrency (balanced CPU/I/O)
- **Hash**: Medium concurrency (CPU-bound operations)
---
## Pipeline Stages
The pipeline consists of four stages, each optimized for its specific workload:
```
┌─────────────┐ ┌────────────┐ ┌──────────┐ ┌──────┐ ┌──────┐
│ Discovery │───▶│ Validation │───▶│ Metadata │───▶│ Hash │───▶│ Done │
│ (Stream) │ │ (32 queue) │ │(16 queue)│ │(16 q)│ │ │
└─────────────┘ └────────────┘ └──────────┘ └──────┘ └──────┘
Async Gen Extension Quick Meta Content Final
Yields files Filtering Extraction Hashing Storage
```
### Stage 0: Discovery (Streaming)
**Location**: `packages/meta-hash/src/lib/folder-watcher/FolderWatcher.ts`
**Purpose**: Discover all files in watched directories and emit paths as async generator.
**Implementation**:
```typescript
async *discoverFiles(directories: string[]): AsyncGenerator<string> {
for (const dir of directories) {
yield* this.walkDirectory(dir);
}
}
```
**Characteristics**:
- **Concurrency**: N/A (single-threaded recursive walk)
- **I/O Pattern**: Sequential directory reads with async/await
- **Output**: File path strings
- **Filtering**: None (emits all files regardless of type)
- **Duration**: Depends on filesystem speed and directory depth
**Key Design Choice**: No filtering logic at this layer keeps the component generic and reusable. All business logic happens in downstream stages.
---
### Stage 1: Validation (32 Workers)
**Location**: `packages/meta-sort/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts::validateFile()`
**Purpose**: Fast extension-based filtering to eliminate unsupported file types.
**Implementation**:
```typescript
private async validateFile(filePath: string): Promise<void> {
// Extension check (instant, no I/O)
const ext = path.extname(filePath).toLowerCase();
if (!this.config.supportedExtensions.has(ext)) {
return; // Skip unsupported files
}
// Mark as pending
this.config.stateManager.addPending(filePath);
this.validatedCount++;
// Stage 2: Metadata (fire and forget)
this.metadataQueue.add(() => this.extractMetadata(filePath));
}
```
**Characteristics**:
- **Concurrency**: 32 workers (CPU count × 2)
- **I/O Pattern**: Zero disk I/O (string operations only)
- **Processing Time**: < 1ms per file
- **Success Rate**: ~65% (video/subtitle/torrent files only)
- **Queue Behavior**: Typically empty (keeps up with discovery)
**Key Design Choice**: High concurrency (32 workers) is acceptable because validation is pure CPU-bound string operations with no I/O. This ensures validation never becomes a bottleneck.
**MIME Validation (Skipped)**: The original implementation included optional MIME type validation (reading file headers). This was removed because:
- Requires disk I/O (read first 4KB of each file)
- Adds ~25% overhead to discovery time
- Extension checking is sufficient for 99.9% of cases
- Edge cases are handled gracefully in later stages
---
### Stage 2: Metadata Extraction (16 Workers)
**Location**: `packages/meta-sort/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts::extractMetadata()`
**Purpose**: Extract lightweight metadata without computing content hash.
**Implementation**:
```typescript
private async extractMetadata(filePath: string): Promise<void> {
this.config.stateManager.startLightProcessing(filePath);
// Call existing light processing logic (quick metadata extraction)
const quickMeta = await this.config.fileProcessor.processFileQuick(filePath);
// Mark light processing as complete and move to hash queue
this.config.stateManager.completeLightProcessing(filePath);
this.metadataCount++;
// Stage 3: Hash (fire and forget)
this.hashQueue.add(() => this.computeHash(filePath));
}
```
**What Metadata is Extracted**:
1. **Filename Parsing** (via `filename-tool`):
- Series title, season, episode number
- Movie title, year
- Quality indicators (1080p, 4K, etc.)
- Language codes
- Release group
2. **File System Stats**:
- File size
- Modification time
- Permissions
3. **Basic Media Info** (via FFmpeg):
- Duration
- Container format
- Codec information (video/audio)
- Resolution (width × height)
- Bitrate
**Characteristics**:
- **Concurrency**: 16 workers (CPU count)
- **I/O Pattern**: Light disk reads + FFmpeg process spawning
- **Processing Time**: ~470ms average per file
- **Storage**: Temporary metadata stored in etcd with `tempId`
- **Virtual FS**: Files appear in virtual filesystem immediately after this stage
**Key Design Choice**: This stage extracts "just enough" metadata to populate the virtual filesystem and make files available to users, while deferring expensive hash computation to the next stage. This provides immediate value while processing continues in the background.
---
### Stage 3: Hash Computation (16 Workers)
**Location**: `packages/meta-sort/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts::computeHash()`
**Purpose**: Compute content hashes and perform deep media analysis.
**Implementation**:
```typescript
private async computeHash(filePath: string): Promise<void> {
const current = this.hashCount + 1;
const queueSize = this.discoveredCount;
await this.config.fileProcessor.processFile(filePath, current, queueSize);
this.hashCount++;
this.config.stateManager.completeHashProcessing(filePath);
}
```
**What is Computed**:
1. **Content Hashes** (via `meta-hash`):
- SHA-256 (primary hash, used for deduplication)
- SHA-1 (compatibility with legacy systems)
- MD5 (fast duplicate detection)
- CRC32 (checksum verification)
2. **Deep Media Analysis** (via FFmpeg):
- Audio track details (language, codec, channels)
- Subtitle track details (language, format)
- Chapter information
- Embedded metadata (title, artist, etc.)
**Characteristics**:
- **Concurrency**: 16 workers (CPU count)
- **I/O Pattern**: Full file read (streaming hash computation)
- **Processing Time**: Varies by file size (50MB/s typical throughput)
- **Storage**: Final metadata stored in etcd indexed by content hash
- **Memory**: Streaming hashing (fixed memory footprint regardless of file size)
**Key Design Choice**: Hash computation is CPU and I/O intensive, so we limit concurrency to CPU count to avoid thrashing. The hash is computed in streaming fashion (read chunks → update hash → repeat) to maintain constant memory usage even for 100GB+ files.
**Duplicate Detection**: Once a hash is computed, the system can:
- Detect exact duplicates (same hash)
- Detect version duplicates (same title/episode but different quality)
- Build virtual folder structure organized by content
---
### Stage 4: Completion
**Location**: `packages/meta-sort/packages/meta-sort-core/src/logic/UnifiedProcessingStateManager.ts::completeHashProcessing()`
**Purpose**: Mark file as fully processed and finalize metadata.
**Actions**:
1. Delete temporary metadata (`/tmp/{tempId}` key in etcd)
2. Store final metadata (`/file/{hash}` key in etcd)
3. Update virtual filesystem structure
4. Record performance metrics
5. Trigger duplicate detection (if enabled)
**Final Metadata Structure**:
```typescript
{
hash: "sha256:abc123...", // Primary content hash
filePath: "/data/watch/...", // Original file location
virtualPath: "/Anime/Series/S01/", // Virtual filesystem path
size: 1234567890, // File size in bytes
duration: 1440, // Duration in seconds
title: "Series Name", // Parsed title
season: 1, // Season number
episode: 5, // Episode number
resolution: "1920x1080", // Video resolution
codec: "h264", // Video codec
// ... many more fields
}
```
---
## Component Design
### FolderWatcher (meta-hash)
**File**: `packages/meta-hash/src/lib/folder-watcher/FolderWatcher.ts`
**Responsibility**: Generic file discovery with no business logic.
**Public API**:
```typescript
class FolderWatcher {
// Discover all files in directories (returns async generator)
async *discoverFiles(directories: string[]): AsyncGenerator<string>
// Watch directories for changes (using chokidar)
watch(
directories: string[],
callbacks: {
onAdd?: (filePath: string) => void | Promise<void>;
onChange?: (filePath: string) => void | Promise<void>;
onUnlink?: (filePath: string) => void | Promise<void>;
},
options?: WatchOptions
): chokidar.FSWatcher
}
```
**Design Rationale**:
- **Generic**: No knowledge of video files, extensions, or Meta Mesh concepts
- **Reusable**: Can be used in other projects (e.g., backup tools, file indexers)
- **Minimal Dependencies**: Only depends on Node.js built-ins and chokidar
- **Error Handling**: Gracefully handles permission errors, missing directories, etc.
---
### StreamingPipeline (meta-mesh)
**File**: `packages/meta-sort/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts`
**Responsibility**: Orchestrate multi-stage file processing with application-specific logic.
**Public API**:
```typescript
class StreamingPipeline {
// Start processing files from discovery stream
async start(discoveryStream: AsyncGenerator<string>): Promise<void>
// Handle file change events
async handleFileAdded(filePath: string): Promise<void>
async handleFileChanged(filePath: string): Promise<void>
async handleFileDeleted(filePath: string): Promise<void>
// Get pipeline statistics
getStats(): PipelineStats
}
```
**Design Rationale**:
- **Application-Specific**: Knows about video files, subtitles, torrents
- **Configurable**: Concurrency levels, supported extensions, validation rules
- **Observable**: Provides real-time statistics via `getStats()`
- **Fire-and-Forget**: Stages don't block each other (errors logged but don't halt pipeline)
---
### UnifiedProcessingStateManager (meta-mesh)
**File**: `packages/meta-sort/packages/meta-sort-core/src/logic/UnifiedProcessingStateManager.ts`
**Responsibility**: Track files through the 4-state processing lifecycle.
**States**:
```typescript
type FileState =
| 'pending' // Discovered, waiting for validation
| 'lightProcessing' // Extracting metadata
| 'hashProcessing' // Computing content hash
| 'done' // Fully processed
```
**Public API**:
```typescript
class UnifiedProcessingStateManager {
addPending(filePath: string): void
startLightProcessing(filePath: string): void
completeLightProcessing(filePath: string, tempId?: string): void
completeHashProcessing(filePath: string, hash?: string): void
removeFile(filePath: string): void
getSnapshot(): UnifiedProcessingSnapshot
}
```
**Design Rationale**:
- **Centralized State**: Single source of truth for file processing status
- **Performance Metrics**: Records timestamps for each stage transition
- **Observability**: Provides snapshot API for monitoring UIs
- **Memory Efficient**: Limits history to last 100 completed files
---
## Data Flow
### Initial Scan (Cold Start)
```
┌─────────────────────────────────────────────────────────────────────┐
│ 1. FolderWatcher starts async generator │
│ for await (const file of folderWatcher.discoverFiles([...])) │
└────────────────────────────┬────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────┐
│ 2. Discovery yields file paths one-by-one │
│ "/data/watch/Anime/Series/S01E01.mkv" ────────┐ │
│ "/data/watch/Anime/Series/S01E02.mkv" ────────┤ │
│ "/data/watch/Movies/Film.mp4" ────────┤ │
│ ... │ │
└────────────────────────────────────────────────────┼────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────┐
│ 3. StreamingPipeline.start() consumes generator │
│ - Validates each file (extension check) │
│ - Adds to metadata queue (fire-and-forget) │
│ - Logs progress every 1000 files │
└────────────────────────────┬────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────┐
│ 4. Validation Queue (32 workers) processes files │
│ - Filter by extension (.mkv, .mp4, .srt, etc.) │
│ - State: pending → ready for metadata │
│ - Queue typically empty (keeps up with discovery) │
└────────────────────────────┬────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────┐
│ 5. Metadata Queue (16 workers) processes files │
│ - Extract filename metadata (title, season, episode) │
│ - Run FFmpeg for basic media info │
│ - Store temp metadata in etcd (/tmp/{tempId}) │
│ - State: lightProcessing │
│ - Duration: ~470ms average │
└────────────────────────────┬────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────┐
│ 6. Hash Queue (16 workers) processes files │
│ - Compute content hashes (SHA-256, SHA-1, MD5, CRC32) │
│ - Deep FFmpeg analysis (all tracks, chapters) │
│ - Delete temp metadata, store final metadata (/file/{hash}) │
│ - Update virtual filesystem structure │
│ - State: hashProcessing → done │
│ - Duration: Varies by file size │
└─────────────────────────────────────────────────────────────────────┘
```
### File Change Event (Hot Path)
```
┌─────────────────────────────────────────────────────────────────────┐
│ 1. chokidar detects file system change │
│ - add: New file created │
│ - change: Existing file modified │
│ - unlink: File deleted │
└────────────────────────────┬────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────┐
│ 2. FolderWatcher invokes callback │
│ onAdd(filePath) ──▶ pipeline.handleFileAdded() │
│ onChange(filePath) ──▶ pipeline.handleFileChanged() │
│ onUnlink(filePath) ──▶ pipeline.handleFileDeleted() │
└────────────────────────────┬────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────┐
│ 3. Pipeline handles event │
│ - Added: Process new file through validation queue │
│ - Changed: Remove old metadata, reprocess through validation │
│ - Deleted: Remove from state manager and etcd │
└─────────────────────────────────────────────────────────────────────┘
```
---
## State Management
### File State Transitions
```
┌─────────┐ addPending() ┌────────────────┐
│ PENDING ├────────────────────────▶│ LIGHT │
└─────────┘ │ PROCESSING │
▲ └────────┬───────┘
│ │
│ removeFile() │ completeLightProcessing()
│ (on error) │
│ ▼
│ ┌────────────────┐
│ │ HASH │
└────────────────────────────────┤ PROCESSING │
└────────┬───────┘
│
│ completeHashProcessing()
│
▼
┌────────────────┐
│ DONE │
└────────────────┘
```
### State Storage Locations
| State | In-Memory Map | etcd Key | Notes |
|-------|--------------|----------|-------|
| pending | ✅ `Map<path, state>` | ❌ | Lightweight, no disk I/O |
| lightProcessing | ✅ `Map<path, state>` | ✅ `/tmp/{tempId}` | Temp metadata for virtual FS |
| hashProcessing | ✅ `Map<path, state>` | ✅ `/tmp/{tempId}` | Same as light processing |
| done | ✅ `Array<state>` (last 100) | ✅ `/file/{hash}` | Final metadata, temp deleted |
### State Metadata
Each state tracks detailed timing information:
```typescript
interface UnifiedFileState {
filePath: string;
state: FileState;
tempId?: string; // Temporary etcd key
hash?: string; // Final content hash (SHA-256)
virtualPath?: string; // Path in virtual filesystem
error?: string; // Error message if processing failed
// Timestamps
discoveredAt?: number;
lightProcessingStartedAt?: number;
lightProcessingCompletedAt?: number;
hashProcessingStartedAt?: number;
hashProcessingCompletedAt?: number;
// Processing times (milliseconds)
lightProcessingTime?: number;
hashProcessingTime?: number;
totalProcessingTime?: number;
}
```
This enables detailed performance analysis and bottleneck identification.
---
## Concurrency Model
### Queue Configuration
The pipeline uses **PQueue** (p-queue) for concurrency control with stage-specific limits:
```typescript
// packages/meta-sort/packages/meta-sort-core/src/index.ts
const cpuCount = os.cpus().length;
const validationConcurrency = config.MAX_WORKER_THREADS
? config.MAX_WORKER_THREADS * 2
: cpuCount * 2; // 32 on 16-core system
const metadataConcurrency = config.MAX_WORKER_THREADS
|| cpuCount; // 16 on 16-core system
const hashConcurrency = config.MAX_WORKER_THREADS
|| cpuCount; // 16 on 16-core system
```
### Concurrency Rationale
| Stage | Concurrency | Rationale |
|-------|-------------|-----------|
| **Validation** | CPU × 2 | Pure CPU-bound string operations, no I/O. High concurrency ensures validation never blocks discovery. |
| **Metadata** | CPU × 1 | Mixed CPU/I/O workload. Spawns FFmpeg processes. One per core prevents context switching overhead. |
| **Hash** | CPU × 1 | CPU and I/O intensive (reads entire file). One per core maximizes throughput without thrashing. |
### Backpressure Handling
PQueue provides natural backpressure:
```typescript
// Fire-and-forget pattern with error handling
this.metadataQueue.add(() => this.extractMetadata(filePath))
.catch(err => console.error(`[Pipeline] Metadata error:`, err.message));
```
- **Queue Size**: Number of pending tasks
- **Pending**: Number of tasks currently executing
- **Behavior**: When queue fills up, new tasks wait (no memory explosion)
### Memory Characteristics
For 900k files with 16 workers:
```
Memory Usage ≈ Queue Overhead + Worker Memory
Queue Overhead:
- Validation queue: ~0 (keeps up with discovery)
- Metadata queue: ~100-500 tasks = ~50KB
- Hash queue: ~1000-5000 tasks = ~500KB
Worker Memory:
- Metadata workers: 16 × ~50MB = 800MB
- Hash workers: 16 × ~100MB = 1.6GB
Total: ~2.5GB (well within 16GB Node.js heap limit)
```
---
## Performance Characteristics
### Discovery Phase
**Throughput**: ~1000-5000 files/second (depends on filesystem)
**Bottleneck**: Filesystem I/O (directory reads)
**Optimization**: Async generator yields to event loop, preventing blocking
**Metrics** (from logs):
```
[Pipeline] Progress: discovered=13000, validated=11937, metadata=1604, hash=0
```
- 13,000 files discovered
- 11,937 validated (91% match extensions)
- 1,604 metadata extracted (12% complete)
- 0 hashes computed (still in queue)
### Metadata Extraction Phase
**Throughput**: ~34 files/second (16 workers ÷ 470ms average)
**Bottleneck**: FFmpeg process spawning + basic media analysis
**Optimization**: Limited concurrency prevents process thrashing
**Average Times**:
- Filename parsing: ~5ms
- File stats: ~1ms
- FFmpeg basic info: ~460ms
- etcd write: ~4ms
- **Total**: ~470ms
### Hash Computation Phase
**Throughput**: Varies by file size
| File Size | Hash Time | Throughput |
|-----------|-----------|------------|
| 100MB | ~2 seconds | 50MB/s |
| 1GB | ~20 seconds | 50MB/s |
| 10GB | ~200 seconds | 50MB/s |
| 50GB | ~1000 seconds | 50MB/s |
**Bottleneck**: Disk read speed + SHA-256 computation
**Optimization**: Streaming hash (constant memory), parallel workers
### End-to-End Performance
For a typical dataset (46,000 files, ~5TB total):
| Metric | Time |
|--------|------|
| Discovery | ~30 seconds |
| First result | ~5 seconds |
| 50% complete | ~20 minutes |
| 100% complete | ~40 minutes |
**Key Improvement**: Users see results in 5 seconds (vs. 15+ minutes in old batch system)
---
## Architecture Decisions
### Decision 1: Streaming vs. Batching
**Problem**: Original batch processing had 15-minute discovery time before any processing started.
**Options Considered**:
1. **Optimize batch processing** (parallel directory traversal, caching)
2. **Hybrid approach** (batch discovery with parallel processing)
3. **Full streaming** (process as discovered)
**Decision**: Full streaming (Option 3)
**Rationale**:
- **User Experience**: Results appear within seconds
- **Memory Efficiency**: Bounded queue size prevents memory explosion
- **Simplicity**: Async generators are easier to reason about than batching logic
- **Scalability**: Works equally well for 100 files or 1 million files
---
### Decision 2: Separate Generic and Application Logic
**Problem**: Original `FolderWatcher` had hardcoded video file extensions, making it non-reusable.
**Options Considered**:
1. **Keep extension filtering in FolderWatcher** (simple, but not reusable)
2. **Pass extension filter to FolderWatcher** (more generic, but clutters API)
3. **Remove all filtering from FolderWatcher** (cleanest separation)
**Decision**: Remove all filtering (Option 3)
**Rationale**:
- **Reusability**: `meta-hash` can be used for non-media file projects
- **Single Responsibility**: FolderWatcher does one thing (discover files)
- **Flexibility**: Application can implement arbitrary filtering logic
- **Performance**: Minimal overhead (extension check is < 1ms)
---
### Decision 3: Three Separate Queues
**Problem**: Single queue with uniform concurrency doesn't match workload characteristics.
**Options Considered**:
1. **Single queue** (simple, but suboptimal)
2. **Two queues** (light + hash)
3. **Three queues** (validation + metadata + hash)
**Decision**: Three queues (Option 3)
**Rationale**:
- **Validation**: High concurrency (32) keeps up with discovery
- **Metadata**: Medium concurrency (16) balances FFmpeg overhead
- **Hash**: Medium concurrency (16) maximizes disk throughput
- **Independence**: Each stage can progress at its own pace
- **Observability**: Clear metrics per stage
---
### Decision 4: Skip MIME Validation
**Problem**: Extension checking might allow incorrect file types (e.g., `.mkv` file that's actually a text file).
**Options Considered**:
1. **Always validate MIME type** (accurate, but slow)
2. **Optional MIME validation** (configurable, but complex)
3. **Skip MIME validation** (fast, handle errors in later stages)
**Decision**: Skip MIME validation (Option 3)
**Rationale**:
- **Performance**: MIME validation requires reading 4KB of each file (~25% overhead)
- **Accuracy**: Extension checking is 99.9% accurate in practice
- **Error Handling**: FFmpeg will fail gracefully on invalid files
- **User Control**: Users control watched directories (low risk of wrong extensions)
---
### Decision 5: Fire-and-Forget Stage Progression
**Problem**: Should errors in one stage halt processing or be logged and continue?
**Options Considered**:
1. **Halt on error** (safe, but brittle)
2. **Retry with backoff** (robust, but complex)
3. **Log and continue** (fire-and-forget)
**Decision**: Log and continue (Option 3)
**Rationale**:
- **Resilience**: One corrupt file doesn't stop processing 899,999 good files
- **Observability**: Errors are logged with full context
- **Metrics**: Error count tracked in state manager
- **Manual Recovery**: Failed files can be manually reprocessed via API
---
### Decision 6: Async Generator vs. Event Emitter
**Problem**: How should FolderWatcher emit discovered files?
**Options Considered**:
1. **Event Emitter** (traditional Node.js pattern)
2. **Callback** (simple, but nested callbacks)
3. **Async Generator** (modern, composable)
**Decision**: Async Generator (Option 3)
**Rationale**:
- **Backpressure**: Consumer controls consumption rate via `for await`
- **Composability**: Can be chained, filtered, transformed with async iterators
- **Simplicity**: No callback hell, no event listener management
- **Performance**: Natural event loop yielding prevents blocking
```typescript
// ✅ Async Generator (Clean)
for await (const file of folderWatcher.discoverFiles([...])) {
await processFile(file);
}
// ❌ Event Emitter (More boilerplate)
folderWatcher.on('file', async (file) => {
await processFile(file);
});
folderWatcher.on('end', () => {
console.log('Discovery complete');
});
folderWatcher.start([...]);
```
---
## Appendix: Key Code Locations
| Component | File Path | Lines | Description |
|-----------|-----------|-------|-------------|
| FolderWatcher | `packages/meta-hash/src/lib/folder-watcher/FolderWatcher.ts` | 170 | Generic file discovery |
| StreamingPipeline | `packages/meta-sort/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts` | 237 | Multi-stage orchestrator |
| PipelineConfig | `packages/meta-sort/packages/meta-sort-core/src/logic/pipeline/PipelineConfig.ts` | 62 | Configuration interface |
| StateManager | `packages/meta-sort/packages/meta-sort-core/src/logic/UnifiedProcessingStateManager.ts` | 214 | State tracking |
| SupportedFileTypes | `packages/meta-sort/packages/meta-sort-core/src/config/SupportedFileTypes.ts` | 57 | Extension configuration |
| Main Entry | `packages/meta-sort/packages/meta-sort-core/src/index.ts` | 109 | Pipeline initialization |
---
## Glossary
- **Async Generator**: JavaScript function that returns an async iterable (uses `async function*` and `yield`)
- **Backpressure**: Flow control mechanism that prevents fast producers from overwhelming slow consumers
- **Event Loop**: Node.js single-threaded event processing loop (must not be blocked by long synchronous operations)
- **Fire-and-Forget**: Pattern where a function is called without awaiting its result (errors handled separately)
- **PQueue**: Third-party library for promise-based concurrency control (limits simultaneous async operations)
- **Streaming**: Processing data incrementally as it arrives (vs. batching, which waits for all data)
---
## References
- [Node.js Async Iterators](https://nodejs.org/api/esm.html#esm_async_iterators)
- [p-queue Documentation](https://github.com/sindresorhus/p-queue)
- [chokidar File Watching](https://github.com/paulmillr/chokidar)
- [FFmpeg Documentation](https://ffmpeg.org/documentation.html)
- [etcd Key-Value Store](https://etcd.io/docs/)
packages/meta-sort-core/src/api/UnifiedAPIServer.ts
+7 -15
View File
@@ -988,8 +988,6 @@ export class UnifiedAPIServer {
url: string; url: string;
api: string; api: string;
status: string; status: string;
capabilities: string[];
version: string;
}> = []; }> = [];
try { try {
@@ -998,17 +996,11 @@ export class UnifiedAPIServer {
const allServices = await serviceDiscovery.discoverAllServices(); const allServices = await serviceDiscovery.discoverAllServices();
for (const svc of allServices) { for (const svc of allServices) {
// Build dashboard URL from API URL
const apiUrl = svc.api || '';
const dashboardPath = svc.endpoints?.dashboard || '/';
services.push({ services.push({
name: svc.name || 'Unknown', name: svc.name || 'Unknown',
url: apiUrl + dashboardPath, url: svc.baseUrl || '',
api: apiUrl, api: svc.baseUrl || '',
status: svc.status || 'unknown', status: svc.status || 'unknown',
capabilities: svc.capabilities || [],
version: svc.version || '',
}); });
} }
} }
@@ -1018,17 +1010,17 @@ export class UnifiedAPIServer {
// Get leader info // Get leader info
const leaderInfo = this.kvManager!.getLeaderInfo(); const leaderInfo = this.kvManager!.getLeaderInfo();
const isLeader = this.kvManager!.isLeader();
return { return {
services, services,
current: 'meta-sort', current: 'meta-sort',
leader: leaderInfo ? { leader: leaderInfo ? {
host: leaderInfo.host, hostname: leaderInfo.hostname,
api: leaderInfo.api, baseUrl: leaderInfo.baseUrl,
http: leaderInfo.http, apiUrl: leaderInfo.apiUrl,
redisUrl: leaderInfo.redisUrl,
webdavUrl: leaderInfo.webdavUrl,
} : null, } : null,
isLeader,
}; };
}); });
} }
packages/meta-sort-core/src/index.ts
+17 -13
View File
@@ -141,26 +141,22 @@ process.on('SIGINT', async () => {
// Start initial discovery and processing (streaming pipeline) // Start initial discovery and processing (streaming pipeline)
(async () => { (async () => {
try { try {
// Validate META_CORE_URL is configured // Validate META_CORE_PATH is configured (required for leader discovery)
if (!config.META_CORE_URL) { if (!config.META_CORE_PATH) {
console.error('[Startup] ERROR: META_CORE_URL is required. Set META_CORE_URL environment variable.'); console.error('[Startup] ERROR: META_CORE_PATH is required. Set META_CORE_PATH environment variable.');
console.error('[Startup] Example: META_CORE_URL=http://meta-core'); console.error('[Startup] Example: META_CORE_PATH=/meta-core');
process.exit(1); process.exit(1);
} }
// Initialize KV Manager if configured (Redis + leader election) // Initialize KV Manager (Redis via leader discovery)
if (config.META_CORE_PATH) { if (config.META_CORE_PATH) {
console.log('[Startup] Initializing KV Manager (Redis + leader election)...'); console.log('[Startup] Initializing KV Manager (leader discovery)...');
kvManager = new KVManager({ kvManager = new KVManager({
metaCorePath: config.META_CORE_PATH, metaCorePath: config.META_CORE_PATH,
filesPath: config.FILES_PATH, filesPath: config.FILES_PATH,
serviceName: config.SERVICE_NAME, serviceName: config.SERVICE_NAME,
version: config.SERVICE_VERSION,
apiPort: config.FUSE_API_PORT, apiPort: config.FUSE_API_PORT,
redisPort: config.REDIS_PORT,
redisUrl: config.REDIS_URL,
baseUrl: config.BASE_URL, baseUrl: config.BASE_URL,
capabilities: ['write', 'mount', 'monitor']
}); });
// Wait for KV to be ready // Wait for KV to be ready
@@ -172,7 +168,7 @@ process.on('SIGINT', async () => {
if (client) { if (client) {
fileProcessor.setKVClient(client); fileProcessor.setKVClient(client);
pipeline.setKVClient(client); pipeline.setKVClient(client);
console.log(`[Startup] KV Manager ready (role: ${kvManager.isLeader() ? 'LEADER' : 'FOLLOWER'})`); console.log('[Startup] KV Manager ready');
} }
} }
@@ -266,9 +262,17 @@ process.on('SIGINT', async () => {
} }
// Connect to meta-core for file events via SSE (Architecture V3) // Connect to meta-core for file events via SSE (Architecture V3)
console.log(`[Startup] Connecting to meta-core at ${config.META_CORE_URL}...`); // Get meta-core URL from leader discovery
const leaderInfo = kvManager?.getLeaderInfo();
if (!leaderInfo) {
console.error('[Startup] ERROR: No leader info available. Cannot connect to meta-core.');
console.error('[Startup] Make sure meta-core is running and kv-leader.info exists.');
process.exit(1);
}
const metaCoreUrl = leaderInfo.baseUrl;
console.log(`[Startup] Connecting to meta-core at ${metaCoreUrl}...`);
eventSubscriber = new EventSubscriber({ eventSubscriber = new EventSubscriber({
metaCoreUrl: config.META_CORE_URL, metaCoreUrl: metaCoreUrl,
pipeline: pipeline, pipeline: pipeline,
requestInitialScan: true, // Request initial scan from meta-core requestInitialScan: true, // Request initial scan from meta-core
reconnectDelayMs: 5000, reconnectDelayMs: 5000,
packages/meta-sort-core/src/kv/IKVClient.ts
+19 -29
View File
@@ -139,20 +139,24 @@ export interface KVClientConfig {
} }
/** /**
* Leader lock file content format * Leader lock file content format (kv-leader.info)
* Written by meta-core leader, read by other services
*/ */
export interface LeaderLockInfo { export interface LeaderLockInfo {
/** Hostname of the leader */ /** Hostname of the leader */
host: string; hostname: string;
/** Base URL for the leader service (e.g., http://localhost:8180) */
baseUrl: string;
/** meta-core API URL (port 9000) */
apiUrl: string;
/** Redis connection URL (e.g., redis://10.0.1.50:6379) */ /** Redis connection URL (e.g., redis://10.0.1.50:6379) */
api: string; redisUrl: string;
/** HTTP API URL for the leader service */ /** WebDAV URL for file access */
http: string; webdavUrl: string;
/** Configured BASE_URL for stable service discovery (optional, falls back to http) */
baseUrl?: string;
/** Timestamp when leadership was acquired */ /** Timestamp when leadership was acquired */
timestamp: number; timestamp: number;
@@ -162,36 +166,22 @@ export interface LeaderLockInfo {
} }
/** /**
* Service registration info * Service registration info (simplified)
* Full URLs are obtained via the /urls API endpoint
*/ */
export interface ServiceInfo { export interface ServiceInfo {
/** Service name (e.g., 'meta-sort', 'meta-fuse') */ /** Service name (e.g., 'meta-sort', 'meta-fuse') */
name: string; name: string;
/** Service version */
version: string;
/** HTTP API URL */
api: string;
/** Current status */
status: 'starting' | 'running' | 'stopping' | 'stopped';
/** Process ID */
pid: number;
/** Hostname */ /** Hostname */
hostname: string; hostname: string;
/** Start timestamp (ISO string) */ /** Base URL for the service */
startedAt: string; baseUrl: string;
/** Current status */
status: 'running' | 'stale' | 'stopped';
/** Last heartbeat timestamp (ISO string) */ /** Last heartbeat timestamp (ISO string) */
lastHeartbeat: string; lastHeartbeat: string;
/** Service capabilities */
capabilities: string[];
/** Named endpoints */
endpoints: Record<string, string>;
} }
packages/meta-sort-core/src/kv/KVManager.ts
+88 -149
View File
@@ -1,11 +1,14 @@
/** /**
* KV Manager - Unified manager for KV storage, leader election, and service discovery * KV Manager - Unified manager for KV storage and service discovery
* *
* This is the main entry point for the KV subsystem. It: * This is the main entry point for the KV subsystem. It:
* 1. Participates in leader election (spawns Redis if leader) * 1. Uses LeaderClient to read leader info from meta-core
* 2. Creates and manages the KV client (Redis) * 2. Creates and manages the KV client (Redis)
* 3. Handles service discovery registration * 3. Handles service discovery registration
* 4. Manages reconnection on leader failure * 4. Manages reconnection on leader changes
*
* Note: Leader election is now handled by meta-core (Go sidecar).
* This service only reads the leader info and connects to Redis.
* *
* Usage: * Usage:
* ```typescript * ```typescript
@@ -23,8 +26,8 @@
* ``` * ```
*/ */
import { networkInterfaces } from 'os'; import { hostname } from 'os';
import { LeaderElection } from './LeaderElection.js'; import { LeaderClient } from './LeaderClient.js';
import { ServiceDiscovery } from './ServiceDiscovery.js'; import { ServiceDiscovery } from './ServiceDiscovery.js';
import { RedisKVClient } from './RedisClient.js'; import { RedisKVClient } from './RedisClient.js';
import type { IKVClient, LeaderLockInfo } from './IKVClient.js'; import type { IKVClient, LeaderLockInfo } from './IKVClient.js';
@@ -39,34 +42,22 @@ interface KVManagerConfig {
/** Service name (e.g., 'meta-sort', 'meta-fuse') */ /** Service name (e.g., 'meta-sort', 'meta-fuse') */
serviceName: string; serviceName: string;
/** Service version */
version?: string;
/** HTTP API port */ /** HTTP API port */
apiPort: number; apiPort: number;
/** Redis port (default: 6379) */
redisPort?: number;
/** Service capabilities */
capabilities?: string[];
/** Whether to skip leader election and use provided Redis URL */
redisUrl?: string;
/** Base URL for service discovery (overrides auto-detected URL) */ /** Base URL for service discovery (overrides auto-detected URL) */
baseUrl?: string; baseUrl?: string;
/** /** Optional direct Redis URL (bypasses leader discovery) */
* Hostname to advertise in lock file for other services to connect. redisUrl?: string;
* Use Docker service name (e.g., 'meta-sort-dev') for stable DNS resolution.
*/ /** meta-core API URL (e.g., http://localhost:9000) */
advertiseHost?: string; metaCoreUrl?: string;
} }
export class KVManager { export class KVManager {
private config: KVManagerConfig; private config: KVManagerConfig;
private leaderElection: LeaderElection | null = null; private leaderClient: LeaderClient | null = null;
private serviceDiscovery: ServiceDiscovery | null = null; private serviceDiscovery: ServiceDiscovery | null = null;
private kvClient: IKVClient | null = null; private kvClient: IKVClient | null = null;
private isStarted = false; private isStarted = false;
@@ -77,32 +68,7 @@ export class KVManager {
private onReconnectCallbacks: (() => void)[] = []; private onReconnectCallbacks: (() => void)[] = [];
constructor(config: KVManagerConfig) { constructor(config: KVManagerConfig) {
this.config = { this.config = config;
version: '1.0.0',
redisPort: 6379,
capabilities: [],
...config
};
}
/**
* Get the local IP for API URL construction
*/
private getLocalIP(): string {
const interfaces = networkInterfaces();
for (const name of Object.keys(interfaces)) {
const addrs = interfaces[name];
if (!addrs) continue;
for (const addr of addrs) {
if (addr.family === 'IPv4' && !addr.internal) {
return addr.address;
}
}
}
return 'localhost';
} }
/** /**
@@ -130,106 +96,86 @@ export class KVManager {
console.log(`[KVManager] Starting for ${this.config.serviceName}...`); console.log(`[KVManager] Starting for ${this.config.serviceName}...`);
// Check for direct Redis URL (skip leader election) // Check for direct Redis URL (skip leader discovery)
if (this.config.redisUrl) { if (this.config.redisUrl) {
console.log(`[KVManager] Using direct Redis URL: ${this.config.redisUrl}`); console.log(`[KVManager] Using direct Redis URL: ${this.config.redisUrl}`);
this.kvClient = await this.createRedisClient(this.config.redisUrl); this.kvClient = await this.createRedisClient(this.config.redisUrl);
// NOTE: Service discovery is handled by Go meta-core sidecar
// The Go binary writes to /meta-core/services/{serviceName}.json with correct internal IPs
// TypeScript ServiceDiscovery would overwrite with external BASE_URL, breaking internal discovery
// Keep ServiceDiscovery instance for reading/discovery only, but don't call start()
const apiUrl = this.config.baseUrl || `http://localhost:${this.config.apiPort}`;
this.serviceDiscovery = new ServiceDiscovery({
metaCorePath: this.config.metaCorePath,
serviceName: this.config.serviceName,
version: this.config.version!,
apiUrl,
capabilities: this.config.capabilities,
endpoints: {
health: '/health',
dashboard: '/',
api: '/api',
metrics: '/api/metrics',
processing: '/api/processing/status'
}
});
// Don't call start() - Go meta-core handles registration
// await this.serviceDiscovery.start();
this.isStarted = true; this.isStarted = true;
this.notifyReady(); this.notifyReady();
return; return;
} }
// Initialize leader election // Initialize leader client
const localIP = this.getLocalIP(); this.leaderClient = new LeaderClient({
metaCorePath: this.config.metaCorePath,
metaCoreUrl: this.config.metaCoreUrl
});
this.leaderElection = new LeaderElection({ // Watch for leader changes
this.leaderClient.onChange(async () => {
console.log('[KVManager] Leader changed, reconnecting...');
await this.reconnect();
});
// Wait for leader and connect
try {
const leaderInfo = await this.leaderClient.waitForLeader(30000);
console.log(`[KVManager] Connecting to Redis at ${leaderInfo.redisUrl}...`);
this.kvClient = await this.createRedisClient(leaderInfo.redisUrl);
} catch (error) {
console.error('[KVManager] Failed to connect to leader:', error);
throw error;
}
// Start watching for leader changes
this.leaderClient.startWatching();
// Initialize service discovery (meta-core handles actual registration)
// We keep this for service discovery (reading other services)
const apiUrl = this.config.baseUrl || `http://${hostname()}:${this.config.apiPort}`;
this.serviceDiscovery = new ServiceDiscovery({
metaCorePath: this.config.metaCorePath, metaCorePath: this.config.metaCorePath,
serviceName: this.config.serviceName, serviceName: this.config.serviceName,
apiPort: this.config.apiPort, version: '1.0.0',
redisPort: this.config.redisPort, apiUrl
baseUrl: this.config.baseUrl
}); });
// Note: We don't start ServiceDiscovery - meta-core handles registration
// Setup leader election callbacks this.isStarted = true;
this.leaderElection.onLeader(async () => {
console.log('[KVManager] We are the leader, connecting to local Redis...');
// Connect to local Redis (spawned by leader election)
const redisUrl = `redis://127.0.0.1:${this.config.redisPort}`;
this.kvClient = await this.createRedisClient(redisUrl);
this.notifyReady(); this.notifyReady();
}); }
this.leaderElection.onFollower(async (leaderInfo: LeaderLockInfo) => { /**
console.log(`[KVManager] We are a follower, connecting to leader at ${leaderInfo.api}...`); * Reconnect to Redis after leader change
*/
// Connect to leader's Redis private async reconnect(): Promise<void> {
this.kvClient = await this.createRedisClient(leaderInfo.api); if (this.isShuttingDown || !this.leaderClient) {
return;
this.notifyReady(); }
});
this.leaderElection.onLostLeader(async () => {
console.log('[KVManager] Leader lost, disconnecting...');
// Close existing connection
if (this.kvClient) { if (this.kvClient) {
await this.kvClient.close(); await this.kvClient.close();
this.kvClient = null; this.kvClient = null;
} }
// Leader election will handle re-election and reconnection // Wait for new leader
}); try {
const leaderInfo = await this.leaderClient.waitForLeader(30000);
console.log(`[KVManager] Reconnecting to Redis at ${leaderInfo.redisUrl}...`);
this.kvClient = await this.createRedisClient(leaderInfo.redisUrl);
// Initialize service discovery // Notify reconnect callbacks
// Use BASE_URL if provided, otherwise fall back to auto-detected local IP for (const callback of this.onReconnectCallbacks) {
const apiUrl = this.config.baseUrl || `http://${localIP}:${this.config.apiPort}`; try {
callback();
this.serviceDiscovery = new ServiceDiscovery({ } catch (error) {
metaCorePath: this.config.metaCorePath, console.error('[KVManager] Error in reconnect callback:', error);
serviceName: this.config.serviceName, }
version: this.config.version!, }
apiUrl, } catch (error) {
capabilities: this.config.capabilities, console.error('[KVManager] Failed to reconnect:', error);
endpoints: {
health: '/health',
dashboard: '/',
api: '/api',
metrics: '/api/metrics',
processing: '/api/processing/status'
} }
});
// Start leader election first
await this.leaderElection.start();
// Start service discovery
await this.serviceDiscovery.start();
this.isStarted = true;
} }
/** /**
@@ -241,10 +187,10 @@ export class KVManager {
console.log('[KVManager] Stopping...'); console.log('[KVManager] Stopping...');
this.isShuttingDown = true; this.isShuttingDown = true;
// Stop service discovery // Stop leader client
if (this.serviceDiscovery) { if (this.leaderClient) {
await this.serviceDiscovery.stop(); this.leaderClient.close();
this.serviceDiscovery = null; this.leaderClient = null;
} }
// Close KV client // Close KV client
@@ -253,10 +199,10 @@ export class KVManager {
this.kvClient = null; this.kvClient = null;
} }
// Stop leader election (releases lock, stops Redis if leader) // Service discovery stop (if it was started)
if (this.leaderElection) { if (this.serviceDiscovery) {
await this.leaderElection.stop(); await this.serviceDiscovery.stop();
this.leaderElection = null; this.serviceDiscovery = null;
} }
this.isStarted = false; this.isStarted = false;
@@ -337,25 +283,18 @@ export class KVManager {
return this.kvClient; return this.kvClient;
} }
/**
* Check if this service is the leader
*/
isLeader(): boolean {
return this.leaderElection?.isLeader() ?? false;
}
/**
* Check if this service is a follower
*/
isFollower(): boolean {
return this.leaderElection?.isFollower() ?? false;
}
/** /**
* Get the current leader info * Get the current leader info
*/ */
getLeaderInfo(): LeaderLockInfo | null { getLeaderInfo(): LeaderLockInfo | null {
return this.leaderElection?.getLeaderInfo() ?? null; return this.leaderClient?.getCachedLeaderInfo() ?? null;
}
/**
* Get the leader client
*/
getLeaderClient(): LeaderClient | null {
return this.leaderClient;
} }
/** /**
packages/meta-sort-core/src/kv/LeaderClient.ts
+216
View File
@@ -0,0 +1,216 @@
/**
* LeaderClient - Client for reading leader info from meta-core
*
* This replaces the old LeaderElection.ts by delegating leader election
* to meta-core (Go sidecar) and just reading the results.
*
* Features:
* - Reads kv-leader.info file for leader URLs
* - Calls meta-core /urls API for current leader URLs
* - Watches for leader changes via file system watcher
*/
import { promises as fs } from 'fs';
import { watch, FSWatcher } from 'fs';
import { dirname } from 'path';
import type { LeaderLockInfo } from './IKVClient.js';
export interface LeaderClientConfig {
/** Path to META_CORE_VOLUME (e.g., /meta-core) */
metaCorePath: string;
/** meta-core API URL (e.g., http://meta-core:9000 or http://localhost:9000) */
metaCoreUrl?: string;
}
export interface URLsResponse {
hostname: string;
baseUrl: string;
apiUrl: string;
redisUrl: string;
webdavUrl: string;
isLeader: boolean;
}
export class LeaderClient {
private config: LeaderClientConfig;
private infoFilePath: string;
private leaderInfo: LeaderLockInfo | null = null;
private watcher: FSWatcher | null = null;
private onChangeCallbacks: (() => void)[] = [];
constructor(config: LeaderClientConfig) {
this.config = config;
this.infoFilePath = `${config.metaCorePath}/locks/kv-leader.info`;
}
/**
* Read leader info from the info file
*/
async getLeaderInfo(): Promise<LeaderLockInfo | null> {
try {
const content = await fs.readFile(this.infoFilePath, 'utf-8');
this.leaderInfo = JSON.parse(content) as LeaderLockInfo;
return this.leaderInfo;
} catch (error) {
console.error('[LeaderClient] Failed to read leader info:', error);
return null;
}
}
/**
* Get Redis URL from leader info
*/
async getRedisUrl(): Promise<string | null> {
const info = await this.getLeaderInfo();
return info?.redisUrl ?? null;
}
/**
* Get WebDAV URL from leader info
*/
async getWebdavUrl(): Promise<string | null> {
const info = await this.getLeaderInfo();
return info?.webdavUrl ?? null;
}
/**
* Get meta-core API URL from leader info
*/
async getApiUrl(): Promise<string | null> {
const info = await this.getLeaderInfo();
return info?.apiUrl ?? null;
}
/**
* Call meta-core /urls API to get current URLs
* Useful for getting URLs when you don't want to read the file directly
*/
async getUrls(): Promise<URLsResponse | null> {
// First try using configured metaCoreUrl
let apiUrl = this.config.metaCoreUrl;
// Fall back to reading from leader info
if (!apiUrl) {
const info = await this.getLeaderInfo();
apiUrl = info?.apiUrl;
}
if (!apiUrl) {
console.error('[LeaderClient] No meta-core API URL available');
return null;
}
try {
const response = await fetch(`${apiUrl}/urls`, {
method: 'GET',
headers: { 'Accept': 'application/json' },
signal: AbortSignal.timeout(5000)
});
if (!response.ok) {
console.error('[LeaderClient] Failed to get URLs:', response.status, response.statusText);
return null;
}
return await response.json() as URLsResponse;
} catch (error) {
console.error('[LeaderClient] Error calling /urls API:', error);
return null;
}
}
/**
* Wait for leader info to be available
*/
async waitForLeader(timeoutMs: number = 30000): Promise<LeaderLockInfo> {
const startTime = Date.now();
const pollInterval = 500;
while (Date.now() - startTime < timeoutMs) {
const info = await this.getLeaderInfo();
if (info) {
console.log(`[LeaderClient] Leader found: ${info.hostname} at ${info.redisUrl}`);
return info;
}
console.log('[LeaderClient] Waiting for leader...');
await new Promise(resolve => setTimeout(resolve, pollInterval));
}
throw new Error(`No leader found within ${timeoutMs}ms`);
}
/**
* Start watching for leader changes
*/
startWatching(): void {
if (this.watcher) {
return; // Already watching
}
const lockDir = dirname(this.infoFilePath);
try {
this.watcher = watch(lockDir, (eventType, filename) => {
if (filename === 'kv-leader.info') {
console.log('[LeaderClient] Leader info changed, reloading...');
this.getLeaderInfo().then(() => {
this.notifyChange();
}).catch(console.error);
}
});
console.log(`[LeaderClient] Watching for leader changes in ${lockDir}`);
} catch (error) {
console.error('[LeaderClient] Failed to start watching:', error);
}
}
/**
* Stop watching for leader changes
*/
stopWatching(): void {
if (this.watcher) {
this.watcher.close();
this.watcher = null;
console.log('[LeaderClient] Stopped watching for leader changes');
}
}
/**
* Register callback for leader changes
*/
onChange(callback: () => void): this {
this.onChangeCallbacks.push(callback);
return this;
}
/**
* Notify all change callbacks
*/
private notifyChange(): void {
for (const callback of this.onChangeCallbacks) {
try {
callback();
} catch (error) {
console.error('[LeaderClient] Error in change callback:', error);
}
}
}
/**
* Get cached leader info (without re-reading file)
*/
getCachedLeaderInfo(): LeaderLockInfo | null {
return this.leaderInfo;
}
/**
* Clean up resources
*/
close(): void {
this.stopWatching();
this.onChangeCallbacks = [];
}
}
packages/meta-sort-core/src/kv/LeaderElection.ts
-615
View File
@@ -1,615 +0,0 @@
/**
* Leader Election using flock on shared filesystem
*
* This implements a distributed leader election mechanism using:
* 1. flock (advisory file locking) for mutual exclusion
* 2. Shared filesystem as the consensus layer
* 3. Separate info file for service discovery
*
* Architecture:
* - Service tries to acquire exclusive flock on /meta-core/locks/kv-leader.lock
* - If acquired → becomes LEADER, spawns Redis, writes endpoint to info file
* - If blocked → becomes FOLLOWER, reads leader endpoint from info file
* - flock automatically releases when process dies (no stale lock issues!)
*
* Key advantage over O_EXCL approach:
* - Lock is automatically released when process exits/crashes
* - No need for timestamp-based stale detection
* - No race conditions during takeover
*/
import { promises as fs } from 'fs';
import { dirname } from 'path';
import { hostname, networkInterfaces } from 'os';
import { spawn, ChildProcess, execSync } from 'child_process';
import * as IORedis from 'ioredis';
import type { LeaderLockInfo } from './IKVClient.js';
// ESM/CJS interop for ioredis
const Redis = (IORedis as any).default ?? IORedis;
interface LeaderElectionConfig {
/** Path to META_CORE_VOLUME (e.g., /meta-core) */
metaCorePath: string;
/** Service name for identification */
serviceName: string;
/** HTTP API port */
apiPort: number;
/** Redis port (default: 6379) */
redisPort?: number;
/** Redis data directory (relative to META_CORE_VOLUME/db/redis) */
redisDataDir?: string;
/** Health check interval in ms (default: 5000) */
healthCheckInterval?: number;
/** Base URL for stable service discovery (e.g., http://localhost:8180) */
baseUrl?: string;
}
type LeaderRole = 'leader' | 'follower' | 'unknown';
export class LeaderElection {
private config: LeaderElectionConfig;
private lockFilePath: string;
private infoFilePath: string;
private flockProcess: ChildProcess | null = null;
private role: LeaderRole = 'unknown';
private leaderInfo: LeaderLockInfo | null = null;
private redisProcess: ChildProcess | null = null;
private healthCheckTimer: NodeJS.Timeout | null = null;
private isShuttingDown = false;
// Event callbacks
private onBecomeLeader?: () => void;
private onBecomeFollower?: (leaderInfo: LeaderLockInfo) => void;
private onLeaderLost?: () => void;
constructor(config: LeaderElectionConfig) {
this.config = {
redisPort: 6379,
healthCheckInterval: 5000,
...config
};
this.lockFilePath = `${this.config.metaCorePath}/locks/kv-leader.lock`;
this.infoFilePath = `${this.config.metaCorePath}/locks/kv-leader.info`;
}
/**
* Get the local machine's IP address
*/
private getLocalIP(): string {
const interfaces = networkInterfaces();
// Prefer non-internal IPv4 addresses
for (const name of Object.keys(interfaces)) {
const addrs = interfaces[name];
if (!addrs) continue;
for (const addr of addrs) {
if (addr.family === 'IPv4' && !addr.internal) {
return addr.address;
}
}
}
// Fallback to hostname or localhost
return hostname() || 'localhost';
}
/**
* Build leader info for this service
*/
private buildLeaderInfo(): LeaderLockInfo {
const ip = this.getLocalIP();
return {
host: hostname(),
api: `redis://${ip}:${this.config.redisPort}`,
http: `http://${ip}:${this.config.apiPort}`,
baseUrl: this.config.baseUrl,
timestamp: Date.now(),
pid: process.pid
};
}
/**
* Ensure lock directory exists
*/
private async ensureLockDir(): Promise<void> {
const lockDir = dirname(this.lockFilePath);
await fs.mkdir(lockDir, { recursive: true });
}
/**
* Try to acquire the leader lock using flock
* Returns true if lock acquired (we are leader), false otherwise
*
* Uses `flock -n -x` for non-blocking exclusive lock attempt.
* The lock is held by keeping a child process alive with the lock fd open.
*/
private async tryAcquireLock(): Promise<boolean> {
await this.ensureLockDir();
// Ensure lock file exists (flock needs an existing file)
try {
await fs.writeFile(this.lockFilePath, '', { flag: 'a' });
} catch {
// Ignore if file already exists
}
return new Promise((resolve) => {
// Spawn a process that tries to acquire the lock and holds it
// Using flock with file descriptor: open fd 200 on lock file, then sleep forever
// The lock is held as long as fd 200 is open (process alive)
this.flockProcess = spawn('bash', [
'-c',
`exec 200>"${this.lockFilePath}" && flock -n -x 200 && echo LOCKED && while true; do sleep 86400; done`
], {
stdio: ['ignore', 'pipe', 'pipe']
});
let resolved = false;
// Check stdout for "LOCKED" message indicating we got the lock
this.flockProcess.stdout?.on('data', (data) => {
const output = data.toString().trim();
if (output === 'LOCKED' && !resolved) {
resolved = true;
console.log('[LeaderElection] Acquired flock on', this.lockFilePath);
resolve(true);
}
});
this.flockProcess.stderr?.on('data', (data) => {
const msg = data.toString().trim();
if (msg) {
console.error('[LeaderElection] flock stderr:', msg);
}
});
// If process exits quickly, we didn't get the lock
this.flockProcess.on('exit', (code) => {
if (!resolved) {
resolved = true;
console.log('[LeaderElection] Could not acquire flock (lock held by another process)');
this.flockProcess = null;
resolve(false);
} else if (!this.isShuttingDown && this.role === 'leader') {
// Lock process died unexpectedly while we were leader
console.error('[LeaderElection] flock process died unexpectedly!');
this.flockProcess = null;
this.handleLockLost();
}
});
// Timeout: if we don't get the lock within 2 seconds, assume it's held
setTimeout(() => {
if (!resolved) {
resolved = true;
if (this.flockProcess) {
this.flockProcess.kill();
this.flockProcess = null;
}
console.log('[LeaderElection] Timeout acquiring flock (lock held by another process)');
resolve(false);
}
}, 2000);
});
}
/**
* Handle losing the lock unexpectedly
*/
private async handleLockLost(): Promise<void> {
if (this.isShuttingDown) return;
console.log('[LeaderElection] Lost leadership, cleaning up...');
this.role = 'unknown';
// Stop Redis
await this.stopRedis();
// Notify callback
this.onLeaderLost?.();
// Try to re-elect
const acquired = await this.tryAcquireLock();
if (acquired) {
await this.transitionToLeader();
} else {
await this.transitionToFollower();
}
}
/**
* Check if the lock is currently held by someone
* Uses flock -n which fails immediately if lock is held
*/
private isLockHeld(): boolean {
try {
// Try to acquire lock non-blocking - if it succeeds, no one holds it
execSync(`flock -n -x "${this.lockFilePath}" -c "exit 0"`, {
timeout: 1000,
stdio: 'ignore'
});
// Lock was free (we briefly held it and released)
return false;
} catch {
// Lock is held by someone else
return true;
}
}
/**
* Read leader info from info file
*/
private async readLeaderInfo(): Promise<LeaderLockInfo | null> {
try {
const content = await fs.readFile(this.infoFilePath, 'utf-8');
return JSON.parse(content) as LeaderLockInfo;
} catch {
return null;
}
}
/**
* Write leader info to info file (atomic write)
*/
private async writeLeaderInfo(): Promise<void> {
this.leaderInfo = this.buildLeaderInfo();
// Atomic write: write to temp file, then rename
const tempPath = `${this.infoFilePath}.tmp`;
await fs.writeFile(tempPath, JSON.stringify(this.leaderInfo, null, 2));
await fs.rename(tempPath, this.infoFilePath);
console.log('[LeaderElection] Wrote leader info to', this.infoFilePath);
}
/**
* Update leader timestamp (heartbeat) using atomic write
*/
private async updateLeaderTimestamp(): Promise<void> {
if (this.role !== 'leader' || !this.leaderInfo) return;
this.leaderInfo.timestamp = Date.now();
try {
// Atomic write: write to temp file, then rename
const tempPath = `${this.infoFilePath}.tmp`;
await fs.writeFile(tempPath, JSON.stringify(this.leaderInfo, null, 2));
await fs.rename(tempPath, this.infoFilePath);
} catch (error) {
console.error('[LeaderElection] Failed to update timestamp:', error);
}
}
/**
* Spawn Redis server as child process
*/
private async spawnRedis(): Promise<void> {
const redisDataDir = `${this.config.metaCorePath}/db/redis`;
// Ensure Redis data directory exists
await fs.mkdir(redisDataDir, { recursive: true });
console.log(`[LeaderElection] Spawning Redis on port ${this.config.redisPort}...`);
// Spawn Redis server
this.redisProcess = spawn('redis-server', [
'--port', String(this.config.redisPort),
'--bind', '0.0.0.0',
'--dir', redisDataDir,
'--appendonly', 'yes',
'--appendfilename', 'appendonly.aof',
'--dbfilename', 'dump.rdb',
'--save', '60', '1', // Save after 60 seconds if at least 1 key changed
'--loglevel', 'warning'
], {
stdio: ['ignore', 'pipe', 'pipe'],
detached: false
});
this.redisProcess.stdout?.on('data', (data) => {
console.log(`[Redis] ${data.toString().trim()}`);
});
this.redisProcess.stderr?.on('data', (data) => {
console.error(`[Redis] ${data.toString().trim()}`);
});
this.redisProcess.on('exit', (code, signal) => {
console.log(`[Redis] Process exited with code ${code}, signal ${signal}`);
this.redisProcess = null;
if (!this.isShuttingDown && this.role === 'leader') {
// Redis crashed - try to restart
console.log('[LeaderElection] Redis crashed, attempting restart...');
this.spawnRedis().catch(console.error);
}
});
// Wait for Redis to be ready
await this.waitForRedisReady();
}
/**
* Wait for Redis to be ready to accept connections
*/
private async waitForRedisReady(): Promise<void> {
const maxAttempts = 30;
const ip = this.getLocalIP();
for (let i = 0; i < maxAttempts; i++) {
try {
const client = new Redis({
host: ip,
port: this.config.redisPort,
connectTimeout: 1000,
maxRetriesPerRequest: 1
});
const pong = await client.ping();
await client.quit();
if (pong === 'PONG') {
console.log('[LeaderElection] Redis is ready');
return;
}
} catch {
// Redis not ready yet
}
await new Promise(resolve => setTimeout(resolve, 1000));
}
throw new Error('Redis failed to start within timeout');
}
/**
* Stop Redis server
*/
private async stopRedis(): Promise<void> {
if (this.redisProcess) {
console.log('[LeaderElection] Stopping Redis...');
this.redisProcess.kill('SIGTERM');
// Wait for graceful shutdown
await new Promise<void>((resolve) => {
const timeout = setTimeout(() => {
if (this.redisProcess) {
this.redisProcess.kill('SIGKILL');
}
resolve();
}, 5000);
this.redisProcess!.on('exit', () => {
clearTimeout(timeout);
resolve();
});
});
this.redisProcess = null;
}
}
/**
* Release the leader lock by killing the flock process
*/
private async releaseLock(): Promise<void> {
if (this.flockProcess) {
console.log('[LeaderElection] Releasing flock...');
// Kill the flock holder process - this releases the lock
this.flockProcess.kill('SIGTERM');
// Wait briefly for graceful exit
await new Promise<void>((resolve) => {
const timeout = setTimeout(() => {
if (this.flockProcess) {
this.flockProcess.kill('SIGKILL');
}
resolve();
}, 1000);
if (this.flockProcess) {
this.flockProcess.on('exit', () => {
clearTimeout(timeout);
resolve();
});
} else {
clearTimeout(timeout);
resolve();
}
});
this.flockProcess = null;
}
// Clean up info file (optional, but good hygiene)
try {
await fs.unlink(this.infoFilePath);
} catch {
// Ignore if file doesn't exist
}
}
/**
* Start health check loop
*/
private startHealthCheck(): void {
this.healthCheckTimer = setInterval(async () => {
if (this.isShuttingDown) return;
try {
if (this.role === 'leader') {
// Update timestamp as heartbeat
await this.updateLeaderTimestamp();
// Verify we still hold the lock
if (!this.flockProcess || this.flockProcess.exitCode !== null) {
console.error('[LeaderElection] Lost flock process!');
await this.handleLockLost();
}
} else if (this.role === 'follower') {
// Check if lock is still held (leader still alive)
if (!this.isLockHeld()) {
console.log('[LeaderElection] Lock is free, attempting to become leader...');
this.onLeaderLost?.();
// Try to become leader
const acquired = await this.tryAcquireLock();
if (acquired) {
await this.transitionToLeader();
} else {
// Another service became leader - reconnect
await this.transitionToFollower();
}
}
}
} catch (error) {
console.error('[LeaderElection] Health check error:', error);
}
}, this.config.healthCheckInterval);
}
/**
* Transition to leader role
*/
private async transitionToLeader(): Promise<void> {
console.log('[LeaderElection] Becoming LEADER');
this.role = 'leader';
// Write leader info BEFORE spawning Redis so followers can find us
await this.writeLeaderInfo();
// Spawn Redis
await this.spawnRedis();
// Notify callback
this.onBecomeLeader?.();
}
/**
* Transition to follower role
*/
private async transitionToFollower(): Promise<void> {
console.log('[LeaderElection] Becoming FOLLOWER');
this.role = 'follower';
// Wait a bit for leader to write info file
await new Promise(resolve => setTimeout(resolve, 500));
// Read leader info
this.leaderInfo = await this.readLeaderInfo();
if (this.leaderInfo) {
console.log(`[LeaderElection] Leader is ${this.leaderInfo.host} at ${this.leaderInfo.api}`);
this.onBecomeFollower?.(this.leaderInfo);
} else {
console.warn('[LeaderElection] Could not read leader info, will retry...');
// Retry reading after a delay
setTimeout(async () => {
this.leaderInfo = await this.readLeaderInfo();
if (this.leaderInfo) {
console.log(`[LeaderElection] Leader is ${this.leaderInfo.host} at ${this.leaderInfo.api}`);
this.onBecomeFollower?.(this.leaderInfo);
}
}, 2000);
}
}
/**
* Start the leader election process
*/
async start(): Promise<void> {
console.log(`[LeaderElection] Starting election for ${this.config.serviceName}...`);
console.log(`[LeaderElection] Lock file: ${this.lockFilePath}`);
console.log(`[LeaderElection] Info file: ${this.infoFilePath}`);
// Try to become leader
const acquired = await this.tryAcquireLock();
if (acquired) {
await this.transitionToLeader();
} else {
await this.transitionToFollower();
}
// Start health check
this.startHealthCheck();
}
/**
* Stop and cleanup
*/
async stop(): Promise<void> {
console.log('[LeaderElection] Stopping...');
this.isShuttingDown = true;
// Stop health check
if (this.healthCheckTimer) {
clearInterval(this.healthCheckTimer);
this.healthCheckTimer = null;
}
// If we were leader, cleanup
if (this.role === 'leader') {
await this.stopRedis();
await this.releaseLock();
}
this.role = 'unknown';
console.log('[LeaderElection] Stopped');
}
// ========================================================================
// Event Registration
// ========================================================================
onLeader(callback: () => void): this {
this.onBecomeLeader = callback;
return this;
}
onFollower(callback: (leaderInfo: LeaderLockInfo) => void): this {
this.onBecomeFollower = callback;
return this;
}
onLostLeader(callback: () => void): this {
this.onLeaderLost = callback;
return this;
}
// ========================================================================
// Getters
// ========================================================================
getRole(): LeaderRole {
return this.role;
}
isLeader(): boolean {
return this.role === 'leader';
}
isFollower(): boolean {
return this.role === 'follower';
}
getLeaderInfo(): LeaderLockInfo | null {
return this.leaderInfo;
}
getRedisUrl(): string | null {
if (this.leaderInfo) {
return this.leaderInfo.api;
}
return null;
}
}
packages/meta-sort-core/src/kv/ServiceDiscovery.ts
+80 -50
View File
@@ -1,16 +1,19 @@
/** /**
* Service Discovery using shared filesystem * Service Discovery using shared filesystem
* *
* Each service registers itself in /meta-core/services/{service-name}.json * Each service registers itself in /meta-core/services/{service-name}-{hostname}.json
* with its API endpoint, status, and heartbeat timestamp. * with its base URL, status, and heartbeat timestamp.
* *
* Services can discover each other by reading these JSON files. * Services can discover each other by reading these JSON files.
* Stale detection: if lastHeartbeat > 60 seconds ago, service is considered dead. * Stale detection: if lastHeartbeat > 60 seconds ago, service is considered stale.
*
* Note: meta-core (Go sidecar) handles actual service registration.
* This class is primarily used for discovering other services.
*/ */
import { promises as fs } from 'fs'; import { promises as fs } from 'fs';
import { hostname } from 'os'; import { hostname } from 'os';
import { join } from 'path'; import { join, basename } from 'path';
import type { ServiceInfo } from './IKVClient.js'; import type { ServiceInfo } from './IKVClient.js';
import { isNodeError } from '../types/ExtendedInterfaces.js'; import { isNodeError } from '../types/ExtendedInterfaces.js';
@@ -21,18 +24,12 @@ interface ServiceDiscoveryConfig {
/** Service name */ /** Service name */
serviceName: string; serviceName: string;
/** Service version */ /** Service version (optional, not used in simplified format) */
version: string; version?: string;
/** HTTP API URL */ /** Base URL for the service */
apiUrl: string; apiUrl: string;
/** Service capabilities */
capabilities?: string[];
/** Named endpoints */
endpoints?: Record<string, string>;
/** Heartbeat interval in ms (default: 30000) */ /** Heartbeat interval in ms (default: 30000) */
heartbeatInterval?: number; heartbeatInterval?: number;
@@ -46,18 +43,19 @@ export class ServiceDiscovery {
private serviceFilePath: string; private serviceFilePath: string;
private heartbeatTimer: NodeJS.Timeout | null = null; private heartbeatTimer: NodeJS.Timeout | null = null;
private isShuttingDown = false; private isShuttingDown = false;
private currentHostname: string;
constructor(config: ServiceDiscoveryConfig) { constructor(config: ServiceDiscoveryConfig) {
this.config = { this.config = {
heartbeatInterval: 30000, heartbeatInterval: 30000,
staleThreshold: 60000, staleThreshold: 60000,
capabilities: [],
endpoints: {},
...config ...config
}; };
this.currentHostname = hostname();
this.servicesDir = `${this.config.metaCorePath}/services`; this.servicesDir = `${this.config.metaCorePath}/services`;
this.serviceFilePath = `${this.servicesDir}/${this.config.serviceName}.json`; // Use hostname-based file naming
this.serviceFilePath = `${this.servicesDir}/${this.config.serviceName}-${this.currentHostname}.json`;
} }
/** /**
@@ -73,28 +71,25 @@ export class ServiceDiscovery {
private buildServiceInfo(status: ServiceInfo['status']): ServiceInfo { private buildServiceInfo(status: ServiceInfo['status']): ServiceInfo {
return { return {
name: this.config.serviceName, name: this.config.serviceName,
version: this.config.version, hostname: this.currentHostname,
api: this.config.apiUrl, baseUrl: this.config.apiUrl,
status, status,
pid: process.pid, lastHeartbeat: new Date().toISOString()
hostname: hostname(),
startedAt: new Date().toISOString(),
lastHeartbeat: new Date().toISOString(),
capabilities: this.config.capabilities || [],
endpoints: this.config.endpoints || {}
}; };
} }
/** /**
* Register this service * Register this service
* Note: In the new architecture, meta-core handles registration.
* This is kept for backwards compatibility and testing.
*/ */
async register(): Promise<void> { async register(): Promise<void> {
await this.ensureServicesDir(); await this.ensureServicesDir();
const info = this.buildServiceInfo('starting'); const info = this.buildServiceInfo('running');
await fs.writeFile(this.serviceFilePath, JSON.stringify(info, null, 2)); await fs.writeFile(this.serviceFilePath, JSON.stringify(info, null, 2));
console.log(`[ServiceDiscovery] Registered ${this.config.serviceName}`); console.log(`[ServiceDiscovery] Registered ${this.config.serviceName}-${this.currentHostname}`);
} }
/** /**
@@ -129,7 +124,6 @@ export class ServiceDiscovery {
// If file was deleted, re-register // If file was deleted, re-register
if (isNodeError(error) && error.code === 'ENOENT') { if (isNodeError(error) && error.code === 'ENOENT') {
await this.register(); await this.register();
await this.updateStatus('running');
} else { } else {
console.error('[ServiceDiscovery] Heartbeat failed:', error); console.error('[ServiceDiscovery] Heartbeat failed:', error);
} }
@@ -164,17 +158,23 @@ export class ServiceDiscovery {
this.stopHeartbeat(); this.stopHeartbeat();
try { try {
// Update status to stopped before removing // Remove the service file
await this.updateStatus('stopped'); await fs.unlink(this.serviceFilePath);
console.log(`[ServiceDiscovery] Unregistered ${this.config.serviceName}-${this.currentHostname}`);
// Optionally remove the file (or keep for debugging)
// await fs.unlink(this.serviceFilePath);
console.log(`[ServiceDiscovery] Unregistered ${this.config.serviceName}`);
} catch (error) { } catch (error) {
if (!isNodeError(error) || error.code !== 'ENOENT') {
console.error('[ServiceDiscovery] Failed to unregister:', error); console.error('[ServiceDiscovery] Failed to unregister:', error);
} }
} }
}
/**
* Check if a service info is stale
*/
private isStale(info: ServiceInfo): boolean {
const lastHeartbeat = new Date(info.lastHeartbeat).getTime();
return Date.now() - lastHeartbeat > this.config.staleThreshold!;
}
// ======================================================================== // ========================================================================
// Discovery Methods // Discovery Methods
@@ -182,24 +182,48 @@ export class ServiceDiscovery {
/** /**
* Discover a service by name * Discover a service by name
* Looks for files matching pattern: {name}-*.json (hostname-based naming)
*/ */
async discoverService(name: string): Promise<ServiceInfo | null> { async discoverService(name: string): Promise<ServiceInfo | null> {
const servicePath = join(this.servicesDir, `${name}.json`);
try { try {
const content = await fs.readFile(servicePath, 'utf-8'); // First try exact match for backward compatibility
const exactPath = join(this.servicesDir, `${name}.json`);
try {
const content = await fs.readFile(exactPath, 'utf-8');
const service = JSON.parse(content) as ServiceInfo; const service = JSON.parse(content) as ServiceInfo;
// Check if service is alive (heartbeat within threshold) if (this.isStale(service)) {
const lastHeartbeat = new Date(service.lastHeartbeat).getTime(); service.status = 'stale';
const isAlive = Date.now() - lastHeartbeat < this.config.staleThreshold!;
if (!isAlive) {
console.warn(`[ServiceDiscovery] Service ${name} appears stale (last heartbeat: ${service.lastHeartbeat})`);
return null;
} }
return service.status === 'running' ? service : null;
} catch {
// Try hostname-based files
}
// Search for hostname-based files: name-*.json
const files = await fs.readdir(this.servicesDir);
const matchingFiles = files.filter(f =>
f.startsWith(`${name}-`) && f.endsWith('.json')
);
// Return first valid (non-stale) service
for (const file of matchingFiles) {
const filePath = join(this.servicesDir, file);
const content = await fs.readFile(filePath, 'utf-8');
const service = JSON.parse(content) as ServiceInfo;
if (this.isStale(service)) {
service.status = 'stale';
continue; // Skip stale services
}
if (service.status === 'running') {
return service; return service;
}
}
return null;
} catch (error) { } catch (error) {
if (isNodeError(error) && error.code === 'ENOENT') { if (isNodeError(error) && error.code === 'ENOENT') {
return null; // Service not registered return null; // Service not registered
@@ -219,11 +243,18 @@ export class ServiceDiscovery {
for (const file of files) { for (const file of files) {
if (!file.endsWith('.json')) continue; if (!file.endsWith('.json')) continue;
const name = file.replace('.json', ''); const filePath = join(this.servicesDir, file);
const service = await this.discoverService(name); try {
const content = await fs.readFile(filePath, 'utf-8');
const service = JSON.parse(content) as ServiceInfo;
if (this.isStale(service)) {
service.status = 'stale';
}
if (service && service.status === 'running') {
services.push(service); services.push(service);
} catch (error) {
console.error(`[ServiceDiscovery] Failed to read ${file}:`, error);
} }
} }
@@ -266,8 +297,7 @@ export class ServiceDiscovery {
if (!service) return false; if (!service) return false;
// Try to ping the service's health endpoint // Try to ping the service's health endpoint
const healthEndpoint = service.endpoints.health || '/health'; const url = `${service.baseUrl}/health`;
const url = `${service.api}${healthEndpoint}`;
try { try {
const response = await fetch(url, { const response = await fetch(url, {
@@ -285,10 +315,10 @@ export class ServiceDiscovery {
/** /**
* Full startup sequence * Full startup sequence
* Note: In the new architecture, meta-core handles registration.
*/ */
async start(): Promise<void> { async start(): Promise<void> {
await this.register(); await this.register();
await this.updateStatus('running');
this.startHeartbeat(); this.startHeartbeat();
} }
packages/meta-sort-core/src/kv/index.ts
+3 -3
View File
@@ -1,10 +1,10 @@
/** /**
* KV Module - Unified key-value storage with leader election * KV Module - Unified key-value storage with leader client
* *
* This module provides: * This module provides:
* - IKVClient interface for KV storage abstraction * - IKVClient interface for KV storage abstraction
* - RedisKVClient for Redis-based storage * - RedisKVClient for Redis-based storage
* - LeaderElection for distributed leader election via flock * - LeaderClient for reading leader info from meta-core
* - ServiceDiscovery for service registration and discovery * - ServiceDiscovery for service registration and discovery
* - KVManager for unified management of all components * - KVManager for unified management of all components
* - MetadataUtils for flattening/reconstructing nested metadata * - MetadataUtils for flattening/reconstructing nested metadata
@@ -12,7 +12,7 @@
export * from './IKVClient.js'; export * from './IKVClient.js';
export * from './RedisClient.js'; export * from './RedisClient.js';
export * from './LeaderElection.js'; export * from './LeaderClient.js';
export * from './ServiceDiscovery.js'; export * from './ServiceDiscovery.js';
export * from './KVManager.js'; export * from './KVManager.js';
export * from './MetadataUtils.js'; export * from './MetadataUtils.js';