host
06710cf47b
- Docker /tmp tmpfs 移到全部 -v 之前（父挂载在前，seed 子挂载在后）； 宿主级 bwrap /tmp 仍宣告未隔离，口径见 tmp_isolation_status() - workers 表补 host/pid/address（老库幂等补列），心跳透传与保留语义， _claim 写 claimed_by_worker；orphaned_containers 精确识别失联容器， Worker.cleanup_orphaned_containers 默认只杀 pid_verified 的精确容器名 - docker_run_timeout() 纯函数抽取（行为与原内联一致），README 补边界与 未验证清单；新增 test_sandbox_scale.py 20 项
436 lines
18 KiB
Python
436 lines
18 KiB
Python
"""沙箱规模单测：/tmp 隔离顺序、注册表寻址、失联识别、超时边界。
|
||
|
||
运行（在 ~/workspace/codex-app-server-sandboxscale 下）：
|
||
python3 -m unittest discover -s cloud/tests -p "test_sandbox_scale.py" -v
|
||
|
||
说明：全部用注入的假 runner / 假 pid 检查，不调用真实 docker、
|
||
不跑真实模型回合、不联网；env 映射里的值只是占位假名，
|
||
绝不是真实密钥。任务 id 用 32 位 hex（容器名正则的精确
|
||
形状），不用短 id——短 id 的容器名本来就不在失联识别范围内。
|
||
|
||
覆盖（与本轮改动一一对应）：
|
||
- Docker /tmp tmpfs 在所有 -v 挂载之前（父挂载在前，seed 子挂载
|
||
在后，不被遮掉）；host_bwrap 仍宣告未隔离（tmp_isolation_status）。
|
||
- workers 注册/心跳：host/pid/address 写进库、心跳未给保留旧值、
|
||
老库（无三列）迁移补列；stale_workers 只列心跳过期行。
|
||
- 失联容器识别：只认精确 codex-<32 位 hex> + 认领到失联 worker
|
||
+ running/interrupted；基础设施容器、非任务名、已完成任务、
|
||
无认领、owner 存活的一律排除；pid_verified 只在“本机 host +
|
||
已知 pid + pid 明确已死”时为真。
|
||
- Worker 精确清理：verified_only 默认只杀已验证的精确容器名，
|
||
未验证的只列不杀；显式 verified_only=False 才杀未验证的。
|
||
- 认领写 claimed_by_worker（CAS 语义不变：二次认领仍失败）。
|
||
- 超时边界纯函数 docker_run_timeout：取较小值、None 语义、
|
||
不限时语义、非法值回落（行为与原内联一致）。
|
||
- web 自适应基线：console.html 含 viewport meta 与小屏 media
|
||
查询（跨视口手工验证矩阵见 cloud/README.md 新增节，不宣称
|
||
本文件已做浏览器实测）。
|
||
"""
|
||
|
||
import http.client
|
||
import json
|
||
import os
|
||
import socket
|
||
import sqlite3
|
||
import sys
|
||
import tempfile
|
||
import threading
|
||
import time
|
||
import unittest
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
from unittest import mock
|
||
|
||
try:
|
||
from cloud.auth import AuthStore
|
||
from cloud.sandbox import DockerExecutor, tmp_isolation_status
|
||
from cloud.server import create_server, init_db, sweep_stale_tasks
|
||
from cloud.worker import Worker, docker_run_timeout
|
||
from cloud.workers import (default_pid_alive, heartbeat_worker,
|
||
list_workers, orphaned_containers,
|
||
register_worker, stale_workers)
|
||
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||
from cloud.auth import AuthStore
|
||
from cloud.sandbox import DockerExecutor, tmp_isolation_status
|
||
from cloud.server import create_server, init_db, sweep_stale_tasks
|
||
from cloud.worker import Worker, docker_run_timeout
|
||
from cloud.workers import (default_pid_alive, heartbeat_worker,
|
||
list_workers, orphaned_containers,
|
||
register_worker, stale_workers)
|
||
|
||
DEAD_TASK = "aa" * 16
|
||
LIVE_TASK = "bb" * 16
|
||
DONE_TASK = "cc" * 16
|
||
UNCLAIMED_TASK = "dd" * 16
|
||
MISSING_TASK = "ee" * 16
|
||
|
||
|
||
def _insert_task(db_path, task_id, status="queued", claimed=None,
|
||
heartbeat_ago=None):
|
||
now = time.time()
|
||
heartbeat = None
|
||
if heartbeat_ago is not None:
|
||
heartbeat = now - heartbeat_ago
|
||
elif status == "running":
|
||
heartbeat = now
|
||
conn = sqlite3.connect(db_path)
|
||
try:
|
||
conn.execute(
|
||
"INSERT INTO tasks(id, project_id, prompt, status,"
|
||
" idempotency_key, depends_on, priority, created_by,"
|
||
" created_at, updated_at, last_heartbeat, claimed_by_worker)"
|
||
" VALUES(?, 'p1', 'do work', ?, NULL, '[]', 0, 'tester',"
|
||
" ?, ?, ?, ?)",
|
||
(task_id, status, now, now, heartbeat, claimed),
|
||
)
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _set_worker_old(db_path, worker_id, seconds_ago):
|
||
conn = sqlite3.connect(db_path)
|
||
try:
|
||
conn.execute(
|
||
"UPDATE workers SET last_seen = ? WHERE worker_id = ?",
|
||
(time.time() - seconds_ago, worker_id))
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _seed_orphan_db(db_path, local_host):
|
||
"""失联场景：dead worker（本机 host、pid 已死）与 live worker
|
||
各认领一个 running 任务，另有已完成、无认领任务各一。"""
|
||
init_db(db_path)
|
||
register_worker(db_path, "dead-worker", executor="docker",
|
||
host=local_host, pid=999999, address="h1:9001")
|
||
register_worker(db_path, "live-worker", executor="docker",
|
||
host=local_host, pid=123456, address="h1:9002")
|
||
_set_worker_old(db_path, "dead-worker", 10000)
|
||
_insert_task(db_path, DEAD_TASK, status="running",
|
||
claimed="dead-worker", heartbeat_ago=10000)
|
||
_insert_task(db_path, LIVE_TASK, status="running",
|
||
claimed="live-worker")
|
||
_insert_task(db_path, DONE_TASK, status="completed",
|
||
claimed="dead-worker")
|
||
_insert_task(db_path, UNCLAIMED_TASK, status="running",
|
||
claimed=None)
|
||
|
||
|
||
class TmpIsolationTestCase(unittest.TestCase):
|
||
def test_tmpfs_before_all_volume_mounts(self):
|
||
seed_dir = tempfile.mkdtemp()
|
||
try:
|
||
with mock.patch("cloud.sandbox.shutil.which",
|
||
return_value="/usr/bin/docker"):
|
||
ex = DockerExecutor(
|
||
"/fake/host.py",
|
||
runner=lambda cmd, timeout=None: SimpleNamespace(
|
||
returncode=0, stdout="{}", stderr=""),
|
||
env={"XCM_API_KEY": "placeholder-not-a-real-key"})
|
||
cmd = ex.build_run_command(
|
||
{"id": "t1", "worktree": "/tmp/wt"},
|
||
["echo", "hi"],
|
||
extra_mounts=["%s:/tmp/codex-seed:ro" % seed_dir])
|
||
tmpfs_idx = cmd.index("--tmpfs")
|
||
self.assertEqual(cmd[tmpfs_idx + 1], "/tmp")
|
||
seed_idx = cmd.index("%s:/tmp/codex-seed:ro" % seed_dir)
|
||
self.assertLess(tmpfs_idx, seed_idx)
|
||
self.assertIn("/var/tmp", cmd)
|
||
finally:
|
||
os.rmdir(seed_dir)
|
||
|
||
def test_host_bwrap_still_declared_not_isolated(self):
|
||
status = tmp_isolation_status()
|
||
self.assertTrue(status["docker"])
|
||
self.assertFalse(status["host_bwrap"])
|
||
|
||
|
||
class RegistryAddressTestCase(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 test_register_and_heartbeat_address_fields(self):
|
||
row = register_worker(self.db_path, "w1", executor="docker",
|
||
host="host-a", pid=111, address="h:1")
|
||
self.assertEqual(row["host"], "host-a")
|
||
self.assertEqual(row["pid"], 111)
|
||
self.assertEqual(row["address"], "h:1")
|
||
# 心跳未给 host/pid/address 时保留旧值，不洗成 None。
|
||
row = heartbeat_worker(self.db_path, "w1", executor="docker")
|
||
self.assertEqual(row["host"], "host-a")
|
||
self.assertEqual(row["pid"], 111)
|
||
self.assertEqual(row["address"], "h:1")
|
||
# 心跳显式给新 pid 时刷新（重启后同 worker_id 复用不残留旧 pid）。
|
||
row = heartbeat_worker(self.db_path, "w1", pid=222)
|
||
self.assertEqual(row["pid"], 222)
|
||
self.assertEqual(row["host"], "host-a")
|
||
|
||
def test_old_workers_table_migrates_address_columns(self):
|
||
conn = sqlite3.connect(self.db_path)
|
||
try:
|
||
conn.execute(
|
||
"CREATE TABLE workers(worker_id TEXT PRIMARY KEY,"
|
||
" executor TEXT, capacity INTEGER,"
|
||
" last_seen DOUBLE PRECISION,"
|
||
" created_at DOUBLE PRECISION)")
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
row = register_worker(self.db_path, "w-old", host="host-b",
|
||
pid=333)
|
||
self.assertEqual(row["host"], "host-b")
|
||
self.assertEqual(row["pid"], 333)
|
||
self.assertIsNone(row["address"])
|
||
|
||
def test_stale_workers_only_expired(self):
|
||
register_worker(self.db_path, "fresh", host="h")
|
||
register_worker(self.db_path, "old", host="h")
|
||
_set_worker_old(self.db_path, "old", 10000)
|
||
stale = stale_workers(self.db_path, alive_seconds=120)
|
||
self.assertEqual([w["worker_id"] for w in stale], ["old"])
|
||
self.assertFalse(stale[0]["alive"])
|
||
live = list_workers(self.db_path, alive_seconds=120)
|
||
self.assertTrue([w for w in live
|
||
if w["worker_id"] == "fresh"][0]["alive"])
|
||
|
||
def test_default_pid_alive_current_process(self):
|
||
self.assertTrue(default_pid_alive(os.getpid()))
|
||
self.assertIsNone(default_pid_alive("not-a-pid"))
|
||
self.assertIsNone(default_pid_alive(None))
|
||
|
||
|
||
class OrphanIdentifyTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.db_path = os.path.join(self.tmp.name, "control.db")
|
||
self.host = socket.gethostname()
|
||
_seed_orphan_db(self.db_path, self.host)
|
||
|
||
def tearDown(self):
|
||
self.tmp.cleanup()
|
||
|
||
def _names(self):
|
||
return ["codex-" + DEAD_TASK, "codex-" + LIVE_TASK,
|
||
"codex-" + DONE_TASK, "codex-" + UNCLAIMED_TASK,
|
||
"codex-" + MISSING_TASK, "codex-postgres",
|
||
"codex-not-a-task", "other-" + DEAD_TASK]
|
||
|
||
def test_only_dead_owner_exact_container_verified(self):
|
||
found = orphaned_containers(
|
||
self.db_path, self._names(), alive_seconds=120,
|
||
pid_alive=lambda pid: False)
|
||
self.assertEqual(len(found), 1)
|
||
self.assertEqual(found[0]["container"], "codex-" + DEAD_TASK)
|
||
self.assertEqual(found[0]["task_id"], DEAD_TASK)
|
||
self.assertEqual(found[0]["worker_id"], "dead-worker")
|
||
self.assertTrue(found[0]["pid_verified"])
|
||
|
||
def test_live_pid_not_verified_but_still_listed(self):
|
||
found = orphaned_containers(
|
||
self.db_path, self._names(), alive_seconds=120,
|
||
pid_alive=lambda pid: True)
|
||
self.assertEqual(len(found), 1)
|
||
self.assertFalse(found[0]["pid_verified"])
|
||
|
||
def test_remote_host_never_pid_verified(self):
|
||
conn = sqlite3.connect(self.db_path)
|
||
try:
|
||
conn.execute(
|
||
"UPDATE workers SET host = ? WHERE worker_id = ?",
|
||
("some-other-host", "dead-worker"))
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
found = orphaned_containers(
|
||
self.db_path, self._names(), alive_seconds=120,
|
||
pid_alive=lambda pid: False)
|
||
self.assertEqual(len(found), 1)
|
||
self.assertFalse(found[0]["pid_verified"])
|
||
|
||
def test_sweep_marks_dead_owner_task_interrupted(self):
|
||
swept = sweep_stale_tasks(self.db_path, threshold_seconds=300)
|
||
self.assertIn(DEAD_TASK, swept)
|
||
self.assertNotIn(LIVE_TASK, swept)
|
||
# sweep 后失联容器仍可识别（interrupted 状态也算失联）。
|
||
found = orphaned_containers(
|
||
self.db_path, self._names(), alive_seconds=120,
|
||
pid_alive=lambda pid: False)
|
||
self.assertEqual(len(found), 1)
|
||
|
||
|
||
class WorkerOrphanCleanupTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.db_path = os.path.join(self.tmp.name, "control.db")
|
||
self.host = socket.gethostname()
|
||
_seed_orphan_db(self.db_path, self.host)
|
||
self.killed = []
|
||
|
||
def fake(cmd, timeout=None):
|
||
if len(cmd) >= 2 and cmd[1] == "ps":
|
||
return SimpleNamespace(
|
||
returncode=0,
|
||
stdout="\n".join(
|
||
["codex-" + DEAD_TASK, "codex-" + LIVE_TASK,
|
||
"codex-postgres"]) + "\n",
|
||
stderr="")
|
||
if len(cmd) >= 2 and cmd[1] in ("kill", "rm"):
|
||
self.killed.append(cmd[-1])
|
||
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
||
|
||
self.ex = DockerExecutor(
|
||
"/fake/host.py", runner=fake,
|
||
env={"XCM_API_KEY": "placeholder-not-a-real-key"})
|
||
self.worker = Worker(
|
||
self.db_path, "/fake/host.py",
|
||
env={"XCM_API_KEY": "placeholder-not-a-real-key"},
|
||
executor=self.ex, worker_id="cleaner",
|
||
worker_address="cleaner-host:9009")
|
||
|
||
def tearDown(self):
|
||
self.tmp.cleanup()
|
||
|
||
def test_verified_only_kills_exact_dead_container(self):
|
||
cleaned = self.worker.cleanup_orphaned_containers(
|
||
pid_alive=lambda pid: False, alive_seconds=120)
|
||
self.assertEqual(cleaned, ["codex-" + DEAD_TASK])
|
||
self.assertIn("codex-" + DEAD_TASK, self.killed)
|
||
self.assertNotIn("codex-" + LIVE_TASK, self.killed)
|
||
self.assertNotIn("codex-postgres", self.killed)
|
||
|
||
def test_unverified_skipped_by_default(self):
|
||
cleaned = self.worker.cleanup_orphaned_containers(
|
||
pid_alive=lambda pid: True, alive_seconds=120)
|
||
self.assertEqual(cleaned, [])
|
||
self.assertEqual(self.killed, [])
|
||
|
||
def test_explicit_unverified_opt_in_kills(self):
|
||
cleaned = self.worker.cleanup_orphaned_containers(
|
||
verified_only=False, pid_alive=lambda pid: True,
|
||
alive_seconds=120)
|
||
self.assertEqual(cleaned, ["codex-" + DEAD_TASK])
|
||
|
||
def test_claim_writes_claimed_by_worker(self):
|
||
_insert_task(self.db_path, "f1" * 16, status="queued")
|
||
worker = Worker(
|
||
self.db_path, "/fake/host.py",
|
||
env={"XCM_API_KEY": "placeholder-not-a-real-key"},
|
||
worker_id="claimer-1")
|
||
self.assertTrue(worker._claim("f1" * 16))
|
||
task = worker.get_task("f1" * 16)
|
||
self.assertEqual(task["claimed_by_worker"], "claimer-1")
|
||
# CAS 语义不变：已 running 的任务二次认领仍失败。
|
||
self.assertFalse(worker._claim("f1" * 16))
|
||
|
||
|
||
class TimeoutBoundaryTestCase(unittest.TestCase):
|
||
def test_min_wins_and_none_semantics(self):
|
||
self.assertEqual(docker_run_timeout(1800, None), 1800)
|
||
self.assertEqual(docker_run_timeout(1800, 600), 600)
|
||
self.assertEqual(docker_run_timeout(600, 1800), 600)
|
||
self.assertIsNone(docker_run_timeout(None, 600))
|
||
self.assertIsNone(docker_run_timeout(None, None))
|
||
|
||
def test_no_time_limit_overrides_both(self):
|
||
self.assertIsNone(docker_run_timeout(1800, 600,
|
||
no_time_limit=True))
|
||
|
||
def test_invalid_value_falls_back_to_task_timeout(self):
|
||
self.assertEqual(docker_run_timeout(1800, "bad"), 1800)
|
||
|
||
|
||
class ViewportBaselineTestCase(unittest.TestCase):
|
||
def test_console_has_viewport_and_small_screen_query(self):
|
||
console = (Path(__file__).resolve().parents[2]
|
||
/ "cloud" / "console.html")
|
||
text = console.read_text(encoding="utf-8")
|
||
self.assertIn('name="viewport"', text)
|
||
self.assertIn("@media(max-width:", text)
|
||
|
||
|
||
class HeartbeatHttpTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.db_path = os.path.join(self.tmp.name, "control.db")
|
||
self.auth = AuthStore(self.db_path)
|
||
self.owner_id, self.owner_key = self.auth.create_user("owner")
|
||
self._old_admins = os.environ.get("CODEX_GLOBAL_ADMINS")
|
||
os.environ["CODEX_GLOBAL_ADMINS"] = self.owner_id
|
||
self.server = create_server(self.db_path, "127.0.0.1", 0)
|
||
self.port = self.server.server_address[1]
|
||
self.thread = threading.Thread(
|
||
target=self.server.serve_forever, daemon=True)
|
||
self.thread.start()
|
||
|
||
def tearDown(self):
|
||
self.server.shutdown()
|
||
self.server.server_close()
|
||
self.thread.join(timeout=5)
|
||
try:
|
||
self.auth.conn.close()
|
||
except Exception:
|
||
pass
|
||
self.tmp.cleanup()
|
||
if self._old_admins is None:
|
||
os.environ.pop("CODEX_GLOBAL_ADMINS", None)
|
||
else:
|
||
os.environ["CODEX_GLOBAL_ADMINS"] = self._old_admins
|
||
|
||
def _req(self, method, path, body=None, key=None):
|
||
conn = http.client.HTTPConnection("127.0.0.1", self.port,
|
||
timeout=10)
|
||
hdrs = {}
|
||
payload = None
|
||
if body is not None:
|
||
payload = json.dumps(body).encode("utf-8")
|
||
hdrs["Content-Type"] = "application/json"
|
||
if key is not None:
|
||
hdrs["X-API-Key"] = key
|
||
try:
|
||
conn.request(method, path, body=payload, headers=hdrs)
|
||
resp = conn.getresponse()
|
||
data = resp.read()
|
||
finally:
|
||
conn.close()
|
||
try:
|
||
parsed = json.loads(data.decode("utf-8")) if data else None
|
||
except ValueError:
|
||
parsed = None
|
||
return resp.status, parsed
|
||
|
||
def test_heartbeat_address_fields_roundtrip(self):
|
||
status, body = self._req(
|
||
"POST", "/workers/heartbeat",
|
||
{"worker_id": "hw1", "host": "host-x", "pid": 4242,
|
||
"address": "host-x:9001"},
|
||
self.owner_key)
|
||
self.assertEqual(status, 200, body)
|
||
self.assertEqual(body["host"], "host-x")
|
||
self.assertEqual(body["pid"], 4242)
|
||
self.assertEqual(body["address"], "host-x:9001")
|
||
status, body = self._req("GET", "/workers",
|
||
key=self.owner_key)
|
||
self.assertEqual(status, 200, body)
|
||
row = [w for w in body["workers"]
|
||
if w["worker_id"] == "hw1"][0]
|
||
self.assertEqual(row["host"], "host-x")
|
||
self.assertEqual(row["pid"], 4242)
|
||
|
||
def test_heartbeat_invalid_pid_400(self):
|
||
status, _ = self._req(
|
||
"POST", "/workers/heartbeat",
|
||
{"worker_id": "hw2", "pid": "not-an-int"},
|
||
self.owner_key)
|
||
self.assertEqual(status, 400)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|