Parallelize the snapshot rebuild across cores (opt-in, validated)

backend/app/services/charts_stats.py

@@ -641,6 +641,50 @@ def new_accumulator() -> dict[str, Any]:
     return {b: _new_acc_one() for b in _BLOB_BRACKETS}
 
 
+# Merging two accumulators (from parallel run-chunk walks) adds per key. Most
+# charts cells are `key -> [numbers]` (element-wise add); enc_hist and death_room
+# are `key -> count` (scalar add).
+_CHART_LIST_FIELDS = (
+    "hp_floor",
+    "enc",
+    "traj",
+    "elites",
+    "smiths",
+    "events",
+    "entity_week",
+    "week_totals",
+    "copies",
+    "ench",
+)
+_CHART_COUNTER_FIELDS = ("enc_hist", "death_room")
+
+
+def _merge_list_dict(dst: dict, src: dict) -> None:
+    """dst[key] += src[key] element-wise (both are lists of the same length)."""
+    for k, v in src.items():
+        cur = dst.get(k)
+        if cur is None:
+            dst[k] = list(v)
+        else:
+            for i, x in enumerate(v):
+                cur[i] += x
+
+
+def merge(dst: dict, src: dict) -> None:
+    """Fold accumulator `src` into `dst` (both from new_accumulator())."""
+    for bracket, s in src.items():
+        d = dst.get(bracket)
+        if d is None:
+            dst[bracket] = s
+            continue
+        for field in _CHART_LIST_FIELDS:
+            _merge_list_dict(d[field], s[field])
+        for field in _CHART_COUNTER_FIELDS:
+            df = d[field]
+            for k, v in s[field].items():
+                df[k] = df.get(k, 0) + v
+
+
 def _bump2(d: dict, key: Any, win: bool) -> None:
     cell = d.setdefault(key, [0, 0])
     cell[0] += 1

backend/app/services/community_stats.py

@@ -112,6 +112,67 @@ def new_accumulator() -> dict[str, Any]:
     return {b: _new_acc_one() for b in _BLOB_BRACKETS}
 
 
+# Field groups for merging two accumulators (from parallel run-chunk walks).
+# `_INT_FIELDS` add; `_LIST_DICT_FIELDS` add element-wise per key; `events` is a
+# nested event -> {option -> count}; the counter dicts add per key; the three
+# records keep the best (min run_time / max run_time / max deck_size).
+_COMMUNITY_INT_FIELDS = ("total_runs", "total_wins", "reward_screens", "reward_skips")
+_COMMUNITY_LIST_DICT_FIELDS = ("by_ascension", "by_character", "map_danger", "rest")
+_COMMUNITY_COUNTER_FIELDS = (
+    "deaths_encounter",
+    "deaths_event",
+    "ancient",
+    "removed",
+    "stolen",
+)
+
+
+def merge(dst: dict, src: dict) -> None:
+    """Fold accumulator `src` into `dst` (both from new_accumulator()). Called in
+    run-chunk order, so records tie-break to the earlier chunk (matching the
+    serial walk's first-seen-wins on an exact tie)."""
+    for bracket, s in src.items():
+        d = dst.get(bracket)
+        if d is None:
+            dst[bracket] = s
+            continue
+        for f in _COMMUNITY_INT_FIELDS:
+            d[f] += s[f]
+        for f in _COMMUNITY_LIST_DICT_FIELDS:
+            df = d[f]
+            for k, v in s[f].items():
+                cur = df.get(k)
+                if cur is None:
+                    df[k] = list(v)
+                else:
+                    for i, x in enumerate(v):
+                        cur[i] += x
+        for f in _COMMUNITY_COUNTER_FIELDS:
+            df = d[f]
+            for k, v in s[f].items():
+                df[k] = df.get(k, 0) + v
+        # events: event_id -> {option_id -> count}
+        de = d["events"]
+        for eid, opts in s["events"].items():
+            cur = de.setdefault(eid, {})
+            for opt, n in opts.items():
+                cur[opt] = cur.get(opt, 0) + n
+        # Records: replace only when src is STRICTLY better, so the earlier chunk
+        # (already in dst) wins ties, matching the serial walk.
+        if s["fastest_win"] is not None and (
+            d["fastest_win"] is None or s["fastest_win"][0] < d["fastest_win"][0]
+        ):
+            d["fastest_win"] = s["fastest_win"]
+        if s["longest_run"] is not None and (
+            d["longest_run"] is None or s["longest_run"][0] > d["longest_run"][0]
+        ):
+            d["longest_run"] = s["longest_run"]
+        if s["biggest_deck"] is not None and (
+            d["biggest_deck"] is None or s["biggest_deck"][0] > d["biggest_deck"][0]
+        ):
+            d["biggest_deck"] = s["biggest_deck"]
+
+
 def _bump(d: dict, key: Any, n: int = 1) -> None:
     d[key] = d.get(key, 0) + n
 

backend/app/services/encounter_stats.py

@@ -57,6 +57,25 @@ def new_accumulator() -> dict[str, Any]:
     return {b: _new_acc_one() for b in _BLOB_BRACKETS}
 
 
+def merge(dst: dict, src: dict) -> None:
+    """Fold accumulator `src` into `dst` (both from new_accumulator()). The only
+    field is `cells`: key -> [total, fatal, total_damage, total_turns], a plain
+    element-wise add — so parallel run-chunk walks combine losslessly."""
+    for bracket, s in src.items():
+        d = dst.get(bracket)
+        if d is None:
+            dst[bracket] = s
+            continue
+        dcells = d["cells"]
+        for k, v in s["cells"].items():
+            cur = dcells.get(k)
+            if cur is None:
+                dcells[k] = list(v)
+            else:
+                for i, x in enumerate(v):
+                    cur[i] += x
+
+
 def accumulate(
     acc: dict[str, Any],
     blob: dict,