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/main.go b/cmd/ingestor/main.go index 68a298de5..d4513e575 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -202,14 +202,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. @@ -572,6 +568,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_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) + } +} 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) + } +}