address comments

Signed-off-by: nyagamunene <stevenyaga2014@gmail.com>

Author: nyagamunene

pkg/events/nats/subscriber.go

@@ -22,16 +22,18 @@ var _ events.Subscriber = (*subEventStore)(nil)
 var (
 	eventsPrefix = "events"
 
-	jsStreamConfig = jetstream.StreamConfig{
-		Name:              "events",
-		Description:       "SuperMQ stream for sending and receiving messages in between SuperMQ events",
-		Subjects:          []string{"events.>"},
-		Retention:         jetstream.LimitsPolicy,
-		MaxMsgsPerSubject: 1e9,
-		MaxAge:            time.Hour * 24,
-		MaxMsgSize:        1024 * 1024,
-		Discard:           jetstream.DiscardOld,
-		Storage:           jetstream.FileStorage,
+	jsStreamConfig = broker.SMQJetStreamConfig{
+		StreamConfig: jetstream.StreamConfig{
+			Name:              "events",
+			Description:       "SuperMQ stream for sending and receiving messages in between SuperMQ events",
+			Subjects:          []string{"events.>"},
+			Retention:         jetstream.LimitsPolicy,
+			MaxMsgsPerSubject: 1e9,
+			MaxAge:            time.Hour * 24,
+			MaxMsgSize:        1024 * 1024,
+			Discard:           jetstream.DiscardOld,
+			Storage:           jetstream.FileStorage,
+		},
 	}
 
 	// ErrEmptyStream is returned when stream name is empty.

pkg/messaging/nats/options.go

@@ -14,63 +14,64 @@ import (
 var (
 	// ErrInvalidType is returned when the provided value is not of the expected type.
 	ErrInvalidType = errors.New("invalid type")
-
-	jsStreamConfig = jetstream.StreamConfig{
-		Name:              "m",
-		Description:       "SuperMQ stream for sending and receiving messages in between SuperMQ channels",
-		Subjects:          []string{"m.>"},
-		Retention:         jetstream.LimitsPolicy,
-		MaxMsgsPerSubject: 1e6,
-		MaxAge:            time.Hour * 24,
-		MaxMsgSize:        1024 * 1024,
-		Discard:           jetstream.DiscardOld,
-		Storage:           jetstream.FileStorage,
-	}
-)
-
-const (
-	msgPrefix                          = "m"
-	defaultMaxPendingMsgs              = 1000
-	defaultMaxPendingBytes             = 5 * 1024 * 1024
-	defaultEnableDroppedMsgTracking    = true
-	defaultEnableSlowConsumerDetection = true
 )
 
-type options struct {
-	prefix             string
-	jsStreamConfig     jetstream.StreamConfig
-	slowConsumerConfig *SlowConsumerConfig
+// SMQJetStreamConfig extends jetstream.StreamConfig with slow consumer monitoring settings
+type SMQJetStreamConfig struct {
+	jetstream.StreamConfig
+	
+	SlowConsumer SlowConsumerConfig `json:"slow_consumer,omitempty"`
 }
 
 type SlowConsumerConfig struct {
-	// MaxPendingMsgs maps to JetStream ConsumerConfig.MaxAckPending
-	// Controls the maximum number of outstanding unacknowledged messages
-	// Also used for slow consumer detection at 70% of this value
-	MaxPendingMsgs int
-
 	// MaxPendingBytes maps to JetStream ConsumerConfig.MaxRequestMaxBytes
 	// Controls the maximum bytes per batch request (closest JetStream equivalent)
-	MaxPendingBytes int
+	MaxPendingBytes int `json:"max_pending_bytes,omitempty"`
 
 	// EnableDroppedMsgTracking enables logging of message redeliveries
 	// which can indicate slow consumer behavior in JetStream
-	EnableDroppedMsgTracking bool
+	EnableDroppedMsgTracking bool `json:"enable_dropped_msg_tracking,omitempty"`
+}
 
-	// EnableSlowConsumerDetection enables automatic slow consumer detection
-	// and logging when consumer lag exceeds 70% of MaxPendingMsgs
-	EnableSlowConsumerDetection bool
+var (
+	defaultJetStreamConfig = SMQJetStreamConfig{
+		StreamConfig: jetstream.StreamConfig{
+			Name:              "m",
+			Description:       "SuperMQ stream for sending and receiving messages in between SuperMQ channels",
+			Subjects:          []string{"m.>"},
+			Retention:         jetstream.LimitsPolicy,
+			MaxMsgsPerSubject: 1e6,
+			MaxAge:            time.Hour * 24,
+			MaxMsgSize:        1024 * 1024,
+			Discard:           jetstream.DiscardOld,
+			Storage:           jetstream.FileStorage,
+			ConsumerLimits: jetstream.StreamConsumerLimits{
+				MaxAckPending: defaultMaxPendingMsgs,
+			},
+		},
+		SlowConsumer: SlowConsumerConfig{
+			MaxPendingBytes:          defaultMaxPendingBytes,
+			EnableDroppedMsgTracking: defaultEnableDroppedMsgTracking,
+		},
+	}
+)
+
+const (
+	msgPrefix                       = "m"
+	defaultMaxPendingMsgs           = 1000
+	defaultMaxPendingBytes          = 5 * 1024 * 1024
+	defaultEnableDroppedMsgTracking = true
+)
+
+type options struct {
+	prefix             string
+	jsStreamConfig     SMQJetStreamConfig
 }
 
 func defaultOptions() options {
 	return options{
 		prefix:         msgPrefix,
-		jsStreamConfig: jsStreamConfig,
-		slowConsumerConfig: &SlowConsumerConfig{
-			MaxPendingMsgs:              defaultMaxPendingMsgs,
-			MaxPendingBytes:             defaultMaxPendingBytes,
-			EnableDroppedMsgTracking:    defaultEnableDroppedMsgTracking,
-			EnableSlowConsumerDetection: defaultEnableSlowConsumerDetection,
-		},
+		jsStreamConfig: defaultJetStreamConfig,
 	}
 }
 
@@ -90,8 +91,8 @@ func Prefix(prefix string) messaging.Option {
 	}
 }
 
-// JSStreamConfig sets the JetStream for the publisher or subscriber.
-func JSStreamConfig(jsStreamConfig jetstream.StreamConfig) messaging.Option {
+// JSStreamConfig sets the JetStream configuration for the publisher or subscriber.
+func JSStreamConfig(jsStreamConfig SMQJetStreamConfig) messaging.Option {
 	return func(val any) error {
 		switch v := val.(type) {
 		case *publisher:
@@ -106,14 +107,30 @@ func JSStreamConfig(jsStreamConfig jetstream.StreamConfig) messaging.Option {
 	}
 }
 
-// WithSlowConsumerConfig sets the slow consumer configuration.
+// WithSlowConsumerConfig sets the slow consumer configuration within the JetStream config.
 func WithSlowConsumerConfig(config SlowConsumerConfig) messaging.Option {
 	return func(val interface{}) error {
 		switch v := val.(type) {
 		case *publisher:
-			v.slowConsumerConfig = &config
+			v.jsStreamConfig.SlowConsumer = config
+		case *pubsub:
+			v.jsStreamConfig.SlowConsumer = config
+		default:
+			return ErrInvalidType
+		}
+
+		return nil
+	}
+}
+
+// WithConsumerLimits sets the built-in JetStream consumer limits (MaxAckPending, etc.).
+func WithConsumerLimits(limits jetstream.StreamConsumerLimits) messaging.Option {
+	return func(val interface{}) error {
+		switch v := val.(type) {
+		case *publisher:
+			v.jsStreamConfig.ConsumerLimits = limits
 		case *pubsub:
-			v.slowConsumerConfig = &config
+			v.jsStreamConfig.ConsumerLimits = limits
 		default:
 			return ErrInvalidType
 		}

pkg/messaging/nats/publisher.go

@@ -55,7 +55,7 @@ func NewPublisher(ctx context.Context, url string, opts ...messaging.Option) (me
 	if err != nil {
 		return nil, err
 	}
-	if _, err := js.CreateStream(ctx, pub.jsStreamConfig); err != nil {
+	if _, err := js.CreateStream(ctx, pub.jsStreamConfig.StreamConfig); err != nil {
 		return nil, err
 	}
 	pub.js = js

pkg/messaging/nats/pubsub.go

@@ -62,7 +62,7 @@ func NewPubSub(ctx context.Context, url string, logger *slog.Logger, opts ...mes
 	if err != nil {
 		return nil, err
 	}
-	stream, err := js.CreateStream(ctx, ps.jsStreamConfig)
+	stream, err := js.CreateStream(ctx, ps.jsStreamConfig.StreamConfig)
 	if err != nil {
 		return nil, err
 	}
@@ -115,19 +115,24 @@ func (ps *pubsub) Subscribe(ctx context.Context, cfg messaging.SubscriberConfig)
 		FilterSubject: cfg.Topic,
 	}
 
-	if ps.slowConsumerConfig != nil {
-		consumerConfig.MaxAckPending = ps.slowConsumerConfig.MaxPendingMsgs
+	// Apply consumer limits from the built-in ConsumerLimits
+	if ps.jsStreamConfig.ConsumerLimits.MaxAckPending > 0 {
+		consumerConfig.MaxAckPending = ps.jsStreamConfig.ConsumerLimits.MaxAckPending
+	}
 
-		if ps.slowConsumerConfig.MaxPendingBytes > 0 {
-			consumerConfig.MaxRequestMaxBytes = ps.slowConsumerConfig.MaxPendingBytes
-		}
+	// Apply additional monitoring configuration
+	monitoring := &ps.jsStreamConfig.SlowConsumer
+	if monitoring.MaxPendingBytes > 0 {
+		consumerConfig.MaxRequestMaxBytes = monitoring.MaxPendingBytes
+	}
 
+	// Log the applied configuration
+	if ps.jsStreamConfig.ConsumerLimits.MaxAckPending > 0 {
 		ps.logger.Info("Applied slow consumer throttling to JetStream consumer",
 			slog.String("consumer", consumerConfig.Name),
 			slog.Int("max_ack_pending", consumerConfig.MaxAckPending),
 			slog.Int("max_request_max_bytes", consumerConfig.MaxRequestMaxBytes),
-			slog.Bool("dropped_msg_tracking", ps.slowConsumerConfig.EnableDroppedMsgTracking),
-			slog.Bool("slow_consumer_detection", ps.slowConsumerConfig.EnableSlowConsumerDetection),
+			slog.Bool("dropped_msg_tracking", monitoring.EnableDroppedMsgTracking),
 		)
 	}
 
@@ -151,10 +156,6 @@ func (ps *pubsub) Subscribe(ctx context.Context, cfg messaging.SubscriberConfig)
 		return fmt.Errorf("failed to consume: %w", err)
 	}
 
-	if ps.slowConsumerConfig != nil && ps.slowConsumerConfig.EnableSlowConsumerDetection {
-		go ps.checkConsumerLag(ctx, consumerConfig.Name)
-	}
-
 	return nil
 }
 
@@ -190,20 +191,7 @@ func (ps *pubsub) natsHandler(h messaging.MessageHandler) func(m jetstream.Msg)
 				slog.Uint64("consumer_seq", meta.Sequence.Consumer),
 			)
 
-			if ps.slowConsumerConfig != nil && ps.slowConsumerConfig.EnableSlowConsumerDetection {
-				sequenceLag := meta.Sequence.Stream - meta.Sequence.Consumer
-				lagThreshold := uint64(float64(ps.slowConsumerConfig.MaxPendingMsgs) * 0.7)
-				if sequenceLag > lagThreshold {
-					args = append(args,
-						slog.Uint64("sequence_lag", sequenceLag),
-						slog.Uint64("lag_threshold", lagThreshold),
-						slog.String("slow_consumer_detection", "lag_threshold_exceeded"),
-					)
-					ps.logger.Warn("JetStream slow consumer detected - sequence lag exceeds threshold", args...)
-				}
-			}
-
-			if ps.slowConsumerConfig != nil && ps.slowConsumerConfig.EnableDroppedMsgTracking {
+			if ps.jsStreamConfig.SlowConsumer.EnableDroppedMsgTracking {
 				if meta.NumDelivered > 1 {
 					args = append(args,
 						slog.Uint64("delivery_count", meta.NumDelivered),
@@ -287,40 +275,4 @@ func formatConsumerName(topic, id string) string {
 	return fmt.Sprintf("%s-%s", topic, id)
 }
 
-// checkConsumerLag checks the consumer lag and logs warnings if thresholds are exceeded.
-// This provides additional slow consumer detection for JetStream consumers.
-func (ps *pubsub) checkConsumerLag(ctx context.Context, consumerName string) {
-	if ps.slowConsumerConfig == nil || !ps.slowConsumerConfig.EnableSlowConsumerDetection {
-		return
-	}
-
-	consumer, err := ps.stream.Consumer(ctx, consumerName)
-	if err != nil {
-		ps.logger.Error("failed to get consumer for lag check",
-			slog.String("consumer", consumerName),
-			slog.String("error", err.Error()),
-		)
-		return
-	}
 
-	info, err := consumer.Info(ctx)
-	if err != nil {
-		ps.logger.Error("failed to get consumer info for lag check",
-			slog.String("consumer", consumerName),
-			slog.String("error", err.Error()),
-		)
-		return
-	}
-
-	pending := info.NumPending
-	lagThreshold := uint64(float64(ps.slowConsumerConfig.MaxPendingMsgs) * 0.7)
-	if pending > lagThreshold {
-		ps.logger.Warn("JetStream consumer lag threshold exceeded",
-			slog.String("consumer", consumerName),
-			slog.Uint64("pending_messages", pending),
-			slog.Int("ack_pending", info.NumAckPending),
-			slog.Int("redelivered", info.NumRedelivered),
-			slog.Uint64("threshold", lagThreshold),
-		)
-	}
-}