fix(agent): код 73 при занятом журнале, одна строка темпа, источник ошибки хаба в refresh (M-7 #22; M-2, M-3, M-5, M-6 #18)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
c51651717c
commit
b90c639eaa
38
src/agent.py
38
src/agent.py
@ -34,6 +34,7 @@ from datetime import datetime, timezone
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
AGENT_VERSION = 9
|
||||
EXIT_LOCKED = 73 # «журнал занят другим процессом»: для run.py это не сбой версии (design 8)
|
||||
EXIT_UPDATE = 75 # «обновление готово»: run.py подменит agent.py на agent.py.new
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
API = "/rest/api/2"
|
||||
@ -283,7 +284,7 @@ class FolderTransport:
|
||||
os.remove(path)
|
||||
except OSError:
|
||||
pass
|
||||
status, text = int(resp.get("status") or 0), resp.get("body") or ""
|
||||
text = body or "" # status и body проверены выше (M-5 #18)
|
||||
if not 200 <= status < 300:
|
||||
raise HttpError(status, text)
|
||||
try:
|
||||
@ -502,6 +503,15 @@ class ConfigGap(Exception):
|
||||
"""В настройках хаба нет того, что нужно команде (R-11): команда падает с понятным текстом, а не молча теряет поле."""
|
||||
|
||||
|
||||
class HubError(Exception):
|
||||
"""Ошибка хаба внутри команды (refresh: known_keys, state): источник — хаб, не Jira (M-6 #18, design 12).
|
||||
Повтор — только при недоступности хаба и 5xx; refresh в Jira не пишет, повтор безопасен."""
|
||||
|
||||
def __init__(self, cause: Exception):
|
||||
super().__init__("хаб: %s" % cause)
|
||||
self.retryable = isinstance(cause, ConnectionError) or (isinstance(cause, HttpError) and cause.code >= 500)
|
||||
|
||||
|
||||
class Executor:
|
||||
def __init__(self, jira: JiraClient, settings: dict, hub: Optional[HubClient] = None):
|
||||
self.jira, self.s, self.hub = jira, settings, hub
|
||||
@ -519,6 +529,11 @@ class Executor:
|
||||
return handler(cmd.get("params") or {}, cmd)
|
||||
except ConfigGap as e:
|
||||
return self._out("failed", error=str(e))
|
||||
except HubError as e:
|
||||
out = self._out("failed", error=str(e))
|
||||
out["retryable"] = e.retryable
|
||||
out["peer"] = "хабом"
|
||||
return out
|
||||
except HttpError as e:
|
||||
out = self._out("failed", error="HTTP %d %s" % (e.code, e.text[:1500]))
|
||||
out["retryable"] = e.code >= 500 # исход неизвестен: Runner повторит через проверку (R-10)
|
||||
@ -532,6 +547,13 @@ class Executor:
|
||||
return {"status": status, "result": result, "error": error,
|
||||
"jira_calls": self.jira.calls, "rate_limited": self.jira.rate_limited}
|
||||
|
||||
def _hub(self, fn, *args):
|
||||
"""Вызов хаба из команды: его сбой — HubError, чтобы не выдать его за обрыв связи с Jira."""
|
||||
try:
|
||||
return fn(*args)
|
||||
except (ConnectionError, HttpError) as e:
|
||||
raise HubError(e) from None
|
||||
|
||||
def gap(self, cmd_type: str) -> None:
|
||||
"""Зазор между записями внутри одной команды — по политике хаба для этого типа."""
|
||||
seconds = int((self.s.get("pacing") or {}).get(cmd_type, 0) or 0)
|
||||
@ -907,7 +929,7 @@ class Executor:
|
||||
if not raws and errors: # ни один ключ не найден: это конфигурация/запрос, а не удалённые задачи
|
||||
raise errors[0]
|
||||
fetched = {r["key"] for r in raws}
|
||||
known = set(self.hub.known_keys()) if self.hub else set()
|
||||
known = set(self._hub(self.hub.known_keys)) if self.hub else set()
|
||||
linked = set()
|
||||
for r in raws:
|
||||
for l in self.links_from(r):
|
||||
@ -935,7 +957,7 @@ class Executor:
|
||||
chunk = issues[i:i + 100]
|
||||
chunk_keys = {x["key"] for x in chunk}
|
||||
pages += 1
|
||||
self.hub.state({"command_id": cmd.get("id"), "agent_version": AGENT_VERSION, "issues": chunk,
|
||||
self._hub(self.hub.state, {"command_id": cmd.get("id"), "agent_version": AGENT_VERSION, "issues": chunk,
|
||||
"links": [l for l in links if l["source_key"] in chunk_keys or l["target_key"] in chunk_keys],
|
||||
"neighbor_keys": [k for k in neighbor_keys if k in chunk_keys],
|
||||
"page": pages, "last_page": i + 100 >= len(issues)})
|
||||
@ -1136,7 +1158,8 @@ class Runner:
|
||||
ago = (datetime.now(timezone.utc) - at).total_seconds()
|
||||
if ago < 0:
|
||||
log("журнал: время последней записи в будущем (%s) — пауза перед первой командой полная" % at.isoformat())
|
||||
ago = 0
|
||||
self.jira.last_call_at = time.monotonic()
|
||||
return # вторая строка «не позже <будущее>» противоречила бы первой (M-2 #18)
|
||||
self.jira.last_call_at = time.monotonic() - ago
|
||||
log("темп: последний вызов Jira не позже %s (по журналу)"
|
||||
% at.astimezone(timezone.utc).replace(microsecond=0).isoformat())
|
||||
@ -1207,14 +1230,15 @@ class Runner:
|
||||
out = self.run_once(cmd)
|
||||
calls += out["jira_calls"]
|
||||
limited += out["rate_limited"]
|
||||
peer = out.pop("peer", "Jira") # с кем оборвалась связь: Jira или хаб (refresh)
|
||||
if not out.pop("retryable", False):
|
||||
break
|
||||
if attempt >= self.RETRY_ATTEMPTS:
|
||||
out = {"status": "failed", "result": None,
|
||||
"error": "результат неизвестен (обрыв связи с Jira), повтор безопасен: %s" % out.get("error")}
|
||||
"error": "результат неизвестен (обрыв связи с %s), повтор безопасен: %s" % (peer, out.get("error"))}
|
||||
break
|
||||
attempt += 1
|
||||
log(" ! обрыв связи с Jira, повтор %d/%d через проверку: %s" % (attempt, self.RETRY_ATTEMPTS, out.get("error")))
|
||||
log(" ! обрыв связи с %s, повтор %d/%d через проверку: %s" % (peer, attempt, self.RETRY_ATTEMPTS, out.get("error")))
|
||||
# F-1: обрыв не обновляет last_call_at, поэтому ждём фиксированно от момента отказа, а не от последнего вызова
|
||||
if not self.wait_seconds(max(float(cmd.get("pause_seconds") or 0), self.RETRY_MIN_WAIT)):
|
||||
return # остановка: finished не пишем, команду подхватит recover
|
||||
@ -1337,7 +1361,7 @@ def cmd_run(cfg: dict, journal_path: str, config_path: Optional[str] = None) ->
|
||||
lock = acquire_lock(journal_path)
|
||||
if lock is None:
|
||||
log("агент уже запущен с журналом %s — второй процесс не стартует" % journal_path)
|
||||
return 1
|
||||
return EXIT_LOCKED
|
||||
try:
|
||||
return _run(cfg, journal_path, config_path)
|
||||
finally:
|
||||
|
||||
@ -184,6 +184,29 @@ class RefreshTest(unittest.TestCase):
|
||||
self.assertEqual(self.hub.commands[cid]["status"], "done")
|
||||
self.assertEqual(self.hub.results[cid]["status"], "done")
|
||||
|
||||
def test_hub_unreachable_in_refresh_names_hub(self):
|
||||
self.fake.add("AURORA-1")
|
||||
self.ex.hub = agent.HubClient({"hub_url": "http://127.0.0.1:1", "hub_token": "t", "timeout": 2})
|
||||
out = self.ex.run({"id": 1, "type": "refresh", "params": {"keys": ["AURORA-1"]}})
|
||||
self.assertEqual(out["status"], "failed")
|
||||
self.assertTrue(out["error"].startswith("хаб: "), out["error"])
|
||||
self.assertTrue(out["retryable"])
|
||||
self.assertEqual(out["peer"], "хабом")
|
||||
|
||||
def test_hub_4xx_in_refresh_not_retryable(self):
|
||||
self.fake.add("AURORA-1")
|
||||
self.ex.hub = agent.HubClient({"hub_url": self.hub.base_url, "hub_token": "wrong-token", "timeout": 5})
|
||||
out = self.ex.run({"id": 1, "type": "refresh", "params": {"keys": ["AURORA-1"]}})
|
||||
self.assertEqual(out["status"], "failed")
|
||||
self.assertTrue(out["error"].startswith("хаб: HTTP 401"), out["error"])
|
||||
self.assertFalse(out["retryable"])
|
||||
|
||||
def test_jira_error_has_no_peer(self):
|
||||
self.fake.fail_next.append((500, {}))
|
||||
out = self.ex.run({"id": 1, "type": "refresh", "params": {"jql": "project = AURORA"}})
|
||||
self.assertTrue(out["retryable"])
|
||||
self.assertNotIn("peer", out)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@ -400,6 +400,7 @@ class RunnerTest(unittest.TestCase):
|
||||
r = self.hub.results[1]
|
||||
self.assertEqual(r["status"], "failed")
|
||||
self.assertIn("результат неизвестен", r["error"])
|
||||
self.assertIn("(обрыв связи с Jira)", r["error"])
|
||||
self.assertEqual(r["jira_calls"], 4)
|
||||
|
||||
def test_stop_during_retry_wait_leaves_command_for_recovery(self):
|
||||
@ -507,12 +508,55 @@ class RunnerTest(unittest.TestCase):
|
||||
self.assertIsNone(agent.acquire_lock(self.journal.path))
|
||||
cfg = jira_cfg(self.fake, hub_url=self.hub.base_url, hub_token=self.hub.token)
|
||||
with mock.patch.object(agent.Runner, "loop", lambda r: self.fail("цикл не должен стартовать")):
|
||||
self.assertEqual(agent.cmd_run(cfg, self.journal.path), 1)
|
||||
self.assertEqual(agent.cmd_run(cfg, self.journal.path), agent.EXIT_LOCKED)
|
||||
self.assertEqual(agent.EXIT_LOCKED, 73)
|
||||
finally:
|
||||
held.close()
|
||||
again = agent.acquire_lock(self.journal.path)
|
||||
self.assertIsNotNone(again) # после выхода первого — снова можно
|
||||
again.close()
|
||||
|
||||
def test_future_journal_time_single_log_line(self):
|
||||
at = (datetime.now(timezone.utc) + timedelta(hours=2)).replace(microsecond=0).isoformat()
|
||||
self.journal.append({"kind": "acked", "command_id": 99, "at": at})
|
||||
logs = []
|
||||
orig = agent.log
|
||||
agent.log = logs.append
|
||||
try:
|
||||
self.runner.restore_pace()
|
||||
finally:
|
||||
agent.log = orig
|
||||
self.assertEqual(len(logs), 1, logs) # M-2 #18: без «не позже <будущее>»
|
||||
self.assertIn("в будущем", logs[0])
|
||||
self.assertIsNotNone(self.runner.jira.last_call_at)
|
||||
|
||||
def test_future_journal_time_then_stop_in_pause(self):
|
||||
at = (datetime.now(timezone.utc) + timedelta(hours=2)).replace(microsecond=0).isoformat()
|
||||
self.journal.append({"kind": "acked", "command_id": 99, "at": at})
|
||||
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}, pause=30)
|
||||
self.runner.sleep = lambda s: setattr(self.runner, "stop", True) # Ctrl-C в первую секунду паузы
|
||||
self.runner.loop()
|
||||
self.assertEqual([r["kind"] for r in self.journal.records() if r["command_id"] == 1], ["claimed"])
|
||||
self.assertEqual(self.fake.requests, []) # в Jira ни одного вызова (M-3 #18)
|
||||
self.assertEqual(self.hub.commands[1]["status"], "claimed")
|
||||
|
||||
def test_retry_exhausted_on_hub_names_hub(self):
|
||||
self.runner.load_settings()
|
||||
logs = []
|
||||
orig = agent.log
|
||||
agent.log = logs.append
|
||||
self.runner.executor.run = lambda cmd: {"status": "failed", "result": None, "error": "хаб: не ответил",
|
||||
"jira_calls": 0, "rate_limited": 0, "retryable": True, "peer": "хабом"}
|
||||
self.hub.enqueue("refresh", {"keys": ["AURORA-1"]})
|
||||
try:
|
||||
self.runner.loop()
|
||||
finally:
|
||||
agent.log = orig
|
||||
r = self.hub.results[1]
|
||||
self.assertEqual(r["error"], "результат неизвестен (обрыв связи с хабом), повтор безопасен: хаб: не ответил")
|
||||
self.assertTrue(any("обрыв связи с хабом, повтор 1/3" in l for l in logs), logs)
|
||||
self.assertNotIn("peer", r)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user