diff --git a/cloud/autoscaler.py b/cloud/autoscaler.py index d81dec1..85ecc7a 100644 --- a/cloud/autoscaler.py +++ b/cloud/autoscaler.py @@ -62,6 +62,29 @@ from cloud.logging_config import get_logger, log_event, setup_logging from cloud.queue import ready_tasks from cloud.workers import DEFAULT_ALIVE_SECONDS as _WORKER_ALIVE_SECONDS +# CHV 混合模式每宿主预热数 W（DESIGN-HYBRID.md §1.2 已定值： +# 2 台/宿主；池目标数 P = W × 在线宿主数）。 +CHV_WARM_POOL_PER_HOST = 2 + + +def pool_target_for_hosts(online_hosts, + per_host=CHV_WARM_POOL_PER_HOST): + """池目标数 P = W × 在线宿主数（§1.2 口径）。 + + 在线宿主数/每宿主预热数为负或非整数时按 0 计， + 绝不算出负的池目标。""" + try: + hosts = int(online_hosts) + except (TypeError, ValueError): + return 0 + try: + per = int(per_host) + except (TypeError, ValueError): + return 0 + if hosts <= 0 or per <= 0: + return 0 + return hosts * per + def queue_depth(db_path, executor=None): """当前就绪任务数：依赖已满足且状态为 queued 的任务。 @@ -126,19 +149,35 @@ def queue_depth(db_path, executor=None): def desired_workers(depth, per_worker_capacity, - min_workers, max_workers): + min_workers, max_workers, pool_target=0): """按深度算目标 worker 数。 depth=0 时也要保留 min_workers（默认 1）：冷启动要快， 且 scaler 缩到 0 后再拉起仍有进程启动延迟；想缩到 0 可以把 min_workers 配成 0，scaler 仍在跑所以能再拉起。 + + CHV 混合模式（DESIGN-HYBRID.md §1.2，C 轨）：期望值 + 作用域扩展——队列深度算出的 fleet 需要量之外，再加 + 池目标数 pool_target（= W × 在线宿主数，W=2，见 + pool_target_for_hosts）。pool_target=0 时与旧公式 + 一字不差。池目标数计入 max_workers 上限：max_workers + 须配得下 min_workers + pool_target，否则期望值会被 + 上限截断、池子补不满（配错时 state 文件里的 + pool_target 与 desired 可见此截断，不静默多扩）。 """ if per_worker_capacity <= 0: per_worker_capacity = 1 + try: + pool_target = int(pool_target) if pool_target is not None else 0 + except (TypeError, ValueError): + pool_target = 0 + if pool_target < 0: + pool_target = 0 if depth <= 0: want = min_workers else: want = math.ceil(depth / per_worker_capacity) + want += pool_target return max(min_workers, min(max_workers, want)) @@ -264,7 +303,31 @@ class Autoscaler: per_worker_capacity=2, interval=30, scale_down_after=3, drain_timeout=60, state_file=None, log_dir=None, worker_cmd=None, executor=None, - guard_alive_seconds=_WORKER_ALIVE_SECONDS): + guard_alive_seconds=_WORKER_ALIVE_SECONDS, + pool_target=None, online_hosts=None, + warm_pool_per_host=CHV_WARM_POOL_PER_HOST): + self.db_path = db_path + self.worker_args = list(worker_args) + self.min_workers = min_workers + self.max_workers = max_workers + self.per_worker_capacity = per_worker_capacity + self.interval = interval + self.scale_down_after = scale_down_after + self.drain_timeout = drain_timeout + self.guard_alive_seconds = guard_alive_seconds + # CHV 混合模式池目标数（§1.2）：显式 pool_target 优先； + # 未给时按 online_hosts × warm_pool_per_host（W=2）算； + # 两者皆无即 0（旧行为）。L2 仍不自动加主机。 + if pool_target is not None: + try: + self.pool_target = max(0, int(pool_target)) + except (TypeError, ValueError): + self.pool_target = 0 + elif online_hosts is not None: + self.pool_target = pool_target_for_hosts( + online_hosts, warm_pool_per_host) + else: + self.pool_target = 0 self.db_path = db_path self.worker_args = list(worker_args) self.min_workers = min_workers @@ -407,7 +470,8 @@ class Autoscaler: self._write_state() return self.last_depth, self.last_desired, len(self.workers) desired = desired_workers(depth, self.per_worker_capacity, - self.min_workers, self.max_workers) + self.min_workers, self.max_workers, + pool_target=self.pool_target) self.last_depth = depth self.last_desired = desired self._reap() @@ -477,8 +541,9 @@ class Autoscaler: if action != "none": log_event(self.logger, "autoscaler_reconcile", message="autoscaler reconcile: depth=%s desired=%s " - "actual=%s action=%s" - % (depth, desired, len(self.workers), action)) + "actual=%s action=%s pool_target=%s" + % (depth, desired, len(self.workers), action, + self.pool_target)) self._write_state() return depth, desired, len(self.workers) @@ -489,6 +554,7 @@ class Autoscaler: "timestamp": time.time(), "depth": self.last_depth, "desired": self.last_desired, + "pool_target": self.pool_target, "actual": len(self.workers), "workers": [{"id": wid, "pid": entry["proc"].pid, "started_at": entry["started_at"]} @@ -571,6 +637,16 @@ def main(argv=None): "已显式给 --executor 时以显式为准）；" "不给时按全部任务算（可能轻微多扩，" "受 max-workers 封顶）") + parser.add_argument("--pool-target", type=int, default=None, + help="CHV 混合模式池目标数（§1.2）：期望值" + " = 队列深度算出的 fleet 需要量 + 该数；" + "不给时按 --online-hosts × " + "--warm-pool-per-host 算；两者皆无即 0") + parser.add_argument("--online-hosts", type=int, default=None, + help="在线 CHV 宿主数：与 --warm-pool-per-host" + " 相乘得池目标数") + parser.add_argument("--warm-pool-per-host", type=int, default=2, + help="每宿主预热数 W，已定值 2 台/宿主（§1.2）") parser.add_argument("worker_args", nargs=argparse.REMAINDER, help="传给 cloud.worker 的参数（写在 -- 之后）") args = parser.parse_args(argv) @@ -595,7 +671,9 @@ def main(argv=None): per_worker_capacity=args.per_worker_capacity, interval=args.interval, scale_down_after=args.scale_down_after, drain_timeout=args.drain_timeout, state_file=args.state_file, - log_dir=args.log_dir, executor=args.executor) + log_dir=args.log_dir, executor=args.executor, + pool_target=args.pool_target, online_hosts=args.online_hosts, + warm_pool_per_host=args.warm_pool_per_host) def _on_signal(signum, frame): scaler.request_stop() diff --git a/cloud/chv/executor.py b/cloud/chv/executor.py index ac91212..77b35f3 100644 --- a/cloud/chv/executor.py +++ b/cloud/chv/executor.py @@ -69,6 +69,17 @@ from cloud.chv.hardening import ( validate_net_entries, validate_snapshot_url, ) +from cloud.chv.hybrid_events import ( + RESIDENT_UPGRADE_HINT, + SNAPSHOT_TENANT_MISMATCH, + VM_RESIDENT_DEMOTED, + VM_RESIDENT_PROMOTED, + VM_RESIDENT_SUSPENDED, + VM_RESTORE_DENIED, + WARM_POOL_ACQUIRE, + WARM_POOL_MISS, + WARM_POOL_RELEASE, +) from cloud.chv.ssh_transport import build_ssh_command as _build_ssh_cmd from cloud.chv.ssh_transport import known_hosts_path as _known_hosts_path from cloud.chv.ssh_transport import ( @@ -2524,7 +2535,7 @@ class CloudHypervisorExecutor(Executor): "gitea_synced": bool(record.get("gitea_synced"))}, now=_time.time()) - self._emit("warm_pool_acquire", task, + self._emit(WARM_POOL_ACQUIRE, task, "vm=%s mode=resumed resume_seconds=%.2f" " warm_hit=%s" % (vm, resume_seconds, @@ -2747,7 +2758,7 @@ class CloudHypervisorExecutor(Executor): "tier": tier, "state": "resident-active", "last_active_at": now, "lease_expires_at": now + suspend + demote}, now=now) - self._emit("vm_resident_promoted", + self._emit(VM_RESIDENT_PROMOTED, {"id": "persist-%s" % tenant_id, "project_id": tenant_id}, "host_id=%s tenant=%s vm=%s tier=%s" @@ -2807,7 +2818,7 @@ class CloudHypervisorExecutor(Executor): tenant, {"state": "resident-paused"}, now=now) self._emit( - "vm_resident_suspended", + VM_RESIDENT_SUSPENDED, {"id": "persist-%s" % tenant, "project_id": tenant}, "host_id=%s tenant=%s vm=%s idle_seconds=%.0f" @@ -2893,7 +2904,7 @@ class CloudHypervisorExecutor(Executor): if snapshot: updates["snapshot_url"] = snapshot self._write_resident_record(tenant_id, updates, now=now) - self._emit("vm_resident_demoted", task, + self._emit(VM_RESIDENT_DEMOTED, task, "host_id=%s tenant=%s vm=%s snapshot=%s" % (self._host_id(), tenant_id, vm, snapshot)) return {"demoted": True, "vm": vm, "snapshot": snapshot} @@ -3110,7 +3121,7 @@ class CloudHypervisorExecutor(Executor): metric = self._upgrade_hint_metric(tasks, now) data["last_hint_at"] = now self._emit( - "resident_upgrade_hint", + RESIDENT_UPGRADE_HINT, {"id": "pool-%s" % tenant_id, "project_id": tenant_id}, "host_id=%s tenant=%s tier=Free metric=%s " @@ -3862,7 +3873,7 @@ class CloudHypervisorExecutor(Executor): slot = self._clean_pool_slot_for_vm(vm) if slot is not None: self._clear_clean_pool_record(slot) - self._emit("warm_pool_release", task, + self._emit(WARM_POOL_RELEASE, task, "vm=%s host_id=%s tenant=%s" % (vm, self._host_id(), tenant)) return {"released": True, "vm": vm, "tenant_id": tenant} @@ -3966,7 +3977,7 @@ class CloudHypervisorExecutor(Executor): tenant = self._tenant_id(task) candidate = self._clean_pool_candidate() if candidate is None or not tenant: - self._emit("warm_pool_miss", task, + self._emit(WARM_POOL_MISS, task, "host_id=%s tenant=%s reason=%s" % (self._host_id(), tenant, "pool_empty" if candidate is None @@ -3986,7 +3997,7 @@ class CloudHypervisorExecutor(Executor): "gitea_synced": False, "data_disk": data_disk, "updated_at": time.time()}) self._check_clean_pool_sync(vm, slot, sync_workspace) - self._emit("warm_pool_acquire", task, + self._emit(WARM_POOL_ACQUIRE, task, "vm=%s host_id=%s tenant=%s mode=resumed " "resume_seconds=%.2f warm_hit=true" % (vm, self._host_id(), tenant, resume_seconds)) @@ -4088,16 +4099,33 @@ class CloudHypervisorExecutor(Executor): sandbox=("read-only" if task.get("plan") else None)) def recycle_persistent_vm(self, task_or_name: dict | str, - tenant_id: str | None = None) -> dict: + tenant_id: str | None = None, + extra_env: dict | None = None, + sync: bool = True) -> dict: """显式回收持久 VM（唯一删用户盘的入口）：先 shutdown （stop_vm，keep_disk=False），再删 overlay 与数据盘、 清 tenant 标记与持久登记。返回 {vm, state: recycled, deleted_disks}。与 recycle_vm 的区别：后者只关 VM - 不删持久盘（持久盘须显式回收才删，用户数据不静默丢）。""" + 不删持久盘（持久盘须显式回收才删，用户数据不静默丢）。 + + C 轨新增（§3.3 同款回收同步门，见 recycle_vm）：sync + 默认开；配了 Gitea 同步且 push 失败时不 shutdown、 + 不删盘，返回 sync_failed（vm_retained=True， + deleted_disks=[]），等人工/重试。 + """ vm = self.vm_name(task_or_name) tenant = tenant_id or self._tenant_id( task_or_name if isinstance(task_or_name, dict) else {"id": task_or_name}) + if sync: + task_dict = task_or_name if isinstance( + task_or_name, dict) else {"id": task_or_name} + ok, sync_detail = self._recycle_sync_gate( + task_dict, extra_env=extra_env) + if not ok: + return self._sync_failed_result(vm, sync_detail, []) + else: + sync_detail = "sync skipped by caller" self.shutdown_vm(task_or_name) deleted = [] if delete_task_disk(self.state_dir, vm): @@ -4115,7 +4143,10 @@ class CloudHypervisorExecutor(Executor): else {"id": task_or_name}, "vm=%s deleted_disks=%s" % (vm, deleted)) return {"vm": vm, "state": "recycled", - "deleted_disks": deleted} + "deleted_disks": deleted, + "sync_ok": True, + "sync_detail": sync_detail, + "vm_retained": False} def snapshot_vm(self, task_or_name: dict | str, snapshot_url: str) -> str: """给（已暂停的）VM 做快照：ch-remote snapshot <裸 URL>。 @@ -4206,12 +4237,12 @@ class CloudHypervisorExecutor(Executor): 只带 host_id + 请求租户 + 原因，不带快照内容（不带 URL、 不带归属 tenant 值）；错误码一律 snapshot_tenant_mismatch。""" requester = self._tenant_id(task) - self._emit("vm_restore_denied", task, + self._emit(VM_RESTORE_DENIED, task, "host_id=%s tenant=%s reason=%s" % (self._host_id(), requester, reason), - "snapshot_tenant_mismatch") + SNAPSHOT_TENANT_MISMATCH) raise ChvFailure( - "snapshot_tenant_mismatch", + SNAPSHOT_TENANT_MISMATCH, "快照 tenant 校验失败（%s），拒绝恢复：请求方 " "tenant=%s" % (reason, requester)) @@ -4242,9 +4273,9 @@ class CloudHypervisorExecutor(Executor): " tenant=%s，拒绝跨用户恢复" % (self.vm_name(task), snapshot_url, marked, requester), - "snapshot_tenant_mismatch") + SNAPSHOT_TENANT_MISMATCH) raise ChvFailure( - "snapshot_tenant_mismatch", + SNAPSHOT_TENANT_MISMATCH, "快照归属 tenant=%s，与请求方 tenant=%s 不一致，" "拒绝跨用户恢复：%s" % (marked, requester, snapshot_url)) @@ -4833,7 +4864,108 @@ class CloudHypervisorExecutor(Executor): 只改数据库状态，必须真关 VM。""" return self.shutdown_vm(task_or_name) - def recycle_vm(self, task: dict) -> dict: + def sync_guest_to_gitea(self, task: dict, + extra_env: dict | None = None) -> tuple[bool, str]: + """回收前把客内已提交的工作区改动 push 回 Gitea + （DESIGN-HYBRID.md §3.3 回收流程第 1 步）。 + + 复用客内 chv-sync 的 push 腿（chv-guest/chv-sync.sh）： + 经 SSH 跑 `chv-sync push`；非 fast-forward 即失败， + 绝不 --force、不自动 rebase（与 chv-sync.sh 的 + fail-closed 语义同款）。返回 (ok, detail)： + - 未配 Gitea 同步（extract_gitea_sync_env 为空）→ + (True, "no gitea sync configured")，无物可推， + 不拦回收； + - VM 已不运行 → (True, "vm not running")：死掉的 + 客机没有可 push 的活工作树，拦回收也推不出去， + 不拿已死的 VM 卡清理（其盘仍在 state_dir 里， + 由显式回收/GC 按原口径处理）； + - push rc==0 → (True, ...)；rc!=0 或 SSH 本身失败 + → (False, 末行输出截断 200 字符）。chv-sync 的 + remote URL 是干净 https、token 只走 credential + store 文件，所以输出里没有 token（与 syncd 的 + _run_sync 同款截断口径）。 + 边界（诚实写明）：push 只推已提交的提交，未提交的 + 工作树改动不参与同步——与客内 syncd 常驻循环同款 + 口径，本方法不自动 commit（自动 commit 是新语义， + 设计未定）；同步失败的判定只看 push 本身。 + 本方法不重试：push 是幂等的，失败后调用方 + （recycle_vm / 人工）重新调回收即重试。 + """ + # 方法内 import：与 wake_vm 同款，避免顶层循环导入。 + from cloud.chv.gitea_sync import extract_gitea_sync_env + sync = extract_gitea_sync_env( + task, extra_env, + config_sync=getattr(self, "gitea_sync_env", None)) + if not sync: + return True, "no gitea sync configured" + if not self._vm_is_running(task): + return True, "vm not running: nothing to sync" + try: + ssh_cmd = self.build_ssh_command(task, ["chv-sync", "push"]) + result = self._call_runner(ssh_cmd, + timeout=self.task_timeout) + except Exception as exc: + return False, "ssh failed: %s" % str(exc)[:200] + if getattr(result, "returncode", 1) == 0: + return True, "chv-sync push ok" + tail = ((getattr(result, "stdout", "") or "") + + (getattr(result, "stderr", "") or "")).strip() + tail = tail.splitlines()[-1] if tail else "" + return False, "chv-sync push failed: %s" % tail[:200] + + def _emit_recycle_sync_failed(self, task, vm, reason): + """回收同步失败的审计（§3.3“同步失败记审计”）： + 事件 vm_recycle_sync_failed（见 hybrid_events）， + 带 host_id + vm + tenant_id（有则带），绝不带密钥值。 + 审计本身 best-effort：格式化失败时退化成裸 vm 明细， + 绝不因此掩盖 sync_failed 的返回。""" + task_dict = task if isinstance(task, dict) else {"id": task} + try: + from cloud.chv.hybrid_events import ( + VM_RECYCLE_SYNC_FAILED, format_detail) + fields = {"vm": vm, "reason": reason} + tenant = self._tenant_id(task_dict) + if tenant: + fields["tenant_id"] = tenant + detail = format_detail(VM_RECYCLE_SYNC_FAILED, fields) + except Exception: + detail = "vm=%s reason=%s" % (vm, str(reason)[:200]) + try: + self._emit(VM_RECYCLE_SYNC_FAILED, task_dict, detail, + kind="sync_failed") + except Exception: + pass + + @staticmethod + def _sync_failed_result(vm, detail, deleted_disks=None): + """同步失败的回收返回（§3.3）：state/status 均为 + sync_failed，VM 保留。recycle_vm 与 + recycle_persistent_vm 共用，免得两处拼同一份 dict； + 各自再补自己的字段（preserved/lost 或 deleted_disks）。""" + out = {"vm": vm, "state": "sync_failed", + "status": "sync_failed", "sync_ok": False, + "sync_error": detail, "vm_retained": True} + if deleted_disks is not None: + out["deleted_disks"] = deleted_disks + return out + + def _recycle_sync_gate(self, task, extra_env=None): + """回收同步门（§3.3 新增语义的共用判定）： + 同步失败 → (False, detail)；可回收 → (True, detail)。 + 调用方在 False 时必须不 shutdown、不删盘、不发 + recycled 审计，只记 sync_failed 审计并返回 + state/status 均为 sync_failed 的结果。""" + ok, detail = self.sync_guest_to_gitea(task, + extra_env=extra_env) + if ok: + return True, detail + self._emit_recycle_sync_failed(task, self.vm_name(task), + detail) + return False, detail + + def recycle_vm(self, task: dict, extra_env: dict | None = None, + sync: bool = True) -> dict: """回收 VM（任务结束后的销毁语义）： 先 shutdown，再返回保留/丢失清单。保留的只有宿主侧 的 task_checkpoints / task_attempts（数据库里的断点与 @@ -4841,9 +4973,30 @@ class CloudHypervisorExecutor(Executor): 丢失的是客机内的临时文件与未拷出的 worktree 临时物 ——recycle 绝不声称“恢复临时文件”。需要拷出的工件 必须在 shutdown 之前由调用方经 SSH 拷出，本方法不 - 假装替调用方拷了。""" + 假装替调用方拷了。 + + C 轨新增（DESIGN-HYBRID.md §3.3）：sync=True（默认） + 时先过回收同步门——配了 Gitea 同步且 `chv-sync push` + 失败时，任务标同步失败（返回 state/status 均为 + sync_failed，sync_ok=False，vm_retained=True），VM + 不回收、不复用，等人工/重试，绝不丢改动后静默回收； + 审计记 vm_recycle_sync_failed。未配同步、VM 已死、 + 或 sync=False（调用方显式跳过）时走旧回收路径。 + """ if not isinstance(task, dict) or task.get("id") is None: raise ValueError("任务缺少 id") + if sync: + ok, sync_detail = self._recycle_sync_gate( + task, extra_env=extra_env) + if not ok: + out = self._sync_failed_result(self.vm_name(task), + sync_detail) + out["preserved"] = ["task_checkpoints", + "task_attempts"] + out["lost"] = [] + return out + else: + sync_detail = "sync skipped by caller" self.shutdown_vm(task) self._emit("vm_recycled", task, "vm=%s" % self.vm_name(task)) @@ -4852,6 +5005,9 @@ class CloudHypervisorExecutor(Executor): "state": "recycled", "preserved": ["task_checkpoints", "task_attempts"], "lost": ["guest /tmp", "guest worktree temp files"], + "sync_ok": True, + "sync_detail": sync_detail, + "vm_retained": False, } def _stop_dead_socket_vms(self, entries): diff --git a/cloud/chv/hybrid_events.py b/cloud/chv/hybrid_events.py new file mode 100644 index 0000000..df2821e --- /dev/null +++ b/cloud/chv/hybrid_events.py @@ -0,0 +1,146 @@ +"""CHV 混合模式审计事件目录（C 轨，feat-chv-hybrid-c）。 + +唯一依据：~/workspace/chv-multi-host/DESIGN-HYBRID.md +§7 与 §9.1 第 13 项。事件名字符串与设计原文一字不差， +供 A/B 两轨的发射点 import（A/B 工作树里还没有本文件， +他们先按设计原文的事件名字符串直接发射，协调员集成时 +统一收到本模块名下）。 + +规则（设计 §7）： +- 全部事件带 host_id（宿主名，默认 socket.gethostname()）。 +- 绝不带密钥值：字段名命中密钥名（见 SECRET_FIELD_NAMES） + 即拒绝，错误信息只含字段名、不含值；调用方仍须遵守 + executor._emit 的既有契约（detail 只含 VM 名、路径、 + URL、kind，不含任何 key 值）。 +- 本模块只定名 + 必填字段 + 校验/格式化，不发事件、 + 不读库、不碰 VM；发射仍走各轨现有的 _emit / 日志路径。 + +另有一枚 C 轨事件 vm_recycle_sync_failed：不在 §9.1-13 +的名单里，但 §3.3 要求“同步失败 → 任务标同步失败 +（审计）”，回收同步失败的审计须有个名字，故在此单列 +并注明来源，不混进设计名单。 +""" + +import socket + +# -- 设计原文事件名（§9.1 第 13 项 / §7，一字不差） ------------ +WARM_POOL_ACQUIRE = "warm_pool_acquire" +WARM_POOL_MISS = "warm_pool_miss" +WARM_POOL_RELEASE = "warm_pool_release" +VM_RESIDENT_SUSPENDED = "vm_resident_suspended" +VM_RESIDENT_DEMOTED = "vm_resident_demoted" +VM_RESIDENT_PROMOTED = "vm_resident_promoted" +RESIDENT_UPGRADE_HINT = "resident_upgrade_hint" +SNAPSHOT_TENANT_MISMATCH = "snapshot_tenant_mismatch" +VM_RESTORE_DENIED = "vm_restore_denied" + +# 设计 §9.1-13 名单一共 9 个事件（acquire/miss/release +# 是 §7“warm_pool_acquire/miss/release”的三枚展开， +# miss 的全名 warm_pool_miss 见 §7“须记 warm_pool_miss +# 审计”）。 +HYBRID_EVENTS = ( + WARM_POOL_ACQUIRE, + WARM_POOL_MISS, + WARM_POOL_RELEASE, + VM_RESIDENT_SUSPENDED, + VM_RESIDENT_DEMOTED, + VM_RESIDENT_PROMOTED, + RESIDENT_UPGRADE_HINT, + SNAPSHOT_TENANT_MISMATCH, + VM_RESTORE_DENIED, +) + +# -- C 轨回收同步审计（§3.3，非 §9.1-13 名单） --------------- +VM_RECYCLE_SYNC_FAILED = "vm_recycle_sync_failed" + +ALL_KNOWN_EVENTS = HYBRID_EVENTS + (VM_RECYCLE_SYNC_FAILED,) + +# -- 必填字段：全部事件都须带 host_id ------------------------- +# 其余字段按事件语义：vm = VM 名，tenant_id = 租户， +# reason = 失败原因短文本（已截断，绝不含密钥值）。 +REQUIRED_FIELDS = { + WARM_POOL_ACQUIRE: ("host_id", "vm"), + WARM_POOL_MISS: ("host_id",), + WARM_POOL_RELEASE: ("host_id", "vm"), + VM_RESIDENT_SUSPENDED: ("host_id", "tenant_id", "vm"), + VM_RESIDENT_DEMOTED: ("host_id", "tenant_id", "vm"), + VM_RESIDENT_PROMOTED: ("host_id", "tenant_id", "vm"), + RESIDENT_UPGRADE_HINT: ("host_id", "tenant_id"), + SNAPSHOT_TENANT_MISMATCH: ("host_id", "tenant_id", "vm"), + VM_RESTORE_DENIED: ("host_id", "tenant_id", "vm"), + VM_RECYCLE_SYNC_FAILED: ("host_id", "vm"), +} + +# 密钥字段名：命中即拒绝。exact 匹配 + token/secret/ +# password 子串匹配（host_key 等公开指纹名不在此列， +# 不可误拦）。 +SECRET_FIELD_NAMES = frozenset({ + "gitea_token", "token", "secret", "password", "api_key", + "apikey", "private_key", "access_key", "session_key", + "credential", "credentials", "seed", "auth", +}) +_SECRET_SUBSTRINGS = ("token", "secret", "password") + + +def _is_secret_field(name: str) -> bool: + lowered = str(name).lower() + if lowered in SECRET_FIELD_NAMES: + return True + return any(part in lowered for part in _SECRET_SUBSTRINGS) + + +def default_host_id() -> str: + """宿主名：审计事件的 host_id 缺省值。""" + return socket.gethostname() + + +def validate_event_fields(event: str, fields: dict) -> dict: + """校验一个事件的字段，返回清洗后的副本（去 None 值）。 + + 未知事件名、缺必填字段、host_id 为空、字段名命中密钥 + 名时一律 ValueError；错误信息只含事件名/字段名， + 绝不含字段值。""" + if event not in ALL_KNOWN_EVENTS: + raise ValueError("未知审计事件：%s" % event) + cleaned = {k: v for k, v in dict(fields or {}).items() + if v is not None} + secret = sorted(k for k in cleaned if _is_secret_field(k)) + if secret: + raise ValueError( + "审计事件 %s 不得带密钥字段：%s" % (event, ",".join(secret))) + missing = [k for k in REQUIRED_FIELDS[event] + if k not in cleaned + or (isinstance(cleaned[k], str) + and not cleaned[k].strip())] + if missing: + raise ValueError( + "审计事件 %s 缺必填字段：%s" % (event, ",".join(missing))) + return cleaned + + +def build_event(event: str, fields: dict | None = None, + host_id: str | None = None) -> dict: + """组装一个事件的字段 dict：host_id 未给时取本机宿主名， + 再过 validate_event_fields。返回的 dict 可直接用于 + format_detail 或交审计回调。""" + merged = dict(fields or {}) + if host_id is not None: + merged["host_id"] = host_id + elif not merged.get("host_id"): + merged["host_id"] = default_host_id() + return validate_event_fields(event, merged) + + +def format_detail(event: str, fields: dict | None = None, + host_id: str | None = None) -> str: + """把事件字段格式化成 executor._emit 的 detail 字符串 + （"k=v k=v"，键排序，值截断 200 字符、去换行）。 + 先过 build_event 校验，所以带 host_id、不带密钥字段。""" + cleaned = build_event(event, fields, host_id=host_id) + parts = [] + for key in sorted(cleaned): + value = str(cleaned[key]).replace("\n", " ").replace("\r", " ") + if len(value) > 200: + value = value[:200] + parts.append("%s=%s" % (key, value)) + return " ".join(parts) diff --git a/cloud/tests/test_chv_hybrid_c.py b/cloud/tests/test_chv_hybrid_c.py new file mode 100644 index 0000000..7949470 --- /dev/null +++ b/cloud/tests/test_chv_hybrid_c.py @@ -0,0 +1,573 @@ +"""CHV 混合模式 C 轨测试（feat-chv-hybrid-c）。 + +只覆盖本轨的三个改动点（DESIGN-HYBRID.md §9.1 第 10–13 项）： +- autoscaler desired_workers 作用域扩展（§1.2，W=2）； +- 回收同步失败即任务标同步失败、VM 不回收（§3.3）； +- 审计事件目录（§7，事件名与设计原文一致、带 host_id、 + 不带密钥值）与 orphans facade 核实（§1.3/ADR-5/ADR-12）。 + +只用标准库与 mock runner/launcher/api_client，绝不启动 +真实 VM、不联网、不碰生产库。真机验证（WSL 3 台排布、 +热恢复 ≤2 秒等）不在本文件声称已验证。 +""" + +import json +import os +import socket +import subprocess +import sys +import tempfile +import time +import unittest +from pathlib import Path +from unittest import mock + +try: + from cloud.autoscaler import ( + CHV_WARM_POOL_PER_HOST, + Autoscaler, + desired_workers, + pool_target_for_hosts, + ) + from cloud.chv import hybrid_events + from cloud.chv.executor import CloudHypervisorExecutor + from cloud.chv.orphans import reconcile_orphans_for_executor + from cloud.server import init_db +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.autoscaler import ( + CHV_WARM_POOL_PER_HOST, + Autoscaler, + desired_workers, + pool_target_for_hosts, + ) + from cloud.chv import hybrid_events + from cloud.chv.executor import CloudHypervisorExecutor + from cloud.chv.orphans import reconcile_orphans_for_executor + from cloud.server import init_db + + +class FakeResult: + def __init__(self, returncode=0, stdout="", stderr=""): + self.returncode = returncode + self.stdout = stdout + self.stderr = stderr + + +class FakeProc: + def __init__(self, alive=True): + self._alive = alive + self.terminated = False + self.killed = False + + def poll(self): + return None if self._alive else 1 + + def terminate(self): + self.terminated = True + self._alive = False + + def wait(self, timeout=None): + return 0 + + def kill(self): + self.killed = True + self._alive = False + + +def _write(path, content=b"x"): + with open(path, "wb") as fh: + fh.write(content) + os.chmod(path, 0o755) + return path + + +# -- 11. desired_workers + 池目标数 -------------------------------------- +class DesiredWorkersPoolTestCase(unittest.TestCase): + def test_pool_target_zero_unchanged(self): + # pool_target=0（缺省）时与旧公式一字不差。 + self.assertEqual(desired_workers(0, 2, 1, 2), 1) + self.assertEqual(desired_workers(0, 2, 0, 2), 0) + self.assertEqual(desired_workers(1, 2, 1, 2), 1) + self.assertEqual(desired_workers(3, 2, 1, 2), 2) + self.assertEqual(desired_workers(10, 2, 1, 2), 2) + self.assertEqual(desired_workers(5, 1, 1, 3), 3) + self.assertEqual(desired_workers(5, 1, 1, 3, pool_target=0), 3) + + def test_pool_target_additive(self): + # 期望值 = fleet 需要量 + 池目标数（§1.2）。 + self.assertEqual(desired_workers(0, 2, 1, 20, pool_target=12), 13) + self.assertEqual(desired_workers(4, 2, 1, 20, pool_target=12), 14) + self.assertEqual(desired_workers(3, 2, 1, 20, pool_target=2), 4) + # depth=0 且 min_workers=0 时池目标仍维持（预热不因 + # 队列空而归零）。 + self.assertEqual(desired_workers(0, 2, 0, 20, pool_target=12), 12) + + def test_pool_target_capped_by_max_workers(self): + # 池目标数计入 max_workers 上限：配不下即截断，不静默多扩。 + self.assertEqual(desired_workers(0, 2, 1, 2, pool_target=12), 2) + self.assertEqual(desired_workers(10, 2, 1, 3, pool_target=12), 3) + + def test_pool_target_negative_or_none_is_zero(self): + self.assertEqual(desired_workers(0, 2, 1, 2, pool_target=-5), 1) + self.assertEqual(desired_workers(0, 2, 1, 2, pool_target=None), 1) + + +class PoolTargetForHostsTestCase(unittest.TestCase): + def test_w_is_two_per_host(self): + self.assertEqual(CHV_WARM_POOL_PER_HOST, 2) + + def test_pool_target_is_w_times_online_hosts(self): + self.assertEqual(pool_target_for_hosts(6), 12) + self.assertEqual(pool_target_for_hosts(22), 44) + self.assertEqual(pool_target_for_hosts(1), 2) + self.assertEqual(pool_target_for_hosts(0), 0) + + def test_invalid_hosts_is_zero(self): + self.assertEqual(pool_target_for_hosts(-1), 0) + self.assertEqual(pool_target_for_hosts(None), 0) + self.assertEqual(pool_target_for_hosts("not-a-number"), 0) + + +def _insert_task(db_path, task_id): + now = time.time() + import sqlite3 + 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)" + " VALUES(?, 'p1', 'x', 'queued', NULL, '[]', 0," + " 'tester', ?, ?, NULL)", + (task_id, now, now)) + conn.commit() + finally: + conn.close() + + +class AutoscalerPoolReconcileTestCase(unittest.TestCase): + """L1 其余语义不变：30 秒一轮（interval 缺省）、scale up + fast / scale down slow（scale_down_after 冷却）原样复用， + 只验证期望值多了池目标数。""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + init_db(self.db_path) + self.state_file = os.path.join(self.tmp.name, "state.json") + self.fake_cmd = [sys.executable, "-c", + "import time; time.sleep(120)"] + + def tearDown(self): + if hasattr(self, "scaler"): + for wid in list(self.scaler.workers): + try: + self.scaler._stop_worker(wid) + except Exception: + pass + self.tmp.cleanup() + + def _scaler(self, **kwargs): + kwargs.setdefault("min_workers", 1) + kwargs.setdefault("max_workers", 5) + kwargs.setdefault("per_worker_capacity", 2) + kwargs.setdefault("scale_down_after", 1) + kwargs.setdefault("drain_timeout", 5) + kwargs.setdefault("state_file", self.state_file) + self.scaler = Autoscaler( + self.db_path, [], worker_cmd=self.fake_cmd, **kwargs) + return self.scaler + + def test_empty_queue_still_holds_pool(self): + scaler = self._scaler(pool_target=2) + depth, desired, actual = scaler.reconcile() + self.assertEqual((depth, desired, actual), (0, 3, 3)) + + def test_depth_plus_pool_and_scale_down_slow(self): + scaler = self._scaler(pool_target=2, scale_down_after=2) + for tid in ("A", "B", "C", "D"): + _insert_task(self.db_path, tid) + depth, desired, actual = scaler.reconcile() + self.assertEqual((depth, desired, actual), (4, 4, 4)) + # 任务全完成：深度归零但池目标仍在，期望值回落到 3 + # 而不是 1；缩容仍须等满 2 轮冷却（scale down slow）。 + import sqlite3 + conn = sqlite3.connect(self.db_path) + try: + conn.execute("UPDATE tasks SET status='completed'") + conn.commit() + finally: + conn.close() + scaler.reconcile() + self.assertEqual(len(scaler.workers), 4) + self.assertEqual(scaler.last_action, "scale_down_pending_1/2") + depth, desired, actual = scaler.reconcile() + self.assertEqual((depth, desired, actual), (0, 3, 3)) + + def test_online_hosts_derives_pool_target(self): + scaler = self._scaler(online_hosts=6, max_workers=20) + self.assertEqual(scaler.pool_target, 12) + depth, desired, actual = scaler.reconcile() + self.assertEqual((depth, desired, actual), (0, 13, 13)) + + def test_explicit_pool_target_wins_over_hosts(self): + scaler = self._scaler(pool_target=2, online_hosts=6) + self.assertEqual(scaler.pool_target, 2) + + def test_state_file_records_pool_target(self): + scaler = self._scaler(pool_target=2) + scaler.reconcile() + with open(self.state_file) as fh: + state = json.load(fh) + self.assertEqual(state["pool_target"], 2) + self.assertEqual(state["desired"], 3) + + def test_main_passes_pool_flags(self): + import cloud.autoscaler as autoscaler_mod + with mock.patch.object( + autoscaler_mod, "Autoscaler") as mock_cls, \ + mock.patch("signal.signal"): + mock_cls.return_value.run_loop = mock.Mock() + autoscaler_mod.main([ + "--db", self.db_path, "--pool-target", "12", + "--", "--interval", "10"]) + kwargs = mock_cls.call_args.kwargs + self.assertEqual(kwargs["pool_target"], 12) + with mock.patch.object( + autoscaler_mod, "Autoscaler") as mock_cls, \ + mock.patch("signal.signal"): + mock_cls.return_value.run_loop = mock.Mock() + autoscaler_mod.main([ + "--db", self.db_path, "--online-hosts", "6", + "--", "--interval", "10"]) + kwargs = mock_cls.call_args.kwargs + self.assertEqual(kwargs["online_hosts"], 6) + self.assertEqual(kwargs["warm_pool_per_host"], 2) + + +# -- 12. 回收同步 --------------------------------------------------------- +class RecycleSyncCase(unittest.TestCase): + """mock runner/launcher/api_client，不启动真实 VM。 + push_rc 控制客内 `chv-sync push` 的返回值。""" + + TOKEN = "SECRET-TOKEN-XYZ-DO-NOT-LEAK" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.dir = self.tmp.name + self.binary = _write(os.path.join(self.dir, "cloud-hypervisor")) + self.ch_remote = _write(os.path.join(self.dir, "ch-remote")) + self.kernel = _write(os.path.join(self.dir, "vmlinux")) + self.rootfs = _write(os.path.join(self.dir, "disk.raw"), b"R" * 100) + self.ssh_key = _write(os.path.join(self.dir, "id_chv")) + self.kvm = _write(os.path.join(self.dir, "kvm")) + self.cpuinfo = os.path.join(self.dir, "cpuinfo") + with open(self.cpuinfo, "w") as fh: + fh.write("flags : fpu vmx sse\n") + self.state_dir = os.path.join(self.dir, "state") + self.calls = [] + self.audit = [] + self.proc = FakeProc(alive=True) + self.push_rc = 0 + self.push_stdout = "已 push" + self.push_stderr = "" + self.raise_on_push = False + + def tearDown(self): + self.tmp.cleanup() + + def _runner(self, cmd, timeout=None): + self.calls.append(list(cmd)) + joined = " ".join(str(c) for c in cmd) + if "chv-sync" in joined: + if self.raise_on_push: + raise RuntimeError("ssh broken") + return FakeResult(self.push_rc, self.push_stdout, + self.push_stderr) + return FakeResult(0, "", "") + + def _make_exec(self, sync_configured=True): + env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888", + "CHV_HOST_KEY_MODE": "keyscan", + "CHV_SSH_FORWARD": "1"} + if sync_configured: + env.update({ + "CHV_GITEA_URL": "https://gitea.tose.top", + "CHV_GITEA_USER": "tester", + "CHV_GITEA_TOKEN": self.TOKEN, + "CHV_SYNC_REPO": "tester/repo", + }) + ex = CloudHypervisorExecutor( + binary=self.binary, ch_remote=self.ch_remote, + kernel=self.kernel, rootfs=self.rootfs, + ssh_key=self.ssh_key, kvm_path=self.kvm, + cpuinfo_path=self.cpuinfo, state_dir=self.state_dir, + runner=self._runner, + launcher=lambda cmd: self.proc, + api_client=lambda *a, **k: (200, {}), + sleeper=lambda s: None, env=env) + ex.audit_callback = lambda e, t, d, k=None: self.audit.append( + (e, t.get("id") if isinstance(t, dict) else t, d, k)) + return ex + + def _task(self): + return {"id": "abc123", "prompt": "x", "project_id": "proj1"} + + def _mark_vm_running(self, ex, vm_name="chv-abc123"): + os.makedirs(self.state_dir, exist_ok=True) + with open(os.path.join(self.state_dir, vm_name + ".sock"), + "w") as fh: + fh.write("") + ex._vms[vm_name] = self.proc + + def _audit_events(self): + return [e for e, _t, _d, _k in self.audit] + + def _audit_details(self): + return [d for _e, _t, d, _k in self.audit] + + def test_no_sync_configured_recycles_as_before(self): + # 未配 Gitea 同步：无物可推，走旧回收路径（回归）。 + ex = self._make_exec(sync_configured=False) + self._mark_vm_running(ex) + out = ex.recycle_vm(self._task()) + self.assertEqual(out["state"], "recycled") + self.assertTrue(out["sync_ok"]) + self.assertFalse(out["vm_retained"]) + self.assertTrue(self.proc.terminated) + self.assertIn("vm_recycled", self._audit_events()) + self.assertFalse(any("chv-sync" in " ".join(c) + for c in self.calls)) + + def test_sync_push_ok_then_recycle(self): + ex = self._make_exec() + self._mark_vm_running(ex) + self.push_rc = 0 + out = ex.recycle_vm(self._task()) + self.assertEqual(out["state"], "recycled") + self.assertTrue(out["sync_ok"]) + self.assertFalse(out["vm_retained"]) + self.assertTrue(self.proc.terminated) + self.assertIn("vm_recycled", self._audit_events()) + self.assertTrue(any("chv-sync" in " ".join(c) + for c in self.calls)) + self.assertNotIn("vm_recycle_sync_failed", + self._audit_events()) + + def test_sync_push_failed_marks_sync_failed_and_retains_vm(self): + # §3.3 新增语义：同步失败即任务标同步失败、VM 不回收。 + ex = self._make_exec() + self._mark_vm_running(ex) + self.push_rc = 1 + self.push_stderr = "push 被拒绝（远端已有新提交）" + out = ex.recycle_vm(self._task()) + self.assertEqual(out["state"], "sync_failed") + self.assertEqual(out["status"], "sync_failed") + self.assertFalse(out["sync_ok"]) + self.assertTrue(out["vm_retained"]) + self.assertIn("push", out["sync_error"]) + # VM 保留：进程未停、socket 未删、未发 recycled/shutdown。 + self.assertFalse(self.proc.terminated) + self.assertTrue(os.path.exists(os.path.join( + self.state_dir, "chv-abc123.sock"))) + self.assertNotIn("vm_recycled", self._audit_events()) + self.assertNotIn("vm_shutdown", self._audit_events()) + self.assertNotIn("vm_destroy", self._audit_events()) + # 记审计：事件名正确、带 host_id、不带密钥值。 + self.assertIn("vm_recycle_sync_failed", self._audit_events()) + details = " ".join(self._audit_details()) + self.assertIn("host_id=", details) + self.assertNotIn(self.TOKEN, details) + self.assertNotIn(self.TOKEN, json.dumps(self.audit)) + + def test_sync_ssh_error_also_retains_vm(self): + ex = self._make_exec() + self._mark_vm_running(ex) + self.raise_on_push = True + out = ex.recycle_vm(self._task()) + self.assertEqual(out["state"], "sync_failed") + self.assertTrue(out["vm_retained"]) + self.assertFalse(self.proc.terminated) + self.assertIn("vm_recycle_sync_failed", self._audit_events()) + + def test_sync_failed_then_retry_after_push_ok(self): + # 人工/重试路径：push 修好后再次回收即成功，不卡死。 + ex = self._make_exec() + self._mark_vm_running(ex) + self.push_rc = 1 + out = ex.recycle_vm(self._task()) + self.assertEqual(out["state"], "sync_failed") + self.push_rc = 0 + out = ex.recycle_vm(self._task()) + self.assertEqual(out["state"], "recycled") + self.assertTrue(self.proc.terminated) + + def test_sync_skip_flag_recycles_despite_push_failure(self): + ex = self._make_exec() + self._mark_vm_running(ex) + self.push_rc = 1 + out = ex.recycle_vm(self._task(), sync=False) + self.assertEqual(out["state"], "recycled") + self.assertTrue(self.proc.terminated) + + def test_vm_not_running_does_not_block_cleanup(self): + # VM 已死：没有活工作树可推，不拿死 VM 卡清理。 + ex = self._make_exec() + out = ex.recycle_vm(self._task()) + self.assertEqual(out["state"], "recycled") + + def test_recycle_persistent_vm_sync_failed_keeps_disks(self): + ex = self._make_exec() + self._mark_vm_running(ex, vm_name="chv-persist-proj1") + self.push_rc = 1 + out = ex.recycle_persistent_vm({"id": "chv-persist-proj1"}) + self.assertEqual(out["state"], "sync_failed") + self.assertEqual(out["status"], "sync_failed") + self.assertTrue(out["vm_retained"]) + self.assertEqual(out["deleted_disks"], []) + self.assertFalse(self.proc.terminated) + self.assertNotIn("vm_persistent_recycled", + self._audit_events()) + self.assertIn("vm_recycle_sync_failed", self._audit_events()) + + def test_recycle_vm_missing_id_still_raises(self): + ex = self._make_exec() + with self.assertRaises(ValueError): + ex.recycle_vm({}) + + +# -- 13. 审计事件目录 ------------------------------------------------------ +class HybridEventsTestCase(unittest.TestCase): + def test_design_event_names_verbatim(self): + # 事件名与 DESIGN-HYBRID.md §9.1-13 原文一字不差。 + self.assertEqual(hybrid_events.HYBRID_EVENTS, ( + "warm_pool_acquire", + "warm_pool_miss", + "warm_pool_release", + "vm_resident_suspended", + "vm_resident_demoted", + "vm_resident_promoted", + "resident_upgrade_hint", + "snapshot_tenant_mismatch", + "vm_restore_denied", + )) + + def test_every_event_requires_host_id(self): + for event in hybrid_events.HYBRID_EVENTS: + self.assertIn("host_id", + hybrid_events.REQUIRED_FIELDS[event], + event) + self.assertIn( + "host_id", + hybrid_events.REQUIRED_FIELDS[ + hybrid_events.VM_RECYCLE_SYNC_FAILED]) + + def test_build_event_defaults_host_id(self): + fields = hybrid_events.build_event( + "warm_pool_miss", {"tenant_id": "t1"}) + self.assertEqual(fields["host_id"], socket.gethostname()) + self.assertEqual(fields["tenant_id"], "t1") + + def test_build_event_rejects_missing_required(self): + with self.assertRaises(ValueError): + hybrid_events.build_event("warm_pool_acquire", {}) + with self.assertRaises(ValueError): + hybrid_events.build_event( + "vm_resident_suspended", + {"host_id": "h1", "tenant_id": "t1"}) + with self.assertRaises(ValueError): + hybrid_events.build_event("no_such_event", + {"host_id": "h1"}) + + def test_secret_field_rejected_without_leaking_value(self): + sentinel = "SECRET-TOKEN-XYZ-DO-NOT-LEAK" + with self.assertRaises(ValueError) as ctx: + hybrid_events.build_event( + "warm_pool_acquire", + {"vm": "chv-x", "GITEA_TOKEN": sentinel}) + self.assertNotIn(sentinel, str(ctx.exception)) + self.assertIn("GITEA_TOKEN", str(ctx.exception)) + + def test_format_detail_has_host_id_and_no_secret(self): + detail = hybrid_events.format_detail( + "vm_resident_promoted", + {"tenant_id": "t1", "vm": "chv-persist-t1"}) + self.assertIn("host_id=%s" % socket.gethostname(), detail) + self.assertIn("vm=chv-persist-t1", detail) + self.assertIn("tenant_id=t1", detail) + + +# -- 10. orphans facade 核实 ---------------------------------------------- +class OrphansFacadeTestCase(unittest.TestCase): + def test_facade_exists_and_cleans_only_orphans(self): + # facade 已存在：读 registry → 只清理 warm/assigned + # 条目（只清理、不分配），非法条目丢弃不连坐。 + with tempfile.TemporaryDirectory() as tmp: + registry = { + "warm-1": {"state": "warm", "vm": "chv-a", + "task_id": "t1"}, + "asg-1": {"state": "assigned", "vm": "chv-b", + "task_id": "t2"}, + "bad-1": {"state": "warm"}, + "dead-1": {"state": "recycled", "vm": "chv-c"}, + } + with open(os.path.join(tmp, "warm-pool.json"), + "w") as fh: + json.dump(registry, fh) + + class Exec: + state_dir = tmp + cleaned = [] + + def shutdown_vm(self, vm): + self.cleaned.append(vm) + return True + + def _emit(self, event, task, detail, kind=None): + pass + + ex = Exec() + cleaned = reconcile_orphans_for_executor(ex) + self.assertEqual(cleaned, ["chv-a", "chv-b"]) + self.assertEqual(ex.cleaned, ["chv-a", "chv-b"]) + with open(os.path.join(tmp, "warm-pool.json")) as fh: + self.assertEqual(json.load(fh), {}) + + def test_facade_quarantines_corrupt_registry(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "warm-pool.json") + with open(path, "w") as fh: + fh.write("{not json") + + class Exec: + state_dir = tmp + + def shutdown_vm(self, vm): + raise AssertionError("must not be called") + + self.assertEqual(reconcile_orphans_for_executor(Exec()), []) + self.assertTrue(os.path.exists(path + ".corrupt")) + + def test_warm_pool_still_not_wired_to_dispatch(self): + from cloud.chv.experimental_pool import WarmPool + self.assertIs(WarmPool.wired_to_dispatch, False) + + def test_orphans_module_does_not_import_experimental_pool(self): + # 生产 facade 不 import 实验模块（ADR-12）：全新解释器 + # 里只 import orphans，experimental_pool 不得被加载。 + code = ( + "import cloud.chv.orphans, sys;" + " assert 'cloud.chv.experimental_pool' not in sys.modules;" + " print('orphans-clean')" + ) + proc = subprocess.run( + [sys.executable, "-c", code], + capture_output=True, text=True, cwd=str( + Path(__file__).resolve().parents[2])) + self.assertEqual(proc.returncode, 0, proc.stderr) + self.assertIn("orphans-clean", proc.stdout) + + +if __name__ == "__main__": + unittest.main()