fix: prevent concurrent scans with Redis lock, add analysis start endpoint

- Add POST /api/v1/library/analysis/start endpoint to manually trigger analysis
- Use Redis-based locking to prevent concurrent scans across uvicorn workers
- Trigger analysis after scan for relocated tracks (not just new/updated)
- Check lock expiry and heartbeat staleness to avoid permanent locks

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: jeffcrouse

backend/app/api/routes/library.py

@@ -393,6 +393,44 @@ async def get_analysis_status(db: DbSession) -> AnalysisStatus:
     )
 
 
+class AnalysisStartResponse(BaseModel):
+    """Response for starting analysis."""
+
+    status: str
+    queued: int = 0
+    message: str
+
+
+@router.post("/analysis/start", response_model=AnalysisStartResponse)
+async def start_analysis(limit: int = 500) -> AnalysisStartResponse:
+    """Manually trigger analysis for unanalyzed tracks.
+
+    This queues tracks for analysis in the background. Use GET /analysis/status
+    to monitor progress.
+    """
+    from app.services.tasks import queue_unanalyzed_tracks
+
+    try:
+        queued = await queue_unanalyzed_tracks(limit=limit)
+        if queued == 0:
+            return AnalysisStartResponse(
+                status="complete",
+                queued=0,
+                message="All tracks are already analyzed",
+            )
+        return AnalysisStartResponse(
+            status="started",
+            queued=queued,
+            message=f"Queued {queued} tracks for analysis",
+        )
+    except Exception as e:
+        return AnalysisStartResponse(
+            status="error",
+            queued=0,
+            message=f"Failed to start analysis: {e}",
+        )
+
+
 # ============================================================================
 # Missing Tracks API
 # ============================================================================

backend/app/services/background.py

@@ -120,21 +120,54 @@ async def shutdown(self) -> None:
         logger.info("BackgroundManager shutdown complete")
 
     def is_scan_running(self) -> bool:
-        """Check if a library scan is currently running."""
+        """Check if a library scan is currently running (across all workers).
+
+        Uses Redis to coordinate across multiple uvicorn workers.
+        """
+        # Check local task first
         if self._current_scan_task and not self._current_scan_task.done():
             return True
 
-        # Also check Redis for stale progress
+        # Check Redis lock (shared across all workers)
         try:
+            if self.redis.get("familiar:scan:lock"):
+                return True
+
+            # Also check progress status
             data: bytes | None = self.redis.get("familiar:scan:progress")  # type: ignore[assignment]
             if data:
                 progress = json.loads(data)
-                return progress.get("status") == "running"
+                if progress.get("status") == "running":
+                    # Check if heartbeat is recent (within 2 minutes)
+                    heartbeat = progress.get("last_heartbeat")
+                    if heartbeat:
+                        from datetime import datetime, timedelta
+                        try:
+                            hb_time = datetime.fromisoformat(heartbeat)
+                            if datetime.now() - hb_time < timedelta(minutes=2):
+                                return True
+                        except (ValueError, TypeError):
+                            pass
         except Exception:
             pass
 
         return False
 
+    def _acquire_scan_lock(self) -> bool:
+        """Try to acquire the scan lock in Redis. Returns True if acquired."""
+        try:
+            # SET NX (only if not exists) with 1 hour expiry
+            return bool(self.redis.set("familiar:scan:lock", "1", nx=True, ex=3600))
+        except Exception:
+            return False
+
+    def _release_scan_lock(self) -> None:
+        """Release the scan lock in Redis."""
+        try:
+            self.redis.delete("familiar:scan:lock")
+        except Exception:
+            pass
+
     async def run_cpu_bound(self, func: Callable, *args: Any) -> Any:
         """Run CPU-bound function in process pool (spawned, not forked).
 
@@ -148,11 +181,17 @@ async def run_scan(self, full_scan: bool = False) -> dict[str, Any]:
         """Start a library scan in the background.
 
         Returns immediately with status. Progress is reported via Redis.
+        Uses Redis-based locking to prevent concurrent scans across workers.
         """
         async with self._lock:
+            # Check if already running (local or other worker)
             if self.is_scan_running():
                 return {"status": "already_running"}
 
+            # Try to acquire Redis lock
+            if not self._acquire_scan_lock():
+                return {"status": "already_running"}
+
             # Create task and store reference
             self._current_scan_task = asyncio.create_task(
                 self._do_scan(full_scan)
@@ -178,6 +217,7 @@ async def _do_scan(self, full_scan: bool) -> dict[str, Any]:
             return {"status": "error", "error": str(e)}
         finally:
             self._current_scan_task = None
+            self._release_scan_lock()
 
     async def run_analysis(self, track_id: str) -> dict[str, Any]:
         """Queue a track for analysis.

backend/app/services/tasks.py

@@ -355,9 +355,11 @@ async def run_library_scan(full_scan: bool = False) -> dict[str, Any]:
             recovered=results.get("recovered", 0),
         )
 
-        # Queue analysis for new/updated tracks
-        if results["new"] > 0 or results["updated"] > 0:
-            queued_count = await queue_unanalyzed_tracks(limit=results["new"] + results["updated"])
+        # Queue analysis for unanalyzed tracks (new, updated, or relocated)
+        # Also catches any tracks that were missed in previous scans
+        unanalyzed_count = results["new"] + results["updated"] + results.get("relocated", 0)
+        if unanalyzed_count > 0:
+            queued_count = await queue_unanalyzed_tracks(limit=max(unanalyzed_count, 500))
             logger.info(f"Queued {queued_count} tracks for analysis")
 
         logger.info(f"Scan complete: {results}")