✨ Update Go module dependencies and enhance NATS consumer metrics. Add github.com/kylelemons/godebug as an indirect dependency. Upgrade github.com/prometheus/common to version 0.67.3. Change NATS consumer mode to "jetstream" in the development configuration. Improve observability by ensuring metrics for pending messages are initialized and recorded correctly in JetStream mode. Update documentation to reflect changes in metrics handling and consumer behavior.

configs/config.dev.toml

@@ -15,7 +15,7 @@ url = "nats://localhost:4222"
 #     * Dead-letter queue (DLQ) support
 #     * Batch processing and consumer monitoring
 #     * Recommended for production environments
-mode = "core"
+mode = "jetstream"
 client = "serial-client"
 cluster = "tele-cluster"
 stream = "TELEGRAM"

docs/observability.md

@@ -32,7 +32,7 @@ The service exposes Prometheus metrics via the monitoring HTTP server (default `
   Count of general publish failures (not DLQ-specific), labelled by message category.
 
 - `caatsm_nats_consumer_pending_messages{stream,consumer}`  
-  Current pending message count for each JetStream consumer (useful for lag/backlog alerts).
+  Current pending message count for each JetStream consumer (useful for lag/backlog alerts). A zero sample is emitted only once at startup so the series exists; if JetStream stats queries fail later, the last known value is preserved rather than force-setting the gauge to `0`, which prevents false “queue cleared” alerts.
 
 Additional OTEL metrics are emitted via the configured OTEL endpoint, including:
 
@@ -282,4 +282,3 @@ Logging is done with Zap. The `internal/infra/log` package standardises fields v
 
 Handler and consumer logs should always be emitted through `WithMessageContext` to ensure these fields are present where applicable.
 
-

docs/reliability.md

@@ -107,14 +107,16 @@ Recommended pattern:
   restarted and returns `ErrNoResponders`), the consumer uses an exponential
   backoff when retrying `Fetch` calls (roughly `1s, 2s, 4s, ...` up to
   around `30s`) to avoid log spam while allowing the system to recover.
-- In dev/test environments, if the stream or consumer is detected as missing at
-  runtime (for example after `docker compose down -v`), the consumer calls the
-  shared `EnsureStream` and `ensureConsumer` logic to recreate them and
-  re-establish subscriptions.
-- In production environments, missing streams/consumers are treated as
-  configuration or operational errors:
-  - They are **not** auto-recreated.
-  - Errors are logged prominently so operators can diagnose and fix the issue.
+- At startup the receiver always calls `StreamManager.EnsureStream` and
+  `ConsumerManager.EnsureConsumer`. If the JetStream account allows it,
+  missing streams are created with the configured retention limits
+  (`max_msgs`, `max_bytes`, `max_age`, discard/storage policy, replicas) before
+  the durable consumer is created. This keeps dev/test clusters self-healing
+  after `docker compose down -v` and removes the race where a consumer was
+  created without its stream.
+- When the JetStream account lacks permissions to create protected resources
+  (a common production posture), the same code path fails fast with a clear
+  error message so operators know they must provision the stream out-of-band.
 - On the publishing side, JetStream `ErrNoResponders` and similar errors are
   treated as temporary by the processor:
   - Such errors cause the consumer to NAK messages and rely on the configured
@@ -142,4 +144,3 @@ Dashboards should combine:
 - NATS consumer statistics (pending, redelivered, ack_pending).  
 - DB health indicators (latency, error counts, connection usage).
 
-

go.mod

@@ -63,6 +63,7 @@ require (
 	github.com/jackc/puddle/v2 v2.2.2 // indirect
 	github.com/klauspost/compress v1.18.1 // indirect
 	github.com/knadh/koanf/maps v0.1.2 // indirect
+	github.com/kylelemons/godebug v1.1.0 // indirect
 	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
 	github.com/magiconair/properties v1.8.7 // indirect
 	github.com/mitchellh/copystructure v1.2.0 // indirect
@@ -82,7 +83,7 @@ require (
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
 	github.com/prometheus/client_model v0.6.2 // indirect
-	github.com/prometheus/common v0.67.2 // indirect
+	github.com/prometheus/common v0.67.3 // indirect
 	github.com/prometheus/procfs v0.19.2 // indirect
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
 	github.com/shirou/gopsutil/v3 v3.23.12 // indirect

internal/infra/metrics/metrics.go

@@ -255,7 +255,21 @@ func RecordJSAPICall(operation string) {
 // JetStream consumer as a gauge, enabling backlog / lag alerts.
 func RecordNATSConsumerPending(stream, consumer string, pending uint64) {
 	ensureCollectors()
-	natsConsumerPending.WithLabelValues(labelValue(stream), labelValue(consumer)).Set(float64(pending))
+	
+	// Validate inputs to ensure metric is recorded correctly
+	streamLabel := labelValue(stream)
+	consumerLabel := labelValue(consumer)
+	
+	// Ensure metric is always set, even with empty labels (will be "unknown")
+	if streamLabel == "" {
+		streamLabel = "unknown"
+	}
+	if consumerLabel == "" {
+		consumerLabel = "unknown"
+	}
+	
+	// Set the metric value
+	natsConsumerPending.WithLabelValues(streamLabel, consumerLabel).Set(float64(pending))
 }
 
 func labelValue(value string) string {

internal/infra/metrics/metrics_test.go

@@ -3,9 +3,12 @@ package metrics
 import (
 	"net/http"
 	"net/http/httptest"
+	"sync"
 	"testing"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus/testutil"
+
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 )
@@ -16,6 +19,10 @@ func TestMetrics(t *testing.T) {
 }
 
 var _ = Describe("Metrics", func() {
+	BeforeEach(func() {
+		resetMetricsForTest()
+	})
+
 	Describe("Handler", func() {
 		It("serves metrics with the correct content type", func() {
 			req := httptest.NewRequest(http.MethodGet, "/metrics", nil)
@@ -29,25 +36,69 @@ var _ = Describe("Metrics", func() {
 	})
 
 	Describe("labelValue helper", func() {
-		It("returns unknown for empty values and lowercases input", func() {
+		It("normalizes empty and uppercase values", func() {
 			Expect(labelValue("   ")).To(Equal("unknown"))
 			Expect(labelValue("SOME_VALUE")).To(Equal("some_value"))
 		})
 	})
 
 	Describe("record helpers", func() {
-		It("can be invoked without panicking", func() {
-			Expect(func() {
-				RecordProcessed("parsed", "ARR", 150*time.Millisecond)
+		It("record counters and histograms for message processing", func() {
+			RecordProcessed("PARSED", "ARR", 150*time.Millisecond)
 			RecordFailure("parser")
-				RecordMessageHandled("TEST", "consumer", ResultOK, 205*time.Millisecond)
-				RecordRetry("TEST", "consumer", RetryReasonProcessorError)
-				RecordDLQMessage("TEST", "consumer")
-				RecordDLQPublishFailure("TEST", "consumer")
-				RecordDBQuery("insert", DBResultOK, 10*time.Millisecond)
+			RecordMessageHandled("stream1", "consumer1", ResultOK, 80*time.Millisecond)
+			RecordRetry("stream1", "consumer1", RetryReasonProcessorError)
+			RecordDLQMessage("stream1", "consumer1")
+			RecordDLQPublishFailure("stream1", "consumer1")
+			RecordPublishFailure("flightPlan")
+			RecordDBQuery("insert_one", DBResultOK, 10*time.Millisecond)
 			RecordJSAPICall("publish")
-				RecordNATSConsumerPending("TEST", "consumer", 7)
-			}).NotTo(Panic())
+
+			Expect(testutil.ToFloat64(processedCounter.WithLabelValues("parsed", "arr"))).To(BeNumerically("==", 1))
+			Expect(testutil.CollectAndCount(parseLatency, MetricParseLatencySeconds)).To(Equal(1))
+
+			Expect(testutil.ToFloat64(failureCounter.WithLabelValues("parser"))).To(BeNumerically("==", 1))
+
+			Expect(testutil.ToFloat64(messagesTotal.WithLabelValues("stream1", "consumer1", ResultOK))).To(BeNumerically("==", 1))
+			Expect(testutil.CollectAndCount(handleLatency, MetricHandleLatencySeconds)).To(Equal(1))
+
+			Expect(testutil.ToFloat64(retriesTotal.WithLabelValues("stream1", "consumer1", RetryReasonProcessorError))).To(BeNumerically("==", 1))
+
+			Expect(testutil.ToFloat64(dlqMessagesTotal.WithLabelValues("stream1", "consumer1"))).To(BeNumerically("==", 1))
+			Expect(testutil.ToFloat64(dlqPublishFailures.WithLabelValues("stream1", "consumer1"))).To(BeNumerically("==", 1))
+
+			Expect(testutil.ToFloat64(publishFailuresTotal.WithLabelValues("flightplan"))).To(BeNumerically("==", 1))
+
+			Expect(testutil.ToFloat64(dbQueriesTotal.WithLabelValues("insert_one", DBResultOK))).To(BeNumerically("==", 1))
+			Expect(testutil.CollectAndCount(dbQueryLatency, MetricDBQueryLatencySeconds)).To(Equal(1))
+
+			Expect(testutil.ToFloat64(jsAPICallsTotal.WithLabelValues("publish"))).To(BeNumerically("==", 1))
+		})
+
+		It("records gauge values for pending messages", func() {
+			RecordNATSConsumerPending("STREAM_A", "consumerA", 42)
+			Expect(testutil.ToFloat64(natsConsumerPending.WithLabelValues("stream_a", "consumera"))).To(BeNumerically("==", 42))
+
+			RecordNATSConsumerPending("STREAM_A", "consumerA", 5)
+			Expect(testutil.ToFloat64(natsConsumerPending.WithLabelValues("stream_a", "consumera"))).To(BeNumerically("==", 5))
 		})
 	})
 })
+
+func resetMetricsForTest() {
+	once = sync.Once{}
+	registry = nil
+	processedCounter = nil
+	failureCounter = nil
+	parseLatency = nil
+	messagesTotal = nil
+	handleLatency = nil
+	retriesTotal = nil
+	jsAPICallsTotal = nil
+	dlqMessagesTotal = nil
+	dlqPublishFailures = nil
+	publishFailuresTotal = nil
+	dbQueriesTotal = nil
+	dbQueryLatency = nil
+	natsConsumerPending = nil
+}

internal/infra/nats/consumer.go

@@ -76,6 +76,27 @@ func ProvideConsumer(
 	}
 	consumer.initCollaborators()
 
+	// Initialize the pending messages metric early (set to 0) so it appears in Prometheus
+	// even before the consumer starts. This ensures the metric is always visible.
+	// We do this first, before any operations that might fail, to ensure the metric exists.
+	// Initialize the metric unconditionally when in JetStream mode, even if js is nil,
+	// as it will be updated later when js becomes available.
+	if normCfg.mode == "jetstream" {
+		logger.Info("Initializing NATS consumer pending messages metric",
+			zap.String("stream", normCfg.streamName),
+			zap.String("consumer", normCfg.consumerName),
+			zap.Uint64("pending", 0),
+			zap.Bool("js_available", js != nil),
+		)
+		// Always initialize the metric in JetStream mode to ensure it appears in Prometheus
+		// The metric will be updated with actual values when the consumer starts
+		obsmetrics.RecordNATSConsumerPending(normCfg.streamName, normCfg.consumerName, 0)
+	} else {
+		logger.Debug("Skipping NATS consumer pending messages metric initialization (not JetStream mode)",
+			zap.String("mode", normCfg.mode),
+		)
+	}
+
 	// Initialize managers
 	if consumer.config.mode == "jetstream" {
 		consumer.consumerManager = NewConsumerManager(js, normCfg.streamName, normCfg.consumerName, normCfg.subject, logger)
@@ -93,6 +114,19 @@ func ProvideConsumer(
 			fetcher.streamManager = consumer.streamManager
 		}
 
+		// Ensure stream exists before creating consumer
+		streamCfg := &StreamConfig{
+			MaxMsgs:  cfg.NATS.StreamLimits.MaxMsgs,
+			MaxBytes: cfg.NATS.StreamLimits.MaxBytes,
+			MaxAge:   cfg.NATS.StreamLimits.MaxAge,
+			Discard:  cfg.NATS.StreamLimits.Discard,
+			Storage:  cfg.NATS.StreamLimits.Storage,
+			Replicas: cfg.NATS.StreamLimits.Replicas,
+		}
+		if err := consumer.streamManager.EnsureStream(streamCfg); err != nil {
+			return nil, fmt.Errorf("failed to ensure stream: %w", err)
+		}
+
 		// Create consumer if it doesn't exist
 		consumerConfig := consumer.buildConsumerConfig()
 		if err := consumer.consumerManager.EnsureConsumer(consumerConfig); err != nil {
@@ -269,14 +303,9 @@ func (c *Consumer) startCore(ctx context.Context) error {
 	}
 
 	handler := func(msg *nats.Msg) {
-		if err := c.batchProcessor.ProcessMessage(ctx, msg); err != nil {
-			isPermanent := app.IsPermanent(err)
-			c.logger.Error("Failed to process message (core mode)",
-				zap.String("subject", msg.Subject),
-				zap.Error(err),
-				zap.Bool("permanent", isPermanent),
-			)
-		}
+		// Use ProcessBatch to ensure metrics are recorded via processSingleMessage
+		// ProcessBatch handles error recording and metrics for both success and failure cases
+		c.batchProcessor.ProcessBatch(ctx, []*nats.Msg{msg})
 	}
 
 	sub, err := c.conn.QueueSubscribe(c.config.subject, queueGroup, handler)
@@ -335,6 +364,23 @@ func (c *Consumer) startJetStream(ctx context.Context) error {
 		zap.String("stream", c.config.streamName),
 	)
 
+	// Record initial pending messages metric immediately
+	// This ensures the metric appears in Prometheus right away
+	if info, err := c.js.ConsumerInfo(c.config.streamName, c.config.consumerName); err == nil {
+		c.logger.Info("Recording initial NATS consumer pending messages metric",
+			zap.String("stream", c.config.streamName),
+			zap.String("consumer", c.config.consumerName),
+			zap.Uint64("pending", info.NumPending),
+		)
+		obsmetrics.RecordNATSConsumerPending(c.config.streamName, c.config.consumerName, info.NumPending)
+	} else {
+		c.logger.Warn("Failed to fetch initial consumer info for pending messages metric",
+			zap.String("stream", c.config.streamName),
+			zap.String("consumer", c.config.consumerName),
+			zap.Error(err),
+		)
+	}
+
 	statsCtx, statsCancel := context.WithCancel(ctx)
 	defer statsCancel()
 	go c.emitConsumerStats(statsCtx)
@@ -379,6 +425,14 @@ func (c *Consumer) emitConsumerStats(ctx context.Context) {
 	ticker := time.NewTicker(c.config.monitorInterval)
 	defer ticker.Stop()
 
+	// Record initial metric (0) to ensure it appears in Prometheus even before first tick
+	c.logger.Info("Starting NATS consumer stats emission goroutine, recording initial pending metric",
+		zap.String("stream", c.config.streamName),
+		zap.String("consumer", c.config.consumerName),
+		zap.Duration("interval", c.config.monitorInterval),
+	)
+	obsmetrics.RecordNATSConsumerPending(c.config.streamName, c.config.consumerName, 0)
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -386,9 +440,19 @@ func (c *Consumer) emitConsumerStats(ctx context.Context) {
 		case <-ticker.C:
 			info, err := c.js.ConsumerInfo(c.config.streamName, c.config.consumerName)
 			if err != nil {
+				c.logger.Warn("Failed to fetch consumer info for pending messages metric",
+					zap.String("stream", c.config.streamName),
+					zap.String("consumer", c.config.consumerName),
+					zap.Error(err),
+				)
 				continue
 			}
 			// Record pending messages for monitoring
+			c.logger.Debug("Recording NATS consumer pending messages metric",
+				zap.String("stream", c.config.streamName),
+				zap.String("consumer", c.config.consumerName),
+				zap.Uint64("pending", info.NumPending),
+			)
 			obsmetrics.RecordNATSConsumerPending(c.config.streamName, c.config.consumerName, info.NumPending)
 		}
 	}

internal/infra/nats/message_processor.go

@@ -74,11 +74,19 @@ func (p *defaultBatchProcessor) processSingleMessage(ctx context.Context, msg *n
 		*p.consecutiveProcessErrors = 0
 	}
 
-	// ACK the message
+	elapsed := time.Since(start)
+
+	// ACK the message (only in JetStream mode; Core NATS doesn't support ACK)
+	if p.mode == "jetstream" {
 		if ackErr := msg.Ack(); ackErr != nil {
 			p.logger.Error("Failed to ACK message", zap.Error(ackErr))
+			// Still record metrics even if ACK fails
+			p.telemetry.RecordMessageHandled(ctx, p.streamName, p.consumerName, "ok", elapsed)
 		} else {
-		elapsed := time.Since(start)
+			p.telemetry.RecordMessageHandled(ctx, p.streamName, p.consumerName, "ok", elapsed)
+		}
+	} else {
+		// Core NATS mode: record metrics without ACK (ACK not supported)
 		p.telemetry.RecordMessageHandled(ctx, p.streamName, p.consumerName, "ok", elapsed)
 	}
 }
@@ -206,6 +214,14 @@ func (p *defaultBatchProcessor) handlePermanentError(ctx context.Context, msg *n
 	if p.consecutiveProcessErrors != nil {
 		*p.consecutiveProcessErrors = 0
 	}
+
+	if p.mode != "jetstream" {
+		p.logger.Debug("Permanent-error message in core mode; skipping DLQ/ACK (not supported)",
+			zap.String("subject", msg.Subject),
+		)
+		return
+	}
+
 	// Poison/permanent message: route to DLQ if configured, then ACK
 	if p.dlqHandler != nil {
 		if dlqErr := p.dlqHandler.RouteToDLQ(ctx, msg, err); dlqErr != nil {
@@ -243,6 +259,13 @@ func (p *defaultBatchProcessor) handleTransientError(ctx context.Context, msg *n
 		}
 	}
 
+	if p.mode != "jetstream" {
+		p.logger.Debug("Transient-error message in core mode; skipping retry (ACK/NAK unsupported)",
+			zap.String("subject", msg.Subject),
+		)
+		return
+	}
+
 	// Transient error: request redelivery with optional delay
 	p.telemetry.RecordRetry(ctx, p.streamName, p.consumerName, obsmetrics.RetryReasonProcessorError)
 	if nakErr := p.nakWithStrategy(msg); nakErr != nil {

internal/infra/nats/stream_manager.go

@@ -1,7 +1,9 @@
 package nats
 
 import (
+	"errors"
 	"fmt"
+	"time"
 
 	"github.com/nats-io/nats.go"
 	"go.uber.org/zap"
@@ -15,6 +17,16 @@ type StreamManager struct {
 	logger     *zap.Logger
 }
 
+// StreamConfig holds configuration for creating a JetStream stream
+type StreamConfig struct {
+	MaxMsgs   int64
+	MaxBytes  int64
+	MaxAge    time.Duration
+	Discard   string // "old" or "new"
+	Storage   string // "file" or "memory"
+	Replicas  int
+}
+
 // NewStreamManager creates a new stream manager
 func NewStreamManager(js nats.JetStreamContext, streamName string, subjects []string, logger *zap.Logger) *StreamManager {
 	return &StreamManager{
@@ -25,16 +37,63 @@ func NewStreamManager(js nats.JetStreamContext, streamName string, subjects []st
 	}
 }
 
-// EnsureStream ensures that the configured JetStream stream exists
-func (sm *StreamManager) EnsureStream() error {
+// EnsureStream ensures that the configured JetStream stream exists, creating it if necessary
+func (sm *StreamManager) EnsureStream(cfg *StreamConfig) error {
+	// Check if stream already exists
 	_, err := sm.js.StreamInfo(sm.streamName)
-	if err != nil {
-		return fmt.Errorf("stream %s not found or inaccessible: %w", sm.streamName, err)
-	}
-
+	if err == nil {
 		sm.logger.Info("JetStream stream verified",
 			zap.String("stream", sm.streamName),
 			zap.Strings("subjects", sm.subjects),
 		)
 		return nil
 	}
+
+	// If stream doesn't exist, create it
+	if errors.Is(err, nats.ErrStreamNotFound) {
+		streamCfg := &nats.StreamConfig{
+			Name:     sm.streamName,
+			Subjects: sm.subjects,
+		}
+
+		// Apply limits if provided
+		if cfg != nil {
+			if cfg.MaxMsgs > 0 {
+				streamCfg.MaxMsgs = cfg.MaxMsgs
+			}
+			if cfg.MaxBytes > 0 {
+				streamCfg.MaxBytes = cfg.MaxBytes
+			}
+			if cfg.MaxAge > 0 {
+				streamCfg.MaxAge = cfg.MaxAge
+			}
+			if cfg.Discard == "new" {
+				streamCfg.Discard = nats.DiscardNew
+			} else {
+				streamCfg.Discard = nats.DiscardOld
+			}
+			if cfg.Storage == "memory" {
+				streamCfg.Storage = nats.MemoryStorage
+			} else {
+				streamCfg.Storage = nats.FileStorage
+			}
+			if cfg.Replicas > 0 {
+				streamCfg.Replicas = cfg.Replicas
+			}
+		}
+
+		_, err := sm.js.AddStream(streamCfg)
+		if err != nil {
+			return fmt.Errorf("failed to create stream %s: %w", sm.streamName, err)
+		}
+
+		sm.logger.Info("Created JetStream stream",
+			zap.String("stream", sm.streamName),
+			zap.Strings("subjects", sm.subjects),
+		)
+		return nil
+	}
+
+	// Other error (e.g., permission denied)
+	return fmt.Errorf("stream %s not found or inaccessible: %w", sm.streamName, err)
+}

test/integration/nats_recovery_test.go

@@ -39,7 +39,7 @@ func TestNATSConsumerRecovery(t *testing.T) {
 
 	// Test stream recovery
 	streamManager := natsinfra.NewStreamManager(js, "TEST_STREAM", []string{"test.subject"}, nil)
-	err = streamManager.EnsureStream()
+	err = streamManager.EnsureStream(nil) // nil uses default stream configuration
 	assert.NoError(t, err)
 
 	// Test consumer recovery