weichenandCursor
2a03dd58f9
Why: U1 requires account ops to hit manifest data_root/DSN with OpenExisting and refuse missing flock or default ./data. Evidence: python3 -m unittest tests.test_svc_user_binding (T1–T5) + test_svc_identity Impact: svc <env> user … calls ENV_BIN with --open-existing; deploy/user fail closed without flock. Refs: docs/svc-deploy/u1-instance-account-binding.md; Backend OpenExisting (PR #171) Co-authored-by: Cursor <cursoragent@cursor.com>
397 lines
16 KiB
Python
397 lines
16 KiB
Python
"""svc.sh 自恢复回归：autostart 期望态 + recover + watchdog（2026-10-11）。
|
||
|
||
验证崩溃（SIGKILL，模拟 OOM 被杀）后 recover 按持久化登记拉回服务、显式 stop
|
||
不被复活、并发生命周期锁防抢跑、watchdog 单元生成正确。全部使用临时目录 +
|
||
桩二进制（bash + python http.server 模拟 agent-system；HTTP 子进程随父进程
|
||
退出——真实产物是单进程二进制，被杀即端口释放，桩须对齐该语义）。
|
||
"""
|
||
import fcntl
|
||
import os
|
||
import shutil
|
||
import socket
|
||
import subprocess
|
||
import tempfile
|
||
import time
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
SOURCE = Path(__file__).resolve().parents[1]
|
||
SVC_SCRIPT = SOURCE / "scripts" / "svc.sh"
|
||
|
||
# 桩 agent-system：version / config explain（环境层 DSN 感知）/ serve（记录参数后
|
||
# 用 http.server 应答健康探测）。TERM 转发给子进程保证 stop 干净；bash 自身不
|
||
# exec——pidcheck（二进制路径）须留在 cmdline。HTTP 子进程轮询 ppid：父进程被
|
||
# SIGKILL 后 0.5s 内自行退出（对齐单进程产物“进程死=端口释放”的语义）。
|
||
STUB = r"""#!/usr/bin/env bash
|
||
set -u
|
||
trace_dir="${SVC_STUB_TRACE:-$PWD/.svc-stub-trace}"
|
||
mkdir -p "$trace_dir"
|
||
case "${1:-}" in
|
||
version)
|
||
echo "stub-agent-system 1.0"
|
||
exit 0 ;;
|
||
config)
|
||
if [ "${2:-}" = "explain" ]; then
|
||
echo "config file: (none)"
|
||
echo "field value source"
|
||
if [ -n "${AGENT_SYSTEM_DSN:-}" ]; then
|
||
printf 'dsn postgresql://%s env\n' "$AGENT_SYSTEM_DSN"
|
||
else
|
||
echo "dsn - default"
|
||
fi
|
||
fi
|
||
exit 0 ;;
|
||
serve)
|
||
shift
|
||
case "${1:-}" in
|
||
-h|--help)
|
||
echo "Usage: agent-system serve -addr H:P -data DIR -plugins DIR -auth MODE -config FILE"
|
||
exit 0 ;;
|
||
esac
|
||
printf 'args: %s\n' "$*" > "$trace_dir/serve.txt"
|
||
addr=""
|
||
while [ $# -gt 0 ]; do
|
||
case "$1" in --addr) addr="$2"; shift 2 ;; *) shift ;; esac
|
||
done
|
||
port="${addr##*:}"; host="${addr%:*}"
|
||
trap 'kill "$child" 2>/dev/null; exit 0' TERM INT
|
||
python3 - "$port" "$host" <<'PYSRV' >/dev/null 2>&1 &
|
||
import os, sys, threading, time
|
||
from http.server import BaseHTTPRequestHandler, HTTPServer
|
||
port, host = int(sys.argv[1]), sys.argv[2]
|
||
class H(BaseHTTPRequestHandler):
|
||
def _ok(self, body=b'{"ok":true}'):
|
||
self.send_response(200)
|
||
self.send_header('Content-Type', 'application/json')
|
||
self.send_header('Content-Length', str(len(body)))
|
||
self.end_headers()
|
||
if self.command != 'HEAD':
|
||
self.wfile.write(body)
|
||
def do_GET(self):
|
||
self._ok()
|
||
def do_HEAD(self):
|
||
self._ok(b'')
|
||
def do_PUT(self):
|
||
n = int(self.headers.get('Content-Length') or 0)
|
||
self.rfile.read(n)
|
||
self._ok()
|
||
def log_message(self, *a):
|
||
pass
|
||
ppid = os.getppid()
|
||
def watch():
|
||
while True:
|
||
try:
|
||
if os.getppid() != ppid:
|
||
os._exit(0)
|
||
except Exception:
|
||
os._exit(0)
|
||
time.sleep(0.3)
|
||
threading.Thread(target=watch, daemon=True).start()
|
||
try:
|
||
HTTPServer((host, port), H).serve_forever()
|
||
except Exception:
|
||
os._exit(1)
|
||
PYSRV
|
||
child=$!
|
||
wait "$child"
|
||
;;
|
||
user)
|
||
printf 'user %s\n' "$*" >> "$trace_dir/cli.txt"
|
||
case "${2:-}" in
|
||
list) echo '[{"owner":"op","name":"op","isAdmin":true,"mustChangePassword":false}]' ;;
|
||
create) echo '{"owner":"new","issued":true}' ;;
|
||
esac
|
||
exit 0 ;;
|
||
*)
|
||
exit 0 ;;
|
||
esac
|
||
"""
|
||
|
||
|
||
class SvcAutoRecoverTests(unittest.TestCase):
|
||
def setUp(self):
|
||
tmp = tempfile.mkdtemp(prefix="svc-autorecover-")
|
||
self.addCleanup(shutil.rmtree, tmp, True)
|
||
self.root = Path(tmp)
|
||
(self.root / "scripts").mkdir(parents=True)
|
||
shutil.copy2(SVC_SCRIPT, self.root / "scripts" / "svc.sh")
|
||
self.trace = self.root / "trace"
|
||
self.stub = self.root / "stub-agent-system"
|
||
self.stub.write_text(STUB)
|
||
self.stub.chmod(0o755)
|
||
(self.root / "plugins").mkdir()
|
||
self.env = os.environ.copy()
|
||
self.env.update({
|
||
"GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": "/dev/null",
|
||
"GIT_TERMINAL_PROMPT": "0",
|
||
"GIT_AUTHOR_NAME": "svc recover", "GIT_COMMITTER_NAME": "svc recover",
|
||
"GIT_AUTHOR_EMAIL": "svc-test@example.invalid",
|
||
"GIT_COMMITTER_EMAIL": "svc-test@example.invalid",
|
||
"SVC_STUB_TRACE": str(self.trace),
|
||
"SVC_SRC_DIR": str(self.root / "srccache"),
|
||
# 隔离 watchdog 单元目录（--print 不落盘，但断言要在受控目录下做）。
|
||
"XDG_CONFIG_HOME": str(self.root / "xdg"),
|
||
})
|
||
for key in list(self.env):
|
||
if key.startswith("SVC_") and key not in ("SVC_STUB_TRACE", "SVC_SRC_DIR"):
|
||
self.env.pop(key)
|
||
for key in ("GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE"):
|
||
self.env.pop(key, None)
|
||
|
||
def tearDown(self):
|
||
# 兜底清理：停掉本测试遗留的服务进程，避免端口跨用例泄漏。
|
||
runtime = self.root / "runtime"
|
||
if runtime.is_dir():
|
||
for d in sorted(runtime.iterdir()):
|
||
if (d / "svc.pid").exists():
|
||
try:
|
||
subprocess.run(
|
||
["bash", str(self.root / "scripts" / "svc.sh"), d.name, "stop"],
|
||
cwd=self.root, env=self.env, timeout=30,
|
||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||
except subprocess.TimeoutExpired:
|
||
pass # 兜底强杀（见下）
|
||
# 直接 SIGKILL 兜底：确保桩进程（及其锁 fd/端口）不跨用例泄漏。
|
||
try:
|
||
pidf = d / "svc.pid"
|
||
if pidf.exists():
|
||
pid = int(pidf.read_text().strip() or 0)
|
||
if pid:
|
||
os.kill(pid, 9)
|
||
except (ValueError, ProcessLookupError, PermissionError, FileNotFoundError):
|
||
pass
|
||
|
||
# ---- 夹具 ----
|
||
|
||
def svc(self, *args, check=True, env=None):
|
||
proc = subprocess.run(
|
||
["bash", str(self.root / "scripts" / "svc.sh"), *map(str, args)],
|
||
cwd=self.root, env=env if env is not None else self.env, text=True,
|
||
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=120)
|
||
if check:
|
||
self.assertEqual(proc.returncode, 0, proc.stdout)
|
||
return proc
|
||
|
||
def runtime(self, name):
|
||
return self.root / "runtime" / name
|
||
|
||
def kv(self, name, key):
|
||
envfile = self.runtime(name) / "svc.env"
|
||
if not envfile.exists():
|
||
return None
|
||
for line in envfile.read_text().splitlines():
|
||
if line.startswith(key + "="):
|
||
return line[len(key) + 1:]
|
||
return None
|
||
|
||
def seed_agent(self, name, port):
|
||
d = self.runtime(name)
|
||
(d / "bin").mkdir(parents=True, exist_ok=True)
|
||
(d / "data").mkdir(exist_ok=True)
|
||
(d / "plugins").mkdir(exist_ok=True)
|
||
shutil.copy2(self.stub, d / "bin" / "agent-system")
|
||
(d / "svc.env").write_text("".join("%s=%s\n" % kv for kv in [
|
||
("kind", "agent"), ("env", name), ("pid", ""),
|
||
("addr", "127.0.0.1:%d" % port), ("tags", ""),
|
||
("source", "local@deadbee"), ("version", "stub"), ("started", ""),
|
||
("pidcheck", str(d / "bin" / "agent-system")),
|
||
("exec", "--addr 127.0.0.1:%d --data %s --plugins %s"
|
||
% (port, d / "data", d / "plugins")),
|
||
]))
|
||
|
||
def start_ok(self, name):
|
||
result = self.svc(name, "start")
|
||
self.assertIn("已启动", result.stdout)
|
||
return result
|
||
|
||
def pid_of(self, name):
|
||
pidf = self.runtime(name) / "svc.pid"
|
||
if not pidf.exists():
|
||
return None
|
||
text = pidf.read_text().strip()
|
||
return int(text) if text else None
|
||
|
||
def pid_alive(self, pid):
|
||
if not pid:
|
||
return False
|
||
try:
|
||
os.kill(pid, 0)
|
||
except (ProcessLookupError, PermissionError):
|
||
return False
|
||
# SIGKILL 后的孤儿在 CI 容器里可能成僵尸（PID 1 不收割）——僵尸按死处理，
|
||
# 与 svc.sh pid_alive_mark 的口径一致（僵尸 cmdline 为空 = 不匹配 = 已死）。
|
||
try:
|
||
state = Path("/proc/%d/stat" % pid).read_text().rsplit(")", 1)[1].split()[0]
|
||
if state == "Z":
|
||
return False
|
||
except (OSError, IndexError):
|
||
pass
|
||
return True
|
||
|
||
def kill9(self, name):
|
||
"""SIGKILL 服务主进程（模拟 OOM killer），保留 pidfile 与登记。"""
|
||
pid = self.pid_of(name)
|
||
self.assertTrue(self.pid_alive(pid), "服务应先在运行")
|
||
os.kill(pid, 9)
|
||
for _ in range(100): # 等 bash 主进程消失（HTTP 子进程由 ppid 看护自杀）
|
||
if not self.pid_alive(pid):
|
||
break
|
||
time.sleep(0.05)
|
||
self.assertFalse(self.pid_alive(pid))
|
||
time.sleep(0.6)
|
||
return pid
|
||
|
||
def history_events(self, name, event):
|
||
f = self.runtime(name) / "history.jsonl"
|
||
if not f.exists():
|
||
return []
|
||
out = []
|
||
for line in f.read_text().splitlines():
|
||
if ('"event":"%s"' % event) in line:
|
||
out.append(line)
|
||
return out
|
||
|
||
# ---- 1. 期望态：start 成功置位 / 显式 stop 清除（含进程已死的 stop）----
|
||
|
||
def test_start_records_autostart_and_stop_clears(self):
|
||
self.seed_agent("dev", 18801)
|
||
self.start_ok("dev")
|
||
self.assertEqual(self.kv("dev", "autostart"), "1")
|
||
self.svc("dev", "stop")
|
||
self.assertIsNone(self.kv("dev", "autostart"))
|
||
|
||
def test_stop_on_dead_process_clears_autostart(self):
|
||
self.seed_agent("dev", 18802)
|
||
self.start_ok("dev")
|
||
self.kill9("dev")
|
||
# 进程已死但 autostart 保留（等待 recover）；显式 stop 表示"保持停止"。
|
||
self.assertEqual(self.kv("dev", "autostart"), "1")
|
||
self.svc("dev", "stop")
|
||
self.assertIsNone(self.kv("dev", "autostart"))
|
||
|
||
# ---- 2. recover：崩溃拉回 / 运行中不动 / 显式停止不复活 / 未启动不碰 ----
|
||
|
||
def test_recover_restarts_sigkilled_service(self):
|
||
self.seed_agent("dev", 18803)
|
||
self.start_ok("dev")
|
||
old_pid = self.kill9("dev")
|
||
result = self.svc("recover", "--quiet")
|
||
self.assertEqual(result.returncode, 0, result.stdout)
|
||
self.assertIn("recover 汇总：拉回 1", result.stdout) # c_info 在 --quiet 下仍输出
|
||
new_pid = self.pid_of("dev")
|
||
self.assertTrue(self.pid_alive(new_pid))
|
||
self.assertNotEqual(new_pid, old_pid)
|
||
self.assertEqual(self.kv("dev", "autostart"), "1")
|
||
ok_events = self.history_events("dev", "recover")
|
||
self.assertTrue(any('"result":"ok"' in l for l in ok_events), ok_events)
|
||
# 恢复用的仍是持久化 exec=（svc.env 登记未漂移）。
|
||
self.assertIn("--addr 127.0.0.1:18803", self.kv("dev", "exec"))
|
||
|
||
def test_recover_noop_while_running(self):
|
||
self.seed_agent("dev", 18804)
|
||
self.start_ok("dev")
|
||
pid = self.pid_of("dev")
|
||
result = self.svc("recover")
|
||
self.assertEqual(result.returncode, 0, result.stdout)
|
||
self.assertEqual(self.pid_of("dev"), pid) # 未重启
|
||
self.assertEqual(self.history_events("dev", "recover"), [])
|
||
|
||
def test_recover_respects_explicit_stop(self):
|
||
self.seed_agent("dev", 18805)
|
||
self.start_ok("dev")
|
||
self.svc("dev", "stop")
|
||
result = self.svc("recover")
|
||
self.assertEqual(result.returncode, 0, result.stdout)
|
||
self.assertFalse(self.pid_alive(self.pid_of("dev")))
|
||
self.assertEqual(self.history_events("dev", "recover"), [])
|
||
|
||
def test_recover_ignores_never_started_service(self):
|
||
self.seed_agent("dev", 18806) # 从未 start：无 autostart 键
|
||
result = self.svc("recover")
|
||
self.assertEqual(result.returncode, 0, result.stdout)
|
||
self.assertFalse(self.pid_alive(self.pid_of("dev")))
|
||
self.assertEqual(self.history_events("dev", "recover"), [])
|
||
|
||
# ---- 3. 并发防护：生命周期锁被持有时 recover 跳过 ----
|
||
|
||
def test_recover_skips_when_lock_held(self):
|
||
self.seed_agent("dev", 18807)
|
||
self.start_ok("dev")
|
||
self.kill9("dev")
|
||
lock = self.runtime("dev") / ".start.lock"
|
||
with open(lock, "w") as fh:
|
||
fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||
result = self.svc("recover")
|
||
self.assertEqual(result.returncode, 0, result.stdout)
|
||
self.assertIn("锁被占用", result.stdout)
|
||
self.assertFalse(self.pid_alive(self.pid_of("dev")))
|
||
self.assertEqual(self.history_events("dev", "recover"), [])
|
||
fcntl.flock(fh, fcntl.LOCK_UN)
|
||
# 锁释放后同一命令即可拉回。
|
||
result = self.svc("recover")
|
||
self.assertEqual(result.returncode, 0, result.stdout)
|
||
self.assertTrue(self.pid_alive(self.pid_of("dev")))
|
||
|
||
# ---- 4. 恢复失败路径：端口被外部进程占用 → fail 事件 + 非零退出 ----
|
||
|
||
def test_recover_fails_when_port_stolen(self):
|
||
self.seed_agent("dev", 18808)
|
||
self.start_ok("dev")
|
||
self.kill9("dev")
|
||
thief = subprocess.Popen(
|
||
["python3", "-m", "http.server", "18808", "--bind", "127.0.0.1"],
|
||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||
self.addCleanup(thief.terminate)
|
||
# 等 thief 真正完成 bind 再 recover——慢环境里 Popen 返回时端口尚未监听，
|
||
# recover 会抢先起服务，把“端口被抢”测成竞态（CI 首轮实测）。
|
||
deadline = time.monotonic() + 5.0
|
||
bound = False
|
||
while time.monotonic() < deadline:
|
||
s = socket.socket()
|
||
s.settimeout(0.2)
|
||
try:
|
||
s.connect(("127.0.0.1", 18808))
|
||
bound = True
|
||
except OSError:
|
||
pass
|
||
finally:
|
||
s.close()
|
||
if bound:
|
||
break
|
||
time.sleep(0.05)
|
||
self.assertTrue(bound, "thief 未能在 5s 内监听 18808")
|
||
try:
|
||
result = self.svc("recover", check=False)
|
||
self.assertEqual(result.returncode, 1, result.stdout)
|
||
fail_events = self.history_events("dev", "recover")
|
||
self.assertTrue(any('"result":"fail"' in l for l in fail_events), fail_events)
|
||
self.assertFalse(self.pid_alive(self.pid_of("dev"))) # 未强抢端口
|
||
finally:
|
||
thief.terminate()
|
||
thief.wait(timeout=10)
|
||
|
||
# ---- 5. watchdog：单元文件生成（--print 走 main 分发全链路）----
|
||
|
||
def test_watchdog_print_plan(self):
|
||
result = self.svc("watchdog", "install", "--print")
|
||
self.assertIn(
|
||
"ExecStart=%s/scripts/svc.sh recover --quiet" % self.root, result.stdout)
|
||
self.assertIn("OnUnitActiveSec=60s", result.stdout)
|
||
self.assertIn("OnBootSec=1min", result.stdout)
|
||
self.assertIn("WantedBy=timers.target", result.stdout)
|
||
# oneshot 结束不得连带杀掉刚拉起的服务（cgroup 陷阱）。
|
||
self.assertIn("KillMode=process", result.stdout)
|
||
# 只打印，不落盘（受控 XDG_CONFIG_HOME 下无单元文件）。
|
||
unit_dir = self.root / "xdg" / "systemd" / "user"
|
||
self.assertFalse((unit_dir / "agent-system-svc-recover.timer").exists())
|
||
self.assertFalse((unit_dir / "agent-system-svc-recover.service").exists())
|
||
|
||
def test_recover_reserved_name_rejected(self):
|
||
self.seed_agent("recover", 18809) # 保留字不可作服务名
|
||
result = self.svc("recover", "recover", check=False)
|
||
self.assertEqual(result.returncode, 2, result.stdout)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|