test(ingestor): join the watchdog loop goroutine instead of only asking it to stop (#2003)

Verified before merging: on upstream/master `go test ./cmd/ingestor -run TestMQTTStallWatchdog -count=20` fails; on this branch the same command passes. The flake blocked CI on #2000.

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 {

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).

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
+	}
+}