fix(insights): stop rarity-scores 500s and blind data-completeness timeouts (#375)

Production insights logs showed two recurring scheduler failures:

1. `release_rarity` failed every run with a 500 from
   `GET /api/internal/insights/rarity-scores`. The API logs traced it to a
   Neo4j `TransientError: MemoryPoolOutOfMemoryError` — `fetch_all_rarity_signals`
   fired eight full-graph Cypher scans concurrently via `asyncio.gather`, summing
   their working sets past `dbms.memory.transaction.total.max` (16 GiB). Two of
   the scans also used `size([(r)-[]-() | 1])`, which materialises a list element
   per relationship for every Release.

   Fix: run the eight queries sequentially (caps peak transaction memory at one
   query; fine for a daily background job) and replace the list-comprehension
   degree with `COUNT { }`. The endpoint now also catches `TransientError` and
   returns 503 instead of a bare 500, so the caller logs a meaningful, retryable
   failure.

2. `data_completeness` failed intermittently with `error=""` — an empty,
   undiagnosable log line. The cause was an `httpx.ReadTimeout` (whose `str()`
   is empty) at the 600s client timeout; the underlying full-table scan varies
   from ~230s to >600s.

   Fix: a `_describe_error` helper that always includes the exception type
   (applied to both the data-completeness and rarity handlers), and a larger
   timeout (1200s) giving the cold-cache scan headroom. The rarity client
   timeout is likewise raised, since its scans now run sequentially.

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: SimplicityGuy

api/queries/rarity_queries.py

@@ -11,7 +11,6 @@
   (Release)-[:DERIVED_FROM]->(Master)
 """
 
-import asyncio
 import bisect
 from datetime import UTC, datetime
 from typing import Any
@@ -218,18 +217,22 @@ async def fetch_all_rarity_signals(driver: Any, pool: Any = None) -> list[dict[s
     RETURN release_id, year, latest_sibling_year
     """
 
-    # 5. Graph degree per release
+    # 5. Graph degree per release.
+    # Use COUNT {} rather than size([(r)-[]-() | 1]): the list comprehension
+    # materialises a list element per relationship for every Release, which over
+    # the full graph exhausts the Neo4j transaction memory pool. COUNT {} counts
+    # without building the intermediate list.
     degree_query = """
     MATCH (r:Release)
-    WITH r, size([(r)-[]-() | 1]) AS degree
+    WITH r, COUNT { (r)--() } AS degree
     RETURN r.id AS release_id, degree
     """
 
     # Quality signals for hidden gem scoring
     # 6. Max artist degree per release
     artist_degree_query = """
     MATCH (r:Release)-[:BY]->(a:Artist)
-    WITH r.id AS release_id, max(size([(a)-[]-() | 1])) AS artist_max_degree
+    WITH r.id AS release_id, max(COUNT { (a)--() }) AS artist_max_degree
     RETURN release_id, artist_max_degree
     """
 
@@ -249,25 +252,20 @@ async def fetch_all_rarity_signals(driver: Any, pool: Any = None) -> list[dict[s
 
     logger.info("🔍 Fetching rarity signals from Neo4j...")
 
-    (
-        pressing_rows,
-        label_rows,
-        format_rows,
-        temporal_rows,
-        degree_rows,
-        artist_degree_rows,
-        label_size_rows,
-        genre_count_rows,
-    ) = await asyncio.gather(
-        run_query(driver, pressing_query, database="neo4j"),
-        run_query(driver, label_query, database="neo4j"),
-        run_query(driver, format_query, database="neo4j"),
-        run_query(driver, temporal_query, database="neo4j"),
-        run_query(driver, degree_query, database="neo4j"),
-        run_query(driver, artist_degree_query, database="neo4j"),
-        run_query(driver, label_size_query, database="neo4j"),
-        run_query(driver, genre_count_query, database="neo4j"),
-    )
+    # Run sequentially, not via asyncio.gather: each of these is a full-graph
+    # scan, and running all eight concurrently sums their working sets against
+    # the single dbms.memory.transaction.total.max pool — which tips it into a
+    # TransientError MemoryPoolOutOfMemoryError. Sequential execution caps peak
+    # transaction memory at one query at a time. This runs in a daily background
+    # computation, so the extra wall-clock time is acceptable.
+    pressing_rows = await run_query(driver, pressing_query, database="neo4j")
+    label_rows = await run_query(driver, label_query, database="neo4j")
+    format_rows = await run_query(driver, format_query, database="neo4j")
+    temporal_rows = await run_query(driver, temporal_query, database="neo4j")
+    degree_rows = await run_query(driver, degree_query, database="neo4j")
+    artist_degree_rows = await run_query(driver, artist_degree_query, database="neo4j")
+    label_size_rows = await run_query(driver, label_size_query, database="neo4j")
+    genre_count_rows = await run_query(driver, genre_count_query, database="neo4j")
 
     logger.info(
         "📊 Rarity signal data fetched",

api/routers/insights_compute.py

@@ -11,6 +11,7 @@
 from fastapi import APIRouter, Query, Request
 from fastapi.responses import JSONResponse
 import httpx
+from neo4j.exceptions import TransientError
 from psycopg.rows import dict_row
 import structlog
 
@@ -139,7 +140,17 @@ async def rarity_scores(request: Request) -> JSONResponse:  # noqa: ARG001
     """Return computed rarity scores for all releases from Neo4j."""
     if not _neo4j:
         return JSONResponse(content={"error": "Service not ready"}, status_code=503)
-    results = await fetch_all_rarity_signals(_neo4j, _pool)
+    try:
+        results = await fetch_all_rarity_signals(_neo4j, _pool)
+    except TransientError:
+        # e.g. MemoryPoolOutOfMemoryError under DB pressure — transient, not a
+        # server bug. Return 503 so the caller logs a meaningful, retryable
+        # failure instead of a bare 500.
+        logger.warning("⚠️ Rarity computation hit a transient Neo4j error", exc_info=True)
+        return JSONResponse(
+            content={"error": "Neo4j temporarily unavailable (transient error)"},
+            status_code=503,
+        )
     return JSONResponse(content={"items": results})
 
 

insights/computations.py

@@ -63,6 +63,18 @@ async def _fetch_from_api(
     return items
 
 
+def _describe_error(exc: Exception) -> str:
+    """Return a non-empty, diagnosable description of *exc*.
+
+    ``str(exc)`` is empty for several exceptions raised here — notably
+    ``httpx.ReadTimeout`` — which previously produced blank ``error=""`` log
+    lines that were impossible to triage. Always prefix the exception type so
+    the failure mode is identifiable even when the message is empty.
+    """
+    detail = str(exc)
+    return f"{type(exc).__name__}: {detail}" if detail else type(exc).__name__
+
+
 async def compute_and_store_artist_centrality(client: httpx.AsyncClient, pool: Any, limit: int = 100) -> int:
     """Compute artist centrality and store results."""
     started_at = datetime.now(UTC)
@@ -260,9 +272,11 @@ async def compute_and_store_data_completeness(client: httpx.AsyncClient, pool: A
     """Compute data completeness and store results."""
     started_at = datetime.now(UTC)
     try:
-        # Data completeness does full sequential scans (~400s for releases table alone);
-        # use a longer timeout than the client default to avoid ReadTimeout errors.
-        results = await _fetch_from_api(client, "/api/internal/insights/data-completeness", timeout=600.0)
+        # Data completeness does full sequential scans whose duration varies widely
+        # (observed ~230s on a warm day, but exceeding 600s on others). The endpoint
+        # caches results in Redis for 6h, so only the cold path is slow; give the
+        # cold path generous headroom to avoid spurious ReadTimeouts.
+        results = await _fetch_from_api(client, "/api/internal/insights/data-completeness", timeout=1200.0)
         if not results:
             await _log_computation(pool, "data_completeness", "completed", started_at, 0)
             return 0
@@ -294,11 +308,11 @@ async def compute_and_store_data_completeness(client: httpx.AsyncClient, pool: A
         await _log_computation(pool, "data_completeness", "completed", started_at, len(results))
         return len(results)
     except Exception as e:
-        logger.error("❌ Data completeness computation failed", error=str(e))
+        logger.error("❌ Data completeness computation failed", error=_describe_error(e))
         try:
-            await _log_computation(pool, "data_completeness", "failed", started_at, error_message=str(e))
+            await _log_computation(pool, "data_completeness", "failed", started_at, error_message=_describe_error(e))
         except Exception as log_err:
-            logger.warning("⚠️ Failed to log computation error", error=str(log_err))
+            logger.warning("⚠️ Failed to log computation error", error=_describe_error(log_err))
         raise
 
 
@@ -326,7 +340,9 @@ async def compute_and_store_rarity(client: httpx.AsyncClient, pool: Any) -> int:
     """Compute release rarity scores and store results."""
     started_at = datetime.now(UTC)
     try:
-        results = await _fetch_from_api(client, "/api/internal/insights/rarity-scores", timeout=600.0)
+        # The API runs eight full-graph Neo4j scans sequentially (to cap transaction
+        # memory), so allow generous headroom over the client default.
+        results = await _fetch_from_api(client, "/api/internal/insights/rarity-scores", timeout=1200.0)
         if not results:
             logger.info("📊 No rarity score results to store")
             await _log_computation(pool, "release_rarity", "completed", started_at, 0)
@@ -367,11 +383,11 @@ async def compute_and_store_rarity(client: httpx.AsyncClient, pool: Any) -> int:
         await _log_computation(pool, "release_rarity", "completed", started_at, len(results))
         return len(results)
     except Exception as e:
-        logger.error("❌ Release rarity computation failed", error=str(e))
+        logger.error("❌ Release rarity computation failed", error=_describe_error(e))
         try:
-            await _log_computation(pool, "release_rarity", "failed", started_at, error_message=str(e))
+            await _log_computation(pool, "release_rarity", "failed", started_at, error_message=_describe_error(e))
         except Exception as log_err:
-            logger.warning("⚠️ Failed to log computation error", error=str(log_err))
+            logger.warning("⚠️ Failed to log computation error", error=_describe_error(log_err))
         raise
 
 

tests/api/test_insights_compute.py

@@ -245,6 +245,19 @@ def test_503_when_not_ready(self, test_client: TestClient) -> None:
         finally:
             ic_router._neo4j = original
 
+    def test_503_on_transient_neo4j_error(self, test_client: TestClient) -> None:
+        """A transient Neo4j error (e.g. out-of-memory) returns 503, not 500."""
+        from neo4j.exceptions import TransientError
+
+        err = TransientError("MemoryPoolOutOfMemoryError")
+        with patch(
+            "api.routers.insights_compute.fetch_all_rarity_signals",
+            new=AsyncMock(side_effect=err),
+        ):
+            response = test_client.get("/api/internal/insights/rarity-scores")
+        assert response.status_code == 503
+        assert "error" in response.json()
+
 
 class TestAnniversariesValidation:
     """Tests for milestones query parameter validation (lines 82-83)."""

tests/insights/test_computations.py

@@ -119,6 +119,24 @@ async def test_returns_empty_list_when_no_items_key(self) -> None:
         assert result == []
 
 
+class TestDescribeError:
+    """Tests for _describe_error — guards against blank error log lines."""
+
+    def test_includes_type_when_message_empty(self) -> None:
+        """An exception with an empty str() (e.g. httpx.ReadTimeout) still names its type."""
+        import httpx
+
+        from insights.computations import _describe_error
+
+        # httpx.ReadTimeout("") has an empty str(), which previously logged as error="".
+        assert _describe_error(httpx.ReadTimeout("")) == "ReadTimeout"
+
+    def test_includes_type_and_message_when_present(self) -> None:
+        from insights.computations import _describe_error
+
+        assert _describe_error(ValueError("boom")) == "ValueError: boom"
+
+
 class TestComputeAndStoreArtistCentrality:
     @pytest.mark.asyncio
     async def test_fetches_from_api_and_stores_results(self) -> None:

tests/insights/test_rarity_computation.py

@@ -61,7 +61,7 @@ async def test_fetches_and_stores(self) -> None:
         mock_fetch.assert_called_once_with(
             mock_client,
             "/api/internal/insights/rarity-scores",
-            timeout=600.0,
+            timeout=1200.0,
         )
 
     @pytest.mark.asyncio