efi-kernel/内核/状态.py

"""驱动状态: 落盘快照 (运行.efi.json) + 状态机 + 断电收尸判定.

[两个真相, 谁说了算 (最容易搞混的地方)]
    活的真相 = PG 的 driver_state 表 (内核的内存; 判活 / 调度 / 级联都看它).
    落盘快照 = 驱动文件夹里的 运行.efi.json (**不是状态源**). 它的用处只有三个:
        1. 文件夹自包含 -- 拷走一个驱动目录, 状态跟着走;
        2. PG 不在 / 断电之后, 离线也能看出它上次什么状态;
        3. 给人看 (cat 一下就知道上次谁在跑).
    **判活一律回 /proc 复核**, 绝不拿快照当依据 (快照可能是几小时前写的).

[为什么快照要原子写]
    直接 open(w) 写一半断电 = 留个半截 json, 下次读出来就是"损坏".
    所以先写 运行.efi.json.tmp, 再 os.replace 换名 -- 同目录 rename 是原子的,
    要么是旧内容, 要么是完整新内容, 不会出现半截 (踩过: 快照损坏被当成"没状态", 白丢现场).

[谁写快照]
    内核 (扫描 / 启动 / 停止 / 收尸之后各刷一次). 驱动作者手改它没有意义, 下次刷新就覆盖.

[状态机 (设计 01 第 4 节)]
              ┌────────► stopped ◄────────┐
              │             │             │
     pid 不在 │        start│             │stop 成功
              │             ▼             │
    crashed ◄─┘        starting ──► running ─┤
    exited  ◄─ pid 被复用    │         │    │
                          失败│    崩了 │    │
                             ▼         ▼    │
                          failed   crashed ─┘
"""

from __future__ import annotations

import json
import os
from datetime import datetime
from pathlib import Path
from typing import Any, cast

import 进程

# ─────────────────────────────── 常量 ───────────────────────────────

配置名 = "配置.efi.json"  # 驱动作者写 (内核只读, 用来算 config_hash)
快照名 = "运行.efi.json"  # 内核写 (给人看 / 离线读)
日志目录名 = "logs"  # 驱动 stdout/stderr 落这儿 (内核重定向, 不解析内容)
快照版本 = 1  # 快照格式版本 (efi 字段)

# 状态取值: 存 PG 用英文 (值稳定, 不受界面措辞影响), 给老板看的时候过 显示状态() 转中文
停止 = "stopped"  # 没在跑, 正常态
启动中 = "starting"  # 已 spawn, 还没确认存活
运行 = "running"  # /proc/<pid> 在, 且 cmdline 校验通过
已退出 = "exited"  # 进程没了 (正常/异常退出), 退出码在 exit_code
崩了 = "crashed"  # 断电 / 内核重启后的判定 (旧状态说在跑, /proc 里没有)
失败 = "failed"  # 拉起就失败 (入口不存在 / 解释器缺 / 立刻非零退出)
无效 = "invalid"  # 配置不合法, 内核拒载 (原因在 last_error), 不影响别的驱动

# 英文状态 -> 中文显示 (界面必须中文; 存库保持英文)
中文表: dict[str, str] = {
    停止: "停止",
    启动中: "启动中",
    运行: "运行",
    已退出: "已退出",
    崩了: "崩了",
    失败: "失败",
    无效: "无效",
}


def 显示状态(值: Any) -> str:
    """把库里的英文状态转成中文给人看; 不认识的取值原样返回 (别吞掉陌生状态)."""
    文本 = str(值 or 停止)
    return 中文表.get(文本, 文本)


def 现在文本() -> str:
    """当前时间, ISO8601 带时区偏移 (快照的 updated_at 用; 老板要看得出是哪一刻写的)."""
    return datetime.now().astimezone().isoformat(timespec="seconds")


def _时刻文本(值: Any) -> str | None:
    """把 PG 里取出来的 timestamptz 转成 ISO 字符串; None / 空 返回 None.

    为什么要在这一层转:
        json 里存不了 datetime 对象; 快照是给人看和离线读的, ISO 文本最省事.
    """
    if 值 is None:
        return None
    if isinstance(值, datetime):
        return 值.astimezone().isoformat(timespec="seconds")
    return str(值)


# ─────────────────────────────── 路径 ───────────────────────────────


def 驱动根(驱动: dict[str, Any]) -> Path:
    """驱动文件夹绝对路径 (drivers.dir)."""
    return Path(str(驱动.get("dir") or ""))


def 入口路径(驱动: dict[str, Any]) -> Path:
    """入口文件绝对路径 = 驱动根 / entry.

    用途: 传给 进程.判活 / 进程.停止 做 cmdline 校验 -- 这是"防 pid 复用误杀"的关键,
    所以必须是**和启动时同一个**路径 (启动用 驱动根/entry, 这里也是).
    """
    return 驱动根(驱动) / str(驱动.get("entry") or "")


def 日志目录(驱动: dict[str, Any]) -> Path:
    """驱动日志目录 (配置里的 logs 字段, 默认 logs/)."""
    名 = str(驱动.get("logs") or 日志目录名)
    return 驱动根(驱动) / 名


def 日志路径(驱动: dict[str, Any]) -> Path:
    """驱动日志文件: <驱动根>/logs/<驱动名>.log (stdout/stderr 都进这一个)."""
    return 日志目录(驱动) / f"{驱动.get('name')}.log"


# ─────────────────────────────── 快照读写 ───────────────────────────────


def 读快照(目录: Path) -> dict[str, Any] | None:
    """读一份 运行.efi.json.

    参数:
        目录: 驱动文件夹绝对路径.
    返回:
        快照字典; 文件不在 / 不是合法 JSON / 顶层不是对象 一律返回 None.

    为什么把"损坏"也归成 None:
        调用方要的行为是同一个: 当成"没有快照", 用 PG 重建一份.
        损坏不是致命错误 (PG 才是真相), 所以这里不抛异常,也不在外面分两种情况处理.
    """
    try:
        原文 = (目录 / 快照名).read_text(encoding="utf-8")
    except OSError:
        return None
    try:
        数据: Any = json.loads(原文)
    except json.JSONDecodeError:
        return None
    # isinstance 窄化出来的只是 dict[Unknown, Unknown]; cast 一次, 免得检查器判"类型部分未知"
    return cast(dict[str, Any], 数据) if isinstance(数据, dict) else None


def 写快照(目录: Path, 数据: dict[str, Any]) -> None:
    """整份重写 运行.efi.json (原子替换).

    参数:
        目录: 驱动文件夹绝对路径.
        数据: 组装快照() 出来的完整字典.
    返回:
        无.

    步骤:
        1. 写 <快照名>.tmp (同目录, 保证 os.replace 是同一个文件系统内的 rename);
        2. os.replace 换名覆盖正式文件.
    其它:
        目录不存在 (驱动被删了) 时抛 OSError -- 该报错就报错, 不静默吞.
    """
    临时 = 目录 / (快照名 + ".tmp")
    文本 = json.dumps(数据, ensure_ascii=False, indent=2) + "\n"
    临时.write_text(文本, encoding="utf-8")
    os.replace(临时, 目录 / 快照名)


def 组装快照(驱动: dict[str, Any], 状态行: dict[str, Any] | None, 内核版本: str) -> dict[str, Any]:
    """按"注册表 + 运行时状态"拼一份完整的快照 (键表见设计 01 第 4 节).

    参数:
        驱动: drivers 表的一行 (dir/entry/config_hash/entry_hash 从这里来).
        状态行: driver_state 表的一行; None = 还没有状态记录 (当"停止"处理).
        内核版本: 写进 kernel 字段 (哪一版内核写的, 排查时有用).
    返回:
        可以直接 json.dumps 的字典.

    为什么同时写 config_hash 和 boot_hash:
        config_hash = **当前**配置的指纹 (从注册表来), boot_hash = **起进程那一刻**的指纹.
        两个一比就知道"配置改过了但还没重启" (设计里那句"配置已改, 待重启"),
        而且快照自包含 -- 离线也能自己判, 不用回 PG 查.
    """
    行: dict[str, Any] = 状态行 or {}
    pid值 = 行.get("pid")
    return {
        "efi": 快照版本,
        "name": str(驱动.get("name") or ""),
        "state": str(行.get("state") or 停止),
        "pid": int(pid值) if isinstance(pid值, int) else None,
        "pgid": 行.get("pgid"),
        "started_at": _时刻文本(行.get("started_at")),
        "stopped_at": _时刻文本(行.get("stopped_at")),
        "exit_code": 行.get("exit_code"),
        "restarts": int(行.get("restarts") or 0),
        "config_hash": 驱动.get("config_hash"),
        "boot_hash": 行.get("boot_hash"),
        "entry_hash": 驱动.get("entry_hash"),
        "list_version": 行.get("list_version"),
        "kernel": 内核版本,
        "last_error": 行.get("last_error"),
        "updated_at": 现在文本(),
    }


def 待重启(驱动: dict[str, Any], 状态行: dict[str, Any] | None) -> bool:
    """配置改过了但进程还在用旧配置跑 -> True (列表里显示"待重启").

    判据: 进程确实在跑 (有 boot_hash), 且 boot_hash != 当前 config_hash.
    boot_hash 为空 = 从没起过 (或没记录), 不算"待重启" (那是"没跑过").
    """
    行: dict[str, Any] = 状态行 or {}
    if str(行.get("state") or 停止) != 运行:
        return False
    起时 = 行.get("boot_hash")
    当前 = 驱动.get("config_hash")
    if not 起时 or not 当前:
        return False
    return str(起时) != str(当前)


# ─────────────────────────────── 收尸判定 ───────────────────────────────


def 复核(驱动: dict[str, Any], 状态行: dict[str, Any] | None) -> tuple[str, str]:
    """拿 PG 里的旧状态 + /proc 的实际情况, 判出这个驱动**现在**到底什么状态.

    什么时候跑: 内核每次 扫描 / boot 都先跑一遍 (本机 22:30 断电, 这是必做项).
    参数:
        驱动: drivers 表一行 (要它的 dir/entry 去 /proc 里核对).
        状态行: driver_state 表一行; None = 没记录.
    返回:
        (状态, 原因). 原因只在"需要解释"时有内容 (要落 last_error / 写事件), 比如
        "上次在跑, 现在 /proc 里没了 (断电 / 被杀)".

    判定表 (设计 02 第 5 节):
        旧状态是"在跑"(running/starting):
            进程在且 cmdline 匹配 -> 运行 (认领, 不重起)
            进程不在             -> 崩了 (断电 / 被杀)
            进程在但 cmdline 不对 -> 已退出 (pid 被复用: 清掉, **不杀**)
            僵尸                 -> 崩了 (已经死了, 只是没被 wait 回收)
        旧状态是"没在跑"(停止/已退出/崩了/失败/无效):
            进程在且匹配 -> 运行 (认领回来: 有人绕过内核对它做了什么)
            否则         -> 维持旧状态 (不动)
    """
    旧 = str((状态行 or {}).get("state") or 停止)
    pid值 = (状态行 or {}).get("pid")
    pid = int(pid值) if isinstance(pid值, int) else None
    入口 = 入口路径(驱动)
    活 = 进程.判活(pid, 入口)

    if 旧 not in (运行, 启动中):
        if 活 == 进程.运行中:
            return 运行, "状态里没写它在跑, 但 /proc 里确实在 -- 认领回来"
        return 旧, ""

    if 活 == 进程.运行中:
        return 运行, ""
    if 活 == 进程.僵尸:
        return 崩了, f"进程 {pid} 已成僵尸 (已死, 父进程没回收)"
    if 活 == 进程.被复用:
        return 已退出, f"pid {pid} 已被系统分给别人 (不认领, 也不杀)"
    if str(驱动.get("mode") or "resident") == "oneshot":
        return 已退出, "oneshot 跑完就退 (正常, 不用重起)"
    return 崩了, f"状态里写着在跑, 但 /proc/{pid} 没了 (断电 / 被杀)"


def 该拉起(驱动: dict[str, Any], 状态行: dict[str, Any] | None, 判定: str) -> str:
    """收尸之后判断"该不该顺手把它拉起来", 返回理由 ("" = 不拉).

    参数:
        驱动: drivers 表一行 (autostart / restart 两个策略字段).
        状态行: driver_state 表一行 (要 exit_code 来区分"断电"和"自己崩").
        判定: 复核() 给出的新状态.
    返回:
        该拉 -> 一句人读的理由 (写进 events); 不该拉 -> 空串.

    规则 (设计 02 第 5 节 + 第 2 节):
        * 断电 / 被杀留下的 crashed: exit_code 是空的 -> 看 autostart (开机自启的才拉);
        * 进程自己崩的 (exit_code 有值): 看 restart == "on-failure";
        * 其它状态 (停止 / 已退出 / 无效 / 失败): 一律不自动拉.
          "失败"不自动重试是故意的 -- 配置错了就别反复撞墙 (省得把日志刷满).
    """
    行: dict[str, Any] = 状态行 or {}
    if 判定 not in (崩了, 失败):
        return ""
    if 判定 == 失败:
        return ""
    退出码 = 行.get("exit_code")
    if 退出码 is None:
        if bool(驱动.get("autostart", False)):
            return "断电/被杀后留下的进程, 而且 autostart=true -- 拉起来"
        return ""
    if str(驱动.get("restart") or "no") == "on-failure":
        return f"上次非正常退出 (退出码 {退出码}), restart=on-failure -- 拉起来"
    return ""