Re-freeze phase 2: claim each cell with an exclusive lock so parallel workers cannot buy the same call twice with the second payment invisible to the ledger

eval/tenant_panel/money.py

@@ -183,6 +183,23 @@ _TRANSIENT = ("503", "502", "504", "overloaded", "high demand", "timeout", "Time
               "RemoteProtocolError", "APIConnectionError")
 
 
+def _claim(tag: str) -> bool:
+    """Атомарно занять клетку. False — её уже покупает другой процесс.
+
+    ⚠ Заведено исполнением: панель покупалась НЕСКОЛЬКИМИ процессами, и проверка «файла ещё нет»
+    у `buy.purchase` не атомарна — два воркера могли одновременно пройти её и купить одну клетку
+    ДВАЖДЫ. На диске остаётся одна запись, поэтому вторая оплата в леджер не попадает вовсе:
+    деньги потрачены, а касса их не видит. Это тот же класс, что «расход фазы мимо глоба», только
+    хуже — незаметный по построению.
+    """
+    try:
+        fd = os.open(str(OUT / f"{tag}.lock"), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
+    except FileExistsError:
+        return False
+    os.close(fd)
+    return True
+
+
 def purchase(led: Guarded, tag: str, model: str, client, kw: dict, **extra) -> dict:
     """Один платный вызов. Кэш → фриз-гейт → проекционный гард → покупка (+ отступ на 5xx)."""
     f = OUT / f"{tag}.json"
@@ -191,17 +208,29 @@ def purchase(led: Guarded, tag: str, model: str, client, kw: dict, **extra) -> d
     if not any(tag.startswith(p.split("*")[0]) for p in TAGS[led.phase]):
         raise SystemExit(f"⛔ тег {tag!r} не покрыт глобом кассы {led.phase} — впиши в TAGS")
     _refuse_unfrozen()
+    if not _claim(tag):                                        # клетку уже покупает другой процесс
+        f = OUT / f"{tag}.json"
+        for _ in range(240):
+            if f.exists():
+                return json.loads(f.read_text(encoding="utf-8"))
+            time.sleep(5)
+        print(f"    ⚠ замок {tag} протух — покупаю сам")
+        (OUT / f"{tag}.lock").unlink(missing_ok=True)
+        _claim(tag)
     led.next_model = model                                     # гард считает ожидание по ЭТОЙ модели
     rec = {}
-    for k in range(4):
-        rec = BUY.purchase(led, tag, model, client, kw, **extra)
-        err = rec.get("error", "") if rec.get("finish") == "error" else ""
-        if not err or not any(t in err for t in _TRANSIENT):
-            return rec
-        wait = 15 * (k + 1)
-        print(f"    транзиентный отказ ({err[:60]}) — отступ {wait} с")
-        time.sleep(wait)
-    return rec
+    try:
+        for k in range(4):
+            rec = BUY.purchase(led, tag, model, client, kw, **extra)
+            err = rec.get("error", "") if rec.get("finish") == "error" else ""
+            if not err or not any(t in err for t in _TRANSIENT):
+                return rec
+            wait = 15 * (k + 1)
+            print(f"    транзиентный отказ ({err[:60]}) — отступ {wait} с")
+            time.sleep(wait)
+        return rec
+    finally:
+        (OUT / f"{tag}.lock").unlink(missing_ok=True)
 
 
 def report() -> None: