diff --git a/cmd/ingestor/mqtt_watchdog_1749_test.go b/cmd/ingestor/mqtt_watchdog_1749_test.go index 105c18ad..43e40782 100644 --- a/cmd/ingestor/mqtt_watchdog_1749_test.go +++ b/cmd/ingestor/mqtt_watchdog_1749_test.go @@ -58,15 +58,8 @@ func TestMQTTStallWatchdog_EscalateOnPersistentDisconnect_1749(t *testing.T) { t.Fatalf("setup: %v", err) } - tick := make(chan time.Time) - done := make(chan struct{}) - defer close(done) - - exited := make(chan struct{}) - go func() { - runLivenessWatchdogLoop(tick, done, threshold, func(args ...any) {}) - close(exited) - }() + tick, stopLoop := startWatchdogTestLoop(t, threshold, func(args ...any) {}) + defer stopLoop() // Feed ticks spanning > (multiplier × threshold) of wall clock so // the escalation path fires. We control the `now` parameter by @@ -114,10 +107,8 @@ func TestMQTTStallWatchdog_DisconnectedEscalationThrottled_1749(t *testing.T) { t.Fatalf("setup: %v", err) } - tick := make(chan time.Time) - done := make(chan struct{}) - defer close(done) - go runLivenessWatchdogLoop(tick, done, threshold, func(args ...any) {}) + tick, stopLoop := startWatchdogTestLoop(t, threshold, func(args ...any) {}) + defer stopLoop() base := time.Now() // Pre-stamp DisconnectedSinceUnix so that the first tick is @@ -259,10 +250,8 @@ func TestMQTTStallWatchdog_LastTickUnixExposed_1749(t *testing.T) { watchdogLastTickUnix.Store(before) }) - tick := make(chan time.Time) - done := make(chan struct{}) - defer close(done) - go runLivenessWatchdogLoop(tick, done, time.Minute, func(args ...any) {}) + tick, stopLoop := startWatchdogTestLoop(t, time.Minute, func(args ...any) {}) + defer stopLoop() stamp := time.Now().Add(48 * time.Hour) // guaranteed > before select { diff --git a/cmd/ingestor/mqtt_watchdog_1810_test.go b/cmd/ingestor/mqtt_watchdog_1810_test.go index aeefd4a0..b68fa75d 100644 --- a/cmd/ingestor/mqtt_watchdog_1810_test.go +++ b/cmd/ingestor/mqtt_watchdog_1810_test.go @@ -212,10 +212,8 @@ func TestWatchdog_EscalationWarnThrottled_1810(t *testing.T) { } } - tick := make(chan time.Time) - done := make(chan struct{}) - defer close(done) - go runLivenessWatchdogLoop(tick, done, threshold, emit) + tick, stopLoop := startWatchdogTestLoop(t, threshold, emit) + defer stopLoop() base := time.Now() // First tick: stamps DisconnectedSinceUnix (no escalation yet). diff --git a/cmd/ingestor/mqtt_watchdog_testhelpers_test.go b/cmd/ingestor/mqtt_watchdog_testhelpers_test.go index 348c2f2d..1db9b99a 100644 --- a/cmd/ingestor/mqtt_watchdog_testhelpers_test.go +++ b/cmd/ingestor/mqtt_watchdog_testhelpers_test.go @@ -1,6 +1,7 @@ package main import ( + "sync" "testing" "time" ) @@ -45,3 +46,31 @@ func sendTickOrFail(t *testing.T, tick chan<- time.Time, stamp time.Time, timeou t.Fatalf("%s: tick blocked after %s — loop dead?", label, timeout) } } + +// startWatchdogTestLoop is setupWatchdogTestLoop for the common case: a test +// that wants the loop running for its own duration and nothing more. The +// returned stop closes done AND waits for the loop goroutine to return. It is +// safe to call more than once. +// +// Waiting is the part that must not be skipped. Closing done only asks the +// loop to stop; the goroutine can still be inside a scan, and that scan walks +// the package-level livenessRegistry. The next test registers its own source +// there, so a loop that has not returned yet will process that source on a +// tick carrying the previous test's fabricated clock. +// +// That is measured, not theoretical. With four call sites closing done and +// walking away, TestMQTTStallWatchdog_DisconnectedEscalationThrottled_1749 +// failed 2 to 3 times per 20 runs of the watchdog tests and never once in 50 +// runs on its own. Debug tracing showed two loops reaching maybeForceReconnect +// for the same source with tick clocks 420s apart, both reading +// LastForceReconnectUnix as 0 before either wrote it, so the throttle the test +// asserts on let both through. +func startWatchdogTestLoop(t *testing.T, threshold time.Duration, emit func(...any)) (tick chan time.Time, stop func()) { + t.Helper() + tick, done, exited := setupWatchdogTestLoop(t, threshold, emit) + var once sync.Once + return tick, func() { + once.Do(func() { close(done) }) + <-exited + } +}