diff --git a/.env b/.env index 84cc4b4..c11e4e0 100644 --- a/.env +++ b/.env @@ -2,7 +2,7 @@ # See .env.example for the full list of knobs and their docs. # REQUIRED — host path to your photo library. -PHOTO_DIRS=/Users/dtoro/Pictures/MulitaTest +PHOTO_DIRS=/mnt/library/homecloud/admin/files/ # Ports — change if 3000 / 8001 collide with other services on the host. FRONTEND_PORT=3000 @@ -16,5 +16,7 @@ ALLOWED_ORIGINS=* LOG_LEVEL=INFO TZ=UTC -# Celery worker pool. -CELERYD_CONCURRENCY=4 +# Celery worker pools — split worker-light (IO) and worker-vision (CPU). +# Defaults target a 6-core / 16 GB host. +CELERY_LIGHT_CONCURRENCY=2 +CELERY_VISION_CONCURRENCY=5 diff --git a/.env.example b/.env.example index 6418ed3..f38a881 100644 --- a/.env.example +++ b/.env.example @@ -65,11 +65,27 @@ TZ=UTC # ── WORKER CONCURRENCY ─────────────────────────────────────────────────────── - -# How many parallel Celery worker processes to spin up. Each one can run -# one scan / thumbnail / metadata job at a time. Bump on a beefy host with a -# big library; lower on a Pi. -CELERYD_CONCURRENCY=4 +# +# The ingestion pipeline runs on two Celery worker services with separate +# concurrency knobs so heavy vision tasks can't starve cheap IO tasks: +# +# worker-light (default / high / low queues) +# Runs: scan, thumbnails, EXIF, pHash, duplicate regrouping. +# Mostly IO-bound — 2 prefork children keep a library streaming in. +# +# worker-vision (vision queue) +# Runs: embeddings, object detection, OCR, face extraction, content +# classification. Each prefork child loads ~2 GB of ONNX model weights, +# so set this to roughly (physical_cores − 1) and watch RAM. +# +# Defaults target a ~6 core / 16 GB host. Raise these, then +# docker compose up -d worker-light worker-vision +# to pick them up. Lower for a Pi; go higher on a workstation. +# +# The old `CELERYD_CONCURRENCY=N` single-worker variable is no longer +# read — delete it from your .env if it's set. +CELERY_LIGHT_CONCURRENCY=2 +CELERY_VISION_CONCURRENCY=5 # ── INTERNAL (rarely overridden) ───────────────────────────────────────────── diff --git a/backend/app/database.py b/backend/app/database.py index d016fb9..a6b5b9b 100644 --- a/backend/app/database.py +++ b/backend/app/database.py @@ -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, diff --git a/backend/app/routers/library.py b/backend/app/routers/library.py index 6b98aa8..9f3d35e 100644 --- a/backend/app/routers/library.py +++ b/backend/app/routers/library.py @@ -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 diff --git a/backend/app/services/vision/bootstrap_models.py b/backend/app/services/vision/bootstrap_models.py index bff5bf5..bb9bf77 100644 --- a/backend/app/services/vision/bootstrap_models.py +++ b/backend/app/services/vision/bootstrap_models.py @@ -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) diff --git a/backend/app/services/vision/export_models.py b/backend/app/services/vision/export_models.py index eaa6cbf..9408e38 100644 --- a/backend/app/services/vision/export_models.py +++ b/backend/app/services/vision/export_models.py @@ -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) diff --git a/backend/app/tasks/scan.py b/backend/app/tasks/scan.py index 59e82ed..bbbaa27 100644 --- a/backend/app/tasks/scan.py +++ b/backend/app/tasks/scan.py @@ -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()] diff --git a/backend/app/tasks/thumbs.py b/backend/app/tasks/thumbs.py index 41c8961..0a8bdb8 100644 --- a/backend/app/tasks/thumbs.py +++ b/backend/app/tasks/thumbs.py @@ -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() diff --git a/backend/app/tasks/vision.py b/backend/app/tasks/vision.py index 94b1748..384091a 100644 --- a/backend/app/tasks/vision.py +++ b/backend/app/tasks/vision.py @@ -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}" diff --git a/backend/requirements.txt b/backend/requirements.txt index b48df03..3c88c42 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -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 diff --git a/docker-compose.yml b/docker-compose.yml index 2286b17..106b832 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -57,12 +57,34 @@ services: - mulita-network restart: unless-stopped - worker: + # ── Celery workers ───────────────────────────────────────────────────── + # + # The ingestion pipeline is split across two worker services so CPU-heavy + # vision tasks (embed / detect / OCR / faces / classify) cannot starve + # the fast IO-bound tasks (scan / thumbnails / EXIF / phash / duplicates). + # + # worker-light listens on default,high,low — IO-bound, cheap + # worker-vision listens on vision — CPU-bound, loads ONNX + # + # Both share the same image, photo volume, and model cache, so there's + # no disk duplication and model weights are loaded lazily only by + # worker-vision. Each service has its own concurrency knob; both + # workers ship their heartbeat to the same Redis broker so the + # Settings > Workers panel lists them side-by-side. + # + # Sizing defaults target a 6-core / 16 GB host: + # CELERY_LIGHT_CONCURRENCY=2 (enough for parallel thumbnail + EXIF) + # CELERY_VISION_CONCURRENCY=5 (5 × ~2GB ONNX = ~10GB RAM, 5/6 cores) + # Raise these in .env and run `docker compose up -d worker-light worker-vision` + # to scale. Keep light under ~4 and vision under your physical core + # count; more just thrashes. + worker-light: build: context: ./backend dockerfile: Dockerfile - container_name: mulita-worker - command: sh -c "python -m app.services.vision.bootstrap_models && celery -A app.tasks.celery worker --loglevel=${LOG_LEVEL:-info} --concurrency=${CELERYD_CONCURRENCY:-4} -Q default,high,low,vision" + 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" volumes: - ./mulita.yml:/app/config/mulita.yml:ro - ${PHOTO_DIRS:-./photos}:/photos:rw @@ -76,9 +98,52 @@ services: - CELERY_BROKER_URL=redis://redis:6379 - CELERY_RESULT_BACKEND=redis://redis:6379 - PHOTO_DIRS=${PHOTO_DIRS:-/photos} - - CELERYD_CONCURRENCY=${CELERYD_CONCURRENCY:-4} - LOG_LEVEL=${LOG_LEVEL:-INFO} - TZ=${TZ:-UTC} + # NullPool — see app/database.py for rationale. + - MULITA_CELERY_WORKER=1 + depends_on: + redis: + condition: service_started + backend: + condition: service_started + db: + condition: service_healthy + networks: + - mulita-network + restart: unless-stopped + + worker-vision: + build: + context: ./backend + dockerfile: Dockerfile + image: mule-image-worker + container_name: mulita-worker-vision + command: sh -c "python -m app.services.vision.bootstrap_models && celery -A app.tasks.celery worker --loglevel=${LOG_LEVEL:-info} --concurrency=${CELERY_VISION_CONCURRENCY:-5} -Q vision -n vision@%h" + volumes: + - ./mulita.yml:/app/config/mulita.yml:ro + - ${PHOTO_DIRS:-./photos}:/photos:rw + - thumbs_data:/data/thumbs + - proxies_data:/data/proxies + - db_data:/data/db + - models_data:/data/models + 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=${PHOTO_DIRS:-/photos} + - LOG_LEVEL=${LOG_LEVEL:-INFO} + - TZ=${TZ:-UTC} + - MULITA_CELERY_WORKER=1 + # Pin each ONNX session to one intra-op thread so N prefork children + # × default-all-cores doesn't oversubscribe the box. With + # concurrency=5 and OMP=1, vision peaks at 5 busy cores, leaving + # one for worker-light + system. These env vars cover the three + # threading runtimes ONNX Runtime might pick up on first use. + - OMP_NUM_THREADS=1 + - OPENBLAS_NUM_THREADS=1 + - MKL_NUM_THREADS=1 depends_on: redis: condition: service_started diff --git a/frontend/src/components/dialogs/SettingsDialog.tsx b/frontend/src/components/dialogs/SettingsDialog.tsx index f3312ee..684a991 100644 --- a/frontend/src/components/dialogs/SettingsDialog.tsx +++ b/frontend/src/components/dialogs/SettingsDialog.tsx @@ -1,4 +1,4 @@ -import { useEffect, useState, useCallback } from 'react' +import { useEffect, useState, useCallback, useRef } from 'react' import { X, RefreshCw, @@ -13,12 +13,17 @@ import { CheckCircle2, Copy, Sparkles, + Activity, + FolderSearch, } from 'lucide-react' import clsx from 'clsx' import { useQuery, useQueryClient } from '@tanstack/react-query' import { library, type MediaType, + type PipelineStage, + type ScanStatus, + type WorkerStatus, } from '../../services/api' import { toast } from '../ToastContainer' @@ -29,6 +34,8 @@ const SETTINGS_THUMB_STATS_KEY = ['settings', 'thumbnail-stats'] as const const SETTINGS_LIB_STATS_KEY = ['settings', 'library-stats'] as const const SETTINGS_WORKER_STATUS_KEY = ['settings', 'worker-status'] as const const SETTINGS_MISSING_STATS_KEY = ['settings', 'missing-stats'] as const +const SETTINGS_PIPELINE_STATS_KEY = ['settings', 'pipeline-stats'] as const +const SETTINGS_SCAN_STATUS_KEY = ['settings', 'scan-status'] as const // Shared with the DuplicatesView so a regroup invalidates the same cache // the grid renders from. Imported via the canonical hook key. import { DUPLICATE_GROUPS_QUERY_KEY } from '../../hooks/useDuplicateGroupsQuery' @@ -91,6 +98,28 @@ export function SettingsDialog({ isOpen, onClose }: SettingsDialogProps) { refetchInterval: isOpen ? 5000 : false, staleTime: 0, }) + // Pipeline progress polls on the same 5s cadence as the worker status + // so both cards update together. Cheap query — ten COUNT(*)s on + // indexed columns. + const pipelineStatsQuery = useQuery({ + queryKey: SETTINGS_PIPELINE_STATS_KEY, + queryFn: library.maintenance.pipelineStats, + enabled: isOpen, + refetchInterval: isOpen ? 5000 : false, + staleTime: 0, + }) + // Scan status — polls fast (2s) so the progress bar feels live during + // a scan, and slow (15s) when idle to cut chatter. `isScanning` is + // read from the latest fetched value so the cadence flips on its own + // the moment a scan kicks off or finishes. + const scanStatusQuery = useQuery({ + queryKey: SETTINGS_SCAN_STATUS_KEY, + queryFn: library.scanStatus, + enabled: isOpen, + refetchInterval: (q) => + isOpen ? ((q.state.data as ScanStatus | undefined)?.is_scanning ? 2000 : 15000) : false, + staleTime: 0, + }) // Duplicates: shares the cache with DuplicatesView so a regroup // triggered from Settings updates the grid view immediately. const duplicatesQuery = useQuery({ @@ -104,6 +133,19 @@ export function SettingsDialog({ isOpen, onClose }: SettingsDialogProps) { const libStats = libStatsQuery.data const workerStatus = workerStatusQuery.data const missingStats = missingStatsQuery.data + const pipelineStats = pipelineStatsQuery.data + const scanStatus = scanStatusQuery.data + + // Throughput — tasks/minute across the fleet, derived by diffing the + // `total` counters on `workers[*].processed` between successive polls. + // We keep the previous sample in a ref so recomputation happens inside + // the existing 5s polling rhythm without introducing extra state. + // First sample returns null (we need two points for a rate). + const throughputSampleRef = useRef<{ + totals: Record + at: number + } | null>(null) + const throughput = computeThroughput(workerStatus, throughputSampleRef) // "loading" in the UI sense = fetching AND no cached data yet. Background // refetches on top of cached data shouldn't flip the refresh spinners. const loadingStats = @@ -116,11 +158,13 @@ export function SettingsDialog({ isOpen, onClose }: SettingsDialogProps) { const refreshStats = useCallback(() => { queryClient.invalidateQueries({ queryKey: SETTINGS_THUMB_STATS_KEY }) queryClient.invalidateQueries({ queryKey: SETTINGS_LIB_STATS_KEY }) + queryClient.invalidateQueries({ queryKey: SETTINGS_PIPELINE_STATS_KEY }) queryClient.invalidateQueries({ queryKey: DUPLICATE_GROUPS_QUERY_KEY }) }, [queryClient]) const refreshWorkers = useCallback(() => { queryClient.invalidateQueries({ queryKey: SETTINGS_WORKER_STATUS_KEY }) queryClient.invalidateQueries({ queryKey: SETTINGS_MISSING_STATS_KEY }) + queryClient.invalidateQueries({ queryKey: SETTINGS_SCAN_STATUS_KEY }) }, [queryClient]) // Surface fetch errors once (React Query de-dupes retries but we still @@ -246,6 +290,33 @@ export function SettingsDialog({ isOpen, onClose }: SettingsDialogProps) { } /> + + {/* Inline scan progress. Rendered as a full-width sub-card + when a scan is active so the user sees the same info + they'd get from the floating widget without leaving + settings. Hidden when idle to keep the section tight. */} + {scanStatus?.is_scanning && ( +
+
+ + Scanning +
+
+ + {scanStatus.current_folder ?? '—'} + + + {scanStatus.processed_files.toLocaleString()} /{' '} + {scanStatus.total_files.toLocaleString()} + +
+ +
+ )} +
+ {/* ----------------------------------------------------- */} + {/* Pipeline progress — per-stage done/total */} + {/* ----------------------------------------------------- */} +
} + title="Pipeline progress" + right={ + + } + > + {pipelineStats ? ( +
+ {pipelineStats.stages.map((stage) => ( + + ))} +

+ Partial stages (marked *) only apply to a subset of + photos — e.g. not every photo has GPS, text, or detectable + objects. Face clusters and duplicate groups are output + counts rather than ratios. +

+
+ ) : ( +
+ + Loading pipeline progress… +
+ )} +
+ {/* ----------------------------------------------------- */} {/* Duplicate detection */} {/* ----------------------------------------------------- */} @@ -430,7 +546,7 @@ export function SettingsDialog({ isOpen, onClose }: SettingsDialogProps) { } > {/* Top-line health */} -
+
0 + ? workerStatus.workers.reduce( + (acc, w) => acc + (w.concurrency ?? 0), + 0 + ) : undefined } + /> + 0 ? 'ok' : 'muted' } />
+ {/* How to scale — explicit, because this question comes up + every time a large import is running and the user wants + it to go faster. The ingestion pipeline runs on two + worker services; each has its own concurrency env var. */} +

+ Two worker services share the load:{' '} + worker-light{' '} + (scan, thumbnails, EXIF — set by{' '} + + CELERY_LIGHT_CONCURRENCY + + ) and{' '} + worker-vision{' '} + (embed, detect, OCR, faces — set by{' '} + + CELERY_VISION_CONCURRENCY + + ) in{' '} + .env. To + scale, bump those values then run{' '} + + docker compose up -d worker-light worker-vision + + . Keep vision below your physical core count — each fork + loads ~2 GB of ONNX models. The vision queue is usually the + bottleneck; watch its depth below. +

+ {/* Inline error banners for the obvious failure modes */} {workerStatus?.broker_error && ( )} - {/* Queue depth */} + {/* Queue depth. `vision` is called out with a highlight + because it's where the serious backlog lives — every + embed / tag / OCR / face task routes here. */} {workerStatus && (
Queue depth
-
+
{Object.entries(workerStatus.queues).map(([name, depth]) => (
0 + ? 'bg-surface-2' + : 'bg-surface' + )} > {name} 0 ? 'text-text' : 'text-text-muted' )} > - {depth} + {depth.toLocaleString()}
))} @@ -797,6 +953,116 @@ function ErrorBanner({ title, detail }: { title: string; detail: string }) { ) } +/** + * Compute the total task-completion rate across the worker fleet by + * diffing the per-task `processed` counters between successive poll + * samples. Returns null on the first call (we need two samples for a + * rate) and 0 when nothing has moved since the last poll. + * + * The sample is kept in a ref — not state — because we don't want to + * re-render on every update; we want the number to settle into the + * existing render cycle that React Query already drives. + */ +function computeThroughput( + status: WorkerStatus | undefined, + ref: React.MutableRefObject<{ + totals: Record + at: number + } | null> +): number | null { + if (!status || status.workers.length === 0) { + return null + } + + // Flatten processed counts across all workers into one dict keyed by + // task name so worker restarts (which reset individual counters) are + // absorbed by the total. + const totals: Record = {} + for (const w of status.workers) { + for (const [task, count] of Object.entries(w.processed ?? {})) { + totals[task] = (totals[task] ?? 0) + (count ?? 0) + } + } + + const now = Date.now() + const prev = ref.current + // Update the ref BEFORE returning so the next call has a baseline. + ref.current = { totals, at: now } + + if (!prev) { + return null + } + + const elapsedSec = (now - prev.at) / 1000 + if (elapsedSec <= 0) { + return null + } + + let delta = 0 + for (const [task, count] of Object.entries(totals)) { + const before = prev.totals[task] ?? 0 + // Guard against counter resets (worker restart) — negative diffs + // are clamped to zero rather than dragging the rate down. + delta += Math.max(0, count - before) + } + + return (delta / elapsedSec) * 60 +} + +function PipelineRow({ stage }: { stage: PipelineStage }) { + const isStandalone = stage.standalone === true + const isComplete = !isStandalone && stage.total > 0 && stage.done >= stage.total + const isActive = !isStandalone && stage.done > 0 && stage.done < stage.total + + return ( +
+
+
+ + {stage.label} + {stage.partial && *} + +
+ + {isStandalone + ? stage.done.toLocaleString() + : `${stage.done.toLocaleString()} / ${stage.total.toLocaleString()}`} + +
+ {!isStandalone && ( +
+ +
+ )} +
+ ) +} + +function ProgressBar({ done, total }: { done: number; total: number }) { + const pct = total > 0 ? Math.min(100, (done / total) * 100) : 0 + const complete = total > 0 && done >= total + return ( +
+
+
+ ) +} + function ActionButton({ loading, disabled, diff --git a/frontend/src/services/api.ts b/frontend/src/services/api.ts index d86122e..6dc403c 100644 --- a/frontend/src/services/api.ts +++ b/frontend/src/services/api.ts @@ -297,6 +297,37 @@ export interface WorkerStatus { scan_errors: string[] } +export interface PipelineStage { + key: string + label: string + done: number + total: number + hint: string + /** True when the stage legitimately runs on a subset of photos — e.g. + * GPS / tags / OCR / faces — so 100% coverage is never expected and + * the UI should not frame "missing" as a problem. */ + partial?: boolean + /** True when the stage doesn't have a done/total progress semantic + * (e.g. face clusters, duplicate groups — those are output counts, + * not ratios). The UI renders a plain count instead of a bar. */ + standalone?: boolean +} + +export interface PipelineStats { + total_photos: number + total_images: number + embedder_model: string + stages: PipelineStage[] +} + +export interface ScanStatus { + is_scanning: boolean + current_folder: string | null + processed_files: number + total_files: number + errors: string[] +} + export const library = { scan: async () => { const response = await api.post('/library/scan') @@ -344,6 +375,14 @@ export const library = { return response.data }, + /** Per-stage ingestion progress — thumbnails, EXIF, GPS, phash, + * embeddings, object tags, OCR, faces, face clusters, duplicate + * groups. Drives the Pipeline Progress card in Settings. */ + pipelineStats: async (): Promise => { + const response = await api.get('/library/maintenance/pipeline-stats') + return response.data + }, + /** Dry-run count of photo rows whose files are no longer on disk * (under a mounted source root). */ missingStats: async (): Promise => {