diff --git a/cmd/homed/bootstrap.go b/cmd/homed/bootstrap.go index f1e4f34f..6b2fc1a6 100644 --- a/cmd/homed/bootstrap.go +++ b/cmd/homed/bootstrap.go @@ -834,17 +834,25 @@ func startSupervisorRuntime(sup *supervisor.Daemon, trk *tracker.Tracker, agent func startHeartbeat(dataDir string, ctx context.Context) func() { // 心跳:每 5s 触碰 /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: diff --git a/cmd/homed/guard_safety_test.go b/cmd/homed/guard_safety_test.go index 66793f4f..a0866c38 100644 --- a/cmd/homed/guard_safety_test.go +++ b/cmd/homed/guard_safety_test.go @@ -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") + } +} diff --git a/cmd/homed/main.go b/cmd/homed/main.go index ed270024..ab51c0cb 100644 --- a/cmd/homed/main.go +++ b/cmd/homed/main.go @@ -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) }