diff --git a/eval/tenant_panel/money.py b/eval/tenant_panel/money.py index 980709dd..44020fe2 100644 --- a/eval/tenant_panel/money.py +++ b/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: