mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-11-01 06:47:29 +00:00
feat(webui+guard): 内核重启/关机按钮，指令走心跳复用既有协议
## 需求
WebUI 加关机与重启按钮，内核提供对应内部接口。有些状态变更只能靠重启
生效（换 LLM 源、换主密钥），没有这条通道时用户只能 ssh 上去敲 systemctl。
## 设计：复用心跳，不新开通道（用户指出的方向）
guard 与 worker 之间已经有一条现成的、每 5s 一次的单向通报（心跳文件，
内容原本是 `t=<unix>`，只用来刷 mtime）。用户点按钮本质就是"我要你在
下一次心跳时知道这件事"——再开 unix socket / 信号 / 临时文件，只会多出
一套要防重、防丢、防竞态的机制。
WebUI 点按钮
→ POST /api/v1/kernel/{restart,shutdown}
→ sdk.KernelControlAPI（只投递，立即返回）
→ 下一次心跳把 `cmd=restart|shutdown reason=...` 写进 <data>/heartbeat
→ 重启：worker 清理 → exit 42 → guard 重拉（**既有机制，零改动**）
关机：guard 在 workerCleanExit 分支读到 cmd=shutdown
→ systemctl stop → 不再拉起
★ 关机为什么必须由 guard 做：
unit 是 Restart=always，guard 自己 return 会被 systemd 立刻拉起
（用户点了"关机"、几秒后服务又活了）。必须显式 systemctl stop。
而不让 worker 自己停：它是被守护方，且**只有 guard 知道这次该不该重拉**。
★ 指令必须一次性消费：guard 读完立即原子重写文件清除 cmd 行，
否则每次心跳都读到同一条 ⇒ 无限重启/关机。worker 侧还要检查
"指令写过但文件里没了" ⇒ 已被消费 ⇒ 清本地 pending，否则又写回去。
## 顺带修的真缺陷：心跳文件无限增长
生产实测 **14.6MB**（纯 O_APPEND，约 70 天）。心跳的唯一职责是刷 mtime，
历史行没有任何读者 ⇒ 每次 consume 要整读 14.6MB（实测 27ms）。
改为 O_TRUNC 滚动写：只保留当前一行。真机验证 15,322,164 → 13 字节。
consume 也加了尺寸守卫（大文件只读尾部 8KB）。
## 安全：能力分级 + 白名单（两道）
1. proc 桥的 method 能力表：新增 `CapKernelControl` 一档。
★ 判据 `TestCapability_AllMethodsClassified` 当场报出
"未登记 ⇒ 会按 CapCore 放行，绕过权限检查" —— 那意味着任何插件都能关机。
2. 内置插件走 `SDKConfig.KernelControl`，第一版我写成"有就发"，
等于全部内置插件都能关机。改为按名字白名单（目前只有 webui）。
## 判据（7 条）
心跳：解析 / 消费后必须清除 / worker 不得写回已消费指令 / 未消费时必须续写
能力：只有白名单插件拿得到 / 未装配时返回 nil 不 panic / SDK nil 安全
心跳文件：不得无限增长 / 大文件只读尾部且读后收敛
变异 6/6 咬住。全量 53 包 ok。
## UI
导航底栏两个图标按钮（与主题/语言/退出并列）：重启有确认框 + 轮询等它
回来并自动刷新；关机有**二次确认且写清后果**（必须人工 systemctl start）。
两者都先把响应写回浏览器并 Flush，再让内核退出 —— 否则用户看到的是
connection reset，无法区分"正在重启"和"真出故障了"。
This commit is contained in:
JianFeeeee
@ -227,7 +227,30 @@ func startHeartbeat(dataDir string, ctx context.Context) func() {
|
||||
// 心跳：每 5s 触碰 <data>/heartbeat，guard 据此判定工作进程是否存活/卡死
|
||||
hbPath := filepath.Join(dataDir, "heartbeat")
|
||||
writeHB := func() {
|
||||
if f, err := os.OpenFile(hbPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644); err == nil {
|
||||
// ★ 若有待发的控制指令，**整文件重写**（保留 t= 行 + cmd= 行）。
|
||||
//
|
||||
// 为什么不能继续 append：guard 消费指令后会重写文件清除 cmd 行；
|
||||
// 若这里仍 append，历史 cmd 行会越积越多（每次读到的还是第一条）。
|
||||
// 无指令时保持原来的 append 语义（历史行为不变，只刷 mtime）。
|
||||
if content := nextHeartbeatWrite(hbPath, time.Now().Unix()); content != "" {
|
||||
if err := os.WriteFile(hbPath, []byte(content), 0644); err == nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
// ★ 心跳文件必须**滚动**，不能无限 append（2026-10-17 实测：14.6MB）。
|
||||
//
|
||||
// heartbeat 的唯一职责是让 guard 看到 mtime 变新（见下面的注释）。
|
||||
// 历史行**没有任何读者** —— 唯一读内容的是 consumeHeartbeatCmd，
|
||||
// 而它只关心 cmd= 行。所以累加历史行纯属浪费：
|
||||
// · 每 5s 12 字节 ⇒ 约 70 天长到 14.6MB（生产实测就是这个量级）
|
||||
// · consumeHeartbeatCmd 要 ReadFile 整个文件 ⇒ 白白读 14.6MB
|
||||
// · 长期跑下去还会一直涨（磁盘与读取成本都随天数线性增长）
|
||||
//
|
||||
// 改成截断重写：只保留**当前**这一行。mtime 语义完全不变
|
||||
// （文件仍在每次心跳被写），而体积恒定在十几字节。
|
||||
//
|
||||
// 用 O_TRUNC 而不是 O_APPEND：内容不再需要累积。
|
||||
if f, err := os.OpenFile(hbPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0644); err == nil {
|
||||
fmt.Fprintf(f, "t=%d\n", time.Now().Unix())
|
||||
f.Close()
|
||||
}
|
||||
|
||||
@ -256,7 +256,25 @@ func runGuard(dataDir string) {
|
||||
|
||||
switch outcome {
|
||||
case workerCleanExit:
|
||||
// ★ 先看 worker 有没有留下控制指令（用户点了关机）。
|
||||
//
|
||||
// 这条指令来自心跳文件：worker 在退出前最后几次心跳里写了
|
||||
// `cmd=shutdown`。此时 worker 已经退干净，由 guard 决定
|
||||
// "这次退出去该不该被 systemd 重新拉起"。
|
||||
//
|
||||
// 为什么要读心跳而不是让 worker 直接 systemctl stop：
|
||||
// · worker 是被守护方，自己停服务等于在父进程眼皮底下自杀
|
||||
// · 只有 guard 知道防护上下文（unit 名、当前是否 failback 中）
|
||||
// · 决策集中在一处，不散在两个进程里
|
||||
if c := consumeHeartbeatCmd(hbPath); c.OK && c.Cmd == hbCmdShutdown {
|
||||
log.Printf("[guard] worker 已退出且留下关机指令（%s）—— 执行 systemctl stop，不再拉起",
|
||||
c.Reason)
|
||||
stopOwnUnit(ownUnit)
|
||||
return
|
||||
}
|
||||
log.Printf("[guard] worker exited cleanly, guard exiting")
|
||||
// ★ 注意：这里 `return` 后 guard 退出，systemd（Restart=always）
|
||||
// 会重新拉起整个 unit —— 这正是"重启"的既有语义，保持不动。
|
||||
return
|
||||
case workerRestartRequested:
|
||||
log.Printf("[guard] worker requested restart, respawning")
|
||||
|
||||
253
cmd/homed/heartbeat_cmd.go
Normal file
253
cmd/homed/heartbeat_cmd.go
Normal file
@ -0,0 +1,253 @@
|
||||
package main
|
||||
|
||||
// 心跳里的**控制指令**：worker → guard 的唯一意图通道。
|
||||
//
|
||||
// ## 为什么走心跳，而不是另开一条通道
|
||||
//
|
||||
// guard 与 worker 之间已经有一条现成的、每 5s 一次的单向通报（心跳文件）。
|
||||
// 用户点"重启/关机"本质就是"我要你在下一次心跳时知道这件事"——
|
||||
// 开第二条通道（unix socket / 信号 / 临时文件）只会多一套要防重、防丢、
|
||||
// 防竞态的机制，而这条已经存在且有 mtime 存活语义。
|
||||
//
|
||||
// ## 格式演进（必须向后兼容）
|
||||
//
|
||||
// 原格式每行 `t=<unix>\n`，**只用来刷 mtime**，内容从没有人读。
|
||||
// 现在追加一行指令：
|
||||
//
|
||||
// t=<unix>
|
||||
// cmd=restart reason=<理由>
|
||||
//
|
||||
// ★ 兼容性：老 guard 只 stat mtime、不读内容 ⇒ 追加行完全无害。
|
||||
// 新 guard 读不到 cmd 行 ⇒ 行为与改动前逐字一致。
|
||||
//
|
||||
// ## ★ 指令为何必须**一次性消费**
|
||||
//
|
||||
// 如果 guard 读到 cmd 就执行、却不清理，那么下一次心跳（或下一次重启后的
|
||||
// 第一个心跳）仍看得到同一条 cmd ⇒ 无限重启。
|
||||
// ⇒ 消费方（guard）读完立即**原子地**把 cmd 行抹掉。
|
||||
// 用 O_TRUNC 重写整个文件（只保留 t= 行）——文件只有一个写者（worker 的
|
||||
// ticker）和一个读者（guard），且两者不会同时写（worker 退出后才轮到 guard 读）。
|
||||
//
|
||||
// ## 为什么关机要 guard 来做，而不是 worker 直接调 systemctl
|
||||
//
|
||||
// worker 是被守护方。它自己调 `systemctl stop` 等于在父进程眼皮底下自杀，
|
||||
// 且 systemd 可能把它判成"异常退出"而先计入一次失败。
|
||||
// 更根本的是：**只有 guard 知道这次退出该不该重拉**。让决策集中在一个地方。
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"log"
|
||||
"os"
|
||||
"os/exec"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// 心跳指令取值。
|
||||
const (
|
||||
hbCmdRestart = "restart"
|
||||
hbCmdShutdown = "shutdown"
|
||||
)
|
||||
|
||||
// hbCommand 是从心跳文件里读出的指令。
|
||||
type hbCommand struct {
|
||||
Cmd string // "" | "restart" | "shutdown"
|
||||
Reason string
|
||||
// OK 表示确实读到了合法指令（Cmd 非空）。
|
||||
OK bool
|
||||
}
|
||||
|
||||
// parseHeartbeatCmd 解析心跳文件内容，取出指令。
|
||||
//
|
||||
// 容错：任何畸形行都忽略（心跳文件是"尽力而为"的通道，不该因为
|
||||
// 一个坏行就让 guard 崩掉）。
|
||||
func parseHeartbeatCmd(content string) hbCommand {
|
||||
sc := bufio.NewScanner(strings.NewReader(content))
|
||||
for sc.Scan() {
|
||||
line := strings.TrimSpace(sc.Text())
|
||||
if !strings.HasPrefix(line, "cmd=") {
|
||||
continue
|
||||
}
|
||||
rest := strings.TrimPrefix(line, "cmd=")
|
||||
var cmd, reason string
|
||||
if i := strings.Index(rest, " reason="); i >= 0 {
|
||||
cmd = strings.TrimSpace(rest[:i])
|
||||
reason = strings.TrimSpace(rest[i+len(" reason="):])
|
||||
} else {
|
||||
cmd = strings.TrimSpace(rest)
|
||||
}
|
||||
switch cmd {
|
||||
case hbCmdRestart, hbCmdShutdown:
|
||||
return hbCommand{Cmd: cmd, Reason: reason, OK: true}
|
||||
}
|
||||
}
|
||||
return hbCommand{}
|
||||
}
|
||||
|
||||
// consumeHeartbeatCmd 读出指令并**立即清除**它（一次性语义）。
|
||||
//
|
||||
// 清除方式：重写文件为只含当前时间戳的心跳行，这样 mtime 仍是新的
|
||||
// （不会让 guard 误判 worker 卡死），而 cmd 行消失。
|
||||
func consumeHeartbeatCmd(path string) hbCommand {
|
||||
// ★ 先看大小再决定读不读（2026-10-17 修正）。
|
||||
//
|
||||
// 修复前：无条件 ReadFile(path) —— 生产上那个文件长到 14.6MB，
|
||||
// 每次 worker 退出都要整读一遍（实测 27ms，纯属浪费）。
|
||||
//
|
||||
// 滚动写入修好后文件只有十几到几十字节，但这里仍保留尺寸守卫：
|
||||
// 老部署上那个巨型文件还在，第一次读完就会被清除，不该为此再读几次。
|
||||
const maxRead = 8 << 10 // 8KB 足够容纳 t= 行 + cmd= 行
|
||||
fi, err := os.Stat(path)
|
||||
if err != nil {
|
||||
return hbCommand{}
|
||||
}
|
||||
if fi.Size() > maxRead {
|
||||
// 巨型历史文件：指令若有，一定在**末尾**（worker 最后几次心跳）。
|
||||
// 只读尾部 8KB，避免整读。
|
||||
return consumeTail(path, maxRead)
|
||||
}
|
||||
return consumeFrom(path)
|
||||
}
|
||||
|
||||
// consumeTail 只读文件尾部（老部署的巨型心跳文件）。
|
||||
func consumeTail(path string, n int64) hbCommand {
|
||||
f, err := os.Open(path)
|
||||
if err != nil {
|
||||
return hbCommand{}
|
||||
}
|
||||
defer f.Close()
|
||||
fi, _ := f.Stat()
|
||||
off := fi.Size() - n
|
||||
if off < 0 {
|
||||
off = 0
|
||||
}
|
||||
buf := make([]byte, n)
|
||||
read, _ := f.ReadAt(buf, off)
|
||||
c := parseHeartbeatCmd(string(buf[:read]))
|
||||
if c.OK {
|
||||
clearHeartbeatCmd(path)
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
||||
// consumeFrom 读小文件（正常路径）。
|
||||
func consumeFrom(path string) hbCommand {
|
||||
b, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return hbCommand{}
|
||||
}
|
||||
c := parseHeartbeatCmd(string(b))
|
||||
if c.OK {
|
||||
clearHeartbeatCmd(path)
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
||||
// clearHeartbeatCmd 把心跳文件重写为只含当前时间戳的一行。
|
||||
//
|
||||
// 原子（写临时文件再 rename）：避免 guard 读到半截内容。
|
||||
// 同时把巨型历史文件**一次性收敛**成十几字节 —— 老部署升级后第一次
|
||||
// 消费指令就顺带清理掉了。
|
||||
func clearHeartbeatCmd(path string) {
|
||||
tmp := path + ".cmdtmp"
|
||||
now := []byte("t=" + strconv.FormatInt(time.Now().Unix(), 10) + "\n")
|
||||
if err := os.WriteFile(tmp, now, 0644); err != nil {
|
||||
return
|
||||
}
|
||||
if err := os.Rename(tmp, path); err != nil {
|
||||
os.Remove(tmp)
|
||||
}
|
||||
}
|
||||
|
||||
// ---- worker 侧的待发指令队列 ----
|
||||
|
||||
// pendingCmd 是"用户点了按钮，但 guard 还没消费"的指令。
|
||||
//
|
||||
// 用 mutex + 单槽（不是队列）：重启/关机都是**目标态**，连点两下不该
|
||||
// 排队两次。后来的请求覆盖前者即可（都指向"要重启"这件事）。
|
||||
var (
|
||||
pendingCmdMu sync.Mutex
|
||||
pendingCmd hbCommand
|
||||
// pendingWritten 标记这条指令是否**已经写进过心跳文件**。
|
||||
//
|
||||
// ★ 为什么需要它（否则指令永远清不掉）：
|
||||
// guard 与 worker 是**两个进程**。guard 消费后把文件重写成纯 t= 行，
|
||||
// 但 worker 内存里的 pendingCmd 还是满的 ⇒ 下一次 tick 又写回去
|
||||
// ⇒ guard 又读到 ⇒ **无限重启/关机**。
|
||||
// 真正的"已消费"信号只能从**文件**上读：写过了、而文件里的 cmd 行没了
|
||||
// ⇒ 说明被 guard 吃了 ⇒ 清本地 pending。
|
||||
pendingWritten bool
|
||||
)
|
||||
|
||||
// requestHeartbeatCmd 由一个外部调用（WebUI 经内核接口）投递指令。
|
||||
//
|
||||
// 立即返回：真正的写入发生在下一次心跳（≤5s）。
|
||||
func requestHeartbeatCmd(cmd, reason string) {
|
||||
pendingCmdMu.Lock()
|
||||
pendingCmd = hbCommand{Cmd: cmd, Reason: reason, OK: true}
|
||||
pendingWritten = false
|
||||
pendingCmdMu.Unlock()
|
||||
}
|
||||
|
||||
// nextHeartbeatWrite 决定这一次心跳该写什么内容。
|
||||
//
|
||||
// 返回 (content, writeFull)：
|
||||
// - content 非空 ⇒ 用**整文件重写**（保留 t= 行 + cmd= 行）
|
||||
// - content 为空 ⇒ 走原来的 append 语义（只刷 mtime）
|
||||
//
|
||||
// 消费检测也在这里：若指令写过、而文件里已没有 cmd 行，说明 guard 已消费，
|
||||
// 清掉本地 pending，本次回到 append。
|
||||
func nextHeartbeatWrite(hbPath string, now int64) string {
|
||||
pendingCmdMu.Lock()
|
||||
defer pendingCmdMu.Unlock()
|
||||
|
||||
if pendingCmd.Cmd == "" {
|
||||
return ""
|
||||
}
|
||||
if pendingWritten {
|
||||
// 检查 guard 是否已消费（guard 消费 = 把文件重写成纯 t= 行）
|
||||
if b, err := os.ReadFile(hbPath); err == nil && !strings.Contains(string(b), "cmd=") {
|
||||
log.Printf("[homed] 控制指令 %s 已被 guard 消费，清除本地待发状态", pendingCmd.Cmd)
|
||||
pendingCmd = hbCommand{}
|
||||
pendingWritten = false
|
||||
return ""
|
||||
}
|
||||
}
|
||||
pendingWritten = true
|
||||
return "t=" + strconv.FormatInt(now, 10) + "\ncmd=" + pendingCmd.Cmd + " reason=" + pendingCmd.Reason + "\n"
|
||||
}
|
||||
|
||||
// stopOwnUnit 让 systemd 停下来（不再自动拉起）。
|
||||
//
|
||||
// ★ 为什么必须显式 stop，不能只是 return：
|
||||
//
|
||||
// 生产 unit 是 `Restart=always`。guard 干净退出后 systemd 会立刻拉起整个
|
||||
// unit —— 用户点了"关机"，几秒后服务又活了，看起来像按钮没生效。
|
||||
//
|
||||
// 用 `systemctl stop <unit>`：
|
||||
//
|
||||
// · 不依赖 unit 里配 RestartPreventExitStatus（那种做法要求 drop-in 与
|
||||
// 主 unit 一起交付，部署时容易漏）
|
||||
// · 走正常停机序列（SIGTERM → TimeoutStopSec），与 guard 自己收尾不冲突
|
||||
//
|
||||
// unit 名为空（非 systemd 环境 / 探测失败）时只记日志：
|
||||
// 那种情况下 guard 退出后进程就真的没了，无需额外动作。
|
||||
func stopOwnUnit(unit string) {
|
||||
if unit == "" {
|
||||
log.Printf("[guard] 非 systemd 环境（unit 未知）：guard 退出即可，systemd 不会拉起")
|
||||
return
|
||||
}
|
||||
log.Printf("[guard] 执行 systemctl stop %s", unit)
|
||||
// 同步执行：guard 此时已经没有别的事要做，且必须在**自己退出之前**
|
||||
// 让 systemd 收到停止请求。异步会与 return 竞争 ——
|
||||
// guard 退了、systemctl 还没跑，systemd 可能已经判定 unit 需要重启。
|
||||
cmd := exec.Command("systemctl", "stop", unit)
|
||||
cmd.Stdout = os.Stderr
|
||||
cmd.Stderr = os.Stderr
|
||||
if err := cmd.Run(); err != nil {
|
||||
log.Printf("[guard] systemctl stop %s 失败: %v（guard 仍退出；"+
|
||||
"若 unit 是 Restart=always 可能被重新拉起）", unit, err)
|
||||
}
|
||||
}
|
||||
208
cmd/homed/heartbeat_cmd_test.go
Normal file
208
cmd/homed/heartbeat_cmd_test.go
Normal file
@ -0,0 +1,208 @@
|
||||
package main
|
||||
|
||||
// 心跳控制指令的判据（2026-10-17）。
|
||||
//
|
||||
// ## 为什么这两条判据必须存在
|
||||
//
|
||||
// 指令通道有三个**静默失效**点，每一个都表现为"按钮点了没反应"：
|
||||
//
|
||||
// 1. 解析不到 → guard 不知道用户想关机 → 照常重拉（用户以为没生效）
|
||||
// 2. **消费后不清除** → guard 每次心跳都读到同一条 → 无限重启/关机
|
||||
// 3. 清除了但 worker 内存里还留着 → 下一次心跳又写回去 → 同上
|
||||
//
|
||||
// 第 2、3 点是同一个 bug 的两面，而且**本机测不出来**（只有真跑才会
|
||||
// 陷入重启循环）。所以用判据把它们钉住。
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// ★★★ 判据：能解析出合法指令，畸形行被忽略。
|
||||
func TestParseHeartbeatCmd(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
in string
|
||||
want hbCommand
|
||||
}{
|
||||
{"关机", "t=100\ncmd=shutdown reason=用户点击\n", hbCommand{Cmd: "shutdown", Reason: "用户点击", OK: true}},
|
||||
{"重启", "t=100\ncmd=restart reason=升级配置\n", hbCommand{Cmd: "restart", Reason: "升级配置", OK: true}},
|
||||
{"无指令", "t=100\nt=101\n", hbCommand{}},
|
||||
{"未知指令", "t=100\ncmd=explode\n", hbCommand{}},
|
||||
{"畸形行", "garbage\ncmd=\n", hbCommand{}},
|
||||
{"无 reason", "cmd=shutdown\n", hbCommand{Cmd: "shutdown", OK: true}},
|
||||
{"空内容", "", hbCommand{}},
|
||||
}
|
||||
for _, c := range cases {
|
||||
got := parseHeartbeatCmd(c.in)
|
||||
if got.Cmd != c.want.Cmd || got.OK != c.want.OK {
|
||||
t.Errorf("%s: 得到 %+v，应以 %+v", c.name, got, c.want)
|
||||
}
|
||||
if c.want.Reason != "" && got.Reason != c.want.Reason {
|
||||
t.Errorf("%s: reason 得到 %q，应 %q", c.name, got.Reason, c.want.Reason)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ★★★ 判据：consumeHeartbeatCmd 读完必须**清除** cmd 行（否则无限重启）。
|
||||
func TestConsumeHeartbeatCmdClearsCommand(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
p := filepath.Join(dir, "heartbeat")
|
||||
if err := os.WriteFile(p, []byte("t=100\ncmd=shutdown reason=x\n"), 0644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
c := consumeHeartbeatCmd(p)
|
||||
if !c.OK || c.Cmd != hbCmdShutdown {
|
||||
t.Fatalf("未读到关机指令: %+v", c)
|
||||
}
|
||||
|
||||
// 文件必须已被清除
|
||||
b, err := os.ReadFile(p)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if parseHeartbeatCmd(string(b)).OK {
|
||||
t.Fatalf("★ 消费后 cmd 行仍在（%q）—— guard 每次心跳都会读到同一条 ⇒ 无限关机", string(b))
|
||||
}
|
||||
// mtime 语义必须保留：文件里仍要有 t= 行，否则 guard 可能判 worker 卡死
|
||||
if len(b) == 0 {
|
||||
t.Fatal("★ 清空成空文件 —— 心跳文件的 mtime 语义被破坏")
|
||||
}
|
||||
|
||||
// 第二次消费必须读不到（幂等）
|
||||
if parseHeartbeatCmd(readAll(t, p)).OK {
|
||||
t.Fatal("★ 第二次仍能读到指令")
|
||||
}
|
||||
}
|
||||
|
||||
// ★★★ 判据：worker 侧写过的指令，在 guard 清除后**不该被再写回**。
|
||||
//
|
||||
// 这是最容易漏的一面：guard 与 worker 是两个进程，
|
||||
// guard 清的是**文件**，worker 内存里的 pendingCmd 还在。
|
||||
// 若 worker 不检查文件状态，下一次 tick 就把指令写回去 ⇒ 无限循环。
|
||||
func TestWorkerDoesNotRewriteConsumedCommand(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
p := filepath.Join(dir, "heartbeat")
|
||||
// 模拟：已经没有 cmd 行（guard 消费过）
|
||||
if err := os.WriteFile(p, []byte("t=200\n"), 0644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// 投递一条指令，并标记"已经写过一次"
|
||||
requestHeartbeatCmd(hbCmdShutdown, "test")
|
||||
// ★ 用 defer 而不是 t.Cleanup：Cleanup 里的手动 Lock 一旦前面的
|
||||
// t.Fatalf 提前返回就可能与未释放的锁叠加（静态分析也提示了这一点）。
|
||||
// defer 的语义更直白：本函数怎么退出都会解锁并复位。
|
||||
defer resetPendingCmdForTest()
|
||||
pendingCmdMu.Lock()
|
||||
pendingWritten = true
|
||||
pendingCmdMu.Unlock()
|
||||
|
||||
content := nextHeartbeatWrite(p, 300)
|
||||
if content != "" {
|
||||
t.Fatalf("★ guard 已消费（文件无 cmd 行），worker 却又写回指令：%q\n"+
|
||||
" 后果：无限关机/重启循环 —— 本机单测测不出，只有真跑才暴露", content)
|
||||
}
|
||||
}
|
||||
|
||||
// ★★★ 判据：指令未被消费时必须持续写在心跳里（不能写完一次就丢）。
|
||||
func TestWorkerKeepsWritingUntilConsumed(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
p := filepath.Join(dir, "heartbeat")
|
||||
os.WriteFile(p, []byte("t=100\ncmd=shutdown reason=test\n"), 0644)
|
||||
|
||||
requestHeartbeatCmd(hbCmdShutdown, "test")
|
||||
defer resetPendingCmdForTest()
|
||||
|
||||
// 第一次：文件里还有 cmd= ⇒ 继续写（还没被消费）
|
||||
c1 := nextHeartbeatWrite(p, 200)
|
||||
if c1 == "" {
|
||||
t.Fatal("★ 指令还在文件里（未消费）却不继续写 —— guard 可能错过它")
|
||||
}
|
||||
}
|
||||
|
||||
func readAll(t *testing.T, p string) string {
|
||||
t.Helper()
|
||||
b, err := os.ReadFile(p)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return string(b)
|
||||
}
|
||||
|
||||
// resetPendingCmdForTest 复位 pending 状态，保证测试之间不串。
|
||||
func resetPendingCmdForTest() {
|
||||
pendingCmdMu.Lock()
|
||||
pendingCmd = hbCommand{}
|
||||
pendingWritten = false
|
||||
pendingCmdMu.Unlock()
|
||||
}
|
||||
|
||||
// ★★★ 判据：心跳文件必须**滚动**，不能无限增长。
|
||||
//
|
||||
// 2026-10-17 生产实测：心跳文件长到 **14.6MB**（纯 O_APPEND，约 70 天）。
|
||||
// 心跳的唯一职责是刷 mtime，历史行没有任何读者 ⇒ 累加纯属浪费，
|
||||
// 且 consumeHeartbeatCmd 会被迫整读 14.6MB。
|
||||
func TestHeartbeatFileDoesNotGrowUnbounded(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
p := filepath.Join(dir, "heartbeat")
|
||||
|
||||
// 模拟 worker 的滚动写：连续写 200 次（每次都是"当前这一行"）
|
||||
for i := 0; i < 200; i++ {
|
||||
content := "t=" + itoa(int64(1000+i)) + "\n"
|
||||
if err := os.WriteFile(p, []byte(content), 0644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
fi, err := os.Stat(p)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// 200 次心跳后文件仍应是单行（约 10 字节），不是 200 行
|
||||
if fi.Size() > 64 {
|
||||
t.Fatalf("★ 心跳文件增长到 %d 字节 —— 必须是滚动写入（只保留当前行）\n"+
|
||||
" 生产实测这种累加会到 14.6MB", fi.Size())
|
||||
}
|
||||
lines := strings.Count(readAll(t, p), "\n")
|
||||
if lines != 1 {
|
||||
t.Fatalf("★ 心跳文件有 %d 行，应为 1 行", lines)
|
||||
}
|
||||
}
|
||||
|
||||
// ★★★ 判据：巨型历史心跳文件只读尾部（不整读），且读后收敛成小文件。
|
||||
func TestConsumeLargeHeartbeatReadsTailOnly(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
p := filepath.Join(dir, "heartbeat")
|
||||
|
||||
// 造一个 1MB 的历史文件，末尾追加指令
|
||||
var sb strings.Builder
|
||||
for i := 0; i < 100000; i++ {
|
||||
sb.WriteString("t=" + itoa(int64(i)) + "\n")
|
||||
}
|
||||
sb.WriteString("t=999999\ncmd=shutdown reason=bigfile\n")
|
||||
if err := os.WriteFile(p, []byte(sb.String()), 0644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
c := consumeHeartbeatCmd(p)
|
||||
if !c.OK || c.Cmd != hbCmdShutdown {
|
||||
t.Fatalf("★ 大文件里的指令没读到: %+v", c)
|
||||
}
|
||||
// 读后必须收敛（clearHeartbeatCmd 重写成单行）
|
||||
fi, _ := os.Stat(p)
|
||||
if fi.Size() > 64 {
|
||||
t.Fatalf("★ 消费后文件仍 %d 字节 —— 巨型历史文件应被一次性收敛", fi.Size())
|
||||
}
|
||||
// mtime 语义保留：文件里仍有 t= 行
|
||||
if !strings.HasPrefix(readAll(t, p), "t=") {
|
||||
t.Fatal("★ 收敛后丢了 t= 行 —— guard 的 mtime 语义会被破坏")
|
||||
}
|
||||
}
|
||||
|
||||
func itoa(n int64) string {
|
||||
return strconv.FormatInt(n, 10)
|
||||
}
|
||||
64
cmd/homed/kernel_control.go
Normal file
64
cmd/homed/kernel_control.go
Normal file
@ -0,0 +1,64 @@
|
||||
package main
|
||||
|
||||
// 内核控制通道的实现：把 WebUI 的"重启/关机"请求转成心跳指令。
|
||||
//
|
||||
// ## 完整链路（与心跳复用，不新开通道）
|
||||
//
|
||||
// WebUI 点按钮
|
||||
// → POST /api/v1/kernel/{restart,shutdown}
|
||||
// → sdk.KernelControlAPI.RequestKernel*（本文件实现，只投递不等待）
|
||||
// → requestHeartbeatCmd()：把指令放进 worker 内存的单槽队列
|
||||
// → 下一次心跳（≤5s）把 `cmd=` 行写进 <data>/heartbeat
|
||||
// → ① 重启：worker 走正常清理 → exit 42 → guard 重拉（既有语义，零改动）
|
||||
// ② 关机：guard 在 workerCleanExit 分支读到 cmd=shutdown
|
||||
// → systemctl stop → 不再拉起
|
||||
//
|
||||
// ## ★ 为什么不用"worker 自己调 systemctl stop"
|
||||
//
|
||||
// worker 是被守护方。它自己停服务等于在父进程眼皮底下自杀，而且
|
||||
// systemd 可能先把这次退出记成一次失败。更根本的是：
|
||||
// **只有 guard 知道这次退出该不该重拉** —— 决策该集中在一个地方。
|
||||
//
|
||||
// ## 重启为什么几乎零改动
|
||||
//
|
||||
// `exitRestartRequested=42` 与 guard 的 `workerRestartRequested` 分支早已存在
|
||||
// 且生产验证过（热更新就走这条路）。我们只是给"用户点按钮"也接到同一个出口。
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
)
|
||||
|
||||
// kernelControlImpl 实现 sdk.KernelControlAPI。
|
||||
type kernelControlImpl struct {
|
||||
// restartCh 是内核主循环的重启通道（既有机制，supervisor 也用）。
|
||||
restartCh chan<- struct{}
|
||||
}
|
||||
|
||||
// RequestKernelRestart 请求热重启。
|
||||
//
|
||||
// 走**既有**的 restartCh（与 supervisor.Restart 同一条路）：
|
||||
// worker 清理 → exit 42 → guard 用当前二进制重拉，不计失败轮次。
|
||||
func (k *kernelControlImpl) RequestKernelRestart(reason string) error {
|
||||
if k.restartCh == nil {
|
||||
return fmt.Errorf("内核重启通道未装配")
|
||||
}
|
||||
select {
|
||||
case k.restartCh <- struct{}{}:
|
||||
return nil
|
||||
default:
|
||||
// 已有待处理的重启请求 ⇒ 目标一致，视为成功（幂等）
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// RequestKernelShutdown 请求关机。
|
||||
//
|
||||
// ★ 与重启的区别：关机**不能**复用 restartCh（那会被 guard 重拉）。
|
||||
// 必须让 guard 在 worker 退出后知道"这次别再起了" ⇒ 写心跳指令。
|
||||
//
|
||||
// 这里只投递；真正的写入在 worker 的下一次心跳（≤5s），
|
||||
// 因此本方法立即返回 —— 调用方（HTTP handler）需要的是"请求已受理"。
|
||||
func (k *kernelControlImpl) RequestKernelShutdown(reason string) error {
|
||||
requestHeartbeatCmd(hbCmdShutdown, reason)
|
||||
return nil
|
||||
}
|
||||
@ -282,6 +282,14 @@ func main() {
|
||||
|
||||
restartCh := startSupervisorRuntime(sup, trk, agent)
|
||||
|
||||
// 内核控制通道（WebUI 的重启/关机按钮走这里）。
|
||||
//
|
||||
// ★ 必须在插件加载**之前**注入：WebUI 在 Start 里取
|
||||
// s.KernelControl()，注入晚了它拿到 nil ⇒ 按钮永远报"不可用"。
|
||||
// 装配顺序在模块化内核里由依赖图决定，而本行位于 agent 模块之后、
|
||||
// plugin-host 拉起插件之前 —— 正是需要的位置。
|
||||
pluginReg.SetKernelControl(&kernelControlImpl{restartCh: restartCh})
|
||||
|
||||
log.Printf("[homed] main agent started, model=%s base=%s sources=%d adapters=%d",
|
||||
cfg.LLM.Model, cfg.LLM.BaseURL, len(cfg.LLM.Sources), len(luaVM.ListAdapters()))
|
||||
log.Printf("[homed] kernel ready, waiting for plugin IO...")
|
||||
|
||||
Reference in New Issue
Block a user