fix(queue): довыдача refresh не удваивает счётчики, предел выдач закрывает sync_run (#13)

M-4, M-5, M-10 (hub#12).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev 2026-10-14 12:46:29 +04:00
parent 4f0b873db6
commit 19aa032a25
2 changed files with 30 additions and 1 deletions

View File

@ -149,6 +149,8 @@ def _redeliver(conn, now) -> dict | None:
return cmd
conn.execute("UPDATE command SET status = 'failed', finished_at = ?, error = ? WHERE id = ?",
(now, f"агент не отчитался после {MAX_DELIVERIES} выдач", cmd["id"]))
if cmd["type"] == "refresh": # загрузка закончилась ничем — журнал синхронизаций закрыт (M-4 #12)
conn.execute("UPDATE sync_run SET finished_at = ? WHERE command_id = ?", (now, cmd["id"]))
conn.execute("UPDATE agent_status SET current_command_id = NULL WHERE id = 1 AND current_command_id = ?", (cmd["id"],))
log.warning("команда %s: агент не отчитался после %s выдач — failed", cmd["id"], MAX_DELIVERIES)
@ -165,6 +167,10 @@ def claim(conn, agent_version=None, settings_version=None, last_jira_call_at=Non
pause = int(settings.pacing(conn).get(cmd["type"], 0))
conn.execute("UPDATE command SET claimed_at = ?, pause_seconds = ?, deliveries = deliveries + 1 WHERE id = ?",
(now, pause, cmd["id"]))
if cmd["type"] == "refresh":
# повторное выполнение загрузит то же заново: счётчики с нуля; started_at — граница пометки соседей,
# не сдвигается (M-5 #12)
conn.execute("UPDATE sync_run SET issues_count = 0, links_count = 0 WHERE command_id = ?", (cmd["id"],))
conn.execute("UPDATE agent_status SET current_command_id = ? WHERE id = 1", (cmd["id"],))
log.info("команда %s выдана снова (выдача %s)", cmd["id"], cmd["deliveries"] + 1)
else:

View File

@ -264,7 +264,7 @@ def _two_batches_with_dependent(conn):
return a, b, first
def test_fresh_definer_still_blocks_queue():
def test_redelivered_definer_blocks_dependents():
conn = fresh()
a, b, first = _two_batches_with_dependent(conn)
assert queue.claim(conn)["id"] == first["id"] # довыдача определителя, зависимые ждут
@ -433,3 +433,26 @@ def test_empty_full_refresh_keeps_neighbors():
r = queue.claim(conn)
queue.report_result(conn, r["id"], "done") # загрузок состояния не было: 0 задач
assert conn.execute("SELECT neighbor FROM issue WHERE key = 'A-1'").fetchone()["neighbor"] == 0
def test_redelivered_refresh_resets_counts():
conn = fresh()
queue.create_batch(conn, "обновить", "claude", [{"type": "refresh", "params": {"keys": ["A-1"]}}])
r = queue.claim(conn)
queue.record_state(conn, r["id"], 150, 40)
started = conn.execute("SELECT started_at FROM sync_run WHERE command_id = ?", (r["id"],)).fetchone()["started_at"]
assert queue.claim(conn)["id"] == r["id"] # отчёт потерян — довыдача
queue.record_state(conn, r["id"], 150, 40)
run = conn.execute("SELECT * FROM sync_run WHERE command_id = ?", (r["id"],)).fetchone()
assert (run["issues_count"], run["links_count"]) == (150, 40)
assert run["started_at"] == started
def test_refresh_failed_by_limit_finishes_sync_run():
conn = fresh()
queue.create_batch(conn, "обновить", "claude", [{"type": "refresh", "params": {"keys": ["A-1"]}}])
r = queue.claim(conn)
for _ in range(queue.MAX_DELIVERIES):
queue.claim(conn)
assert queue.get_command(conn, r["id"])["status"] == "failed"
assert conn.execute("SELECT finished_at FROM sync_run WHERE command_id = ?", (r["id"],)).fetchone()["finished_at"]