dtoro/mule-image
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: harden pipeline — retries, acks_late, time limits, session safety

Browse Source 
Addresses 16 robustness, transparency, and performance issues across
the Celery media processing pipeline:

Critical:
- Singleton DB engine in vision tasks (was leaking one per task call)
- acks_late + task_reject_on_worker_lost so crashed workers don't lose tasks
- Global soft/hard time limits (5/10 min) to prevent hung worker slots
- Thumbnail copy-before-resize (in-place mutation degraded larger sizes)
- backfill_vision now checks each task type independently (OCR, faces, etc.)
- Parameterized LIMIT in backfill_vision (was f-string SQL injection)

High:
- try/except + retry(max=3) on all vision inference tasks
- extract_metadata writes processing_error on exiftool failure
- PIL Image handles closed in _load_thumb/_load_original
- Scan progress Redis keys auto-expire after 1 hour
- Watcher lock renewal is wall-clock based (30s) not event-count based
- worker_process_init signal warms up vision models on startup

Medium:
- Explicit task_routes for every task name (wildcards never matched)
- app.services.metadata added to Celery include list
- POST /maintenance/recover-stuck endpoint for photos stuck in processing
- Docker healthchecks for worker-light, worker-vision, and Redis
- Task ID in vision log lines for distributed tracing
- Bare except:pass narrowed to specific exceptions

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
root
2026-04-13 08:57:10 +02:00
parent e974ffbfd2
commit f090a809a9
8 changed files with 366 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
backend/app/tasks/celery.py
View File

@@ -1,15 +1,26 @@
"""
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']
include=[
'app.tasks.scan',
'app.tasks.thumbs',
'app.tasks.vision',
'app.services.metadata', # extract_metadata lives here
]
)
# Configure Celery
@@ -19,16 +30,34 @@ celery_app.conf.update(
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={
'app.tasks.thumbs.*': {'queue': 'high'},
'app.tasks.scan.*': {'queue': 'low'},
'app.tasks.vision.*': {'queue': 'vision'},
# Vision queue — GPU/CPU-bound inference
'embed_photo': {'queue': 'vision'},
'ocr_photo': {'queue': 'vision'},
'detect_objects': {'queue': 'vision'},
'extract_faces': {'queue': 'vision'},
'classify_content': {'queue': 'vision'},
'vision_fanout': {'queue': 'vision'},
'recluster_faces': {'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'},
},
task_default_queue='default',
@@ -36,4 +65,22 @@ celery_app.conf.update(
task_default_exchange_type='direct',
task_default_routing_key='default',
broker_connection_retry_on_startup=True,
)
)
@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")

36
backend/app/tasks/scan.py
View File

@@ -95,11 +95,13 @@ async def _scan_folder_async(folder_path: str, source_root_id: Optional[str], ta
r = _get_redis()
PROGRESS_TTL = 3600 # 1 hour — auto-expire if scan crashes
def progress_set(key: str, value) -> None:
if r is None:
return
try:
r.set(key, str(value))
r.set(key, str(value), ex=PROGRESS_TTL)
except Exception as e:
logger.debug(f"scan progress set failed: {e}")
@@ -467,12 +469,20 @@ async def _scan_all_source_roots_async():
except Exception as e:
logger.warning(f"Could not queue post-scan face recluster: {e}")
# Re-extract metadata for photos missing GPS coordinates.
# Runs on every startup so photos scanned before the GPS fix
# eventually get their coordinates populated.
try:
backfill_gps.apply_async(countdown=30)
except Exception as e:
logger.warning(f"Could not queue post-scan GPS backfill: {e}")
WATCHER_LOCK_KEY = "mulita:watch_folders:lock"
WATCHER_LOCK_TTL = 300 # 5 min — renewed every 60s
WATCHER_LOCK_TTL = 60 # 1 min — renewed every event batch via wall-clock check
@shared_task(name='watch_folders', bind=True)
@shared_task(name='watch_folders', bind=True, soft_time_limit=None, time_limit=None)
def watch_folders(self):
"""
Watch folders for changes using watchfiles. Long-running task that
@@ -527,16 +537,20 @@ def watch_folders(self):
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:
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:
pass
logger.warning("watch_folders: failed to renew Redis lock")
for change_type, filepath in changes:
filepath = str(filepath)
@@ -557,7 +571,7 @@ def watch_folders(self):
try:
lock.release()
except Exception:
pass
logger.warning("watch_folders: could not release Redis lock (may have expired)")
async def handle_file_deletion(filepath: str):
"""Handle deletion of a file from the filesystem"""

25
backend/app/tasks/thumbs.py
View File

@@ -218,23 +218,28 @@ def auto_rotate_image(image: Image.Image) -> Image.Image:
if orientation in rotation_map:
image = image.rotate(rotation_map[orientation], expand=True)
except:
except (AttributeError, KeyError, TypeError):
pass # No orientation data available
return image
def generate_thumbnail(image: Image.Image, size: int, output_path: str):
"""Generate a thumbnail of the specified size"""
# Maintain aspect ratio
image.thumbnail((size, size), Image.Resampling.LANCZOS)
# Save as WebP with specified quality
image.save(
"""Generate a thumbnail of the specified size.
Works on a copy so the caller's image is never mutated — this is
critical because the thumbnail loop iterates multiple sizes and
in-place shrinking would degrade later (larger) sizes.
"""
img = image.copy()
img.thumbnail((size, size), Image.Resampling.LANCZOS)
img.save(
output_path,
'WEBP',
quality=settings.thumbnails.quality,
method=4 # Balance between speed and compression
)
img.close()
@shared_task(bind=True, name='generate_thumbnails')
def generate_thumbnails(self, photo_id: str):
@@ -325,11 +330,11 @@ async def _generate_thumbnails_async(photo_id: str, task):
photo.processing_status = 'completed'
photo.processing_error = None
await session.commit()
logger.info(f"Thumbnails generated for photo {photo_id}")
# Dispatch vision pipeline (embedding, OCR, detection, faces)
# after thumbs are ready so vision tasks have images to read.
# Dispatch vision pipeline only after thumbnails succeeded —
# vision tasks need the generated thumbnails to run inference.
try:
from app.tasks.vision import vision_fanout
vision_fanout.delay(photo_id)

270
backend/app/tasks/vision.py
View File

@@ -23,12 +23,33 @@ from app.config import settings
logger = logging.getLogger(__name__)
VISION_READY_KEY = "mulita:vision:ready"
def _vision_worker_ready() -> bool:
"""Check whether the vision worker has finished model bootstrap."""
try:
import redis as _redis
return bool(_redis.from_url(settings.redis_url).exists(VISION_READY_KEY))
except Exception:
return False
_sync_engine = None
def _get_sync_engine():
"""Return a module-level singleton engine (one per worker process)."""
global _sync_engine
if _sync_engine is None:
sync_url = settings.database_url.replace("+asyncpg", "+psycopg2").replace("+aiosqlite", "")
_sync_engine = create_engine(sync_url, pool_pre_ping=True, pool_size=3, max_overflow=5)
return _sync_engine
def _get_sync_session() -> Session:
"""Create a sync DB session for use in Celery workers."""
sync_url = settings.database_url.replace("+asyncpg", "+psycopg2").replace("+aiosqlite", "")
engine = create_engine(sync_url, pool_pre_ping=True)
return sessionmaker(bind=engine)()
"""Create a sync DB session backed by the shared engine."""
return sessionmaker(bind=_get_sync_engine())()
def _load_thumb(photo_id: str, size: str = "medium") -> np.ndarray | None:
@@ -37,12 +58,19 @@ def _load_thumb(photo_id: str, size: str = "medium") -> np.ndarray | None:
if not thumb_path.exists():
logger.warning("Thumbnail not found: %s", thumb_path)
return None
img = Image.open(thumb_path).convert("RGB")
return np.array(img)
try:
img = Image.open(thumb_path).convert("RGB")
img.load() # force decode to catch corruption early
arr = np.array(img)
img.close()
return arr
except Exception as e:
logger.warning("Corrupt or unreadable thumbnail for %s: %s", photo_id, e)
return None
@shared_task(name='embed_photo', queue='vision')
def embed_photo(photo_id: str):
@shared_task(name='embed_photo', queue='vision', bind=True, max_retries=3)
def embed_photo(self, photo_id: str):
"""Generate CLIP embedding for a photo and store in pgvector."""
if not settings.vision.enabled:
return {'status': 'skipped', 'reason': 'vision disabled'}
@@ -51,9 +79,13 @@ def embed_photo(photo_id: str):
if image is None:
return {'status': 'error', 'message': 'thumbnail not found'}
from app.services.vision.registry import registry
embedder = registry.get_embedder()
vector = embedder.embed_image(image)
try:
from app.services.vision.registry import registry
embedder = registry.get_embedder()
vector = embedder.embed_image(image)
except Exception as exc:
logger.exception("embed_photo failed for %s", photo_id)
raise self.retry(exc=exc, countdown=60)
model_name = settings.vision.embedder.name
@@ -72,10 +104,13 @@ def embed_photo(photo_id: str):
)
session.add(emb)
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
logger.info("Embedded photo %s with %s", photo_id, model_name)
logger.info("[%s] Embedded photo %s with %s", self.request.id, photo_id, model_name)
return {'status': 'success', 'photo_id': photo_id}
@@ -99,8 +134,8 @@ def vision_fanout(photo_id: str):
return {'status': 'dispatched', 'photo_id': photo_id}
@shared_task(name='ocr_photo', queue='vision')
def ocr_photo(photo_id: str):
@shared_task(name='ocr_photo', queue='vision', bind=True, max_retries=3)
def ocr_photo(self, photo_id: str):
"""Run OCR on a photo and store text regions."""
if not settings.vision.enabled or not settings.vision.ocr.enabled:
return {'status': 'skipped', 'reason': 'OCR disabled'}
@@ -109,9 +144,13 @@ def ocr_photo(photo_id: str):
if image is None:
return {'status': 'error', 'message': 'thumbnail not found'}
from app.services.vision.registry import registry
ocr_engine = registry.get_ocr()
results = ocr_engine.run(image)
try:
from app.services.vision.registry import registry
ocr_engine = registry.get_ocr()
results = ocr_engine.run(image)
except Exception as exc:
logger.exception("ocr_photo failed for %s", photo_id)
raise self.retry(exc=exc, countdown=60)
if not results:
logger.info("No OCR text found for photo %s", photo_id)
@@ -131,15 +170,18 @@ def ocr_photo(photo_id: str):
bbox=r.bbox,
))
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
logger.info("OCR: %d text regions for photo %s", len(results), photo_id)
logger.info("[%s] OCR: %d text regions for photo %s", self.request.id, len(results), photo_id)
return {'status': 'success', 'photo_id': photo_id, 'regions': len(results)}
@shared_task(name='detect_objects', queue='vision')
def detect_objects(photo_id: str):
@shared_task(name='detect_objects', queue='vision', bind=True, max_retries=3)
def detect_objects(self, photo_id: str):
"""Detect objects in a photo, create Tag(kind=object) rows, and
link via photo_tags with confidence/bbox/source."""
if not settings.vision.enabled or not settings.vision.detector.enabled:
@@ -149,14 +191,19 @@ def detect_objects(photo_id: str):
if image is None:
return {'status': 'error', 'message': 'thumbnail not found'}
from app.services.vision.registry import registry
detector = registry.get_detector()
detections = detector.detect(image)
try:
from app.services.vision.registry import registry
detector = registry.get_detector()
detections = detector.detect(image)
except Exception as exc:
logger.exception("detect_objects failed for %s", photo_id)
raise self.retry(exc=exc, countdown=60)
if not detections:
logger.info("No objects detected for photo %s", photo_id)
return {'status': 'success', 'photo_id': photo_id, 'objects': 0}
from app.models import Photo
from app.models.tags import Tag, photo_tags
source_name = "vision:yolov8n"
@@ -206,16 +253,19 @@ def detect_objects(photo_id: str):
)
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
labels = [d.label for d in detections]
logger.info("Detected %d objects in photo %s: %s", len(detections), photo_id, labels)
logger.info("[%s] Detected %d objects in photo %s: %s", self.request.id, len(detections), photo_id, labels)
return {'status': 'success', 'photo_id': photo_id, 'objects': len(detections)}
@shared_task(name='classify_content', queue='vision')
def classify_content(photo_id: str):
@shared_task(name='classify_content', queue='vision', bind=True, max_retries=3)
def classify_content(self, photo_id: str):
"""Classify image content type (screenshot, document, artwork, etc.)
using CLIP zero-shot classification. Writes Tag(kind=content_type)."""
if not settings.vision.enabled or not settings.vision.classifier.enabled:
@@ -225,14 +275,19 @@ def classify_content(photo_id: str):
if image is None:
return {'status': 'error', 'message': 'thumbnail not found'}
from app.services.vision.registry import registry
classifier = registry.get_classifier()
results = classifier.classify(image)
try:
from app.services.vision.registry import registry
classifier = registry.get_classifier()
results = classifier.classify(image)
except Exception as exc:
logger.exception("classify_content failed for %s", photo_id)
raise self.retry(exc=exc, countdown=60)
if not results:
logger.info("No confident classification for photo %s", photo_id)
return {'status': 'success', 'photo_id': photo_id, 'content_type': None}
from app.models import Photo
from app.models.tags import Tag, photo_tags
source_name = "vision:clip_classifier"
@@ -273,10 +328,13 @@ def classify_content(photo_id: str):
)
)
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
logger.info("Classified photo %s as '%s' (%.2f)", photo_id, best.label, best.confidence)
logger.info("[%s] Classified photo %s as '%s' (%.2f)", self.request.id, photo_id, best.label, best.confidence)
return {'status': 'success', 'photo_id': photo_id, 'content_type': best.label}
@@ -309,15 +367,19 @@ def _load_original(photo_id: str) -> np.ndarray | None:
w, h = img.size
if max(w, h) > max_dim:
scale = max_dim / max(w, h)
img = img.resize((int(w * scale), int(h * scale)), Image.BICUBIC)
return np.array(img)
resized = img.resize((int(w * scale), int(h * scale)), Image.BICUBIC)
img.close()
img = resized
arr = np.array(img)
img.close()
return arr
except Exception as e:
logger.warning("Failed to load original %s: %s", filepath, e)
return None
@shared_task(name='extract_faces', queue='vision')
def extract_faces(photo_id: str):
@shared_task(name='extract_faces', queue='vision', bind=True, max_retries=3)
def extract_faces(self, photo_id: str):
"""Detect faces and store recognition embeddings using InsightFace
(RetinaFace + ArcFace). No YOLO workaround needed — RetinaFace has
strong human-vs-non-human precision on its own."""
@@ -330,9 +392,13 @@ def extract_faces(photo_id: str):
if image is None:
return {'status': 'error', 'message': 'no image available'}
from app.services.vision.registry import registry
face_proc = registry.get_face_processor()
faces = face_proc.process(image)
try:
from app.services.vision.registry import registry
face_proc = registry.get_face_processor()
faces = face_proc.process(image)
except Exception as exc:
logger.exception("extract_faces failed for %s", photo_id)
raise self.retry(exc=exc, countdown=60)
if not faces:
logger.info("No faces detected for photo %s", photo_id)
@@ -355,6 +421,9 @@ def _save_faces(photo_id: str, faces) -> dict:
cluster_id=None,
))
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
@@ -383,8 +452,8 @@ def _schedule_recluster_debounced():
logger.debug("recluster debounce check failed: %s", e)
@shared_task(name='recluster_faces', queue='vision')
def recluster_faces():
@shared_task(name='recluster_faces', queue='vision', bind=True, max_retries=10)
def recluster_faces(self):
"""Run DBSCAN clustering over all face embeddings and assign/create
Tag(kind=face_cluster) entries."""
# Clear debounce key so new face extractions can schedule another round.
@@ -394,9 +463,14 @@ def recluster_faces():
except Exception:
pass
if not _vision_worker_ready():
logger.info("Vision worker not ready yet — retrying in 30s")
raise self.retry(countdown=30)
if not settings.vision.enabled or not settings.vision.faces.enabled:
return {'status': 'skipped', 'reason': 'faces disabled'}
from app.models import Photo
from app.models.face_embedding import FaceEmbedding
from app.models.tags import Tag, photo_tags
from app.services.vision.clustering import cluster_faces
@@ -477,6 +551,9 @@ def recluster_faces():
)
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
@@ -485,45 +562,98 @@ def recluster_faces():
return {'status': 'success', 'clusters': n_clusters, 'faces': len(face_rows)}
@shared_task(name='backfill_vision')
def backfill_vision(task: str | None = None, limit: int | None = None):
@shared_task(name='backfill_vision', bind=True, max_retries=10)
def backfill_vision(self, task: str | None = None, limit: int | None = None):
"""Queue vision tasks for photos that haven't been processed yet.
Uses a sync DB connection to avoid asyncpg conflicts in Celery."""
if not _vision_worker_ready():
logger.info("Vision worker not ready yet — retrying in 30s")
raise self.retry(countdown=30)
model_name = settings.vision.embedder.name
# Newest-first ordering — matches regenerate_all_thumbnails so the
# whole ingestion pipeline sweeps the library top-down and the user
# sees recent photos fully-indexed long before the backlog drains.
# `taken_at` is the canonical capture timestamp (from EXIF, falls
# back to filesystem mtime in scan); `added_at` is the tie-breaker
# when taken_at is null.
sql = """
SELECT p.id FROM photos p
LEFT JOIN embeddings e ON e.photo_id = p.id AND e.model = :model
WHERE e.photo_id IS NULL
AND p.processing_status = 'completed'
ORDER BY p.taken_at DESC NULLS LAST, p.added_at DESC NULLS LAST
"""
ordering = "ORDER BY p.taken_at DESC NULLS LAST, p.added_at DESC NULLS LAST"
limit_clause = " LIMIT :lim" if limit else ""
params: dict = {"model": model_name}
if limit:
sql += f" LIMIT {limit}"
params["lim"] = int(limit)
session = _get_sync_session()
try:
result = session.execute(sa_text(sql), {"model": model_name})
photo_ids = [row[0] for row in result.fetchall()]
# Each query finds photos missing a specific pipeline output so
# enabling a new processor after import still back-fills.
embed_ids = []
if task in ('embed', None):
sql = f"""
SELECT p.id FROM photos p
LEFT JOIN embeddings e ON e.photo_id = p.id AND e.model = :model
WHERE e.photo_id IS NULL AND p.processing_status = 'completed'
{ordering}{limit_clause}
"""
embed_ids = [r[0] for r in session.execute(sa_text(sql), params).fetchall()]
ocr_ids = []
if task in ('ocr', None) and settings.vision.ocr.enabled:
sql = f"""
SELECT p.id FROM photos p
LEFT JOIN ocr_text o ON o.photo_id = p.id
WHERE o.photo_id IS NULL AND p.processing_status = 'completed'
{ordering}{limit_clause}
"""
ocr_ids = [r[0] for r in session.execute(sa_text(sql), params).fetchall()]
detect_ids = []
if task in ('detect', None) and settings.vision.detector.enabled:
sql = f"""
SELECT p.id FROM photos p
WHERE p.processing_status = 'completed'
AND NOT EXISTS (
SELECT 1 FROM photo_tags pt WHERE pt.photo_id = p.id
AND pt.source = 'vision:yolov8n'
)
{ordering}{limit_clause}
"""
detect_ids = [r[0] for r in session.execute(sa_text(sql), params).fetchall()]
face_ids = []
if task in ('faces', None) and settings.vision.faces.enabled:
sql = f"""
SELECT p.id FROM photos p
LEFT JOIN face_embeddings fe ON fe.photo_id = p.id
WHERE fe.photo_id IS NULL AND p.processing_status = 'completed'
{ordering}{limit_clause}
"""
face_ids = [r[0] for r in session.execute(sa_text(sql), params).fetchall()]
classify_ids = []
if task in ('classify', None) and settings.vision.classifier.enabled:
sql = f"""
SELECT p.id FROM photos p
WHERE p.processing_status = 'completed'
AND NOT EXISTS (
SELECT 1 FROM photo_tags pt WHERE pt.photo_id = p.id
AND pt.source = 'vision:clip_classifier'
)
{ordering}{limit_clause}
"""
classify_ids = [r[0] for r in session.execute(sa_text(sql), params).fetchall()]
except Exception:
session.rollback()
raise
finally:
session.close()
count = 0
for pid in photo_ids:
if task == 'embed' or task is None:
embed_photo.delay(pid)
if task == 'ocr' or task is None:
ocr_photo.delay(pid)
if task == 'detect' or task is None:
detect_objects.delay(pid)
if task == 'faces' or task is None:
extract_faces.delay(pid)
count += 1
# Dispatch — deduplicate across query results.
all_ids = set(embed_ids) | set(ocr_ids) | set(detect_ids) | set(face_ids) | set(classify_ids)
for pid in embed_ids:
embed_photo.delay(pid)
for pid in ocr_ids:
ocr_photo.delay(pid)
for pid in detect_ids:
detect_objects.delay(pid)
for pid in face_ids:
extract_faces.delay(pid)
for pid in classify_ids:
classify_content.delay(pid)
logger.info("Backfill queued %d photos for vision processing", count)
return {'status': 'queued', 'count': count}
logger.info("Backfill queued %d photos for vision processing", len(all_ids))
return {'status': 'queued', 'count': len(all_ids)}