From 6345c6fb05750d86f2cd08ffd183ee9909315072 Mon Sep 17 00:00:00 2001 From: Kpa-clawbot Date: Fri, 1 May 2026 00:01:07 -0700 Subject: [PATCH] fix(ingestor): observability + bounded backoff for MQTT reconnect (#947) (#949) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Fixes #947 — MQTT ingestor silently stalls after `pingresp not received` disconnect due to paho's default 10-minute reconnect backoff and zero observability of reconnect attempts. ## Changes ### `cmd/ingestor/main.go` - **Extract `buildMQTTOpts()`** — encapsulates MQTT client option construction for testability - **`SetMaxReconnectInterval(30s)`** — bounds paho's default 10-minute exponential backoff (source: `options.go:137` in `paho.mqtt.golang@v1.5.0`) - **`SetConnectTimeout(10s)`** — prevents stuck connect attempts from blocking reconnect cycle - **`SetWriteTimeout(10s)`** — prevents stuck publish writes - **`SetReconnectingHandler`** — logs `MQTT [] reconnecting to ` on every reconnect attempt, giving operators visibility into retry behavior - **Enhanced `SetConnectionLostHandler`** — now includes broker address in log line for multi-source disambiguation ### `cmd/ingestor/mqtt_opts_test.go` (new) - Tests verify `MaxReconnectInterval`, `ConnectTimeout`, `WriteTimeout` are set correctly - Tests verify credential and TLS configuration - Anti-tautology: tests fail if timing settings are removed from `buildMQTTOpts()` ## Operator impact After this change, a pingresp disconnect produces: ``` MQTT [staging] disconnected from tcp://broker:1883: pingresp not received, disconnecting MQTT [staging] reconnecting to tcp://broker:1883 MQTT [staging] reconnecting to tcp://broker:1883 MQTT [staging] connected to tcp://broker:1883 MQTT [staging] subscribed to meshcore/# ``` Max gap between disconnect and first reconnect attempt: ~30s (was up to 10 minutes). --------- Co-authored-by: you --- cmd/ingestor/main.go | 50 ++++++++++++++-------- cmd/ingestor/mqtt_opts_test.go | 76 ++++++++++++++++++++++++++++++++++ 2 files changed, 108 insertions(+), 18 deletions(-) create mode 100644 cmd/ingestor/mqtt_opts_test.go diff --git a/cmd/ingestor/main.go b/cmd/ingestor/main.go index b0b94bed..b9dad90f 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -129,23 +129,7 @@ func main() { tag = source.Broker } - opts := mqtt.NewClientOptions(). - AddBroker(source.Broker). - SetAutoReconnect(true). - SetConnectRetry(true). - SetOrderMatters(true) - - if source.Username != "" { - opts.SetUsername(source.Username) - } - if source.Password != "" { - opts.SetPassword(source.Password) - } - if source.RejectUnauthorized != nil && !*source.RejectUnauthorized { - opts.SetTLSConfig(&tls.Config{InsecureSkipVerify: true}) - } else if strings.HasPrefix(source.Broker, "ssl://") { - opts.SetTLSConfig(&tls.Config{}) - } + opts := buildMQTTOpts(source) opts.SetOnConnectHandler(func(c mqtt.Client) { log.Printf("MQTT [%s] connected to %s", tag, source.Broker) @@ -165,7 +149,11 @@ func main() { }) opts.SetConnectionLostHandler(func(c mqtt.Client, err error) { - log.Printf("MQTT [%s] disconnected: %v", tag, err) + log.Printf("MQTT [%s] disconnected from %s: %v", tag, source.Broker, err) + }) + + opts.SetReconnectingHandler(func(c mqtt.Client, options *mqtt.ClientOptions) { + log.Printf("MQTT [%s] reconnecting to %s", tag, source.Broker) }) // Capture source for closure @@ -206,6 +194,32 @@ func main() { log.Println("Done.") } +// buildMQTTOpts creates MQTT client options for a source with bounded reconnect +// backoff, connect timeout, and TLS/auth configuration. +func buildMQTTOpts(source MQTTSource) *mqtt.ClientOptions { + opts := mqtt.NewClientOptions(). + AddBroker(source.Broker). + SetAutoReconnect(true). + SetConnectRetry(true). + SetOrderMatters(true). + SetMaxReconnectInterval(30 * time.Second). + SetConnectTimeout(10 * time.Second). + SetWriteTimeout(10 * time.Second) + + if source.Username != "" { + opts.SetUsername(source.Username) + } + if source.Password != "" { + opts.SetPassword(source.Password) + } + if source.RejectUnauthorized != nil && !*source.RejectUnauthorized { + opts.SetTLSConfig(&tls.Config{InsecureSkipVerify: true}) + } else if strings.HasPrefix(source.Broker, "ssl://") { + opts.SetTLSConfig(&tls.Config{}) + } + return opts +} + func handleMessage(store *Store, tag string, source MQTTSource, m mqtt.Message, channelKeys map[string]string, cfg *Config) { defer func() { if r := recover(); r != nil { diff --git a/cmd/ingestor/mqtt_opts_test.go b/cmd/ingestor/mqtt_opts_test.go new file mode 100644 index 00000000..e8aceec4 --- /dev/null +++ b/cmd/ingestor/mqtt_opts_test.go @@ -0,0 +1,76 @@ +package main + +import ( + "testing" + "time" +) + +func TestBuildMQTTOpts_ReconnectSettings(t *testing.T) { + source := MQTTSource{ + Broker: "tcp://localhost:1883", + Name: "test", + } + opts := buildMQTTOpts(source) + + if opts.MaxReconnectInterval != 30*time.Second { + t.Errorf("MaxReconnectInterval = %v, want 30s", opts.MaxReconnectInterval) + } + if opts.ConnectTimeout != 10*time.Second { + t.Errorf("ConnectTimeout = %v, want 10s", opts.ConnectTimeout) + } + if opts.WriteTimeout != 10*time.Second { + t.Errorf("WriteTimeout = %v, want 10s", opts.WriteTimeout) + } + if !opts.AutoReconnect { + t.Error("AutoReconnect should be true") + } + if !opts.ConnectRetry { + t.Error("ConnectRetry should be true") + } +} + +func TestBuildMQTTOpts_Credentials(t *testing.T) { + source := MQTTSource{ + Broker: "tcp://broker:1883", + Username: "user1", + Password: "pass1", + } + opts := buildMQTTOpts(source) + + if opts.Username != "user1" { + t.Errorf("Username = %q, want %q", opts.Username, "user1") + } + if opts.Password != "pass1" { + t.Errorf("Password = %q, want %q", opts.Password, "pass1") + } +} + +func TestBuildMQTTOpts_TLS_InsecureSkipVerify(t *testing.T) { + f := false + source := MQTTSource{ + Broker: "ssl://broker:8883", + RejectUnauthorized: &f, + } + opts := buildMQTTOpts(source) + + if opts.TLSConfig == nil { + t.Fatal("TLSConfig should be set") + } + if !opts.TLSConfig.InsecureSkipVerify { + t.Error("InsecureSkipVerify should be true when RejectUnauthorized=false") + } +} + +func TestBuildMQTTOpts_TLS_SSL_Prefix(t *testing.T) { + source := MQTTSource{ + Broker: "ssl://broker:8883", + } + opts := buildMQTTOpts(source) + + if opts.TLSConfig == nil { + t.Fatal("TLSConfig should be set for ssl:// brokers") + } + if opts.TLSConfig.InsecureSkipVerify { + t.Error("InsecureSkipVerify should be false by default") + } +}