lb-autoheal: keep A's history fresh + cheap history-only re-seed for deltas
Browse filesProxy mode had disabled A's maintenance worker AND its 600MB save, which
is also where the hourly/daily history snapshots were taken -- so A stopped
producing history and could no longer seed the mirror's d1/d7 deltas after
a mirror restart. Two fixes in the auto-heal loop:
- _sm_lb_keep_history_fresh(): re-take the cheap, index-independent hourly
+ daily snapshots each tick so A stays the durable bootstrap source.
- history-only re-seed: when the mirror has rows but no d1 baseline (fresh
restart), push just the history files -- no 1.6M row re-push, no
starvation risk. The full mirror sync now also pushes history FIRST, so a
heal interrupted mid-rows still lands the deltas.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
app.py
CHANGED
|
@@ -12494,6 +12494,28 @@ def _sm_lb_proxy_fetch(request: Request, path: str, inject_counts: bool = False)
|
|
| 12494 |
_SM_LB_MIRROR_STATUS = {"running": False, "pushed_rows": 0, "pushed_hist": 0, "total_rows": 0,
|
| 12495 |
"done": False, "error": ""}
|
| 12496 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 12497 |
def _sm_lb_mirror_sync_worker(target: str, chunk: int):
|
| 12498 |
st = _SM_LB_MIRROR_STATUS
|
| 12499 |
st.update({"running": True, "pushed_rows": 0, "pushed_hist": 0, "total_rows": 0,
|
|
@@ -12501,6 +12523,12 @@ def _sm_lb_mirror_sync_worker(target: str, chunk: int):
|
|
| 12501 |
sess = _requests.Session()
|
| 12502 |
hdr = {"X-Ingest-Key": SM_LEADERBOARD_INGEST_KEY, "Content-Type": "application/json"}
|
| 12503 |
try:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 12504 |
with _SM_LEADERBOARD_LOCK:
|
| 12505 |
rows_src = list((_sm_leaderboard_load_unlocked().get("players") or {}).values())
|
| 12506 |
st["total_rows"] = len(rows_src)
|
|
@@ -12532,22 +12560,6 @@ def _sm_lb_mirror_sync_worker(target: str, chunk: int):
|
|
| 12532 |
flush()
|
| 12533 |
time.sleep(0.3) # gentle: don't hammer the mirror during the one-time sync
|
| 12534 |
flush()
|
| 12535 |
-
# history snapshots -> immediate deltas on the mirror
|
| 12536 |
-
try:
|
| 12537 |
-
import gzip as _gz
|
| 12538 |
-
for f in sorted(SM_LB_HISTORY_DIR.glob("*.json")):
|
| 12539 |
-
data = json.loads(f.read_text(encoding="utf-8"))
|
| 12540 |
-
sess.post(f"{target}/lb-history-import",
|
| 12541 |
-
json={"kind": "daily", "name": f.stem, "data": data}, headers=hdr, timeout=90).raise_for_status()
|
| 12542 |
-
st["pushed_hist"] += 1
|
| 12543 |
-
for f in sorted(SM_LB_HISTORY_HOURLY_DIR.glob("*.json.gz")):
|
| 12544 |
-
data = json.loads(_gz.decompress(f.read_bytes()).decode("utf-8"))
|
| 12545 |
-
sess.post(f"{target}/lb-history-import",
|
| 12546 |
-
json={"kind": "hourly", "name": f.name[:-len('.json.gz')], "data": data},
|
| 12547 |
-
headers=hdr, timeout=90).raise_for_status()
|
| 12548 |
-
st["pushed_hist"] += 1
|
| 12549 |
-
except Exception as he: # noqa: BLE001 -- history is best-effort; accounts already synced
|
| 12550 |
-
print(f"[lb-mirror-sync] history push warning: {type(he).__name__}: {str(he)[:150]}")
|
| 12551 |
st["done"] = True
|
| 12552 |
print(f"[lb-mirror-sync] done: {st['pushed_rows']}/{st['total_rows']} rows, "
|
| 12553 |
f"{st['pushed_hist']} history files -> {target}")
|
|
@@ -12618,13 +12630,43 @@ def _sm_lb_mirror_stored_count():
|
|
| 12618 |
except Exception: # noqa: BLE001
|
| 12619 |
return 0, False
|
| 12620 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 12621 |
def _sm_lb_autoheal_worker():
|
| 12622 |
if not (SM_LEADERBOARD_PROXY_URL and SM_LEADERBOARD_INGEST_KEY):
|
| 12623 |
return
|
| 12624 |
time.sleep(75) # let boot + the pusher's first sweep settle before judging the mirror
|
| 12625 |
last_sync = 0.0
|
|
|
|
| 12626 |
while True:
|
| 12627 |
try:
|
|
|
|
| 12628 |
if not _SM_LB_MIRROR_STATUS.get("running"):
|
| 12629 |
mirror_rows, reachable = _sm_lb_mirror_stored_count()
|
| 12630 |
local_rows = _sm_lb_local_stored_count()
|
|
@@ -12639,6 +12681,19 @@ def _sm_lb_autoheal_worker():
|
|
| 12639 |
_sm_lb_mirror_sync_worker(SM_LEADERBOARD_PROXY_URL, 5000)
|
| 12640 |
last_sync = time.time()
|
| 12641 |
_sm_lb_proxy_healthy(force=True) # refresh the serving gate
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 12642 |
except Exception as exc: # noqa: BLE001
|
| 12643 |
print(f"[lb-autoheal] {type(exc).__name__}: {str(exc)[:150]}")
|
| 12644 |
time.sleep(SM_LB_AUTOHEAL_INTERVAL)
|
|
|
|
| 12494 |
_SM_LB_MIRROR_STATUS = {"running": False, "pushed_rows": 0, "pushed_hist": 0, "total_rows": 0,
|
| 12495 |
"done": False, "error": ""}
|
| 12496 |
|
| 12497 |
+
def _sm_lb_push_history(target: str, sess, hdr, st: dict = None):
|
| 12498 |
+
"""Push A's daily + hourly history snapshots to the mirror (immediate d1/d7 deltas). Small +
|
| 12499 |
+
quick; done BEFORE the heavy row push so a heal interrupted mid-rows still lands the deltas."""
|
| 12500 |
+
n = 0
|
| 12501 |
+
import gzip as _gz
|
| 12502 |
+
for f in sorted(SM_LB_HISTORY_DIR.glob("*.json")):
|
| 12503 |
+
data = json.loads(f.read_text(encoding="utf-8"))
|
| 12504 |
+
sess.post(f"{target}/lb-history-import",
|
| 12505 |
+
json={"kind": "daily", "name": f.stem, "data": data}, headers=hdr, timeout=90).raise_for_status()
|
| 12506 |
+
n += 1
|
| 12507 |
+
if st is not None:
|
| 12508 |
+
st["pushed_hist"] = n
|
| 12509 |
+
for f in sorted(SM_LB_HISTORY_HOURLY_DIR.glob("*.json.gz")):
|
| 12510 |
+
data = json.loads(_gz.decompress(f.read_bytes()).decode("utf-8"))
|
| 12511 |
+
sess.post(f"{target}/lb-history-import",
|
| 12512 |
+
json={"kind": "hourly", "name": f.name[:-len('.json.gz')], "data": data},
|
| 12513 |
+
headers=hdr, timeout=90).raise_for_status()
|
| 12514 |
+
n += 1
|
| 12515 |
+
if st is not None:
|
| 12516 |
+
st["pushed_hist"] = n
|
| 12517 |
+
return n
|
| 12518 |
+
|
| 12519 |
def _sm_lb_mirror_sync_worker(target: str, chunk: int):
|
| 12520 |
st = _SM_LB_MIRROR_STATUS
|
| 12521 |
st.update({"running": True, "pushed_rows": 0, "pushed_hist": 0, "total_rows": 0,
|
|
|
|
| 12523 |
sess = _requests.Session()
|
| 12524 |
hdr = {"X-Ingest-Key": SM_LEADERBOARD_INGEST_KEY, "Content-Type": "application/json"}
|
| 12525 |
try:
|
| 12526 |
+
# history FIRST: it's tiny and gives deltas immediately; if the long row push below is
|
| 12527 |
+
# interrupted (e.g. A gets restarted under the load) the mirror still has its d1/d7.
|
| 12528 |
+
try:
|
| 12529 |
+
_sm_lb_push_history(target, sess, hdr, st)
|
| 12530 |
+
except Exception as he: # noqa: BLE001 -- best-effort; rows still follow
|
| 12531 |
+
print(f"[lb-mirror-sync] history push warning: {type(he).__name__}: {str(he)[:150]}")
|
| 12532 |
with _SM_LEADERBOARD_LOCK:
|
| 12533 |
rows_src = list((_sm_leaderboard_load_unlocked().get("players") or {}).values())
|
| 12534 |
st["total_rows"] = len(rows_src)
|
|
|
|
| 12560 |
flush()
|
| 12561 |
time.sleep(0.3) # gentle: don't hammer the mirror during the one-time sync
|
| 12562 |
flush()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 12563 |
st["done"] = True
|
| 12564 |
print(f"[lb-mirror-sync] done: {st['pushed_rows']}/{st['total_rows']} rows, "
|
| 12565 |
f"{st['pushed_hist']} history files -> {target}")
|
|
|
|
| 12630 |
except Exception: # noqa: BLE001
|
| 12631 |
return 0, False
|
| 12632 |
|
| 12633 |
+
def _sm_lb_mirror_missing_d1():
|
| 12634 |
+
"""True if the mirror is up but has no ~24h-ago snapshot (d1 deltas blank). A fresh mirror
|
| 12635 |
+
(post-restart, before it accumulates its own hourly snapshots) has this until A re-seeds its
|
| 12636 |
+
history -- cheap to fix without re-pushing 1.6M rows."""
|
| 12637 |
+
try:
|
| 12638 |
+
global _SM_LB_PROXY_SESSION
|
| 12639 |
+
if _SM_LB_PROXY_SESSION is None:
|
| 12640 |
+
_SM_LB_PROXY_SESSION = _requests.Session()
|
| 12641 |
+
r = _SM_LB_PROXY_SESSION.get(f"{SM_LEADERBOARD_PROXY_URL}/leaderboard?limit=1&measure=d1", timeout=8)
|
| 12642 |
+
r.raise_for_status()
|
| 12643 |
+
return (r.json() or {}).get("history", {}).get("d1_age") is None
|
| 12644 |
+
except Exception: # noqa: BLE001
|
| 12645 |
+
return False # unreachable -> not our concern here (the thin/serving gates handle it)
|
| 12646 |
+
|
| 12647 |
+
def _sm_lb_keep_history_fresh():
|
| 12648 |
+
"""Proxy mode disables A's heavy maintenance worker AND its 600MB save -- which is also where
|
| 12649 |
+
the hourly/daily history snapshots were taken. So A stops producing history and, days later,
|
| 12650 |
+
can no longer seed the mirror's d1/d7 after a restart. Re-take the (cheap, index-independent)
|
| 12651 |
+
snapshots here so A always holds fresh bootstrap history. Once/hour + once/day of real work;
|
| 12652 |
+
a file-exists check otherwise."""
|
| 12653 |
+
try:
|
| 12654 |
+
_sm_leaderboard_snapshot_hourly()
|
| 12655 |
+
with _SM_LEADERBOARD_LOCK:
|
| 12656 |
+
state = _sm_leaderboard_load_unlocked()
|
| 12657 |
+
_sm_leaderboard_snapshot_daily(state)
|
| 12658 |
+
except Exception as exc: # noqa: BLE001
|
| 12659 |
+
print(f"[lb-autoheal] snapshot keep-fresh: {type(exc).__name__}: {str(exc)[:120]}")
|
| 12660 |
+
|
| 12661 |
def _sm_lb_autoheal_worker():
|
| 12662 |
if not (SM_LEADERBOARD_PROXY_URL and SM_LEADERBOARD_INGEST_KEY):
|
| 12663 |
return
|
| 12664 |
time.sleep(75) # let boot + the pusher's first sweep settle before judging the mirror
|
| 12665 |
last_sync = 0.0
|
| 12666 |
+
last_hist = 0.0
|
| 12667 |
while True:
|
| 12668 |
try:
|
| 12669 |
+
_sm_lb_keep_history_fresh() # A stays the durable source of bootstrap history
|
| 12670 |
if not _SM_LB_MIRROR_STATUS.get("running"):
|
| 12671 |
mirror_rows, reachable = _sm_lb_mirror_stored_count()
|
| 12672 |
local_rows = _sm_lb_local_stored_count()
|
|
|
|
| 12681 |
_sm_lb_mirror_sync_worker(SM_LEADERBOARD_PROXY_URL, 5000)
|
| 12682 |
last_sync = time.time()
|
| 12683 |
_sm_lb_proxy_healthy(force=True) # refresh the serving gate
|
| 12684 |
+
elif reachable and not thin and (time.time() - last_hist) > SM_LB_AUTOHEAL_COOLDOWN \
|
| 12685 |
+
and _sm_lb_mirror_missing_d1():
|
| 12686 |
+
# Rows are fine but the mirror has no d1 baseline (fresh restart): re-seed just
|
| 12687 |
+
# the history -- tiny, no 1.6M row re-push, no starvation risk.
|
| 12688 |
+
print("[lb-autoheal] mirror missing d1 history -> re-seeding history only")
|
| 12689 |
+
try:
|
| 12690 |
+
_sess = _requests.Session()
|
| 12691 |
+
_hdr = {"X-Ingest-Key": SM_LEADERBOARD_INGEST_KEY, "Content-Type": "application/json"}
|
| 12692 |
+
n = _sm_lb_push_history(SM_LEADERBOARD_PROXY_URL, _sess, _hdr)
|
| 12693 |
+
print(f"[lb-autoheal] re-seeded {n} history files")
|
| 12694 |
+
except Exception as he: # noqa: BLE001
|
| 12695 |
+
print(f"[lb-autoheal] history re-seed failed: {type(he).__name__}: {str(he)[:120]}")
|
| 12696 |
+
last_hist = time.time()
|
| 12697 |
except Exception as exc: # noqa: BLE001
|
| 12698 |
print(f"[lb-autoheal] {type(exc).__name__}: {str(exc)[:150]}")
|
| 12699 |
time.sleep(SM_LB_AUTOHEAL_INTERVAL)
|