Make debug snapshot reads retry-safe

app.py

@@ -100,6 +100,7 @@ from database.db import (
     insert_bet,
     next_bet_id,
     read_table,
+    read_table_retryable,
     read_cached_odds,
     read_cached_probable_starters,
     read_cached_probable_starters_meta,
@@ -3832,17 +3833,75 @@ def main() -> None:
         _debug_baseline_bundle = load_shared_baseline_bundle_from_snapshots(
             max_age_seconds=max(STATCAST_TTL_SECONDS, 60 * 60)
         )
-        _debug_hitter_df = read_table(conn, "shared_hitter_baseline_event_rows").drop(
+        _debug_read_status = {
+            "hitter_event_rows": {
+                "table_name": "shared_hitter_baseline_event_rows",
+                "read_source": "baseline_bundle",
+                "read_attempts": 1,
+                "retry_used": False,
+                "read_error": "",
+                "snapshot_built_at": None,
+                "source_status": str(_debug_baseline_bundle.get("snapshot_source_status") or ""),
+            },
+            "pitcher_event_rows": {
+                "table_name": "shared_pitcher_baseline_event_rows",
+                "read_source": "baseline_bundle",
+                "read_attempts": 1,
+                "retry_used": False,
+                "read_error": "",
+                "snapshot_built_at": None,
+                "source_status": str(_debug_baseline_bundle.get("snapshot_source_status") or ""),
+            },
+        }
+        _baseline_status_df = _debug_baseline_bundle.get("snapshot_status", pd.DataFrame())
+        if isinstance(_baseline_status_df, pd.DataFrame) and not _baseline_status_df.empty:
+            for _table_key, _status_key in [
+                ("shared_hitter_baseline_event_rows", "hitter_event_rows"),
+                ("shared_pitcher_baseline_event_rows", "pitcher_event_rows"),
+            ]:
+                _row = _baseline_status_df[_baseline_status_df["table_name"] == _table_key]
+                if not _row.empty:
+                    _debug_read_status[_status_key]["snapshot_built_at"] = _row.iloc[0].get("snapshot_built_at")
+                    _debug_read_status[_status_key]["source_status"] = _row.iloc[0].get("source_status")
+
+        _debug_hitter_df = _debug_baseline_bundle.get("blended_batter_df", pd.DataFrame()).copy()
+        _debug_pitcher_df = _debug_baseline_bundle.get("blended_pitcher_df", pd.DataFrame()).copy()
+        if not isinstance(_debug_hitter_df, pd.DataFrame):
+            _debug_hitter_df = pd.DataFrame()
+        if not isinstance(_debug_pitcher_df, pd.DataFrame):
+            _debug_pitcher_df = pd.DataFrame()
+        _debug_hitter_df = _debug_hitter_df.drop(
             columns=["snapshot_built_at", "snapshot_version", "source_status"],
             errors="ignore",
         )
-        _debug_pitcher_df = read_table(conn, "shared_pitcher_baseline_event_rows").drop(
+        _debug_pitcher_df = _debug_pitcher_df.drop(
             columns=["snapshot_built_at", "snapshot_version", "source_status"],
             errors="ignore",
         )
+
+        if _debug_hitter_df.empty:
+            _debug_hitter_df, _debug_read_status["hitter_event_rows"] = read_table_retryable(
+                conn,
+                "shared_hitter_baseline_event_rows",
+            )
+            _debug_hitter_df = _debug_hitter_df.drop(
+                columns=["snapshot_built_at", "snapshot_version", "source_status"],
+                errors="ignore",
+            )
+        if _debug_pitcher_df.empty:
+            _debug_pitcher_df, _debug_read_status["pitcher_event_rows"] = read_table_retryable(
+                conn,
+                "shared_pitcher_baseline_event_rows",
+            )
+            _debug_pitcher_df = _debug_pitcher_df.drop(
+                columns=["snapshot_built_at", "snapshot_version", "source_status"],
+                errors="ignore",
+            )
         if _debug_hitter_df.empty:
+            _debug_read_status["hitter_event_rows"]["read_source"] = "load_statcast_recent_fallback"
             _debug_hitter_df = load_statcast_recent()
         if _debug_pitcher_df.empty:
+            _debug_read_status["pitcher_event_rows"]["read_source"] = "debug_hitter_fallback"
             _debug_pitcher_df = _debug_hitter_df
         loader["update"]("Rendering debug tables...", 0.75)
         render_debug(
@@ -3857,6 +3916,7 @@ def main() -> None:
             grade_outcomes_fn=grade_final_game_outcomes_from_scores,
             grade_props_fn=grade_batter_prop_outcomes_from_audit,
             fill_realized_fn=fill_batter_prop_realized_outcomes,
+            debug_event_row_status=_debug_read_status,
         )
         loader["clear"]()
 

data/shared_baseline.py

@@ -8,7 +8,7 @@ from typing import Any
 import pandas as pd
 from sqlalchemy import text
 
-from database.db import get_connection, replace_table_contents, upsert_dataframe
+from database.db import get_connection, replace_table_contents, safe_read_sql, upsert_dataframe
 from features.pitch_features import add_pitch_features
 from models.rolling_form_model import (
     build_batter_rolling_form_row,
@@ -1015,8 +1015,8 @@ def _read_snapshot_table(
         query = text(
             f"SELECT * FROM {table_name} WHERE player_name IN ({', '.join(clauses)}) ORDER BY player_name"
         )
-        return pd.read_sql(query, conn, params=params)
-    return pd.read_sql(text(f"SELECT * FROM {table_name} ORDER BY player_name"), conn)
+        return safe_read_sql(query, conn, params=params)
+    return safe_read_sql(text(f"SELECT * FROM {table_name} ORDER BY player_name"), conn)
 
 
 def _replace_snapshot_scope(

database/db.py

@@ -23,10 +23,12 @@ so they can be added later without structural changes.
 from __future__ import annotations
 
 import json
+import time
 from typing import Any, Iterable, Mapping
 
 import pandas as pd
 from sqlalchemy import text
+from sqlalchemy.exc import DBAPIError, OperationalError
 
 from database import remote_db
 from utils.helpers import utc_now_iso
@@ -37,6 +39,8 @@ from utils.helpers import utc_now_iso
 # ---------------------------------------------------------------------------
 
 _INSERT_CHUNK_SIZE = 500
+_READ_RETRY_ATTEMPTS = 4
+_READ_RETRY_BASE_DELAY_SECONDS = 0.20
 
 
 # ---------------------------------------------------------------------------
@@ -76,6 +80,94 @@ def _bulk_insert(conn, table: str, df: pd.DataFrame) -> None:
         conn.execute(sql, records[i : i + _INSERT_CHUNK_SIZE])
 
 
+def _is_retryable_read_error(exc: Exception) -> bool:
+    text_value = str(exc or "").lower()
+    retry_markers = (
+        "serializationfailure",
+        "restart transaction",
+        "transactionretrywithprotorefresherror",
+        "readwithinuncertaintyintervalerror",
+        "retry txn",
+        "retry_serializable",
+    )
+    return any(marker in text_value for marker in retry_markers)
+
+
+def safe_read_sql(
+    sql,
+    conn,
+    params: Mapping[str, Any] | None = None,
+    *,
+    max_attempts: int = _READ_RETRY_ATTEMPTS,
+    base_delay_seconds: float = _READ_RETRY_BASE_DELAY_SECONDS,
+) -> pd.DataFrame:
+    last_error: Exception | None = None
+    for attempt in range(1, max(1, int(max_attempts)) + 1):
+        try:
+            return pd.read_sql(sql, conn, params=params)
+        except (OperationalError, DBAPIError) as exc:
+            last_error = exc
+            if attempt >= max_attempts or not _is_retryable_read_error(exc):
+                raise
+            time.sleep(base_delay_seconds * attempt)
+    if last_error is not None:
+        raise last_error
+    return pd.DataFrame()
+
+
+def read_table_retryable(
+    conn,
+    table_name: str,
+    *,
+    where_sql: str | None = None,
+    params: Mapping[str, Any] | None = None,
+    max_attempts: int = _READ_RETRY_ATTEMPTS,
+) -> tuple[pd.DataFrame, dict[str, Any]]:
+    sql_text = f"SELECT * FROM {table_name}"
+    if where_sql:
+        sql_text += f" WHERE {where_sql}"
+
+    attempts = 0
+    retry_used = False
+    last_error = ""
+    for attempt in range(1, max(1, int(max_attempts)) + 1):
+        attempts = attempt
+        try:
+            df = safe_read_sql(
+                text(sql_text),
+                conn,
+                params=params,
+                max_attempts=1,
+            )
+            return df, {
+                "table_name": table_name,
+                "read_source": "db_retryable",
+                "read_attempts": attempts,
+                "retry_used": retry_used,
+                "read_error": "",
+            }
+        except (OperationalError, DBAPIError) as exc:
+            last_error = str(exc)
+            if attempt >= max_attempts or not _is_retryable_read_error(exc):
+                return pd.DataFrame(), {
+                    "table_name": table_name,
+                    "read_source": "db_retryable_failed",
+                    "read_attempts": attempts,
+                    "retry_used": retry_used,
+                    "read_error": last_error,
+                }
+            retry_used = True
+            time.sleep(_READ_RETRY_BASE_DELAY_SECONDS * attempt)
+
+    return pd.DataFrame(), {
+        "table_name": table_name,
+        "read_source": "db_retryable_failed",
+        "read_attempts": attempts,
+        "retry_used": retry_used,
+        "read_error": last_error,
+    }
+
+
 # ---------------------------------------------------------------------------
 # Schema initialization
 # ---------------------------------------------------------------------------
@@ -582,7 +674,7 @@ def replace_table_contents(
 
 
 def read_table(conn, table_name: str) -> pd.DataFrame:
-    return pd.read_sql(text(f"SELECT * FROM {table_name}"), conn)
+    return safe_read_sql(text(f"SELECT * FROM {table_name}"), conn)
 
 
 def _safe_json_dump(value: Any) -> str:

visualization/debug_page.py

@@ -316,6 +316,7 @@ def render_debug(
     grade_outcomes_fn: Callable | None = None,
     grade_props_fn: Callable | None = None,
     fill_realized_fn: Callable | None = None,
+    debug_event_row_status: dict[str, dict[str, Any]] | None = None,
 ) -> None:
     """
     Full Debug Dashboard page.
@@ -1038,6 +1039,28 @@ def render_debug(
                 else:
                     st.info("Open the Props page in this session to capture HR health diagnostics.")
 
+            with st.expander("Debug Event Row Read Status", expanded=False):
+                read_status = debug_event_row_status or {}
+                if read_status:
+                    status_rows = pd.DataFrame(
+                        [
+                            {
+                                "section": key,
+                                "table_name": value.get("table_name"),
+                                "read_source": value.get("read_source"),
+                                "read_attempts": value.get("read_attempts"),
+                                "retry_used": value.get("retry_used"),
+                                "snapshot_built_at": value.get("snapshot_built_at"),
+                                "source_status": value.get("source_status"),
+                                "read_error": value.get("read_error"),
+                            }
+                            for key, value in read_status.items()
+                        ]
+                    )
+                    st.dataframe(status_rows, use_container_width=True, hide_index=True)
+                else:
+                    st.info("No debug event-row read status captured in this session.")
+
             with st.expander("Cached Source Freshness", expanded=False):
                 freshness_rows: list[dict[str, Any]] = []