From 91d6d602237330de6b9cf8c445d12475632b93e4 Mon Sep 17 00:00:00 2001 From: Jonathan Herlin Date: Wed, 2 Sep 2026 10:28:00 +0200 Subject: [PATCH 1/2] 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. (cherry picked from commit 647841c99033c3574897f288b144f5e82721b4c4) Co-Authored-By: Claude Opus 5 --- cmd/ingestor/main.go | 60 ++++++- .../mqtt_force_reconnect_race_test.go | 150 ++++++++++++++++++ 2 files changed, 202 insertions(+), 8 deletions(-) create mode 100644 cmd/ingestor/mqtt_force_reconnect_race_test.go diff --git a/cmd/ingestor/main.go b/cmd/ingestor/main.go index ac102da01..f0f6e3cf4 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -201,14 +201,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. @@ -554,6 +550,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. diff --git a/cmd/ingestor/mqtt_force_reconnect_race_test.go b/cmd/ingestor/mqtt_force_reconnect_race_test.go new file mode 100644 index 000000000..8697cd3ea --- /dev/null +++ b/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) + } +} From cb7dc5187eee51e7661ae81cbfbdd905be3ea351 Mon Sep 17 00:00:00 2001 From: dborup Date: Sat, 26 Sep 2026 07:34:04 +0000 Subject: [PATCH 2/2] test(ingestor): pin #1897 force-reconnect behaviour against a real paho client The fake-client tests only assert the IsConnectionOpen/Disconnect/Connect call sequence. A Disconnect moved into a goroutine keeps that sequence and still kills paho's retry loop (review mutants M15/M18), and CI runs the ingestor suite without -race, so nothing caught it. New tests drive the production watchdog action (maybeForceReconnect -> buildForceReconnectFn) against a real paho client built by buildMQTTOpts and a small in-test MQTT broker, counting CONNECT packets at the broker: - broker down, paho retrying, watchdog fires three times: CONNECTs keep arriving and the client reconnects on its own when the broker returns; - half-open socket (IsConnectionOpen true, broker silent): the old socket is dropped and a new session comes up; - paho stopped (disconnected, #1749 escalation): force-reconnect starts a fresh connect without blocking while the broker is down. CI: add a separate -race pass for the ForceReconnect tests only; the full ingestor suite stays without -race. Fork guards untouched. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0133zLreBhuYtXESMXVBfqno --- .github/workflows/deploy.yml | 10 + .../mqtt_force_reconnect_paho_test.go | 391 ++++++++++++++++++ 2 files changed, 401 insertions(+) create mode 100644 cmd/ingestor/mqtt_force_reconnect_paho_test.go diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml index 97c3086a5..7e003b6b7 100644 --- a/.github/workflows/deploy.yml +++ b/.github/workflows/deploy.yml @@ -74,6 +74,16 @@ jobs: echo "--- Go Ingestor Coverage ---" go tool cover -func=ingestor-coverage.out | tail -1 + # The full ingestor suite runs without -race (too slow), so the + # force-reconnect tests (upstream #1897) get their own -race pass: a + # Disconnect moved into a goroutine races paho's retry loop, and the + # race detector backs up the behavioural real-paho tests. + - name: Race-check Go ingestor force-reconnect + run: | + set -e -o pipefail + cd cmd/ingestor + go test -race -count=1 -timeout 5m -run 'ForceReconnect' ./... + - name: Build and test channel library + decrypt CLI run: | set -e -o pipefail diff --git a/cmd/ingestor/mqtt_force_reconnect_paho_test.go b/cmd/ingestor/mqtt_force_reconnect_paho_test.go new file mode 100644 index 000000000..86012ebcf --- /dev/null +++ b/cmd/ingestor/mqtt_force_reconnect_paho_test.go @@ -0,0 +1,391 @@ +package main + +import ( + "fmt" + "net" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + mqtt "github.com/eclipse/paho.mqtt.golang" + "github.com/eclipse/paho.mqtt.golang/packets" +) + +// Behavioural tests for buildForceReconnectFn (upstream #1897) against a REAL +// paho client and a small in-test broker. +// +// The fake-client tests in mqtt_force_reconnect_race_test.go pin the call +// sequence. They cannot see what the fix is actually for: whether paho's own +// retry loop is still alive after the watchdog has fired. A Disconnect moved +// into a goroutine keeps the call sequence intact and kills the retry loop +// all the same (review of PR #29: mutants M15/M18). These tests assert the +// outcome instead, counted at the broker: +// +// - broker down, paho retrying, watchdog fires repeatedly: CONNECT packets +// keep arriving, and the client reconnects on its own once the broker is +// back (no further trigger); +// - genuine half-open socket (IsConnectionOpen()==true, broker silent): the +// force-reconnect drops the old socket and a new session comes up; +// - paho has given up (status disconnected, the #1749 escalation case): +// the force-reconnect starts a fresh connect without blocking. +// +// Timings are shortened from buildMQTTOpts' production values so the whole +// file runs in a few seconds. The one ratio that matters is kept: +// forceReconnectTestRetryInterval is well above Disconnect's 250ms quiesce, as +// production's 30s MaxReconnectInterval is, so a Disconnect issued while paho +// sleeps between attempts is still blocked in "disconnecting" when Connect() +// runs — the exact #1897 failure. + +const ( + forceReconnectTestRetryInterval = 800 * time.Millisecond + forceReconnectTestDeadline = 4 * time.Second +) + +type testBrokerMode int32 + +const ( + testBrokerUp testBrokerMode = iota // CONNECT → CONNACK accepted + testBrokerDown // CONNECT is read and counted, then the socket is closed +) + +// forceReconnectTestBroker speaks just enough MQTT 3.1.1 for paho: CONNECT → +// CONNACK, PINGREQ → PINGRESP, DISCONNECT → close. +type forceReconnectTestBroker struct { + ln net.Listener + mode atomic.Int32 + + mu sync.Mutex + connects int // CONNECT packets received, in any mode + accepted int // sessions that got a CONNACK + sessions []*testBrokerSession + wg sync.WaitGroup +} + +type testBrokerSession struct { + conn net.Conn + silent atomic.Bool // half-open: read everything, answer nothing + closed chan struct{} // closed once the client side has gone away +} + +func newForceReconnectTestBroker(t *testing.T) *forceReconnectTestBroker { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("test broker listen: %v", err) + } + b := &forceReconnectTestBroker{ln: ln} + b.wg.Add(1) + go b.acceptLoop() + t.Cleanup(b.close) + return b +} + +func (b *forceReconnectTestBroker) url() string { return "tcp://" + b.ln.Addr().String() } + +func (b *forceReconnectTestBroker) acceptLoop() { + defer b.wg.Done() + for { + conn, err := b.ln.Accept() + if err != nil { + return + } + b.wg.Add(1) + go b.serve(conn) + } +} + +func (b *forceReconnectTestBroker) serve(conn net.Conn) { + defer b.wg.Done() + defer conn.Close() + pkt, err := packets.ReadPacket(conn) + if err != nil { + return + } + if _, ok := pkt.(*packets.ConnectPacket); !ok { + return + } + b.mu.Lock() + b.connects++ + b.mu.Unlock() + if testBrokerMode(b.mode.Load()) == testBrokerDown { + return + } + connack := packets.NewControlPacket(packets.Connack).(*packets.ConnackPacket) + connack.ReturnCode = packets.Accepted + if connack.Write(conn) != nil { + return + } + s := &testBrokerSession{conn: conn, closed: make(chan struct{})} + defer close(s.closed) + b.mu.Lock() + b.accepted++ + b.sessions = append(b.sessions, s) + b.mu.Unlock() + for { + pkt, err := packets.ReadPacket(conn) + if err != nil { + return + } + if s.silent.Load() { + continue + } + switch pkt.(type) { + case *packets.PingreqPacket: + if packets.NewControlPacket(packets.Pingresp).Write(conn) != nil { + return + } + case *packets.DisconnectPacket: + return + } + } +} + +func (b *forceReconnectTestBroker) counts() (connects, accepted int) { + b.mu.Lock() + defer b.mu.Unlock() + return b.connects, b.accepted +} + +func (b *forceReconnectTestBroker) session(i int) *testBrokerSession { + b.mu.Lock() + defer b.mu.Unlock() + return b.sessions[i] +} + +// goDown drops every live session (paho sees EOF and starts its reconnect +// loop) and makes later CONNECTs fail. +func (b *forceReconnectTestBroker) goDown() { + b.mode.Store(int32(testBrokerDown)) + b.mu.Lock() + defer b.mu.Unlock() + for _, s := range b.sessions { + s.conn.Close() + } +} + +func (b *forceReconnectTestBroker) goUp() { b.mode.Store(int32(testBrokerUp)) } + +// silenceSessions turns every live session half-open: the socket stays up +// but the broker never sends another byte. +func (b *forceReconnectTestBroker) silenceSessions() { + b.mu.Lock() + defer b.mu.Unlock() + for _, s := range b.sessions { + s.silent.Store(true) + } +} + +func (b *forceReconnectTestBroker) close() { + b.ln.Close() + b.mu.Lock() + for _, s := range b.sessions { + s.conn.Close() + } + b.mu.Unlock() + b.wg.Wait() +} + +// newForceReconnectTestClient builds the client exactly as main does +// (buildMQTTOpts: AutoReconnect, ConnectRetry, 30s keepalive) with shortened +// timeouts, and waits for the first session. +func newForceReconnectTestClient(t *testing.T, b *forceReconnectTestBroker, tag string) mqtt.Client { + t.Helper() + opts := buildMQTTOpts(MQTTSource{Broker: b.url(), Name: tag}). + SetConnectTimeout(500 * time.Millisecond). + SetWriteTimeout(500 * time.Millisecond). + SetMaxReconnectInterval(forceReconnectTestRetryInterval). + SetConnectRetryInterval(200 * time.Millisecond) + client := mqtt.NewClient(opts) + client.Connect() + t.Cleanup(func() { client.Disconnect(0) }) + if !pollUntil(forceReconnectTestDeadline, client.IsConnectionOpen) { + t.Fatal("test setup: client never connected to the test broker") + } + return client +} + +// forceReconnectTestWatchdog drives the production watchdog action +// (maybeForceReconnect → ForceReconnectFn in a goroutine) with a synthetic +// clock that steps past forceReconnectThrottle, so each trigger really fires. +type forceReconnectTestWatchdog struct { + state *SourceLivenessState + now time.Time + + mu sync.Mutex + emits []string +} + +func newForceReconnectTestWatchdog(client mqtt.Client, tag string) *forceReconnectTestWatchdog { + s := &SourceLivenessState{ + Tag: tag, + Broker: "test", + IsConnectedFn: client.IsConnected, + ForceReconnectFn: buildForceReconnectFn(client, tag), + } + atomic.StoreInt64(&s.LastMessageUnix, time.Now().Add(-time.Hour).Unix()) + return &forceReconnectTestWatchdog{state: s, now: time.Now()} +} + +func (w *forceReconnectTestWatchdog) emit(args ...any) { + w.mu.Lock() + defer w.mu.Unlock() + if len(args) > 0 { + if s, ok := args[0].(string); ok { + w.emits = append(w.emits, s) + } + } +} + +func (w *forceReconnectTestWatchdog) count(substr string) int { + w.mu.Lock() + defer w.mu.Unlock() + n := 0 + for _, e := range w.emits { + if strings.Contains(e, substr) { + n++ + } + } + return n +} + +func (w *forceReconnectTestWatchdog) classify() LivenessKind { + _, kind := checkSourceLiveness(w.state, 5*time.Minute, time.Now()) + return kind +} + +// trigger fires one forced reconnect and waits for ForceReconnectFn to +// return. It must return promptly in every state: the watchdog issues it from +// a goroutine, and one that never returns leaks per trigger. +func (w *forceReconnectTestWatchdog) trigger(t *testing.T) { + t.Helper() + issued := w.count("reconnect attempt issued") + maybeForceReconnect(w.state, w.now, w.emit) + w.now = w.now.Add(forceReconnectThrottle + time.Second) + if !pollUntil(2*time.Second, func() bool { return w.count("reconnect attempt issued") > issued }) { + t.Fatal("ForceReconnectFn did not return within 2s") + } +} + +func pollUntil(timeout time.Duration, cond func() bool) bool { + deadline := time.Now().Add(timeout) + for { + if cond() { + return true + } + if time.Now().After(deadline) { + return false + } + time.Sleep(5 * time.Millisecond) + } +} + +// waitForConnects waits until the broker has seen more than `than` CONNECT +// packets and returns the new count. +func waitForConnects(t *testing.T, b *forceReconnectTestBroker, than int, why string) int { + t.Helper() + var n int + if !pollUntil(forceReconnectTestDeadline, func() bool { n, _ = b.counts(); return n > than }) { + t.Fatalf("%s: no CONNECT attempt within %s (still %d) — paho's retry loop is dead", why, forceReconnectTestDeadline, n) + } + return n +} + +// The #1897 production failure. Old code: after the first trigger the broker +// never sees another CONNECT and the client never comes back. +func TestForceReconnect_RealPaho_RetryLoopSurvivesRepeatedTriggers(t *testing.T) { + b := newForceReconnectTestBroker(t) + client := newForceReconnectTestClient(t, b, "force-reconnect-retrying") + w := newForceReconnectTestWatchdog(client, "force-reconnect-retrying") + + before, _ := b.counts() + b.goDown() + n := waitForConnects(t, b, before, "after broker went down") + + // Premise: while paho retries, IsConnected() stays true, so the watchdog + // classifies the silent source as Stalled and fires; IsConnectionOpen() + // is what tells the two cases apart. + if client.IsConnectionOpen() { + t.Fatal("IsConnectionOpen()==true while the broker is down") + } + if kind := w.classify(); kind != LivenessStalled { + t.Fatalf("watchdog classified a retrying client as %v, want LivenessStalled", kind) + } + + // Fire right after each observed attempt, i.e. while paho sleeps + // forceReconnectTestRetryInterval before the next one. A Disconnect here + // blocks in "disconnecting" until that sleep ends. + for i := 1; i <= 3; i++ { + w.trigger(t) + n = waitForConnects(t, b, n, fmt.Sprintf("after force-reconnect #%d", i)) + } + waitForConnects(t, b, n, "second attempt after the last force-reconnect") + + // Broker back: paho's own loop must reconnect, with no further trigger. + _, acceptedBefore := b.counts() + b.goUp() + if !pollUntil(forceReconnectTestDeadline, client.IsConnectionOpen) { + c, a := b.counts() + t.Fatalf("client did not reconnect within %s after the broker came back (CONNECTs %d, sessions %d)", forceReconnectTestDeadline, c, a) + } + if _, a := b.counts(); a <= acceptedBefore { + t.Fatalf("IsConnectionOpen()==true but the broker accepted no new session (%d)", a) + } +} + +// The #1335 case the force-reconnect exists for: paho thinks the socket is +// open, the broker is silent. The old socket must be dropped and replaced. +func TestForceReconnect_RealPaho_ReplacesHalfOpenSocket(t *testing.T) { + b := newForceReconnectTestBroker(t) + client := newForceReconnectTestClient(t, b, "force-reconnect-half-open") + w := newForceReconnectTestWatchdog(client, "force-reconnect-half-open") + + b.silenceSessions() + if !client.IsConnectionOpen() { + t.Fatal("test setup: half-open session must still look open to paho") + } + if kind := w.classify(); kind != LivenessStalled { + t.Fatalf("watchdog classified a half-open client as %v, want LivenessStalled", kind) + } + + w.trigger(t) + + if !pollUntil(forceReconnectTestDeadline, func() bool { _, a := b.counts(); return a >= 2 && client.IsConnectionOpen() }) { + c, a := b.counts() + t.Fatalf("no replacement session within %s (CONNECTs %d, sessions %d, open %v)", forceReconnectTestDeadline, c, a, client.IsConnectionOpen()) + } + select { + case <-b.session(0).closed: + case <-time.After(forceReconnectTestDeadline): + t.Fatal("client never dropped the half-open socket") + } +} + +// The #1749 escalation case: paho has stopped retrying (status disconnected). +// The force-reconnect must start a fresh connect — and must not block on it +// while the broker is still down. +func TestForceReconnect_RealPaho_RestartsClientPahoGaveUpOn(t *testing.T) { + b := newForceReconnectTestBroker(t) + client := newForceReconnectTestClient(t, b, "force-reconnect-gave-up") + w := newForceReconnectTestWatchdog(client, "force-reconnect-gave-up") + + client.Disconnect(250) + b.goDown() + if !pollUntil(forceReconnectTestDeadline, func() bool { return !client.IsConnected() }) { + t.Fatal("test setup: client did not settle to disconnected") + } + if kind := w.classify(); kind != LivenessDisconnected { + t.Fatalf("watchdog classified a stopped client as %v, want LivenessDisconnected", kind) + } + + before, _ := b.counts() + w.trigger(t) + n := waitForConnects(t, b, before, "after force-reconnect of a stopped client") + waitForConnects(t, b, n, "ConnectRetry after force-reconnect of a stopped client") + + b.goUp() + if !pollUntil(forceReconnectTestDeadline, client.IsConnectionOpen) { + t.Fatalf("client did not connect within %s after the broker came back", forceReconnectTestDeadline) + } +}