fix: comprehensive bug fixes across all services (#233)

* fix: comprehensive bug fixes across all services

Critical fixes:
- brainzgraphinator: fix wrong keys for mbid, mb_type, begin_date/end_date (always null/wrong)
- brainztableinator: fix begin_date/end_date reading from wrong location
- insights: fix infinite loop in cache invalidate_all (string/int cursor comparison)
- insights: include metric param in top-artists cache key to prevent wrong cached responses
- api: fix KeyError on missing 'total' column in queue history endpoint
- extractor: gate state marker completion behind success check (prevent silent data loss)
- extractor: use DataType::discogs() instead of DataType::all() to avoid publishing to non-existent exchange
- docker-compose.prod: add missing production secret overrides for 3 MusicBrainz services
- scripts: fix SQL/command injection in reset-password.sh
- dashboard: fix PostgreSQL connection leak in get_database_info (called every 2s)
- schema-init: set autocommit for DDL to prevent cascading InFailedSqlTransaction

Important fixes:
- explore: fix _doExplore -> _loadExplore (credits profile broken)
- explore: fix window.graphController -> window.exploreApp (NLQ entity links broken)
- explore: fix SSE stream eventType not persisting across chunks, flush buffer on done
- api/sync: release Redis lock after task creation, add password-change token check
- common: fix active_connections counter growing monotonically in PostgreSQL pool
- common: lazy-init asyncio.Lock in CircuitBreaker (RuntimeError on Python 3.12+)
- common: fix CircuitBreaker HALF_OPEN race condition
- graphinator: fix DETACH DELETE count always returning 0, use summary.counters
- graphinator: use MERGE instead of MATCH for Master node in process_release
- graphinator: add None guard before graph.close()
- brainz services: reset active_connection/active_channel to None after recovery failure
- dashboard: handle JSONDecodeError in admin proxy handlers (return 400 not 500)
- dashboard: consume Neo4j result before issuing second query on same session
- docker-compose.prod: remove unnecessary Neo4j credentials from explore service
- mcp-server: add error handling for API failures in all tool handlers
- insights: wrap _log_computation in error handlers to prevent exception masking
- tableinator: parse started_at as datetime with UTC timezone for purge_stale_rows
- utilities: fix connection leak in debug_message.py
- utilities: replace hardcoded Neo4j password in system_monitor.py
- tests: fix test names, mock cursor types, and assertions to match bug fixes

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* test: fix JS test and add coverage for new bug fix code

- Fix credits.test.js: update mock from _doExplore to _loadExplore and
  fix assertion order to match the corrected credits.js code
- Add CircuitBreaker tests: lazy asyncio.Lock init, HALF_OPEN race fix
  for both sync and async paths
- Add insights computation tests: verify original exception propagates
  when _log_computation also fails
- Add MCP server tests: HTTP error and network error return error dicts
- Add admin proxy tests: malformed JSON body returns 400 for all proxy
  endpoints
- Add sync password_changed test: token issued before password change
  returns 401
- Add tableinator purge test: naive datetime (no timezone) gets UTC

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* test: cover remaining uncovered lines from bug fixes

- insights/computations.py: add tests for label_longevity, anniversaries,
  data_completeness, and rarity exception masking prevention
- common/db_resilience.py: add tests for lazy lock init in
  _on_success_async and _on_failure_async
- explore/app.js: add test for NLQ entity click calling _switchPane
  and _loadExplore on exploreApp
- explore/api-client.js: add tests for SSE buffer flush on stream
  completion and eventType persisting across chunk boundaries

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

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

Author: SimplicityGuy

api/queries/metrics_queries.py

@@ -99,7 +99,7 @@ async def get_queue_history(pool: Any, range_value: str) -> dict[str, Any]:
             "ts": row["ts"],
             "ready": _round_or_int(row["ready"], is_raw=is_raw),
             "unacked": _round_or_int(row["unacked"], is_raw=is_raw),
-            "total": _round_or_int(row["total"], is_raw=is_raw),
+            "total": _round_or_int(row.get("ready", 0), is_raw=is_raw) + _round_or_int(row.get("unacked", 0), is_raw=is_raw),
             "publish_rate": _round_or_int(row["publish_rate"], is_raw=is_raw),
             "deliver_rate": _round_or_int(row["deliver_rate"], is_raw=is_raw),
         }

api/routers/sync.py

@@ -64,6 +64,17 @@ async def _get_current_user(
                     detail="Token has been revoked",
                     headers={"WWW-Authenticate": "Bearer"},
                 )
+        # Check if password was changed after token was issued
+        if user_id and _redis:
+            pw_changed = await _redis.get(f"password_changed:{user_id}")
+            if pw_changed:
+                iat = payload.get("iat")
+                if iat and int(iat) < int(pw_changed):
+                    raise HTTPException(
+                        status_code=status.HTTP_401_UNAUTHORIZED,
+                        detail="Token has been revoked",
+                        headers={"WWW-Authenticate": "Bearer"},
+                    )
         return payload
     except ValueError as exc:
         raise HTTPException(
@@ -142,6 +153,7 @@ async def trigger_sync(
     # Set per-user cooldown to prevent rapid re-triggers
     if _redis:
         await _redis.setex(f"sync:cooldown:{user_id}", 600, "1")
+        await _redis.delete(f"sync:lock:{user_id}")
 
     logger.info("🔄 Sync triggered", user_id=user_id, sync_id=sync_id)
     return JSONResponse(content={"sync_id": sync_id, "status": "started"}, status_code=status.HTTP_202_ACCEPTED)

brainzgraphinator/brainzgraphinator.py

@@ -292,11 +292,13 @@ async def enrich_artist(tx: Any, record: dict[str, Any]) -> bool:
         "    a.mb_updated_at = $mb_updated_at "
         "RETURN a.id AS matched_id",
         discogs_id=discogs_id,
-        mbid=record.get("mbid"),
-        mb_type=record.get("type"),
+        mbid=record.get("mbid", record.get("id")),
+        mb_type=record.get("mb_type", record.get("type")),
         mb_gender=record.get("gender"),
-        mb_begin_date=record.get("begin_date"),
-        mb_end_date=record.get("end_date"),
+        mb_begin_date=record.get(
+            "begin_date", (record.get("life_span") or {}).get("begin")
+        ),
+        mb_end_date=record.get("end_date", (record.get("life_span") or {}).get("end")),
         mb_area=record.get("area"),
         mb_begin_area=record.get("begin_area"),
         mb_end_area=record.get("end_area"),
@@ -338,11 +340,13 @@ async def enrich_label(tx: Any, record: dict[str, Any]) -> bool:
         "    l.mb_updated_at = $mb_updated_at "
         "RETURN l.id AS matched_id",
         discogs_id=discogs_id,
-        mbid=record.get("mbid"),
-        mb_type=record.get("type"),
+        mbid=record.get("mbid", record.get("id")),
+        mb_type=record.get("mb_type", record.get("type")),
         mb_label_code=record.get("label_code"),
-        mb_begin_date=record.get("begin_date"),
-        mb_end_date=record.get("end_date"),
+        mb_begin_date=record.get(
+            "begin_date", (record.get("life_span") or {}).get("begin")
+        ),
+        mb_end_date=record.get("end_date", (record.get("life_span") or {}).get("end")),
         mb_area=record.get("area"),
         mb_updated_at=datetime.now(UTC).isoformat(),
     )
@@ -373,7 +377,7 @@ async def enrich_release(tx: Any, record: dict[str, Any]) -> bool:
         "    r.mb_updated_at = $mb_updated_at "
         "RETURN r.id AS matched_id",
         discogs_id=discogs_id,
-        mbid=record.get("mbid"),
+        mbid=record.get("mbid", record.get("id")),
         mb_barcode=record.get("barcode"),
         mb_status=record.get("status"),
         mb_updated_at=datetime.now(UTC).isoformat(),
@@ -448,8 +452,8 @@ async def enrich_release_group(tx: Any, record: dict[str, Any]) -> bool:
         "    m.mb_updated_at = $mb_updated_at "
         "RETURN m.id AS matched_id",
         discogs_id=discogs_id,
-        mbid=record.get("mbid"),
-        mb_type=record.get("type"),
+        mbid=record.get("mbid", record.get("id")),
+        mb_type=record.get("mb_type", record.get("type")),
         mb_secondary_types=record.get("secondary_types", []),
         mb_first_release_date=record.get("first_release_date"),
         mb_disambiguation=record.get("disambiguation"),
@@ -763,6 +767,8 @@ async def _recover_consumers() -> None:
             await temp_connection.close()
         except Exception:  # nosec: B110
             pass
+        active_connection = None
+        active_channel = None
 
 
 async def main() -> None:

brainztableinator/brainztableinator.py

@@ -427,6 +427,8 @@ async def _recover_consumers() -> None:
             await temp_connection.close()
         except Exception:  # nosec: B110
             pass
+        active_connection = None
+        active_channel = None
 
 
 async def _insert_relationship(
@@ -498,9 +500,11 @@ async def process_artist(conn: Any, record: dict[str, Any]) -> None:
                 record.get("sort_name", ""),
                 record.get("mb_type", ""),
                 record.get("gender", ""),
-                record.get("begin_date"),
-                record.get("end_date"),
-                record.get("ended", False),
+                record.get("begin_date", (record.get("life_span") or {}).get("begin")),
+                record.get("end_date", (record.get("life_span") or {}).get("end")),
+                record.get(
+                    "ended", (record.get("life_span") or {}).get("ended", False)
+                ),
                 record.get("area", ""),
                 record.get("begin_area", ""),
                 record.get("end_area", ""),
@@ -543,9 +547,11 @@ async def process_label(conn: Any, record: dict[str, Any]) -> None:
                 record.get("name", ""),
                 record.get("mb_type", ""),
                 record.get("label_code"),
-                record.get("begin_date"),
-                record.get("end_date"),
-                record.get("ended", False),
+                record.get("begin_date", (record.get("life_span") or {}).get("begin")),
+                record.get("end_date", (record.get("life_span") or {}).get("end")),
+                record.get(
+                    "ended", (record.get("life_span") or {}).get("ended", False)
+                ),
                 record.get("area", ""),
                 record.get("disambiguation", ""),
                 record.get("discogs_label_id"),

common/db_resilience.py

@@ -45,17 +45,38 @@ def __init__(self, config: CircuitBreakerConfig):
         self.last_failure_time: datetime | None = None
         self.state = CircuitState.CLOSED
         self._lock = Lock()
-        self._async_lock = asyncio.Lock()
+        self._async_lock: asyncio.Lock | None = None
 
     def call(self, func: Callable[..., T], *args: Any, **kwargs: Any) -> T:
         """Execute function with circuit breaker protection."""
+        execute_under_lock = False
         with self._lock:
             if self.state == CircuitState.OPEN:
                 if self._should_attempt_reset():
                     self.state = CircuitState.HALF_OPEN
                     logger.info(f"🔄 {self.config.name}: Circuit breaker entering HALF_OPEN state")
+                    execute_under_lock = True
                 else:
                     raise Exception(f"{self.config.name}: Circuit breaker is OPEN")
+            elif self.state == CircuitState.HALF_OPEN:
+                execute_under_lock = True
+
+            if execute_under_lock:
+                # Execute trial call under lock to prevent HALF_OPEN race
+                try:
+                    result = func(*args, **kwargs)
+                    self.failure_count = 0
+                    if self.state != CircuitState.CLOSED:
+                        logger.info(f"✅ {self.config.name}: Circuit breaker reset to CLOSED")
+                        self.state = CircuitState.CLOSED
+                    return result
+                except self.config.expected_exception:
+                    self.failure_count += 1
+                    self.last_failure_time = datetime.now(UTC)
+                    if self.failure_count >= self.config.failure_threshold and self.state != CircuitState.OPEN:
+                        logger.error(f"❌ {self.config.name}: Circuit breaker OPEN after {self.failure_count} failures")
+                        self.state = CircuitState.OPEN
+                    raise
 
         try:
             result = func(*args, **kwargs)
@@ -67,13 +88,39 @@ def call(self, func: Callable[..., T], *args: Any, **kwargs: Any) -> T:
 
     async def call_async(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
         """Execute async function with circuit breaker protection."""
+        if self._async_lock is None:
+            self._async_lock = asyncio.Lock()
+        execute_under_lock = False
         async with self._async_lock:
             if self.state == CircuitState.OPEN:
                 if self._should_attempt_reset():
                     self.state = CircuitState.HALF_OPEN
                     logger.info(f"🔄 {self.config.name}: Circuit breaker entering HALF_OPEN state")
+                    execute_under_lock = True
                 else:
                     raise Exception(f"{self.config.name}: Circuit breaker is OPEN")
+            elif self.state == CircuitState.HALF_OPEN:
+                execute_under_lock = True
+
+            if execute_under_lock:
+                # Execute trial call under lock to prevent HALF_OPEN race
+                try:
+                    if asyncio.iscoroutinefunction(func):
+                        result = await func(*args, **kwargs)
+                    else:
+                        result = func(*args, **kwargs)
+                    self.failure_count = 0
+                    if self.state != CircuitState.CLOSED:
+                        logger.info(f"✅ {self.config.name}: Circuit breaker reset to CLOSED")
+                        self.state = CircuitState.CLOSED
+                    return result
+                except self.config.expected_exception:
+                    self.failure_count += 1
+                    self.last_failure_time = datetime.now(UTC)
+                    if self.failure_count >= self.config.failure_threshold and self.state != CircuitState.OPEN:
+                        logger.error(f"❌ {self.config.name}: Circuit breaker OPEN after {self.failure_count} failures")
+                        self.state = CircuitState.OPEN
+                    raise
 
         try:
             if asyncio.iscoroutinefunction(func):
@@ -110,6 +157,8 @@ def _on_failure(self) -> None:
 
     async def _on_success_async(self) -> None:
         """Handle successful async call."""
+        if self._async_lock is None:
+            self._async_lock = asyncio.Lock()
         async with self._async_lock:
             self.failure_count = 0
             if self.state != CircuitState.CLOSED:
@@ -118,6 +167,8 @@ async def _on_success_async(self) -> None:
 
     async def _on_failure_async(self) -> None:
         """Handle failed async call."""
+        if self._async_lock is None:
+            self._async_lock = asyncio.Lock()
         async with self._async_lock:
             self.failure_count += 1
             self.last_failure_time = datetime.now(UTC)

common/postgres_resilient.py

@@ -212,6 +212,8 @@ def connection(self) -> Generator[psycopg.Connection[Any]]:
             if conn and not conn.closed and not self._closed:
                 try:
                     self.connections.put_nowait(conn)
+                    with self._lock:
+                        self.active_connections = max(0, self.active_connections - 1)
                 except Full:
                     # Pool is full, close connection
                     with contextlib.suppress(Exception):
@@ -338,7 +340,6 @@ async def initialize(self) -> None:
                 conn = await self._create_connection()
                 if conn:
                     await self.connections.put(conn)
-                    self.active_connections += 1
             except asyncio.QueueFull:
                 break
             except Exception as e:

dashboard/admin_proxy.py

@@ -13,6 +13,7 @@
 
 from fastapi import APIRouter, Query, Request, Response
 import httpx
+from starlette.responses import JSONResponse
 import structlog
 
 
@@ -91,7 +92,10 @@ async def proxy_login(request: Request) -> Response:
     """Proxy login requests to the API service."""
     url = _build_url("/api/admin/auth/login")
     headers = _auth_headers(request)
-    sanitised_body = await _validated_json_body(request)
+    try:
+        sanitised_body = await _validated_json_body(request)
+    except json.JSONDecodeError:
+        return JSONResponse(content={"detail": "Malformed JSON in request body"}, status_code=400)
     try:
         async with httpx.AsyncClient(timeout=30.0) as client:
             if sanitised_body:
@@ -154,7 +158,10 @@ async def proxy_trigger(request: Request) -> Response:
     """Proxy extraction trigger requests to the API service."""
     url = _build_url("/api/admin/extractions/trigger")
     headers = _auth_headers(request)
-    sanitised_body = await _validated_json_body(request)
+    try:
+        sanitised_body = await _validated_json_body(request)
+    except json.JSONDecodeError:
+        return JSONResponse(content={"detail": "Malformed JSON in request body"}, status_code=400)
     try:
         async with httpx.AsyncClient(timeout=30.0) as client:
             if sanitised_body:
@@ -173,7 +180,10 @@ async def proxy_trigger_musicbrainz(request: Request) -> Response:
     """Proxy MusicBrainz extraction trigger requests to the API service."""
     url = _build_url("/api/admin/extractions/trigger")
     headers = _auth_headers(request)
-    sanitised_body = await _validated_json_body(request)
+    try:
+        sanitised_body = await _validated_json_body(request)
+    except json.JSONDecodeError:
+        return JSONResponse(content={"detail": "Malformed JSON in request body"}, status_code=400)
     body_dict: dict = json.loads(sanitised_body) if sanitised_body else {}
     body_dict["source"] = "musicbrainz"
     payload = json.dumps(body_dict, separators=(",", ":")).encode()

dashboard/dashboard.py

@@ -348,6 +348,7 @@ async def get_database_info(self) -> list[DatabaseInfo]:
         databases = []
 
         # Check PostgreSQL
+        conn = None
         try:
             if self.postgres_conn is None:
                 raise RuntimeError("PostgreSQL resilient connection not initialized")
@@ -392,13 +393,18 @@ async def get_database_info(self) -> list[DatabaseInfo]:
                     error=str(e),
                 )
             )
+        finally:
+            if conn is not None:
+                with contextlib.suppress(Exception):
+                    await conn.rollback()
 
         # Check Neo4j
         try:
             if self.neo4j_driver:
                 async with self.neo4j_driver.session() as session:
                     result = await session.run("CALL dbms.components() YIELD name, versions")
-                    await result.single()  # Consume result
+                    await result.single()
+                    await result.consume()
 
                     # Get database size
                     result = await session.run("CALL apoc.meta.stats() YIELD nodeCount, relCount")