feat: 沙箱规模——/tmp 顺序加严、注册表寻址与失联精确清理、超时边界抽取

- 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 项
This commit is contained in:
host committed 2026-11-04 20:54:33 +00:00
1 parent e079b9b44a
commit b42b650fb7
7 files changed
+897 -48

No files matched your search

+138 -22
View File
@@ -223,7 +223,8 @@ from cloud.sandbox import CloudExecutor, DockerExecutor, LocalExecutor
from cloud.uploads import upload_root as _upload_root
from cloud.sandbox import _default_runner as _sandbox_default_runner
from cloud.caps import effective_capacity, resolve_capacity
from cloud.workers import heartbeat_worker, register_worker
from cloud.workers import (heartbeat_worker, orphaned_containers,
register_worker)
# 与 server.TERMINAL_STATUSES 对齐:这些状态的任务不再被认领或取消。
TERMINAL_STATUSES = frozenset(
@@ -270,6 +271,39 @@ def _uuid_hex():
return uuid.uuid4().hex
def docker_run_timeout(task_timeout, docker_timeout,
no_time_limit=False):
"""本次 docker run 超时(纯函数,可单测,长任务天花板边界)。
规则(与 _run_attempt 原内联逻辑完全一致,本轮只抽取,
不改行为):
- no_time_limit=1(任务不限时)时一律 None:不限的是任务
总墙钟与总轮次,不是单轮无限——host 单轮 turn_timeout
仍是护栏,不在本函数里。
- task_timeout 为 None 时 None(Docker 路径此时也不再用
executor.timeout 兜底?注意:原内联只在 task_timeout
非 None 时算 min,task_timeout 为 None 时 effective 恒
为 None——本函数忠实保留该语义,绝不悄悄把 docker_timeout
扶正为有效超时)。
- 否则取 min(task_timeout, docker_timeout)(docker_timeout
为 None 时取 task_timeout);任一值非法时回落 task_timeout。
取消与超时的状态语义不在此:超时 -> failed(不是取消),
取消 -> interrupted(执行结果绝不覆盖它),见 _run_attempt
与 cancel();恢复靠断点续跑(goal-resume / run 续跑),
不是靠把超时任务悄悄重跑一遍。
"""
if no_time_limit:
return None
if task_timeout is None:
return None
if docker_timeout is None:
return task_timeout
try:
return min(float(task_timeout), float(docker_timeout))
except (TypeError, ValueError):
return task_timeout
def _worker_host_name(executor, task):
"""host 任务名:优先 executor._host_name(与 sandbox 同款
派生),没有 executor 时回落任务 id(uuid hex 本就是安全名)。"""
@@ -462,6 +496,9 @@ def _task_from_row(row):
"interrupted_reason": _col("interrupted_reason"),
"auto_resume_count": _col("auto_resume_count") or 0,
"no_time_limit": _col("no_time_limit") or 0,
# 认领 owner:老行缺列时 None(= 认领前或老任务,无 owner
# 可查,失联识别宁可漏不猜)。
"claimed_by_worker": _col("claimed_by_worker"),
# 协调员:老行缺列时 coordinator_parts 为 [](= 普通任务),
# parent_task_id/coordinator_part 缺列时为 None。
"coordinator_parts": _coord_parse_parts(_col("coordinator_parts")),
@@ -577,7 +614,8 @@ class Worker:
outbox_max_attempts=3,
quota_poll_interval_seconds=30,
auto_resume_max=50,
ignore_host_budget=True):
ignore_host_budget=True,
worker_address=None):
self.db_path = str(db_path)
self.host_py_path = host_py_path
self.config_path = config_path
@@ -822,6 +860,11 @@ class Worker:
# 自然变 stale,不要用随机 uuid 让注册表无法对应同一台机器。
self.worker_id = worker_id or "%s-%d" % (
socket.gethostname(), os.getpid())
# 寻址串:调用方显式给才有(如运维约定的 host:port),
# 默认 None(不猜、不从环境变量里拼密钥类地址);注册/
# 心跳时原样写进 workers.address,供失联排查时定位
# owner 在哪台主机、哪个入口,绝不是访问凭证。
self.worker_address = worker_address
# 并发上限配置来源(见 cloud/caps.py):显式构造参数 /
# CLI --worker-capacity > 环境变量 CLOUD_WORKER_CAPACITY >
# 默认 1;非法值归一为 1。self._env 在上面已定好,单测
@@ -884,7 +927,8 @@ class Worker:
self.db_path, self.worker_id,
executor=self._executor_name(), capacity=self.capacity,
running=self._running_count(),
host=self.host_name, pid=os.getpid())
host=self.host_name, pid=os.getpid(),
address=self.worker_address)
except Exception as exc:
log_event(self.logger, "worker_register_failed",
message="worker register failed: %s" % exc)
@@ -896,13 +940,17 @@ class Worker:
时退化为 socket.gethostname(),容器内即容器名,与
worktree_host 同款 caveat),pid 用本进程号;两者每轮
刷新,供失联判定做本机 + pid 已死的 pid_verified 校验。
"""
每次都显式传本机 host 与本进程 pid:worker 重启后
同 worker_id 复用时,旧 pid 不会残留在注册表里误导
失联判断(旧 pid 残留 + 心跳新鲜 = 活着,尚可;旧 pid
残留 + 心跳过期 = pid 校验会误判——每次刷新杜绝此例)。 """
try:
heartbeat_worker(
self.db_path, self.worker_id,
executor=self._executor_name(), capacity=self.capacity,
running=self._running_count(),
host=self.host_name, pid=os.getpid())
host=self.host_name, pid=os.getpid(),
address=self.worker_address)
except Exception as exc:
log_event(self.logger, "worker_registry_heartbeat_failed",
message="worker registry heartbeat failed: %s" % exc)
@@ -1044,31 +1092,37 @@ class Worker:
conn = _db_connect(self.db_path)
try:
self._ensure_result_columns(conn)
# 认领同时写 claimed_by_worker:失联容器识别的唯一
# owner 依据(见 workers.orphaned_containers)。CAS
# 语义不变(仍只在 queued 时改行),认领失败时行不
# 动、旧认领记录也不被覆盖。
if executor is None:
cur = conn.execute(
"UPDATE tasks SET status = 'running',"
" last_heartbeat = ?, updated_at = ?,"
" worktree_host = ?"
" worktree_host = ?, claimed_by_worker = ?"
" WHERE id = ? AND status = 'queued'",
(now, now, self.host_name, task_id),
(now, now, self.host_name, self.worker_id, task_id),
)
elif self.claim_executor_only:
cur = conn.execute(
"UPDATE tasks SET status = 'running',"
" last_heartbeat = ?, updated_at = ?,"
" worktree_host = ?"
" worktree_host = ?, claimed_by_worker = ?"
" WHERE id = ? AND status = 'queued'"
" AND executor = ?",
(now, now, self.host_name, task_id, executor),
(now, now, self.host_name, self.worker_id, task_id,
executor),
)
else:
cur = conn.execute(
"UPDATE tasks SET status = 'running',"
" last_heartbeat = ?, updated_at = ?,"
" worktree_host = ?"
" worktree_host = ?, claimed_by_worker = ?"
" WHERE id = ? AND status = 'queued'"
" AND (executor IS NULL OR executor = ?)",
(now, now, self.host_name, task_id, executor),
(now, now, self.host_name, self.worker_id, task_id,
executor),
)
conn.commit()
claimed = cur.rowcount == 1
@@ -1636,6 +1690,7 @@ class Worker:
"goal_max_rounds": "INTEGER",
"goal_mode": "INTEGER",
"worktree_host": "TEXT",
"claimed_by_worker": "TEXT",
"model": "TEXT",
"effort": "TEXT",
"child_model": "TEXT",
@@ -3278,12 +3333,71 @@ class Worker:
pass
return
def cleanup_orphaned_containers(self, verified_only=True,
pid_alive=None, alive_seconds=None):
"""精确失联清理:只杀 workers.orphaned_containers() 认定的
容器(按精确容器名逐个杀,绝不按前缀展开)。
verified_only=True(默认):只杀 pid_verified 的容器——
本机 host + 已知 pid 且 pid 明确已死,owner 死亡有本地
证据;远端 host / pid 未知 / 未验证的只列不杀(返回里
不含它们,日志里记 unverified 数量供运维人工处理)。
verified_only=False:把未验证的也杀掉——只应在“本机
单机部署、注册表 pid 列是老版本缺失”这类调用方已确认
单机语义的场景显式开,跨主机场景开它等于按心跳过期
猜死,绝不能默认开。
本方法只写识别与杀容器,不改任务状态:任务状态由
sweep_stale_tasks(心跳过期 -> interrupted)负责,两者
分工,互不越权。执行器无 list_containers/kill_container
时返回 [](Local/Cloud 执行器本来就没有容器可清)。
返回已清理的精确容器名列表。
"""
list_fn = getattr(self.executor, "list_containers", None)
kill_fn = getattr(self.executor, "kill_container", None)
if list_fn is None or kill_fn is None:
return []
try:
names = list_fn()
except Exception as exc:
log_event(self.logger, "worker_orphan_list_failed",
message="orphan container list failed: %s" % exc)
return []
try:
orphans = orphaned_containers(
self.db_path, names, alive_seconds=alive_seconds,
pid_alive=pid_alive)
except Exception as exc:
log_event(self.logger, "worker_orphan_identify_failed",
message="orphan identify failed: %s" % exc)
return []
cleaned = []
unverified = 0
for item in orphans:
if verified_only and not item.get("pid_verified"):
unverified += 1
continue
try:
if kill_fn(item["container"]):
cleaned.append(item["container"])
except Exception:
continue
if cleaned or unverified:
log_event(self.logger, "worker_orphan_cleanup",
message="cleaned orphan containers: %s;"
" unverified skipped: %d"
% (",".join(cleaned), unverified))
return cleaned
def cleanup_stale_containers(self):
"""启动清理兜底:清掉上一轮残留的 codex- 容器。
只在执行器有 cleanup_stale_containers 时做(Docker);
列出或删除失败只记日志,绝不抛异常炸掉 worker。
返回已清理的容器名列表(非 Docker 执行器返回 [])。
注意:这是旧的任务正则兜底(删所有 codex-<任务 id>
容器,不区分 owner),精确失联清理请用
cleanup_orphaned_containers();本方法保留是为兼容
认领记录缺失的老任务容器,不改其行为。
"""
cleanup = getattr(self.executor, "cleanup_stale_containers", None)
if cleanup is None:
@@ -5418,18 +5532,15 @@ class Worker:
# Docker 超时取较小值:两个超时都不应静默延长另一个。
# 注意绝不能改共享的 executor.timeout(并发下两个任务会
# 互相覆盖对方的超时);超时只作为本次调用的参数传下去。
effective_docker_timeout = None
# 规则抽到模块级 docker_run_timeout()(纯函数可单测),
# 这里只接线,行为与原内联完全一致。
is_docker = isinstance(self.executor, DockerExecutor)
if is_docker and _effective_task_timeout is not None:
if self.executor.timeout is None:
effective_docker_timeout = _effective_task_timeout
else:
try:
effective_docker_timeout = min(
float(_effective_task_timeout),
float(self.executor.timeout))
except (TypeError, ValueError):
effective_docker_timeout = _effective_task_timeout
effective_docker_timeout = (
docker_run_timeout(
_effective_task_timeout,
self.executor.timeout if is_docker else None,
no_time_limit=_no_time_limit)
if is_docker else None)
# P0 安全批后:Docker 同步路径起取消看门狗线程
# (阻塞在 docker run 里时,库状态变 interrupted 就
# 按容器名真杀;非 Docker 路径不启动,避免空线程)。
@@ -9488,6 +9599,10 @@ def main(argv=None):
parser.add_argument("--worker-id", default=None,
help="多 worker 池身份(默认:主机名-进程号;"
"生产部署应配稳定 id,否则重启后旧行变 stale)")
parser.add_argument("--worker-address", default=None,
help="worker 寻址串(可选,显式给才写进注册表"
" workers.address,如运维约定的 host:port;"
"不给则为 NULL,绝不猜、不含任何凭证)")
parser.add_argument("--worker-capacity", type=int, default=None,
help="worker 并发上限:同一 worker 内最多同时跑"
"几个任务(未给时读环境变量"
@@ -9621,6 +9736,7 @@ def main(argv=None):
repos_dir=args.repos_dir,
heartbeat_file=args.heartbeat_file,
worker_id=args.worker_id,
worker_address=args.worker_address,
capacity=args.worker_capacity,
host_name=args.host_name,
pr_jobs_only=args.pr_jobs_only,