fix: harden request handling, pipeline containment, and startup boundaries

backend/app/pipeline/grounding.py

@@ -7,8 +7,10 @@ from collections.abc import Callable
 from dataclasses import dataclass
 from enum import StrEnum
 
+import httpx2
 import structlog
 
+from app.adapters.spotify.errors import SpotifyError
 from app.config import Settings
 from app.domain.matching import candidate_key, judge_candidate_match, track_key
 from app.domain.models import Familiarity, Track, TrackCandidate
@@ -22,6 +24,7 @@ class ResolutionStatus(StrEnum):
     RESOLVED = "resolved"
     MISS = "miss"
     MISMATCH = "mismatch"
+    FAILED = "failed"
     QUOTA = "quota"
 
 
@@ -32,6 +35,7 @@ class GroundingMetrics:
     attempted_count: int
     miss_count: int
     mismatch_guard_count: int
+    failed_count: int
     cache_hit_count: int
     did_reach_deadline: bool
     did_exhaust_quota: bool
@@ -111,6 +115,7 @@ class Grounder:
             settings.resolution_cache_ttl_seconds,
             settings.resolution_cache_max_entries,
         )
+        self._semaphore = asyncio.Semaphore(settings.grounding_concurrency)
 
     async def ground(
         self,
@@ -119,6 +124,7 @@ class Grounder:
         known_track_ids: frozenset[str],
         familiarity: Familiarity,
         pool_target: int,
+        deadline_at: float,
     ) -> GroundingResult:
         """Resolve candidates until the target, deadline, or quota boundary."""
         accepted: dict[int, Track] = {}
@@ -127,18 +133,18 @@ class Grounder:
         metrics = _MutableMetrics()
         pending: dict[asyncio.Task[_ResolutionAttempt], int] = {}
         next_index = 0
-        deadline_at = time.monotonic() + self.settings.request_deadline_seconds
-        semaphore = asyncio.Semaphore(self.settings.grounding_concurrency)
         try:
             while next_index < len(candidates) or pending:
                 if len(accepted) >= pool_target:
                     break
+                if time.monotonic() >= deadline_at:
+                    metrics.did_reach_deadline = True
+                    break
                 next_index = self._launch_tasks(
                     catalog,
                     candidates,
                     pending,
                     next_index,
-                    semaphore,
                 )
                 if not pending:
                     break
@@ -183,12 +189,9 @@ class Grounder:
         candidates: tuple[TrackCandidate, ...],
         pending: dict[asyncio.Task[_ResolutionAttempt], int],
         next_index: int,
-        semaphore: asyncio.Semaphore,
     ) -> int:
         while next_index < len(candidates) and len(pending) < self.settings.grounding_concurrency:
-            task = asyncio.create_task(
-                self._resolve(catalog, next_index, candidates[next_index], semaphore)
-            )
+            task = asyncio.create_task(self._resolve(catalog, next_index, candidates[next_index]))
             pending[task] = next_index
             next_index += 1
         return next_index
@@ -228,9 +231,8 @@ class Grounder:
         catalog: MusicCatalog,
         index: int,
         candidate: TrackCandidate,
-        semaphore: asyncio.Semaphore,
     ) -> _ResolutionAttempt:
-        async with semaphore:
+        async with self._semaphore:
             return await self._resolve_with_slot(catalog, index, candidate)
 
     async def _resolve_with_slot(
@@ -265,6 +267,15 @@ class Grounder:
                 bare_results = []
         except CatalogQuotaExhaustedError:
             return _ResolutionAttempt(index=index, status=ResolutionStatus.QUOTA)
+        except (SpotifyError, httpx2.HTTPError) as error:
+            structlog.get_logger().info(
+                "candidate_unresolved",
+                title=candidate.title,
+                artist=candidate.artist,
+                status=ResolutionStatus.FAILED,
+                error_type=type(error).__name__,
+            )
+            return _ResolutionAttempt(index=index, status=ResolutionStatus.FAILED)
 
         if matched_track is not None:
             self.cache.put(key, matched_track)
@@ -289,6 +300,7 @@ class _MutableMetrics:
     attempted_count: int = 0
     miss_count: int = 0
     mismatch_guard_count: int = 0
+    failed_count: int = 0
     cache_hit_count: int = 0
     did_reach_deadline: bool = False
     did_exhaust_quota: bool = False
@@ -297,6 +309,7 @@ class _MutableMetrics:
         self.attempted_count += 1
         self.miss_count += attempt.status is ResolutionStatus.MISS
         self.mismatch_guard_count += attempt.status is ResolutionStatus.MISMATCH
+        self.failed_count += attempt.status is ResolutionStatus.FAILED
         self.cache_hit_count += attempt.is_cache_hit
         self.did_exhaust_quota = self.did_exhaust_quota or attempt.status is ResolutionStatus.QUOTA
 
@@ -305,6 +318,7 @@ class _MutableMetrics:
             attempted_count=self.attempted_count,
             miss_count=self.miss_count,
             mismatch_guard_count=self.mismatch_guard_count,
+            failed_count=self.failed_count,
             cache_hit_count=self.cache_hit_count,
             did_reach_deadline=self.did_reach_deadline,
             did_exhaust_quota=self.did_exhaust_quota,
@@ -340,6 +354,7 @@ def _log_grounding(result: GroundingResult) -> None:
         attempted_count=metrics.attempted_count,
         miss_rate=metrics.miss_rate,
         mismatch_guard_rate=metrics.mismatch_guard_rate,
+        failed_count=metrics.failed_count,
         cache_hits=metrics.cache_hit_count,
         deadline_reached=metrics.did_reach_deadline,
         quota_exhausted=metrics.did_exhaust_quota,