diff --git a/src/jira_hub/api.py b/src/jira_hub/api.py index a66b193..1898b7d 100644 --- a/src/jira_hub/api.py +++ b/src/jira_hub/api.py @@ -111,7 +111,8 @@ def h_agent_result(hub, conn, m, q, body): def h_agent_state(hub, conn, m, q, body): with db.transaction(conn): - out = mirror.apply_state(conn, body.get("issues") or [], body.get("links") or [], body.get("neighbor_keys") or []) + out = mirror.apply_state(conn, body.get("issues") or [], body.get("links") or [], body.get("neighbor_keys") or [], + authoritative=queue.is_full_refresh(conn, body.get("command_id"))) if body.get("command_id"): queue.record_state(conn, body["command_id"], out["issues"], out["links"]) return 200, out diff --git a/src/jira_hub/mirror.py b/src/jira_hub/mirror.py index a8961ef..701e2f3 100644 --- a/src/jira_hub/mirror.py +++ b/src/jira_hub/mirror.py @@ -12,9 +12,12 @@ _FILTERS = {"sprint": "i.sprint = ?", "assignee": "i.assignee = ?", "status": "i "work_kind": "n.work_kind = ?", "fact_status": "n.fact_status = ?"} -def apply_state(conn, issues, links, neighbor_keys=(), fetched_at=None) -> dict: +def apply_state(conn, issues, links, neighbor_keys=(), fetched_at=None, authoritative=True) -> dict: """Upsert задач; связи с source_key среди присланных задач заменяются присланными. - Связи, у которых источник не прислан, не трогаются до загрузки их источника.""" + Связи, у которых источник не прислан, не трогаются до загрузки их источника. + authoritative=True — загрузка по базовой JQL: флаг neighbor пересчитывается по neighbor_keys (задача, вошедшая + в базовую выборку, перестаёт быть соседней). authoritative=False — точечное обновление (по ключам, по + произвольной JQL): флаг только поднимается, никогда не снижается.""" fetched_at = fetched_at or db.now() neighbors = set(neighbor_keys) keys = [] @@ -25,9 +28,11 @@ def apply_state(conn, issues, links, neighbor_keys=(), fetched_at=None) -> dict: row["fetched_at"] = fetched_at row["neighbor"] = 1 if it["key"] in neighbors else 0 cols = list(row) + sets = [f"{c} = excluded.{c}" for c in cols if c not in ("key", "neighbor")] + sets.append("neighbor = excluded.neighbor" if authoritative else "neighbor = MAX(issue.neighbor, excluded.neighbor)") conn.execute( f"INSERT INTO issue({', '.join(cols)}) VALUES ({', '.join('?' * len(cols))}) " - "ON CONFLICT(key) DO UPDATE SET " + ", ".join(f"{c} = excluded.{c}" for c in cols if c != "key"), + "ON CONFLICT(key) DO UPDATE SET " + ", ".join(sets), [row[c] for c in cols]) keys.append(it["key"]) conn.executemany("DELETE FROM issue_link WHERE source_key = ?", [(k,) for k in keys]) @@ -95,7 +100,7 @@ def set_note(conn, key, fields: dict, updated_by: str) -> dict: unknown = set(fields) - set(NOTE_FIELDS) if unknown: raise ValueError(f"неизвестные поля заметки: {', '.join(sorted(unknown))}") - with db.transaction(conn): # чтение, слияние и запись — атомарно (соединение общее для потоков) + with db.transaction(conn): # чтение, слияние и запись — атомарно (каждый запрос и поток открывает своё соединение, R-2) cur = get_note(conn, key) or {"key": key} cur.update(fields) cur["updated_at"], cur["updated_by"] = db.now(), updated_by diff --git a/src/jira_hub/queue.py b/src/jira_hub/queue.py index 2977e5c..9521e85 100644 --- a/src/jira_hub/queue.py +++ b/src/jira_hub/queue.py @@ -174,6 +174,15 @@ def claim(conn, agent_version=None, settings_version=None, last_jira_call_at=Non # --- отчёты ----------------------------------------------------------------- +def is_full_refresh(conn, command_id) -> bool: + """Команда — refresh по базовой JQL (полный прогон): только он вправе сбрасывать флаг neighbor.""" + if not command_id: + return False + cmd = get_command(conn, command_id) + return bool(cmd and cmd["type"] == "refresh" and cmd["params"].get("jql") + and cmd["params"]["jql"] == settings.get(conn, "base_jql")) + + def record_state(conn, command_id, issues_count, links_count) -> None: conn.execute("UPDATE sync_run SET issues_count = issues_count + ?, links_count = links_count + ? WHERE command_id = ?", (int(issues_count), int(links_count), command_id)) @@ -198,6 +207,11 @@ def report_result(conn, command_id, status, result=None, error=None, jira_calls= if cmd["type"] == "refresh": conn.execute("UPDATE sync_run SET finished_at = ?, jira_calls = ?, rate_limited = ? WHERE command_id = ?", (now, int(jira_calls or 0), int(rate_limited or 0), command_id)) + if status == "done" and is_full_refresh(conn, command_id): + # полный прогон не обновил задачу — в базовую JQL она больше не входит: сосед + run = conn.execute("SELECT started_at FROM sync_run WHERE command_id = ?", (command_id,)).fetchone() + if run: + conn.execute("UPDATE issue SET neighbor = 1 WHERE fetched_at < ?", (run["started_at"],)) if cmd["type"] == "refresh" and status == "done": reconcile(conn) return get_command(conn, command_id) diff --git a/tests/test_api.py b/tests/test_api.py index 00c55d7..055eb2c 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -241,3 +241,36 @@ def test_settings_put_invalid_is_4xx_and_atomic(hub): c.put("/settings", {"pacing": {"refresh": 3}, "tokens": {}}) assert 400 <= e.value.status < 500 assert c.get("/settings")["settings_version"] == v0 + + +def _run_refresh(hub, body, issues, neighbor_keys=()): + """Поставить refresh, взять его агентом, загрузить состояние, отчитаться done.""" + hub["client"].post("/refresh", body) + cmd = hub["agent"].post("/agent/claim", {})["command"] + assert cmd["type"] == "refresh" + hub["agent"].post("/agent/state", {"command_id": cmd["id"], "issues": issues, "links": [], + "neighbor_keys": list(neighbor_keys)}) + hub["agent"].post(f"/agent/commands/{cmd['id']}/result", {"status": "done"}) + + +def test_refresh_by_keys_keeps_neighbor_flag(hub): + """I-2(b): точечное обновление соседней задачи не опускает флаг.""" + _run_refresh(hub, {"full": True}, [{"key": "A-1", "summary": "т", "labels": []}, + {"key": "B-1", "summary": "сосед", "labels": []}], neighbor_keys=["B-1"]) + assert hub["client"].get("/issues/B-1")["neighbor"] is True + _run_refresh(hub, {"keys": ["B-1"]}, [{"key": "B-1", "summary": "сосед 2", "labels": []}]) + issue = hub["client"].get("/issues/B-1") + assert issue["summary"] == "сосед 2" and issue["neighbor"] is True + + +def test_full_refresh_marks_unseen_issue_as_neighbor(hub): + """I-2(a): полный прогон по базовой JQL: не обновлённая им задача получает neighbor=1.""" + conn = hub["hub"].conn() + conn.execute("INSERT INTO issue(key, summary, labels, fetched_at) VALUES ('X-1', 'вышла из выборки', '[]', '2020-01-01T00:00:00+00:00')") + conn.close() + _run_refresh(hub, {"full": True}, [{"key": "A-1", "summary": "т", "labels": []}]) + assert hub["client"].get("/issues/X-1")["neighbor"] is True + assert hub["client"].get("/issues/A-1")["neighbor"] is False + # частичный refresh (по ключам) других задач ничего не помечает + _run_refresh(hub, {"keys": ["A-1"]}, [{"key": "A-1", "summary": "т2", "labels": []}]) + assert hub["client"].get("/issues/A-1")["neighbor"] is False diff --git a/tests/test_mirror.py b/tests/test_mirror.py index d825f62..710b2f7 100644 --- a/tests/test_mirror.py +++ b/tests/test_mirror.py @@ -44,6 +44,15 @@ def test_apply_state_marks_neighbors(): assert mirror.known_keys(conn) == ["A-1", "B-9"] +def test_partial_load_never_lowers_neighbor(): + conn = db.connect(":memory:") + mirror.apply_state(conn, [issue("B-9")], [], neighbor_keys={"B-9"}) + mirror.apply_state(conn, [issue("B-9", summary="новая")], [], authoritative=False) # refresh {keys} + assert mirror.get_issue(conn, "B-9")["neighbor"] is True + mirror.apply_state(conn, [issue("B-9")], [], authoritative=True) # полный прогон: вошла в базовую + assert mirror.get_issue(conn, "B-9")["neighbor"] is False + + def test_note_survives_refresh_and_renames(): conn = db.connect(":memory:") mirror.set_note(conn, "new:7", {"fact_status": "В работе", "work_kind": "разработка"}, "import")