From 787543ea89784e5a0295bd52772e3404f8b59f82 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Tue, 13 Oct 2026 00:39:36 +0400 Subject: [PATCH] =?UTF-8?q?feat(relay):=20=D1=80=D0=B5=D0=BB=D0=B5=20?= =?UTF-8?q?=D1=87=D0=B5=D1=80=D0=B5=D0=B7=20=D0=BE=D0=B1=D1=89=D1=83=D1=8E?= =?UTF-8?q?=20=D0=BF=D0=B0=D0=BF=D0=BA=D1=83=20=E2=80=94=20=D0=B7=D0=B0?= =?UTF-8?q?=D0=BF=D1=80=D0=BE=D1=81=D1=8B=20=D0=B0=D0=B3=D0=B5=D0=BD=D1=82?= =?UTF-8?q?=D0=B0=20=D0=B8=D0=B7=20=D1=84=D0=B0=D0=B9=D0=BB=D0=BE=D0=B2=20?= =?UTF-8?q?=D0=B2=20=D1=81=D0=B2=D0=BE=D0=B9=20HTTP=20=D0=B8=20=D0=BE?= =?UTF-8?q?=D0=B1=D1=80=D0=B0=D1=82=D0=BD=D0=BE=20(#5)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- src/jira_hub/relay.py | 223 ++++++++++++++++++++++++++++++++++++++ tests/test_relay.py | 243 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 466 insertions(+) create mode 100644 src/jira_hub/relay.py create mode 100644 tests/test_relay.py diff --git a/src/jira_hub/relay.py b/src/jira_hub/relay.py new file mode 100644 index 0000000..8ab6f29 --- /dev/null +++ b/src/jira_hub/relay.py @@ -0,0 +1,223 @@ +"""Реле через общую папку (specs/hub-api, «Реле через общую папку»): файлы-запросы агента из /req/ +пересылаются в HTTP этого же хаба, ответы кладутся в /resp/. HTTP API, обработчики и токены не меняются; +реле для них — ещё один клиент. Нужно, когда VPN на виртуалке закрывает сеть до хаба (design.md folder-transport).""" +import json +import os +import threading +import time +import traceback +import urllib.error +import urllib.request +from datetime import datetime, timezone +from pathlib import Path + +from . import settings + +CLAIM_PATH = "/agent/claim" +CLAIM_MIN_LEFT = 35 # с: claim пересылается, только если до deadline осталось не меньше (хаб держит claim ≤ 30 с) +SETTINGS_EVERY = 5 # с: как часто перечитывать relay_dir +CLEANUP_EVERY = 60 # с: как часто убирать мусор +RESP_TTL = 600 # с: ответ, за которым агент не пришёл +REQ_GRACE = 60 # с после deadline: запрос, который никто не взял +TMP_TTL = 600 # с: брошенные временные файлы (имя на точку) +BAD_JSON_WAIT = 2 # с: сколько перечитывать нечитаемый запрос +FORWARD_TIMEOUT = 90 # с: ожидание ответа своего HTTP (claim держит до 30 с) +METHODS = ("GET", "POST", "PUT") + + +def say(msg: str) -> None: + print(f"[{time.strftime('%d/%b/%Y %H:%M:%S')}] {msg}", flush=True) + + +def parse_deadline(value) -> float | None: + """ISO 8601 → секунды эпохи; без зоны считается UTC; мусор → None.""" + if not isinstance(value, str): + return None + try: + dt = datetime.fromisoformat(value) + except ValueError: + return None + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt.timestamp() + + +def write_atomic(directory: Path, name: str, payload: dict) -> None: + """Записать JSON под временным именем на точку и переименовать: читатель видит файл только целиком.""" + tmp = Path(directory) / f".{name}.tmp" + with open(tmp, "w", encoding="utf-8") as f: + json.dump(payload, f, ensure_ascii=False) + f.flush() + os.fsync(f.fileno()) + os.replace(tmp, Path(directory) / name) + + +class Relay: + def __init__(self, hub, port: int, host: str = "127.0.0.1", poll: float = 0.1): + self.hub, self.port, self.host, self.poll = hub, port, host, poll + self.root: Path | None = None + self._settings_at: float | None = None + self._cleanup_at = 0.0 + self._ready = False + self._warned = False + self.workers: list[threading.Thread] = [] + + # --- цикл --------------------------------------------------------------------- + def run(self, stop: threading.Event) -> None: + while not stop.is_set(): + try: + self.step() + except Exception: # noqa: BLE001 — сбой одного шага не останавливает реле + traceback.print_exc() + stop.wait(self.poll) + + def step(self) -> None: + now = time.time() + if self._settings_at is None or now - self._settings_at >= SETTINGS_EVERY: + self._settings_at = now + self._reload() + if self.root is None: + return + if not self.root.is_dir(): + if not self._warned: + say(f"relay: папки {self.root} нет — жду") + self._warned = True + self._ready = False + return + if not self._ready: + self._prepare() + for path in sorted((self.root / "req").iterdir()): + if path.name.startswith(".") or not path.name.endswith(".json"): + continue + self._take(path) + if now - self._cleanup_at >= CLEANUP_EVERY: + self._cleanup_at = now + self.cleanup(now) + + def join(self, timeout: float = 5) -> None: + for w in list(self.workers): + w.join(timeout) + + # --- папка ---------------------------------------------------------------------- + def _reload(self) -> None: + conn = self.hub.conn() + try: + value = settings.get(conn, "relay_dir") + finally: + conn.close() + root = Path(value) if value else None + if root != self.root: + self.root, self._ready, self._warned = root, False, False + say(f"relay: папка {root}" if root else "relay: выключено") + + def _prepare(self) -> None: + for sub in ("req", "resp", "work"): + (self.root / sub).mkdir(exist_ok=True) + for stale in (self.root / "work").iterdir(): + stale.unlink(missing_ok=True) + say(f"relay: удалён незавершённый запрос {stale.name} (перезапуск реле)") + self._ready, self._warned = True, False + + def _take(self, path: Path) -> None: + target = self.root / "work" / path.name + try: + os.rename(path, target) + except FileNotFoundError: # агент успел удалить свой просроченный запрос + return + worker = threading.Thread(target=self.process, args=(self.root, target), daemon=True) + worker.start() + self.workers = [w for w in self.workers if w.is_alive()] + [worker] + + # --- обработка запроса ------------------------------------------------------------- + def process(self, root: Path, work: Path) -> None: + started = time.monotonic() + try: + req = self._read(work) + if req is None: + say(f"relay: нечитаемый запрос {work.name} — выброшен") + return + method, path = req.get("method"), req.get("path") + deadline = parse_deadline(req.get("deadline")) + left = deadline - time.time() if deadline is not None else -1.0 + if left <= 0 or (method == "POST" and path == CLAIM_PATH and left < CLAIM_MIN_LEFT): + say(f"relay {method} {path} просрочен (осталось {left:.0f} с) — выброшен") + return + status, body = self.forward(method, path, req.get("body"), req.get("authorization")) + write_atomic(root / "resp", work.name, {"id": req.get("id"), "status": status, "body": body}) + say(f"relay {method} {path} {status} {time.monotonic() - started:.1f} с") + except OSError as e: # папка пропала на ходу — агент дождётся своего срока + say(f"relay: сбой файла {work.name}: {e}") + finally: + work.unlink(missing_ok=True) + + def _read(self, work: Path) -> dict | None: + end = time.monotonic() + BAD_JSON_WAIT + while True: + try: + with open(work, encoding="utf-8") as f: + data = json.load(f) + return data if isinstance(data, dict) else None + except (ValueError, OSError): + if time.monotonic() >= end: + return None + time.sleep(0.1) + + def forward(self, method, path, body, authorization) -> tuple[int, str]: + if method not in METHODS or not isinstance(path, str) or not path.startswith("/"): + return 400, _error("bad_request", f"неверный метод или путь: {method} {path}") + data = json.dumps(body, ensure_ascii=False).encode("utf-8") if body is not None else None + try: + req = urllib.request.Request(f"http://{self.host}:{self.port}{path}", data=data, method=method) + if authorization: + req.add_header("Authorization", str(authorization)) + if data is not None: + req.add_header("Content-Type", "application/json; charset=utf-8") + with urllib.request.urlopen(req, timeout=FORWARD_TIMEOUT) as resp: + return resp.status, resp.read().decode("utf-8", "replace") + except urllib.error.HTTPError as e: + try: + return e.code, e.read().decode("utf-8", "replace") + finally: + e.close() + except ValueError as e: # недопустимый заголовок (перевод строки) или URL + return 400, _error("bad_request", str(e)) + except (urllib.error.URLError, OSError) as e: + return 502, _error("relay", f"хаб недоступен изнутри: {e}") + + # --- уборка ------------------------------------------------------------------------ + def cleanup(self, now: float) -> None: + if self.root is None or not self.root.is_dir(): + return + for p in _entries(self.root / "resp"): + ttl = TMP_TTL if p.name.startswith(".") else RESP_TTL + if _age(p, now) > ttl: + p.unlink(missing_ok=True) + for p in _entries(self.root / "req"): + if p.name.startswith("."): + if _age(p, now) > TMP_TTL: + p.unlink(missing_ok=True) + continue + try: + deadline = parse_deadline(json.loads(p.read_text(encoding="utf-8")).get("deadline")) + except (OSError, ValueError, AttributeError): + deadline = None + if deadline is None or now - deadline > REQ_GRACE: + p.unlink(missing_ok=True) + + +def _error(error: str, detail: str) -> str: + return json.dumps({"error": error, "detail": detail}, ensure_ascii=False) + + +def _entries(directory: Path) -> list[Path]: + try: + return list(directory.iterdir()) + except FileNotFoundError: + return [] + + +def _age(p: Path, now: float) -> float: + try: + return now - p.stat().st_mtime + except FileNotFoundError: + return 0.0 diff --git a/tests/test_relay.py b/tests/test_relay.py new file mode 100644 index 0000000..a032004 --- /dev/null +++ b/tests/test_relay.py @@ -0,0 +1,243 @@ +import json +import os +import threading +import time +from datetime import datetime, timedelta, timezone + +import pytest + +from jira_hub import api, relay, settings +from jira_hub.client import HubClient + + +@pytest.fixture +def env(tmp_path): + h = api.Hub(tmp_path / "hub.sqlite", backups_dir=tmp_path / "backups") + h.claim_wait = 0 + conn = h.conn() + root = tmp_path / "bridge" + root.mkdir() + settings.set_(conn, "relay_dir", str(root)) + tok = settings.tokens(conn) + conn.close() + srv = api.make_server(h, "127.0.0.1", 0) + th = threading.Thread(target=srv.serve_forever, daemon=True) + th.start() + r = relay.Relay(h, srv.server_port) + client = HubClient(f"http://127.0.0.1:{srv.server_port}", tok["client"]) + yield {"hub": h, "root": root, "relay": r, "agent": "Bearer " + tok["agent"], "client": client, + "port": srv.server_port} + srv.shutdown() + srv.server_close() + th.join(5) + + +def deadline(seconds): + return (datetime.now(timezone.utc) + timedelta(seconds=seconds)).isoformat() + + +def put_request(root, rid, method, path, auth, body=None, seconds=60, **extra): + req = {"id": rid, "method": method, "path": path, "body": body, "authorization": auth, + "deadline": deadline(seconds), **extra} + relay.write_atomic(root / "req", f"{rid}.json", req) + + +def response(root, rid): + p = root / "resp" / f"{rid}.json" + return json.loads(p.read_text(encoding="utf-8")) if p.exists() else None + + +def run_once(r): + r.step() + r.join() + + +def test_request_roundtrip(env): + r, root = env["relay"], env["root"] + r.step() # подготовка: req/, resp/, work/ + put_request(root, "a1", "GET", "/agent/settings", env["agent"]) + run_once(r) + resp = response(root, "a1") + assert resp["status"] == 200 and resp["id"] == "a1" + assert json.loads(resp["body"])["project_key"] == "AURORA" + assert not list((root / "work").iterdir()) and not list((root / "req").iterdir()) + + +def test_status_passes_through(env): + r, root = env["relay"], env["root"] + r.step() + put_request(root, "u1", "GET", "/issues", env["agent"]) # агентский токен на клиентский маршрут + put_request(root, "u2", "GET", "/agent/commands/999", env["agent"]) + run_once(r) + assert response(root, "u1")["status"] == 401 + assert response(root, "u2")["status"] == 404 + + +def test_expired_and_short_claim_dropped(env): + r, root, c = env["relay"], env["root"], env["client"] + b = c.post("/batches", {"title": "t", "author": "test", + "commands": [{"type": "ensure_fields", "params": {"key": "A-1", "duedate": "2026-10-23"}}]}) + r.step() + put_request(root, "c1", "POST", "/agent/claim", env["agent"], body={}, seconds=20) + put_request(root, "s1", "GET", "/agent/settings", env["agent"], seconds=-1) + run_once(r) + assert response(root, "c1") is None and response(root, "s1") is None + statuses = [x["status"] for x in c.get(f"/batches/{b['id']}")["commands"]] + assert statuses[0] == "queued" + assert not list((root / "work").iterdir()) + + +def test_claim_with_enough_time_is_forwarded(env): + r, root = env["relay"], env["root"] + r.step() + put_request(root, "c2", "POST", "/agent/claim", env["agent"], body={"agent_version": 2}, seconds=50) + run_once(r) + assert response(root, "c2")["status"] == 200 + + +def test_long_claim_does_not_block_other_requests(env): + env["hub"].claim_wait = 3 + r, root = env["relay"], env["root"] + r.step() + put_request(root, "c3", "POST", "/agent/claim", env["agent"], body={}, seconds=60) + r.step() + put_request(root, "s3", "GET", "/agent/settings", env["agent"]) + t0 = time.monotonic() + while response(root, "s3") is None and time.monotonic() - t0 < 2.5: + r.step() + time.sleep(0.05) + assert response(root, "s3")["status"] == 200 + assert response(root, "c3") is None # claim ещё ждёт + r.join() + assert response(root, "c3")["status"] == 200 + + +def test_unreachable_hub_gives_502(env): + r = relay.Relay(env["hub"], 1) # на порту 1 никто не слушает + root = env["root"] + r.step() + put_request(root, "p1", "GET", "/agent/settings", env["agent"]) + run_once(r) + resp = response(root, "p1") + assert resp["status"] == 502 and "хаб недоступен" in resp["body"] + + +def test_dotfiles_ignored_and_bad_json_dropped(env, monkeypatch, capsys): + monkeypatch.setattr(relay, "BAD_JSON_WAIT", 0.2) + r, root = env["relay"], env["root"] + r.step() + (root / "req" / ".x1.json.tmp").write_text("{}", encoding="utf-8") + (root / "req" / "bad.json").write_text("{не json", encoding="utf-8") + run_once(r) + assert (root / "req" / ".x1.json.tmp").exists() + assert response(root, "bad") is None and not list((root / "work").iterdir()) + assert "нечитаемый запрос bad.json" in capsys.readouterr().out + + +def test_bad_request_fields_get_400(env): + r, root = env["relay"], env["root"] + r.step() + put_request(root, "b1", "GET", "agent/settings", env["agent"]) # путь без / + put_request(root, "b2", "DELETE", "/agent/settings", env["agent"]) # метод не из списка + put_request(root, "b3", "GET", "/agent/settings", "Bearer x\r\nX-Evil: 1") # перевод строки в заголовке + run_once(r) + for rid in ("b1", "b2", "b3"): + assert response(root, rid)["status"] == 400, rid + + +def test_missing_deadline_is_dropped(env): + r, root = env["relay"], env["root"] + r.step() + relay.write_atomic(root / "req", "d1.json", {"id": "d1", "method": "GET", "path": "/agent/settings", + "authorization": env["agent"]}) + relay.write_atomic(root / "req", "d2.json", {"id": "d2", "method": "GET", "path": "/agent/settings", + "authorization": env["agent"], "deadline": "завтра"}) + run_once(r) + assert response(root, "d1") is None and response(root, "d2") is None + + +def test_vanished_request_is_skipped(env, capsys): + r, root = env["relay"], env["root"] + r.step() + put_request(root, "v1", "GET", "/agent/settings", env["agent"]) + orig = os.rename + + def vanish(src, dst): + os.remove(src) # агент удалил свой запрос раньше реле + return orig(src, dst) + + os.rename = vanish + try: + run_once(r) + finally: + os.rename = orig + assert response(root, "v1") is None + assert "Traceback" not in capsys.readouterr().err + + +def test_work_cleared_on_prepare(env, capsys): + r, root = env["relay"], env["root"] + (root / "work").mkdir() + (root / "work" / "old.json").write_text("{}", encoding="utf-8") + r.step() + assert not (root / "work" / "old.json").exists() + assert "old.json" in capsys.readouterr().out + + +def test_missing_dir_warns_once_then_works(env, capsys): + r, root = env["relay"], env["root"] + gone = root.parent / "bridge-later" + conn = env["hub"].conn() + settings.set_(conn, "relay_dir", str(gone)) + conn.close() + for _ in range(5): + r.step() + out = capsys.readouterr().out + assert out.count("нет — жду") == 1 + gone.mkdir() + r.step() + assert (gone / "req").is_dir() and (gone / "resp").is_dir() and (gone / "work").is_dir() + put_request(gone, "m1", "GET", "/agent/settings", env["agent"]) + run_once(r) + assert response(gone, "m1")["status"] == 200 + + +def test_disabled_and_reenabled_without_restart(env, monkeypatch): + r, root = env["relay"], env["root"] + conn = env["hub"].conn() + settings.set_(conn, "relay_dir", None) + conn.close() + r.step() + assert r.root is None and not (root / "req").exists() + monkeypatch.setattr(relay, "SETTINGS_EVERY", 0) + conn = env["hub"].conn() + settings.set_(conn, "relay_dir", str(root)) + conn.close() + r.step() + assert r.root == root and (root / "req").is_dir() + + +def test_cleanup_by_age(env): + r, root = env["relay"], env["root"] + r.step() + old, fresh = root / "resp" / "old.json", root / "resp" / "fresh.json" + tmp_old = root / "resp" / ".t.json.tmp" + for p in (old, fresh, tmp_old): + p.write_text("{}", encoding="utf-8") + now = time.time() + os.utime(old, (now - 700, now - 700)) + os.utime(tmp_old, (now - 700, now - 700)) + stale_req = root / "req" / ".q.json.tmp" + stale_req.write_text("{}", encoding="utf-8") + os.utime(stale_req, (now - 700, now - 700)) + r.cleanup(now) + assert not old.exists() and not tmp_old.exists() and not stale_req.exists() + assert fresh.exists() + + +def test_log_line(env, capsys): + r, root = env["relay"], env["root"] + r.step() + put_request(root, "l1", "GET", "/agent/settings", env["agent"]) + run_once(r) + assert "relay GET /agent/settings 200" in capsys.readouterr().out