fix(hub): сбой резервной копии не блокирует выдачу, SAVEPOINT для вложенных транзакций, настоящий тест исключительной выдачи

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014Typ5sXWNMFirr2va2uL8o

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:

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"]}
 
 

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"]

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")])
+    queue.create_batch(conn, "x", "claude", [fields("A-1")])
-    first, second = queue.claim(conn), queue.claim(conn)
+
-    assert first is not None and second is None
+    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():