From 9ea0a05fb176408114131edfa7416a81c7a6136e Mon Sep 17 00:00:00 2001 From: host Date: Thu, 29 Oct 2026 15:57:46 +0800 Subject: [PATCH] host: fix state-loss from tmp collision and cross-container flock failure 2026-10-29 task 2 lost its host-state entry (version 390 -> single digits) while running under Docker/gVisor. Root cause, proven live: - save_state tmp name contained only pid, and every containerized host.py runs as pid 1, so concurrent writers shared one tmp file and interleaved writes produced corrupt JSON; - fcntl.flock does not exclude across runsc containers sharing a bind-mounted lock file (both entered at 0.0s in a live probe); - _read_state_file silently treated corrupt JSON as empty state, so the next save legitimately rewrote the file with only the surviving writer's tasks. Fix: - tmp name now hostname + pid + thread + uuid4 (backup tmp same); - mkdir atomic lock replaces flock (atomic on the host fs, holds across bind mounts/gofer), with stale-lock recovery after 120s and a loud timeout instead of unlocked writes; reads need no lock because publication is tmp + os.replace; - save path uses a strict disk read: corrupt JSON, non-object top level, or wrong-shaped tasks/automations now raise instead of overwriting (recovery stays with restore_state + .bak). Tests: tests/test_host_state_lock.py, 7 items (tmp uniqueness in process and across processes, 8-process concurrent saves all survive with version 8, stale lock recovery, fresh lock timeout raises, corrupt/wrong-shaped disk refuses to overwrite). Full suites: cloud 1168 (skipped 4), host 104, deploy 39, all green. --- host.py | 108 +++++++++++++++++++++++------ tests/test_host_state_lock.py | 123 ++++++++++++++++++++++++++++++++++ 2 files changed, 210 insertions(+), 21 deletions(-) create mode 100644 tests/test_host_state_lock.py diff --git a/host.py b/host.py index 0606b12..320b605 100644 --- a/host.py +++ b/host.py @@ -44,6 +44,7 @@ turn/completed 通知，所以多个 thread 的 turn 可以并发跑。 CODEX_BIN=.../bin/codex（start.sh 已自动优先用 bin/codex）。 """ import argparse +import contextlib import json import os import queue @@ -852,31 +853,93 @@ class Server: # 覆盖、自动化按名合并，这样两个进程各改各的任务时，后保存 # 的不会把先保存的覆盖掉。tmp 文件带 pid，避免两个进程用 # 同一个 .tmp 互相踩。锁文件在 state 文件旁边，名如 -# host-state.json.lock，用 fcntl.flock（Linux/macOS 都有）。 -import contextlib -try: - import fcntl -except ImportError: # Windows 等无 fcntl 的平台：退化为无锁 - fcntl = None +# 跨进程互斥绝不能再靠 fcntl.flock：2026-10-29 实测，runsc +# （gVisor）容器里 flock 只在各自 Sentry 内模拟，两个共享 +# bind 挂载锁文件的容器能同时拿到排他锁；而 tmp 文件名又只 +# 含 pid（容器内 host.py 全是 pid 1），并发保存写同一个 tmp， +# 交错出坏 JSON，读到坏 JSON 的下一次保存静默当成空 state +# 重写，全部任务条目一起丢（任务 2 实弹，版本 390 跌回个位数）。 +# 改用 mkdir 原子锁：mkdir 在宿主文件系统上原子成功才算拿到， +# 经 bind 挂载/gofer 也一样成立；持有者被杀留下的陈旧锁按 +# mtime 超龄清理；超时未拿到则大声失败，绝不无锁裸写。 +_STATE_LOCK_TIMEOUT = 30.0 +_STATE_LOCK_STALE = 120.0 + + +def _state_tmp_path(path): + # tmp 文件名必须每个写者唯一：旧写法只含 pid，容器内 host.py + # 全是 pid 1，并发保存写同一个 tmp 是断点丢失的直接原因。 + # hostname + pid + 线程 + uuid 四重区分；备份 tmp 同款。 + try: + import socket + host_name = socket.gethostname() + except Exception: + host_name = "unknown" + return (f"{path}.tmp.{host_name}.{os.getpid()}." + f"{threading.get_ident()}.{uuid.uuid4().hex}") @contextlib.contextmanager def _state_lock(path, exclusive=True): - lock_path = path + ".lock" - os.makedirs(os.path.dirname(os.path.abspath(lock_path)), exist_ok=True) - f = open(lock_path, "a+") + if not exclusive: + # 读不加锁：保存是 tmp + os.replace 原子发布，读到的 + # 永远是完整旧版或完整新版，不会读到半截文件。 + yield + return + lock_dir = path + ".lockdir" + os.makedirs(os.path.dirname(os.path.abspath(lock_dir)), exist_ok=True) + deadline = time.time() + _STATE_LOCK_TIMEOUT + while True: + try: + os.mkdir(lock_dir) + break + except FileExistsError: + try: + age = time.time() - os.path.getmtime(lock_dir) + except OSError: + age = 0 # 刚被别人删掉，重试 + if age > _STATE_LOCK_STALE: + try: + os.rmdir(lock_dir) + except OSError: + pass # 别人先清掉了，重试 + continue + if time.time() >= deadline: + raise RuntimeError( + f"state 锁超时未拿到（{lock_dir}），拒绝无锁写入，" + "防止并发保存互相覆盖") + time.sleep(0.05) try: - if fcntl is not None: - fcntl.flock(f.fileno(), - fcntl.LOCK_EX if exclusive else fcntl.LOCK_SH) yield finally: - if fcntl is not None: - try: - fcntl.flock(f.fileno(), fcntl.LOCK_UN) - except Exception: - pass - f.close() + try: + os.rmdir(lock_dir) + except OSError: + pass + + +def _read_state_file_for_save(path): + # 保存路径的严格读：文件存在但 JSON 损坏、顶层不是对象、 + # tasks/automations 类型不对时一律抛错，绝不当成空 state + # 去覆盖——旧的静默读空是断点丢失的最后一环：坏文件一旦 + # 被当成空 state 保存，全部任务条目就被合法地重写没了。 + # 文件不存在仍当空 state（首次保存）；.bak 恢复走 + # restore_state，不走这里。 + if not os.path.exists(path): + return {"tasks": {}, "automations": []} + try: + with open(path, encoding="utf-8") as f: + data = json.load(f) + except (ValueError, UnicodeDecodeError) as e: + raise RuntimeError( + f"state 文件 JSON 已损坏，拒绝覆盖（先用 .bak 恢复）：{path}：{e}") + if not isinstance(data, dict): + raise RuntimeError(f"state 文件顶层不是对象，拒绝覆盖：{path}") + if not isinstance(data.get("tasks", {}), dict): + raise RuntimeError(f"state 文件 tasks 不是对象，拒绝覆盖：{path}") + if not isinstance(data.get("automations", []), list): + raise RuntimeError(f"state 文件 automations 不是列表，拒绝覆盖：{path}") + return data def _read_state_file(path): @@ -1069,7 +1132,10 @@ def save_state(path, state): with _state_lock(path, exclusive=True): # 锁内重读 + 合并：内存任务覆盖同名磁盘任务，磁盘上 # 别的进程新增的任务保留；自动化按名合并，内存优先。 - disk = _read_state_file(path) + # 注意：这里必须用保存路径的严格读——磁盘文件已损坏时 + # 抛错而不是当成空 state，否则就是 2026-10-29 断点丢失 + # 的最后一环（坏文件被合法重写，全部条目消失）。 + disk = _read_state_file_for_save(path) loaded_version = state.get("_loaded_version", disk.get("version", 0)) if disk.get("version", 0) > loaded_version: # 别的写者（另一进程 / 另一台机器共享文件时）已先写过， @@ -1106,7 +1172,7 @@ def save_state(path, state): except Exception: merged["last_writer"] = str(os.getpid()) merged["updated_at"] = time.time() - tmp = f"{path}.tmp.{os.getpid()}" + tmp = _state_tmp_path(path) with open(tmp, "w") as f: json.dump(merged, f, indent=2, ensure_ascii=False) if os.path.exists(path): @@ -1114,7 +1180,7 @@ def save_state(path, state): backup = path + ".bak" backup_1 = backup + ".1" backup_2 = backup + ".2" - backup_tmp = f"{backup}.tmp.{os.getpid()}.{threading.get_ident()}" + backup_tmp = _state_tmp_path(backup) shutil.copy2(path, backup_tmp) if os.path.exists(backup_2): os.remove(backup_2) diff --git a/tests/test_host_state_lock.py b/tests/test_host_state_lock.py new file mode 100644 index 0000000..7758da5 --- /dev/null +++ b/tests/test_host_state_lock.py @@ -0,0 +1,123 @@ +"""host-state 并发保存回归测试（2026-10-29 断点丢失实弹）。 + +旧写法两处同时失效：tmp 文件名只含 pid（容器内 host.py 全是 +pid 1，并发保存写同一个 tmp 交错出坏 JSON）；flock 在 runsc +容器间不互斥（实测双双 0 秒拿到排他锁）。坏 JSON 又被 +_read_state_file 静默当成空 state，下一次保存合法地重写， +全部任务条目一起丢、版本从 390 跌回个位数。 + +本文件锁死三条修法：tmp 名每个写者唯一、mkdir 原子锁跨进程 +互斥、磁盘已损坏时保存直接拒绝覆盖。 +""" + +import glob +import json +import multiprocessing +import os +import sys +import tempfile +import time +import unittest +from pathlib import Path + + +REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO_ROOT)) + +import host + + +def _save_one(args): + path, name = args + state = {"tasks": {name: {"status": "running"}}, "automations": []} + host.save_state(path, state) + return name + + +def _tmp_name(_ignored): + return host._state_tmp_path("/tmp/x/host-state.json") + + +class StateTmpNameTest(unittest.TestCase): + def test_tmp_names_unique_in_process(self): + names = {host._state_tmp_path("/tmp/x/host-state.json") + for _ in range(100)} + self.assertEqual(len(names), 100) + + def test_tmp_names_unique_across_processes(self): + ctx = multiprocessing.get_context("fork") + with ctx.Pool(8) as pool: + names = pool.map(_tmp_name, range(8)) + self.assertEqual(len(set(names)), 8) + + +class StateMkdirLockTest(unittest.TestCase): + def test_concurrent_saves_all_tasks_survive(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "host-state.json") + ctx = multiprocessing.get_context("fork") + names = [f"task-{i}" for i in range(8)] + with ctx.Pool(8) as pool: + pool.map(_save_one, [(path, n) for n in names]) + with open(path, encoding="utf-8") as f: + data = json.load(f) + self.assertEqual(set(data["tasks"]), set(names)) + self.assertEqual(data["version"], 8) + self.assertFalse(glob.glob(path + ".tmp.*")) + self.assertFalse(os.path.exists(path + ".lockdir")) + + def test_stale_lock_is_recovered(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "host-state.json") + lock_dir = path + ".lockdir" + os.mkdir(lock_dir) + old = time.time() - 1000 + os.utime(lock_dir, (old, old)) + host.save_state(path, {"tasks": {"a": {}}, "automations": []}) + self.assertFalse(os.path.exists(lock_dir)) + with open(path, encoding="utf-8") as f: + self.assertIn("a", json.load(f)["tasks"]) + + def test_fresh_lock_timeout_raises_instead_of_writing_unlocked(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "host-state.json") + os.mkdir(path + ".lockdir") + old_timeout = host._STATE_LOCK_TIMEOUT + host._STATE_LOCK_TIMEOUT = 0.3 + try: + with self.assertRaises(RuntimeError): + host.save_state(path, {"tasks": {"a": {}}, + "automations": []}) + finally: + host._STATE_LOCK_TIMEOUT = old_timeout + self.assertFalse(os.path.exists(path)) + + +class StateCorruptRefusalTest(unittest.TestCase): + def test_corrupt_disk_refuses_to_overwrite(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "host-state.json") + garbage = b"{not json, torn write" + with open(path, "wb") as f: + f.write(garbage) + with self.assertRaises(RuntimeError): + host.save_state(path, {"tasks": {"a": {}}, + "automations": []}) + with open(path, "rb") as f: + self.assertEqual(f.read(), garbage) + self.assertFalse(os.path.exists(path + ".bak")) + self.assertFalse(os.path.exists(path + ".lockdir")) + + def test_wrong_shaped_disk_refuses_to_overwrite(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "host-state.json") + with open(path, "w", encoding="utf-8") as f: + json.dump({"tasks": ["not", "an", "object"], + "automations": []}, f) + with self.assertRaises(RuntimeError): + host.save_state(path, {"tasks": {"a": {}}, + "automations": []}) + + +if __name__ == "__main__": + unittest.main()