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
This commit is contained in:
parent
ffc5b20cdb
commit
0a3de8b040
3 changed files with 16 additions and 9 deletions
|
|
@ -60,7 +60,9 @@ func (h *DefaultHandler) OnHostConfigUpdated(incoming *models.HostInfo) {
|
||||||
func (h *DefaultHandler) OnClusterUpdated(incoming *models.ClusterInfo) {
|
func (h *DefaultHandler) OnClusterUpdated(incoming *models.ClusterInfo) {
|
||||||
existing, _ := store.Clusters.Get(incoming.ClusterName)
|
existing, _ := store.Clusters.Get(incoming.ClusterName)
|
||||||
existing.ClusterName = incoming.ClusterName
|
existing.ClusterName = incoming.ClusterName
|
||||||
existing.HeartbeatInterval = incoming.HeartbeatInterval
|
if incoming.HeartbeatInterval > 0 {
|
||||||
|
existing.HeartbeatInterval = incoming.HeartbeatInterval
|
||||||
|
}
|
||||||
existing.Vcpus = incoming.Vcpus
|
existing.Vcpus = incoming.Vcpus
|
||||||
existing.InactiveHeartbeatCount = incoming.InactiveHeartbeatCount
|
existing.InactiveHeartbeatCount = incoming.InactiveHeartbeatCount
|
||||||
store.Clusters.Put(existing.ClusterName, existing)
|
store.Clusters.Put(existing.ClusterName, existing)
|
||||||
|
|
|
||||||
|
|
@ -21,13 +21,17 @@ func SetHeartbeatInterval(secs int) {
|
||||||
// runHeartbeat loops until ctx is cancelled, calling fn(intervalSecs) on each tick.
|
// runHeartbeat loops until ctx is cancelled, calling fn(intervalSecs) on each tick.
|
||||||
// Extracted from clib.go for testability without CGO.
|
// Extracted from clib.go for testability without CGO.
|
||||||
func runHeartbeat(ctx context.Context, fn func(int)) {
|
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 {
|
for {
|
||||||
interval := int(atomic.LoadInt32(&heartbeatIntervalSecs))
|
|
||||||
select {
|
select {
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
return
|
return
|
||||||
case <-time.After(time.Duration(interval) * time.Second):
|
case <-t.C:
|
||||||
fn(interval)
|
fn(interval)
|
||||||
}
|
}
|
||||||
|
interval = int(atomic.LoadInt32(&heartbeatIntervalSecs))
|
||||||
|
t.Reset(time.Duration(interval) * time.Second)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -35,17 +35,18 @@ func TestRunHeartbeat_CallsFunction(t *testing.T) {
|
||||||
atomic.StoreInt32(&heartbeatIntervalSecs, 1)
|
atomic.StoreInt32(&heartbeatIntervalSecs, 1)
|
||||||
|
|
||||||
var callCount int64
|
var callCount int64
|
||||||
fn := func(_ int) {
|
fn := func(_ int) { atomic.AddInt64(&callCount, 1) }
|
||||||
atomic.AddInt64(&callCount, 1)
|
|
||||||
}
|
|
||||||
|
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
defer cancel()
|
done := make(chan struct{})
|
||||||
|
go func() {
|
||||||
go runHeartbeat(ctx, fn)
|
runHeartbeat(ctx, fn)
|
||||||
|
close(done)
|
||||||
|
}()
|
||||||
|
|
||||||
time.Sleep(1500 * time.Millisecond)
|
time.Sleep(1500 * time.Millisecond)
|
||||||
cancel()
|
cancel()
|
||||||
|
<-done // wait for goroutine to exit before the test returns
|
||||||
|
|
||||||
if got := atomic.LoadInt64(&callCount); got < 1 {
|
if got := atomic.LoadInt64(&callCount); got < 1 {
|
||||||
t.Errorf("fn was called %d times, want at least 1", got)
|
t.Errorf("fn was called %d times, want at least 1", got)
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue