FMQ config update for v1.0

broker/localauth/store_test.go

@@ -197,6 +197,7 @@ func TestStoreRejectsInvalidConfiguration(t *testing.T) {
 	current := writeSecret(t, dir, "current", currentSecret)
 	weak := writeSecret(t, dir, "weak", strings.Repeat("x", 31))
 	nul := writeSecret(t, dir, "nul", strings.Repeat("x", 16)+"\x00"+strings.Repeat("x", 16))
+	doubleNewline := writeSecret(t, dir, "double-newline", strings.Repeat("x", 32)+"\n\n")
 
 	tests := []struct {
 		name      string
@@ -235,6 +236,22 @@ func TestStoreRejectsInvalidConfiguration(t *testing.T) {
 			},
 			wantError: "secret file must not contain NUL bytes",
 		},
+		{
+			// Secret contents are checked here rather than in configuration
+			// validation, which never opens the files.
+			name: "more than one terminal newline",
+			configs: func() []config.LocalPrincipalConfig {
+				return []config.LocalPrincipalConfig{principalConfig(doubleNewline, "")}
+			},
+			wantError: "may contain only one terminal newline",
+		},
+		{
+			name: "missing secret file",
+			configs: func() []config.LocalPrincipalConfig {
+				return []config.LocalPrincipalConfig{principalConfig(filepath.Join(dir, "absent"), "")}
+			},
+			wantError: "failed to read secret file",
+		},
 		{
 			name: "same current and previous secret",
 			configs: func() []config.LocalPrincipalConfig {

cluster/message.go

@@ -17,10 +17,13 @@ type Message struct {
 // QueueMessage is a typed envelope for cross-node queue delivery.
 // It separates queue metadata from user-defined message properties.
 type QueueMessage struct {
-	MessageID      string
-	QueueName      string
-	GroupID        string
-	Topic          string
+	MessageID string
+	QueueName string
+	GroupID   string
+	Topic     string
+	// SourceTopic is the topic the message was published to, before queue
+	// addressing. Topic identifies the queue; only this recovers the origin.
+	SourceTopic    string
 	Payload        []byte
 	Sequence       int64
 	UserProperties map[string]string

cluster/transport.go

@@ -1327,6 +1327,10 @@ func encodeRouteQueueMessage(clientID, queueName string, msg *QueueMessage) *clu
 		properties[queueTypes.PropQueueName] = queueName
 	}
 	properties[queueTypes.PropOffset] = fmt.Sprintf("%d", msg.Sequence)
+	// Source topic is broker-owned even when empty. Stamping the zero value is
+	// what clears a publisher-supplied property copied above instead of letting
+	// the decoder promote it into trusted queue metadata.
+	properties[queueTypes.PropSourceTopic] = msg.SourceTopic
 	if msg.Stream {
 		properties[queueTypes.PropStreamOffset] = fmt.Sprintf("%d", msg.StreamOffset)
 		if msg.StreamTimestamp != 0 {
@@ -1381,6 +1385,9 @@ func decodeRouteQueueMessage(wire *clusterv1.RouteQueueMessageRequest) *QueueMes
 	if offset, ok := parseInt64Property(rawProps, queueTypes.PropOffset); ok {
 		msg.Sequence = offset
 	}
+	if sourceTopic := rawProps[queueTypes.PropSourceTopic]; sourceTopic != "" {
+		msg.SourceTopic = sourceTopic
+	}
 	if streamOffset, ok := parseInt64Property(rawProps, queueTypes.PropStreamOffset); ok {
 		msg.Stream = true
 		msg.StreamOffset = streamOffset

cluster/transport_queue_topic_test.go

@@ -3,13 +3,17 @@
 
 package cluster
 
-import "testing"
+import (
+	"testing"
+
+	queueTypes "github.com/absmach/fluxmq/queue/types"
+)
 
 func TestRouteQueueMessageWirePreservesDeliveryTopic(t *testing.T) {
 	const deliveryTopic = "$queue/m/domain/c/channel/tst"
 
 	wire := encodeRouteQueueMessage("consumer", "m", &QueueMessage{
 		MessageID: "m:1",
 		QueueName: "m",
 		GroupID:   "rules-engine",
 		Topic:     deliveryTopic,
@@ -26,3 +30,66 @@
 		t.Fatalf("decoded topic = %q, want %q", decoded.Topic, deliveryTopic)
 	}
 }
+
+// The delivery address cannot be reversed into a source topic, so a consumer on
+// another node depends entirely on the source topic surviving the wire.
+func TestRouteQueueMessageWirePreservesSourceTopic(t *testing.T) {
+	const sourceTopic = "domain/c/channel/tst"
+
+	wire := encodeRouteQueueMessage("consumer", "m", &QueueMessage{
+		MessageID:      "m:1",
+		QueueName:      "m",
+		GroupID:        "rules-engine",
+		Topic:          "$queue/m/domain/c/channel/tst",
+		SourceTopic:    sourceTopic,
+		Payload:        []byte("payload"),
+		Sequence:       1,
+		UserProperties: map[string]string{"user": "kept"},
+	})
+
+	decoded := decodeRouteQueueMessage(wire)
+	if decoded.SourceTopic != sourceTopic {
+		t.Fatalf("decoded source topic = %q, want %q", decoded.SourceTopic, sourceTopic)
+	}
+	// Queue-owned metadata must not leak into the user properties a consumer
+	// sees as its own.
+	if _, leaked := decoded.UserProperties[queueTypes.PropSourceTopic]; leaked {
+		t.Fatal("the source topic leaked into user properties")
+	}
+	if decoded.UserProperties["user"] != "kept" {
+		t.Fatalf("an ordinary user property was dropped: %v", decoded.UserProperties)
+	}
+}
+
+// Source topic is broker-owned even when its real value is empty. The encoder
+// must overwrite a publisher-supplied value before the decoder promotes the
+// reserved property into QueueMessage.SourceTopic.
+func TestRouteQueueMessageWireClearsForgedEmptySourceTopic(t *testing.T) {
+	wire := encodeRouteQueueMessage("consumer", "m", &QueueMessage{
+		MessageID:   "m:1",
+		QueueName:   "m",
+		Topic:       "$queue/m",
+		SourceTopic: "",
+		Payload:     []byte("payload"),
+		Sequence:    1,
+		UserProperties: map[string]string{
+			queueTypes.PropSourceTopic: "forged/topic",
+			"user":                     "kept",
+		},
+	})
+
+	if sourceTopic, ok := wire.Properties[queueTypes.PropSourceTopic]; !ok || sourceTopic != "" {
+		t.Fatalf("wire source topic = %q, present=%t; want an explicit empty broker value", sourceTopic, ok)
+	}
+
+	decoded := decodeRouteQueueMessage(wire)
+	if decoded.SourceTopic != "" {
+		t.Fatalf("decoded source topic = %q, want the real empty source topic", decoded.SourceTopic)
+	}
+	if _, leaked := decoded.UserProperties[queueTypes.PropSourceTopic]; leaked {
+		t.Fatal("the source topic leaked into user properties")
+	}
+	if decoded.UserProperties["user"] != "kept" {
+		t.Fatalf("an ordinary user property was dropped: %v", decoded.UserProperties)
+	}
+}

cmd/main.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"crypto/tls"
 	"encoding/hex"
+	"errors"
 	"flag"
 	"fmt"
 	"log/slog"
@@ -65,6 +66,16 @@ const (
 	listenerMTLS  = "mtls"
 )
 
+// valueOr returns what the operator wrote, or the built-in default when the
+// key was omitted. Configuration keeps absent and zero distinct so a written
+// value is never silently replaced by a different one.
+func valueOr[T any](configured *T, fallback T) T {
+	if configured == nil {
+		return fallback
+	}
+	return *configured
+}
+
 func protocolVersionForMode(mode string) int {
 	switch config.NormalizeProtocolMode(mode) {
 	case config.ProtocolModeV3:
@@ -385,10 +396,23 @@ func releaseShutdownResources(
 
 func main() {
 	configFile := flag.String("config", "", "Path to configuration file")
+	configOptional := flag.Bool("config-optional", false,
+		"Fall back to built-in defaults when --config names a file that does not exist")
 	flag.Parse()
 
-	cfg, err := config.Load(*configFile)
+	load := config.Load
+	if *configOptional {
+		load = config.LoadOptional
+	}
+
+	cfg, err := load(*configFile)
 	if err != nil {
+		if errors.Is(err, config.ErrConfigNotFound) {
+			slog.Error("Configuration file not found",
+				"path", *configFile,
+				"hint", "check the path, or pass --config-optional to start with built-in defaults")
+			os.Exit(1)
+		}
 		slog.Error("Failed to load configuration", "error", err)
 		os.Exit(1)
 	}
@@ -439,14 +463,14 @@ func main() {
 
 	slog.Info("Starting MQTT broker", "version", fluxmq.Version)
 	slog.Info("Configuration loaded",
-		"tcp_v3_listener", cfg.Server.TCP.V3.Addr,
-		"tcp_v5_listener", cfg.Server.TCP.V5.Addr,
-		"tcp_tls_listener", cfg.Server.TCP.TLS.Addr,
-		"tcp_mtls_listener", cfg.Server.TCP.MTLS.Addr,
-		"ws_v3_listener", cfg.Server.WebSocket.V3.Addr,
-		"ws_v5_listener", cfg.Server.WebSocket.V5.Addr,
-		"ws_tls_listener", cfg.Server.WebSocket.TLS.Addr,
-		"ws_mtls_listener", cfg.Server.WebSocket.MTLS.Addr,
+		"tcp_v3_listener", cfg.Server.MQTT.TCP.V3.Addr,
+		"tcp_v5_listener", cfg.Server.MQTT.TCP.V5.Addr,
+		"tcp_tls_listener", cfg.Server.MQTT.TCP.TLS.Addr,
+		"tcp_mtls_listener", cfg.Server.MQTT.TCP.MTLS.Addr,
+		"ws_v3_listener", cfg.Server.MQTT.WebSocket.V3.Addr,
+		"ws_v5_listener", cfg.Server.MQTT.WebSocket.V5.Addr,
+		"ws_tls_listener", cfg.Server.MQTT.WebSocket.TLS.Addr,
+		"ws_mtls_listener", cfg.Server.MQTT.WebSocket.MTLS.Addr,
 		"http_plain_listener", cfg.Server.HTTP.Plain.Addr,
 		"http_tls_listener", cfg.Server.HTTP.TLS.Addr,
 		"http_mtls_listener", cfg.Server.HTTP.MTLS.Addr,
@@ -478,7 +502,7 @@ func main() {
 	case "badger":
 		badgerStore, err := badger.New(badger.Config{
 			Dir:        cfg.Storage.BadgerDir,
-			SyncWrites: cfg.Storage.SyncWrites,
+			SyncWrites: cfg.Storage.BadgerSyncWrites,
 		})
 		if err != nil {
 			slog.Error("Failed to initialize BadgerDB storage", "error", err)
@@ -725,14 +749,10 @@ func main() {
 			}
 		}
 
-		cacheSize := cfg.Auth.External.IdentityCacheSize
-		if cacheSize == 0 {
-			cacheSize = corebroker.DefaultIdentityCacheSize
-		}
-		cacheTTL := cfg.Auth.External.IdentityCacheTTL
-		if cacheTTL == 0 {
-			cacheTTL = corebroker.DefaultIdentityCacheTTL
-		}
+		// An omitted key takes the built-in default; a written one is used as
+		// written. Validation already refused a written zero.
+		cacheSize := valueOr(cfg.Auth.External.IdentityCacheSize, corebroker.DefaultIdentityCacheSize)
+		cacheTTL := valueOr(cfg.Auth.External.IdentityCacheTTL, corebroker.DefaultIdentityCacheTTL)
 		engineOpts := []corebroker.AuthEngineOption{
 			corebroker.WithIdentityCache(cacheSize, cacheTTL),
 		}
@@ -774,7 +794,7 @@ func main() {
 
 		newHookProvider := func() corebroker.BlockingHookProvider {
 			opts := []hook.Option{
-				hook.WithTimeout(cfg.Hooks.Timeout),
+				hook.WithTimeout(valueOr(cfg.Hooks.Timeout, 0)),
 				hook.WithLogger(logger),
 			}
 			switch transport {
@@ -795,7 +815,7 @@ func main() {
 		slog.Info("Blocking hooks configured",
 			"url", cfg.Hooks.URL,
 			"transport", transport,
-			"timeout", cfg.Hooks.Timeout,
+			"timeout", valueOr(cfg.Hooks.Timeout, 0),
 			"fail_mode", cfg.Hooks.FailMode,
 			"protocols", cfg.Hooks.Protocols,
 			"events", cfg.Hooks.Events)
@@ -847,16 +867,11 @@ func main() {
 		// Convert queue configs from main config to queue types
 		queueCfg := queue.DefaultConfig()
 		queueCfg.AutoCommitInterval = cfg.QueueManager.AutoCommitInterval
-		// Zero leaves the dispatcher default in place.
-		if cfg.QueueManager.CaptureWorkers > 0 {
-			queueCfg.CaptureWorkers = cfg.QueueManager.CaptureWorkers
-		}
-		if cfg.QueueManager.CaptureQueueDepth > 0 {
-			queueCfg.CaptureQueueDepth = cfg.QueueManager.CaptureQueueDepth
-		}
-		if cfg.QueueManager.CaptureDrainTimeout > 0 {
-			queueCfg.CaptureDrainTimeout = cfg.QueueManager.CaptureDrainTimeout
-		}
+		// An omitted key leaves the dispatcher default in place; a written one
+		// is used as written, and validation already refused a written zero.
+		queueCfg.CaptureWorkers = valueOr(cfg.QueueManager.CaptureWorkers, queueCfg.CaptureWorkers)
+		queueCfg.CaptureQueueDepth = valueOr(cfg.QueueManager.CaptureQueueDepth, queueCfg.CaptureQueueDepth)
+		queueCfg.CaptureDrainTimeout = valueOr(cfg.QueueManager.CaptureDrainTimeout, queueCfg.CaptureDrainTimeout)
 		queueCfg.WritePolicy = queue.WritePolicy(cfg.Cluster.Raft.WritePolicy)
 		queueCfg.DistributionMode = queue.DistributionMode(cfg.Cluster.Raft.DistributionMode)
 		for _, qc := range cfg.Queues {
@@ -1077,12 +1092,12 @@ func main() {
 
 	tcpSlots := []struct {
 		name string
-		cfg  config.TCPListenerConfig
+		cfg  config.MQTTTCPListenerConfig
 	}{
-		{name: "v3", cfg: cfg.Server.TCP.V3},
-		{name: "v5", cfg: cfg.Server.TCP.V5},
-		{name: listenerTLS, cfg: cfg.Server.TCP.TLS},
-		{name: listenerMTLS, cfg: cfg.Server.TCP.MTLS},
+		{name: "v3", cfg: cfg.Server.MQTT.TCP.V3},
+		{name: "v5", cfg: cfg.Server.MQTT.TCP.V5},
+		{name: listenerTLS, cfg: cfg.Server.MQTT.TCP.TLS},
+		{name: listenerMTLS, cfg: cfg.Server.MQTT.TCP.MTLS},
 	}
 
 	for _, slot := range tcpSlots {
@@ -1124,12 +1139,12 @@ func main() {
 
 	wsSlots := []struct {
 		name string
-		cfg  config.WSListenerConfig
+		cfg  config.MQTTWebSocketListenerConfig
 	}{
-		{name: "v3", cfg: cfg.Server.WebSocket.V3},
-		{name: "v5", cfg: cfg.Server.WebSocket.V5},
-		{name: listenerTLS, cfg: cfg.Server.WebSocket.TLS},
-		{name: listenerMTLS, cfg: cfg.Server.WebSocket.MTLS},
+		{name: "v3", cfg: cfg.Server.MQTT.WebSocket.V3},
+		{name: "v5", cfg: cfg.Server.MQTT.WebSocket.V5},
+		{name: listenerTLS, cfg: cfg.Server.MQTT.WebSocket.TLS},
+		{name: listenerMTLS, cfg: cfg.Server.MQTT.WebSocket.MTLS},
 	}
 
 	for _, slot := range wsSlots {