fix(face-clusters): run global reclustering as async job; remove 10s axios timeout cap (#1345)

backend/app/routes/face_clusters.py

@@ -1,8 +1,11 @@
+import asyncio
 import logging
 from binascii import Error as Base64Error
 import base64
 from concurrent.futures import CancelledError, Future, ProcessPoolExecutor
-from typing import Annotated
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Annotated, Dict, Optional
 import uuid
 import os
 from fastapi import APIRouter, Depends, HTTPException, Query, status
@@ -28,8 +31,10 @@
     ErrorResponse,
     GetClustersResponse,
     GetClustersData,
-    GlobalReclusterResponse,
-    GlobalReclusterData,
+    GlobalReclusterStartData,
+    GlobalReclusterStartResponse,
+    GlobalReclusterStatusData,
+    GlobalReclusterStatusResponse,
     ClusterMetadata,
     GetClusterImagesResponse,
     GetClusterImagesData,
@@ -79,6 +84,78 @@ def _rescore_memories_for_cluster(app_state: State, cluster_id: str) -> None:
         logger.error(f"Failed to queue memory rescore for cluster {cluster_id}: {e}")
 
 
+# Reclustering scans every face embedding and can run past any HTTP timeout
+# on large libraries, so it runs as a background task the client polls.
+@dataclass
+class ReclusterTask:
+    status: str = "running"  # running | complete | error
+    clusters_created: Optional[int] = None
+    faces_skipped: Optional[int] = None
+    message: Optional[str] = None
+    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
+    finished_at: Optional[datetime] = None
+    task: Optional[asyncio.Task] = None
+
+
+recluster_tasks: Dict[str, ReclusterTask] = {}
+
+# Holds the in-flight job's task_id (None when idle). A full recluster rebuilds
+# every cluster, so a second concurrent run would race on the same tables.
+_active_recluster_task_id: Optional[str] = None
+
+# Minutes a finished job's result stays available for polling, counted from
+# completion (not start) so a long job isn't reaped right after it finishes.
+RECLUSTER_TASK_TTL_MINUTES = 15
+
+
+async def _run_global_recluster(task_id: str, app_state: State) -> None:
+    global _active_recluster_task_id
+    entry = recluster_tasks[task_id]
+    try:
+        result, total_faces_skipped = await asyncio.to_thread(
+            cluster_util_face_clusters_sync, force_full_reclustering=True
+        )
+        request_metadata_export(app_state)
+
+        entry.status = "complete"
+        entry.clusters_created = result or 0
+        entry.faces_skipped = total_faces_skipped
+        entry.message = (
+            "No faces found to cluster"
+            if not result
+            else "Global reclustering completed successfully."
+        )
+        logger.info("Global reclustering completed successfully (task_id=%s)", task_id)
+    except Exception as e:
+        logger.error(f"Global reclustering failed: {str(e)}")
+        entry.status = "error"
+        entry.message = f"Global reclustering failed: {str(e)}"
+    finally:
+        # Stamp completion time (cleanup ages results from this) and release
+        # the concurrency guard; the result itself stays for the client to poll.
+        entry.finished_at = datetime.now(timezone.utc)
+        if _active_recluster_task_id == task_id:
+            _active_recluster_task_id = None
+
+
+async def _cleanup_stale_recluster_tasks() -> None:
+    """Drop finished reclustering results once they age out. Running tasks stay:
+    one can legitimately run long, and the concurrency guard caps them at one."""
+    while True:
+        await asyncio.sleep(300)  # run every 5 minutes
+        now = datetime.now(timezone.utc)
+        stale = [
+            tid
+            for tid, entry in recluster_tasks.items()
+            if entry.status != "running"
+            and entry.finished_at is not None
+            and (now - entry.finished_at).total_seconds()
+            > RECLUSTER_TASK_TTL_MINUTES * 60
+        ]
+        for tid in stale:
+            recluster_tasks.pop(tid, None)
+
+
 @router.put(
     "/{cluster_id}",
     response_model=RenameClusterResponse,
@@ -372,52 +449,96 @@ def face_tagging(
 
 @router.post(
     "/global-recluster",
-    response_model=GlobalReclusterResponse,
+    status_code=status.HTTP_202_ACCEPTED,
+    response_model=GlobalReclusterStartResponse,
     responses={code: {"model": ErrorResponse} for code in [500]},
 )
-def trigger_global_reclustering(app_state: State = Depends(get_state)):
+async def trigger_global_reclustering(
+    app_state: State = Depends(get_state),
+) -> GlobalReclusterStartResponse:
     """
-    Manually trigger global face reclustering.
+    Start a global face reclustering job in the background.
     This forces full reclustering regardless of the 24-hour rule.
+
+    Returns immediately with a task_id; poll
+    GET /face-clusters/global-recluster/{task_id} for the result, since
+    reclustering runs over every face embedding and can take a long time
+    on large libraries.
+
+    If a reclustering job is already running, its task_id is returned instead
+    of starting a second (concurrent runs would race on the cluster tables).
     """
-    try:
-        logger.info("Starting manual global face reclustering...")
+    global _active_recluster_task_id
 
-        result, total_faces_skipped = cluster_util_face_clusters_sync(
-            force_full_reclustering=True
+    if _active_recluster_task_id is not None:
+        logger.info(
+            "Global reclustering already in progress (task_id=%s); reusing it",
+            _active_recluster_task_id,
+        )
+        return GlobalReclusterStartResponse(
+            success=True,
+            message="Global reclustering already in progress.",
+            data=GlobalReclusterStartData(task_id=_active_recluster_task_id),
         )
-        request_metadata_export(app_state)
 
-        if result == 0:
-            return GlobalReclusterResponse(
-                success=True,
-                message="No faces found to cluster",
-                data=GlobalReclusterData(
-                    clusters_created=0, faces_skipped=total_faces_skipped
-                ),
-            )
+    task_id = str(uuid.uuid4())
+    entry = ReclusterTask()
+    recluster_tasks[task_id] = entry
+    _active_recluster_task_id = task_id
+    entry.task = asyncio.create_task(_run_global_recluster(task_id, app_state))
 
-        logger.info("Global reclustering completed successfully")
+    logger.info("Started manual global face reclustering (task_id=%s)", task_id)
 
-        return GlobalReclusterResponse(
-            success=True,
-            message="Global reclustering completed successfully.",
-            data=GlobalReclusterData(
-                clusters_created=result, faces_skipped=total_faces_skipped
-            ),
-        )
+    return GlobalReclusterStartResponse(
+        success=True,
+        message="Global reclustering started.",
+        data=GlobalReclusterStartData(task_id=task_id),
+    )
 
-    except Exception as e:
-        logger.error(f"Global reclustering failed: {str(e)}")
+
+@router.get(
+    "/global-recluster/{task_id}",
+    response_model=GlobalReclusterStatusResponse,
+    responses={code: {"model": ErrorResponse} for code in [404, 500]},
+)
+async def get_global_recluster_status(task_id: str) -> GlobalReclusterStatusResponse:
+    """Poll the status of a previously started global reclustering job."""
+    entry = recluster_tasks.get(task_id)
+    if entry is None:
         raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            status_code=status.HTTP_404_NOT_FOUND,
             detail=ErrorResponse(
                 success=False,
-                error="Internal server error",
-                message=f"Global reclustering failed: {str(e)}",
+                error="Task Not Found",
+                message=f"Recluster task '{task_id}' not found or expired.",
             ).model_dump(),
         )
 
+    if entry.status == "running":
+        return GlobalReclusterStatusResponse(
+            success=True,
+            data=GlobalReclusterStatusData(status="running"),
+        )
+
+    # Leave the entry in place so repeated polls (multiple tabs, retries) see
+    # the same result; _cleanup_stale_recluster_tasks reaps it once stale.
+    if entry.status == "error":
+        return GlobalReclusterStatusResponse(
+            success=False,
+            message=entry.message,
+            data=GlobalReclusterStatusData(status="error"),
+        )
+
+    return GlobalReclusterStatusResponse(
+        success=True,
+        message=entry.message,
+        data=GlobalReclusterStatusData(
+            status="complete",
+            clusters_created=entry.clusters_created,
+            faces_skipped=entry.faces_skipped,
+        ),
+    )
+
 
 @router.post(
     "/multi-search",

backend/app/schemas/face_clusters.py

@@ -1,5 +1,5 @@
 from pydantic import BaseModel
-from typing import List, Optional, Dict, Union, Any
+from typing import List, Literal, Optional, Dict, Union, Any
 
 from app.schemas.videos import VideoData
 
@@ -80,16 +80,32 @@ class GetClusterImagesResponse(BaseModel):
     data: Optional[GetClusterImagesData] = None
 
 
-class GlobalReclusterData(BaseModel):
+class GlobalReclusterStartData(BaseModel):
+    task_id: str
+
+
+class GlobalReclusterStartResponse(BaseModel):
+    """Returned immediately when a global reclustering job is started."""
+
+    success: bool
+    message: Optional[str] = None
+    error: Optional[str] = None
+    data: Optional[GlobalReclusterStartData] = None
+
+
+class GlobalReclusterStatusData(BaseModel):
+    status: Literal["running", "complete", "error"]
     clusters_created: Optional[int] = None
     faces_skipped: Optional[int] = None
 
 
-class GlobalReclusterResponse(BaseModel):
+class GlobalReclusterStatusResponse(BaseModel):
+    """Polled to check on the progress of a global reclustering job."""
+
     success: bool
     message: Optional[str] = None
     error: Optional[str] = None
-    data: Optional[GlobalReclusterData] = None
+    data: Optional[GlobalReclusterStatusData] = None
 
 
 class MultiPersonSearchRequest(BaseModel):

backend/main.py

@@ -45,7 +45,10 @@
 from app.routes.albums import router as albums_router
 from app.routes.images import router as images_router
 from app.routes.videos import router as videos_router
-from app.routes.face_clusters import router as face_clusters_router
+from app.routes.face_clusters import (
+    router as face_clusters_router,
+    _cleanup_stale_recluster_tasks,
+)
 from app.routes.user_preferences import router as user_preferences_router
 from app.routes.memories import router as memories_router
 from app.routes.shutdown import router as shutdown_router
@@ -105,7 +108,6 @@ async def lifespan(app: FastAPI):
     semantic_util_sync_vocabulary()
     # New pool, just for folder indexing, so it never queues behind AI tagging
     app.state.indexing_executor = ProcessPoolExecutor(max_workers=INDEXING_MAX_WORKERS)
-    # Create ProcessPoolExecutor and attach it to app.state
     app.state.executor = ProcessPoolExecutor(max_workers=1)
     # Self-gating no-ops unless something is missing/stale (fresh install,
     # checkpoint swap, edited seed). Single-worker executor runs them in
@@ -123,12 +125,16 @@ async def lifespan(app: FastAPI):
 
     # Start the SSE model download cleanup task
     cleanup_task = asyncio.create_task(_cleanup_stale_tasks())
+    recluster_cleanup_task = asyncio.create_task(_cleanup_stale_recluster_tasks())
 
     try:
         yield
     finally:
         cleanup_task.cancel()
-        await asyncio.gather(cleanup_task, return_exceptions=True)
+        recluster_cleanup_task.cancel()
+        await asyncio.gather(
+            cleanup_task, recluster_cleanup_task, return_exceptions=True
+        )
         app.state.metadata_export_debouncer.close()
         app.state.indexing_executor.shutdown(wait=True)
         # Closes the only socket bound outside localhost; the in-memory share

backend/run-server.ps1

@@ -44,13 +44,12 @@ if ($args[0] -eq "--test") {
         Start-Sleep -Seconds 3
     }
 } else {
-    # Get the WORKERS environment variable or use a default value
-    $workers = if ($env:WORKERS) { $env:WORKERS } else { "1" }
-    Write-Host "WORKERS: $workers"
+    # Model-download and reclustering jobs are tracked in memory, per worker, so
+    # a second worker would answer status polls it knows nothing about.
 
     # Start the Hypercorn server
     $process = Start-Process -FilePath "hypercorn" `
-                             -ArgumentList "main:app --workers $workers --bind 0.0.0.0:8000" `
+                             -ArgumentList "main:app --workers 1 --bind 0.0.0.0:8000" `
                              -PassThru
 
     # Wait for process termination or Ctrl+C

backend/run.sh

@@ -21,7 +21,7 @@ if [[ $1 == "--test" ]]; then
         sleep 3
     done
 else
-    # print the value of the WORKERS environment variable
-    echo "WORKERS: ${WORKERS:-1}"
-    hypercorn main:app --workers ${WORKERS:-1} --bind 0.0.0.0:8000
+    # Model-download and reclustering jobs are tracked in memory, per worker, so
+    # a second worker would answer status polls it knows nothing about.
+    hypercorn main:app --workers 1 --bind 0.0.0.0:8000
 fi