From 0a3de8b0400742edbee2a467a6e9cf2dfde38632 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Fri, 17 Apr 2026 14:20:02 +0000 Subject: [PATCH] fix: P1-3 address heartbeat timer leak, test drain, and store consistency - COR-001: drain runHeartbeat goroutine via done channel before test returns, eliminating race against concurrent SetHeartbeatInterval tests - COR-002: replace time.After with time.NewTimer + defer t.Stop() + t.Reset() to prevent timer goroutine leaks on context cancellation - COR-005: guard HeartbeatInterval store write with > 0 check in OnClusterUpdated, keeping store consistent with SetHeartbeatInterval's zero-ignore semantics --- src/frontend/apiserver/backend/handler.go | 4 +++- src/frontend/apiserver/backend/heartbeat.go | 8 ++++++-- src/frontend/apiserver/backend/heartbeat_test.go | 13 +++++++------ 3 files changed, 16 insertions(+), 9 deletions(-) diff --git a/src/frontend/apiserver/backend/handler.go b/src/frontend/apiserver/backend/handler.go index 7e36036..7f41ce8 100644 --- a/src/frontend/apiserver/backend/handler.go +++ b/src/frontend/apiserver/backend/handler.go @@ -60,7 +60,9 @@ func (h *DefaultHandler) OnHostConfigUpdated(incoming *models.HostInfo) { func (h *DefaultHandler) OnClusterUpdated(incoming *models.ClusterInfo) { existing, _ := store.Clusters.Get(incoming.ClusterName) existing.ClusterName = incoming.ClusterName - existing.HeartbeatInterval = incoming.HeartbeatInterval + if incoming.HeartbeatInterval > 0 { + existing.HeartbeatInterval = incoming.HeartbeatInterval + } existing.Vcpus = incoming.Vcpus existing.InactiveHeartbeatCount = incoming.InactiveHeartbeatCount store.Clusters.Put(existing.ClusterName, existing) diff --git a/src/frontend/apiserver/backend/heartbeat.go b/src/frontend/apiserver/backend/heartbeat.go index 7897570..fff98c7 100644 --- a/src/frontend/apiserver/backend/heartbeat.go +++ b/src/frontend/apiserver/backend/heartbeat.go @@ -21,13 +21,17 @@ func SetHeartbeatInterval(secs int) { // runHeartbeat loops until ctx is cancelled, calling fn(intervalSecs) on each tick. // Extracted from clib.go for testability without CGO. func runHeartbeat(ctx context.Context, fn func(int)) { + interval := int(atomic.LoadInt32(&heartbeatIntervalSecs)) + t := time.NewTimer(time.Duration(interval) * time.Second) + defer t.Stop() for { - interval := int(atomic.LoadInt32(&heartbeatIntervalSecs)) select { case <-ctx.Done(): return - case <-time.After(time.Duration(interval) * time.Second): + case <-t.C: fn(interval) } + interval = int(atomic.LoadInt32(&heartbeatIntervalSecs)) + t.Reset(time.Duration(interval) * time.Second) } } diff --git a/src/frontend/apiserver/backend/heartbeat_test.go b/src/frontend/apiserver/backend/heartbeat_test.go index 134db92..0924d3f 100644 --- a/src/frontend/apiserver/backend/heartbeat_test.go +++ b/src/frontend/apiserver/backend/heartbeat_test.go @@ -35,17 +35,18 @@ func TestRunHeartbeat_CallsFunction(t *testing.T) { atomic.StoreInt32(&heartbeatIntervalSecs, 1) var callCount int64 - fn := func(_ int) { - atomic.AddInt64(&callCount, 1) - } + fn := func(_ int) { atomic.AddInt64(&callCount, 1) } ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - - go runHeartbeat(ctx, fn) + done := make(chan struct{}) + go func() { + runHeartbeat(ctx, fn) + close(done) + }() time.Sleep(1500 * time.Millisecond) cancel() + <-done // wait for goroutine to exit before the test returns if got := atomic.LoadInt64(&callCount); got < 1 { t.Errorf("fn was called %d times, want at least 1", got)