From 5d0651e70a8fd1ed5d260d83cc30e671c0b9ad9c Mon Sep 17 00:00:00 2001 From: host Date: Wed, 4 Nov 2026 09:16:25 +0000 Subject: [PATCH] gapfill G7: make _merge_task status commit and success audit atomic (same connection, audit_on_conn, one commit; audit failure rolls back to completed instead of merged-plus-500; ledger-head IntegrityError retries whole segment 5x mirroring AuthStore.audit); success semantics unchanged --- cloud/server.py | 47 ++++- cloud/tests/test_gapfill_g7_merge_atomic.py | 184 ++++++++++++++++++++ 2 files changed, 223 insertions(+), 8 deletions(-) create mode 100644 cloud/tests/test_gapfill_g7_merge_atomic.py diff --git a/cloud/server.py b/cloud/server.py index 8ba70fd..14da35b 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -68,6 +68,7 @@ from cloud.quota import quota_for_owner as _quota_for_owner from cloud.timeouts import STALE_SWEEP_SECONDS as _STALE_SWEEP from cloud.timeouts import approval_default_seconds as _approval_default from cloud.timeouts import validate_server_timeout as _validate_timeout +from cloud.auth import audit_on_conn as _audit_on_conn from cloud.db import connect as _db_connect from cloud.db import create_database as _create_database from cloud.db import init_postgres_schema as _init_postgres_schema @@ -6578,15 +6579,47 @@ class CloudControlPlaneHandler(OAuthMixin, IdentityMemoryMixin, ProjectMixin, Au "merge_gate": gate, }) return + # G7（A4 时序残留）:状态提交与成功审计必须同一事务。 + # 旧写法先 commit 状态、再另开 AuthStore 连接写审计： + # 审计失败时状态已是 merged、响应却是 500——“合并了却 + # 报失败”，调用方无法靠状态判断，也无法安全重试 + # （重复合并 409）。新写法用 audit_on_conn 把审计对 + # 写进同一连接、一次 commit：审计失败则整段回滚，状态 + # 仍是 completed、响应 500（_dispatch_safe 兜底），调用 + # 方可原样重试合并。链头唯一冲突的重试口径与 + # AuthStore.audit 完全一致（5 次、只重试 + # IntegrityError，其它错误直接抛）。成功语义不变： + # 200 体、审计内容（merge_task/task_id）、门禁响应全同。 + # 前提：tasks 与 audit_log/audit_ledger 在同一库 + # （_db 与 _auth_store 都连 server.db_path；PG 模式下 + # _connect 同样分发到同一 PG 库，? 占位由适配层转 %s）。 now = time.time() conn = self._db() try: - conn.execute( - "UPDATE tasks SET status = 'merged', updated_at = ?" - " WHERE id = ?", - (now, task_id), - ) - conn.commit() + last_exc = None + for _attempt in range(5): + try: + conn.execute( + "UPDATE tasks SET status = 'merged'," + " updated_at = ? WHERE id = ?", + (now, task_id), + ) + _audit_on_conn(conn, user_id, task["project_id"], + "merge_task", task_id) + conn.commit() + break + except Exception as exc: + last_exc = exc + try: + conn.rollback() + except Exception: + pass + if not isinstance( + exc, (sqlite3.IntegrityError, + _PgIntegrityError)): + raise + else: + raise last_exc task = _task_to_dict( conn.execute( "SELECT * FROM tasks WHERE id = ?", (task_id,) @@ -6594,8 +6627,6 @@ class CloudControlPlaneHandler(OAuthMixin, IdentityMemoryMixin, ProjectMixin, Au ) finally: conn.close() - with _auth_store(self.server.db_path) as store: - store.audit(user_id, task["project_id"], "merge_task", task_id) task["merge_gate"] = gate self._send_json(200, task) diff --git a/cloud/tests/test_gapfill_g7_merge_atomic.py b/cloud/tests/test_gapfill_g7_merge_atomic.py new file mode 100644 index 0000000..74b1fc7 --- /dev/null +++ b/cloud/tests/test_gapfill_g7_merge_atomic.py @@ -0,0 +1,184 @@ +"""gapfill G7：_merge_task 状态提交与成功审计同一事务。 + +旧写法（A4 时序残留）：先 commit 状态为 merged、再另开 +AuthStore 连接写 merge_task 审计；审计失败时状态已是 +merged、响应却是 500——“合并了却报失败”，调用方无法靠 +状态判断，也无法安全重试（重复合并 409）。 + +新写法：audit_on_conn 把审计对（audit_log 行 + 链条目） +写进同一连接、一次 commit；审计失败则整段回滚，状态仍是 +completed、异常上抛后由 _dispatch_safe 转 500 JSON，调用方 +可原样重试合并。链头唯一冲突的重试口径与 AuthStore.audit +完全一致（5 次、只重试 IntegrityError，其它错误直接抛）。 +成功语义不变：200 体、审计内容、门禁响应全同；本文件第 +一项钉住成功路径与旧行为一致，后三项钉住失败路径的新行为。 + +真实模型 0 轮，不联网；不跑真实 HTTP（handler 直调 + +真实 AuthStore/DB，审计行直接查库验证）。 +""" + +import os +import sqlite3 +import sys +import tempfile +import time +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest import mock + +REPO_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(REPO_ROOT)) + +import cloud.server as server_mod # noqa: E402 +from cloud.auth import AuthStore, audit_on_conn # noqa: E402 +from cloud.server import CloudControlPlaneHandler, init_db # noqa: E402 + +NO_BLOCK = {"merge_blocked": False, "release_blocked": False, + "blocked_by": None, "round": None, "failed_checks": []} + +HARD_BODY = { + "files_changed": ["apps/web/a.py"], + "owned_dir": "apps/web", + "branch_protection": "protected", + "ci_status": "success", + "ci_checks": [{"name": "ci", "status": "success"}], + "required_checks": ["ci"], + "contract_version": "v2", + "contract_frozen": True, + "coordinator_approved": True, + "unresolved_review_comments": 0, +} + + +class MergeAtomicTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + init_db(self.db_path) + self.auth = AuthStore(self.db_path) + self.owner_id, _key = self.auth.create_user("owner") + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "INSERT INTO projects(id, name, owner_id, created_at," + " repo_url, auto_pr) VALUES(?, ?, ?, ?, ?, 0)", + ("p1", "P", self.owner_id, time.time(), None)) + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " created_by, created_at, updated_at)" + " VALUES(?, ?, ?, ?, ?, ?, ?)", + ("t1", "p1", "do work", "completed", self.owner_id, + time.time(), time.time())) + conn.commit() + finally: + conn.close() + self.auth.create_project("p1", self.owner_id) + + def tearDown(self): + try: + self.auth.conn.close() + except Exception: + pass + self.tmp.cleanup() + + def _handler(self): + h = CloudControlPlaneHandler.__new__(CloudControlPlaneHandler) + h.server = SimpleNamespace(db_path=self.db_path) + sent = [] + h._send_json = lambda status, obj, extra_headers=None: sent.append( + (status, obj)) + h._read_json = lambda: (dict(HARD_BODY), None) + h.sent = sent + return h + + def _merge(self, h): + with mock.patch.object(server_mod, "_pushback_block", + return_value=dict(NO_BLOCK)): + h._merge_task(self.owner_id, "t1") + + def _status(self): + conn = sqlite3.connect(self.db_path) + try: + return conn.execute( + "SELECT status FROM tasks WHERE id = 't1'").fetchone()[0] + finally: + conn.close() + + def _merge_audits(self): + return [a for a in self.auth.list_audit("p1") + if a["action"] == "merge_task"] + + def test_success_unchanged(self): + # 成功语义与旧写法一致：200、merged、审计有行、门禁在体里。 + h = self._handler() + self._merge(h) + status, obj = h.sent[-1] + self.assertEqual(status, 200) + self.assertEqual(obj["status"], "merged") + self.assertIn("merge_gate", obj) + self.assertEqual(self._status(), "merged") + self.assertEqual(len(self._merge_audits()), 1) + self.assertEqual(self._merge_audits()[0]["detail"], "t1") + + def test_audit_failure_rolls_back_status(self): + # 旧行为是 merged + 500；新行为是 completed + 抛错 + # （_dispatch_safe 转 500 JSON），调用方可原样重试。 + h = self._handler() + with mock.patch.object( + server_mod, "_audit_on_conn", + side_effect=RuntimeError("audit down")): + with self.assertRaises(RuntimeError): + self._merge(h) + self.assertEqual(h.sent, []) + self.assertEqual(self._status(), "completed") + self.assertEqual(self._merge_audits(), []) + # 原样重试（审计恢复后）必须成功：状态仍是 completed， + # 不是卡在 merged 导致重复合并 409。 + h2 = self._handler() + self._merge(h2) + self.assertEqual(h2.sent[-1][0], 200) + self.assertEqual(self._status(), "merged") + + def test_ledger_conflict_retries_then_succeeds(self): + # 链头唯一冲突重试整段（与 AuthStore.audit 同口径）： + # 前两次冲突、第三次成功，整体仍是一次合并、一条审计。 + calls = {"n": 0} + + def flaky(conn, user_id, project_id, action, detail="", + created_at=None): + calls["n"] += 1 + if calls["n"] <= 2: + raise sqlite3.IntegrityError("chain head race") + return audit_on_conn(conn, user_id, project_id, action, + detail=detail, created_at=created_at) + + h = self._handler() + with mock.patch.object(server_mod, "_audit_on_conn", + side_effect=flaky): + self._merge(h) + self.assertEqual(h.sent[-1][0], 200) + self.assertEqual(calls["n"], 3) + self.assertEqual(self._status(), "merged") + self.assertEqual(len(self._merge_audits()), 1) + + def test_ledger_conflict_exhausted_rolls_back(self): + # 5 次冲突耗尽：抛错、状态回滚、审计无行、不留半截合并。 + calls = {"n": 0} + + def always_conflict(*args, **kwargs): + calls["n"] += 1 + raise sqlite3.IntegrityError("chain head race") + + h = self._handler() + with mock.patch.object(server_mod, "_audit_on_conn", + side_effect=always_conflict): + with self.assertRaises(sqlite3.IntegrityError): + self._merge(h) + self.assertEqual(calls["n"], 5) + self.assertEqual(self._status(), "completed") + self.assertEqual(self._merge_audits(), []) + + +if __name__ == "__main__": + unittest.main()