diff --git a/backend/app/routes/face_clusters.py b/backend/app/routes/face_clusters.py index 46cb62506..0cbe90f7d 100644 --- a/backend/app/routes/face_clusters.py +++ b/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", diff --git a/backend/app/schemas/face_clusters.py b/backend/app/schemas/face_clusters.py index acd342be3..dbf251c4b 100644 --- a/backend/app/schemas/face_clusters.py +++ b/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): diff --git a/backend/main.py b/backend/main.py index 00c0f63d0..ae061f5c2 100644 --- a/backend/main.py +++ b/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 diff --git a/backend/run-server.ps1 b/backend/run-server.ps1 index d21d635b6..de58c38a3 100644 --- a/backend/run-server.ps1 +++ b/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 diff --git a/backend/run.sh b/backend/run.sh index dd3e0283d..68c348057 100644 --- a/backend/run.sh +++ b/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 \ No newline at end of file diff --git a/backend/tests/test_face_clusters.py b/backend/tests/test_face_clusters.py index dc76ce38d..2dbc08c71 100644 --- a/backend/tests/test_face_clusters.py +++ b/backend/tests/test_face_clusters.py @@ -1,9 +1,14 @@ +import asyncio +import contextlib import sqlite3 from contextlib import closing import pytest +import threading +import time import numpy as np -from unittest.mock import patch +from datetime import datetime, timedelta, timezone +from unittest.mock import AsyncMock, patch from fastapi import FastAPI import app.database.connection as connection_db import app.database.face_clusters as face_clusters_db @@ -23,6 +28,7 @@ ) from app.utils.face_quality import face_passes_quality_gate from fastapi.testclient import TestClient +from app.routes import face_clusters as face_clusters_module from app.routes.face_clusters import router as face_clusters_router app = FastAPI() @@ -483,6 +489,162 @@ def test_unsupported_http_methods(self, method, endpoint): assert response.status_code == 405 +# ============================================================================ +# Global Reclustering Background Job Tests +# ============================================================================ + + +def poll_until_terminal(job_client, task_id, timeout=10.0): + """Poll a reclustering task until it leaves the 'running' state.""" + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + response = job_client.get(f"/face_clusters/global-recluster/{task_id}") + assert response.status_code == 200 + payload = response.json() + if payload["data"]["status"] != "running": + return payload + time.sleep(0.05) + + raise AssertionError(f"Task {task_id} did not reach a terminal state in {timeout}s") + + +class TestGlobalReclusterAPI: + """TestClient runs as a context manager so POST and the later GET share one + event loop; a per-request loop would drop the job's asyncio task between them.""" + + @pytest.fixture(autouse=True) + def reset_recluster_state(self): + """Reset module-level job state so tests cannot leak into each other.""" + face_clusters_module.recluster_tasks.clear() + face_clusters_module._active_recluster_task_id = None + yield + face_clusters_module.recluster_tasks.clear() + face_clusters_module._active_recluster_task_id = None + + @patch("app.routes.face_clusters.cluster_util_face_clusters_sync") + def test_start_global_recluster_returns_202_and_completes(self, mock_recluster): + """Starting a job returns 202 with a task_id, and polling that task_id + eventually yields the clustering result.""" + mock_recluster.return_value = (3, 1) + + with TestClient(app) as job_client: + start_response = job_client.post("/face_clusters/global-recluster") + + assert start_response.status_code == 202 + start_payload = start_response.json() + assert start_payload["success"] is True + task_id = start_payload["data"]["task_id"] + assert task_id + + status_payload = poll_until_terminal(job_client, task_id) + + assert status_payload["success"] is True + assert status_payload["data"]["status"] == "complete" + assert status_payload["data"]["clusters_created"] == 3 + assert status_payload["data"]["faces_skipped"] == 1 + mock_recluster.assert_called_once_with(force_full_reclustering=True) + + @patch("app.routes.face_clusters.cluster_util_face_clusters_sync") + def test_second_trigger_rejoins_running_job(self, mock_recluster): + """A trigger while a job is in flight rejoins its task_id instead of + starting a second pass that would race on the cluster tables.""" + release = threading.Event() + + def blocking_recluster(**_kwargs): + release.wait(timeout=10) + return (2, 0) + + mock_recluster.side_effect = blocking_recluster + + with TestClient(app) as job_client: + first_response = job_client.post("/face_clusters/global-recluster") + second_response = job_client.post("/face_clusters/global-recluster") + + assert first_response.status_code == 202 + assert second_response.status_code == 202 + + first_task_id = first_response.json()["data"]["task_id"] + assert second_response.json()["data"]["task_id"] == first_task_id + + # Let the blocked job finish so the client context can exit cleanly. + release.set() + status_payload = poll_until_terminal(job_client, first_task_id) + + assert status_payload["data"]["status"] == "complete" + assert mock_recluster.call_count == 1 + + @patch("app.routes.face_clusters.cluster_util_face_clusters_sync") + def test_failed_job_reports_error_status(self, mock_recluster): + """A job that raises is reported as an error result to the poller + rather than surfacing as a failed HTTP request.""" + mock_recluster.side_effect = RuntimeError("clustering blew up") + + with TestClient(app) as job_client: + start_response = job_client.post("/face_clusters/global-recluster") + task_id = start_response.json()["data"]["task_id"] + + status_payload = poll_until_terminal(job_client, task_id) + + assert status_payload["success"] is False + assert status_payload["data"]["status"] == "error" + assert "clustering blew up" in status_payload["message"] + + def test_status_unknown_task_id_returns_404(self): + """Polling an unknown task_id returns 404 rather than hanging or 500.""" + response = client.get("/face_clusters/global-recluster/does-not-exist") + + assert response.status_code == 404 + detail = response.json()["detail"] + assert detail["success"] is False + assert detail["error"] == "Task Not Found" + + @pytest.mark.asyncio + async def test_cleanup_reaps_tasks_past_ttl(self): + """A finished task past RECLUSTER_TASK_TTL_MINUTES is reaped and a fresher + one kept. asyncio.sleep is patched so the 5-minute loop runs in the test.""" + stale_entry = face_clusters_module.ReclusterTask(status="complete") + stale_entry.finished_at = datetime.now(timezone.utc) - timedelta( + minutes=face_clusters_module.RECLUSTER_TASK_TTL_MINUTES + 1 + ) + fresh_entry = face_clusters_module.ReclusterTask(status="complete") + fresh_entry.finished_at = datetime.now(timezone.utc) + + face_clusters_module.recluster_tasks["stale-task"] = stale_entry + face_clusters_module.recluster_tasks["fresh-task"] = fresh_entry + + # A bare AsyncMock never suspends, so the loop's `while True` would + # starve the event loop; route it through a real zero-duration sleep. + real_sleep = asyncio.sleep + + async def instant_sleep(*_args, **_kwargs): + await real_sleep(0) + + with patch( + "app.routes.face_clusters.asyncio.sleep", + new=AsyncMock(side_effect=instant_sleep), + ): + cleanup_task = asyncio.create_task( + face_clusters_module._cleanup_stale_recluster_tasks() + ) + try: + + async def until_reaped(): + while "stale-task" in face_clusters_module.recluster_tasks: + await asyncio.sleep(0) + + await asyncio.wait_for(until_reaped(), timeout=2) + finally: + cleanup_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await cleanup_task + + assert "stale-task" not in face_clusters_module.recluster_tasks + assert "fresh-task" in face_clusters_module.recluster_tasks + + response = client.get("/face_clusters/global-recluster/stale-task") + assert response.status_code == 404 + + # ============================================================================ # Algorithmic Logic Tests # ============================================================================ @@ -555,7 +717,7 @@ def test_folder_size_regression(self, mock_db_get): ), f"The folder-size bug is present: expected 2 clusters, got {len(isolated_clusters)} in isolated run" # Verify points were assigned correctly (10 points per cluster) - cluster_counts = {} + cluster_counts: dict[str, int] = {} for r in results_isolated: cluster_counts[r.cluster_uuid] = cluster_counts.get(r.cluster_uuid, 0) + 1 @@ -642,7 +804,7 @@ def test_adaptive_eps_clamping_regression(self, mock_db_get): pt_b2 /= np.linalg.norm(pt_b2) # 5 Singletons (mutually orthogonal to each other and A/B) - singletons = [] + singletons: list[np.ndarray] = [] for _ in range(5): vec = np.random.randn(dim) vec -= np.dot(vec, center_a) * center_a @@ -666,7 +828,7 @@ def test_adaptive_eps_clamping_regression(self, mock_db_get): ) # Group face_ids by their cluster UUIDs - clusters = {} + clusters: dict[str, list[int]] = {} for r in results: if r.cluster_uuid not in clusters: clusters[r.cluster_uuid] = [] diff --git a/backend/tests/test_metadata_export_routes.py b/backend/tests/test_metadata_export_routes.py index 4891e97f5..9b6758ae9 100644 --- a/backend/tests/test_metadata_export_routes.py +++ b/backend/tests/test_metadata_export_routes.py @@ -80,6 +80,15 @@ def _wait_until_idle(client: TestClient) -> dict: raise AssertionError("export never finished") +def _wait_for_recluster(client: TestClient, task_id: str) -> str: + for _ in range(200): + data = client.get(f"/face-clusters/global-recluster/{task_id}").json()["data"] + if data["status"] != "running": + return data["status"] + time.sleep(0.02) + raise AssertionError("recluster never finished") + + class TestExportEndpoints: def test_status_counts_png_images(self, client, library): _photo(library, "a") @@ -225,7 +234,11 @@ def test_person_rename_and_recluster(self, client, library, monkeypatch): "app.routes.face_clusters.cluster_util_face_clusters_sync", lambda force_full_reclustering=False: (0, 0), ) - self._expect(client, "POST", "/face-clusters/global-recluster", 1) + before = self._notified(client) + r = client.post("/face-clusters/global-recluster") + # Reclustering runs as a background job and requests the export when done. + assert _wait_for_recluster(client, r.json()["data"]["task_id"]) == "complete" + assert self._notified(client) - before == 1 def test_turning_the_toggle_on_catches_up_the_library(self, client, library): self._expect( diff --git a/docs/backend/backend_python/openapi.json b/docs/backend/backend_python/openapi.json index 28082477e..9adc50664 100644 --- a/docs/backend/backend_python/openapi.json +++ b/docs/backend/backend_python/openapi.json @@ -1741,15 +1741,15 @@ "Face Clusters" ], "summary": "Trigger Global Reclustering", - "description": "Manually trigger global face reclustering.\nThis forces full reclustering regardless of the 24-hour rule.", + "description": "Start a global face reclustering job in the background.\nThis forces full reclustering regardless of the 24-hour rule.\n\nReturns immediately with a task_id; poll\nGET /face-clusters/global-recluster/{task_id} for the result, since\nreclustering runs over every face embedding and can take a long time\non large libraries.\n\nIf a reclustering job is already running, its task_id is returned instead\nof starting a second (concurrent runs would race on the cluster tables).", "operationId": "trigger_global_reclustering_face_clusters_global_recluster_post", "responses": { - "200": { + "202": { "description": "Successful Response", "content": { "application/json": { "schema": { - "$ref": "#/components/schemas/GlobalReclusterResponse" + "$ref": "#/components/schemas/GlobalReclusterStartResponse" } } } @@ -1767,6 +1767,69 @@ } } }, + "/face-clusters/global-recluster/{task_id}": { + "get": { + "tags": [ + "Face Clusters" + ], + "summary": "Get Global Recluster Status", + "description": "Poll the status of a previously started global reclustering job.", + "operationId": "get_global_recluster_status_face_clusters_global_recluster__task_id__get", + "parameters": [ + { + "name": "task_id", + "in": "path", + "required": true, + "schema": { + "type": "string", + "title": "Task Id" + } + } + ], + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/GlobalReclusterStatusResponse" + } + } + } + }, + "404": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/app__schemas__face_clusters__ErrorResponse" + } + } + }, + "description": "Not Found" + }, + "500": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/app__schemas__face_clusters__ErrorResponse" + } + } + }, + "description": "Internal Server Error" + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + } + } + }, "/face-clusters/multi-search": { "post": { "tags": [ @@ -4207,8 +4270,76 @@ "title": "GetUserPreferencesResponse", "description": "Response model for getting user preferences" }, - "GlobalReclusterData": { + "GlobalReclusterStartData": { "properties": { + "task_id": { + "type": "string", + "title": "Task Id" + } + }, + "type": "object", + "required": [ + "task_id" + ], + "title": "GlobalReclusterStartData" + }, + "GlobalReclusterStartResponse": { + "properties": { + "success": { + "type": "boolean", + "title": "Success" + }, + "message": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Message" + }, + "error": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Error" + }, + "data": { + "anyOf": [ + { + "$ref": "#/components/schemas/GlobalReclusterStartData" + }, + { + "type": "null" + } + ] + } + }, + "type": "object", + "required": [ + "success" + ], + "title": "GlobalReclusterStartResponse", + "description": "Returned immediately when a global reclustering job is started." + }, + "GlobalReclusterStatusData": { + "properties": { + "status": { + "type": "string", + "enum": [ + "running", + "complete", + "error" + ], + "title": "Status" + }, "clusters_created": { "anyOf": [ { @@ -4233,9 +4364,12 @@ } }, "type": "object", - "title": "GlobalReclusterData" + "required": [ + "status" + ], + "title": "GlobalReclusterStatusData" }, - "GlobalReclusterResponse": { + "GlobalReclusterStatusResponse": { "properties": { "success": { "type": "boolean", @@ -4266,7 +4400,7 @@ "data": { "anyOf": [ { - "$ref": "#/components/schemas/GlobalReclusterData" + "$ref": "#/components/schemas/GlobalReclusterStatusData" }, { "type": "null" @@ -4278,7 +4412,8 @@ "required": [ "success" ], - "title": "GlobalReclusterResponse" + "title": "GlobalReclusterStatusResponse", + "description": "Polled to check on the progress of a global reclustering job." }, "HTTPValidationError": { "properties": { diff --git a/frontend/src/api/api-functions/face_clusters.ts b/frontend/src/api/api-functions/face_clusters.ts index 7d521059e..0da44b846 100644 --- a/frontend/src/api/api-functions/face_clusters.ts +++ b/frontend/src/api/api-functions/face_clusters.ts @@ -1,5 +1,5 @@ import { faceClustersEndpoints } from '../apiEndpoints'; -import { apiClient } from '../axiosConfig'; +import { apiClient, LONG_REQUEST_TIMEOUT_MS } from '../axiosConfig'; import { APIResponse } from '@/types/API'; import { BackendRes } from '@/hooks/useQueryExtension'; import type { Image, ImageMetadata, Video } from '@/types/Media'; @@ -90,6 +90,7 @@ export const fetchSearchedFaces = async ( const response = await apiClient.post( faceClustersEndpoints.searchForFaces, request, + { timeout: LONG_REQUEST_TIMEOUT_MS }, ); return response.data; }; @@ -100,24 +101,43 @@ export const fetchSearchedFacesBase64 = async ( const response = await apiClient.post( faceClustersEndpoints.searchForFacesBase64, request, + { timeout: LONG_REQUEST_TIMEOUT_MS }, ); return response.data; }; -export interface GlobalReclusterData { +export interface GlobalReclusterStartData { + task_id: string; +} + +export interface GlobalReclusterStatusData { + status: 'running' | 'complete' | 'error'; clusters_created: number | null; faces_skipped: number | null; } -export const triggerGlobalReclustering = async (): Promise< - BackendRes +// Reclustering scans every embedding and can exceed any HTTP timeout, so the +// backend runs it as a background job; this kicks it off and returns a task_id. +export const startGlobalReclustering = async (): Promise< + BackendRes > => { - const response = await apiClient.post>( + const response = await apiClient.post>( faceClustersEndpoints.globalRecluster, ); return response.data; }; +// Poll this with the task_id returned by startGlobalReclustering until +// status is 'complete' or 'error'. +export const getGlobalReclusterStatus = async ( + taskId: string, +): Promise> => { + const response = await apiClient.get>( + faceClustersEndpoints.globalReclusterStatus(taskId), + ); + return response.data; +}; + export interface MultiPersonSearchRequest { cluster_ids: string[]; match_mode: 'match_any' | 'match_all'; @@ -129,6 +149,7 @@ export const fetchMultiPersonSearch = async ( const response = await apiClient.post>( faceClustersEndpoints.multiPersonSearch, request, + { timeout: LONG_REQUEST_TIMEOUT_MS }, ); return response.data; }; diff --git a/frontend/src/api/apiEndpoints.ts b/frontend/src/api/apiEndpoints.ts index 47e4067b7..5fbf52cfe 100644 --- a/frontend/src/api/apiEndpoints.ts +++ b/frontend/src/api/apiEndpoints.ts @@ -29,6 +29,8 @@ export const faceClustersEndpoints = { renameCluster: (clusterId: string) => `/face-clusters/${clusterId}`, getClusterImages: (clusterId: string) => `/face-clusters/${clusterId}/images`, globalRecluster: '/face-clusters/global-recluster', + globalReclusterStatus: (taskId: string) => + `/face-clusters/global-recluster/${encodeURIComponent(taskId)}`, multiPersonSearch: '/face-clusters/multi-search', }; diff --git a/frontend/src/api/axiosConfig.ts b/frontend/src/api/axiosConfig.ts index bc7fc31fb..cf31af971 100644 --- a/frontend/src/api/axiosConfig.ts +++ b/frontend/src/api/axiosConfig.ts @@ -1,10 +1,15 @@ import axios from 'axios'; import { BACKEND_URL, SYNC_MICROSERVICE_URL } from '@/config/Backend'; +const DEFAULT_TIMEOUT_MS = 30000; + +// Passed per call by endpoints that scan the whole library, like face search. +export const LONG_REQUEST_TIMEOUT_MS = 120000; + // Create simple axios instance with basic configuration export const apiClient = axios.create({ baseURL: BACKEND_URL, - timeout: 10000, // 10 seconds timeout + timeout: DEFAULT_TIMEOUT_MS, headers: { 'Content-Type': 'application/json', Accept: 'application/json', @@ -12,7 +17,7 @@ export const apiClient = axios.create({ }); export const syncApiClient = axios.create({ baseURL: SYNC_MICROSERVICE_URL, - timeout: 10000, // 10 seconds timeout + timeout: DEFAULT_TIMEOUT_MS, headers: { 'Content-Type': 'application/json', Accept: 'application/json', diff --git a/frontend/src/hooks/__tests__/useGlobalRecluster.test.tsx b/frontend/src/hooks/__tests__/useGlobalRecluster.test.tsx new file mode 100644 index 000000000..440e4741f --- /dev/null +++ b/frontend/src/hooks/__tests__/useGlobalRecluster.test.tsx @@ -0,0 +1,188 @@ +import React from 'react'; +import { renderHook, act } from '@testing-library/react'; +import { QueryClient, QueryClientProvider } from '@tanstack/react-query'; +import { useGlobalRecluster } from '@/hooks/useGlobalRecluster'; +import { + startGlobalReclustering, + getGlobalReclusterStatus, +} from '@/api/api-functions/face_clusters'; + +jest.mock('@/api/api-functions/face_clusters', () => ({ + startGlobalReclustering: jest.fn(), + getGlobalReclusterStatus: jest.fn(), +})); + +const mockStart = startGlobalReclustering as jest.MockedFunction< + typeof startGlobalReclustering +>; +const mockStatus = getGlobalReclusterStatus as jest.MockedFunction< + typeof getGlobalReclusterStatus +>; + +const POLL_INTERVAL_MS = 2000; + +const startOk = { success: true, message: 'started', data: { task_id: 'abc' } }; +const running = { + success: true, + data: { + status: 'running' as const, + clusters_created: null, + faces_skipped: null, + }, +}; +const complete = { + success: true, + message: 'done', + data: { status: 'complete' as const, clusters_created: 3, faces_skipped: 1 }, +}; +const errored = { + success: false, + message: 'reclustering failed', + data: { + status: 'error' as const, + clusters_created: null, + faces_skipped: null, + }, +}; + +function setup() { + const queryClient = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }); + const invalidateSpy = jest.spyOn(queryClient, 'invalidateQueries'); + const wrapper = ({ children }: { children: React.ReactNode }) => ( + {children} + ); + const utils = renderHook(() => useGlobalRecluster(), { wrapper }); + return { ...utils, invalidateSpy }; +} + +// Flush pending promises and advance fake timers together. +const flush = async (ms = 0) => { + await act(async () => { + await Promise.resolve(); + await jest.advanceTimersByTimeAsync(ms); + }); +}; + +beforeEach(() => { + jest.useFakeTimers(); + mockStart.mockReset(); + mockStatus.mockReset(); +}); + +afterEach(() => { + jest.useRealTimers(); +}); + +describe('useGlobalRecluster', () => { + test('polls until the job completes successfully', async () => { + mockStart.mockResolvedValue(startOk); + mockStatus.mockResolvedValueOnce(running).mockResolvedValueOnce(complete); + + const { result, invalidateSpy } = setup(); + + act(() => { + result.current.trigger(); + }); + expect(result.current.isPending).toBe(true); + + await flush(); // start resolves, first poll -> running + expect(mockStatus).toHaveBeenCalledTimes(1); + expect(result.current.isPending).toBe(true); + + await flush(POLL_INTERVAL_MS); // scheduled poll -> complete + expect(result.current.isSuccess).toBe(true); + expect(result.current.successData).toEqual(complete.data); + expect(result.current.successMessage).toBe('done'); + expect(invalidateSpy).toHaveBeenCalledWith({ queryKey: ['clusters'] }); + }); + + test('sets an error when the job reports failure', async () => { + mockStart.mockResolvedValue(startOk); + mockStatus.mockResolvedValue(errored); + + const { result, invalidateSpy } = setup(); + act(() => { + result.current.trigger(); + }); + await flush(); + + expect(result.current.isError).toBe(true); + expect(result.current.errorMessage).toBe('reclustering failed'); + expect(invalidateSpy).toHaveBeenCalledWith({ queryKey: ['clusters'] }); + }); + + test('sets an error when starting the job fails', async () => { + mockStart.mockRejectedValue(new Error('network down')); + + const { result } = setup(); + act(() => { + result.current.trigger(); + }); + await flush(); + + expect(result.current.isError).toBe(true); + expect(mockStatus).not.toHaveBeenCalled(); + }); + + test('sets an error when the status request itself rejects (e.g. 404 for an aged-out task)', async () => { + mockStart.mockResolvedValue(startOk); + const notFound = { + isAxiosError: true, + message: 'Request failed with status code 404', + code: 'ERR_BAD_REQUEST', + response: { status: 404, data: {} }, + }; + mockStatus.mockRejectedValue(notFound); + + const { result, invalidateSpy } = setup(); + act(() => { + result.current.trigger(); + }); + await flush(); + + expect(result.current.isError).toBe(true); + expect(result.current.errorMessage).toBe( + 'ERR_BAD_REQUEST: Request failed with status code 404', + ); + expect(result.current.error).toBeInstanceOf(Error); + expect(invalidateSpy).not.toHaveBeenCalled(); + }); + + test('stops polling after unmount', async () => { + mockStart.mockResolvedValue(startOk); + mockStatus.mockResolvedValue(running); + + const { result, unmount } = setup(); + act(() => { + result.current.trigger(); + }); + await flush(); // first poll + expect(mockStatus).toHaveBeenCalledTimes(1); + + unmount(); + await flush(POLL_INTERVAL_MS * 3); + // No further polls once unmounted. + expect(mockStatus).toHaveBeenCalledTimes(1); + }); + + test('a rapid second trigger does not leave a second poll loop running', async () => { + mockStart.mockResolvedValue(startOk); + mockStatus.mockResolvedValue(running); + + const { result } = setup(); + // Both triggers fire before startGlobalReclustering resolves. + act(() => { + result.current.trigger(); + result.current.trigger(); + }); + + await flush(); // both starts resolve; only the latest run may poll + expect(mockStatus).toHaveBeenCalledTimes(1); + + const callsBefore = mockStatus.mock.calls.length; + await flush(POLL_INTERVAL_MS); // exactly one loop advances + expect(mockStatus.mock.calls.length).toBe(callsBefore + 1); + }); +}); diff --git a/frontend/src/hooks/useGlobalRecluster.tsx b/frontend/src/hooks/useGlobalRecluster.tsx new file mode 100644 index 000000000..71e4ebf5f --- /dev/null +++ b/frontend/src/hooks/useGlobalRecluster.tsx @@ -0,0 +1,138 @@ +import { useCallback, useEffect, useRef, useState } from 'react'; +import { useQueryClient } from '@tanstack/react-query'; +import { + startGlobalReclustering, + getGlobalReclusterStatus, + GlobalReclusterStatusData, +} from '@/api/api-functions/face_clusters'; +import { getErrorMessage } from '@/lib/utils'; + +const POLL_INTERVAL_MS = 2000; + +interface ReclusterState { + isPending: boolean; + isSuccess: boolean; + isError: boolean; + error: Error | undefined; + successData: GlobalReclusterStatusData | undefined; + successMessage: string | undefined; + errorMessage: string | undefined; +} + +const idleState: ReclusterState = { + isPending: false, + isSuccess: false, + isError: false, + error: undefined, + successData: undefined, + successMessage: undefined, + errorMessage: undefined, +}; + +export type UseGlobalReclusterResult = ReclusterState & { trigger: () => void }; + +// Rejections can be plain objects, not Errors, but state promises an Error. +const toError = (err: unknown): Error => + err instanceof Error ? err : new Error(getErrorMessage(err)); + +// Starts the backend's recluster job and polls it until it completes or fails. +export function useGlobalRecluster(): UseGlobalReclusterResult { + const queryClient = useQueryClient(); + const pollTimeoutRef = useRef | null>(null); + // Identifies the latest trigger() call. A callback bails if a newer trigger + // (or unmount) has since bumped this, so stale runs can't leak updates. + const runIdRef = useRef(0); + const [state, setState] = useState(idleState); + + const stopPolling = useCallback(() => { + if (pollTimeoutRef.current) { + clearTimeout(pollTimeoutRef.current); + pollTimeoutRef.current = null; + } + }, []); + + // On unmount, invalidate any in-flight run and stop polling. + useEffect(() => { + return () => { + runIdRef.current += 1; + stopPolling(); + }; + }, [stopPolling]); + + const trigger = useCallback(() => { + const runId = runIdRef.current + 1; + runIdRef.current = runId; + const isActive = () => runId === runIdRef.current; + + stopPolling(); + setState({ ...idleState, isPending: true }); + + startGlobalReclustering() + .then((startRes) => { + if (!isActive()) return; + + const taskId = startRes.data?.task_id; + if (!taskId) { + throw new Error('Backend did not return a task_id for reclustering.'); + } + + // Self-scheduling poll: the next tick is only queued after the current + // request resolves, so requests can't stack up or overlap. + const poll = async () => { + try { + const statusRes = await getGlobalReclusterStatus(taskId); + if (!isActive()) return; + + if (statusRes.data?.status === 'running') { + pollTimeoutRef.current = setTimeout(poll, POLL_INTERVAL_MS); + return; + } + + stopPolling(); + queryClient.invalidateQueries({ queryKey: ['clusters'] }); + + if (!statusRes.success || statusRes.data?.status === 'error') { + setState({ + ...idleState, + isError: true, + error: new Error( + statusRes.message || 'Global reclustering failed.', + ), + errorMessage: statusRes.message, + }); + return; + } + + setState({ + ...idleState, + isSuccess: true, + successData: statusRes.data, + successMessage: statusRes.message, + }); + } catch (err: unknown) { + if (!isActive()) return; + stopPolling(); + setState({ + ...idleState, + isError: true, + error: toError(err), + errorMessage: getErrorMessage(err), + }); + } + }; + + poll(); + }) + .catch((err: unknown) => { + if (!isActive()) return; + setState({ + ...idleState, + isError: true, + error: toError(err), + errorMessage: getErrorMessage(err), + }); + }); + }, [stopPolling, queryClient]); + + return { trigger, ...state }; +} diff --git a/frontend/src/pages/SettingsPage/components/ApplicationControlsCard.tsx b/frontend/src/pages/SettingsPage/components/ApplicationControlsCard.tsx index dc81ab7d7..525a68801 100644 --- a/frontend/src/pages/SettingsPage/components/ApplicationControlsCard.tsx +++ b/frontend/src/pages/SettingsPage/components/ApplicationControlsCard.tsx @@ -1,4 +1,4 @@ -import React, { useState } from 'react'; +import React, { useEffect, useState } from 'react'; import { Settings as SettingsIcon, RefreshCw, Users } from 'lucide-react'; import { Button } from '@/components/ui/button'; @@ -9,8 +9,7 @@ import { useUpdater } from '@/hooks/useUpdater'; import { useDispatch } from 'react-redux'; import { showLoader, hideLoader } from '@/features/loaderSlice'; import { showInfoDialog } from '@/features/infoDialogSlice'; -import { triggerGlobalReclustering } from '@/api/api-functions/face_clusters'; -import { usePictoMutation } from '@/hooks/useQueryExtension'; +import { useGlobalRecluster } from '@/hooks/useGlobalRecluster'; import { useMutationFeedback } from '@/hooks/useMutationFeedback'; import { showGlobalAlert } from '@/features/globalAlertSlice'; @@ -32,11 +31,11 @@ const ApplicationControlsCard: React.FC = () => { const [updateDialogOpen, setUpdateDialogOpen] = useState(false); - const reclusterMutation = usePictoMutation({ - mutationFn: triggerGlobalReclustering, - autoInvalidateTags: ['clusters'], - onSuccess: (data) => { - const facesSkipped = data.data?.faces_skipped; + const reclusterJob = useGlobalRecluster(); + + useEffect(() => { + if (reclusterJob.isSuccess) { + const facesSkipped = reclusterJob.successData?.faces_skipped; if (facesSkipped != null && facesSkipped > 0) { dispatch( @@ -46,34 +45,29 @@ const ApplicationControlsCard: React.FC = () => { }), ); } - }, - }); + } + }, [reclusterJob.isSuccess, reclusterJob.successData, dispatch]); const feedbackOptions = React.useMemo( () => ({ - loadingMessage: 'Starting global face reclustering...', + loadingMessage: + 'Reclustering faces... this can take a while on large libraries.', successTitle: 'Reclustering Completed', successMessage: - reclusterMutation.successMessage || + reclusterJob.successMessage || 'Global face reclustering completed successfully.', errorTitle: 'Reclustering Failed', errorMessage: - reclusterMutation.errorMessage || + reclusterJob.errorMessage || 'Failed to complete global face reclustering.', - // You can use reclusterMutation.successData?.clusters_created to show the number of clusters created if needed - // Example: `Clusters created: ${reclusterMutation.successData?.clusters_created}` }), - [ - reclusterMutation.successMessage, - reclusterMutation.errorMessage, - reclusterMutation.successData, - ], + [reclusterJob.successMessage, reclusterJob.errorMessage], ); - useMutationFeedback(reclusterMutation, feedbackOptions); + useMutationFeedback(reclusterJob, feedbackOptions); const onGlobalReclusterClick = () => { - reclusterMutation.mutate(undefined); + reclusterJob.trigger(); }; const onCheckUpdatesClick = () => {