Files
mule-image/backend/app/tasks/celery.py
Claudio 09a00f7419 feat(nextcloud): hard-delete SourceRoot + reliable delete sync
Two related fixes for the Nextcloud library lifecycle.

1. DELETE /api/v1/nextcloud/source-roots/{id} now actually deletes
   the SourceRoot, every Folder under it, and every Photo in those
   folders (Nextcloud files untouched). Was a soft-deactivate
   (is_active=false) that left the rows around forever, so re-adding
   the same path resurrected ghosts and prune-missing reported zero.
   Returns {deleted_photos, deleted_folders}; the Settings UI toasts
   the count and invalidates photos/folders/stats so cached lists
   don't show ghosts. photo_tags and heap_photos already cascade via
   ON DELETE CASCADE; FolderShare uses a stringly-typed folder_id
   with no FK so cleaned up explicitly.

2. The watcher (watch_folders task) was getting killed every five
   minutes by the global task_soft_time_limit=300 in app/tasks/celery.py
   despite passing soft_time_limit=None on the decorator (None falls
   back to the worker default in this Celery version). Override with
   soft_time_limit=0, time_limit=0 (= unlimited) so the watch loop
   actually stays alive. The 'Soft time limit (300s) exceeded' /
   'Worker exited prematurely' lines should stop in worker-watcher
   logs.

3. Added discard_missing_photos() in services/cleanup.py — a soft
   variant of prune_missing_photos that walks every present source
   root, checks os.path.exists for each non-discarded Photo, and
   flips is_discarded=true on the missing ones (UPDATE not DELETE).
   Wired as discard_missing_photos_beat in tasks/scan.py and
   scheduled every 30 min via celery beat. Beat runs in-process on
   worker-watcher (--beat flag in compose) — there's only ever one
   watcher and we don't need a separate container.

Hard delete remains manual via prune-missing for users who want to
review before committing. The beat catch-up only soft-discards (file
gone -> mule-image trash, restorable).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-10 22:19:58 +02:00

91 lines
3.2 KiB
Python

"""
Celery configuration and app initialization
"""
import logging
import os
from celery import Celery
from celery.signals import worker_process_init
from app.config import settings
logger = logging.getLogger(__name__)
# Create Celery app
celery_app = Celery(
'mulita',
broker=settings.celery_broker_url,
backend=settings.celery_result_backend,
include=[
'app.tasks.scan',
'app.tasks.thumbs',
'app.tasks.vision',
'app.services.metadata', # extract_metadata lives here
]
)
# Configure Celery
celery_app.conf.update(
task_serializer='json',
accept_content=['json'],
result_serializer='json',
timezone='UTC',
enable_utc=True,
# Robust acknowledgment: keep message in broker until task succeeds.
task_acks_late=True,
task_reject_on_worker_lost=True,
# Global time limits — individual tasks can override via decorator.
task_soft_time_limit=300, # 5 min — raises SoftTimeLimitExceeded
task_time_limit=600, # 10 min — SIGKILL
# Explicit routes for every task name. Wildcard patterns don't match
# short names produced by @shared_task(name='...').
task_routes={
# Vision queue — CPU-bound binary classification
'classify_content': {'queue': 'vision'},
'vision_fanout': {'queue': 'vision'},
# High-priority queue — thumbnails & duplicates
'generate_thumbnails': {'queue': 'high'},
'regenerate_all_thumbnails': {'queue': 'high'},
'backfill_phashes': {'queue': 'high'},
'regroup_duplicates': {'queue': 'high'},
'incremental_regroup_duplicates': {'queue': 'high'},
# Low-priority queue — scans
'scan_folder': {'queue': 'low'},
'scan_all_source_roots': {'queue': 'low'},
'backfill_gps': {'queue': 'low'},
# Dedicated watcher queue
'watch_folders': {'queue': 'watcher'},
'discard_missing_photos_beat': {'queue': 'low'},
},
task_default_queue='default',
task_default_exchange='default',
task_default_exchange_type='direct',
task_default_routing_key='default',
broker_connection_retry_on_startup=True,
# Periodic catch-up so external file deletions in Nextcloud get
# reflected even when the real-time watcher missed the event
# (worker restart window, mount transient, etc).
beat_schedule={
'discard-missing-photos-every-30min': {
'task': 'discard_missing_photos_beat',
'schedule': 30 * 60,
},
},
)
@worker_process_init.connect
def _warmup_vision_models(**kwargs):
"""Pre-load vision models in the worker process so the first task
doesn't pay cold-start latency. Only runs on the vision queue."""
# The worker name contains the queue — only warm up vision workers.
worker_queues = os.environ.get("CELERY_QUEUES", "")
if "vision" not in worker_queues:
# Heuristic: check the celery command line for -Q vision
import sys
if "vision" not in " ".join(sys.argv):
return
try:
from app.services.vision.registry import registry
registry.warmup()
except Exception:
logger.exception("Vision model warmup failed")