185 lines
8.9 KiB
Python
185 lines
8.9 KiB
Python
# -*- 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)
|