Fixes

.env

@@ -0,0 +1,20 @@
+# VideoDetect — Docker Compose environment variables
+# Copy this file to .env and fill in real values before running docker-compose up.
+
+# -----------------------------------------------------
+# Database
+# -----------------------------------------------------
+DB_ROOT_PASSWORD=changeme_root
+DB_PASSWORD=changeme_videodetect
+
+# -----------------------------------------------------
+# Storage paths (host-side mounts)
+# -----------------------------------------------------
+NAS_OUTPUT_PATH=${PWD}/output
+MODELS_PATH=${PWD}/models
+TRAINING_PATH=${PWD}/training
+
+# -----------------------------------------------------
+# Grafana
+# -----------------------------------------------------
+GRAFANA_PASSWORD=changeme_grafana

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,25 +70,35 @@ 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
 # -----------------------------------------------------
 codec:
   whitelist:
-    - avc1      # H.264
-    - hevc      # H.265
+    - avc1      # H.264 (MP4 container tag)
+    - h264      # H.264 (ffprobe codec name)
+    - hevc      # H.265 (MP4 container tag)
+    - h265      # H.265 (ffprobe codec name)
     - vp8
     - vp9
     - av01      # AV1
     - mjpeg
     - mp4v
+    - wmv3      # Windows Media Video 7
+    - mpeg4     # MPEG-4 Part 2
   default_status_on_error: UNSCANNABLE
 
 # -----------------------------------------------------

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
 -- -----------------------------------------------------

docker-compose.yml

@@ -1,5 +1,3 @@
-version: '3.8'
-
 services:
   mariadb:
     image: mariadb:10.11
@@ -29,11 +27,10 @@ services:
 
   worker:
     build:
-      context: ./worker
-      dockerfile: Dockerfile
+      context: .
+      dockerfile: worker/Dockerfile
     container_name: videodetect-worker
     restart: unless-stopped
-    runtime: nvidia
     environment:
       - NVIDIA_VISIBLE_DEVICES=all
       - NVIDIA_DRIVER_CAPABILITIES=compute,utility
@@ -45,10 +42,10 @@ services:
       - CONFIG_PATH=/app/config.yaml
     volumes:
       - /dev/null:/dev/null  # tmpfs mounted at /scratch in container
-      - ${NAS_INPUT_PATH:-/mnt/nas/input}:/data/input:ro
-      - ${NAS_OUTPUT_PATH:-/mnt/nas/output}:/data/output
-      - ${MODELS_PATH:-/mnt/nas/models}:/models
-      - ${TRAINING_PATH:-/mnt/nas/training}:/data/training
+      - nas_input:/data/input:ro
+      - ${NAS_OUTPUT_PATH:-./output}:/data/output
+      - ${MODELS_PATH:-./models}:/models
+      - ${TRAINING_PATH:-./training}:/data/training
     tmpfs:
       - /scratch:noexec,nosuid,size=100G
     deploy:
@@ -57,11 +54,14 @@ services:
           devices:
             - driver: nvidia
               count: all
-              capabilities: [gpu]
+              capabilities: ["gpu"]
         limits:
           memory: 24G
     networks:
       - videodetect-network
+    depends_on:
+      mariadb:
+        condition: service_healthy
     healthcheck:
       test: ["CMD", "python3", "-c", "import torch; print(torch.cuda.is_available())"]
       interval: 30s
@@ -86,7 +86,7 @@ services:
       - FLASK_ENV=production
     volumes:
       - ./ui:/app
-      - ${NAS_INPUT_PATH:-/mnt/nas/input}:/data/input:ro
+      - nas_input:/data/input:ro
       - ${NAS_OUTPUT_PATH:-/mnt/nas/output}:/data/output
     networks:
       - videodetect-network
@@ -142,6 +142,12 @@ volumes:
     driver: local
   grafana_data:
     driver: local
+  nas_input:
+    driver: local
+    driver_opts:
+      type: nfs
+      o: addr=10.0.0.2,ro,nfsvers=4,hard,intr
+      device: ":/mnt/Bulk/Homes/ryan/Prawns"
 
 networks:
   videodetect-network:

src/db_connector.py

@@ -3,7 +3,7 @@ Database connection layer with connection pooling for VideoDetect.
 
 Provides:
 - Connection pooling via DBUtils + PyMySQL
-- Automatic reconnection on disconnect
+- Automatic reconnection on disconnect (ping + retry on transient errors)
 - Context manager support
 - Prepared statements for all queries
 - Transaction support for atomic state transitions
@@ -11,6 +11,7 @@ Provides:
 
 import contextlib
 import logging
+import time
 from typing import Optional
 
 from dbutils.pooled_db import PooledDB
@@ -22,6 +23,12 @@ logger = logging.getLogger(__name__)
 class DBConnector:
     """Thread-safe database connection pool manager."""
 
+    # Transient errors that indicate a dropped/stale connection and are safe
+    # to retry with a fresh connection from the pool.
+    _RETRYABLE_ERRORS = (pymysql.err.InterfaceError, pymysql.err.OperationalError)
+    _MAX_RETRIES = 3
+    _RETRY_BACKOFF_SECONDS = 0.5
+
     def __init__(
         self,
         host: str = "mariadb",
@@ -33,27 +40,37 @@ class DBConnector:
         pool_min: int = 5,
         pool_recycle: int = 3600,
     ):
-        self._pool = PooledDB(
-            creator=pymysql,
-            maxconnections=pool_size,
-            mincached=pool_min,
-            maxcached=pool_size,
-            maxusage=200,
-            blocking=True,
-            max_idle_time=pool_recycle,
-            connection_timeout=10,
-            charset="utf8mb4",
-            cursorclass=pymysql.cursors.DictCursor,
-            host=host,
-            port=port,
-            database=database,
-            user=user,
-            password=password,
-            read_timeout=30,
-            write_timeout=30,
-        )
+        # PooledDB parameters
+        pool_config = {
+            "creator": pymysql,
+            "maxconnections": pool_size,
+            "mincached": pool_min,
+            "maxcached": pool_size,
+            "maxusage": 200,
+            "blocking": True,
+            # ping(0) = check if connection is alive BEFORE returning it from
+            # the pool. This is the correct DBUtils mechanism for detecting
+            # dead/stale connections (idle timeout, network blip, DB restart).
+            "ping": 0,
+        }
+
+        # PyMySQL connection parameters
+        connection_params = {
+            "host": host,
+            "port": port,
+            "database": database,
+            "user": user,
+            "password": password,
+            "charset": "utf8mb4",
+            "cursorclass": pymysql.cursors.DictCursor,
+            "read_timeout": 30,
+            "write_timeout": 30,
+        }
+
+        self._pool = PooledDB(**pool_config, **connection_params)
+        self._pool_recycle = pool_recycle
         logger.info(
-            "DB pool initialized: host=%s db=%s pool_size=%d min=%d",
+            "DB pool initialized: host=%s db=%s pool_size=%d min=%d ping=0",
             host, database, pool_size, pool_min,
         )
 
@@ -61,10 +78,85 @@ class DBConnector:
         """Get a connection from the pool."""
         return self._pool.connection()
 
+    def _run_query(self, query: str, params, transaction: bool, fetch: str):
+        """Run a single query on a fresh connection, with retry on transient errors.
+
+        A pooled connection can be dropped by the server (idle timeout, network
+        blip, DB restart). When that happens the query fails with an
+        InterfaceError/OperationalError. We discard the dead connection and
+        retry with a fresh one from the pool.
+
+        Args:
+            query: SQL statement.
+            params: Bound parameters (or None).
+            transaction: If True, wrap the query in a transaction.
+            fetch: One of "none", "one", or "all" controlling the return value.
+
+        Returns:
+            - "none": the affected row count.
+            - "one": a single row (dict) or None.
+            - "all": a list of rows (dicts).
+        """
+        last_exc = None
+        for attempt in range(1, self._MAX_RETRIES + 1):
+            conn = self.get_connection()
+            try:
+                cursor = conn.cursor()
+                if transaction:
+                    conn.begin()
+                cursor.execute(query, params or ())
+                if fetch == "one":
+                    result = cursor.fetchone()
+                elif fetch == "all":
+                    result = cursor.fetchall()
+                else:
+                    result = cursor.rowcount
+                if transaction:
+                    conn.commit()
+                cursor.close()
+                conn.close()
+                return result
+            except self._RETRYABLE_ERRORS as e:
+                last_exc = e
+                logger.warning(
+                    "Transient DB error (attempt %d/%d): %s",
+                    attempt, self._MAX_RETRIES, e,
+                )
+                # Discard the dead connection; a fresh one is taken on retry.
+                self._safe_close(conn)
+                if attempt < self._MAX_RETRIES:
+                    time.sleep(self._RETRY_BACKOFF_SECONDS * attempt)
+            except Exception:
+                # Non-transient error: roll back if in a transaction and re-raise.
+                if transaction:
+                    self._safe_rollback(conn)
+                self._safe_close(conn)
+                raise
+        raise last_exc
+
+    @staticmethod
+    def _safe_close(conn):
+        try:
+            conn.close()
+        except Exception:
+            pass
+
+    @staticmethod
+    def _safe_rollback(conn):
+        try:
+            conn.rollback()
+        except Exception:
+            pass
+
     @contextlib.contextmanager
     def get_cursor(self, transaction: bool = False):
-        """Context manager for getting a cursor with optional transaction support."""
+        """Context manager for getting a cursor with optional transaction support.
+
+        Note: this does not retry on transient errors; use execute/fetchone/
+        fetchall for retry-safe single-statement queries.
+        """
         conn = self.get_connection()
+        cursor = None
         try:
             cursor = conn.cursor()
             if transaction:
@@ -74,11 +166,15 @@ class DBConnector:
                 conn.commit()
         except Exception:
             if transaction:
-                conn.rollback()
+                self._safe_rollback(conn)
             raise
         finally:
+            if cursor is not None:
+                try:
                     cursor.close()
-            conn.close()
+                except Exception:
+                    pass
+            self._safe_close(conn)
 
     @contextlib.contextmanager
     def transaction(self):
@@ -89,28 +185,22 @@ class DBConnector:
             yield conn
             conn.commit()
         except Exception:
-            conn.rollback()
+            self._safe_rollback(conn)
             raise
         finally:
-            conn.close()
+            self._safe_close(conn)
 
     def execute(self, query: str, params=None, transaction: bool = False):
-        """Execute a query and return affected rows."""
-        with self.get_cursor(transaction=transaction) as cursor:
-            cursor.execute(query, params or ())
-            return cursor.rowcount
+        """Execute a query and return affected rows. Retries on transient errors."""
+        return self._run_query(query, params, transaction, fetch="none")
 
     def fetchone(self, query: str, params=None):
-        """Execute a query and return one row."""
-        with self.get_cursor() as cursor:
-            cursor.execute(query, params or ())
-            return cursor.fetchone()
+        """Execute a query and return one row. Retries on transient errors."""
+        return self._run_query(query, params, transaction=False, fetch="one")
 
     def fetchall(self, query: str, params=None):
-        """Execute a query and return all rows."""
-        with self.get_cursor() as cursor:
-            cursor.execute(query, params or ())
-            return cursor.fetchall()
+        """Execute a query and return all rows. Retries on transient errors."""
+        return self._run_query(query, params, transaction=False, fetch="all")
 
     def initialize_schema(self, schema_path: str = "db/schema.sql"):
         """Initialize the database schema from SQL file."""

src/logging_config.py

@@ -14,8 +14,6 @@ import logging.handlers
 import sys
 from datetime import datetime, timezone
 
-from python_json_logger import json_formatter
-
 
 def setup_logging(
     level: str = "INFO",

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/prober.py

@@ -7,6 +7,7 @@ Handles errors gracefully for corrupt or unsupported files.
 
 import json
 import logging
+import os
 import subprocess
 import time
 from dataclasses import dataclass, field

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",
-                     self.input_path, self.scan_interval, self.walker_threads)
+        logger.info(
+            "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,
         }

worker/Dockerfile

@@ -35,7 +35,8 @@ RUN groupadd -g 1000 appuser && \
 WORKDIR /app
 
 # Create necessary directories
-RUN mkdir -p /scratch /models /data/training /data/output /logs
+RUN mkdir -p /scratch /models /data/training /data/output /logs && \
+    chown -R appuser:appuser /app /scratch /models /data /logs
 
 # Copy requirements first for better caching
 COPY worker/requirements.txt /app/requirements.txt