mirror ledger to private dataset

app/ledger.py

@@ -34,6 +34,84 @@ ROOT = _root()
 LOCK = threading.Lock()
 
 
+# ─────────────────────── 원장 미러 ───────────────────────
+# 🔴 HF Space의 /data 는 영속 스토리지를 붙이지 않으면 재시작에 사라진다.
+#    실측했다 — 재시작 후 부팅 표식이 없어졌다(boots 1 -> 1).
+#    유료 스토리지 대신 비공개 데이터셋 저장소에 원장을 미러링한다.
+#    시즌 4개월치 기록이 컨테이너 재시작 한 번에 날아가면 대회가 끝난다.
+MIRROR = os.environ.get("FINCHAL_MIRROR")        # 예: FINAL-Bench/finchal-ledger
+_MIR = {"ok": False, "dirty": False, "at": 0.0, "n": 0, "err": None}
+
+
+def _api():
+    from huggingface_hub import HfApi
+    tok = os.environ.get("HF_TOKEN") or os.environ.get("FINCHAL_HF_TOKEN")
+    return HfApi(token=tok) if tok else None
+
+
+def restore():
+    """부팅 시 미러에서 원장을 되살린다. 이게 실패하면 미러에 쓰지 않는다.
+
+    🔑 복원 실패 후 flush 하면 빈 원장이 미러를 덮어 시즌이 통째로 사라진다.
+       한 번의 잘못된 덮어쓰기가 4개월을 지운다.
+    """
+    if not MIRROR:
+        return "미러 없음"
+    try:
+        from huggingface_hub import snapshot_download
+        a = _api()
+        if a is None:
+            _MIR["err"] = "HF_TOKEN 없음"
+            return _MIR["err"]
+        try:
+            a.repo_info(MIRROR, repo_type="dataset")
+        except Exception:
+            a.create_repo(MIRROR, repo_type="dataset", private=True, exist_ok=True)
+        d = snapshot_download(MIRROR, repo_type="dataset",
+                              token=a.token, local_dir=None)
+        import shutil, glob
+        os.makedirs(os.path.join(ROOT, "positions"), exist_ok=True)
+        n = 0
+        for src in glob.glob(os.path.join(d, "**", "*"), recursive=True):
+            if not os.path.isfile(src) or os.sep + ".cache" in src:
+                continue
+            rel = os.path.relpath(src, d)
+            if rel.startswith(".git"):
+                continue
+            dst = os.path.join(ROOT, rel)
+            os.makedirs(os.path.dirname(dst), exist_ok=True)
+            shutil.copyfile(src, dst)
+            n += 1
+        _MIR["ok"] = True
+        _MIR["n"] = n
+        return "복원 %d개" % n
+    except Exception as e:
+        _MIR["err"] = "%s: %s" % (type(e).__name__, e)
+        return _MIR["err"]
+
+
+def flush(force=False):
+    """원장을 미러에 올린다. 제출마다 올리면 커밋이 폭주하므로 모아서 보낸다."""
+    if not (MIRROR and _MIR["ok"]) or (not _MIR["dirty"] and not force):
+        return False
+    try:
+        a = _api()
+        a.upload_folder(repo_id=MIRROR, repo_type="dataset", folder_path=ROOT,
+                        commit_message="ledger sync",
+                        ignore_patterns=["*.tmp", ".cache/*"])
+        _MIR["dirty"] = False
+        _MIR["at"] = time.time()
+        return True
+    except Exception as e:
+        _MIR["err"] = "%s: %s" % (type(e).__name__, e)
+        return False
+
+
+def mirror_status():
+    return {"repo": MIRROR, "restored": _MIR["ok"], "files": _MIR["n"],
+            "pending": _MIR["dirty"], "error": _MIR["err"]}
+
+
 def boot_check():
     """원장이 재시작을 견디는지 **실측**한다.
 
@@ -107,7 +185,8 @@ def register(user_id, strategy, ptype, assets):
             "visits": prev.get("visits", []),
         }
         _write(_p("users.json"), u)
-        return u[user_id]
+    _MIR["dirty"] = True
+    return u[user_id]
 
 
 def user_by_key(key):
@@ -143,6 +222,7 @@ def submit(user_id, asset, position, source="api"):
     with LOCK:
         with open(_p("positions", "%s.jsonl" % asset), "a", encoding="utf-8") as f:
             f.write(json.dumps(rec, ensure_ascii=False) + "\n")
+    _MIR["dirty"] = True
     touch(user_id)
     return rec
 

app/main.py

@@ -336,14 +336,31 @@ def _anchor_loop():
         time.sleep(6 * 3600)
 
 
+# 🔑 원장 복원은 **다른 무엇보다 먼저**다. 복원 전에 부팅 표식을 쓰면
+#    그 표식이 미러를 덮어 원장을 지운다.
+_RESTORE = L.restore()
 _DUR = L.boot_check()
 
 
+def _flush_loop():
+    """원장을 주기적으로 미러에 올린다. 제출마다 커밋하면 저장소가 터진다."""
+    import traceback
+    while True:
+        time.sleep(90)
+        try:
+            L.flush()
+        except Exception:
+            traceback.print_exc()
+
+
 @app.on_event("startup")
 def _startup():
+    import threading
+    print("[finchal] 원장 %s · 미러 %s" % (L.ROOT, _RESTORE))
+    if L.MIRROR:
+        threading.Thread(target=_flush_loop, daemon=True).start()
     if os.environ.get("FINCHAL_NO_ANCHOR"):
         return
-    import threading
     threading.Thread(target=_anchor_loop, daemon=True).start()
 
 
@@ -351,6 +368,7 @@ def _startup():
 def health():
     return {"ok": True, "season": S1["name"], "assets": list(ASSETS),
             "ledger": L.ROOT, "durability": _DUR[0], "boots": _DUR[1],
-            "warn": None if _DUR[0] == "확인됨" else
-                    "원장이 재시작을 견디는지 아직 확인되지 않았습니다. "
-                    "영속 스토리지 없이 시즌을 열면 기록이 사라질 수 있습니다."}
+            "mirror": L.mirror_status(), "restore": _RESTORE,
+            "warn": None if (_DUR[0] == "확인됨" or L.mirror_status()["restored"])
+                    else "원장이 재시작을 견딘다는 증거가 없습니다. "
+                         "미러(FINCHAL_MIRROR)나 영속 스토리지를 붙이십시오."}