After updates

README.md

@@ -82,6 +82,213 @@ open http://localhost:3000  (admin / your_grafana_password)
 open http://localhost:9090
 ```
 
+## System Operation
+
+### How Processes Start
+
+**Service Initialization:**
+
+1. **MariaDB** starts first with health check
+2. **Worker** initializes via `src/main.py`:
+   - Loads `config.yaml`
+   - Sets up JSON logging with rotation
+   - Connects to MariaDB (connection pooling)
+   - Initializes database schema
+   - Verifies GPU availability (CUDA/PyTorch)
+   - Starts the `DirectoryScanner` in a background thread (scans permanent storage in place)
+   - Creates `WorkerPool` with 1 worker thread
+   - Enters job processing loop
+3. **UI** starts Flask review interface via Gunicorn (2 workers)
+4. **Monitoring** starts Prometheus and Grafana independently
+
+### Processing Pipeline
+
+The worker follows this flow for each video:
+
+```
+Pending → Lock → Probe → Sample → Detect → Classify → Aggregate → Route → Persist → Export → Cleanup → Completed
+```
+
+**Detailed Steps:**
+
+1. **Job Queue** - Atomically lock `PENDING` jobs via `UPDATE status = 'PROCESSING'`
+   - Priority: newest files first (`last_scan_time DESC`)
+   - Max concurrent: 1 per GPU
+
+2. **Probe Video** - Extract metadata via FFprobe
+   - Duration, codec, resolution
+   - Validate against codec whitelist (H.264, H.265, VP8/9, AV1)
+   - Mark `UNSCANNABLE` if invalid
+
+3. **Sample Frames** - Extract frames at configured interval (default: 30s)
+   - Save as JPEG to `/scratch/{video_id}/frames/`
+   - Quality: 2 (lower=better)
+
+4. **Detect Faces** - YOLOv8n TensorRT inference (FP32)
+   - Batch size auto-tuned by GPU memory monitor
+   - NMS filtering (IoU: 0.45, confidence: 0.25)
+   - Cap: 10 faces/frame, 100 faces/video
+
+5. **Extract Crops** - Resize detected faces to 224×224
+   - Save to `/scratch/{video_id}/crops/`
+
+6. **Classify Crops** - MobileNetV3-Small TensorRT inference
+   - Temperature-scaled softmax (T=1.0)
+   - Returns confidence per crop
+
+7. **Aggregate Confidence** - Combine crop confidences into video-level score
+   - Strategy: `max` (most conservative)
+   - Alternatives: `weighted_mean`, `top_k_mean`
+
+8. **Route Decision** - Threshold-based routing:
+   - `C ≥ 0.75` → **MATCH**
+   - `0.45 ≤ C < 0.75` → **REVIEW** (human annotation)
+   - `C < 0.45` → **SKIP**
+   - No faces → **SKIP**
+
+9. **Persist Results** - Atomic transaction:
+   - Update `videos` table (confidence, routing, status)
+   - Insert `processing_logs` row (audit trail)
+   - State guard: only update if `status='PROCESSING'`
+
+10. **Export** - Buffer and batch export (default: 100 videos)
+    - Format: Parquet with Snappy compression
+    - Path: `/data/output/{model_version}/`
+    - Fallback: JSONL if Parquet fails
+
+11. **Cleanup** - Delete `/scratch/{video_id}/` directory
+    - Only after successful persistence
+    - Prevents orphaned scratch files
+
+**Directory Scanner Service (runs alongside the worker):**
+- Scans the permanent storage location **in place** (no staging/copy step)
+- Walks `/data/input` every **2 hours** by default (configurable via `scanner.scan_interval_seconds`)
+- Detects new, modified, and removed video files by comparing against the DB
+- Filters to video files by extension
+- Computes SHA256 hash, probes metadata, validates codec
+- Queues any video that has not been scanned yet as `PENDING` for the worker pool
+- **Single-instance guard:** an in-process lock plus a DB lock (with a lease) ensure only one scan runs at a time — a long-running scan never overlaps another, even across multiple worker replicas. The lock lease is refreshed via heartbeats during the scan and is taken over automatically if a scanner crashes.
+
+### Configuration Reference
+
+All configuration is in `config.yaml`. Environment variable override format: `VD_<SECTION>_<KEY>` (e.g., `VD_SAMPLING_INTERVAL_SECONDS=60`).
+
+#### Key Configuration Sections
+
+**Sampling & Thresholds:**
+```yaml
+sampling:
+  interval_seconds: 30          # Frame extraction frequency
+  quality: 2                    # JPEG quality (1-31, lower=better)
+  format: jpeg
+
+thresholds:
+  T_high: 0.75                  # MATCH threshold
+  T_low: 0.45                   # REVIEW threshold
+```
+
+**GPU & Batching:**
+```yaml
+gpu:
+  max_memory_gb: 18             # Target VRAM usage
+  batch_size: auto              # Auto-tune based on available VRAM
+
+batching:
+  max_batch_size: 16            # Maximum batch size
+  vram_target_gb: 16            # Target VRAM for batch tuning
+  vram_reduce_threshold_gb: 16  # Reduce batch if above
+  vram_increase_threshold_gb: 10 # Increase batch if below
+```
+
+**Storage Paths:**
+```yaml
+storage:
+  scratch_path: /scratch        # Temporary processing (tmpfs)
+  input_path: /data/input       # Source videos (NFS)
+  output_path: /data/output     # Results (local/NAS)
+  models_path: /models          # TensorRT models
+  training_path: /data/training # Training data
+```
+
+**Database:**
+```yaml
+database:
+  host: mariadb
+  port: 3306
+  name: videodetect
+  user: videodetect
+  password: videodetect123
+  pool_size: 20                 # Connection pool size
+  pool_min: 5
+  pool_recycle: 3600            # Recycle connections after 1h
+```
+
+**Face Detection:**
+```yaml
+face_detection:
+  model: yolo8n
+  model_path: /models/face_detector/face_detector.trt
+  input_size: 640
+  confidence_threshold: 0.25
+  iou_threshold: 0.45
+  max_faces_per_frame: 10
+  max_faces_per_video: 100
+```
+
+**Classification & Aggregation:**
+```yaml
+classifier:
+  model: mobilenetv3-small
+  model_path: /models/classifier/classifier.trt
+  input_size: 224
+  temperature: 1.0              # Calibration temperature
+
+aggregation:
+  strategy: max                 # max, weighted_mean, top_k_mean
+  alpha: 1.0                    # weighted_mean weight for mean
+  beta: 0.1                     # weighted_mean weight for variance
+  top_k: 3                      # top_k_mean: average top 3 scores
+```
+
+**Export:**
+```yaml
+export:
+  format: parquet               # parquet, jsonl, or both
+  compression: snappy
+  batch_size: 100               # Export after N videos
+  include_frame_confidences: true
+```
+
+**Review UI:**
+```yaml
+review_ui:
+  host: "0.0.0.0"
+  port: 5000
+  per_page: 20                  # Pagination
+  top_k_frames: 5               # Show top-k contributing frames
+  auth_enabled: false           # No auth per TC-06
+  ssl_enabled: false            # Internal LAN only
+```
+
+#### Volume Mounts
+
+From `docker-compose.yml`:
+
+- **Input**: NFS mount → `/data/input` (read-only)
+- **Output**: `./output` → `/data/output`
+- **Models**: `./models` → `/models`
+- **Training**: `./training` → `/data/training`
+- **Scratch**: 100GB tmpfs at `/scratch` (RAM disk)
+
+### Key Design Principles
+
+- **Atomic state transitions** - Database locks prevent race conditions
+- **Crash recovery** - `PROCESSING` jobs automatically requeued on restart
+- **Idempotent** - Re-running same video produces same result
+- **Stateless** - Scratch cleanup after each job
+- **Fail-safe** - 3 retry attempts before marking `ERROR`
+- **No auth/SSL** - Internal LAN deployment per TC-06
+
 ## Project Structure
 
 ```

config.yaml

@@ -70,12 +70,18 @@ model:
 # -----------------------------------------------------
 # Directory Scanner
 # -----------------------------------------------------
+# Scans the permanent storage location in place (no staging/copy step).
+# The corpus is large (~163k files / ~41TB), so the default interval is 2 hours.
+# A single-instance guard (in-process lock + DB lock with a lease) ensures a
+# long-running scan never overlaps another scan, even across replicas.
 scanner:
-  scan_interval_seconds: 60
+  scan_interval_seconds: 7200   # 2 hours (adaptive: raise for very large corpora)
   walker_threads: 8
   ffprobe_timeout_seconds: 10
   hash_algorithm: sha256
   hash_chunk_size_mb: 1
+  lock_lease_seconds: 21600     # 6 hours: max time a scan may hold the lock before it is considered stale
+  heartbeat_interval_files: 500 # refresh the lock lease every N files processed
 
 # -----------------------------------------------------
 # Codec Validation

db/schema.sql

@@ -115,6 +115,20 @@ CREATE TABLE IF NOT EXISTS scan_history (
     error_message TEXT DEFAULT NULL
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
 
+-- -----------------------------------------------------
+-- Table: scanner_lock
+-- Single-instance guard for the directory scanner. Ensures only one scan
+-- runs at a time across all replicas. A lease (locked_at + lease_seconds)
+-- lets a live scanner keep the lock via heartbeats, and lets a crashed
+-- scanner's lock be taken over once it goes stale.
+-- -----------------------------------------------------
+CREATE TABLE IF NOT EXISTS scanner_lock (
+    lock_name VARCHAR(64) PRIMARY KEY,
+    owner VARCHAR(128) NOT NULL COMMENT 'Instance id (host-pid-uuid) holding the lock',
+    locked_at DATETIME NOT NULL COMMENT 'Last time the lock was acquired or heartbeated',
+    lease_seconds INT NOT NULL DEFAULT 21600 COMMENT 'Lock is stale if older than this (6 hours)'
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
+
 -- -----------------------------------------------------
 -- Insert default model entry
 -- -----------------------------------------------------

src/main.py

@@ -8,6 +8,7 @@ Initializes all components and starts the processing pipeline.
 import logging
 import signal
 import sys
+import threading
 import time
 from pathlib import Path
 
@@ -18,6 +19,7 @@ from config_loader import get_config
 from db_connector import DBConnector
 from logging_config import setup_logging
 from orchestrator import WorkerPool
+from scanner import DirectoryScanner
 
 logger = logging.getLogger(__name__)
 
@@ -84,7 +86,27 @@ def main():
     except ImportError:
         logger.warning("PyTorch not installed. GPU features disabled.")
 
-    logger.info("Worker initialization complete. Starting processing loop...")
+    # Initialize the directory scanner.
+    # It scans the permanent storage location in place (no staging/copy step),
+    # periodically discovering new/removed video files and queueing any that
+    # have not been scanned yet as PENDING for the worker pool to pick up.
+    storage_config = config.get_section("storage")
+    scanner_config = config.get_section("scanner")
+    scanner = DirectoryScanner(
+        db_connector=db,
+        config=config.data,
+        input_path=storage_config.get("input_path", "/data/input"),
+        scan_interval=int(scanner_config.get("scan_interval_seconds", 60)),
+        walker_threads=int(scanner_config.get("walker_threads", 8)),
+    )
+
+    logger.info("Worker initialization complete. Starting scanner and processing loop...")
+
+    # Run the scanner in a background thread (its start() is a blocking loop).
+    scanner_thread = threading.Thread(
+        target=scanner.start, name="directory-scanner", daemon=True
+    )
+    scanner_thread.start()
 
     pool = WorkerPool(db, config.data, max_workers=1)
 
@@ -93,6 +115,7 @@ def main():
     except KeyboardInterrupt:
         logger.info("Worker shutting down.")
         pool.stop()
+        scanner.stop()
 
 
 if __name__ == "__main__":

src/scanner.py

@@ -10,9 +10,12 @@ import hashlib
 import json
 import logging
 import os
+import socket
+import threading
 import time
+import uuid
 from concurrent.futures import ThreadPoolExecutor, as_completed
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from typing import Dict, List, Optional, Tuple
 
@@ -54,19 +57,62 @@ class DirectoryScanner:
         self._total_files_modified = 0
         self._total_files_unscannable = 0
 
+        # Single-instance guard: prevents overlapping scans both within this
+        # process (threading lock) and across replicas (DB lock with a lease).
+        self._instance_id = f"{socket.gethostname()}-{os.getpid()}-{uuid.uuid4().hex[:8]}"
+        self._lock_name = "directory_scanner"
+        self._lock_lease_seconds = int(
+            config.get("scanner", {}).get("lock_lease_seconds", 21600)
+        )
+        self._heartbeat_every = int(
+            config.get("scanner", {}).get("heartbeat_interval_files", 500)
+        )
+        self._scan_lock = threading.Lock()
+        self._lock_table_ensured = False
+
     def start(self):
-        """Start the scanner loop."""
+        """Start the scanner loop.
+
+        Each cycle is guarded so that at most one scan runs at a time:
+        - an in-process threading lock prevents re-entrant scans, and
+        - a DB lock (with a lease) prevents overlapping scans across replicas.
+        A long-running scan keeps its lease alive via heartbeats, so the next
+        scheduled tick (or another replica) waits instead of starting a second
+        parallel scan.
+        """
         self._running = True
-        logger.info("Scanner starting: input_path=%s interval=%ds threads=%d",
+        logger.info(
-                     self.input_path, self.scan_interval, self.walker_threads)
+            "Scanner starting: input_path=%s interval=%ds threads=%d instance=%s",
+            self.input_path, self.scan_interval, self.walker_threads, self._instance_id,
+        )
 
         while self._running:
+            # In-process re-entrancy guard: never run two scans at once.
+            if not self._scan_lock.acquire(blocking=False):
+                logger.warning("A scan is already in progress; skipping this cycle.")
+                self._sleep_interval()
+                continue
+
+            try:
+                if self.acquire_lock():
                     try:
                         self._run_scan()
+                    finally:
+                        self.release_lock()
+                else:
+                    logger.info(
+                        "Scanner lock held by another instance; skipping this cycle."
+                    )
             except Exception as e:
                 logger.error("Scanner error: %s", e, exc_info=True)
+            finally:
+                self._scan_lock.release()
 
             # Sleep until next scan
+            self._sleep_interval()
+
+    def _sleep_interval(self):
+        """Sleep for the scan interval, waking early if stopped."""
         for _ in range(self.scan_interval):
             if not self._running:
                 break
@@ -77,6 +123,98 @@ class DirectoryScanner:
         self._running = False
         logger.info("Scanner stopping. Total scans: %d", self._scan_count)
 
+    # ------------------------------------------------------------------
+    # Single-instance lock (cross-process / cross-replica guard)
+    # ------------------------------------------------------------------
+    def _ensure_lock_table(self):
+        """Create the scanner_lock table if it does not already exist."""
+        if self._lock_table_ensured:
+            return
+        self.db.execute(
+            """CREATE TABLE IF NOT EXISTS scanner_lock (
+                   lock_name VARCHAR(64) PRIMARY KEY,
+                   owner VARCHAR(128) NOT NULL,
+                   locked_at DATETIME NOT NULL,
+                   lease_seconds INT NOT NULL DEFAULT 21600
+               ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"""
+        )
+        self._lock_table_ensured = True
+
+    def acquire_lock(self) -> bool:
+        """Attempt to acquire the cross-process scanner lock.
+
+        Returns True if this instance now owns the lock, False otherwise.
+        A stale lock (held longer than the lease) is taken over so a crashed
+        scanner does not block scanning forever.
+        """
+        try:
+            self._ensure_lock_table()
+        except Exception as e:
+            logger.warning("Could not ensure scanner_lock table: %s", e)
+            return True  # fail-open: keep scanning rather than block entirely
+
+        now = datetime.now(timezone.utc).replace(tzinfo=None)
+
+        # 1) Try to insert a fresh lock row.
+        try:
+            self.db.execute(
+                """INSERT INTO scanner_lock (lock_name, owner, locked_at, lease_seconds)
+                   VALUES (%s, %s, %s, %s)""",
+                (self._lock_name, self._instance_id, now, self._lock_lease_seconds),
+                transaction=True,
+            )
+            logger.info("Acquired scanner lock (fresh). owner=%s", self._instance_id)
+            return True
+        except Exception:
+            # Row already exists -> fall through to steal-if-stale.
+            pass
+
+        # 2) Take over the lock if it is stale or already ours.
+        stale_before = now - timedelta(seconds=self._lock_lease_seconds)
+        try:
+            affected = self.db.execute(
+                """UPDATE scanner_lock
+                   SET owner = %s, locked_at = %s
+                   WHERE lock_name = %s
+                     AND (owner = %s OR locked_at < %s)""",
+                (self._instance_id, now, self._lock_name, self._instance_id, stale_before),
+                transaction=True,
+            )
+            if affected and affected > 0:
+                logger.info("Acquired scanner lock (stale takeover). owner=%s",
+                            self._instance_id)
+                return True
+        except Exception as e:
+            logger.warning("Failed to check scanner lock: %s", e)
+            return True  # fail-open
+
+        logger.info("Scanner lock held by another instance; not acquiring.")
+        return False
+
+    def release_lock(self):
+        """Release the scanner lock if we own it."""
+        try:
+            self.db.execute(
+                """DELETE FROM scanner_lock WHERE lock_name = %s AND owner = %s""",
+                (self._lock_name, self._instance_id),
+                transaction=True,
+            )
+        except Exception as e:
+            logger.warning("Failed to release scanner lock: %s", e)
+
+    def _heartbeat(self):
+        """Refresh the lock lease so a long-running scan is not stolen."""
+        try:
+            now = datetime.now(timezone.utc).replace(tzinfo=None)
+            self.db.execute(
+                """UPDATE scanner_lock SET locked_at = %s
+                   WHERE lock_name = %s AND owner = %s""",
+                (now, self._lock_name, self._instance_id),
+                transaction=True,
+            )
+        except Exception as e:
+            logger.debug("Scanner lock heartbeat failed: %s", e)
+
     def _run_scan(self):
         """Execute a single scan cycle."""
         scan_start = time.time()
@@ -181,6 +319,7 @@ class DirectoryScanner:
     def _process_files_batch(self, files: List[Path]) -> List[dict]:
         """Process a batch of files in parallel."""
         results = []
+        processed = 0
 
         with ThreadPoolExecutor(max_workers=self.walker_threads) as executor:
             future_to_file = {
@@ -201,6 +340,11 @@ class DirectoryScanner:
                         "error_message": str(e),
                     })
 
+                processed += 1
+                # Keep the single-instance lock alive during long scans.
+                if self._heartbeat_every and processed % self._heartbeat_every == 0:
+                    self._heartbeat()
+
         return results
 
     def _process_single_file(self, file_path: Path) -> dict:
@@ -384,6 +528,10 @@ class DirectoryScanner:
             "total_files_new": self._total_files_new,
             "total_files_modified": self._total_files_modified,
             "total_files_unscannable": self._total_files_unscannable,
+            "instance_id": self._instance_id,
+            "scan_interval_seconds": self.scan_interval,
+            "lock_lease_seconds": self._lock_lease_seconds,
+            "scan_in_progress": self._scan_lock.locked(),
             "input_path": str(self.input_path),
             "is_running": self._running,
         }