From 57f8852634f6b2999712e4d031221d16cdfb35cd Mon Sep 17 00:00:00 2001 From: host Date: Tue, 27 Oct 2026 11:23:49 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20worker=20=E4=BB=A3=E7=A0=81=E5=8F=98?= =?UTF-8?q?=E5=8C=96=E8=87=AA=E9=80=80=20+=20=E9=83=A8=E7=BD=B2=E6=97=B6?= =?UTF-8?q?=E6=B8=85=E5=AD=A4=E5=84=BF=20worker?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 2026-10-27 暂存清理实弹：05:51 的旧 docker worker 成孤儿后 继续用旧代码抢任务、按老路失败且不删暂存；autoscaler 用 start_new_session=True 拉 worker、部署重启杀不到非 unit 进程。 两层修： 1. worker 常驻循环每轮边界比对 worker.py 文件签名 （mtime_ns + size），部署换目录后必变，变了就以 75 （EX_TEMPFAIL，失败类）退出——检查点在轮次边界，绝不 中断在跑任务。退出后：unit 托管的由 systemd on-failure 重拉新版，autoscaler 托管的由 reconcile 重拉，孤儿直接 消失。附带好处：以后部署不用再手动重启本地 worker。 2. host_deploy.sh 换目录后、启动服务前清孤儿：按 cgroup 过滤杀非 .service 的 cloud.worker 进程（桥接修补前就存 在的孤儿），先 TERM（跑完当前任务再退），30 秒未退的 再 KILL；unit 托管的（如 host-1）不动。pgrep 用 [c]loud.worker 写法，避免检查命令自匹配。 测试：test_worker.py 新增 2 个（签名变了则 75 退出且不认领、 签名不变则跑到 max_seconds）；反向验证：撤掉 worker 修复 时正好前者变红。清扫过滤器主机干跑：两个现役 worker 均 为 unit cgroup，全 KEEP，杀 0 个。全量 deploy 19、host 62、cloud 619 全过。 --- cloud/tests/test_worker.py | 18 ++++++++++++++++++ cloud/worker.py | 33 +++++++++++++++++++++++++++++++++ deploy/host_deploy.sh | 35 +++++++++++++++++++++++++++++++++++ 3 files changed, 86 insertions(+) diff --git a/cloud/tests/test_worker.py b/cloud/tests/test_worker.py index caa0b81..7c5f50b 100644 --- a/cloud/tests/test_worker.py +++ b/cloud/tests/test_worker.py @@ -539,6 +539,24 @@ class WorkerTestCase(unittest.TestCase): self.assertEqual(result["status"], "failed") self.assertFalse(os.path.exists(bundle_dir)) + # -- 代码变化自退 ------------------------------------------------------ + # 2026-10-27 实弹：05:51 的旧 docker worker 成孤儿后继续用旧 + # 代码抢任务。常驻循环每轮边界比对 worker.py 签名，变了就退 + # 出（75，失败类，supervisor 重拉新版），绝不中断在跑任务。 + def test_code_changed_exits_loop_before_claiming(self): + worker = self._worker( + runner=self._recording_runner(_completed_runner)) + worker._code_signature = ("stale-signature",) + with self.assertRaises(SystemExit) as ctx: + worker.run_loop(interval=10, max_seconds=30) + self.assertEqual(ctx.exception.code, 75) + self.assertEqual(self.calls, []) # 自退在认领之前 + + def test_code_unchanged_loop_runs_to_max_seconds(self): + worker = self._worker() + results = worker.run_loop(interval=10, max_seconds=0.4) + self.assertEqual(results, []) + def test_custom_env_key_name_from_arg_and_config(self): # 参数指定的名字出现在报错里（只含名字，不含值）。 _insert_task(self.db_path, "A") diff --git a/cloud/worker.py b/cloud/worker.py index dcd54eb..6da3021 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -160,6 +160,16 @@ DEFAULT_MODEL = "gpt-5.6-terra" _ALREADY_EXISTS_MARKER = "已存在" +def _code_signature(): + """本文件签名（mtime_ns, size）：部署 staging 换目录后必变； + 取不到时返回 None（此时自退检查跳过，绝不误杀）。""" + try: + st = os.stat(os.path.abspath(__file__)) + return (st.st_mtime_ns, st.st_size) + except OSError: + return None + + def _missing_key_error(env_key): """缺 key 的标准错误：只含 env_key 名，与 relay 侧的报错一致。""" return "Missing environment variable: %s" % env_key @@ -403,6 +413,16 @@ class Worker: self.capacity = int(capacity) except (TypeError, ValueError): self.capacity = 1 + # 常驻启动时的代码签名：run_loop 每轮边界比对，变了就自退 + # （见 run_loop 的代码变化自退检查）。 + self._code_signature = _code_signature() + + def _code_changed(self): + """启动后 worker.py 是否被换过（部署换目录/手动改代码）： + 签名取不到时一律当没变，不误杀。""" + now = _code_signature() + return (now is not None and self._code_signature is not None + and now != self._code_signature) def register(self): """向控制面注册本 worker（写 workers 表，失败只记日志）。 @@ -2651,6 +2671,19 @@ class Worker: if max_seconds is not None and \ time.time() - started >= max_seconds: break + # 代码变化自退：部署换目录后，旧代码 worker 若还抢 + # 任务会污染生产（2026-10-27 实弹：05:51 的旧 + # docker worker 成孤儿后继续用旧代码抢任务、按老路 + # 失败且不删暂存）。检查点在轮次边界，绝不中断在跑 + # 任务；退出码 75（EX_TEMPFAIL，失败类）：unit 托管 + # 的由 systemd on-failure 重拉新版，autoscaler 托管 + # 的由 reconcile 重拉，孤儿直接消失。单次 run_all + # 不走这里（进程本来就退出）。 + if self._code_changed(): + log_event(self.logger, "worker_code_changed_exit", + message="worker code changed, exiting " + "for restart") + raise SystemExit(75) try: results = self.run_all() except Exception as exc: diff --git a/deploy/host_deploy.sh b/deploy/host_deploy.sh index 8264dc2..1c905ec 100644 --- a/deploy/host_deploy.sh +++ b/deploy/host_deploy.sh @@ -185,6 +185,41 @@ if id codexsvc >/dev/null 2>&1; then chown -R codexsvc:codexsvc "$APP_DIR" fi +# --- 5b. 清孤儿 worker --------------------------------------------------- +# 非 systemd unit cgroup 的 cloud.worker 进程（手动/setsid 拉起的旧 +# worker）：部署重启杀不到它（2026-10-27 实弹：05:51 的旧 docker +# worker 成孤儿后继续用旧代码抢任务、按老路失败且不删暂存）。 +# unit 托管的 worker（cgroup 含 .service，如 codex-host-worker） +# 不动：它们靠新版 worker 的代码变化自退 + systemd on-failure 重拉。 +# 先 TERM（worker 收到后停认领、跑完当前任务再退），30 秒内未退 +# 的再 KILL；被中断的在跑任务由陈旧回收标 interrupted，可续跑。 +for _pid in $(pgrep -f "[c]loud.worker" 2>/dev/null || true); do + _cg=$(cat /proc/$_pid/cgroup 2>/dev/null || true) + case "$_cg" in + *".service"*) ;; + *) echo "清孤儿 worker：pid=$_pid"; kill $_pid 2>/dev/null || true ;; + esac +done +for _i in $(seq 1 30); do + _left=0 + for _pid in $(pgrep -f "[c]loud.worker" 2>/dev/null || true); do + _cg=$(cat /proc/$_pid/cgroup 2>/dev/null || true) + case "$_cg" in + *".service"*) ;; + *) _left=1 ;; + esac + done + [ "$_left" = "0" ] && break + sleep 1 +done +for _pid in $(pgrep -f "[c]loud.worker" 2>/dev/null || true); do + _cg=$(cat /proc/$_pid/cgroup 2>/dev/null || true) + case "$_cg" in + *".service"*) ;; + *) echo "孤儿 worker 未退，KILL：pid=$_pid"; kill -9 $_pid 2>/dev/null || true ;; + esac +done + rollback() { echo "新代码 health 未通过，开始回滚到备份：$BACKUP_FILE" >&2 systemctl stop "$WORKER_UNIT" $SERVER_UNITS || true