wiki: синхронизация — правки ревью Task 7 (#91)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev 2026-10-24 02:30:58 +04:00
parent bc69d5087a
commit b074b51ec0
9 changed files with 331 additions and 70 deletions

View File

@ -32,6 +32,8 @@ class CommandHooks:
on_late_result: Callable[[Connection, dict, str, dict | None, int, int], bool] on_late_result: Callable[[Connection, dict, str, dict | None, int, int], bool]
# (conn, cmd, status, result, files_dir): до транзакции отчёта — тяжёлая подготовка (файлы, перевод); None — нет # (conn, cmd, status, result, files_dir): до транзакции отчёта — тяжёлая подготовка (файлы, перевод); None — нет
before_result: Callable[[Connection, dict, str, dict | None, Path | None], None] | None = None before_result: Callable[[Connection, dict, str, dict | None, Path | None], None] | None = None
# (conn, cmd): отчёт не дошёл до after_result (ранний выход, исключение) — снять подготовку before_result
abort_result: Callable[[Connection, dict], None] | None = None
@dataclass(frozen=True) @dataclass(frozen=True)

View File

@ -296,37 +296,51 @@ def report_result(conn, command_id, status, result=None, error=None, jira_calls=
raise ValueError("status — done или failed") raise ValueError("status — done или failed")
error = None if error is None else str(error)[:ERROR_LIMIT] error = None if error is None else str(error)[:ERROR_LIMIT]
early = get_command(conn, command_id) early = get_command(conn, command_id)
if early is not None and early["status"] == "claimed": eh = _hooks(early["type"]) if early is not None and early["status"] == "claimed" else None
eh = _hooks(early["type"]) prepared = bool(eh and eh.before_result)
if eh and eh.before_result: # вне транзакции отчёта: модуль готовит тяжёлое, не держа блокировку базы completed = False
eh.before_result(conn, early, status, result, files_dir) try:
with db.transaction(conn): if prepared: # вне транзакции отчёта: модуль готовит тяжёлое, не держа блокировку базы
cmd = get_command(conn, command_id) try:
if cmd is None: eh.before_result(conn, early, status, result, files_dir)
raise KeyError(command_id) except ValueError as e: # не путать с «status — done или failed» (422 bad_status в API)
if cmd["status"] == "queued": raise RuntimeError(f"before_result: {e}") from e
raise NotClaimedError(f"команда {command_id} не выдана") with db.transaction(conn):
h = _hooks(cmd["type"]) cmd = get_command(conn, command_id)
if cmd["status"] in TERMINAL: if cmd is None:
# поздний отчёт по команде, которую хаб уже закрыл (force-отмена, предел выдач): модуль может принять данные, raise KeyError(command_id)
# статус не трогаем if cmd["status"] == "queued":
if h and h.on_late_result(conn, cmd, status, result, jira_calls, rate_limited): raise NotClaimedError(f"команда {command_id} не выдана")
return get_command(conn, command_id) h = _hooks(cmd["type"])
return cmd # идемпотентно; терминальный статус (cancelled/failed) не затирается if cmd["status"] in TERMINAL:
now = db.now() # поздний отчёт по команде, которую хаб уже закрыл (force-отмена, предел выдач): модуль может принять
# данные, статус не трогаем
if h and h.on_late_result(conn, cmd, status, result, jira_calls, rate_limited):
return get_command(conn, command_id)
return cmd # идемпотентно; терминальный статус не затирается
now = db.now()
if h:
status, why = h.check_result(conn, cmd, status, result, files_dir)
if why is not None:
error = why[:ERROR_LIMIT]
conn.execute("UPDATE command SET status = ?, finished_at = ?, result = ?, error = ?, jira_calls = ?, rate_limited = ? WHERE id = ?",
(status, now, _dump(result) if result is not None else None, error, int(jira_calls or 0), int(rate_limited or 0), command_id))
conn.execute("UPDATE agent_status SET current_command_id = NULL, last_seen_at = ? WHERE current_command_id = ?",
(now, command_id))
if h:
h.on_result(conn, cmd, status, result, jira_calls, rate_limited, now, files_dir)
if h: if h:
status, why = h.check_result(conn, cmd, status, result, files_dir) h.after_result(conn, cmd, status)
if why is not None: completed = True
error = why[:ERROR_LIMIT] return get_command(conn, command_id)
conn.execute("UPDATE command SET status = ?, finished_at = ?, result = ?, error = ?, jira_calls = ?, rate_limited = ? WHERE id = ?", finally:
(status, now, _dump(result) if result is not None else None, error, int(jira_calls or 0), int(rate_limited or 0), command_id)) if prepared and not completed: # ранний выход или исключение: модуль снимает подготовку (abort_result)
conn.execute("UPDATE agent_status SET current_command_id = NULL, last_seen_at = ? WHERE current_command_id = ?", ah = eh.abort_result
(now, command_id)) if ah is not None:
if h: try:
h.on_result(conn, cmd, status, result, jira_calls, rate_limited, now, files_dir) ah(conn, early)
if h: except Exception: # noqa: BLE001
h.after_result(conn, cmd, status) log.exception("команда %s: abort_result", command_id)
return get_command(conn, command_id)
def reconcile(conn) -> list[int]: def reconcile(conn) -> list[int]:

View File

@ -77,4 +77,5 @@ HOOKS = CommandHooks(
drop_unresolved=lambda cmd, unresolved: unresolved, retry_params=lambda cmd: cmd["params"], drop_unresolved=lambda cmd, unresolved: unresolved, retry_params=lambda cmd: cmd["params"],
check_confirmed=lambda conn, cmd: (False, None), on_claim=sync.on_claim, on_failed=sync.on_failed, check_confirmed=lambda conn, cmd: (False, None), on_claim=sync.on_claim, on_failed=sync.on_failed,
check_result=sync.check_result, on_result=sync.on_result, after_result=sync.after_result, check_result=sync.check_result, on_result=sync.on_result, after_result=sync.after_result,
on_late_result=lambda *a: False, before_result=sync.before_result) on_late_result=lambda *a: False, before_result=sync.before_result,
abort_result=sync.abort_result)

View File

@ -1,4 +1,5 @@
"""Корень вики проекта: поле wiki_root, значения для GET /projects, хуки создания и правки проекта.""" """Корень вики проекта: поле wiki_root, значения для GET /projects, хуки создания и правки проекта."""
import glob
import logging import logging
import re import re
import shutil import shutil
@ -74,7 +75,8 @@ def _drop_copy(conn, label: str) -> None:
base = paths.project_dir(conn, label) base = paths.project_dir(conn, label)
except RuntimeError: # база без файла (тесты в памяти): копии на диске нет except RuntimeError: # база без файла (тесты в памяти): копии на диске нет
return return
for d in (base, base.with_name(f".{base.name}.new"), base.with_name(f".{base.name}.old")): leftovers = sorted(base.parent.glob(glob.escape(f".{base.name}") + ".new*")) if base.parent.is_dir() else []
for d in (base, *leftovers, base.with_name(f".{base.name}.old")):
try: try:
shutil.rmtree(d) shutil.rmtree(d)
except FileNotFoundError: except FileNotFoundError:

View File

@ -6,6 +6,8 @@ from ...core.http import ApiError
from . import convert from . import convert
MAX_PAGES = 50 MAX_PAGES = 50
MAX_STAGE = 20000 # страниц одной команды в стадии (между пачками)
LIMITS = {"code": 200, "title": 1000, "url": 4000, "parent_code": 200, "eva_id": 200}
def _bad(code, field): def _bad(code, field):
@ -16,9 +18,9 @@ def _text(page, key, code, required=True, nullable=False):
v = page.get(key) v = page.get(key)
if v is None and (nullable or not required): if v is None and (nullable or not required):
return None return None
if not isinstance(v, str) or (required and not v.strip()): if not isinstance(v, str) or (required and not v.strip()) or len(v) > LIMITS.get(key, 10**9):
raise _bad(code, key) raise _bad(code, key)
return v return v if v.strip() else None # пустое необязательное поле (parent_code "") — как отсутствующее
def _check(i, page) -> dict: def _check(i, page) -> dict:
@ -26,7 +28,7 @@ def _check(i, page) -> dict:
if not isinstance(page, dict): if not isinstance(page, dict):
raise ApiError(422, "bad_pages", f"страница {i}: не объект") raise ApiError(422, "bad_pages", f"страница {i}: не объект")
code = page.get("code") code = page.get("code")
if not isinstance(code, str) or not code.strip(): if not isinstance(code, str) or not code.strip() or len(code) > LIMITS["code"]:
raise ApiError(422, "bad_pages", f"страница {i}: code") raise ApiError(422, "bad_pages", f"страница {i}: code")
title = _text(page, "title", code) title = _text(page, "title", code)
html = page.get("html") html = page.get("html")
@ -36,7 +38,7 @@ def _check(i, page) -> dict:
if orderno is not None and (not isinstance(orderno, int) or isinstance(orderno, bool)): if orderno is not None and (not isinstance(orderno, int) or isinstance(orderno, bool)):
raise _bad(code, "orderno") raise _bad(code, "orderno")
eva_id = page.get("eva_id") eva_id = page.get("eva_id")
if eva_id is not None and not isinstance(eva_id, str): if eva_id is not None and (not isinstance(eva_id, str) or len(eva_id) > LIMITS["eva_id"]):
raise _bad(code, "eva_id") raise _bad(code, "eva_id")
attachments, links = page.get("attachments") or [], page.get("links") or [] attachments, links = page.get("attachments") or [], page.get("links") or []
if not isinstance(attachments, list) or not all( if not isinstance(attachments, list) or not all(
@ -58,12 +60,12 @@ def accept(conn, command_id, pages) -> dict:
raise ApiError(422, "bad_request", "command_id — целое") raise ApiError(422, "bad_request", "command_id — целое")
if not isinstance(pages, list) or not pages or len(pages) > MAX_PAGES: if not isinstance(pages, list) or not pages or len(pages) > MAX_PAGES:
raise ApiError(422, "bad_pages", f"pages — список от 1 до {MAX_PAGES} страниц") raise ApiError(422, "bad_pages", f"pages — список от 1 до {MAX_PAGES} страниц")
checked = [_check(i, p) for i, p in enumerate(pages, 1)]
cmd = queue.get_command(conn, command_id) cmd = queue.get_command(conn, command_id)
if cmd is None: if cmd is None:
raise ApiError(404, "not_found", f"command {command_id}") raise ApiError(404, "not_found", f"command {command_id}")
if cmd["type"] != "wiki_sync" or cmd["status"] != "claimed": if cmd["type"] != "wiki_sync" or cmd["status"] != "claimed":
raise ApiError(409, "not_claimed", f"команда {command_id} не выдана агенту как wiki_sync") raise ApiError(409, "not_claimed", f"команда {command_id} не выдана агенту как wiki_sync")
checked = [_check(i, p) for i, p in enumerate(pages, 1)]
rows = [] rows = []
for p in checked: for p in checked:
try: try:
@ -77,5 +79,8 @@ def accept(conn, command_id, pages) -> dict:
cur = conn.execute("SELECT status FROM command WHERE id = ?", (command_id,)).fetchone() cur = conn.execute("SELECT status FROM command WHERE id = ?", (command_id,)).fetchone()
if cur is None or cur["status"] != "claimed": # состояние могло смениться, пока шёл перевод if cur is None or cur["status"] != "claimed": # состояние могло смениться, пока шёл перевод
raise ApiError(409, "not_claimed", f"команда {command_id} уже не в работе") raise ApiError(409, "not_claimed", f"команда {command_id} уже не в работе")
have = conn.execute("SELECT COUNT(*) AS n FROM wiki_sync_stage WHERE command_id = ?", (command_id,)).fetchone()["n"]
if have + len(rows) > MAX_STAGE:
raise ApiError(422, "bad_pages", f"pages: в команде больше {MAX_STAGE} страниц")
conn.executemany("INSERT OR REPLACE INTO wiki_sync_stage(command_id, code, payload) VALUES (?, ?, ?)", rows) conn.executemany("INSERT OR REPLACE INTO wiki_sync_stage(command_id, code, payload) VALUES (?, ?, ?)", rows)
return {"accepted": len(rows)} return {"accepted": len(rows)}

View File

@ -7,6 +7,7 @@ import json
import logging import logging
import re import re
import shutil import shutil
import sqlite3
import threading import threading
from dataclasses import dataclass, field from dataclasses import dataclass, field
from pathlib import Path from pathlib import Path
@ -38,11 +39,21 @@ class Prepared:
kept_ids: set = field(default_factory=set) kept_ids: set = field(default_factory=set)
new_ids: list = field(default_factory=list) # id вложений, добавленных подготовкой new_ids: list = field(default_factory=list) # id вложений, добавленных подготовкой
new_dir: Path | None = None new_dir: Path | None = None
old_dir: Path | None = None # прежняя копия после подмены: удаляется в after_result
committed: bool = False
root_title: str | None = None root_title: str | None = None
_PREPARED: dict[int, Prepared] = {} _PREPARED: dict[int, Prepared] = {}
_BUSY: set[Path] = set() # каталоги .new идущих подготовок (с начала сборки)
_lock = threading.Lock() _lock = threading.Lock()
_label_locks: dict[str, threading.Lock] = {} # подготовка…применение одного проекта — по одной
_tls = threading.local() # блокировка, взятая этим потоком в before_result
def _label_lock(label: str) -> threading.Lock:
with _lock:
return _label_locks.setdefault(label, threading.Lock())
# --- раскладка копии ---------------------------------------------------------- # --- раскладка копии ----------------------------------------------------------
@ -144,17 +155,18 @@ def build_copy(stage_pages: list[dict], codes: list[str], target: Path, ids: dic
# --- подготовка (до транзакции отчёта) --------------------------------------- # --- подготовка (до транзакции отчёта) ---------------------------------------
def _siblings(conn, label: str) -> tuple[Path, Path, Path]: def _siblings(conn, label: str, cmd_id: int) -> tuple[Path, Path, Path]:
"""(копия, каталог сборки этой команды, каталог прежней копии на время подмены)."""
cur = paths.project_dir(conn, label) cur = paths.project_dir(conn, label)
return cur, cur.with_name(f".{cur.name}.new"), cur.with_name(f".{cur.name}.old") return cur, cur.with_name(f".{cur.name}.new.{cmd_id}"), cur.with_name(f".{cur.name}.old")
def _allocate_ids(conn, label, pages) -> tuple[dict, list, list]: def _allocate_ids(conn, label, pages) -> tuple[dict, list, list]:
"""id вложений страниц: существующие по (страница, url) сохраняются, новые вставляются короткой транзакцией.""" """id вложений страниц: существующие по (страница, url) сохраняются, новые вставляются короткой транзакцией."""
ids, new_ids, attach = {}, [], [] ids, new_ids, attach = {}, [], []
existing = {(r["page_code"], r["url"]): r["id"] for r in conn.execute(
"SELECT id, page_code, url FROM wiki_attachment WHERE label = ?", (label,))}
with db.transaction(conn): with db.transaction(conn):
existing = {(r["page_code"], r["url"]): r["id"] for r in conn.execute(
"SELECT id, page_code, url FROM wiki_attachment WHERE label = ?", (label,))}
for page in pages: for page in pages:
for a in page["attachments"]: for a in page["attachments"]:
key = (page["code"], a["url"]) key = (page["code"], a["url"])
@ -182,6 +194,14 @@ def prepare(conn, cmd: dict, result: dict | None) -> Prepared:
"""До транзакции: проверить корень и список кодов, выделить id вложений, собрать `.new`. Ошибка — в p.error.""" """До транзакции: проверить корень и список кодов, выделить id вложений, собрать `.new`. Ошибка — в p.error."""
label, root = cmd["params"]["label"], cmd["params"]["root"] label, root = cmd["params"]["label"], cmd["params"]["root"]
p = Prepared(cmd["id"], label, root) p = Prepared(cmd["id"], label, root)
try:
_, p.new_dir, _ = _siblings(conn, label, cmd["id"])
except RuntimeError as e: # база без файла
p.error = str(e)
return p
with _lock: # с этого момента ни finish_pending_swaps, ни уборка не тронут .new
_PREPARED[cmd["id"]] = p
_BUSY.add(p.new_dir)
row = conn.execute("SELECT root_code FROM wiki_project WHERE label = ?", (label,)).fetchone() row = conn.execute("SELECT root_code FROM wiki_project WHERE label = ?", (label,)).fetchone()
if row is None or row["root_code"] != root: if row is None or row["root_code"] != root:
p.discarded = "корень сменился" p.discarded = "корень сменился"
@ -199,14 +219,17 @@ def prepare(conn, cmd: dict, result: dict | None) -> Prepared:
if c not in stage: if c not in stage:
raise SyncFailed(f"нет страницы {c}") raise SyncFailed(f"нет страницы {c}")
pages = [stage[c] for c in codes] pages = [stage[c] for c in codes]
_, p.new_dir, _ = _siblings(conn, label) cur, _, old = _siblings(conn, label, cmd["id"])
secure_dir(p.new_dir.parent) secure_dir(p.new_dir.parent)
shutil.rmtree(p.new_dir, ignore_errors=True) shutil.rmtree(p.new_dir, ignore_errors=True)
shutil.rmtree(old, ignore_errors=True) # остаток прошлой подмены: убираем здесь, не под блокировкой базы
ids, p.new_ids, p.attachments = _allocate_ids(conn, label, pages) ids, p.new_ids, p.attachments = _allocate_ids(conn, label, pages)
p.kept_ids = set(ids.values()) p.kept_ids = set(ids.values())
p.rows = build_copy(pages, codes, p.new_dir, ids) p.rows = build_copy(pages, codes, p.new_dir, ids)
title = (result or {}).get("root_title") title = (result or {}).get("root_title")
p.root_title = title if isinstance(title, str) and title.strip() else p.rows[0]["title"] p.root_title = title if isinstance(title, str) and title.strip() else p.rows[0]["title"]
except sqlite3.OperationalError:
raise # база занята: 500, агент повторит отчёт; abort_result всё уберёт
except SyncFailed as e: except SyncFailed as e:
p.error = str(e) p.error = str(e)
except ValueError as e: # fill_ids, путь вне копии except ValueError as e: # fill_ids, путь вне копии
@ -222,11 +245,43 @@ def prepare(conn, cmd: dict, result: dict | None) -> Prepared:
def before_result(conn, cmd, status, result, files_dir) -> None: def before_result(conn, cmd, status, result, files_dir) -> None:
"""До транзакции отчёта done: одна подготовка на проект (блокировка до after_result/abort_result)."""
if cmd["type"] != "wiki_sync" or status != "done": if cmd["type"] != "wiki_sync" or status != "done":
return return
p = prepare(conn, cmd, result) lock = _label_lock(cmd["params"]["label"])
lock.acquire()
_tls.held = lock
row = conn.execute("SELECT status FROM command WHERE id = ?", (cmd["id"],)).fetchone()
if row is None or row["status"] != "claimed": # параллельный отчёт уже закрыл команду
return
prepare(conn, cmd, result)
def _release(cmd_id: int) -> Prepared | None:
with _lock: with _lock:
_PREPARED[cmd["id"]] = p p = _PREPARED.pop(cmd_id, None)
if p is not None and p.new_dir is not None:
_BUSY.discard(p.new_dir)
lock = getattr(_tls, "held", None)
if lock is not None:
_tls.held = None
lock.release()
return p
def abort_result(conn, cmd) -> None:
"""Отчёт не дошёл до конца: подготовка снимается — каталог сборки, добавленные ею вложения, блокировка."""
p = _release(cmd["id"])
if p is None:
return
if p.new_dir is not None and not p.committed:
shutil.rmtree(p.new_dir, ignore_errors=True)
if p.new_ids and not p.committed:
try:
with db.transaction(conn):
_abandon(conn, p)
except sqlite3.Error:
log.warning("wiki_sync %s: вложения подготовки не удалены", cmd["id"], exc_info=True)
def _prepared(conn, cmd, result) -> Prepared: def _prepared(conn, cmd, result) -> Prepared:
@ -239,8 +294,6 @@ def check_result(conn, cmd, status, result, files_dir) -> tuple[str, str | None]
if cmd["type"] != "wiki_sync" or status != "done": if cmd["type"] != "wiki_sync" or status != "done":
return status, None return status, None
p = _prepared(conn, cmd, result) p = _prepared(conn, cmd, result)
with _lock:
_PREPARED[cmd["id"]] = p
if p.discarded: if p.discarded:
if isinstance(result, dict): if isinstance(result, dict):
result["discarded"] = p.discarded result["discarded"] = p.discarded
@ -252,9 +305,9 @@ def check_result(conn, cmd, status, result, files_dir) -> tuple[str, str | None]
# --- применение (в транзакции отчёта) ---------------------------------------- # --- применение (в транзакции отчёта) ----------------------------------------
def _swap(cur: Path, new: Path, old: Path) -> None: def _swap(cur: Path, new: Path, old: Path) -> bool:
"""wiki/<метка> → .old, .new → <метка>, .old удаляется. При сбое прежний каталог возвращается на место.""" """wiki/<метка> → .old, .new → <метка>; True, если прежняя копия была (её удалит after_result вне транзакции).
shutil.rmtree(old, ignore_errors=True) При сбое прежний каталог возвращается на место."""
had = cur.exists() had = cur.exists()
try: try:
if had: if had:
@ -264,7 +317,7 @@ def _swap(cur: Path, new: Path, old: Path) -> None:
if had and not cur.exists() and old.exists(): if had and not cur.exists() and old.exists():
old.rename(cur) old.rename(cur)
raise raise
shutil.rmtree(old, ignore_errors=True) return had
def _fail(conn, cmd, error) -> None: def _fail(conn, cmd, error) -> None:
@ -277,7 +330,7 @@ def on_result(conn, cmd, status, result, jira_calls, rate_limited, now, files_di
if cmd["type"] != "wiki_sync": if cmd["type"] != "wiki_sync":
return return
with _lock: with _lock:
p = _PREPARED.pop(cmd["id"], None) p = _PREPARED.get(cmd["id"])
if status != "done": if status != "done":
error = conn.execute("SELECT error FROM command WHERE id = ?", (cmd["id"],)).fetchone()["error"] error = conn.execute("SELECT error FROM command WHERE id = ?", (cmd["id"],)).fetchone()["error"]
on_failed_or_reported_failed(conn, cmd, error or "синхронизация не удалась") on_failed_or_reported_failed(conn, cmd, error or "синхронизация не удалась")
@ -298,6 +351,7 @@ def on_result(conn, cmd, status, result, jira_calls, rate_limited, now, files_di
try: try:
with db.transaction(conn): with db.transaction(conn):
_commit(conn, p, now) _commit(conn, p, now)
p.committed = True
except OSError as e: except OSError as e:
_abandon(conn, p) _abandon(conn, p)
_fail(conn, cmd, f"не удалось подменить копию: {e}") _fail(conn, cmd, f"не удалось подменить копию: {e}")
@ -319,8 +373,9 @@ def _commit(conn, p: Prepared, now: str) -> None:
conn.execute("UPDATE wiki_project SET synced_at = ?, pages_count = ?, root_title = ?, sync_error = NULL, failed_at = NULL " conn.execute("UPDATE wiki_project SET synced_at = ?, pages_count = ?, root_title = ?, sync_error = NULL, failed_at = NULL "
"WHERE label = ?", (now, len(p.rows), p.root_title, p.label)) "WHERE label = ?", (now, len(p.rows), p.root_title, p.label))
conn.execute("DELETE FROM wiki_sync_stage WHERE command_id = ?", (p.cmd_id,)) conn.execute("DELETE FROM wiki_sync_stage WHERE command_id = ?", (p.cmd_id,))
cur, new, old = _siblings(conn, p.label) cur, new, old = _siblings(conn, p.label, p.cmd_id)
_swap(cur, new, old) # последним: сбой откатывает строки (savepoint) if _swap(cur, new, old): # последним: сбой откатывает строки (savepoint)
p.old_dir = old
def on_failed_or_reported_failed(conn, cmd: dict, error: str | None) -> None: def on_failed_or_reported_failed(conn, cmd: dict, error: str | None) -> None:
@ -345,33 +400,42 @@ def on_claim(conn, cmd, now, redelivery) -> None:
def after_result(conn, cmd, status) -> None: def after_result(conn, cmd, status) -> None:
with _lock: """Вне транзакции: прежняя копия после подмены удаляется, блокировка проекта снимается."""
_PREPARED.pop(cmd["id"], None) p = _release(cmd["id"])
if p is not None and p.old_dir is not None:
try:
shutil.rmtree(p.old_dir)
except OSError as e:
log.warning("вики: прежняя копия %s не удалена (уберёт следующий старт): %s", p.old_dir, e)
# --- уборка ---------------------------------------------------------------- # --- уборка ----------------------------------------------------------------
_PENDING = re.compile(r"^\.(.+)\.(new(?:\.\d+)?|old)$", re.DOTALL)
def finish_pending_swaps(root: Path) -> None: def finish_pending_swaps(root: Path) -> None:
"""При старте: `.<имя>.new` без каталога `<имя>` — переименовать (подмена прервана); `.new` рядом с `<имя>` — """При старте: `.<имя>.new[.<id>]` с `_tree.json` (сборка закончена) без каталога `<имя>` — переименовать (подмена
остаток неудачной сборки, удаляется; `.<имя>.old` — удаляется. Каталоги, которые собирает идущая подготовка, не трогаются.""" прервана); без `_tree.json` или рядом с `<имя>` — остаток сборки, удаляется; `.<имя>.old` — удаляется. Каталоги
идущих подготовок не трогаются."""
root = Path(root) root = Path(root)
if not root.is_dir(): if not root.is_dir():
return return
with _lock: with _lock:
busy = {p.new_dir for p in _PREPARED.values() if p.new_dir} busy = set(_BUSY)
for entry in sorted(root.iterdir()): for entry in sorted(root.iterdir()):
n = entry.name m = _PENDING.match(entry.name)
if not n.startswith(".") or entry in busy: if m is None or entry in busy:
continue continue
try: try:
if n.endswith(".new"): if m.group(2) == "old":
target = root / n[1:-4]
if target.exists():
shutil.rmtree(entry, ignore_errors=True)
else:
entry.rename(target)
elif n.endswith(".old"):
shutil.rmtree(entry) shutil.rmtree(entry)
continue
target = root / m.group(1)
if target.exists() or not (entry / "_tree.json").is_file():
shutil.rmtree(entry, ignore_errors=True)
else:
entry.rename(target)
except OSError as e: except OSError as e:
log.warning("вики: не удалось довести подмену %s: %s", entry, e) log.warning("вики: не удалось довести подмену %s: %s", entry, e)

View File

@ -2,7 +2,7 @@
from datetime import timedelta from datetime import timedelta
from ...core import db, settings from ...core import db, settings
from . import commands, paths, sync from . import commands, paths, search, sync
def auto_sync_tick(hub, conn) -> None: def auto_sync_tick(hub, conn) -> None:
@ -34,6 +34,7 @@ def housekeeping_tick(hub, conn) -> None:
if root is not None and str(root) not in _swept: if root is not None and str(root) not in _swept:
_swept.add(str(root)) _swept.add(str(root))
sync.finish_pending_swaps(root) sync.finish_pending_swaps(root)
search.ensure(conn) # нет FTS5 — одно предупреждение в лог, поиск пойдёт по файлам
sync.housekeeping(conn) sync.housekeeping(conn)

View File

@ -51,9 +51,8 @@ def test_route_conflict():
def test_routes_assembled(): def test_routes_assembled():
from jira_hub.core import api from jira_hub.core import api
from jira_hub.modules import jira, wiki
pairs = lambda routes: {(m, p) for m, p, _, _ in routes} # noqa: E731 pairs = lambda routes: {(m, p) for m, p, _, _ in routes} # noqa: E731
mods = jira.MODULE.routes + wiki.MODULE.routes mods = tuple(r for m in registry.modules() for r in m.routes)
assert pairs(api.ROUTES) == pairs(api.CORE_ROUTES) | pairs(mods) assert pairs(api.ROUTES) == pairs(api.CORE_ROUTES) | pairs(mods)
assert not pairs(api.CORE_ROUTES) & pairs(mods) assert not pairs(api.CORE_ROUTES) & pairs(mods)
assert len(api.ROUTES) == len(api.CORE_ROUTES) + len(mods) assert len(api.ROUTES) == len(api.CORE_ROUTES) + len(mods)

View File

@ -239,12 +239,26 @@ def test_layout_orphans_and_cycles(tmp_path):
def test_finish_pending_swap(tmp_path): def test_finish_pending_swap(tmp_path):
root = tmp_path / "wiki" root = tmp_path / "wiki"
(root / ".AWPCA.new" / "x").mkdir(parents=True) (root / ".AWPCA.new" / "x").mkdir(parents=True)
(root / ".AWPCA.new" / "_tree.json").write_text("[]")
(root / ".AWPCA.old").mkdir() (root / ".AWPCA.old").mkdir()
(root / ".B.new").mkdir() (root / ".B.new").mkdir()
(root / "B").mkdir() # остаток неудачной сборки рядом с живой копией (root / "B").mkdir() # остаток неудачной сборки рядом с живой копией
(root / ".C.new.7").mkdir() # сборка без _tree.json прервана: не продвигается, удаляется
(root / ".D.new.8").mkdir()
(root / ".D.new.8" / "_tree.json").write_text("[]") # каталог с суффиксом команды, сборка закончена
sync.finish_pending_swaps(root) sync.finish_pending_swaps(root)
assert (root / "AWPCA" / "x").is_dir() and not (root / ".AWPCA.old").exists() and not (root / ".AWPCA.new").exists() assert (root / "AWPCA" / "x").is_dir() and not (root / ".AWPCA.old").exists() and not (root / ".AWPCA.new").exists()
assert (root / "B").is_dir() and not (root / ".B.new").exists() assert (root / "B").is_dir() and not (root / ".B.new").exists()
assert not (root / "C").exists() and not (root / ".C.new.7").exists() and (root / "D" / "_tree.json").is_file()
busy = root / ".E.new.9"
busy.mkdir()
(busy / "_tree.json").write_text("[]")
sync._BUSY.add(busy)
try:
sync.finish_pending_swaps(root) # идущая подготовка не трогается
finally:
sync._BUSY.discard(busy)
assert busy.is_dir() and not (root / "E").exists()
sync.finish_pending_swaps(root / "нет") # нет каталога — не ошибка sync.finish_pending_swaps(root / "нет") # нет каталога — не ошибка
@ -433,3 +447,162 @@ def test_batch_limit_and_replace(api):
api.wiki.post("/agent/wiki/pages", {"command_id": cmd["id"], "pages": [TREE[0]]}) api.wiki.post("/agent/wiki/pages", {"command_id": cmd["id"], "pages": [TREE[0]]})
api.wiki.post("/agent/wiki/pages", {"command_id": cmd["id"], "pages": [TREE[0] | {"title": "Новый"}]}) # повтор — замена api.wiki.post("/agent/wiki/pages", {"command_id": cmd["id"], "pages": [TREE[0] | {"title": "Новый"}]}) # повтор — замена
assert [json.loads(r["payload"])["title"] for r in api.rows("SELECT payload FROM wiki_sync_stage")] == ["Новый"] assert [json.loads(r["payload"])["title"] for r in api.rows("SELECT payload FROM wiki_sync_stage")] == ["Новый"]
# --- правки ревью (I1–I3, M1–M9) -------------------------------------------------
def _clean(api):
"""Ни подготовок, ни блокировки проекта, ни остатков сборки, ни вложений-сирот."""
assert sync._PREPARED == {} and sync._BUSY == set()
assert not sync._label_lock("AWPCA").locked()
assert not [p.name for p in api.wiki_dir.glob(".*")] if api.wiki_dir.exists() else True
def test_old_copy_removed_outside_transaction(api, monkeypatch):
first_sync(api)
api.enqueue()
seen = []
real = sync.shutil.rmtree
def probe(path, *a, **k):
if str(path).endswith(".old"):
other = sqlite3.connect(api.db_path, timeout=0)
try:
other.execute("BEGIN IMMEDIATE")
other.execute("ROLLBACK")
seen.append("free")
except sqlite3.OperationalError:
seen.append("locked")
finally:
other.close()
return real(path, *a, **k)
monkeypatch.setattr(sync.shutil, "rmtree", probe)
api.run(TREE, CODES)
assert seen and set(seen) == {"free"}
_clean(api)
def _attach_pages():
att = [{"ref": "/files/x/a.png", "url": "https://eva.example.ru/files/x/a.png", "name": "a.png"}]
return [TREE[0], page("DOC-3", "Статусы", "DOC-1", 1, '<img src="/files/x/a.png">', att)]
def test_exception_in_before_result_cleans_up(api, monkeypatch):
cmd = api.claim()
api.wiki.post("/agent/wiki/pages", {"command_id": cmd["id"], "pages": _attach_pages()})
real = sync.build_copy
def boom(*a, **k):
real(*a, **k)
raise sqlite3.OperationalError("database is locked")
monkeypatch.setattr(sync, "build_copy", boom)
with pytest.raises(HubError) as e:
api.report(cmd, "done", {"root_title": "X", "codes": ["DOC-1", "DOC-3"]})
assert e.value.status == 500 # не failed: агент повторит отчёт
assert api.rows("SELECT status FROM command WHERE id = ?", cmd["id"])[0]["status"] == "claimed"
assert api.rows("SELECT * FROM wiki_attachment") == [] # сирота убран
_clean(api)
monkeypatch.setattr(sync, "build_copy", real)
assert api.report(cmd, "done", {"root_title": "X", "codes": ["DOC-1", "DOC-3"]})["status"] == "done"
assert len(api.rows("SELECT * FROM wiki_attachment")) == 1
_clean(api)
def test_value_error_from_hook_is_not_422(api, monkeypatch):
cmd = api.claim()
monkeypatch.setattr(sync, "prepare", lambda *a: (_ for _ in ()).throw(ValueError("случайная")))
with pytest.raises(HubError) as e:
api.report(cmd, "done", {"root_title": "X", "codes": ["DOC-1"]})
assert e.value.status == 500 and e.value.error != "bad_status"
_clean(api)
def test_terminal_command_late_report_cleans_up(api, monkeypatch):
"""Команда закрыта между подготовкой и транзакцией (force-отмена): подготовка снята, копия не тронута."""
cmd = api.claim()
api.wiki.post("/agent/wiki/pages", {"command_id": cmd["id"], "pages": _attach_pages()})
real = sync.before_result
def cancel_after(conn, c, status, result, files_dir):
real(conn, c, status, result, files_dir)
other = api.hub.conn()
queue.cancel_command(other, c["id"], force=True)
other.close()
monkeypatch.setattr(sync, "before_result", lambda *a: cancel_after(*a))
# хуки заданы при импорте: подменяем и в самом наборе хуков
hooks = commands.HOOKS
object.__setattr__(hooks, "before_result", sync.before_result)
try:
out = api.report(cmd, "done", {"root_title": "X", "codes": ["DOC-1", "DOC-3"]})
finally:
object.__setattr__(hooks, "before_result", real)
assert out["status"] == "cancelled" and not api.copy.exists()
assert api.rows("SELECT * FROM wiki_attachment") == [] and api.rows("SELECT * FROM wiki_page") == []
_clean(api)
def test_repeat_done_is_idempotent(api):
cmd = api.claim()
api.wiki.post("/agent/wiki/pages", {"command_id": cmd["id"], "pages": TREE})
res = {"root_title": "АРМ КА", "codes": CODES}
assert api.report(cmd, "done", res)["status"] == "done"
again = api.report(cmd, "done", res)
assert again["status"] == "done" and api.project()["wiki_pages"] == 3
_clean(api)
assert len(api.rows("SELECT * FROM wiki_page")) == 3
def test_concurrent_done_reports(api):
import threading
cmd = api.claim()
api.wiki.post("/agent/wiki/pages", {"command_id": cmd["id"], "pages": _attach_pages()})
res = {"root_title": "АРМ КА", "codes": ["DOC-1", "DOC-3"]}
out, errs = [], []
def go():
try:
out.append(api.report(cmd, "done", res)["status"])
except Exception as e: # noqa: BLE001
errs.append(e)
ts = [threading.Thread(target=go) for _ in range(3)]
[t.start() for t in ts]
[t.join(30) for t in ts]
assert not errs and out == ["done"] * 3
assert api.project()["wiki_pages"] == 2
((att),) = api.rows("SELECT id FROM wiki_attachment")
md = (api.copy / "АРМ КА" / "01_Статусы" / "index.md").read_text(encoding="utf-8")
assert f"(attachment:{att['id']})" in md
_clean(api)
def test_bad_pages_after_state_check(api):
(row,) = api.rows("SELECT id FROM command WHERE type = 'wiki_sync'")
with pytest.raises(HubError) as e: # команда в queued: 409, а не 422 за форму
api.wiki.post("/agent/wiki/pages", {"command_id": row["id"], "pages": [{"code": "X"}]})
assert e.value.status == 409
def test_page_limits_and_free_code_format(api):
cmd = api.claim()
for field, value in (("title", "т" * 1001), ("url", "u" * 4001), ("code", "c" * 201)):
with pytest.raises(HubError) as e:
api.wiki.post("/agent/wiki/pages", {"command_id": cmd["id"], "pages": [TREE[0] | {field: value}]})
assert e.value.status == 422
api.wiki.post("/agent/wiki/pages", {"command_id": cmd["id"], "pages": [TREE[0] | {"code": "ABC-9"}]}) # формат не навязан
def test_empty_parent_code_is_null(api):
run = api.run([TREE[0] | {"parent_code": ""}], ["DOC-1"])
assert run["status"] == "done"
assert api.rows("SELECT parent_code FROM wiki_page")[0]["parent_code"] is None
def test_housekeeping_ensures_fts(open_db, tmp_path):
conn = open_db(tmp_path / "h.sqlite")
assert not conn.execute("SELECT 1 FROM sqlite_master WHERE name = 'wiki_page_fts'").fetchone()
tick.housekeeping_tick(None, conn)
assert conn.execute("SELECT 1 FROM sqlite_master WHERE name = 'wiki_page_fts'").fetchone()