vrr/Pulse
2
0
Fork 0
mirror of https://github.com/rcourtman/Pulse.git synced 2026-08-01 11:13:32 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

Fix Docker agent test timer race by making the timer hook per-Agent

Browse source 
Tests swapped the package-level newTimerFn hook while async goroutines
leaked from earlier tests (backup-cleanup and stop-command paths) were
still reading it in waitForAsyncDelay, tripping the race detector on CI.
Replace the global with a per-Agent newTimerFn seam (nil defaults to
time.NewTimer), make waitForContextDelay an Agent method, and inject the
immediate timer into the tests that previously swapped the global.
Verified with go test ./internal/dockeragent/ -race -count=20.

Contract-Neutral: test seam refactor to fix data race, no public contract delta

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
courtmanr@gmail.com 2026-07-28 16:27:30 +01:00
parent 16ff5544c1
commit 8e5ef365d4
6 changed files with 31 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

20
internal/dockeragent/agent.go
View file

@ -134,6 +134,7 @@ type Agent struct {
manualCheckActiveID string
manualCheckResults map[string]manualUpdateCheckResult
manualCheckCollect func(context.Context) (agentsdocker.Report, error) // test seam for bounded manual checks
newTimerFn func(time.Duration) *time.Timer // test seam; per-Agent so async goroutines never read a shared global (nil = time.NewTimer)
backgroundMu sync.Mutex // protects updateCheckRunning, cleanupTaskRunning
updateCheckRunning bool
cleanupTaskRunning bool
@ -734,7 +735,7 @@ func (a *Agent) Run(ctx context.Context) error {
)
initialDelay := 5*time.Second + randomDurationFn(startupJitterWindow)
updateTimer := newTimerFn(initialDelay)
updateTimer := a.newTimer(initialDelay)
defer stopTimer(updateTimer)
// Periodic cleanup of orphaned backups (every 15 minutes)
@ -811,13 +812,20 @@ func (a *Agent) runAsync(task func(context.Context)) {
}()
}
func (a *Agent) newTimer(delay time.Duration) *time.Timer {
if a.newTimerFn != nil {
return a.newTimerFn(delay)
}
return time.NewTimer(delay)
}
func (a *Agent) waitForAsyncDelay(delay time.Duration) bool {
if delay <= 0 {
return true
}
a.ensureAsyncLifecycle()
timer := newTimerFn(delay)
timer := a.newTimer(delay)
defer stopTimer(timer)
select {
@ -1348,18 +1356,18 @@ func (a *Agent) sendManualUpdateCheckAckWithRetry(ctx context.Context, target Ta
if err == nil {
return nil
}
if attempt+1 == manualUpdateCheckAckAttempts || !waitForContextDelay(ctx, manualUpdateCheckAckRetryDelay*time.Duration(1<<attempt)) {
if attempt+1 == manualUpdateCheckAckAttempts || !a.waitForContextDelay(ctx, manualUpdateCheckAckRetryDelay*time.Duration(1<<attempt)) {
break
}
}
return err
}
func waitForContextDelay(ctx context.Context, delay time.Duration) bool {
func (a *Agent) waitForContextDelay(ctx context.Context, delay time.Duration) bool {
if delay <= 0 {
return ctx.Err() == nil
}
timer := newTimerFn(delay)
timer := a.newTimer(delay)
defer stopTimer(timer)
select {
case <-ctx.Done():
@ -1621,7 +1629,7 @@ func (a *Agent) Close() error {
close(done)
}()
waitTimer := newTimerFn(2 * time.Second)
waitTimer := a.newTimer(2 * time.Second)
select {
case <-done:
stopTimer(waitTimer)

6
internal/dockeragent/agent_http_test.go
View file

@ -727,9 +727,6 @@ func TestHandleStopCommand(t *testing.T) {
t.Run("stop service goroutine executes", func(t *testing.T) {
marker := filepath.Join(t.TempDir(), "called")
writeSystemctl(t, "if [ \"$1\" = \"disable\" ]; then exit 0; fi\nif [ \"$1\" = \"stop\" ]; then : > "+marker+"; exit 2; fi\nexit 0")
swap(t, &newTimerFn, func(time.Duration) *time.Timer {
return time.NewTimer(0)
})
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusOK)
@ -742,8 +739,9 @@ func TestHandleStopCommand(t *testing.T) {
httpClients: map[bool]*http.Client{
false: server.Client(),
},
newTimerFn: immediateTimer,
}
// Close agent BEFORE swap restores globals (t.Cleanup is LIFO)
// Close joins the stop-service goroutine before the fake systemctl is cleaned up
t.Cleanup(func() { _ = agent.Close() })
if err := agent.handleStopCommand(context.Background(), TargetConfig{URL: server.URL, Token: "token"}, agentsdocker.Command{ID: "cmd"}); !errors.Is(err, ErrStopRequested) {

24
internal/dockeragent/container_update_test.go
View file

@ -223,9 +223,6 @@ func TestUpdateContainer_Errors(t *testing.T) {
func TestUpdateContainer_Success(t *testing.T) {
logger := zerolog.Nop()
swap(t, &sleepFn, func(time.Duration) {})
swap(t, &newTimerFn, func(time.Duration) *time.Timer {
return time.NewTimer(0)
})
swap(t, &nowFn, func() time.Time {
return time.Date(2024, 3, 1, 12, 0, 0, 0, time.UTC)
})
@ -271,7 +268,8 @@ func TestUpdateContainer_Success(t *testing.T) {
return err
},
},
logger: logger,
logger: logger,
newTimerFn: immediateTimer,
}
result := agent.updateContainerWithProgress(context.Background(), "container1", nil)
@ -297,9 +295,6 @@ func TestUpdateContainer_Success(t *testing.T) {
func TestUpdateContainer_StoppedContainerPreservesStoppedState(t *testing.T) {
logger := zerolog.Nop()
swap(t, &sleepFn, func(time.Duration) {})
swap(t, &newTimerFn, func(time.Duration) *time.Timer {
return time.NewTimer(0)
})
swap(t, &nowFn, func() time.Time {
return time.Date(2024, 3, 1, 12, 0, 0, 0, time.UTC)
})
@ -348,7 +343,8 @@ func TestUpdateContainer_StoppedContainerPreservesStoppedState(t *testing.T) {
return nil
},
},
logger: logger,
logger: logger,
newTimerFn: immediateTimer,
}
result := agent.updateContainerWithProgress(context.Background(), "container1", nil)
@ -365,9 +361,6 @@ func TestUpdateContainer_StoppedContainerPreservesStoppedState(t *testing.T) {
func TestUpdateContainer_CleanupError(t *testing.T) {
logger := zerolog.Nop()
swap(t, &sleepFn, func(time.Duration) {})
swap(t, &newTimerFn, func(time.Duration) *time.Timer {
return time.NewTimer(0)
})
swap(t, &nowFn, func() time.Time {
return time.Date(2024, 3, 1, 12, 0, 0, 0, time.UTC)
})
@ -405,7 +398,8 @@ func TestUpdateContainer_CleanupError(t *testing.T) {
return cleanupErr
},
},
logger: logger,
logger: logger,
newTimerFn: immediateTimer,
}
result := agent.updateContainerWithProgress(context.Background(), "container1", nil)
@ -621,9 +615,6 @@ func TestHandleUpdateContainerCommand(t *testing.T) {
func TestUpdateContainer_SharedNamespaceCreateConfig(t *testing.T) {
logger := zerolog.Nop()
swap(t, &sleepFn, func(time.Duration) {})
swap(t, &newTimerFn, func(time.Duration) *time.Timer {
return time.NewTimer(0)
})
swap(t, &nowFn, func() time.Time {
return time.Date(2024, 3, 1, 12, 0, 0, 0, time.UTC)
})
@ -691,7 +682,8 @@ func TestUpdateContainer_SharedNamespaceCreateConfig(t *testing.T) {
return nil
},
},
logger: logger,
logger: logger,
newTimerFn: immediateTimer,
}
result := agent.updateContainerWithProgress(context.Background(), "container1", nil)

4
internal/dockeragent/container_update_typed_test.go
View file

@ -60,7 +60,6 @@ func TestTypedContainerUpdatePreflightRefusesRuntimeAndDigestDrift(t *testing.T)
func TestTypedContainerUpdateDelegatesToProductionRecreatePath(t *testing.T) {
swap(t, &sleepFn, func(time.Duration) {})
swap(t, &newTimerFn, func(time.Duration) *time.Timer { return time.NewTimer(0) })
cleanupDone := make(chan struct{})
agent := &Agent{
@ -93,7 +92,8 @@ func TestTypedContainerUpdateDelegatesToProductionRecreatePath(t *testing.T) {
return nil
},
},
logger: zerolog.Nop(),
logger: zerolog.Nop(),
newTimerFn: immediateTimer,
}
var progress []string

1
internal/dockeragent/deps.go
View file

@ -19,7 +19,6 @@ var (
connectRuntimeFn = connectRuntime
hostmetricsCollect = hostmetrics.Collect
newTickerFn = time.NewTicker
newTimerFn = time.NewTimer
randomDurationFn = randomDuration
nowFn = time.Now
sleepFn = time.Sleep

5
internal/dockeragent/test_helpers_test.go
View file

@ -7,6 +7,7 @@ import (
"errors"
"io"
"testing"
"time"
containertypes "github.com/moby/moby/api/types/container"
"github.com/moby/moby/api/types/image"
@ -224,6 +225,10 @@ func statsReader(t *testing.T, stats containertypes.StatsResponse) dockerStatsRe
}
}
func immediateTimer(time.Duration) *time.Timer {
return time.NewTimer(0)
}
func swap[T any](t *testing.T, target *T, value T) {
t.Helper()
prev := *target