feat: auto-start file watcher with Redis lock for live import

Re-enable the watchfiles-based folder watcher with a Redis lock to
prevent multiple instances from stacking up across restarts. The
watcher is now automatically dispatched on startup when scanner.watch
is true (default), and only one instance runs at a time.

- Redis lock (SETNX + TTL renewal) ensures single-instance execution
- Graceful exit if another watcher holds the lock
- New POST /maintenance/start-watcher endpoint for manual control
- Fix: use settings.scanner/vision properties instead of mulita_config

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

backend/app/tasks/scan.py

@@ -468,74 +468,96 @@ async def _scan_all_source_roots_async():
             logger.warning(f"Could not queue post-scan face recluster: {e}")
 
 
-@shared_task(name='watch_folders')
-def watch_folders():
+WATCHER_LOCK_KEY = "mulita:watch_folders:lock"
+WATCHER_LOCK_TTL = 300  # 5 min — renewed every 60s
+
+
+@shared_task(name='watch_folders', bind=True)
+def watch_folders(self):
     """
     Watch folders for changes using watchfiles. Long-running task that
     monitors filesystem events under every active source root.
+
+    Uses a Redis lock to ensure only one instance runs across all
+    workers. The lock is renewed periodically so it survives restarts
+    without leaving orphan watchers.
     """
+    import redis as redis_lib
     from watchfiles import watch
 
-    # Read source roots from the DB instead of the (now-removed) YAML
-    # config. We need both the path and the id so we can dispatch
-    # scan_folder with the source_root_id when an event fires.
-    roots: list[tuple[str, str]] = []
+    r = redis_lib.from_url(settings.redis_url)
+
+    # Acquire exclusive lock — if another watcher is already running,
+    # this instance exits immediately instead of stacking up.
+    lock = r.lock(WATCHER_LOCK_KEY, timeout=WATCHER_LOCK_TTL)
+    if not lock.acquire(blocking=False):
+        logger.info("watch_folders: another instance is already running, exiting")
+        return {'status': 'skipped', 'reason': 'another instance is running'}
+
     try:
-        async def _load_roots():
-            async with AsyncSessionLocal() as session:
-                result = await session.execute(
-                    select(SourceRoot).where(SourceRoot.is_active == True)  # noqa: E712
-                )
-                return [
-                    (os.path.normpath(sr.path), sr.id)
-                    for sr in result.scalars().all()
-                    if os.path.exists(sr.path)
-                ]
-        roots = asyncio.run(_load_roots())
-    except Exception as e:
-        logger.error(f"watch_folders could not load source roots: {e}")
-        return
-
-    if not roots:
-        logger.warning("No valid source roots to watch")
-        return
-
-    paths = [p for p, _ in roots]
-    logger.info(f"Starting folder watcher for: {paths}")
-
-    def find_source_root_for(path: str) -> Optional[str]:
-        """Return the source_root id whose path contains `path`, or None."""
-        normalized = os.path.normpath(path)
-        for root_path, root_id in roots:
-            if normalized == root_path or normalized.startswith(root_path + os.sep):
-                return root_id
-        return None
-
-    for changes in watch(*paths):
-        for change_type, filepath in changes:
-            filepath = str(filepath)
-
-            # Check if it's a supported file type
-            if Path(filepath).suffix.lower() not in SUPPORTED_EXTENSIONS:
-                continue
-
-            if change_type == 'added' or change_type == 'modified':
-                # Queue scan for the parent folder, with the source_root_id
-                # resolved by ancestor lookup so scan_folder doesn't
-                # auto-create a new SourceRoot for an arbitrary subdir.
-                parent_dir = str(Path(filepath).parent)
-                source_root_id = find_source_root_for(parent_dir)
-                if source_root_id is None:
-                    logger.debug(
-                        f"watcher event for {filepath}: parent {parent_dir} "
-                        f"not under any active source root, ignoring"
+        roots: list[tuple[str, str]] = []
+        try:
+            async def _load_roots():
+                async with AsyncSessionLocal() as session:
+                    result = await session.execute(
+                        select(SourceRoot).where(SourceRoot.is_active == True)  # noqa: E712
                     )
+                    return [
+                        (os.path.normpath(sr.path), sr.id)
+                        for sr in result.scalars().all()
+                        if os.path.exists(sr.path)
+                    ]
+            roots = asyncio.run(_load_roots())
+        except Exception as e:
+            logger.error(f"watch_folders could not load source roots: {e}")
+            return
+
+        if not roots:
+            logger.warning("No valid source roots to watch")
+            return
+
+        paths = [p for p, _ in roots]
+        logger.info(f"Starting folder watcher for: {paths}")
+
+        def find_source_root_for(path: str) -> Optional[str]:
+            """Return the source_root id whose path contains `path`, or None."""
+            normalized = os.path.normpath(path)
+            for root_path, root_id in roots:
+                if normalized == root_path or normalized.startswith(root_path + os.sep):
+                    return root_id
+            return None
+
+        renew_counter = 0
+        for changes in watch(*paths):
+            # Renew the Redis lock periodically so it doesn't expire
+            # while the watcher is idle between events.
+            renew_counter += 1
+            if renew_counter % 10 == 0:
+                try:
+                    lock.extend(WATCHER_LOCK_TTL)
+                except Exception:
+                    pass
+
+            for change_type, filepath in changes:
+                filepath = str(filepath)
+
+                if Path(filepath).suffix.lower() not in SUPPORTED_EXTENSIONS:
                     continue
-                scan_folder.delay(parent_dir, source_root_id)
-                logger.info(f"File {change_type}: {filepath}, queued scan for {parent_dir}")
-            elif change_type == 'deleted':
-                # Handle file deletion
-                asyncio.run(handle_file_deletion(filepath))
+
+                if change_type == 'added' or change_type == 'modified':
+                    parent_dir = str(Path(filepath).parent)
+                    source_root_id = find_source_root_for(parent_dir)
+                    if source_root_id is None:
+                        continue
+                    scan_folder.delay(parent_dir, source_root_id)
+                    logger.info(f"File {change_type}: {filepath}, queued scan for {parent_dir}")
+                elif change_type == 'deleted':
+                    asyncio.run(handle_file_deletion(filepath))
+    finally:
+        try:
+            lock.release()
+        except Exception:
+            pass
 
 async def handle_file_deletion(filepath: str):
     """Handle deletion of a file from the filesystem"""