workers: reap pid_verified stale workers with pre-kill audit and post-kill verification

cloud/workers.py

@@ -32,9 +32,21 @@
   前缀枚举/删除容器：失联清理只做判定（见
   classify_stale_workers），判定为 kill 的唯一可杀对象是该行
   的 pid 本人，且必须是本机 + pid 已死（pid_verified）。
+  执行路径（本轨新增，见 reap_stale_workers）：复用
+  classify_stale_workers 的判定，不另造判定函数；只对
+  action == "kill" 的行调 kill_fn，且执行时先用同一 checker
+  再验一次已死（防判定快照与执行之间 pid 被复用），杀前
+  写审计（审计写失败则不杀，fail-closed），杀后用同一
+  checker 复核 pid 已死；异机、无 pid、pid 存活或未知一律
+  只列不杀。执行路径不枚举容器、不按名/前缀匹配、不碰
+  基础设施容器、不删 workers 注册行：杀的是 pid 本人，
+  注册行的去留是另一条决策（删行会改 /workers 列表语义），
+  本轨不擅自删。亦不新增 HTTP 路由（会改 v1.1 契约）：调
+  用方（管理员/清扫器）在有权限的上下文里直接调本函数。
 """
 
 import os
+import signal
 import socket
 import time
 
@@ -350,6 +362,145 @@ def classify_stale_workers(db_path, stale_seconds=None, local_host=None,
     return out
 
 
+def _default_kill(pid):
+    """默认真杀：只给该 pid 本人发 SIGKILL，不发进程组。
+
+    pid 已死时 os.kill 抛 ProcessLookupError，调用方
+    （reap_stale_workers）把它当“已死、无需再杀”处理，
+    不是失败。绝不把信号发给进程组（会误伤同组无关进程），
+    绝不按名/前缀找进程。
+    """
+    os.kill(pid, signal.SIGKILL)
+
+
+def _default_audit(db_path, action, detail):
+    """默认审计写 audit_log：user_id=system、project_id=""。
+
+    表不存在时先建（PG 与 SQLite 各一套 DDL，同
+    cloud/automations.py 的兜底写法：PG DDL 在 SQLite 下
+    失败即回滚再试 SQLite DDL，PG 适配层下第一条即成功）。
+    写失败抛给调用方：reap 的杀前审计失败时不杀
+    （fail-closed），杀后审计失败时只记 audit_after=False。
+    """
+    conn = _db_connect(str(db_path))
+    try:
+        for ddl in (
+                "CREATE TABLE IF NOT EXISTS audit_log("
+                "id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, "
+                "user_id TEXT, project_id TEXT, action TEXT, detail TEXT, "
+                "created_at DOUBLE PRECISION)",
+                "CREATE TABLE IF NOT EXISTS audit_log("
+                "id INTEGER PRIMARY KEY AUTOINCREMENT, "
+                "user_id TEXT, project_id TEXT, action TEXT, detail TEXT, "
+                "created_at REAL)"):
+            try:
+                conn.execute(ddl)
+                conn.execute(
+                    "INSERT INTO audit_log(user_id, project_id, action,"
+                    " detail, created_at) VALUES(?, ?, ?, ?, ?)",
+                    ("system", "", action, detail, time.time()))
+                conn.commit()
+                return
+            except Exception:
+                try:
+                    conn.rollback()
+                except Exception:
+                    pass
+        raise RuntimeError("audit_log write failed")
+    finally:
+        conn.close()
+
+
+def reap_stale_workers(db_path, stale_seconds=None, local_host=None,
+                       pid_is_dead_fn=None, kill_fn=None, audit_fn=None,
+                       now=None):
+    """失联 worker 执行路径：复用判定，只杀 pid_verified。
+
+    流程（每行）：
+    - 先调 classify_stale_workers（不另造判定）：action 不是
+      kill 的行一律只列不杀，executed=False、outcome="listed"，
+      绝不调 kill_fn、不写杀审计。
+    - kill 行执行时先用同一 checker 再验一次 pid 已死：若此时
+      已不死（判定后 pid 被复用等），跳过不杀，
+      outcome="skipped_pid_alive_at_execution"，不调 kill_fn。
+    - 杀前写审计：audit_fn 未给时走 _default_audit 写
+      audit_log（action=worker_pid_kill）；审计写失败则不杀，
+      outcome="audit_failed"（fail-closed）。
+    - 杀：kill_fn 未给时走 _default_kill（SIGKILL 只给该 pid
+      本人）；ProcessLookupError 当已死处理，不是失败；其它
+      OSError 记入 kill_error，仍以后续复核定成败。
+    - 杀后用同一 checker 复核：已死 -> outcome="killed"；
+      还活着 -> outcome="kill_failed_pid_alive"；随后写
+      杀后审计（action=worker_pid_kill_verified，detail 含
+      outcome），审计成败只记 audit_after，不改 outcome。
+    - 本函数不枚举容器、不按名/前缀匹配、不碰基础设施容器、
+      不删 workers 行；返回每行的判定字段 + executed /
+      outcome / verified_dead / audit_before / audit_after /
+      kill_error。verified_dead 仅在杀后复核后才有真值，
+      未执行时为 None（不是“还活着”的断言）。
+    """
+    rows = classify_stale_workers(
+        db_path, stale_seconds=stale_seconds, local_host=local_host,
+        pid_is_dead_fn=pid_is_dead_fn, now=now)
+    checker = pid_is_dead_fn if pid_is_dead_fn is not None else pid_is_dead
+    killer = kill_fn if kill_fn is not None else _default_kill
+    results = []
+    for row in rows:
+        result = dict(row)
+        result["executed"] = False
+        result["verified_dead"] = None
+        result["audit_before"] = False
+        result["audit_after"] = False
+        result["kill_error"] = None
+        if row.get("action") != "kill":
+            result["outcome"] = "listed"
+            results.append(result)
+            continue
+        pid = row.get("pid")
+        if pid is None or not checker(pid):
+            result["outcome"] = "skipped_pid_alive_at_execution"
+            result["verified_dead"] = False
+            results.append(result)
+            continue
+        detail = ("worker_id=%s host=%s pid=%s reason=%s"
+                  % (row.get("worker_id"), row.get("host"), pid,
+                     row.get("reason")))
+        try:
+            if audit_fn is not None:
+                audit_fn("worker_pid_kill", detail)
+            else:
+                _default_audit(db_path, "worker_pid_kill", detail)
+            result["audit_before"] = True
+        except Exception:
+            result["outcome"] = "audit_failed"
+            results.append(result)
+            continue
+        result["executed"] = True
+        try:
+            killer(pid)
+        except ProcessLookupError:
+            pass
+        except Exception as exc:
+            result["kill_error"] = str(exc)
+        verified = bool(checker(pid))
+        result["verified_dead"] = verified
+        result["outcome"] = (
+            "killed" if verified else "kill_failed_pid_alive")
+        after_detail = detail + " outcome=%s verified_dead=%s" % (
+            result["outcome"], verified)
+        try:
+            if audit_fn is not None:
+                audit_fn("worker_pid_kill_verified", after_detail)
+            else:
+                _default_audit(
+                    db_path, "worker_pid_kill_verified", after_detail)
+            result["audit_after"] = True
+        except Exception:
+            result["audit_after"] = False
+        results.append(result)
+    return results
+
+
 def register_worker(db_path, worker_id, executor=None, capacity=1,
                     running=None, host=None, pid=None, address=None):
     """注册（或刷新）一个 worker：不存在 INSERT，存在 UPDATE。