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