From f657e2c0ba521d0a357f7df7cf1ce3d011aa2728 Mon Sep 17 00:00:00 2001 From: Claudio Date: Mon, 11 May 2026 12:28:36 +0200 Subject: [PATCH] feat: retire the watchfiles watcher in favour of NC webhooks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit End-to-end webhook flow is proven on this NC instance (NodeCreated + NodeWritten both fired and dispatched scan_folder on a PUT test), so the watchfiles-based polling layer is no longer needed. - scanner.start_initial_scan no longer queues watch_folders on boot. - scan.watch_folders kept as a one-line no-op shim so any leftover apply_async in flight from the previous deploy doesn't crash a worker. Will be deleted entirely after the queue drains. - celery.py reroutes watch_folders to the `default` queue (worker-light) so the no-op shim actually completes — the `watcher` queue is dead. - docker-compose drops the mulita-worker-watcher service. Its celery --beat responsibility (firing discard_missing_photos_beat every 30 min) moves to worker-light's command. Latency note: NC dispatches webhook events through its background-job queue, currently run by cron */5. After this commit lands you'll want to tighten cron to */1 so new uploads land in mule within ~60s instead of up to 5 min. --- backend/app/services/scanner.py | 22 ++----- backend/app/tasks/celery.py | 7 ++- backend/app/tasks/scan.py | 104 ++++---------------------------- docker-compose.yml | 55 +++-------------- 4 files changed, 31 insertions(+), 157 deletions(-) diff --git a/backend/app/services/scanner.py b/backend/app/services/scanner.py index 853e75e..f0cce1c 100644 --- a/backend/app/services/scanner.py +++ b/backend/app/services/scanner.py @@ -78,26 +78,16 @@ async def bootstrap_default_source_root() -> None: async def start_initial_scan(): - """Start the initial library scan and optionally the file watcher. + """Start the initial library scan. - 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. + The watchfiles-based watcher has been retired in favour of Nextcloud + `webhook_listeners` (see `app.routers.nc_webhook`). NC POSTs every + file event directly to mule, so we no longer keep a long-running + inotify task. The periodic `discard_missing_photos_beat` Celery job + is still there as a safety net for deletions a webhook might miss. """ 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/celery.py b/backend/app/tasks/celery.py index f4f7c80..7439001 100644 --- a/backend/app/tasks/celery.py +++ b/backend/app/tasks/celery.py @@ -52,8 +52,11 @@ celery_app.conf.update( 'scan_folder': {'queue': 'low'}, 'scan_all_source_roots': {'queue': 'low'}, 'backfill_gps': {'queue': 'low'}, - # Dedicated watcher queue - 'watch_folders': {'queue': 'watcher'}, + # `watch_folders` is retired (file events come from NC webhooks) + # but the task definition still exists as a no-op shim for any + # in-flight apply_async. Route it to the default queue so the + # remaining worker actually drains it. + 'watch_folders': {'queue': 'default'}, 'discard_missing_photos_beat': {'queue': 'low'}, }, task_default_queue='default', diff --git a/backend/app/tasks/scan.py b/backend/app/tasks/scan.py index 55507e8..455fcfb 100644 --- a/backend/app/tasks/scan.py +++ b/backend/app/tasks/scan.py @@ -473,100 +473,20 @@ async def _scan_all_source_roots_async(): # again (e.g. another extractor-logic fix lands). -WATCHER_LOCK_KEY = "mulita:watch_folders:lock" -WATCHER_LOCK_TTL = 60 # 1 min — renewed every event batch via wall-clock check - - -@shared_task(name='watch_folders', bind=True, soft_time_limit=0, time_limit=0) +@shared_task(name='watch_folders', bind=True) def watch_folders(self): + """Retired: file events now arrive via NC webhook_listeners. + + Kept as a no-op task so any in-flight queue items (a leftover + apply_async from a restart before this commit, or an admin-button + trigger) don't crash workers. Will be removed entirely once the + queue drains. """ - 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 - - 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: - 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 - - import time - last_renew = time.monotonic() - for changes in watch(*paths, rust_timeout=30_000): - # Renew the Redis lock on a wall-clock schedule (every 30s) - # instead of every N events, so quiet directories don't let - # the lock expire. watchfiles' rust_timeout ensures we wake - # at least every 30s even with no FS events. - now = time.monotonic() - if now - last_renew >= 30: - try: - lock.extend(WATCHER_LOCK_TTL) - last_renew = now - except Exception: - logger.warning("watch_folders: failed to renew Redis lock") - - for change_type, filepath in changes: - filepath = str(filepath) - - if Path(filepath).suffix.lower() not in SUPPORTED_EXTENSIONS: - continue - - 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: - logger.warning("watch_folders: could not release Redis lock (may have expired)") + logger.info( + "watch_folders task is retired; file events come from NC " + "webhook_listeners. No-op." + ) + return {'status': 'retired'} async def handle_file_deletion(filepath: str): """Handle deletion of a file from the filesystem""" diff --git a/docker-compose.yml b/docker-compose.yml index 0a6db08..9cba911 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -144,7 +144,7 @@ services: dockerfile: Dockerfile image: mule-image-worker container_name: mulita-worker-light - command: sh -c "python -m app.services.vision.bootstrap_models && celery -A app.tasks.celery worker --loglevel=${LOG_LEVEL:-info} --concurrency=${CELERY_LIGHT_CONCURRENCY:-2} -Q default,high,low -n light@%h" + command: sh -c "python -m app.services.vision.bootstrap_models && celery -A app.tasks.celery worker --beat --loglevel=${LOG_LEVEL:-info} --concurrency=${CELERY_LIGHT_CONCURRENCY:-2} -Q default,high,low -n light@%h" volumes: - ./mulita.yml:/app/config/mulita.yml:ro - ${PHOTO_DIRS:-./photos}:/photos:rw @@ -191,52 +191,13 @@ services: - "cloud.hubris.network:192.168.8.175" restart: unless-stopped - # Dedicated watcher worker — runs the long-lived watch_folders task - # on its own queue so it never blocks scan/thumbnail workers. - worker-watcher: - build: - context: ./backend - dockerfile: Dockerfile - image: mule-image-worker - container_name: mulita-worker-watcher - # --beat runs the celery beat scheduler in-process alongside the - # watcher worker — there's only ever one watcher (Redis-locked - # singleton) and we don't need a separate container just to fire a - # 30-minute periodic task. Beat schedule lives in app/tasks/celery.py. - command: sh -c "celery -A app.tasks.celery worker --beat --loglevel=${LOG_LEVEL:-info} --concurrency=1 -Q watcher -n watcher@%h" - volumes: - - ./mulita.yml:/app/config/mulita.yml:ro - - ${PHOTO_DIRS:-./photos}:/photos:rw - - ${NEXTCLOUD_USERS_HOST_PATH:-./photos}:/nextcloud-users:rw - - db_data:/data/db - environment: - - DATABASE_URL=postgresql+asyncpg://mulita:mulita@db:5432/mulita - - REDIS_URL=redis://redis:6379 - - CELERY_BROKER_URL=redis://redis:6379 - - CELERY_RESULT_BACKEND=redis://redis:6379 - - PHOTO_DIRS=/photos - - NEXTCLOUD_USERS_ROOT=${NEXTCLOUD_USERS_ROOT:-/nextcloud-users} - - NEXTCLOUD_BASE_URL=${NEXTCLOUD_BASE_URL:-} - - NEXTCLOUD_WEBHOOK_SECRET=${NEXTCLOUD_WEBHOOK_SECRET:-} - - LOG_LEVEL=${LOG_LEVEL:-INFO} - - TZ=${TZ:-UTC} - - MULITA_CELERY_WORKER=1 - depends_on: - redis: - condition: service_started - db: - condition: service_healthy - networks: - - mulita-network - # Pin cloud.hubris.network to the LAN caddy IP. Without this, the - # docker DNS forwards the lookup to the host's resolver, which - # returns the public IONOS VPS IP — but cloud isn't in the VPS - # traefik exposure list, so TLS handshakes against it die with - # "unexpected eof while reading". Caddy on 192.168.8.175 holds the - # cloud.hubris.network cert and proxies to the Nextcloud LXC. - extra_hosts: - - "cloud.hubris.network:192.168.8.175" - restart: unless-stopped + # worker-watcher used to live here — it ran the long-lived + # watchfiles-based `watch_folders` task plus celery `--beat`. Both + # responsibilities moved on Phase 2: + # * file events now come from NC's webhook_listeners → POST + # /api/v1/internal/nc-webhook (see backend/app/routers/nc_webhook.py) + # * `--beat` was folded into worker-light's command so the + # periodic discard_missing_photos_beat job still fires. worker-vision: build: