From d693569f59375ac1609bb9e5c88b4f2276a0e5c0 Mon Sep 17 00:00:00 2001 From: dtoro Date: Sun, 12 Apr 2026 22:24:38 +0200 Subject: [PATCH] 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) --- backend/app/routers/library.py | 13 +++ backend/app/services/duplicates.py | 4 +- backend/app/services/scanner.py | 23 +++-- backend/app/tasks/scan.py | 140 +++++++++++++++++------------ 4 files changed, 112 insertions(+), 68 deletions(-) diff --git a/backend/app/routers/library.py b/backend/app/routers/library.py index 68c1c49..530f449 100644 --- a/backend/app/routers/library.py +++ b/backend/app/routers/library.py @@ -840,4 +840,17 @@ async def trigger_backfill_phashes(current_user: User = Depends(get_current_user return {"status": "queued"} except Exception as e: logger.error(f"Backfill queue failed: {e}") + return {"status": "error", "message": str(e)} + + +@router.post("/maintenance/start-watcher") +async def start_file_watcher(current_user: User = Depends(get_current_user)): + """Start the filesystem watcher. Uses a Redis lock so only one + instance runs at a time — safe to call repeatedly.""" + from app.tasks.scan import watch_folders + try: + watch_folders.apply_async(countdown=2) + return {"status": "queued"} + except Exception as e: + logger.error(f"Watcher queue failed: {e}") return {"status": "error", "message": str(e)} \ No newline at end of file diff --git a/backend/app/services/duplicates.py b/backend/app/services/duplicates.py index c2cf0e7..3f4ae13 100644 --- a/backend/app/services/duplicates.py +++ b/backend/app/services/duplicates.py @@ -125,7 +125,7 @@ async def regroup_duplicates( Idempotent — safe to call as often as you like. Returns a summary dict. """ - embedder_model = settings.mulita_config.vision.embedder.name + embedder_model = settings.vision.embedder.name async with AsyncSessionLocal() as session: # Pull all visible photos with a phash or embedding. @@ -210,7 +210,7 @@ async def incremental_regroup( Much faster than a full regroup for post-scan updates: O(new × log N) via HNSW instead of O(N²). """ - embedder_model = settings.mulita_config.vision.embedder.name + embedder_model = settings.vision.embedder.name async with AsyncSessionLocal() as session: # If no watermark, fall back to full regroup. diff --git a/backend/app/services/scanner.py b/backend/app/services/scanner.py index 6c3ccab..853e75e 100644 --- a/backend/app/services/scanner.py +++ b/backend/app/services/scanner.py @@ -78,17 +78,26 @@ async def bootstrap_default_source_root() -> None: async def start_initial_scan(): - """Start the initial library scan. + """Start the initial library scan and optionally the file watcher. - NOTE: the folder watcher (watch_folders task) is intentionally NOT - dispatched here. It's an infinite loop celery task and every backend - restart was queuing a new instance, eventually pinning every worker - and starving scan_folder dispatches. Re-enabling it needs a Redis - lock or a dedicated long-running container — until then the user - triggers scans manually via "Scan all folders". + The file watcher uses a Redis lock to ensure only one instance runs + across all workers, so it's safe to dispatch on every startup — only + the first one will actually watch, the rest exit immediately. """ try: scan_all_source_roots.delay() logger.info("Initial scan queued successfully") except Exception as e: logger.error(f"Failed to start initial scan: {e}") + + # Start the file watcher if enabled in config. + from app.config import settings + if settings.scanner.watch: + try: + from app.tasks.scan import watch_folders + # Countdown gives the initial scan time to register source roots + # before the watcher tries to load them. + watch_folders.apply_async(countdown=10) + logger.info("File watcher queued (Redis-locked, single instance)") + except Exception as e: + logger.warning(f"Could not queue file watcher: {e}") diff --git a/backend/app/tasks/scan.py b/backend/app/tasks/scan.py index 614308d..8fcbb55 100644 --- a/backend/app/tasks/scan.py +++ b/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"""