From 06dc0ac7c89bedb1586efdab7212ef09dcf503ef Mon Sep 17 00:00:00 2001 From: Chris Hondl Date: Fri, 10 Jul 2026 08:49:18 -0700 Subject: [PATCH 1/5] Add snapshot serialization for pipeline state --- backend/src/google/snapshot.py | 109 +++++++++++++++++++++++++++++++++ 1 file changed, 109 insertions(+) create mode 100644 backend/src/google/snapshot.py diff --git a/backend/src/google/snapshot.py b/backend/src/google/snapshot.py new file mode 100644 index 00000000..83163c80 --- /dev/null +++ b/backend/src/google/snapshot.py @@ -0,0 +1,109 @@ +from enum import Enum +import json +from typing import Any, Dict, List, Mapping, Optional, Tuple, Type +import zlib + +import attr +from sqlalchemy import inspect +from sqlalchemy.sql.sqltypes import Enum as SQLEnum + +from src.data.utils import objs_type +from src.db.models import ETag, Event, Match, Team, TeamEvent, TeamYear, Year +from src.db.models.etag import ETagORM +from src.db.models.event import EventORM +from src.db.models.main import Model, ModelORM +from src.db.models.match import MatchORM +from src.db.models.team import TeamORM +from src.db.models.team_event import TeamEventORM +from src.db.models.team_year import TeamYearORM +from src.db.models.year import YearORM +from src.google.storage import _bucket + +SNAPSHOT_SCHEMA = 1 +SNAPSHOT_PREFIX = "state" + + +def snapshot_key(year: int) -> str: + return f"{SNAPSHOT_PREFIX}/snapshot.{year}" + + +def _enum_fields(orm_type: Type[ModelORM]) -> Dict[str, Type[Enum]]: + fields: Dict[str, Type[Enum]] = {} + for column in inspect(orm_type).columns: + if isinstance(column.type, SQLEnum) and column.type.enum_class is not None: + fields[column.name] = column.type.enum_class + return fields + + +def _load( + model_cls: Type[Model], orm_type: Type[ModelORM], data: Dict[str, Any] +) -> Any: + obj = model_cls.from_dict(data) + for name, enum_cls in _enum_fields(orm_type).items(): + value = getattr(obj, name) + if value is not None and not isinstance(value, enum_cls): + setattr(obj, name, enum_cls(value)) + return obj + + +def _dump_values(objs: Mapping[str, Model]) -> List[Dict[str, Any]]: + return [attr.asdict(o) for o in sorted(objs.values(), key=lambda o: o.pk())] + + +def _load_values( + model_cls: Type[Model], orm_type: Type[ModelORM], rows: List[Dict[str, Any]] +) -> Dict[str, Any]: + loaded = [_load(model_cls, orm_type, row) for row in rows] + return {o.pk(): o for o in loaded} + + +def serialize(objs: objs_type, teams: List[Team]) -> bytes: + year_obj = objs[0] + payload = { + "schema": SNAPSHOT_SCHEMA, + "year": year_obj.year, + "teams": [attr.asdict(t) for t in sorted(teams, key=lambda t: t.team)], + "objs": { + "year": attr.asdict(year_obj), + "team_years": _dump_values(objs[1]), + "events": _dump_values(objs[2]), + "team_events": _dump_values(objs[3]), + "matches": _dump_values(objs[4]), + "etags": _dump_values(objs[5]), + }, + } + return zlib.compress(json.dumps(payload).encode("utf-8")) + + +def deserialize(raw: bytes) -> Tuple[objs_type, List[Team]]: + payload = json.loads(zlib.decompress(raw).decode("utf-8")) + data = payload["objs"] + objs: objs_type = ( + _load(Year, YearORM, data["year"]), + _load_values(TeamYear, TeamYearORM, data["team_years"]), + _load_values(Event, EventORM, data["events"]), + _load_values(TeamEvent, TeamEventORM, data["team_events"]), + _load_values(Match, MatchORM, data["matches"]), + _load_values(ETag, ETagORM, data["etags"]), + ) + teams = [_load(Team, TeamORM, row) for row in payload["teams"]] + return objs, teams + + +def write_snapshot(year: int, objs: objs_type, teams: List[Team]) -> None: + bucket = _bucket() + key = snapshot_key(year) + tmp_key = key + ".tmp" + bucket.blob(tmp_key).upload_from_string( + serialize(objs, teams), "application/octet-stream" + ) + bucket.copy_blob(bucket.blob(tmp_key), bucket, key) + bucket.blob(tmp_key).delete() + + +def read_snapshot(year: int) -> Optional[Tuple[objs_type, List[Team]]]: + try: + raw = _bucket().blob(snapshot_key(year)).download_as_bytes() + except Exception: + return None + return deserialize(raw) From 01d525bf33c81d36f2b463c630b569ffd3eae501 Mon Sep 17 00:00:00 2001 From: Chris Hondl Date: Fri, 10 Jul 2026 08:49:18 -0700 Subject: [PATCH 2/5] Publish blobs from in-memory state, tolerate DB outage --- backend/src/google/storage.py | 83 +++++++++++++++++++++++------------ 1 file changed, 55 insertions(+), 28 deletions(-) diff --git a/backend/src/google/storage.py b/backend/src/google/storage.py index 60ddc2a1..54707ced 100644 --- a/backend/src/google/storage.py +++ b/backend/src/google/storage.py @@ -3,7 +3,7 @@ from datetime import datetime, timedelta, timezone import gzip import json -from typing import Any, Dict, List, Optional +from typing import Any, Callable, Dict, List, Optional, TypeVar import zlib from google.cloud import storage @@ -11,6 +11,7 @@ from src.constants import CURR_YEAR, HIST_EPOCH, PROD from src.data.utils import nan_safe_eq, objs_type from src.db.functions import get_noteworthy_matches, get_upcoming_matches +from src.db.models.team import Team from src.db.read.event import get_events as get_events_db from src.db.read.team import get_teams as get_teams_db from src.db.read.team_year import get_team_years as get_team_years_db @@ -63,6 +64,17 @@ def upload_file_to_gcs(data: Any, object_name: str) -> None: _upload_bytes(_bucket(), object_name, compress(data), None) +T = TypeVar("T") + + +def _best_effort(label: str, fn: Callable[[], T], default: T) -> T: + try: + return fn() + except Exception as e: + print(f"skipped {label}: {e}") + return default + + def read_manifest() -> Optional[Manifest]: try: raw = _bucket().blob(MANIFEST_OBJECT).download_as_bytes() @@ -104,6 +116,7 @@ def _publish(plan: UploadPlan) -> None: def write_objs( objs: objs_type, orig_objs: Optional[objs_type] = None, + teams: Optional[List[Team]] = None, ) -> None: year = CURR_YEAR year_obj = objs[0] @@ -114,7 +127,8 @@ def add(object_name: str, data: Any) -> None: rendered[object_name] = compress(data) # teams/all - teams = get_teams_db() + if teams is None: + teams = _best_effort("teams", get_teams_db, []) add("teams/all", _read_all_teams(teams)) # team_years/{CURR_YEAR} @@ -129,7 +143,9 @@ def add(object_name: str, data: Any) -> None: ) # events/all - add("events/all", _read_all_events(get_events_db())) + all_events = _best_effort("events/all", get_events_db, None) + if all_events is not None: + add("events/all", _read_all_events(all_events)) # events/{CURR_YEAR} events = list(objs[2].values()) @@ -181,38 +197,49 @@ def add(object_name: str, data: Any) -> None: add("team_to_events", team_to_events) # team/{team.team} - all_team_years = get_team_years_db() - team_years_by_team: Dict[int, List[Any]] = defaultdict(list) - for ty in all_team_years: - team_years_by_team[ty.team].append(ty) - teams_by_num = {t.team: t for t in teams} - for num in {ty.team for ty in team_years}: - team_obj = teams_by_num.get(num) - if team_obj is None: - continue - add(f"team/{num}", _read_team(team_obj, team_years_by_team.get(num, []))) + all_team_years = _best_effort("team pages", get_team_years_db, None) + if all_team_years is not None: + team_years_by_team: Dict[int, List[Any]] = defaultdict(list) + for ty in all_team_years: + team_years_by_team[ty.team].append(ty) + teams_by_num = {t.team: t for t in teams} + for num in {ty.team for ty in team_years}: + team_obj = teams_by_num.get(num) + if team_obj is None: + continue + add(f"team/{num}", _read_team(team_obj, team_years_by_team.get(num, []))) # noteworthy_matches/{year} - noteworthy_matches = get_noteworthy_matches( - year=year, country=None, state=None, district=None, elim=None, week=None + noteworthy_matches = _best_effort( + "noteworthy_matches", + lambda: get_noteworthy_matches( + year=year, country=None, state=None, district=None, elim=None, week=None + ), + None, ) - add(f"noteworthy_matches/{year}", _read_noteworthy_matches(noteworthy_matches)) + if noteworthy_matches is not None: + add(f"noteworthy_matches/{year}", _read_noteworthy_matches(noteworthy_matches)) # upcoming_matches?limit=20&metric={predicted_time | max_epa | sum_epa | diff_epa} for metric in ["predicted_time", "max_epa", "sum_epa", "diff_epa"]: - upcoming_matches = get_upcoming_matches( - country=None, - state=None, - district=None, - elim=None, - minutes=-1, - limit=20, - metric=metric, - ) - add( - f"upcoming_matches.limit=20.metric={metric}", - _read_upcoming_matches(upcoming_matches), + upcoming_matches = _best_effort( + f"upcoming_matches.{metric}", + lambda metric=metric: get_upcoming_matches( + country=None, + state=None, + district=None, + elim=None, + minutes=-1, + limit=20, + metric=metric, + ), + None, ) + if upcoming_matches is not None: + add( + f"upcoming_matches.limit=20.metric={metric}", + _read_upcoming_matches(upcoming_matches), + ) cycle = datetime.now(timezone.utc).isoformat() plan = plan_uploads(rendered, prev, cycle, hist_epoch=HIST_EPOCH) From 902dc3e8324984e3bbcb04d39af165b1a3057d4f Mon Sep 17 00:00:00 2001 From: Chris Hondl Date: Fri, 10 Jul 2026 08:49:18 -0700 Subject: [PATCH 3/5] Load state from snapshot, move DB writes off the hot path --- backend/src/data/main.py | 59 +++++++++++++++++++++++++++------------ backend/src/data/utils.py | 13 +++++---- 2 files changed, 49 insertions(+), 23 deletions(-) diff --git a/backend/src/data/main.py b/backend/src/data/main.py index 6b0e64c3..5e6a4a40 100644 --- a/backend/src/data/main.py +++ b/backend/src/data/main.py @@ -1,5 +1,6 @@ from collections import defaultdict from copy import deepcopy +import traceback from typing import Dict, List, Optional, Tuple from src.constants import CURR_YEAR, DISABLE_GCS @@ -38,6 +39,7 @@ update_team_years as update_team_years_db, update_teams as update_teams_db, ) +from src.google.snapshot import read_snapshot, write_snapshot from src.google.storage import write_objs as write_objs_storage @@ -51,13 +53,16 @@ def process_year( all_team_years: Optional[Dict[int, Dict[int, TeamYear]]], ) -> List[Team]: timer = Timer() - orig_objs = deepcopy(objs) + curr_year_gcs = year_num == CURR_YEAR and not DISABLE_GCS + orig_objs = None if curr_year_gcs else deepcopy(objs) if all_team_years is None: all_team_years = defaultdict(dict) - for year in range(max(2002, year_num - 4), year_num): - team_years = get_team_years_db(year=year) - for ty in team_years: - all_team_years[ty.year][ty.team] = ty + try: + for year in range(max(2002, year_num - 4), year_num): + for ty in get_team_years_db(year=year): + all_team_years[ty.year][ty.team] = ty + except Exception: + traceback.print_exc() new_teams, objs = process_year_tba(year_num, teams, objs, tba_partial, cache) teams += new_teams @@ -74,13 +79,23 @@ def process_year( objs = process_year_epa(objs, all_team_years) timer.print(str(year_num) + " EPA") - write_objs_db(year_num, objs, orig_objs if partial else None, not partial) - timer.print(str(year_num) + " Write DB") + if curr_year_gcs: + write_snapshot(year_num, objs, teams) + timer.print(str(year_num) + " Write Snapshot") - if year_num == CURR_YEAR and not DISABLE_GCS: - write_objs_storage(objs, orig_objs if partial else None) + write_objs_storage(objs, None, teams) timer.print(str(year_num) + " Write Storage") + try: + db_orig = read_objs_db(year_num) if partial else None + write_objs_db(year_num, objs, db_orig, not partial) + except Exception: + traceback.print_exc() + timer.print(str(year_num) + " Write DB") + else: + write_objs_db(year_num, objs, orig_objs if partial else None, not partial) + timer.print(str(year_num) + " Write DB") + return teams @@ -138,15 +153,23 @@ def update_curr_year(partial: bool, tba_partial: bool): year = CURR_YEAR timer = Timer() - teams = get_teams_db() - timer.print("Load Teams") - - if partial: - objs: objs_type = read_objs_db(year) - timer.print("Read Objs") - else: - objs = create_objs(year) - timer.print("Create Objs") + objs: Optional[objs_type] = None + teams: Optional[List[Team]] = None + if partial and not DISABLE_GCS: + loaded = read_snapshot(year) + if loaded is not None: + objs, teams = loaded + timer.print("Read Snapshot") + + if objs is None or teams is None: + teams = get_teams_db() + timer.print("Load Teams") + if partial: + objs = read_objs_db(year) + timer.print("Read Objs") + else: + objs = create_objs(year) + timer.print("Create Objs") teams = process_year( year, partial, tba_partial, year < CURR_YEAR, teams, objs, None diff --git a/backend/src/data/utils.py b/backend/src/data/utils.py index 560670b3..ab6a3c8e 100644 --- a/backend/src/data/utils.py +++ b/backend/src/data/utils.py @@ -53,13 +53,16 @@ def read_objs(year: int) -> objs_type: if year_obj is None: raise Exception("Year not found") + def by_pk(objs: list) -> dict: + return {o.pk(): o for o in sorted(objs, key=lambda o: o.pk())} + return ( year_obj, - {t.pk(): t for t in get_team_years_db(year=year)}, - {e.pk(): e for e in get_events_db(year=year)}, - {te.pk(): te for te in get_team_events_db(year=year)}, - {m.pk(): m for m in get_matches_db(year=year)}, - {e.pk(): e for e in get_etags_db(year=year)}, + by_pk(get_team_years_db(year=year)), + by_pk(get_events_db(year=year)), + by_pk(get_team_events_db(year=year)), + by_pk(get_matches_db(year=year)), + by_pk(get_etags_db(year=year)), ) From 2be6721517e9e1298df581abcc9c4a4f45205b05 Mon Sep 17 00:00:00 2001 From: Chris Hondl Date: Fri, 10 Jul 2026 11:19:56 -0700 Subject: [PATCH 4/5] Pass cycle-start state to the publisher; render team pages from fresh rows F1: the event-blob gate needs the pre-cycle objs to tell which events actually changed, so deepcopy them and pass as orig instead of None. team-page lag: build each team's current-year row from the in-memory objs (fresh this cycle) instead of the persisted read (previous cycle), matching the team_years list blob. --- backend/src/data/main.py | 4 ++-- backend/src/google/storage.py | 3 +++ 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/backend/src/data/main.py b/backend/src/data/main.py index 5e6a4a40..7d959689 100644 --- a/backend/src/data/main.py +++ b/backend/src/data/main.py @@ -54,7 +54,7 @@ def process_year( ) -> List[Team]: timer = Timer() curr_year_gcs = year_num == CURR_YEAR and not DISABLE_GCS - orig_objs = None if curr_year_gcs else deepcopy(objs) + orig_objs = deepcopy(objs) if all_team_years is None: all_team_years = defaultdict(dict) try: @@ -83,7 +83,7 @@ def process_year( write_snapshot(year_num, objs, teams) timer.print(str(year_num) + " Write Snapshot") - write_objs_storage(objs, None, teams) + write_objs_storage(objs, orig_objs if partial else None, teams) timer.print(str(year_num) + " Write Storage") try: diff --git a/backend/src/google/storage.py b/backend/src/google/storage.py index 54707ced..53822478 100644 --- a/backend/src/google/storage.py +++ b/backend/src/google/storage.py @@ -201,6 +201,9 @@ def add(object_name: str, data: Any) -> None: if all_team_years is not None: team_years_by_team: Dict[int, List[Any]] = defaultdict(list) for ty in all_team_years: + if ty.year != year: + team_years_by_team[ty.team].append(ty) + for ty in team_years: team_years_by_team[ty.team].append(ty) teams_by_num = {t.team: t for t in teams} for num in {ty.team for ty in team_years}: From fe6fcc4e1f64fd8f401436c1a563e01ad43eae3e Mon Sep 17 00:00:00 2001 From: Chris Hondl Date: Fri, 10 Jul 2026 11:19:56 -0700 Subject: [PATCH 5/5] Validate snapshot schema on read; use a unique tmp key per writer A5: deserialize now checks the embedded schema version and read_snapshot falls back to the DB path (logging why) on mismatch or a corrupt/short payload, instead of silently None-ing missing fields. Bump SNAPSHOT_SCHEMA on any ORM column change. A6: the staging tmp blob key carries pid+uuid so two concurrent publishers cannot race on a shared key. --- backend/src/google/snapshot.py | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/backend/src/google/snapshot.py b/backend/src/google/snapshot.py index 83163c80..d6f76d4c 100644 --- a/backend/src/google/snapshot.py +++ b/backend/src/google/snapshot.py @@ -1,6 +1,8 @@ from enum import Enum import json +import os from typing import Any, Dict, List, Mapping, Optional, Tuple, Type +from uuid import uuid4 import zlib import attr @@ -77,6 +79,11 @@ def serialize(objs: objs_type, teams: List[Team]) -> bytes: def deserialize(raw: bytes) -> Tuple[objs_type, List[Team]]: payload = json.loads(zlib.decompress(raw).decode("utf-8")) + schema = payload.get("schema") + if schema != SNAPSHOT_SCHEMA: + raise ValueError( + f"snapshot schema {schema} != expected {SNAPSHOT_SCHEMA}" + ) data = payload["objs"] objs: objs_type = ( _load(Year, YearORM, data["year"]), @@ -93,7 +100,7 @@ def deserialize(raw: bytes) -> Tuple[objs_type, List[Team]]: def write_snapshot(year: int, objs: objs_type, teams: List[Team]) -> None: bucket = _bucket() key = snapshot_key(year) - tmp_key = key + ".tmp" + tmp_key = f"{key}.{os.getpid()}.{uuid4().hex}.tmp" bucket.blob(tmp_key).upload_from_string( serialize(objs, teams), "application/octet-stream" ) @@ -106,4 +113,8 @@ def read_snapshot(year: int) -> Optional[Tuple[objs_type, List[Team]]]: raw = _bucket().blob(snapshot_key(year)).download_as_bytes() except Exception: return None - return deserialize(raw) + try: + return deserialize(raw) + except Exception as e: + print(f"snapshot unreadable, falling back to DB path: {e}") + return None