fix: stop watchdog force-reconnect from racing paho's own retry loop (#1897)

Relates to #1335, which was already closed by PR #1336 shipping the
naive `client.Disconnect(250); client.Connect()` force-reconnect. That
fix has its own bug: liveness.IsConnectedFn (paho's IsConnected())
reports true for the entire time paho is actively retrying, not just
when genuinely connected, so the watchdog's stall check cannot tell a
half-open TCP socket (the original #1335 case) from a broker that paho
is already correctly reconnecting to. Unconditionally calling
Disconnect(250) then Connect() on that second, transitional case races
paho's status machine and permanently kills its retry loop, requiring
another watchdog trigger to recover, sometimes compounding into 100+
minute outages.

This is a different failure mode from #1749/PR #1853: that bug is a
blocking log.Print() write freezing the entire watchdog loop before
ForceReconnectFn is ever called. This bug only manifests once
ForceReconnectFn does fire, so the two fixes are independent and touch
disjoint files.

buildForceReconnectFn now gates Disconnect() on IsConnectionOpen() (true
only when status is strictly connected) so it only tears down a
genuinely open connection, and logs Connect()'s error token instead of
discarding it.

cmd/ingestor/main.go

@@ -193,14 +193,10 @@ func main() {
 		// half-open TCP socket and re-dial when paho.IsConnected==true
 		// but no messages have flowed past the stall threshold. Throttled
 		// per source by the watchdog itself (forceReconnectThrottle).
-		// Disconnect(250) gives in-flight publishes 250ms to drain;
-		// Connect() returns immediately and paho's reconnect machinery
-		// takes over from there. Captured-by-value `client` is the same
-		// pointer used everywhere else for this source.
-		liveness.ForceReconnectFn = func() {
-			client.Disconnect(250)
-			client.Connect()
-		}
+		// Captured-by-value `client` is the same pointer used everywhere
+		// else for this source. See buildForceReconnectFn for why this is
+		// NOT simply "Disconnect(250) then Connect()".
+		liveness.ForceReconnectFn = buildForceReconnectFn(client, tag)
 		// PR #1216 r2 item 3: tag collisions used to log.Fatalf, which
 		// killed the entire ingestor over one config typo and recreated
 		// the #1212 total-ingest-stop class this PR exists to prevent.
@@ -544,6 +540,54 @@ func buildMQTTOpts(source MQTTSource) *mqtt.ClientOptions {
 	return opts
 }
 
+// buildForceReconnectFn builds the watchdog's forced-reconnect action for a
+// source (#1335, hardened against a race found while investigating a 100+
+// minute reconnect failure).
+//
+// paho's own client.IsConnected() — used as liveness.IsConnectedFn — reports
+// true not only when genuinely connected but for the ENTIRE time paho's
+// background AutoReconnect/ConnectRetry loop is retrying (status
+// reconnecting/connecting). So the watchdog's LivenessStalled classification
+// (IsConnected==true, no messages) fires just as often for "paho is actively,
+// correctly retrying a still-down broker" as it does for the true #1335
+// half-open-TCP case. Naively doing Disconnect(250) then Connect() in the
+// first case is actively harmful: paho's Disconnecting() must block until the
+// CURRENT in-flight connection attempt plus its backoff sleep unwind (up to
+// ConnectTimeout+MaxReconnectInterval, tens of seconds) before status
+// actually reaches `disconnected`. Disconnect(250) returns to the caller
+// after the 250ms quiesce regardless, so the following Connect() usually runs
+// while status is still the transitional `disconnecting` state — paho then
+// returns an error token (silently discarded by the old code) AND, because
+// Disconnect() was called at all, tears down paho's own retry loop for good
+// ("user requested no auto reconnection"). The client is left with nothing
+// retrying until the watchdog's next trigger fires, which can repeat the same
+// race — compounding into very long outages.
+//
+// client.IsConnectionOpen() (unlike IsConnected()) is strictly status ==
+// connected — never true while paho is reconnecting/connecting — so it
+// reliably distinguishes "genuinely connected, maybe half-open" (safe to
+// Disconnect then Connect; Disconnecting() does not need to wait on any
+// in-flight retry loop from status connected, so it completes well within
+// the 250ms quiesce) from "paho is already retrying on its own" (must NOT
+// call Disconnect; Connect() alone is a safe no-op per paho when a retry is
+// already under way, and properly starts a fresh attempt on the rare
+// occasion status has actually settled to disconnected).
+func buildForceReconnectFn(client mqtt.Client, tag string) func() {
+	return func() {
+		if client.IsConnectionOpen() {
+			client.Disconnect(250)
+		}
+		// Connect() resolves synchronously (Error() readable immediately,
+		// no Wait() needed) for both error returns and the "already
+		// retrying, treated as a safe no-op" success case — only a genuine
+		// fresh connection attempt leaves the token pending in the
+		// background, and we must not block this call on that.
+		if token := client.Connect(); token.Error() != nil {
+			log.Printf("MQTT [%s] WATCHDOG force-reconnect Connect() failed: %v", tag, token.Error())
+		}
+	}
+}
+
 func handleMessage(store *Store, tag string, source MQTTSource, m mqtt.Message, channelKeys map[string]string, regionKeys map[string][]byte, cfg *Config) {
 	// Liveness watchdog (#1212): record receipt before any processing so a
 	// slow handler still counts as "source is alive". Cheap atomic store.

cmd/ingestor/mqtt_force_reconnect_race_test.go

@@ -0,0 +1,150 @@
+package main
+
+import (
+	"bytes"
+	"errors"
+	"log"
+	"strings"
+	"testing"
+	"time"
+
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+)
+
+// fakeToken is a minimal mqtt.Token whose Error() is resolvable immediately
+// without calling Wait(), matching how paho's real ConnectToken behaves on
+// both of its synchronous return paths (the "already retrying, treated as a
+// safe no-op" success case and any error case) — see buildForceReconnectFn's
+// doc comment for why the caller must not need to Wait() before reading
+// Error() (doing so would block indefinitely on a genuine, still in-flight
+// reconnect attempt).
+type fakeToken struct{ err error }
+
+func (t *fakeToken) Wait() bool                     { return true }
+func (t *fakeToken) WaitTimeout(time.Duration) bool { return true }
+func (t *fakeToken) Done() <-chan struct{}          { ch := make(chan struct{}); close(ch); return ch }
+func (t *fakeToken) Error() error                   { return t.err }
+
+// fakeClient implements mqtt.Client, recording calls to the three methods
+// buildForceReconnectFn actually uses (IsConnectionOpen, Disconnect,
+// Connect). Every other method panics — buildForceReconnectFn must never
+// touch subscriptions, publishes, or options, so a call there indicates the
+// fix drifted from its intended scope.
+type fakeClient struct {
+	isConnectionOpen bool
+	connectErr       error
+
+	disconnectCalled bool
+	connectCalled    bool
+	callOrder        []string
+}
+
+func (c *fakeClient) IsConnected() bool { panic("not used by buildForceReconnectFn") }
+func (c *fakeClient) IsConnectionOpen() bool {
+	c.callOrder = append(c.callOrder, "IsConnectionOpen")
+	return c.isConnectionOpen
+}
+func (c *fakeClient) Connect() mqtt.Token {
+	c.connectCalled = true
+	c.callOrder = append(c.callOrder, "Connect")
+	return &fakeToken{err: c.connectErr}
+}
+func (c *fakeClient) Disconnect(quiesce uint) {
+	c.disconnectCalled = true
+	c.callOrder = append(c.callOrder, "Disconnect")
+}
+func (c *fakeClient) Publish(topic string, qos byte, retained bool, payload any) mqtt.Token {
+	panic("not used by buildForceReconnectFn")
+}
+func (c *fakeClient) Subscribe(topic string, qos byte, callback mqtt.MessageHandler) mqtt.Token {
+	panic("not used by buildForceReconnectFn")
+}
+func (c *fakeClient) SubscribeMultiple(filters map[string]byte, callback mqtt.MessageHandler) mqtt.Token {
+	panic("not used by buildForceReconnectFn")
+}
+func (c *fakeClient) Unsubscribe(topics ...string) mqtt.Token {
+	panic("not used by buildForceReconnectFn")
+}
+func (c *fakeClient) AddRoute(topic string, callback mqtt.MessageHandler) {
+	panic("not used by buildForceReconnectFn")
+}
+func (c *fakeClient) OptionsReader() mqtt.ClientOptionsReader {
+	panic("not used by buildForceReconnectFn")
+}
+
+// Case 1 (#1335, the genuine half-open-TCP case): paho reports the
+// connection as strictly open. This is the one case where Disconnect() is
+// safe — status==connected means Disconnecting() does not need to wait on
+// any in-flight retry loop, so it completes well within the quiesce window.
+// buildForceReconnectFn must still Disconnect before Connect here.
+func TestBuildForceReconnectFn_DisconnectsWhenConnectionOpen(t *testing.T) {
+	c := &fakeClient{isConnectionOpen: true}
+	fn := buildForceReconnectFn(c, "half-open-source")
+	fn()
+
+	if !c.disconnectCalled {
+		t.Error("expected Disconnect to be called when IsConnectionOpen()==true (half-open TCP case, #1335)")
+	}
+	if !c.connectCalled {
+		t.Error("expected Connect to be called")
+	}
+	want := []string{"IsConnectionOpen", "Disconnect", "Connect"}
+	if len(c.callOrder) != len(want) {
+		t.Fatalf("expected call order %v, got %v", want, c.callOrder)
+	}
+	for i, name := range want {
+		if c.callOrder[i] != name {
+			t.Errorf("expected call order %v, got %v", want, c.callOrder)
+			break
+		}
+	}
+}
+
+// Case 2 (the race this fix closes): paho is already mid-retry
+// (reconnecting, or ConnectRetry's connecting), so IsConnectionOpen() is
+// false even though the looser IsConnected() (used elsewhere for liveness
+// classification) would report true. Calling Disconnect() here is what
+// caused the original bug: it races paho's internal status transition and
+// can permanently kill an in-flight retry loop via a botched, silently
+// swallowed reconnect. buildForceReconnectFn must skip Disconnect and just
+// call Connect(), which paho treats as a safe no-op if a retry is already
+// under way.
+func TestBuildForceReconnectFn_SkipsDisconnectWhenAlreadyRetrying(t *testing.T) {
+	c := &fakeClient{isConnectionOpen: false}
+	fn := buildForceReconnectFn(c, "mid-retry-source")
+	fn()
+
+	if c.disconnectCalled {
+		t.Error("Disconnect must NOT be called while paho is already retrying (IsConnectionOpen()==false) — this races paho's status machine and can kill the in-flight retry loop")
+	}
+	if !c.connectCalled {
+		t.Error("expected Connect to still be called (safe no-op per paho when already retrying, or a real reconnect if paho had actually settled to disconnected)")
+	}
+}
+
+// Case 3: Connect() returns an error token (e.g. the errStatusMustBeDisconnected
+// class this whole bug hinged on — Connect() called while paho's status is
+// transitionally "disconnecting"). The old code discarded the returned token
+// entirely; the fix must surface it via log output instead of silently
+// dropping it.
+func TestBuildForceReconnectFn_LogsConnectError(t *testing.T) {
+	c := &fakeClient{isConnectionOpen: false, connectErr: errors.New("status can only transition to connecting from disconnected")}
+	fn := buildForceReconnectFn(c, "erroring-source")
+
+	var buf bytes.Buffer
+	origOut := log.Writer()
+	origFlags := log.Flags()
+	log.SetOutput(&buf)
+	log.SetFlags(0)
+	defer func() {
+		log.SetOutput(origOut)
+		log.SetFlags(origFlags)
+	}()
+
+	fn()
+
+	logged := buf.String()
+	if !strings.Contains(logged, "erroring-source") || !strings.Contains(logged, "status can only transition") {
+		t.Errorf("expected Connect() error to be logged with the source tag, got: %q", logged)
+	}
+}