diff --git a/src/jira_hub/db.py b/src/jira_hub/db.py index a3b712d..9b16192 100644 --- a/src/jira_hub/db.py +++ b/src/jira_hub/db.py @@ -1,4 +1,5 @@ """Соединение с SQLite и схема хаба. Схема создаётся при первом подключении; версия — в setting.schema_version.""" +import itertools import json import sqlite3 from contextlib import contextmanager @@ -126,13 +127,27 @@ def migrate(conn: sqlite3.Connection) -> None: conn.execute("INSERT OR IGNORE INTO agent_status(id) VALUES (1)") +_savepoints = itertools.count(1) + + @contextmanager def transaction(conn: sqlite3.Connection): """BEGIN IMMEDIATE … COMMIT; при исключении — ROLLBACK. Нужна там, где читаем и пишем вместе (claim, отчёт). - Вложенный вызов (транзакция уже открыта) входит во внешнюю: своих BEGIN/COMMIT/ROLLBACK не делает, - исключение уходит внешнему блоку, который откатит всё.""" + Вложенный вызов (транзакция уже открыта) идёт через SAVEPOINT: при успехе RELEASE, при исключении + ROLLBACK TO + RELEASE и исключение пробрасывается — вызывающий может его перехватить, внешняя + транзакция при этом остаётся целой, а записи вложенного блока не сохраняются. + Соединение не должно делиться между потоками: каждый поток/запрос открывает своё.""" if conn.in_transaction: - yield conn + name = f"sp_{next(_savepoints)}" + conn.execute(f"SAVEPOINT {name}") + try: + yield conn + except BaseException: + conn.execute(f"ROLLBACK TO {name}") + conn.execute(f"RELEASE {name}") + raise + else: + conn.execute(f"RELEASE {name}") return conn.execute("BEGIN IMMEDIATE") try: diff --git a/src/jira_hub/queue.py b/src/jira_hub/queue.py index ba06d74..ed45ff7 100644 --- a/src/jira_hub/queue.py +++ b/src/jira_hub/queue.py @@ -1,12 +1,15 @@ """Очередь команд: пакеты, выдача агенту по одной, отчёты, отмена, повтор, сверка с зеркалом, автообновление. Хаб никогда не повторяет команды сам (design.md, решение 4).""" import json +import logging from datetime import timedelta from . import backup as _backup # noqa: F401 (имя модуля занято параметром backup в claim) from . import commands as C from . import db, mirror, settings +log = logging.getLogger(__name__) + TERMINAL = {"done", "failed", "skipped", "cancelled", "confirmed_by_state"} STALE_MINUTES = 10 # взятая команда старше — не блокирует выдачу, показывается как «нет ответа» ONLINE_MINUTES = 5 @@ -162,7 +165,10 @@ def claim(conn, agent_version=None, settings_version=None, last_jira_call_at=Non if first and has_writes: to_backup = f"batch-{cmd['batch_id']}" if to_backup and backup: - backup(to_backup) + try: + backup(to_backup) + except Exception: # сбой копии не должен держать очередь: команда уже взята + log.warning("резервная копия %s не создана, команда %s выдаётся без неё", to_backup, cmd["id"], exc_info=True) return {"id": cmd["id"], "type": cmd["type"], "params": cmd["params"], "pause_seconds": pause, "batch_id": cmd["batch_id"]} diff --git a/tests/test_db.py b/tests/test_db.py index a402771..0133f79 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -56,3 +56,16 @@ def test_nested_transaction_error_propagates_to_outer(tmp_path): def test_parse_and_utcnow(): assert db.parse("2020-01-01T00:00:00").tzinfo is not None assert db.parse(db.now()) <= db.utcnow() + + +def test_nested_transaction_rollback_is_local(tmp_path): + conn = db.connect(tmp_path / "hub.sqlite") + with db.transaction(conn): + conn.execute("INSERT INTO batch(title, author, created_at) VALUES ('outer', 'y', 'z')") + try: + with db.transaction(conn): + conn.execute("INSERT INTO batch(title, author, created_at) VALUES ('inner', 'y', 'z')") + raise RuntimeError("inner") + except RuntimeError: + pass + assert [r["title"] for r in conn.execute("SELECT title FROM batch")] == ["outer"] diff --git a/tests/test_queue.py b/tests/test_queue.py index 87aea6a..fe697d3 100644 --- a/tests/test_queue.py +++ b/tests/test_queue.py @@ -59,11 +59,37 @@ def test_claim_order_pause_and_report(): assert queue.claim(conn)["params"]["key"] == "A-2" and queue.batch_view(conn, b2["id"])["status"] == "running" -def test_claim_is_exclusive(): +def test_claim_is_exclusive(tmp_path): + import threading + path = tmp_path / "hub.sqlite" + setup = db.connect(path) + settings.ensure_defaults(setup) + queue.create_batch(setup, "x", "claude", [fields("A-1")]) + conns = [db.connect(path), db.connect(path)] + barrier, results = threading.Barrier(2), [] + + def worker(c): + barrier.wait() + results.append(queue.claim(c)) + + threads = [threading.Thread(target=worker, args=(c,)) for c in conns] + [t.start() for t in threads] + [t.join() for t in threads] + assert sorted(r is None for r in results) == [False, True] + assert setup.execute("SELECT COUNT(*) AS n FROM command WHERE status = 'claimed'").fetchone()["n"] == 1 + + +def test_backup_failure_does_not_block_claim(caplog): conn = fresh() - queue.create_batch(conn, "x", "claude", [fields("A-1"), fields("A-2")]) - first, second = queue.claim(conn), queue.claim(conn) - assert first is not None and second is None + queue.create_batch(conn, "x", "claude", [fields("A-1")]) + + def broken(label): + raise OSError("диск полон") + + c = queue.claim(conn, backup=broken) + assert c is not None and c["params"]["key"] == "A-1" + assert queue.get_command(conn, c["id"])["status"] == "claimed" + assert "не создана" in caplog.text def test_stale_claimed_does_not_block():