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

feat: split celery workers, fix asyncpg-in-fork, add pipeline progress UI

Browse Source 
Three overlapping fixes so the ingestion pipeline actually runs and the
user can see what it's doing:

Pipeline recovery
- app/database.py: use NullPool when MULITA_CELERY_WORKER=1 so each
  Celery task opens a fresh asyncpg connection on its own event loop.
  Fixes "another operation in progress" and "Future attached to a
  different loop" errors that were dropping ~every thumbnail +
  extract_metadata task on the floor.
- app/tasks/thumbs.py: initialize photo=None before the try and rollback
  on error so a transport failure in the initial SELECT doesn't raise
  UnboundLocalError in the except block and leak rows stuck in 'pending'.
- app/services/vision/bootstrap_models.py: on missing model files,
  invoke export_models automatically instead of just warning. First
  boot of a fresh install now self-heals.
- app/services/vision/export_models.py: shutil.move instead of
  Path.rename so the YOLO export survives the /app → /data/models
  cross-volume hop.
- requirements.txt: add ultralytics so export works in a stock image.

Worker topology
- docker-compose.yml: replace the single worker with worker-light
  (default/high/low queues, c=2, IO-bound) and worker-vision (vision
  queue, c=5, OMP_NUM_THREADS=1 to avoid oversubscription on 6 cores).
  Vision is pinned to ≤5 parallel inferences so ONNX doesn't each
  spawn an all-cores intra-op pool.
- .env / .env.example: CELERYD_CONCURRENCY replaced with
  CELERY_LIGHT_CONCURRENCY + CELERY_VISION_CONCURRENCY.
- Backfill queries in thumbs / scan / vision now ORDER BY taken_at
  DESC NULLS LAST so newest photos finish first — the library fills
  in top-down in the UI instead of arbitrary insertion order.

Settings visibility
- routers/library.py: new GET /maintenance/pipeline-stats returning
  done/total per stage (thumbnails, exif, gps, phash, embeddings,
  tags, ocr, faces, face clusters, duplicate groups). Worker-status
  now also reports the `vision` queue depth, which was missing.
- services/api.ts: PipelineStats / PipelineStage / ScanStatus types
  and the matching client call.
- components/dialogs/SettingsDialog.tsx:
  - new Pipeline Progress card with one progress bar per stage
  - inline scan banner (processed/total/current folder) inside the
    Library section while a scan is running
  - Tasks/min throughput computed by diffing worker processed counters
    between polls
  - Workers section calls out the vision queue and documents the
    CELERY_LIGHT/VISION_CONCURRENCY + docker compose up -d scale path

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
root
2026-04-11 10:06:45 +02:00
parent afe420c620
commit 07b1e5e02a
13 changed files with 746 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
backend/app/database.py
View File

@@ -15,8 +15,10 @@ SQLite (escape hatch via docker-compose.sqlite.yml): no Alembic. The
historical inline ALTER TABLE block stays in place so existing dev
installs keep upgrading.
"""
import os
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.orm import declarative_base
from sqlalchemy.pool import NullPool
from sqlalchemy import text
import logging
from pathlib import Path
@@ -28,6 +30,27 @@ logger = logging.getLogger(__name__)
_is_sqlite = settings.database_url.startswith("sqlite")
_is_postgres = settings.database_url.startswith("postgresql")
# When running inside a Celery worker we use NullPool rather than the
# default connection pool. The reasons stack up:
#
# 1. Celery's prefork model forks the master *after* imports, so every
# child inherits the same asyncpg Connection objects — they share
# a socket, and two children using one concurrently raises
# "another operation is in progress".
#
# 2. Task bodies run under `asyncio.run()`, which spins up a fresh
# event loop per invocation. A pooled asyncpg Connection created
# on loop A, returned to the pool, and checked out on loop B
# raises "Future attached to a different loop".
#
# NullPool dodges both: every session checkout opens a brand-new
# connection on the *current* loop and the connection is closed at
# session end. Connection setup is cheap compared to task cost, so this
# is the right default for the worker. The FastAPI backend keeps the
# normal pool because it serves many short requests on a single long-
# lived event loop, where pooling is a clear win.
_is_celery_worker = os.environ.get("MULITA_CELERY_WORKER") == "1"
if _is_sqlite:
db_path = Path(settings.database_url.replace("sqlite+aiosqlite:///", ""))
db_path.parent.mkdir(parents=True, exist_ok=True)
@@ -39,6 +62,12 @@ if _is_sqlite:
"timeout": 30,
},
)
elif _is_celery_worker:
engine = create_async_engine(
settings.database_url,
echo=False,
poolclass=NullPool,
)
else:
engine = create_async_engine(
settings.database_url,

206
backend/app/routers/library.py
View File

@@ -378,7 +378,11 @@ async def get_worker_status(db: AsyncSession = Depends(get_db)):
r = _redis.Redis.from_url(settings.redis_url, socket_timeout=1.0)
r.ping()
broker_ok = True
for q in ('default', 'high', 'low'):
# `vision` is the big one — embed / classify / detect / ocr /
# extract_faces all land here, so it's where backlogs actually
# pile up. Leaving it off the dashboard made it look like the
# queue was always empty while the worker was clearly busy.
for q in ('default', 'high', 'low', 'vision'):
try:
queue_depths[q] = int(r.llen(q) or 0)
except Exception:
@@ -445,6 +449,206 @@ async def get_worker_status(db: AsyncSession = Depends(get_db)):
}
@router.get("/maintenance/pipeline-stats")
async def get_pipeline_stats(db: AsyncSession = Depends(get_db)):
"""Per-stage progress across the ingestion pipeline.
Returns a `{stage_key: {done, total, label}}` map so the Settings
panel can render one progress bar per stage. `total` is the number
of non-discarded photos the stage is *expected* to run on — which is
every non-discarded photo for most stages, or a narrower subset when
a stage is image-only (e.g. embeddings don't run on videos).
Keep the shape flat + serialisable; the frontend turns it straight
into a list of rows without needing to know about the models.
"""
from app.config import settings as _settings
from app.models import Embedding, FaceEmbedding, OCRText
from app.models.tags import photo_tags # association Table, not a model
not_discarded = Photo.is_discarded.is_(False)
async def scalar_count(query):
return (await db.execute(query)).scalar() or 0
# Total non-discarded photos — the denominator for most stages.
total_photos = await scalar_count(
select(func.count(Photo.id)).where(not_discarded)
)
# Image-only denominator (embeddings, tags, faces, OCR, phash). We
# exclude videos because those stages either don't apply or run off
# the extracted video frame which is treated separately.
total_images = await scalar_count(
select(func.count(Photo.id)).where(
not_discarded, Photo.media_type != 'video'
)
)
completed = await scalar_count(
select(func.count(Photo.id)).where(
not_discarded, Photo.processing_status == 'completed'
)
)
with_exif = await scalar_count(
select(func.count(Photo.id)).where(
not_discarded, Photo.exif_json.is_not(None)
)
)
with_gps = await scalar_count(
select(func.count(Photo.id)).where(
not_discarded,
Photo.latitude.is_not(None),
Photo.longitude.is_not(None),
)
)
with_phash = await scalar_count(
select(func.count(Photo.id)).where(
not_discarded, Photo.phash.is_not(None)
)
)
# Embeddings: count distinct photos that have a row for the currently
# configured embedder model. A photo can have multiple model rows
# (historical re-embeds) so COUNT(DISTINCT) is the right thing here.
embedder_model = _settings.vision.embedder.name
embeddings_done = await scalar_count(
select(func.count(func.distinct(Embedding.photo_id)))
.select_from(Embedding)
.join(Photo, Photo.id == Embedding.photo_id)
.where(not_discarded, Embedding.model == embedder_model)
)
tagged_photos = await scalar_count(
select(func.count(func.distinct(photo_tags.c.photo_id)))
.select_from(photo_tags)
.join(Photo, Photo.id == photo_tags.c.photo_id)
.where(not_discarded)
)
ocr_done = await scalar_count(
select(func.count(func.distinct(OCRText.photo_id)))
.select_from(OCRText)
.join(Photo, Photo.id == OCRText.photo_id)
.where(not_discarded)
)
# Faces: photos that have at least one face_embeddings row. A photo
# with no faces legitimately finishes face extraction with zero rows,
# so this undercounts by exactly "images with no visible people". We
# surface the photo-with-faces count rather than "images scanned for
# faces" because the latter isn't tracked anywhere.
faces_photos = await scalar_count(
select(func.count(func.distinct(FaceEmbedding.photo_id)))
.select_from(FaceEmbedding)
.join(Photo, Photo.id == FaceEmbedding.photo_id)
.where(not_discarded)
)
face_rows = await scalar_count(select(func.count(FaceEmbedding.id)))
face_clusters = await scalar_count(
select(func.count(func.distinct(FaceEmbedding.cluster_id)))
.where(FaceEmbedding.cluster_id.is_not(None))
)
duplicate_groups = await scalar_count(
select(func.count(func.distinct(Photo.duplicate_group_id))).where(
not_discarded, Photo.duplicate_group_id.is_not(None)
)
)
duplicate_members = await scalar_count(
select(func.count(Photo.id)).where(
not_discarded, Photo.duplicate_group_id.is_not(None)
)
)
# Ordered list so the frontend renders stages in pipeline order
# without needing to know the sequence itself.
stages = [
{
"key": "thumbnails",
"label": "Thumbnails & pHash",
"done": completed,
"total": total_photos,
"hint": "Generated on scan. Unlocks every downstream stage.",
},
{
"key": "exif",
"label": "EXIF metadata",
"done": with_exif,
"total": total_photos,
"hint": "Camera, lens, capture time. Required for GPS + taken_at.",
},
{
"key": "gps",
"label": "GPS coordinates",
"done": with_gps,
"total": total_photos,
"hint": "Subset of EXIF. Drives the map view; many photos legitimately have none.",
"partial": True, # not every photo is expected to have GPS
},
{
"key": "phash",
"label": "Perceptual hashes",
"done": with_phash,
"total": total_images,
"hint": "Feeds duplicate detection.",
},
{
"key": "embeddings",
"label": f"Embeddings ({embedder_model})",
"done": embeddings_done,
"total": total_images,
"hint": "Semantic search + content classification.",
},
{
"key": "tags",
"label": "Object tags (YOLO)",
"done": tagged_photos,
"total": total_images,
"hint": "Auto-generated object labels. Not every photo has a detectable object.",
"partial": True,
},
{
"key": "ocr",
"label": "OCR text",
"done": ocr_done,
"total": total_images,
"hint": "Extracted text from screenshots / documents. Many photos have none.",
"partial": True,
},
{
"key": "faces",
"label": "Face detection",
"done": faces_photos,
"total": total_images,
"hint": f"{face_rows} face rows detected across {faces_photos} photos.",
"partial": True,
},
{
"key": "face_clusters",
"label": "Face clusters",
"done": face_clusters,
"total": face_clusters, # no meaningful "total" — it's just the current count
"hint": "Built by recluster_faces. Run it after backfill to populate the People view.",
"standalone": True,
},
{
"key": "duplicates",
"label": "Duplicate groups",
"done": duplicate_groups,
"total": duplicate_groups, # same — current count, not a progress ratio
"hint": f"{duplicate_members} photos in {duplicate_groups} groups. Run regroup_duplicates after new imports.",
"standalone": True,
},
]
return {
"total_photos": total_photos,
"total_images": total_images,
"embedder_model": embedder_model,
"stages": stages,
}
@router.get("/maintenance/missing-stats")
async def get_missing_stats():
"""Count photos whose files no longer exist on disk under a mounted

33
backend/app/services/vision/bootstrap_models.py
View File

@@ -67,15 +67,38 @@ def bootstrap(models_dir: str | None = None):
if missing:
logger.warning(
"Missing %d model(s) that require manual export via export_models.py:",
"Missing %d model file(s); attempting automatic export:",
len(missing),
)
for rel_path, desc in missing:
logger.warning(" %s — %s", base / rel_path, desc)
logger.warning(
"Run: python -m app.services.vision.export_models --models-dir %s",
base,
)
try:
from app.services.vision import export_models
export_models.export_openclip(base)
export_models.export_yolov8n(base)
except Exception as e:
logger.error(
"Automatic export failed: %s. "
"Run `python -m app.services.vision.export_models "
"--models-dir %s` manually before starting the worker.",
e,
base,
)
return
# Re-check what's still missing after the export pass.
still_missing = [
(rel_path, desc)
for rel_path, desc in EXPORTS
if not (base / rel_path).exists()
]
if still_missing:
for rel_path, desc in still_missing:
logger.error(" still missing: %s — %s", base / rel_path, desc)
else:
logger.info("All model files present in %s", base)
else:
logger.info("All model files present in %s", base)

10
backend/app/services/vision/export_models.py
View File

@@ -138,10 +138,16 @@ def export_yolov8n(models_dir: Path):
model = YOLO("yolov8n.pt")
model.export(format="onnx", imgsz=640, simplify=True)
# ultralytics exports to cwd as yolov8n.onnx — move to target
# ultralytics exports to cwd as yolov8n.onnx — move to target. Use
# shutil.move rather than Path.rename so it works across filesystems
# (the cwd is typically /app inside the container, while the target
# /data/models is a separately-mounted volume — Path.rename raises
# "Invalid cross-device link" in that case).
import shutil
exported = Path("yolov8n.onnx")
if exported.exists():
exported.rename(onnx_path)
shutil.move(str(exported), str(onnx_path))
size_mb = onnx_path.stat().st_size / 1e6
logger.info("YOLOv8n exported (%.1f MB)", size_mb)

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

@@ -484,11 +484,18 @@ def backfill_gps():
async def _backfill_gps_async():
async with AsyncSessionLocal() as session:
# Newest-first so the most recent photos get their GPS + EXIF
# written before the worker climbs back through the archive.
result = await session.execute(
select(Photo.id).where(
select(Photo.id)
.where(
Photo.latitude.is_(None),
Photo.is_discarded.is_(False),
)
.order_by(
Photo.taken_at.desc().nullslast(),
Photo.added_at.desc().nullslast(),
)
)
photo_ids = [row[0] for row in result.all()]

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

@@ -237,6 +237,9 @@ def generate_thumbnails(self, photo_id: str):
async def _generate_thumbnails_async(photo_id: str, task):
"""Async implementation of thumbnail generation"""
async with AsyncSessionLocal() as session:
# Declared up front so the except block below can safely check it
# even if the initial SELECT raises (e.g. asyncpg transport error).
photo: Optional[Photo] = None
try:
# Get photo from database
result = await session.execute(
@@ -330,13 +333,25 @@ async def _generate_thumbnails_async(photo_id: str, task):
except Exception as e:
logger.error(f"Error generating thumbnails for {photo_id}: {e}")
# Update error status
if photo:
photo.processing_status = 'failed'
photo.processing_error = str(e)
await session.commit()
# Update error status. If the session is in a bad state (e.g.
# the original failure was a transport error) rollback first so
# the status write has a clean transaction to commit into.
try:
await session.rollback()
except Exception:
pass
if photo is not None:
try:
photo.processing_status = 'failed'
photo.processing_error = str(e)
await session.commit()
except Exception:
logger.exception(
f"Could not mark photo {photo_id} as failed"
)
return {'status': 'error', 'message': str(e)}
@shared_task(name='regenerate_all_thumbnails')
@@ -345,12 +360,24 @@ def regenerate_all_thumbnails():
return asyncio.run(_regenerate_all_thumbnails_async())
async def _regenerate_all_thumbnails_async():
"""Async implementation of regenerating all thumbnails"""
"""Async implementation of regenerating all thumbnails.
Queue order matters on first-boot and recovery runs: we dispatch
newest-first (by EXIF taken_at, fallback added_at) so the user's
most recent photos become fully-indexed before the 2012 archive even
starts. Picking up the library in pipeline order means the grid,
timeline and All Photos view populate top-down instead of the worker
chewing through random insertion-order rows while the UI still
shows grey placeholders.
"""
async with AsyncSessionLocal() as session:
# Get all photos that need thumbnails
# Get all photos that need thumbnails, newest first.
result = await session.execute(
select(Photo).where(
Photo.processing_status.in_(['pending', 'failed'])
select(Photo)
.where(Photo.processing_status.in_(['pending', 'failed']))
.order_by(
Photo.taken_at.desc().nullslast(),
Photo.added_at.desc().nullslast(),
)
)
photos = result.scalars().all()
@@ -389,10 +416,16 @@ async def _backfill_phashes_async():
async with AsyncSessionLocal() as session:
while True:
# Newest-first so the recent end of the library gets phashes
# (and therefore duplicate detection) ahead of the archive.
result = await session.execute(
select(Photo)
.where(Photo.phash.is_(None))
.where(Photo.processing_status == 'completed')
.order_by(
Photo.taken_at.desc().nullslast(),
Photo.added_at.desc().nullslast(),
)
.limit(BATCH)
)
batch = result.scalars().all()

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

@@ -446,12 +446,18 @@ def backfill_vision(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."""
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.added_at DESC
ORDER BY p.taken_at DESC NULLS LAST, p.added_at DESC NULLS LAST
"""
if limit:
sql += f" LIMIT {limit}"

1
backend/requirements.txt
View File

@@ -37,6 +37,7 @@ watchfiles==0.21.0
# Vision pipeline (ONNX Runtime CPU inference)
onnxruntime==1.18.1
open-clip-torch==2.24.0 # tokenizer + export helper; inference via ONNX
ultralytics==8.4.37 # YOLOv8n export helper; inference via ONNX
rapidocr-onnxruntime==1.3.22
scikit-learn==1.4.0 # DBSCAN for face clustering
insightface>=0.7.3 # RetinaFace + ArcFace face detection/recognition