From 1d843daf2a9c7a3b76380a30f34a714e6d572237 Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Mon, 12 Oct 2026 19:32:04 +0800 Subject: [PATCH] =?UTF-8?q?fix(guard):=20=E5=BF=83=E8=B7=B3=E6=8F=90?= =?UTF-8?q?=E5=89=8D=E5=90=AF=E5=8A=A8=20+=20=E9=A6=96=E6=AC=A1=E5=90=8C?= =?UTF-8?q?=E6=AD=A5=E5=86=99=E5=85=A5=20=E2=80=94=E2=80=94=20=E5=90=AF?= =?UTF-8?q?=E5=8A=A8=E7=AB=9E=E6=80=81=E7=9A=84=20worker=20=E4=BE=A7?= =?UTF-8?q?=E6=A0=B9=E4=BF=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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。 --- cmd/homed/bootstrap.go | 22 ++++++++---- cmd/homed/guard_safety_test.go | 61 ++++++++++++++++++++++++++++++++++ cmd/homed/main.go | 23 +++++++++++-- 3 files changed, 96 insertions(+), 10 deletions(-) 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) }