fix(ingestor): observability + bounded backoff for MQTT reconnect (#947) (#949)

## 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 [<tag>] reconnecting to
<broker>` 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 <you@example.com>

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 {

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