fix: observability lifecycle correctness — health on unexpected exit, uptime semantics, subscribe ordering (#13)

* Initial plan

* fix: apply review feedback - lifecycle hooks, health on unexpected exit, subscribe ordering

Co-authored-by: intel352 <77607+intel352@users.noreply.github.com>

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: intel352 <77607+intel352@users.noreply.github.com>

Author: Copilot

internal/broker_module.go

@@ -68,6 +68,7 @@ func (m *brokerModule) Init() error {
 // Start is a no-op; individual per-topic streams are created on demand.
 func (m *brokerModule) Start(_ context.Context) error {
 	m.health.SetRunning(true)
+	m.metrics.MarkStarted()
 	m.log.LogStreamStart(m.transport)
 	return nil
 }
@@ -88,6 +89,7 @@ func (m *brokerModule) Stop(ctx context.Context) error {
 	m.streams = make(map[string]*service.Stream)
 
 	m.health.SetRunning(false)
+	m.metrics.MarkStopped()
 	snap := m.metrics.Snapshot()
 	m.log.LogStreamStop(snap.MessagesIn+snap.MessagesOut,
 		slog.String("transport", m.transport),
@@ -174,9 +176,19 @@ func (m *brokerModule) ensureStream(ctx context.Context, topic string) (*service
 	)
 
 	go func() {
-		if runErr := stream.Run(ctx); runErr != nil && ctx.Err() == nil {
-			metrics.RecordError()
-			log.LogStreamError(runErr, slog.String("topic", topic))
+		if runErr := stream.Run(ctx); ctx.Err() == nil {
+			// Stream exited without context cancellation; remove it from the
+			// active streams map so it can be recreated on next access.
+			m.mu.Lock()
+			delete(m.streams, topic)
+			m.mu.Unlock()
+			if runErr != nil {
+				metrics.RecordError()
+				log.LogStreamError(runErr, slog.String("topic", topic))
+			}
+			log.LogTopicEvent("stream_stopped", topic,
+				slog.String("reason", "run_exited"),
+			)
 		}
 	}()
 

internal/input_module.go

@@ -136,16 +136,20 @@ func (m *inputModule) Start(ctx context.Context) error {
 	m.cancel = cancel
 
 	m.health.SetRunning(true)
+	m.metrics.MarkStarted()
 	m.log.LogStreamStart("bento.input",
 		slog.String("target_topic", m.targetTopic),
 		slog.String("target_broker", m.targetBroker),
 	)
 
 	go func() {
 		defer close(m.done)
-		if runErr := stream.Run(runCtx); runErr != nil && runCtx.Err() == nil {
-			m.metrics.RecordError()
-			m.log.LogStreamError(runErr)
+		if runErr := stream.Run(runCtx); runCtx.Err() == nil {
+			m.health.SetRunning(false)
+			if runErr != nil {
+				m.metrics.RecordError()
+				m.log.LogStreamError(runErr)
+			}
 		}
 	}()
 
@@ -171,6 +175,7 @@ func (m *inputModule) Stop(ctx context.Context) error {
 	}
 
 	m.health.SetRunning(false)
+	m.metrics.MarkStopped()
 	snap := m.metrics.Snapshot()
 	m.log.LogStreamStop(snap.MessagesOut,
 		slog.String("target_topic", m.targetTopic),

internal/logger_test.go

@@ -189,6 +189,7 @@ func TestMetricsRecordError(t *testing.T) {
 
 func TestMetricsUptime(t *testing.T) {
 	m := newStreamMetrics()
+	m.MarkStarted()
 	time.Sleep(5 * time.Millisecond)
 	uptime := m.Uptime()
 
@@ -215,6 +216,7 @@ func TestMetricsLastMessageTime(t *testing.T) {
 
 func TestMetricsSnapshot(t *testing.T) {
 	m := newStreamMetrics()
+	m.MarkStarted()
 	m.RecordMessageIn()
 	m.RecordMessageIn()
 	m.RecordMessageOut()

internal/metrics.go

@@ -8,23 +8,42 @@ import (
 
 // StreamMetrics tracks runtime statistics for a single stream or module.
 // All counter operations are thread-safe via atomic primitives; the
-// startTime/lastMessageTime fields are guarded by a small mutex.
+// startTime/stopTime/lastMessageTime fields are guarded by a small mutex.
 type StreamMetrics struct {
 	messagesIn  atomic.Int64
 	messagesOut atomic.Int64
 	errors      atomic.Int64
 
 	mu              sync.Mutex
 	startTime       time.Time
+	stopTime        time.Time
 	lastMessageTime time.Time
 }
 
-// newStreamMetrics creates a StreamMetrics instance with the start time set
-// to the current wall clock.
+// newStreamMetrics creates a new StreamMetrics instance. Call MarkStarted
+// when the associated module begins running.
 func newStreamMetrics() *StreamMetrics {
-	m := &StreamMetrics{}
+	return &StreamMetrics{}
+}
+
+// MarkStarted records the moment the stream begins running, resetting any
+// previously frozen stop time. Call this from module Start().
+func (m *StreamMetrics) MarkStarted() {
+	m.mu.Lock()
 	m.startTime = time.Now()
-	return m
+	m.stopTime = time.Time{}
+	m.mu.Unlock()
+}
+
+// MarkStopped freezes the uptime at the current wall clock. After this call
+// Uptime() returns a stable value rather than continuing to grow. Call this
+// from module Stop().
+func (m *StreamMetrics) MarkStopped() {
+	m.mu.Lock()
+	if !m.startTime.IsZero() {
+		m.stopTime = time.Now()
+	}
+	m.mu.Unlock()
 }
 
 // RecordMessageIn increments the inbound message counter and updates
@@ -65,13 +84,18 @@ func (m *StreamMetrics) Errors() int64 {
 	return m.errors.Load()
 }
 
-// Uptime returns how long the stream has been running since Start was called.
+// Uptime returns how long the stream was running. If MarkStopped has been
+// called the duration is frozen; otherwise it grows from MarkStarted.
+// Returns 0 if MarkStarted has not been called.
 func (m *StreamMetrics) Uptime() time.Duration {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	if m.startTime.IsZero() {
 		return 0
 	}
+	if !m.stopTime.IsZero() {
+		return m.stopTime.Sub(m.startTime)
+	}
 	return time.Since(m.startTime)
 }
 
@@ -98,7 +122,11 @@ func (m *StreamMetrics) Snapshot() MetricsSnapshot {
 	m.mu.Lock()
 	uptime := time.Duration(0)
 	if !m.startTime.IsZero() {
-		uptime = time.Since(m.startTime)
+		if !m.stopTime.IsZero() {
+			uptime = m.stopTime.Sub(m.startTime)
+		} else {
+			uptime = time.Since(m.startTime)
+		}
 	}
 	lastMsg := m.lastMessageTime
 	m.mu.Unlock()

internal/output_module.go

@@ -105,22 +105,9 @@ func (m *outputModule) Start(ctx context.Context) error {
 	runCtx, cancel := context.WithCancel(context.Background())
 	m.cancel = cancel
 
-	m.health.SetRunning(true)
-	m.log.LogStreamStart("bento.output",
-		slog.String("source_topic", m.sourceTopic),
-		slog.String("source_broker", m.sourceBroker),
-	)
-
-	go func() {
-		defer close(m.done)
-		if runErr := stream.Run(runCtx); runErr != nil && runCtx.Err() == nil {
-			m.metrics.RecordError()
-			m.log.LogStreamError(runErr)
-		}
-	}()
-
-	// Subscribe to the host EventBus topic. When messages arrive, forward them
-	// to the Bento producer.
+	// Subscribe to the host EventBus topic before starting the stream. When
+	// messages arrive, forward them to the Bento producer. Subscribing first
+	// avoids leaking the stream goroutine if Subscribe returns an error.
 	producerFnRef := m.producerFn
 	metrics := m.metrics
 	log := m.log
@@ -140,9 +127,29 @@ func (m *outputModule) Start(ctx context.Context) error {
 		log.LogMessageProcessed(sourceTopic)
 		return nil
 	}); err != nil {
+		// Cancel the context since the stream goroutine was never launched.
+		cancel()
 		return fmt.Errorf("bento.output %q: subscribe to topic %q: %w", m.name, m.sourceTopic, err)
 	}
 
+	m.health.SetRunning(true)
+	m.metrics.MarkStarted()
+	m.log.LogStreamStart("bento.output",
+		slog.String("source_topic", m.sourceTopic),
+		slog.String("source_broker", m.sourceBroker),
+	)
+
+	go func() {
+		defer close(m.done)
+		if runErr := stream.Run(runCtx); runCtx.Err() == nil {
+			m.health.SetRunning(false)
+			if runErr != nil {
+				m.metrics.RecordError()
+				m.log.LogStreamError(runErr)
+			}
+		}
+	}()
+
 	return nil
 }
 
@@ -168,6 +175,7 @@ func (m *outputModule) Stop(ctx context.Context) error {
 	}
 
 	m.health.SetRunning(false)
+	m.metrics.MarkStopped()
 	snap := m.metrics.Snapshot()
 	m.log.LogStreamStop(snap.MessagesIn,
 		slog.String("source_topic", m.sourceTopic),

internal/stream_module.go

@@ -62,16 +62,20 @@ func (m *streamModule) Start(ctx context.Context) error {
 	m.cancel = cancel
 
 	m.health.SetRunning(true)
+	m.metrics.MarkStarted()
 
 	m.log.LogStreamStart("bento",
 		slog.Int("config_keys", len(m.config)),
 	)
 
 	go func() {
 		defer close(m.done)
-		if runErr := stream.Run(runCtx); runErr != nil && runCtx.Err() == nil {
-			m.metrics.RecordError()
-			m.log.LogStreamError(runErr)
+		if runErr := stream.Run(runCtx); runCtx.Err() == nil {
+			m.health.SetRunning(false)
+			if runErr != nil {
+				m.metrics.RecordError()
+				m.log.LogStreamError(runErr)
+			}
 		}
 	}()
 
@@ -97,9 +101,12 @@ func (m *streamModule) Stop(ctx context.Context) error {
 	}
 
 	m.health.SetRunning(false)
+	m.metrics.MarkStopped()
 	snap := m.metrics.Snapshot()
-	m.log.LogStreamStop(snap.Errors,
+	// Stream modules don't intercept individual messages, so messages_processed is 0.
+	m.log.LogStreamStop(0,
 		slog.Duration("uptime", snap.Uptime),
+		slog.Int64("errors", snap.Errors),
 	)
 
 	return nil