diff --git a/cloud/db.py b/cloud/db.py index 6fe3846..fc6918b 100644 --- a/cloud/db.py +++ b/cloud/db.py @@ -324,7 +324,9 @@ def _load_postgres_driver(): # (语义弱于 REPLACE,调用点只有 project_members 与 # slack_users 两张已知主键表,不会走到退化分支)。 # - ``INSERT OR IGNORE`` -> ``ON CONFLICT DO NOTHING``:无冲突 -# 目标,任何唯一冲突都跳过,不改已有行。 +# 目标,任何唯一冲突都跳过,不改已有行;VALUES 形与 +# SELECT 形两种写法都翻译(SELECT 形漏翻曾致新鲜 PG 库 +# init 失败,见 _OR_IGNORE_SELECT_RE)。 # - ``PRAGMA table_info(<表>)`` -> information_schema 查询, # 返回 (cid, name, type, notnull, dflt_value, pk) 六元组, # 与 sqlite 的 PRAGMA table_info 行形状一致,调用方取 row[1] @@ -422,6 +424,13 @@ _OR_REPLACE_PK = { "memory_settings": ("user_id",), "task_memory_flags": ("task_id",), "memory_extraction_log": ("task_id",), + # 审计循环状态单行表:主键 task_id。未登记时 Postgres 下 + # INSERT OR REPLACE 退化为 ON CONFLICT DO NOTHING,已存在 + # 行的 round / last_decision 等更新静默丢弃(2026-10-31 + # PG 验证实测:跨进程竞态中包 round=2 而状态表 round 恒 + # 为 1、轮次耗尽时 last_decision 仍为 pushback;SQLite + # 原生 OR REPLACE 测不出来,与自动记忆三张表同款坑)。 + "task_audit_loop_state": ("task_id",), } _IDENTITY_TABLES = frozenset( @@ -437,6 +446,13 @@ _OR_IGNORE_RE = _re.compile( r"^\s*INSERT\s+OR\s+IGNORE\s+INTO\s+(\w+)\s*\(([^)]+)\)\s*" r"VALUES\s*\((.*)\)\s*$", _re.IGNORECASE | _re.DOTALL) +# INSERT OR IGNORE ... SELECT 形(VALUES 形之外的第二种写法, +# 如配额回填):PG 没有 OR IGNORE 语法,不翻译会直接报 +# syntax error at or near "OR"(2026-10-31 新鲜 PG 库复现)。 +_OR_IGNORE_SELECT_RE = _re.compile( + r"^\s*INSERT\s+OR\s+IGNORE\s+INTO\s+(\w+)\s*" + r"(\([^)]*\))?\s+SELECT\s+(.*)$", + _re.IGNORECASE | _re.DOTALL) _INSERT_TABLE_RE = _re.compile( r"^\s*INSERT\s+INTO\s+(?:\w+\.)?(\w+)", _re.IGNORECASE) _PRAGMA_INFO_RE = _re.compile( @@ -506,6 +522,22 @@ def _translate_sql(sql): "INSERT INTO %s (%s) VALUES (%s) ON CONFLICT DO NOTHING" % (table, cols, values)) return _convert_placeholders(translated), "normal" + m = _OR_IGNORE_SELECT_RE.match(stripped) + if m: + # INSERT OR IGNORE ... SELECT 形 -> INSERT INTO ... SELECT + # ... ON CONFLICT DO NOTHING:与 VALUES 形同款语义(任何 + # 唯一冲突都跳过、不改已有行)。旧版只认 VALUES 形,SELECT + # 形原样透传到 PG,直接报 syntax error at or near "OR"; + # 配额回填就是 SELECT 形,init_db 里它的 except 又把整个 + # 事务 rollback,migrate 的 ALTER 一并被丢掉(见 server + # _backfill_default_quotas 的保存点注释)。 + table = m.group(1) + cols = m.group(2) or "" + select_body = m.group(3).rstrip().rstrip(";") + translated = ( + "INSERT INTO %s %s SELECT %s ON CONFLICT DO NOTHING" + % (table, cols, select_body)) + return _convert_placeholders(translated), "normal" translated = _AUTOINCREMENT_RE.sub( "BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY", sql) translated = _convert_placeholders(translated) diff --git a/cloud/server.py b/cloud/server.py index 8b73837..5b57450 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -1642,8 +1642,28 @@ def _backfill_default_quotas(conn): 只补“users 里有、user_quotas 里无”的用户,已有行(含 管理员显式设为 NULL 的不限额行)一律不动。返回补写行数。 + + 失败只回滚到保存点,绝不回滚调用方的整个事务:init_db + 里 migrate 的 ALTER 与它同事务,旧版 except 里整事务 + rollback 会把 ALTER 一并丢掉(2026-10-31 新鲜 PG 库实测: + init 后 tasks 缺 paused_by_user / interrupted_reason,合流 + 随后在 PG“语句失败即 abort 整个事务”的语义下连环报 + InFailedSqlTransaction)。成功时也不再由本函数提交,提交 + 由调用方负责(init_db 结尾统一提交;旧版成功时提交等于 + 提前提交 init 的全部 DDL,改后 init 恢复原子性)。调用方 + 只有 init_db 两处(均以提交收尾),改动不影响其它路径。 + 保存点在 SQLite 与 PG 下都支持;保存点本身建不起来时 + (如连接已处 abort 态)退化为整事务 rollback 再返回 0。 """ _ensure_quotas_table(conn) + try: + conn.execute("SAVEPOINT backfill_default_quotas") + except Exception: + try: + conn.rollback() + except Exception: + pass + return 0 try: cur = conn.execute( "INSERT OR IGNORE INTO user_quotas(user_id," @@ -1652,13 +1672,17 @@ def _backfill_default_quotas(conn): " LEFT JOIN user_quotas q ON q.user_id = u.id" " WHERE q.user_id IS NULL", (_default_quota_usd_micros(), time.time())) - conn.commit() + conn.execute("RELEASE SAVEPOINT backfill_default_quotas") return cur.rowcount or 0 except Exception: try: - conn.rollback() + conn.execute("ROLLBACK TO SAVEPOINT backfill_default_quotas") + conn.execute("RELEASE SAVEPOINT backfill_default_quotas") except Exception: - pass + try: + conn.rollback() + except Exception: + pass return 0 diff --git a/cloud/tests/test_pg_init_fix.py b/cloud/tests/test_pg_init_fix.py new file mode 100644 index 0000000..a0356fe --- /dev/null +++ b/cloud/tests/test_pg_init_fix.py @@ -0,0 +1,309 @@ +"""PG init/backfill 回滚与审计状态 upsert 登记修复测试。 + +只覆盖两处修复,不跑真实模型回合、不联网: +- 问题一:新鲜 PG 库 init 后 tasks 缺 paused_by_user / + interrupted_reason,合流随后连环报 InFailedSqlTransaction。 + 根因是两层:(a) INSERT OR IGNORE ... SELECT 形在 PG 适配层 + 下未翻译,直接报 syntax error at or near "OR"; + (b) _backfill_default_quotas 的 except 整事务 rollback 把 + 同事务里 migrate 的 ALTER 一并丢掉。修法:翻译 SELECT 形、 + backfill 改保存点 containment、成功时不再由本函数提交。 +- 问题二:task_audit_loop_state 未登记进 _OR_REPLACE_PK, + PG 下 INSERT OR REPLACE 退化为 ON CONFLICT DO NOTHING, + 已存在行的 round / last_decision 等更新静默丢弃。 + +PG 测试与 test_postgres_integration.py 同款口径:须显式设置 +CLOUD_TEST_PG_DSN 且装有 psycopg 驱动,否则 skip,绝不 fail; +DSN 只从环境变量读,测试不打印它。 +""" + +import json +import os +import sqlite3 +import sys +import tempfile +import time +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest import mock + +try: + from cloud.audit_loop import LoopState, save_loop_state + from cloud.db import _translate_sql + from cloud.server import _backfill_default_quotas, init_db + from cloud.worker import Worker +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.audit_loop import LoopState, save_loop_state + from cloud.db import _translate_sql + from cloud.server import _backfill_default_quotas, init_db + from cloud.worker import Worker + +try: + from cloud.tests.test_postgres_backend import ( + _has_real_driver, + _real_pg_params, + ) + _PG_PARAMS = _real_pg_params() +except Exception: + _PG_PARAMS = None + + +def _pg_env(): + """返回 PG 测试用的环境变量 dict;无真实 PG 时返回 None。""" + if _PG_PARAMS is None or not _has_real_driver(): + return None + host, port, dbname, user, password = _PG_PARAMS + return { + "CLOUD_DB_BACKEND": "postgres", + "CLOUD_DB_HOST": host, + "CLOUD_DB_PORT": str(port), + "CLOUD_DB_NAME": dbname, + "CLOUD_DB_USER": user, + "CLOUD_DB_PASSWORD": password, + } + + +def _completed_runner(cmd): + return SimpleNamespace(returncode=0, stdout="{}", stderr="") + + +class TranslateTestCase(unittest.TestCase): + def test_ignore_select_form_translated(self): + sql, kind = _translate_sql( + "INSERT OR IGNORE INTO user_quotas(user_id," + " quota_usd_micros, updated_at)" + " SELECT u.id, ?, ? FROM users u" + " LEFT JOIN user_quotas q ON q.user_id = u.id" + " WHERE q.user_id IS NULL") + self.assertEqual(kind, "normal") + self.assertNotIn("OR IGNORE", sql) + self.assertIn("ON CONFLICT DO NOTHING", sql) + self.assertIn("%s", sql) + self.assertNotIn("?", sql) + + def test_audit_loop_state_or_replace_updates(self): + sql, kind = _translate_sql( + "INSERT OR REPLACE INTO task_audit_loop_state" + "(task_id, round, passed_gates, criteria, last_decision," + " updated_at) VALUES(?, ?, ?, ?, ?, ?)") + self.assertEqual(kind, "normal") + self.assertIn("ON CONFLICT (task_id) DO UPDATE SET", sql) + self.assertNotIn("DO NOTHING", sql) + self.assertIn("round = EXCLUDED.round", sql) + self.assertIn("last_decision = EXCLUDED.last_decision", sql) + + +class SqliteBackfillTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + + def tearDown(self): + self.tmp.cleanup() + + def _conn_with_users(self): + conn = sqlite3.connect(self.db_path) + conn.execute("CREATE TABLE users(id TEXT PRIMARY KEY)") + conn.execute("INSERT INTO users(id) VALUES('u1')") + conn.commit() + return conn + + def test_backfill_does_not_commit_caller_transaction(self): + # 旧版成功时 conn.commit(),会把调用方未提交的改动一并 + # 提交;改后提交由调用方负责,rollback 后调用方的改动 + # 与回填行应一并消失。 + conn = self._conn_with_users() + try: + conn.execute("INSERT INTO users(id) VALUES('u2')") + self.assertEqual(_backfill_default_quotas(conn), 2) + conn.rollback() + users = [r[0] for r in conn.execute( + "SELECT id FROM users ORDER BY id").fetchall()] + self.assertEqual(users, ["u1"]) + # 回滚连 _ensure 建的 user_quotas 表一起丢掉(同事务), + # 表不在即等于回填行已消失;表若在则行数必为 0。 + tables = [r[0] for r in conn.execute( + "SELECT name FROM sqlite_master" + " WHERE name = 'user_quotas'").fetchall()] + if tables: + quotas = conn.execute( + "SELECT COUNT(*) FROM user_quotas").fetchone()[0] + self.assertEqual(quotas, 0) + finally: + conn.close() + + def test_backfill_failure_rolls_back_only_to_savepoint(self): + # user_quotas 缺 quota_usd_micros 列(_ensure 为 + # CREATE TABLE IF NOT EXISTS,不会补列),INSERT 必失败; + # 旧版整事务 rollback 会把同事务里调用方的改动也丢掉, + # 保存点版只回滚回填本身。 + conn = sqlite3.connect(self.db_path) + try: + conn.execute("CREATE TABLE users(id TEXT PRIMARY KEY)") + conn.execute( + "CREATE TABLE user_quotas(user_id TEXT PRIMARY KEY," + " updated_at DOUBLE PRECISION)") + conn.execute("INSERT INTO users(id) VALUES('u1')") + conn.commit() + conn.execute("INSERT INTO users(id) VALUES('u2')") + self.assertEqual(_backfill_default_quotas(conn), 0) + users = [r[0] for r in conn.execute( + "SELECT id FROM users ORDER BY id").fetchall()] + self.assertEqual(users, ["u1", "u2"]) + finally: + conn.close() + + +class SqliteUpsertTestCase(unittest.TestCase): + def test_save_loop_state_updates_existing_row(self): + conn = sqlite3.connect(":memory:") + try: + conn.execute( + "CREATE TABLE task_audit_loop_state(task_id TEXT" + " PRIMARY KEY, round INTEGER, passed_gates TEXT," + " criteria TEXT, last_decision TEXT," + " updated_at DOUBLE PRECISION)") + save_loop_state(conn, LoopState( + task_id="P1", round=1, passed_gates=["child:a"], + criteria={"child:a": "child:a completes"}, + last_decision="pushback")) + save_loop_state(conn, LoopState( + task_id="P1", round=2, passed_gates=["child:a"], + criteria={"child:a": "child:a completes"}, + last_decision="escalate")) + conn.commit() + row = conn.execute( + "SELECT round, last_decision FROM task_audit_loop_state" + " WHERE task_id = 'P1'").fetchone() + self.assertEqual(row, (2, "escalate")) + finally: + conn.close() + + +class PgInitFixTestCase(unittest.TestCase): + """真实 PG 测试:无 DSN/驱动时全部 skip。""" + + def setUp(self): + env = _pg_env() + if env is None: + self.skipTest("未设置可达的 CLOUD_TEST_PG_DSN 或未装驱动," + "跳过真实 PG 测试") + self._patcher = mock.patch.dict(os.environ, env) + self._patcher.start() + self.addCleanup(self._patcher.stop) + self._old = os.environ.get("CODEX_PUSH_ALLOW_PRIVATE_TARGETS") + os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = "1" + self.addCleanup(self._restore_push_env) + # 一次性库由外部建好并指向 DSN;init_db 直接跑,不做 + # 任何夹具补偿(这正是本修复要证明的点)。 + init_db(":memory:") + self.worker = Worker(":memory:", "/fake/host.py", + runner=_completed_runner, + env={"XCM_API_KEY": "test-placeholder"}, + max_retries=0, retry_backoff_seconds=0) + self.worker._audit = lambda p, a, d="": None + + def _restore_push_env(self): + if self._old is None: + os.environ.pop("CODEX_PUSH_ALLOW_PRIVATE_TARGETS", None) + else: + os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = self._old + + def _pg(self): + from cloud.db import connect + return connect(":memory:") + + def test_fresh_pg_init_then_merge_without_failed_transaction(self): + conn = self._pg() + try: + cols = {r[1] for r in conn.execute( + "PRAGMA table_info(tasks)").fetchall()} + self.assertIn("paused_by_user", cols) + self.assertIn("interrupted_reason", cols) + now = time.time() + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " idempotency_key, depends_on, priority, created_by," + " created_at, updated_at, last_heartbeat, audit_loop," + " parent_task_id, coordinator_part, result_text," + " error_text, coordinator_parts, started_at)" + " VALUES('P1', 'p1', 'do', 'running', NULL, '[]', 0," + " 'tester', ?, ?, NULL, 1, NULL, NULL, NULL, NULL," + " ?, ?)", + (now, now, + json.dumps([{"name": "a", "on_failure": "fail"}]), now)) + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " idempotency_key, depends_on, priority, created_by," + " created_at, updated_at, last_heartbeat, audit_loop," + " parent_task_id, coordinator_part, result_text," + " error_text, coordinator_parts, started_at)" + " VALUES('C1', 'p1', 'do', 'failed', NULL, '[]', 0," + " 'tester', ?, ?, NULL, 0, 'P1', 'a', NULL, 'boom'," + " NULL, ?)", + (now, now, now)) + conn.commit() + finally: + conn.close() + # 不做任何补偿重跑 migrate:直接调合流,旧代码下这里 + # 会因缺列回落连环报 InFailedSqlTransaction。 + out = self.worker._coord_audit_loop_merge( + {"id": "P1", "project_id": "p1", "audit_loop": 1}, + [{"name": "a", "on_failure": "fail"}], + [{"name": "a", "id": "C1"}]) + self.assertEqual(out["decision"], "pushback") + + def test_pg_upsert_updates_existing_state_row(self): + conn = self._pg() + try: + save_loop_state(conn, LoopState( + task_id="P9", round=1, passed_gates=[], + criteria={}, last_decision="pushback")) + conn.commit() + save_loop_state(conn, LoopState( + task_id="P9", round=2, passed_gates=["child:a"], + criteria={"child:a": "child:a completes"}, + last_decision="escalate")) + conn.commit() + row = conn.execute( + "SELECT round, passed_gates, criteria, last_decision" + " FROM task_audit_loop_state WHERE task_id = 'P9'" + ).fetchone() + self.assertEqual(row[0], 2) + self.assertEqual(json.loads(row[1]), ["child:a"]) + self.assertEqual(json.loads(row[2]), + {"child:a": "child:a completes"}) + self.assertEqual(row[3], "escalate") + finally: + conn.close() + + def test_pg_backfill_failure_preserves_caller_transaction(self): + # 与 SQLite 版同款保存点断言,跑在 PG 上:缺列导致 + # INSERT 失败时,同事务里调用方的改动必须还在。 + conn = self._pg() + try: + conn.execute("DROP TABLE IF EXISTS user_quotas") + conn.execute( + "CREATE TABLE user_quotas(user_id TEXT PRIMARY KEY," + " updated_at DOUBLE PRECISION)") + conn.execute("CREATE TABLE IF NOT EXISTS users" + "(id TEXT PRIMARY KEY)") + conn.execute("INSERT INTO users(id) VALUES('u9')" + " ON CONFLICT DO NOTHING") + conn.commit() + conn.execute("INSERT INTO users(id) VALUES('u10')" + " ON CONFLICT DO NOTHING") + self.assertEqual(_backfill_default_quotas(conn), 0) + users = [r[0] for r in conn.execute( + "SELECT id FROM users WHERE id IN ('u9', 'u10')" + " ORDER BY id").fetchall()] + self.assertEqual(users, sorted(["u9", "u10"])) + conn.rollback() + finally: + conn.close() + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/worker.py b/cloud/worker.py index b6ded27..e006d91 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -5772,6 +5772,14 @@ class Worker: " WHERE id = ?", (cid,)).fetchone() fetched.append((cid, row)) except Exception: + # PG 下语句失败即 abort 整个事务:缺列的首次查询 + # 失败后,不先 rollback 就重查只会连环报 + # InFailedSqlTransaction。本连接只做 SELECT,rollback + # 无副作用;SQLite 下同样无害。 + try: + conn.rollback() + except Exception: + pass try: fetched = [] for cid in child_ids: @@ -5781,6 +5789,10 @@ class Worker: " FROM tasks WHERE id = ?", (cid,)).fetchone() fetched.append((cid, row)) except Exception: + try: + conn.rollback() + except Exception: + pass fetched = [] for cid in child_ids: row = conn.execute(