fix(guard): 心跳提前启动 + 首次同步写入 —— 启动竞态的 worker 侧根修

562ba6bd 修了 guard 侧(给旧 mtime 加启动宽限,不误杀),那只是防御。
**根因**在 worker 侧:心跳文件要等全部重型初始化完(原 main 尾部)才
首写,且首次写是异步(在 goroutine 里,与多模态空间/知识库重建竞争
调度)。停机 >30s 后重启时 mtime 停在上一代 worker,guard 的 5s ticker
先到就 SIGKILL。

两层独立生效:
  worker 侧(本提交,治本):dataDir 一确定就启动心跳,首次写同步
  guard 侧(562ba6bd,防御):保留启动宽限期容忍旧 mtime

## 改动

- main.go:startHeartbeat 从尾部(190)移到 dataDir 解析后(54),
  用 context.Background()(心跳生命周期=进程,与 agent ctx 解耦)。
- bootstrap.go:startHeartbeat 首次 writeHB() 提到 goroutine 之外
  同步执行——返回前 mtime 必已变新,main 往下做慢事之前 guard 看到的
  是新心跳。

## 判据(TestHeartbeatFirstWriteSynchronous,变异测试通过)

  (a) 返回后立即查 mtime:必须是新的(同步首写),不许等到第一个 tick
      首写移回 goroutine(异步)→ (a) 当场 FAIL,已实测
  (b) 5s 后 mtime 再变(ticker 在跑)
  (c) stop() 后不再更新(hbStop 生效,不泄漏 goroutine)

判据覆盖完整生命周期:启动即新 → 周期更新 → 停止即停。

cmd/homed 包全量测试通过;全仓生产编译 GREEN。
This commit is contained in:
JianFeeeee
2026-10-12 19:32:04 +08:00
parent 562ba6bd98
commit 1d843daf2a
3 changed files with 96 additions and 10 deletions

View File

@ -834,17 +834,25 @@ func startSupervisorRuntime(sup *supervisor.Daemon, trk *tracker.Tracker, agent
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 {
fmt.Fprintf(f, "t=%d\n", time.Now().Unix())
f.Close()
}
}
// ★ 首次写入必须**同步**(返回前落盘),不能丢进 goroutine:
//
// heartbeat 文件在 worker 侧的唯一职责是「让 guard 看到 mtime 变新」。
// 停机 >heartbeatTimeout 后重启时,文件 mtime 停在上一代 worker(已过期);
// 若首次写是异步的,它可能与重型初始化(多模态空间/知识库重建,冷启动
// 5–8s)竞争调度,迟迟不落盘 —— guard 的 ticker 先到、拿旧 mtime 判 stale。
// (2026-10-12 生产部署实测:6 连杀 + failback 降级。)
// 同步首写保证:main() 继续往下做任何慢事之前,mtime 已经是新的。
writeHB()
hbStop := make(chan struct{})
go func() {
t := time.NewTicker(5 * time.Second)
defer t.Stop()
writeHB := func() {
if f, err := os.OpenFile(hbPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644); err == nil {
fmt.Fprintf(f, "t=%d\n", time.Now().Unix())
f.Close()
}
}
writeHB()
for {
select {
case <-t.C:

View File

@ -3,6 +3,7 @@
package main
import (
"context"
"os"
"path/filepath"
"strings"
@ -169,3 +170,63 @@ func TestFailbackPluginsAreSubsetOfKnownPlugins(t *testing.T) {
}
}
}
// TestHeartbeatFirstWriteSynchronous 钉住「心跳首写必须同步落盘」。
//
// ★ 场景(2026-10-12 生产部署实测的 worker 侧根因):停机 >30s 后重启,
// heartbeat 文件 mtime 停在上一代 worker(已过期)。原实现把首次写放进
// goroutine —— 它与重型初始化(多模态空间/知识库重建,冷启动 5–8s)竞争
// 调度,迟迟不落盘;guard 的 5s ticker 先到、拿旧 mtime 判 stale →
// SIGKILL 加载中的 worker(6 连杀 + failback 降级,guard 侧修复见
// shouldKillStaleHeartbeat 的宽限期;本判据钉 worker 侧根因)。
//
// 不变量:startHeartbeat 返回时,心跳文件 mtime 必已是「现在」——
// main() 往下做任何慢事之前,guard 看到的一定是新心跳。
func TestHeartbeatFirstWriteSynchronous(t *testing.T) {
dir := t.TempDir()
hb := filepath.Join(dir, "heartbeat")
// 造上一代遗留的过期心跳(复现停机重启场景)
if err := os.WriteFile(hb, []byte("old"), 0644); err != nil {
t.Fatal(err)
}
old := time.Now().Add(-2 * time.Minute)
if err := os.Chtimes(hb, old, old); err != nil {
t.Fatal(err)
}
// 启动心跳;返回即应已落盘首写
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
stop := startHeartbeat(dir, ctx)
// (a) 返回后立刻检查 mtime:必须是新的(同步首写),不许等到第一个 tick
fi, err := os.Stat(hb)
if err != nil {
t.Fatalf("心跳文件不存在: %v", err)
}
if age := time.Since(fi.ModTime()); age > 2*time.Second {
t.Errorf("startHeartbeat 返回后心跳仍是旧的(age=%v)—— 首写不是同步的,guard 会拿旧 mtime 误杀加载中的 worker", age)
}
// (b) 5s 后应有第二次写(ticker 还在跑)
time.Sleep(5500 * time.Millisecond)
fi2, err := os.Stat(hb)
if err != nil {
t.Fatalf("5s 后心跳文件消失: %v", err)
}
if !fi2.ModTime().After(fi.ModTime()) {
t.Error("5s 后心跳未更新 —— ticker 未运行,guard 会在 heartbeatTimeout 后误杀健康 worker")
}
// (c) 停止后不再写
stop()
m3 := func() time.Time {
f, _ := os.Stat(hb)
return f.ModTime()
}()
time.Sleep(5500 * time.Millisecond)
if m4, _ := os.Stat(hb); !m4.ModTime().Equal(m3) {
t.Error("停止心跳后文件仍被更新 —— hbStop 未生效,泄漏 goroutine")
}
}

View File

@ -53,6 +53,23 @@ func main() {
opt.dataDir = resolveDataDir(opt.dataDir)
}
// ---- 心跳:dataDir 一确定就启动(★ 启动竞态的 worker 侧根修)----
//
// 心跳文件是 guard 判 worker 存活的唯一依据(mtime 新鲜度 + IPC PING)。
// 必须在**任何重型初始化之前**首写同步落盘:停机 >heartbeatTimeout 后重启,
// 文件 mtime 停在上一代 worker(已过期);若等所有初始化完才首写(原位置在
// main 尾部),guard 的 5s ticker 先到、拿旧 mtime 判 stale → 把还在加载
// 多模态空间/知识库的 worker SIGKILL(2026-10-12 生产实测:6 连杀 +
// failback 降级,详见 562ba6bd)。
//
// guard 侧另有启动宽限期兜底(shouldKillStaleHeartbeat),两层独立生效:
// worker 侧让 mtime 尽快变新(治本),guard 侧容忍旧 mtime(防御)。
//
// 用 context.Background()(而非 main 的 ctx):心跳生命周期 = 进程,
// 与 agent 的 ctx 解耦;停止靠本函数返回的 hbStop(waitForShutdown 调它)。
stopHeartbeatEarly := startHeartbeat(opt.dataDir, context.Background())
defer stopHeartbeatEarly()
if opt.cliSocket == "" {
opt.cliSocket = filepath.Join(opt.dataDir, "cli.sock")
}
@ -186,7 +203,7 @@ func main() {
log.Printf("[homed] kernel ready, waiting for plugin IO...")
// ---- 等待退出信号 ----
stopHeartbeat := startHeartbeat(opt.dataDir, ctx)
waitForShutdown(ctx, opt.dataDir, restartCh, stopHeartbeat, pluginReg, trk, cfgReg, sup)
//
// 心跳已在 main 早期启动(见文件头部 startHeartbeatEarly),此处不再重复启动。
waitForShutdown(ctx, opt.dataDir, restartCh, stopHeartbeatEarly, pluginReg, trk, cfgReg, sup)
}