From d6c8852e1e33b55e3c4df9d2f5e5dff664659799 Mon Sep 17 00:00:00 2001 From: windyboy Date: Fri, 13 Feb 2026 09:50:29 +0800 Subject: [PATCH] Fix quarantine non-atomic operations with two-phase commit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements atomic quarantine/restore operations using two-phase commit pattern to prevent orphaned files when manifest updates fail. Changes to models.py: - Add status field to QuarantineEntry ("pending" | "committed") - Default to "committed" for backward compatibility Changes to quarantine.py: - Rewrite quarantine_file() with three phases: 1. Write pending manifest entry BEFORE moving file 2. Move file to quarantine 3. Mark manifest entry as committed - Rewrite restore_from_quarantine() with same pattern - Add _recover_pending_entries() for auto-recovery on manifest load - Update _load_manifest() and _save_manifest() to handle status field Changes to executor.py: - Add special handling for quarantine rollback using QuarantineManager - Fix bug where executor didn't preserve quarantine destination_path - Return QuarantineManager result directly (includes actual quarantine path) Testing: - Fixed pre-existing test_rollback_quarantine_operation - All 439 tests now pass (was 438 with 1 failure) Atomicity guarantees: - If manifest write fails → operation fails, no file moved - If file move fails → rollback removes pending manifest entry - If commit fails → auto-recovery fixes on next load - No orphaned files possible Co-Authored-By: Claude Sonnet 4.5 --- ARCHITECTURE_REVIEW.md | 1528 ++++++++++++++++++++++++++++++++++++++++ src/vlm/executor.py | 47 +- src/vlm/models.py | 4 +- src/vlm/quarantine.py | 370 +++++++--- 4 files changed, 1848 insertions(+), 101 deletions(-) create mode 100644 ARCHITECTURE_REVIEW.md diff --git a/ARCHITECTURE_REVIEW.md b/ARCHITECTURE_REVIEW.md new file mode 100644 index 0000000..5126500 --- /dev/null +++ b/ARCHITECTURE_REVIEW.md @@ -0,0 +1,1528 @@ +# Video Library Manager - Architecture Review + +**Review Date**: 2026-02-13 +**Reviewer**: Claude (Sonnet 4.5) +**Project Version**: Commit 1f55eab +**Review Scope**: Full architecture, workflow design, implementation quality + +--- + +## Executive Summary + +Video Library Manager (VLM) is a well-architected Python CLI tool with a safety-first design philosophy. The multi-stage pipeline architecture (scan → parse → enrich → analyze → plan → execute) effectively separates concerns and provides human-in-the-loop checkpoints before file modifications. + +**Overall Assessment**: ⭐⭐⭐⭐ (4/5) + +### Key Strengths +- ✅ Clear separation of concerns with single-responsibility modules +- ✅ Safety-first design with dry-run, rollback, and quarantine mechanisms +- ✅ Graceful degradation when optional dependencies (ffprobe) are unavailable +- ✅ Incremental caching for expensive operations (metadata, enrichment) + +### Critical Issues Identified +- ❌ **Data loss during workflow**: Metadata (resolution/codec) lost in analyze stage +- ❌ **No unified I/O layer**: Duplicated CSV/JSON loading, no schema versioning +- ❌ **Path validation missing**: Absolute paths can become stale between stages +- ❌ **Non-atomic operations**: Multi-step operations can fail partially + +### Priority Recommendations +1. **Immediate**: Fix identity reconstruction to preserve video metadata +2. **Short-term**: Consolidate I/O operations, add schema versioning +3. **Long-term**: Implement explicit state machine for workflow progression + +--- + +## 1. Workflow Design Analysis + +### 1.1 Current Architecture + +``` +┌─────────┐ ┌───────┐ ┌─────────┐ ┌─────────┐ ┌──────┐ ┌─────────┐ +│ Scan │────▶│ Parse │────▶│ Enrich │────▶│ Analyze │────▶│ Plan │────▶│ Execute │ +└─────────┘ └───────┘ └─────────┘ └─────────┘ └──────┘ └─────────┘ + │ │ │ │ │ │ +inventory.csv identities.json (in-place) analysis.json plan.json rollback.json +``` + +**Design Pattern**: Linear pipeline with explicit file I/O between stages + +### 1.2 Strengths + +#### ✅ Clear Stage Separation +Each stage has a well-defined input/output contract: +- **Scan**: library_root → inventory.csv (VideoFile list) +- **Parse**: inventory.csv → identities.json (Identity objects) +- **Enrich**: identities.json → identities.json (in-place update) +- **Analyze**: identities.json → analysis.json (gaps + duplicates) +- **Plan**: identities.json + analysis.json → plan.json (operations) +- **Execute**: plan.json → file system changes + rollback.json + +This separation allows: +- Running stages independently +- Manual review of intermediate artifacts +- Debugging at each stage boundary + +#### ✅ Read-First Principle +Most stages are read-only until final execution: +- Scan: read-only file system traversal +- Parse: read CSV, write JSON (no file moves) +- Enrich: update JSON in-place (no file moves) +- Analyze: read-only analysis +- Plan: read-only planning + +**Only Execute stage modifies the file system**, minimizing risk. + +#### ✅ Human-in-the-Loop Checkpoints +Three review points before file modifications: +1. After **analyze**: Review gaps and duplicates in `analysis.json` +2. After **plan**: Review proposed operations in `plan.json` +3. Before **execute**: Dry-run preview with `vlm execute` (no --confirm) + +### 1.3 Weaknesses + +#### ❌ No Transaction Semantics + +**Issue**: Each stage is independent with no rollback for failed intermediate stages. + +**Example Scenario**: +```bash +vlm enrich # Crashes after enriching 5,000 of 10,000 records +# Result: identities.json partially updated, no way to resume +# User must re-run --refresh-all (re-processes all 10,000) +``` + +**Impact**: +- Wasted API calls for already-enriched records +- Potential data inconsistency if crash during file write +- No incremental resume capability + +**Recommendation**: +- Add checkpoint files (e.g., `.enrich_checkpoint`) with last processed index +- Detect checkpoint on startup, resume from last position +- Or: Use atomic file operations (write to temp, rename) + +#### ❌ Implicit State Progression + +**Issue**: Workflow progression is tracked only by file existence, not explicit state. + +**Current State Tracking**: +```python +# state.py exists but is NOT integrated into main workflow +# Main workflow relies on: +if os.path.exists("inventory.csv"): # Scan completed +if os.path.exists("identities.json"): # Parse completed +if os.path.exists("analysis.json"): # Analyze completed +``` + +**Consequences**: +- Can't detect if file was manually edited (invalidating workflow state) +- No way to know if scan completed successfully or crashed mid-way +- State management (`vlm state` commands) is disconnected from main pipeline + +**Recommendation**: +- Add `.vlm/workflow_state.json` tracking stage completion and file hashes +- Validate file integrity before each stage (check hash against expected) +- Integrate StateManager into main workflow, auto-mark files as "processed" + +#### ❌ Lossy Data Reconstruction + +**Critical Issue**: Video metadata lost during analyze stage. + +**Root Cause**: +```python +# analysis.py lines 70-123: detect_duplicates() +# Loads identities.json and reconstructs VideoFile objects +# But io.py reconstruction loses resolution/codec/bitrate + +# io.py:48-60 _create_video_file_from_identity() +vf = VideoFile( + path=Path(record["filename"]), # Only basic fields + filename=record["filename"], + size_bytes=0, # ⚠️ Defaulted to 0! + modified_timestamp=datetime.now(timezone.utc), + category=record.get("category", "other"), + resolution=None, # ⚠️ Lost! + codec=None, # ⚠️ Lost! + duration_seconds=None, + bitrate_kbps=None, +) +``` + +**Impact on Duplicate Resolution**: +- `by_quality` strategy fails because resolution/codec are None +- Quality comparison (analysis.py:115) returns incomplete metadata +- Users relying on `duplicate_keep: by_quality` get incorrect results + +**Evidence**: +```python +# duplicate_resolve.py:67-85 _quality_score() +resolution_rank = { + "4K": 4, "2160p": 4, + "1080p": 3, + "720p": 2, + "480p": 1, +}.get(vf.resolution, 0) # Always returns 0 if resolution=None! +``` + +**Recommendation**: +- **Option 1**: Store full VideoFile in identities.json (not just identity) + ```json + { + "movies": [ + { + "identity": {...}, + "video_file": { + "resolution": "1080p", + "codec": "x265", + ... + } + } + ] + } + ``` +- **Option 2**: Load metadata from inventory.csv during analyze stage + ```python + # Load both inventory.csv and identities.json + # Join on path to get full metadata + ``` + +--- + +## 2. Data Flow Analysis + +### 2.1 Current Data Paths + +``` +Stage 1: Scan + VideoFile(path, filename, size_bytes, modified_timestamp, category, + resolution, codec, duration_seconds, bitrate_kbps) + ↓ serialize to CSV + inventory.csv (9 columns) + +Stage 2: Parse + Load inventory.csv → List[VideoFile] + ↓ parse filenames → extract identity + MovieIdentity(title, year, confidence, needs_review) + SeriesIdentity(title, season, episodes, confidence, needs_review) + ↓ serialize to JSON + identities.json + { + "movies": [ + {"filename": "...", "title": "Inception", "year": 2010, ...} + ], + "series": [...] + } + +Stage 3: Enrich (in-place update) + Load identities.json + ↓ call TMDB API + Add: title_zh, title_en, reputation_score, canonical_id, ... + ↓ serialize back to JSON + identities.json (updated in-place) + +Stage 4: Analyze + Load identities.json + ↓ reconstruct VideoFile (⚠️ loses resolution/codec) + Detect gaps and duplicates + ↓ serialize + analysis.json + { + "completeness": [...], + "duplicates": [ + { + "files": [ + {"filename": "...", "size_bytes": 0, "resolution": null} + ] + } + ] + } + +Stage 5: Plan + Load identities.json + analysis.json + ↓ generate FileOperation list + ExecutionPlan(operations, summary, human_summary) + ↓ serialize + plan.json + +Stage 6: Execute + Load plan.json + ↓ execute file operations + File system changes + rollback.json +``` + +### 2.2 Data Loss Points + +| Stage Transition | Data Preserved | Data Lost | Impact | +|------------------|----------------|-----------|--------| +| Scan → inventory.csv | All VideoFile fields | None | ✅ Good | +| inventory.csv → Parse | VideoFile fully loaded | None | ✅ Good | +| Parse → identities.json | Identity fields | VideoFile metadata (resolution/codec) | ⚠️ **Minor** - still in inventory.csv | +| identities.json → Analyze | Identity fields | VideoFile metadata | ❌ **Critical** - duplicate quality comparison broken | +| Analyze → analysis.json | Duplicate groups, gaps | Enrichment fields (title_zh) | ⚠️ Minor - not needed for plan | +| analysis.json → Plan | Operations, conflicts | Original identity context | ✅ OK - plan references paths | + +### 2.3 Recommendations + +#### Fix 1: Preserve Metadata in Identities +Add `video_file` object to identities.json: +```json +{ + "movies": [ + { + "filename": "/library/movie/Inception (2010).mkv", + "title": "Inception", + "year": 2010, + "confidence": 0.95, + "video_file": { + "resolution": "1080p", + "codec": "x265", + "size_bytes": 2147483648, + "duration_seconds": 8880 + } + } + ] +} +``` + +**Migration Path**: +1. Update `io.save_identities_json()` to include `video_file` field +2. Update `io.load_identities_json()` to reconstruct VideoFile from saved data +3. Add schema version: `{"version": "1.1", "movies": [...]}` for compatibility check + +#### Fix 2: Use Inventory as Source of Truth +Keep inventory.csv as canonical metadata store, join with identities: +```python +# In analyze stage +inventory = io.load_inventory_csv("inventory.csv") +identities = io.load_identities_json("identities.json") + +# Build lookup: path → VideoFile +vf_by_path = {vf.path: vf for vf in inventory} + +# Join: identity + full VideoFile +for identity in identities: + vf = vf_by_path.get(Path(identity["filename"])) + if vf: + # Use vf.resolution, vf.codec for quality comparison +``` + +**Trade-off**: Requires inventory.csv to be present during analyze stage (dependency) + +--- + +## 3. Module Coupling Analysis + +### 3.1 Coupling Matrix + +| Module | High Coupling | Medium Coupling | Low Coupling | +|--------|---------------|-----------------|--------------| +| **cli.py** | commands/*, config | io, models | utils | +| **scanner.py** | config, models | io | utils | +| **parser.py** | models | - | utils | +| **enrichment.py** | cache, providers, models | config | utils | +| **analysis.py** | models, parser | io | - | +| **planner.py** | duplicate_resolve, models, config | io | utils | +| **executor.py** | quarantine, models, config | io | - | +| **io.py** | scanner, models | - | - | + +### 3.2 Problematic Coupling + +#### Issue 1: CLI Bypasses I/O Module + +**Problem**: CLI commands duplicate CSV/JSON loading logic instead of using `io.py`. + +**Evidence**: +```python +# cli.py parse command (lines 185-336) +with open(input_path) as f: + reader = csv.DictReader(f) + files = [] + for row in reader: + # Manual CSV parsing instead of io.load_inventory_csv() +``` + +**Impact**: +- Changes to CSV schema require updates in multiple places +- Inconsistent error handling between cli.py and io.py +- Maintenance burden + +**Fix**: +```python +# cli.py parse command +from vlm.io import load_inventory_csv, save_identities_json + +files = load_inventory_csv(input_path) +# ... parse identities ... +save_identities_json(identities, output_path) +``` + +#### Issue 2: Circular Import Risk + +**Potential Issue**: `io.py` imports `scanner.py` for type hints, `cli.py` imports both. + +```python +# io.py:4 +from vlm.scanner import scan_library # Direct import + +# cli.py:10 +from vlm import scanner +from vlm import io +``` + +**Current Status**: No circular import (yet), but fragile. + +**Recommendation**: +- Use `TYPE_CHECKING` for type-only imports: + ```python + from typing import TYPE_CHECKING + if TYPE_CHECKING: + from vlm.scanner import VideoFile + ``` + +### 3.3 Missing Abstractions + +#### Missing: Unified File I/O Interface + +**Current State**: Each module handles file I/O differently. + +**Examples**: +- `scanner.py`: Uses `csv.writer()` directly +- `enrichment.py`: Uses `json.dump()` directly +- `cli.py`: Uses `csv.DictReader()` directly +- `io.py`: Provides helpers but not used consistently + +**Recommendation**: Create `FileRepository` abstraction: +```python +class FileRepository: + def load_inventory(self, path: Path) -> List[VideoFile]: ... + def save_inventory(self, files: List[VideoFile], path: Path): ... + def load_identities(self, path: Path) -> dict: ... + def save_identities(self, identities: dict, path: Path): ... + def load_analysis(self, path: Path) -> Analysis: ... + def save_analysis(self, analysis: Analysis, path: Path): ... + def load_plan(self, path: Path) -> ExecutionPlan: ... + def save_plan(self, plan: ExecutionPlan, path: Path): ... +``` + +**Benefits**: +- Single source of truth for file formats +- Easy to add schema versioning +- Testable in isolation (mock file system) +- Consistent error handling + +--- + +## 4. Error Handling Review + +### 4.1 Current Error Strategy + +#### Positive Patterns + +✅ **Graceful Degradation in Scanner**: +```python +# scanner.py:401-487 +try: + metadata = _extract_metadata_ffprobe(file_path, logger) +except Exception: + logger.debug(f"ffprobe failed for {file_path}, skipping metadata") + # Continue with None values +``` +**Assessment**: Good - missing ffprobe doesn't crash, just degrades functionality. + +✅ **Explicit Error States**: +```python +# models.py FileOperation +has_conflict: bool = False # Explicit flag for destination conflicts +needs_review: bool = False # Explicit flag for parsing uncertainty +``` +**Assessment**: Good - errors are data, not exceptions. + +✅ **Nested Exception Handling**: +```python +# cli.py execute command +try: + plan = load_plan_json(plan_path) + try: + results = executor.execute_plan(plan, mode="execute", confirmed=confirmed) + except ExecutionError as e: + logger.error(f"Execution failed: {e}") +except PlanLoadError as e: + logger.error(f"Plan load failed: {e}") +``` +**Assessment**: Good - staged error handling with context. + +#### Negative Patterns + +❌ **Generic Exception Catching**: +```python +# enrichment.py:97-120 +try: + payload, api_calls, failures, skip_reason = _enrich_record(...) +except Exception as e: + logger.error(f"Enrichment failed: {e}") + stats["failed"] += 1 + # ⚠️ Can't distinguish network errors from data errors +``` +**Problem**: All errors treated equally, no retry logic for transient failures. + +❌ **Silent Data Reconstruction Failures**: +```python +# io.py:63-82 _movie_identity_from_record() +identity = MovieIdentity( + title=record["title"], # ⚠️ KeyError if missing, crashes silently + year=record["year"], + ... +) +``` +**Problem**: No validation that required fields exist, crashes with unhelpful KeyError. + +❌ **No Exit Code Semantics**: +```python +# cli.py main() +except Exception as e: + logger.error(f"Error: {e}") + sys.exit(1) # ⚠️ All errors exit with same code +``` +**Problem**: Calling scripts can't distinguish error types. + +### 4.2 Recommended Error Hierarchy + +```python +class VLMError(Exception): + """Base exception for all VLM errors.""" + exit_code = 1 + +class VLMConfigError(VLMError): + """Configuration errors (missing config, invalid YAML).""" + exit_code = 2 + +class VLMDataError(VLMError): + """Data format errors (malformed JSON, missing fields).""" + exit_code = 3 + +class VLMSystemError(VLMError): + """System errors (permission denied, disk full).""" + exit_code = 4 + +class VLMAPIError(VLMError): + """External API errors (TMDB timeout, auth failure).""" + exit_code = 5 + is_transient: bool = False # Retry hint +``` + +**Usage Example**: +```python +try: + config = load_config() +except FileNotFoundError: + raise VLMConfigError("Config file not found at ~/.vlm/config.yaml") + +try: + identities = load_identities_json(path) +except json.JSONDecodeError as e: + raise VLMDataError(f"Invalid JSON in {path}: {e}") + +try: + result = tmdb_provider.enrich(identity) +except requests.Timeout: + raise VLMAPIError("TMDB API timeout", is_transient=True) +``` + +**Benefits**: +- Calling code can handle errors by type +- Exit codes inform shell scripts of error category +- `is_transient` flag enables retry logic + +--- + +## 5. State Management Review + +### 5.1 Current State Architecture + +```python +# state.py +class FileState: + file_path: str + status: str # "reviewed", "ignored", "planned", "executed", "quarantined" + reason: Optional[str] + updated_at: str # ISO 8601 timestamp + +class StateManager: + def __init__(self, state_file: Path): + self.state_store = StateStore() # In-memory dict + self.state_file = state_file + self.load() # Load entire file into memory + + def set_file_state(self, file_path: Path, status: str, reason: str): + # Update in-memory dict + self.state_store.set_file_state(...) + self.save() # Write entire dict to file +``` + +### 5.2 Critical Issues + +#### Issue 1: State Not Integrated with Main Workflow + +**Problem**: State tracking is optional and disconnected from scan/parse/execute stages. + +**Evidence**: +- `vlm execute --confirm` does NOT auto-mark files as "executed" +- Users must manually run `vlm state set --status executed` +- No validation that files in execution plan are in correct state + +**Expected Workflow**: +```bash +vlm scan # Should mark files "discovered" +vlm parse # Should mark files "parsed" +vlm enrich # Should mark files "enriched" +vlm execute # Should mark files "executed" +``` + +**Actual Workflow**: +```bash +vlm scan # No state change +vlm parse # No state change +vlm execute # No state change +# State only changes via explicit vlm state set commands +``` + +**Recommendation**: +```python +# executor.py execute_plan() +for operation in plan.operations: + result = self.execute_operation(operation, mode=mode) + if result.success and mode == "execute": + # Auto-update state + state_manager.set_file_state( + operation.source_path, + status="executed", + reason=operation.operation_type, + ) +``` + +#### Issue 2: Non-Atomic State Persistence + +**Problem**: StateManager.save() writes entire file without atomic guarantees. + +**Current Implementation**: +```python +# state.py:166-174 +def save(self): + with open(self.state_file, "w") as f: + json.dump(self.state_store.to_dict(), f, indent=2) + # ⚠️ If interrupted mid-write, file is corrupted +``` + +**Fix with Atomic Write**: +```python +import tempfile +import os + +def save(self): + # Write to temp file first + tmp_fd, tmp_path = tempfile.mkstemp(dir=self.state_file.parent) + try: + with os.fdopen(tmp_fd, "w") as f: + json.dump(self.state_store.to_dict(), f, indent=2) + # Atomic rename (POSIX guarantee) + os.replace(tmp_path, self.state_file) + except: + os.unlink(tmp_path) + raise +``` + +#### Issue 3: Path String vs Path Object Mismatch + +**Problem**: State keys are strings, but API uses Path objects, leading to symlink issues. + +**Example**: +```python +# File accessed via symlink +vlm state set /library/symlink/movie.mkv --status reviewed + +# File accessed via real path +vlm state show /library/real/movie.mkv +# Returns: No state found (different string keys!) +``` + +**Fix**: +```python +# state.py StateStore.set_file_state() +def set_file_state(self, file_path: Path, status: str, reason: str): + # Canonicalize path before using as key + canonical_path = file_path.resolve() + key = str(canonical_path) + self.states[key] = FileState(...) +``` + +### 5.3 Recommended State Machine + +``` +┌──────────────┐ +│ Discovered │ (after scan) +└──────┬───────┘ + │ + ▼ +┌──────────────┐ +│ Parsed │ (after parse, identity extracted) +└──────┬───────┘ + │ + ▼ +┌──────────────┐ +│ Enriched │ (after enrich, TMDB data added) +└──────┬───────┘ + │ + ▼ +┌──────────────┐ +│ Analyzed │ (after analyze, included in gap/dup detection) +└──────┬───────┘ + │ + ├─────────────────┬─────────────────┐ + ▼ ▼ ▼ +┌──────────────┐ ┌─────────────┐ ┌──────────────┐ +│ Planned │ │ Ignored │ │ Reviewed │ +│ (in plan.json)│ │ (user skip) │ │ (user manual)│ +└──────┬───────┘ └─────────────┘ └──────────────┘ + │ + ▼ +┌──────────────┐ +│ Executed │ (file moved/renamed) +└──────┬───────┘ + │ + ├─────────────────┐ + ▼ ▼ +┌──────────────┐ ┌─────────────┐ +│ Quarantined │ │ Rolled Back │ +└──────────────┘ └─────────────┘ +``` + +**Implementation**: +```python +class WorkflowState(Enum): + DISCOVERED = "discovered" + PARSED = "parsed" + ENRICHED = "enriched" + ANALYZED = "analyzed" + PLANNED = "planned" + IGNORED = "ignored" + REVIEWED = "reviewed" + EXECUTED = "executed" + QUARANTINED = "quarantined" + ROLLED_BACK = "rolled_back" + +VALID_TRANSITIONS = { + WorkflowState.DISCOVERED: [WorkflowState.PARSED], + WorkflowState.PARSED: [WorkflowState.ENRICHED, WorkflowState.IGNORED], + WorkflowState.ENRICHED: [WorkflowState.ANALYZED, WorkflowState.IGNORED], + # ... +} + +def validate_transition(from_state: WorkflowState, to_state: WorkflowState): + if to_state not in VALID_TRANSITIONS.get(from_state, []): + raise VLMStateError(f"Invalid transition: {from_state} → {to_state}") +``` + +--- + +## 6. Caching Strategy Review + +### 6.1 Scanner Metadata Cache + +**Implementation** (scanner.py:219-255): +```python +if cached_file is not None and \ + cached_file.size_bytes == size_bytes and \ + int(_normalize_to_utc(cached_file.modified_timestamp).timestamp()) == file_mtime_seconds: + # Reuse cached metadata (resolution, codec, duration, bitrate) + return cached_file +``` + +**Cache Key**: `(path, size_bytes, modified_timestamp)` + +**Strengths**: +- ✅ Simple and effective for unchanged files +- ✅ Avoids expensive ffprobe calls +- ✅ Works across scan invocations + +**Weaknesses**: +- ❌ **No invalidation**: If file content changes but size/mtime stay same (rare), stale cache +- ❌ **In-memory only**: Limited by RAM for large libraries +- ❌ **Path-based key**: Symlinks and relative paths won't cache-hit + +**Recommendations**: +1. Add content hash to cache key (optional, expensive): + ```python + # Only for small files or on explicit --verify flag + if size_bytes < 10MB or verify_cache: + content_hash = hashlib.md5(file.read_bytes()).hexdigest() + cache_key = (path, size_bytes, mtime, content_hash) + ``` + +2. Persist cache to disk (SQLite): + ```python + # ~/.vlm/metadata_cache.db + CREATE TABLE metadata_cache ( + path TEXT PRIMARY KEY, + size_bytes INTEGER, + mtime_seconds INTEGER, + resolution TEXT, + codec TEXT, + duration_seconds INTEGER, + bitrate_kbps INTEGER + ) + ``` + +### 6.2 Enrichment Cache + +**Implementation** (cache.py): +```python +CREATE TABLE identity_enrichment ( + identity_key TEXT PRIMARY KEY, + fingerprint TEXT NOT NULL, + payload_json TEXT NOT NULL, + cached_at TEXT NOT NULL +) +``` + +**Cache Key**: `identity_key` (e.g., "movie:Inception:2010") +**Invalidation**: `fingerprint` (hash of title/year/season/episodes) + +**Strengths**: +- ✅ Persistent SQLite storage +- ✅ Fingerprint-based invalidation detects identity changes +- ✅ Survives across runs + +**Weaknesses**: + +#### Issue 1: Fingerprint Collision Risk + +**Problem**: Hash collision not handled. + +**Example**: +```python +# enrichment.py:216-227 _fingerprint() +def _fingerprint(record: dict, media_type: str) -> str: + payload = json.dumps({ + "media_type": media_type, + "title": record.get("title"), + "year": record.get("year"), + # ... + }, sort_keys=True) + return hashlib.sha256(payload.encode()).hexdigest() +``` + +**Collision Scenario** (theoretical): +- Movie "Title A (2020)" hashes to `abc123...` +- Movie "Title B (2020)" also hashes to `abc123...` (collision) +- Both share same cache entry (wrong data returned) + +**Likelihood**: Extremely low (SHA-256 has 2^256 space), but not zero. + +**Fix**: Use composite key instead of hash: +```sql +CREATE TABLE identity_enrichment ( + media_type TEXT NOT NULL, + title TEXT NOT NULL, + year INTEGER, + season INTEGER, + fingerprint TEXT NOT NULL, + payload_json TEXT NOT NULL, + PRIMARY KEY (media_type, title, year, season) +) +``` + +#### Issue 2: No Cache TTL + +**Problem**: Cached enrichment never expires, even if TMDB data updates. + +**Example**: +- Cache stores reputation_score=7.5 from TMDB on 2025-01-01 +- TMDB rating updates to 8.2 on 2025-06-01 +- VLM continues using stale 7.5 score indefinitely + +**Recommendation**: +```python +# Add cache expiry +CACHE_TTL_DAYS = 90 # Configurable + +CREATE TABLE identity_enrichment ( + ... + cached_at TEXT NOT NULL, + expires_at TEXT NOT NULL -- cached_at + TTL +) + +# Check expiry on load +def get_identity(self, identity_key, fingerprint): + row = self.conn.execute( + "SELECT payload_json, expires_at FROM identity_enrichment WHERE ...", + (identity_key, fingerprint) + ).fetchone() + if row: + if datetime.fromisoformat(row[1]) > datetime.now(timezone.utc): + return json.loads(row[0]) # Cache hit + return None # Expired or not found +``` + +#### Issue 3: Payload Validation Missing + +**Problem**: No schema validation when loading cached payload. + +**Risk**: +```python +# cache.py:62 +payload = json.loads(row[0]) +# ⚠️ If payload_json is corrupted or missing required fields, +# downstream code crashes with KeyError +``` + +**Fix**: Validate payload structure: +```python +from vlm.models import EnrichedIdentity # Hypothetical + +def get_identity(self, identity_key, fingerprint): + row = self.conn.execute(...).fetchone() + if row: + payload = json.loads(row[0]) + try: + # Validate required fields + EnrichedIdentity.from_dict(payload) + return payload + except (KeyError, ValueError) as e: + logger.warning(f"Invalid cached payload for {identity_key}: {e}") + # Invalidate corrupted cache entry + self.invalidate(identity_key) + return None +``` + +--- + +## 7. Safety Mechanisms Review + +### 7.1 Dry-Run Implementation + +**Design** (executor.py:45-85): +```python +def execute_plan(self, plan, mode="dry-run", confirmed=False): + if mode == "execute" and not confirmed: + raise ValueError("Execute mode requires explicit confirmation") + + for op in plan.operations: + result = self.execute_operation(op, mode=mode) + # Same code path for dry-run and execute + # Mode flag controls whether file ops actually happen +``` + +**Strength**: +- ✅ Dry-run uses same logic as execute (what you see is what you get) +- ✅ Explicit confirmation required via `--confirm` flag + +**Weakness**: +- ❌ **No separation of dry-run vs execute logic** + - Both modes call same `execute_operation()` method + - If bug exists in execute branch, dry-run can't detect it + - Dry-run validation is weak (only checks file existence) + +**Recommendation**: Separate validation logic +```python +def validate_operation(self, op: FileOperation) -> ValidationResult: + """Dry-run validation (no file system changes).""" + errors = [] + + # Check source exists + if not op.source_path.exists(): + errors.append(f"Source does not exist: {op.source_path}") + + # Check destination not occupied + if op.destination_path and op.destination_path.exists(): + errors.append(f"Destination conflict: {op.destination_path}") + + # Check parent directory writable + if not os.access(op.destination_path.parent, os.W_OK): + errors.append(f"Destination not writable: {op.destination_path.parent}") + + return ValidationResult(valid=len(errors) == 0, errors=errors) + +def execute_plan(self, plan, mode="dry-run"): + if mode == "dry-run": + return [self.validate_operation(op) for op in plan.operations] + else: + return [self.execute_operation(op) for op in plan.operations] +``` + +### 7.2 Quarantine Safety + +**Implementation** (quarantine.py:45-90): +```python +def quarantine_file(self, file_path: Path, reason: str) -> OperationResult: + # 1. Determine quarantine path + quarantine_path = self._get_quarantine_path(file_path) + + # 2. Create quarantine directory + quarantine_path.parent.mkdir(parents=True, exist_ok=True) + + # 3. Move file + shutil.move(str(file_path), str(quarantine_path)) + + # 4. Update manifest + self.manifest.add_entry(QuarantineEntry(...)) + self._save_manifest() +``` + +**Critical Issue**: **Non-Atomic Operation** + +**Failure Scenario**: +1. File moved to quarantine (step 3 succeeds) +2. Manifest update fails (step 4 fails: disk full, permission denied) +3. **Result**: File is quarantined but not tracked in manifest +4. User runs `vlm quarantine list` → file not shown +5. File is effectively lost (orphaned in quarantine directory) + +**Fix**: Atomic transaction pattern +```python +def quarantine_file(self, file_path: Path, reason: str) -> OperationResult: + quarantine_path = self._get_quarantine_path(file_path) + entry = QuarantineEntry( + original_path=str(file_path), + quarantine_path=str(quarantine_path), + reason=reason, + timestamp=utcnow(), + ) + + # Step 1: Add to manifest FIRST (before moving file) + self.manifest.add_entry(entry) + self._save_manifest() + + try: + # Step 2: Move file + quarantine_path.parent.mkdir(parents=True, exist_ok=True) + shutil.move(str(file_path), str(quarantine_path)) + except Exception as e: + # Rollback: Remove from manifest if file move failed + self.manifest.remove_entry(entry) + self._save_manifest() + raise QuarantineError(f"File move failed: {e}") + + return OperationResult(success=True, ...) +``` + +**Alternative**: Use write-ahead log (WAL) +```python +# Write intent to WAL before action +wal.append({"action": "quarantine", "file": file_path, "dest": quarantine_path}) +# Perform action +shutil.move(...) +# Commit WAL entry +wal.commit() +``` + +### 7.3 Rollback Implementation + +**Design** (executor.py:212-258): +```python +def rollback(self, rollback_log_path: Path) -> List[OperationResult]: + log = self._load_rollback_log(rollback_log_path) + results = [] + + # LIFO: Reverse order of operations + for op in reversed(log.operations): + result = self._rollback_operation(op) + results.append(result) + + return results +``` + +**Strengths**: +- ✅ LIFO order (undo in reverse) +- ✅ Best-effort semantics (continues on failure) +- ✅ Comprehensive logging + +**Weaknesses**: + +#### Issue 1: No Validation Before Rollback + +**Problem**: No check that files are still at destination before rolling back. + +**Failure Scenario**: +```bash +vlm execute --confirm # Moves A.mkv → B.mkv +# User manually moves B.mkv → C.mkv +vlm rollback # Tries to move B.mkv → A.mkv +# FAILS: B.mkv no longer exists +``` + +**Fix**: Validate file state before rollback +```python +def _rollback_operation(self, op: FileOperation) -> OperationResult: + # Validate current state matches expected state + if op.operation_type == "move": + if not op.destination_path.exists(): + return OperationResult( + success=False, + error=f"Cannot rollback: file not at destination {op.destination_path}" + ) + if op.source_path.exists(): + return OperationResult( + success=False, + error=f"Cannot rollback: source path occupied {op.source_path}" + ) + + # Perform rollback + shutil.move(str(op.destination_path), str(op.source_path)) +``` + +#### Issue 2: No Idempotency + +**Problem**: Can't safely re-run rollback if it failed partway. + +**Scenario**: +```bash +vlm rollback # Fails after rolling back 50 of 100 operations +vlm rollback # Re-run: tries to rollback already-rolled-back files +# CHAOS: Files in inconsistent state +``` + +**Fix**: Track rollback progress +```python +class RollbackLog: + operations: List[FileOperation] + rollback_status: Dict[int, bool] = {} # index → rolled_back + +def rollback(self, rollback_log_path: Path): + log = self._load_rollback_log(rollback_log_path) + + for i, op in enumerate(reversed(log.operations)): + # Skip already rolled back + if log.rollback_status.get(i, False): + continue + + result = self._rollback_operation(op) + if result.success: + log.rollback_status[i] = True + self._save_rollback_log(log, rollback_log_path) +``` + +--- + +## 8. Scalability & Performance + +### 8.1 Known Bottlenecks + +#### Bottleneck 1: Serial ffprobe Execution + +**Location**: scanner.py:401-487 `_extract_metadata_ffprobe()` + +**Issue**: +```python +# Called once per video file +def _create_video_file(...): + if use_metadata: + metadata = _extract_metadata_ffprobe(file_path, logger) + # ⚠️ Blocking subprocess.run() call per file +``` + +**Impact**: +- 10,000 files × 0.5s per ffprobe = 5,000 seconds = **83 minutes** +- Single-threaded, CPU idle during I/O wait + +**Recommendation**: Parallel ffprobe with process pool +```python +from concurrent.futures import ProcessPoolExecutor + +def scan_library_parallel(library_root, config, max_workers=8): + files = _discover_files(library_root) + + with ProcessPoolExecutor(max_workers=max_workers) as executor: + # Submit all ffprobe jobs + futures = { + executor.submit(_extract_metadata_ffprobe, f, logger): f + for f in files + } + + # Collect results as they complete + for future in as_completed(futures): + file = futures[future] + try: + metadata = future.result(timeout=10) + yield VideoFile(metadata=metadata, ...) + except TimeoutError: + logger.warning(f"ffprobe timeout for {file}") + yield VideoFile(metadata=None, ...) +``` + +**Expected Speedup**: 8x on 8-core CPU (83 min → 10 min) + +#### Bottleneck 2: Serial API Calls in Enrichment + +**Location**: enrichment.py:67-120 `enrich_identities_data()` + +**Issue**: +```python +for record in records: + # Blocking HTTP request per record + payload = _enrich_record(record, media_type, providers, logger) + # ⚠️ No concurrency despite max_concurrency=6 config +``` + +**Impact**: +- 5,000 uncached records × 1s per API call = **5,000 seconds = 83 minutes** +- Config option `enrichment.max_concurrency` exists but is **NOT IMPLEMENTED** + +**Recommendation**: Async HTTP with concurrency limit +```python +import asyncio +import aiohttp + +async def enrich_identities_async(identities, config): + semaphore = asyncio.Semaphore(config.enrichment_max_concurrency) + + async def enrich_one(record): + async with semaphore: + async with aiohttp.ClientSession() as session: + return await _enrich_record_async(record, session, config) + + tasks = [enrich_one(r) for r in records] + results = await asyncio.gather(*tasks, return_exceptions=True) + return results +``` + +**Expected Speedup**: 6x with max_concurrency=6 (83 min → 14 min) + +**Alternative**: Use threading for I/O-bound tasks +```python +from concurrent.futures import ThreadPoolExecutor + +def enrich_identities_threaded(identities, config): + with ThreadPoolExecutor(max_workers=config.enrichment_max_concurrency) as executor: + futures = [ + executor.submit(_enrich_record, record, providers, logger) + for record in records + ] + for future in as_completed(futures): + result = future.result() + # Process result +``` + +### 8.2 Memory Concerns + +#### Issue 1: In-Memory VideoFile List + +**Location**: scanner.py:36 `scan_library()` return type + +**Problem**: +```python +def scan_library(...) -> List[VideoFile]: + files = [] + for entry in _discover_files(...): + vf = _create_video_file(entry, ...) + files.append(vf) # ⚠️ Accumulates in memory + return files # ⚠️ Entire list returned at once +``` + +**Impact**: +- 100,000 files × 500 bytes per VideoFile = **50 MB** +- Plus metadata strings (resolution, codec): **+50 MB** +- Peak memory: **~100 MB** for large libraries + +**Severity**: Low for most users (modern systems have GB of RAM), but problematic for: +- Very large libraries (1M+ files) +- Constrained environments (embedded systems, containers with memory limits) + +**Recommendation**: Generator pattern for streaming +```python +def scan_library(...) -> Iterator[VideoFile]: + for entry in _discover_files(...): + vf = _create_video_file(entry, ...) + yield vf # ⚠️ Stream one at a time + +# Usage +for vf in scan_library(...): + writer.writerow(vf.to_csv_row()) # Process and discard +``` + +**Trade-off**: Generator requires changes to all consumers of `scan_library()` + +#### Issue 2: Identities JSON in Memory + +**Location**: enrichment.py:22 `enrich_identities_data(identities_data: dict)` + +**Problem**: +```python +# Entire identities.json loaded into memory +with open("identities.json") as f: + identities = json.load(f) # ⚠️ Full parse to dict + +enrich_identities_data(identities, ...) # ⚠️ Mutates dict in place +``` + +**Impact**: +- 50,000 identities × 1 KB each = **50 MB** +- After enrichment (adds title_zh, reputation, etc.): **+50 MB** +- Peak: **~100 MB** during enrich + +**Recommendation**: Stream JSON processing +```python +import ijson # Iterative JSON parser + +def enrich_identities_streaming(input_path, output_path, config): + with open(input_path, "rb") as fin, open(output_path, "w") as fout: + fout.write('{"movies": [') + first = True + + # Stream parse movies array + for record in ijson.items(fin, "movies.item"): + enriched = _enrich_record(record, providers, config) + if not first: + fout.write(',') + json.dump(enriched, fout) + first = False + + fout.write('], "series": [...]}') +``` + +**Trade-off**: More complex code, harder to add summary stats + +--- + +## 9. Missing Features & Architectural Gaps + +### 9.1 Schema Versioning + +**Problem**: CSV and JSON files have no version field. + +**Impact**: +- Can't detect format changes when evolving codebase +- If field added/removed, old files crash on load +- No migration path for legacy data + +**Example Breaking Change**: +```python +# v1.0: identities.json +{"movies": [{"title": "Inception", "year": 2010}]} + +# v1.1: Add confidence field (BREAKING) +{"movies": [{"title": "Inception", "year": 2010, "confidence": 0.95}]} + +# Loading v1.0 file with v1.1 code +identity = record["confidence"] # KeyError! +``` + +**Recommendation**: Add version to all data files +```json +{ + "version": "1.1", + "movies": [...] +} +``` + +**Migration Handler**: +```python +def load_identities_json(path: Path) -> dict: + with open(path) as f: + data = json.load(f) + + version = data.get("version", "1.0") # Default to 1.0 if missing + + # Migrate old formats + if version == "1.0": + data = _migrate_1_0_to_1_1(data) + elif version == "1.1": + pass # Current version + else: + raise VLMDataError(f"Unsupported identities version: {version}") + + return data + +def _migrate_1_0_to_1_1(data: dict) -> dict: + # Add default confidence to old records + for movie in data.get("movies", []): + if "confidence" not in movie: + movie["confidence"] = 0.5 # Unknown confidence + data["version"] = "1.1" + return data +``` + +### 9.2 Path Canonicalization + +**Problem**: Paths not canonicalized, symlinks cause duplicates. + +**Example**: +```bash +/library/movie/real/Inception.mkv # Real file +/library/movie/symlink/Inception.mkv -> ../real/Inception.mkv # Symlink + +vlm scan +# Both appear as separate files in inventory.csv! +``` + +**Impact**: +- Duplicate detection fails (same file counted twice) +- State tracking inconsistent (same file has two state entries) +- Execution plan may target same file via different paths + +**Recommendation**: Canonicalize all paths +```python +from pathlib import Path + +def canonicalize_path(path: Path) -> Path: + """Resolve symlinks and relative components.""" + return path.resolve() + +# Usage in scanner +def _create_video_file(entry, ...): + file_path = canonicalize_path(Path(entry.path)) + # Now symlinks resolve to same canonical path +``` + +**Trade-off**: Performance cost (resolve() requires filesystem access) + +### 9.3 Unified I/O Layer + +**Problem**: CSV/JSON loading duplicated across modules. + +**Evidence**: +- `cli.py parse`: Manual CSV parsing (lines 200-214) +- `commands/analyze.py`: Manual JSON loading +- `io.py`: Provides helpers but not consistently used + +**Impact**: +- Schema changes require updates in multiple places +- Inconsistent error handling +- Hard to add features (e.g., compression, encryption) + +**Recommendation**: FileRepository abstraction (see Section 3.3) + +### 9.4 Configuration Validation + +**Problem**: Config validated only in cli.py main(), not in Config class. + +**Impact**: +- Commands can load invalid config and fail late +- Hard to test config validation in isolation + +**Current**: +```python +# cli.py:150-180 +config = Config.load_from_yaml(config_path) +# ⚠️ No validation at load time + +# Later in command +if not config.library_root: + raise VLMConfigError("library_root not set") +``` + +**Recommendation**: Validate at construction +```python +class Config: + @classmethod + def load_from_yaml(cls, path: Path) -> "Config": + with open(path) as f: + data = yaml.safe_load(f) + + config = cls(data) + config.validate() # ⚠️ Validate immediately + return config + + def validate(self): + errors = [] + + if not self.library_root: + errors.append("library_root is required") + elif not Path(self.library_root).exists(): + errors.append(f"library_root does not exist: {self.library_root}") + + if self.video_extensions and not isinstance(self.video_extensions, list): + errors.append("video_extensions must be a list") + + if errors: + raise VLMConfigError("\n".join(errors)) +``` + +--- + +## 10. Recommendations Summary + +### Priority 1: Critical Fixes (Do Immediately) + +| Issue | Impact | Effort | Fix | +|-------|--------|--------|-----| +| **Identity reconstruction loses metadata** | ❌ Duplicate resolution by quality broken | Medium | Store full VideoFile in identities.json or join with inventory.csv | +| **Quarantine non-atomic** | ❌ File loss risk if manifest update fails | Low | Update manifest before moving file | +| **Path canonicalization missing** | ❌ Symlinks cause duplicates | Low | Use `Path.resolve()` everywhere | +| **No schema versioning** | ❌ Breaking changes crash on old files | Low | Add "version" field to JSON/CSV | + +### Priority 2: Important Improvements (Do Soon) + +| Issue | Impact | Effort | Fix | +|-------|--------|--------|-----| +| **Duplicated I/O logic** | ⚠️ Maintenance burden, inconsistency | Medium | Consolidate to io.py, enforce usage | +| **No exception hierarchy** | ⚠️ Poor error handling, no retry logic | Low | Create VLMError base class | +| **State not integrated** | ⚠️ Manual state tracking, no automation | Medium | Auto-update state in execute stage | +| **Config validation late** | ⚠️ Commands fail after user invocation | Low | Validate in Config.__init__ | + +### Priority 3: Performance Optimizations (Nice to Have) + +| Issue | Impact | Effort | Fix | +|-------|--------|--------|-----| +| **Serial ffprobe calls** | ⚠️ Slow scan (83 min for 10K files) | High | Parallel execution with ProcessPoolExecutor | +| **Serial API calls** | ⚠️ Slow enrichment (83 min for 5K records) | High | Async HTTP with aiohttp or ThreadPoolExecutor | +| **In-memory file lists** | ⚠️ Memory usage (100MB for 100K files) | Medium | Generator pattern for streaming | + +### Priority 4: Architectural Enhancements (Future) + +| Enhancement | Benefit | Effort | Description | +|-------------|---------|--------|-------------| +| **Transaction semantics** | Resume failed stages without re-processing | High | Checkpoint files for each stage | +| **Explicit state machine** | Track workflow progression explicitly | Medium | WorkflowState enum with valid transitions | +| **Repository pattern** | Testable data access, easy to mock | Medium | FileRepository abstraction layer | +| **Cache TTL** | Fresh TMDB data without manual refresh | Low | Add expires_at to enrichment cache | + +--- + +## 11. Conclusion + +### Overall Design Quality: ⭐⭐⭐⭐ (4/5) + +VLM demonstrates **strong architectural fundamentals**: +- Clear separation of concerns +- Safety-first workflow with human checkpoints +- Graceful degradation and comprehensive error logging +- Well-structured modules with mostly reasonable coupling + +### Critical Gaps Requiring Attention: + +1. **Data Loss in Workflow**: Metadata lost during identity reconstruction breaks duplicate resolution +2. **Non-Atomic Operations**: File operations can fail partway, leaving inconsistent state +3. **Missing Abstractions**: No unified I/O layer, no exception hierarchy, no schema versioning +4. **Performance Bottlenecks**: Serial execution of expensive operations (ffprobe, API calls) + +### Recommended Next Steps: + +**Week 1**: Fix critical data loss bug +- Preserve VideoFile metadata in identities.json +- Add integration test for duplicate resolution by quality + +**Week 2**: Add safety and consistency +- Implement atomic quarantine operations +- Add path canonicalization +- Add schema versioning to all data files + +**Week 3**: Consolidate I/O layer +- Move all CSV/JSON operations to io.py +- Enforce usage via code review +- Add validation for all loaded data + +**Month 2**: Performance optimization +- Parallelize ffprobe calls +- Implement async API calls for enrichment +- Benchmark and measure improvements + +### Final Assessment + +VLM is a **well-designed, production-ready tool** with a few critical bugs and several opportunities for improvement. The codebase is maintainable, testable, and follows Python best practices. With the recommended fixes, particularly addressing the metadata loss issue, VLM will be a robust and scalable solution for video library management. + +**Confidence in Current Design**: High (85%) +**Confidence After Priority 1+2 Fixes**: Very High (95%) + +--- + +**End of Review** diff --git a/src/vlm/executor.py b/src/vlm/executor.py index c6b2773..deeb592 100644 --- a/src/vlm/executor.py +++ b/src/vlm/executor.py @@ -169,15 +169,11 @@ class ExecutionEngine: error_message=None, executed_at=executed_at ) - result = self._quarantine_manager.quarantine_file( + # Quarantine the file and return the result directly + # (includes the actual quarantine destination_path for rollback) + return self._quarantine_manager.quarantine_file( operation.source_path, operation.reason ) - return OperationResult( - operation=operation, - success=result.success, - error_message=result.error_message, - executed_at=result.executed_at - ) # Handle conflicted operations if operation.has_conflict: @@ -536,6 +532,43 @@ class ExecutionEngine: executed_at=executed_at ) + # Handle quarantine operations using QuarantineManager + if operation.operation_type == "quarantine": + if not self._quarantine_manager: + error_msg = "Cannot rollback quarantine: QuarantineManager not available" + log_operation( + self.logger, + logging.ERROR, + error_msg, + operation_type="rollback" + ) + return OperationResult( + operation=operation, + success=False, + error_message=error_msg, + executed_at=executed_at + ) + + # Use QuarantineManager to restore the file + if operation.destination_path: + restore_result = self._quarantine_manager.restore_from_quarantine( + operation.destination_path + ) + return OperationResult( + operation=operation, + success=restore_result.success, + error_message=restore_result.error_message, + executed_at=executed_at + ) + else: + error_msg = "Cannot rollback quarantine: no destination path recorded" + return OperationResult( + operation=operation, + success=False, + error_message=error_msg, + executed_at=executed_at + ) + try: # For move/rename operations, reverse the direction # Original: source -> destination diff --git a/src/vlm/models.py b/src/vlm/models.py index 75eb3f6..de6ca71 100644 --- a/src/vlm/models.py +++ b/src/vlm/models.py @@ -183,7 +183,7 @@ class RollbackLog: @dataclass class QuarantineEntry: """Represents a single file in quarantine. - + Attributes: original_path: Original path of the file before quarantine quarantine_path: Path to the file in quarantine directory @@ -191,6 +191,7 @@ class QuarantineEntry: reason: Optional reason for quarantining the file size_bytes: File size in bytes category: Category of the video ("movie" or "series") + status: Operation status ("pending" | "committed") for two-phase commit """ original_path: Path quarantine_path: Path @@ -198,6 +199,7 @@ class QuarantineEntry: reason: Optional[str] size_bytes: int category: str + status: str = "committed" # Default for backward compatibility @dataclass diff --git a/src/vlm/quarantine.py b/src/vlm/quarantine.py index ae875e0..e1113fd 100644 --- a/src/vlm/quarantine.py +++ b/src/vlm/quarantine.py @@ -187,54 +187,31 @@ class QuarantineManager: executed_at=executed_at ) - # Move file to quarantine + # Two-phase commit for atomic quarantine operation + + # PHASE 1: Write pending manifest entry BEFORE moving file + manifest = self._load_manifest(category) + pending_entry = QuarantineEntry( + original_path=file_path, + quarantine_path=quarantine_path, + quarantined_at=executed_at, + reason=reason, + size_bytes=file_size, + category=category, + status="pending" + ) + manifest.entries.append(pending_entry) + try: - file_path.rename(quarantine_path) - + self._save_manifest(category, manifest) log_operation( self.logger, - logging.INFO, - f"Successfully quarantined file: {file_path} -> {quarantine_path}", - operation_type="quarantine", - file_path=file_path + logging.DEBUG, + f"Phase 1: Wrote pending manifest entry for {file_path}", + operation_type="quarantine" ) - - # Update manifest - try: - self._update_manifest( - category=category, - original_path=file_path, - quarantine_path=quarantine_path, - quarantined_at=executed_at, - reason=reason, - size_bytes=file_size - ) - except Exception as e: - # Log manifest update failure but don't fail the operation - # since the file was already moved successfully - log_operation( - self.logger, - logging.WARNING, - f"Failed to update manifest: {str(e)}", - operation_type="quarantine", - file_path=file_path - ) - - return OperationResult( - operation=FileOperation( - operation_type="quarantine", - source_path=file_path, - destination_path=quarantine_path, - reason=reason or "Quarantine", - has_conflict=False - ), - success=True, - error_message=None, - executed_at=executed_at - ) - except Exception as e: - error_msg = f"Failed to move file to quarantine: {str(e)}" + error_msg = f"Failed to write pending manifest entry: {str(e)}" log_operation( self.logger, logging.ERROR, @@ -254,6 +231,91 @@ class QuarantineManager: error_message=error_msg, executed_at=executed_at ) + + # PHASE 2: Move file + try: + file_path.rename(quarantine_path) + log_operation( + self.logger, + logging.INFO, + f"Phase 2: Successfully moved file: {file_path} -> {quarantine_path}", + operation_type="quarantine", + file_path=file_path + ) + except Exception as e: + # Rollback: Remove pending entry from manifest + error_msg = f"Failed to move file to quarantine: {str(e)}" + log_operation( + self.logger, + logging.ERROR, + error_msg, + operation_type="quarantine", + file_path=file_path + ) + + try: + manifest.entries.remove(pending_entry) + self._save_manifest(category, manifest) + log_operation( + self.logger, + logging.INFO, + f"Rollback: Removed pending manifest entry for {file_path}", + operation_type="quarantine" + ) + except Exception as rollback_error: + log_operation( + self.logger, + logging.ERROR, + f"Rollback failed: {str(rollback_error)}", + operation_type="quarantine" + ) + + return OperationResult( + operation=FileOperation( + operation_type="quarantine", + source_path=file_path, + destination_path=quarantine_path, + reason=reason or "Quarantine", + has_conflict=False + ), + success=False, + error_message=error_msg, + executed_at=executed_at + ) + + # PHASE 3: Mark as committed + try: + pending_entry.status = "committed" + self._save_manifest(category, manifest) + log_operation( + self.logger, + logging.DEBUG, + f"Phase 3: Marked manifest entry as committed for {file_path}", + operation_type="quarantine" + ) + except Exception as e: + # File was moved but manifest update failed + # Auto-recovery will handle this on next load + log_operation( + self.logger, + logging.WARNING, + f"Failed to mark entry as committed (auto-recovery will fix): {str(e)}", + operation_type="quarantine", + file_path=file_path + ) + + return OperationResult( + operation=FileOperation( + operation_type="quarantine", + source_path=file_path, + destination_path=quarantine_path, + reason=reason or "Quarantine", + has_conflict=False + ), + success=True, + error_message=None, + executed_at=executed_at + ) def _determine_category(self, file_path: Path) -> str: """Determine the category of a file based on its path. @@ -363,11 +425,16 @@ class QuarantineManager: quarantined_at=datetime.fromisoformat(entry_data['quarantined_at']), reason=entry_data.get('reason'), size_bytes=entry_data['size_bytes'], - category=entry_data['category'] + category=entry_data['category'], + status=entry_data.get('status', 'committed') # Default for backward compatibility ) entries.append(entry) - - return QuarantineManifest(entries=entries) + + # Auto-recovery: Clean up pending entries + manifest = QuarantineManifest(entries=entries) + self._recover_pending_entries(category, manifest) + + return manifest except Exception as e: log_operation( @@ -402,7 +469,8 @@ class QuarantineManager: 'quarantined_at': entry.quarantined_at.isoformat(), 'reason': entry.reason, 'size_bytes': entry.size_bytes, - 'category': entry.category + 'category': entry.category, + 'status': entry.status } for entry in manifest.entries ] @@ -418,7 +486,73 @@ class QuarantineManager: f"Saved manifest for category '{category}' with {len(manifest.entries)} entries", operation_type="quarantine" ) - + + def _recover_pending_entries(self, category: str, manifest: QuarantineManifest) -> None: + """Auto-recovery: Clean up pending entries from incomplete operations. + + Checks each pending entry: + - If file exists in quarantine → mark as committed + - If file doesn't exist → remove orphaned entry + + Args: + category: Category name + manifest: Manifest to recover (modified in place) + """ + pending_entries = [e for e in manifest.entries if e.status == "pending"] + + if not pending_entries: + return + + log_operation( + self.logger, + logging.INFO, + f"Auto-recovery: Found {len(pending_entries)} pending entries for category '{category}'", + operation_type="quarantine" + ) + + recovered = 0 + removed = 0 + + for entry in pending_entries: + if entry.quarantine_path.exists(): + # File exists → mark as committed + entry.status = "committed" + recovered += 1 + log_operation( + self.logger, + logging.INFO, + f"Auto-recovery: Marked as committed: {entry.quarantine_path}", + operation_type="quarantine" + ) + else: + # File doesn't exist → remove orphaned entry + manifest.entries.remove(entry) + removed += 1 + log_operation( + self.logger, + logging.WARNING, + f"Auto-recovery: Removed orphaned entry: {entry.original_path}", + operation_type="quarantine" + ) + + if recovered > 0 or removed > 0: + # Save recovered manifest + try: + self._save_manifest(category, manifest) + log_operation( + self.logger, + logging.INFO, + f"Auto-recovery: Completed for '{category}' - {recovered} committed, {removed} removed", + operation_type="quarantine" + ) + except Exception as e: + log_operation( + self.logger, + logging.ERROR, + f"Auto-recovery: Failed to save manifest: {str(e)}", + operation_type="quarantine" + ) + def _update_manifest( self, category: str, @@ -679,55 +813,20 @@ class QuarantineManager: executed_at=executed_at ) - # Move file back to original location + # Two-phase commit for atomic restore operation + + # PHASE 1: Mark entry as pending restoration + entry.status = "pending" try: - quarantine_path.rename(original_path) - + self._save_manifest(category, manifest) log_operation( self.logger, - logging.INFO, - f"Successfully restored file: {quarantine_path} -> {original_path}", - operation_type="restore", - file_path=quarantine_path + logging.DEBUG, + f"Phase 1: Marked entry as pending restoration for {quarantine_path}", + operation_type="restore" ) - - # Remove entry from manifest - try: - manifest.entries.remove(entry) - self._save_manifest(category, manifest) - - log_operation( - self.logger, - logging.INFO, - f"Removed entry from manifest for category '{category}'", - operation_type="restore" - ) - except Exception as e: - # Log manifest update failure but don't fail the operation - # since the file was already moved successfully - log_operation( - self.logger, - logging.WARNING, - f"Failed to update manifest after restore: {str(e)}", - operation_type="restore", - file_path=quarantine_path - ) - - return OperationResult( - operation=FileOperation( - operation_type="restore", - source_path=quarantine_path, - destination_path=original_path, - reason="Restore", - has_conflict=False - ), - success=True, - error_message=None, - executed_at=executed_at - ) - except Exception as e: - error_msg = f"Failed to restore file: {str(e)}" + error_msg = f"Failed to mark entry as pending: {str(e)}" log_operation( self.logger, logging.ERROR, @@ -747,7 +846,92 @@ class QuarantineManager: error_message=error_msg, executed_at=executed_at ) - + + # PHASE 2: Move file back to original location + try: + quarantine_path.rename(original_path) + log_operation( + self.logger, + logging.INFO, + f"Phase 2: Successfully restored file: {quarantine_path} -> {original_path}", + operation_type="restore", + file_path=quarantine_path + ) + except Exception as e: + # Rollback: Mark entry back as committed + error_msg = f"Failed to restore file: {str(e)}" + log_operation( + self.logger, + logging.ERROR, + error_msg, + operation_type="restore", + file_path=quarantine_path + ) + + try: + entry.status = "committed" + self._save_manifest(category, manifest) + log_operation( + self.logger, + logging.INFO, + f"Rollback: Marked entry back as committed for {quarantine_path}", + operation_type="restore" + ) + except Exception as rollback_error: + log_operation( + self.logger, + logging.ERROR, + f"Rollback failed: {str(rollback_error)}", + operation_type="restore" + ) + + return OperationResult( + operation=FileOperation( + operation_type="restore", + source_path=quarantine_path, + destination_path=original_path, + reason="Restore", + has_conflict=False + ), + success=False, + error_message=error_msg, + executed_at=executed_at + ) + + # PHASE 3: Remove entry from manifest + try: + manifest.entries.remove(entry) + self._save_manifest(category, manifest) + log_operation( + self.logger, + logging.DEBUG, + f"Phase 3: Removed entry from manifest for category '{category}'", + operation_type="restore" + ) + except Exception as e: + # File was restored but manifest update failed + # This is not critical since the file is in the right place + log_operation( + self.logger, + logging.WARNING, + f"Failed to remove entry from manifest (file was restored): {str(e)}", + operation_type="restore", + file_path=quarantine_path + ) + + return OperationResult( + operation=FileOperation( + operation_type="restore", + source_path=quarantine_path, + destination_path=original_path, + reason="Restore", + has_conflict=False + ), + success=True, + error_message=None, + executed_at=executed_at + ) + def _determine_category_from_quarantine(self, quarantine_path: Path) -> Optional[str]: """Determine the category from a quarantine path.