diff --git a/harness/es.py b/harness/es.py index aaa93ed..1287c08 100644 --- a/harness/es.py +++ b/harness/es.py @@ -336,6 +336,95 @@ def append_event(aggregate: str, aggregate_id: str, event_type: str, data: dict) return (eid, version) +def claim_task_if_ready(task_id: str, claim: dict): + """Атомарный условный клейм задачи: TaskClaimed ТОЛЬКО если она сейчас ready. + + t-261015000631-13aef0 (гонка hour/day 2026-10-15 00:00): оба уровня читали + read model ДО клейма друг друга и каждый по своему снапшоту писал TaskClaimed + → L-задача исполнялась дважды (двойной LLM-burn 9router). Read-check-write + в scheduler таймингом не закрыть: снапшот модели к моменту записи — вчерашний. + + Атомарность гарантирует СТОР, не тайминг: проверка статуса и INSERT — в одной + BEGIN IMMEDIATE транзакции. Статус считаем не из read model (она может + отставать), а прямой свёрткой хвоста событий агрегата из events — истина + на момент транзакции. Побеждает ровно один писатель; проигравший получает + False и молча пропускает задачу (её уже ведёт другой уровень). + + Возвращает (event_id, version) при успехе, False — задача не ready + (клеймнута/закрыта/не существует), None — дубль записи (интegrity). + """ + with _LOCK: + _ensure_signature_schema() + db = _conn() + try: + db.execute("BEGIN IMMEDIATE") + except sqlite3.OperationalError: + pass + row = db.execute( + "SELECT event_type, version FROM events " + "WHERE aggregate='task' AND aggregate_id=? ORDER BY version", + (task_id,), + ).fetchall() + # свёртка статуса по потоку событий задачи (та же логика, что + # _apply_event, но без полного read model — только этот агрегат) + status = None + for r in row: + et = r["event_type"] + if et == "TaskSubmitted": + status = "ready" + elif et == "TaskClaimed": + status = "running" + elif et in ("TaskCompleted",): + status = "done" + elif et == "TaskFailed": + status = "failed" + elif et == "TaskClosed": + if status == "failed": + status = "closed" + elif et in ("TaskPromoted", "TaskRequeued"): + status = "ready" + elif et == "TaskBlocked": + status = "blocked" + if status != "ready": + db.rollback() + db.close() + return False + version = (row[-1]["version"] if row else 0) + 1 + ts = datetime.now(timezone.utc).isoformat() + data_str = json.dumps(claim, ensure_ascii=False) + sig_f = {} + try: + from . import audit as _audit + prev = db.execute( + "SELECT sig FROM events ORDER BY id DESC LIMIT 1").fetchone() + sig_f = _audit.sign_event_row("task", task_id, "TaskClaimed", + data_str, ts, version, + prev["sig"] if prev else None) + if sig_f.get("key_id"): + dom = db.execute( + "SELECT key_id FROM events WHERE key_id IS NOT NULL " + "ORDER BY id DESC LIMIT 1").fetchone() + if dom and dom["key_id"] and dom["key_id"] != sig_f["key_id"]: + sig_f = {} + except Exception: + sig_f = {} + try: + cur = db.execute( + "INSERT INTO events (aggregate,aggregate_id,event_type,data,timestamp,version," + "sig,key_id,prev_sig) VALUES (?,?,?,?,?,?,?,?,?)", + ("task", task_id, "TaskClaimed", data_str, ts, version, + sig_f.get("sig"), sig_f.get("key_id"), sig_f.get("prev_sig")), + ) + db.commit() + eid = cur.lastrowid + except sqlite3.IntegrityError: + db.rollback() + db.close() + return None + db.close() + return (eid, version) + + def append_tick(level: str, seq: int): """CycleTicked идемпотентно: aggregate_id = 'level:seq' → UNIQUE даёт защиту.""" return append_event("cycle", f"{level}:{seq}", "CycleTicked", {"level": level, "seq": seq}) diff --git a/harness/scheduler.py b/harness/scheduler.py index c50442b..3be5e32 100644 --- a/harness/scheduler.py +++ b/harness/scheduler.py @@ -522,16 +522,33 @@ def run_level_tick(level_cfg, seq: int, model: dict): batch = [model["tasks"].get(tid) for tid in batch_ids] batch = [t for t in batch if t] claimed = [] + stale = [] # заняты другим уровнем после снапшота модели (гонка клейма) for t in batch: - es.append_event("task", t["id"], "TaskClaimed", { + # t-261015000631-13aef0: атомарный условный клейм НА УРОВНЕ СТОРА + # (проверка статуса + INSERT в одной BEGIN IMMEDIATE транзакции). + # Гонка 2026-10-15 00:00: hour и day читали модель ДО клейма друг + # друга → L-задача t-261014221908 клеймнута дважды (TaskClaimed + # id=2767823/2767828) и исполнена дважды = двойной burn 9router. + # Снапшот модели здесь — вчерашний; истина только в сторе. + res = es.claim_task_if_ready(t["id"], { "level": level_cfg.name, "session_id": f"s-{level_cfg.name}-{seq}", }) + if not res: + # False — уже не ready (взята/закрыта), None — конфликт записи: + # события нет, клейма нет → пропускаем, задача не наша. + stale.append(t["id"]) + continue claimed.append(t["id"]) # TG-visibility: взяли задачу — сообщаем в чат (анти-спам: не чаще 15м на уровень) try: _notify_busy(level_cfg, t) except Exception: pass # уведомление — best effort, не валить тик + if stale: + _log("claim-race: пропущены задачи, занятые другим уровнем/не ready: " + + ", ".join(stale)) + # в промпт — только задачи, чей клейм реально наш + batch = [t for t in batch if t["id"] in set(claimed)] if not claimed: claimed = None prompt = build_context(level_cfg, model, batch) diff --git a/tests/test_seen_indication.py b/tests/test_seen_indication.py index db50946..8da1642 100644 --- a/tests/test_seen_indication.py +++ b/tests/test_seen_indication.py @@ -134,6 +134,12 @@ class _FakeES: def append_event(self, aggregate, aid, evtype, data={}): written.append((evtype, aid)) return "x" + + # t-261015000631-13aef0: условный клейм в стабе всегда успешен + def claim_task_if_ready(self, task_id, claim): + written.append(("TaskClaimed", task_id)) + return ("x", 1) + class _FakeRunner: @staticmethod def run_llm_tick(level_cfg, prompt, session_id=None, trace_prefix=None, timeout=None): diff --git a/tests/test_task_batch.py b/tests/test_task_batch.py index fcfad4e..0846336 100644 --- a/tests/test_task_batch.py +++ b/tests/test_task_batch.py @@ -162,6 +162,12 @@ class _FakeES: written.append((evtype, aid)) return "x" + # t-261015000631-13aef0: клейм теперь условный и атомарный (es.claim_task_if_ready); + # стаб без __getattr__-заглушки: всегда успешен, как готовая задача + def claim_task_if_ready(self, task_id, claim): + written.append(("TaskClaimed", task_id)) + return ("x", 1) + class _FakeRunner: @staticmethod diff --git a/tests/test_task_claim_race.py b/tests/test_task_claim_race.py new file mode 100644 index 0000000..12965b1 --- /dev/null +++ b/tests/test_task_claim_race.py @@ -0,0 +1,185 @@ +# -*- coding: utf-8 -*- +"""t-261015000631-13aef0: атомарный клейм — регресс гонки двойного claim. + +Инцидент 2026-10-15 00:00: L-задача t-261014221908-d20347 клеймнута ДВАЖДЫ +(TaskClaimed id=2767823 hour 00:00:00 и id=2767828 day 00:00:01) — оба уровня +читали read model до клейма друг друга, а старый цикл писал TaskClaimed по +снапшоту без перепроверки → двойное исполнение = двойной burn 9router. + +Проверяет: + 1) es.claim_task_if_ready: ready → успех; повторно/не-ready/несуществующая + → False; после requeue (TaskPromoted) → снова успех (обратимость); + 2) атомарность СТОРА: 8 потоков параллельно клеймят одну ready-задачу — + ровно один truthy-результат, ровно одно событие TaskClaimed; + 3) интеграция — точный сценарий прод-инцидента: hour- и day-тик с ОДНИМ + и тем же устаревшим снапшотом модели (оба видят задачу ready) → + ровно один TaskClaimed и один TaskCompleted, тик-проигравший + отрабатывает вхолостую без падения; + 4) свёртка статуса в сторе согласована с проекциями (done не ре-клеймится). + +Запуск: ./venv/bin/python tests/test_task_claim_race.py +""" +import os +import sys +import threading +from concurrent.futures import ThreadPoolExecutor + +_DB = "/tmp/simbiot-test-claim-race.db" +for _p in (_DB, _DB + "-wal", _DB + "-shm"): + if os.path.exists(_p): + os.unlink(_p) +os.environ["SIMBIOT_DB"] = _DB +sys.path.insert(0, "/root/simbiot-harness") + +from harness import es +from harness import commands + +es.init_db() + +fails = [] + + +def check(name, cond, extra=""): + print(("PASS" if cond else "FAIL"), name, ("" if cond else " | " + str(extra))) + if not cond: + fails.append(name) + + +def events_of(tid, etype=None): + db = es._conn() + try: + if etype: + rows = db.execute( + "SELECT * FROM events WHERE aggregate='task' AND aggregate_id=? " + "AND event_type=?", (tid, etype)).fetchall() + else: + rows = db.execute( + "SELECT * FROM events WHERE aggregate='task' AND aggregate_id=? " + "ORDER BY version", (tid,)).fetchall() + return [dict(r) for r in rows] + finally: + db.close() + + +def status_of(tid): + t = (es.run_projections().get("tasks") or {}).get(tid) + return (t or {}).get("status") + + +# ── 1. Семантика claim_task_if_ready ────────────────────────────────────── +tid, _ = commands.submit_task(title="гонка клейма: атомарный claim", + minutes=60, details="фикс t-261015000631") +check("1.1: ready → клейм успешен", bool(es.claim_task_if_ready( + tid, {"level": "hour", "session_id": "s-hour-1"}))) +check("1.2: running → клейм ОТКАЗАН (второй уровень не клеймнет)", + es.claim_task_if_ready(tid, {"level": "day", "session_id": "s-day-1"}) is False) +check("1.3: событий TaskClaimed ровно одно", + len(events_of(tid, "TaskClaimed")) == 1, str(len(events_of(tid, "TaskClaimed")))) +check("1.4: несуществующая задача → False", + es.claim_task_if_ready("t-nope-000000", {}) is False) + +# закрыли → клейм невозможен; requeue вернул в очередь → снова возможен +es.append_event("task", tid, "TaskCompleted", {"answer": "done"}) +check("1.5: done → клейм ОТКАЗАН", + es.claim_task_if_ready(tid, {"level": "hour", "session_id": "s-hour-2"}) is False) +commands.requeue_task(tid, reason="тест обратимости") +check("1.6: после requeue (TaskPromoted → ready) клейм снова успешен", + bool(es.claim_task_if_ready(tid, {"level": "min", "session_id": "s-min-1"}))) +check("1.7: свёртка стора == проекции (running)", + status_of(tid) == "running", status_of(tid)) + +# ── 2. Атомарность стора: 8 потоков, одна ready-задача ──────────────────── +tid2, _ = commands.submit_task(title="гонка клейма: 8 потоков", + minutes=60, details="один победитель") +barrier = threading.Barrier(8) + + +def _claim(_i): + barrier.wait() # максимальная конкуренция на одной ready-задаче + return es.claim_task_if_ready(tid2, {"level": "race", "session_id": f"s-{_i}"}) + + +with ThreadPoolExecutor(max_workers=8) as ex: + results = list(ex.map(_claim, range(8))) +winners = [r for r in results if r] +losers = [r for r in results if r is False] +check("2.1: ровно ОДИН победитель из 8 параллельных", len(winners) == 1, str(results)) +check("2.2: остальные 7 — чистый False (не None-дубль)", + len(losers) == 7, str(results)) +check("2.3: TaskClaimed в сторе ровно один", + len(events_of(tid2, "TaskClaimed")) == 1, + str(len(events_of(tid2, "TaskClaimed")))) +check("2.4: проекция видит running", status_of(tid2) == "running", status_of(tid2)) + +# ── 3. Интеграция: прод-сценарий hour+day со stale-снапшотом ─────────────── +from harness import scheduler +from harness.levels import LEVELS +import harness.isolated_runner as _ir + +tid3, _ = commands.submit_task(title="гонка клейма: hour vs day", + minutes=60, details="сценарий 2026-10-15 00:00") +# ОДИН снапшот модели для обоих тиков — как в проде: оба уровня читали модель +# ДО клейма друг друга. Снапшот устаревает сразу после первого клейма — +# именно это раньше и рождало второй TaskClaimed. +stale_model = es.run_projections() +check("3.1: подготовка — в снапшоте задача ready", + (stale_model["tasks"].get(tid3) or {}).get("status") == "ready") + +_orig_run_llm_tick = _ir.run_llm_tick + + +def _fake_run_llm_tick(level_cfg, prompt, session_id=None, trace_prefix=None, timeout=None): + return (f"дайджест {level_cfg.name}: отработана пачка", + {"model_calls": 1}, f"trace-{level_cfg.name}") + + +ticks_out = {} +tick_err = {} + + +def _run_tick(level_name, seq): + try: + # fresh-снапшот имитировать НЕ будем: оба тика получают ОДИН stale — + # худший случай. Внутри тика claim пойдёт в стор атомарно. + ticks_out[level_name] = scheduler.run_level_tick( + LEVELS[level_name], seq, stale_model) + except Exception as e: # noqa: BLE001 + tick_err[level_name] = repr(e) + + +try: + _ir.run_llm_tick = staticmethod(_fake_run_llm_tick) + barrier2 = threading.Barrier(2) + + def _run(level_name, seq): + barrier2.wait() + _run_tick(level_name, seq) + + th_hour = threading.Thread(target=_run, args=("hour", 4242)) + th_day = threading.Thread(target=_run, args=("day", 4243)) + th_hour.start(), th_day.start() + th_hour.join(), th_day.join() +finally: + _ir.run_llm_tick = _orig_run_llm_tick + +check("3.2: оба тика завершились без исключений", not tick_err, str(tick_err)) +claims3 = events_of(tid3, "TaskClaimed") +comps3 = events_of(tid3, "TaskCompleted") +check("3.3: TaskClaimed ровно ОДИН (регресс: было два)", + len(claims3) == 1, f"n={len(claims3)}") +check("3.4: TaskCompleted ровно ОДИН (исполнение не задвоилось)", + len(comps3) == 1, f"n={len(comps3)}") +winner_level = (claims3[0]["data"] if claims3 else "").find('"level"') >= 0 +check("3.5: клейм содержит level победителя", bool(winner_level), + claims3[0]["data"][:120] if claims3 else "нет событий") +loser = [lv for lv in ("hour", "day") + if lv != __import__("json").loads(claims3[0]["data"])["level"]][0] \ + if claims3 else "?" +check("3.6: проигравший тик отработал вхолостую (пустой батч, без падения)", + loser in ticks_out and ticks_out[loser][1] in ([], None), + str({k: v[1] for k, v in ticks_out.items()})) +check("3.7: финальный статус задачи done", status_of(tid3) == "done", status_of(tid3)) + +print() +print("FAILS:", len(fails)) +sys.exit(1 if fails else 0)