fix: add panic recovery to unprotected goroutines

intel352 · Copilot · intel352 · commit 9bf4092222ad · 2026-04-04T02:41:34.000-04:00
Prevents observer/callback panics from crashing the entire process.
Matches existing recovery patterns used in other goroutine sites.

Adds defer/recover blocks to goroutines across 20 files:
- Root: reload_orchestrator, contract_verifier, application_observer
- Scheduler: dispatcher, shutdown waiter
- EventLogger: processing loop, startup emission, shutdown waiter
- HTTPServer: runServer
- ConfigWatcher: context listener, event loop
- EventBus: memory/NATS/Kinesis/Kafka/Redis/durable/custom workers,
  shutdown waiters, async handlers, event emission, observer notification
- Cache: memory cleanup goroutine
- ReverseProxy: health checker, dry-run requests, circuit breaker and
  direct proxy goroutines
- LetsEncrypt: certificate renewal goroutine

Co-authored-by: Copilot &lt;223556219+Copilot@users.noreply.github.com&gt;
diff --git a/application_observer.go b/application_observer.go
@@ -134,6 +134,11 @@ func (app *ObservableApplication) NotifyObservers(ctx context.Context, event clo
 func (app *ObservableApplication) emitEvent(ctx context.Context, event cloudevents.Event) {
 	// Use a separate goroutine to avoid blocking application operations
 	go func() {
+		defer func() {
+			if r := recover(); r != nil {
+				app.logger.Error("panic recovered in application event emission", "error", r)
+			}
+		}()
 		if err := app.NotifyObservers(ctx, event); err != nil {
 			app.logger.Error("Failed to notify observers", "event", event.Type(), "error", err)
 		}
diff --git a/configwatcher/configwatcher.go b/configwatcher/configwatcher.go
@@ -70,6 +70,11 @@ func (w *ConfigWatcher) Start(ctx context.Context) error {
 	w.wg.Add(1)
 	go func() {
 		defer w.wg.Done()
+		defer func() {
+			if r := recover(); r != nil {
+				w.logger.Error("panic recovered in config watcher context listener", "error", r)
+			}
+		}()
 		select {
 		case <-ctx.Done():
 			_ = w.stopWatching()
@@ -117,6 +122,11 @@ func (w *ConfigWatcher) stopWatching() error {
 
 func (w *ConfigWatcher) eventLoop() {
 	defer w.wg.Done()
+	defer func() {
+		if r := recover(); r != nil {
+			w.logger.Error("panic recovered in config watcher event loop", "error", r)
+		}
+	}()
 	var timer *time.Timer
 	changedPaths := make(map[string]struct{})
 	var mu sync.Mutex
diff --git a/contract_verifier.go b/contract_verifier.go
@@ -128,6 +128,11 @@ func (v *StandardContractVerifier) runReloadWithGuard(module Reloadable, label s
 	type result struct{ err error }
 	ch := make(chan result, 1)
 	go func() {
+		defer func() {
+			if r := recover(); r != nil {
+				ch <- result{err: fmt.Errorf("Reload panicked: %v", r)}
+			}
+		}()
 		ch <- result{err: module.Reload(ctx, nil)}
 	}()
 
@@ -170,6 +175,11 @@ func (v *StandardContractVerifier) VerifyHealthContract(provider HealthProvider)
 	}
 	ch := make(chan result, 1)
 	go func() {
+		defer func() {
+			if r := recover(); r != nil {
+				ch <- result{err: fmt.Errorf("HealthCheck panicked: %v", r)}
+			}
+		}()
 		reports, err := provider.HealthCheck(ctx)
 		ch <- result{reports, err}
 	}()
diff --git a/eventlogger/module.go b/eventlogger/module.go
@@ -340,6 +340,13 @@ func (m *EventLoggerModule) Start(ctx context.Context) error {
 
 // emitStartupOperationalEvents performs the operational event emission without holding the Start mutex.
 func (m *EventLoggerModule) emitStartupOperationalEvents(ctx context.Context, sync bool, outputsLen, bufferLen int, targetConfigs []OutputTargetConfig) {
+	defer func() {
+		if r := recover(); r != nil {
+			if m.logger != nil {
+				m.logger.Error("panic recovered in event logger startup event emission", "error", r)
+			}
+		}
+	}()
 	// Protect field reads with RLock to avoid race with Stop() which writes m.started under Lock
 	m.mutex.RLock()
 	logger := m.logger
@@ -431,6 +438,13 @@ func (m *EventLoggerModule) Stop(ctx context.Context) error {
 	// Wait for processing with optional timeout
 	done := make(chan struct{})
 	go func() {
+		defer func() {
+			if r := recover(); r != nil {
+				if m.logger != nil {
+					m.logger.Error("panic recovered in event logger shutdown waiter", "error", r)
+				}
+			}
+		}()
 		m.wg.Wait()
 		close(done)
 	}()
@@ -759,6 +773,13 @@ func (m *EventLoggerModule) ObserverID() string {
 // processEvents processes events from both event channels.
 func (m *EventLoggerModule) processEvents(ctx context.Context) {
 	defer m.wg.Done()
+	defer func() {
+		if r := recover(); r != nil {
+			if m.logger != nil {
+				m.logger.Error("panic recovered in event logger processing loop", "error", r)
+			}
+		}
+	}()
 
 	flushTicker := time.NewTicker(m.config.FlushInterval)
 	defer flushTicker.Stop()
diff --git a/httpserver/module.go b/httpserver/module.go
@@ -285,6 +285,11 @@ func (m *HTTPServerModule) Start(ctx context.Context) error {
 
 // runServer starts the HTTP server with appropriate TLS configuration
 func (m *HTTPServerModule) runServer(ctx context.Context, addr string) {
+	defer func() {
+		if r := recover(); r != nil {
+			m.logger.Error("panic recovered in HTTP server", "error", r)
+		}
+	}()
 	m.logger.Info("Starting HTTP server", "address", addr)
 	var err error
 
diff --git a/modules/cache/memory.go b/modules/cache/memory.go
@@ -2,6 +2,7 @@ package cache
 
 import (
 	"context"
+	"log/slog"
 	"sync"
 	"time"
 
@@ -76,6 +77,11 @@ func (c *MemoryCache) Connect(ctx context.Context) error {
 	// Start cleanup goroutine with derived context
 	c.cleanupCtx, c.cancelFunc = context.WithCancel(ctx)
 	go func() {
+		defer func() {
+			if r := recover(); r != nil {
+				slog.Error("panic recovered in cache cleanup goroutine", "error", r)
+			}
+		}()
 		c.startCleanupTimer(c.cleanupCtx)
 	}()
 	return nil
diff --git a/modules/eventbus/custom_memory.go b/modules/eventbus/custom_memory.go
@@ -427,6 +427,11 @@ func (c *CustomMemoryEventBus) matchesTopic(eventTopic, subscriptionTopic string
 
 // handleEvents processes events for a custom subscription
 func (c *CustomMemoryEventBus) handleEvents(sub *customMemorySubscription) {
+	defer func() {
+		if r := recover(); r != nil {
+			slog.Error("panic recovered in custom memory event handler", "error", r, "topic", sub.topic)
+		}
+	}()
 	for {
 		select {
 		case <-c.ctx.Done():
@@ -469,6 +474,11 @@ func (c *CustomMemoryEventBus) handleEvents(sub *customMemorySubscription) {
 
 // metricsCollector periodically logs metrics
 func (c *CustomMemoryEventBus) metricsCollector() {
+	defer func() {
+		if r := recover(); r != nil {
+			slog.Error("panic recovered in custom memory eventbus metrics collector", "error", r)
+		}
+	}()
 	ticker := time.NewTicker(c.config.MetricsInterval)
 	defer ticker.Stop()
 
diff --git a/modules/eventbus/durable_memory.go b/modules/eventbus/durable_memory.go
@@ -235,6 +235,11 @@ func (d *DurableMemoryEventBus) Stop(ctx context.Context) error {
 
 	done := make(chan struct{})
 	go func() {
+		defer func() {
+			if r := recover(); r != nil {
+				slog.Error("panic recovered in durable memory eventbus shutdown waiter", "error", r)
+			}
+		}()
 		d.wg.Wait()
 		close(done)
 	}()
diff --git a/modules/eventbus/kafka.go b/modules/eventbus/kafka.go
@@ -267,6 +267,11 @@ func (k *KafkaEventBus) Stop(ctx context.Context) error {
 	// Wait for all workers to finish
 	done := make(chan struct{})
 	go func() {
+		defer func() {
+			if r := recover(); r != nil {
+				slog.Error("panic recovered in Kafka eventbus shutdown waiter", "error", r)
+			}
+		}()
 		k.wg.Wait()
 		close(done)
 	}()
@@ -368,6 +373,11 @@ func (k *KafkaEventBus) subscribe(ctx context.Context, topic string, handler Eve
 
 // startConsumerGroup starts the Kafka consumer group
 func (k *KafkaEventBus) startConsumerGroup() {
+	defer func() {
+		if r := recover(); r != nil {
+			slog.Error("panic recovered in Kafka consumer group", "error", r)
+		}
+	}()
 	handler := &KafkaConsumerGroupHandler{
 		eventBus:      k,
 		subscriptions: make(map[string]*kafkaSubscription),
@@ -467,5 +477,10 @@ func (k *KafkaEventBus) processEvent(sub *kafkaSubscription, event Event) {
 
 // processEventAsync processes an event asynchronously
 func (k *KafkaEventBus) processEventAsync(sub *kafkaSubscription, event Event) {
+	defer func() {
+		if r := recover(); r != nil {
+			slog.Error("panic recovered in Kafka async event handler", "error", r, "topic", event.Type())
+		}
+	}()
 	k.processEvent(sub, event)
 }
diff --git a/modules/eventbus/kinesis.go b/modules/eventbus/kinesis.go
@@ -242,6 +242,11 @@ func (k *KinesisEventBus) Stop(ctx context.Context) error {
 	// Wait for all workers to finish
 	done := make(chan struct{})
 	go func() {
+		defer func() {
+			if r := recover(); r != nil {
+				slog.Error("panic recovered in Kinesis eventbus shutdown waiter", "error", r)
+			}
+		}()
 		k.wg.Wait()
 		close(done)
 	}()
@@ -402,6 +407,11 @@ func (k *KinesisEventBus) readShard(shardID string) {
 		delete(k.activeShards, shardID)
 		k.shardMutex.Unlock()
 	}()
+	defer func() {
+		if r := recover(); r != nil {
+			slog.Error("panic recovered in Kinesis shard reader", "error", r, "shard", shardID)
+		}
+	}()
 
 	// Get shard iterator
 	iterResp, err := k.client.GetShardIterator(k.ctx, &kinesis.GetShardIteratorInput{
@@ -608,5 +618,10 @@ func (k *KinesisEventBus) processEvent(sub *kinesisSubscription, event Event) {
 
 // processEventAsync processes an event asynchronously
 func (k *KinesisEventBus) processEventAsync(sub *kinesisSubscription, event Event) {
+	defer func() {
+		if r := recover(); r != nil {
+			slog.Error("panic recovered in Kinesis async event handler", "error", r, "topic", event.Type())
+		}
+	}()
 	k.processEvent(sub, event)
 }
diff --git a/modules/eventbus/memory.go b/modules/eventbus/memory.go
@@ -155,6 +155,11 @@ func (m *MemoryEventBus) Stop(ctx context.Context) error {
 	// Wait for all workers to finish
 	done := make(chan struct{})
 	go func() {
+		defer func() {
+			if r := recover(); r != nil {
+				slog.Error("panic recovered in memory eventbus shutdown waiter", "error", r)
+			}
+		}()
 		m.wg.Wait()
 		close(done)
 	}()
@@ -525,6 +530,11 @@ func (m *MemoryEventBus) queueEventHandler(sub *memorySubscription, event Event)
 // worker is a goroutine that processes events from the worker pool
 func (m *MemoryEventBus) worker() {
 	defer m.wg.Done()
+	defer func() {
+		if r := recover(); r != nil {
+			slog.Error("panic recovered in memory eventbus worker", "error", r)
+		}
+	}()
 
 	for {
 		select {
diff --git a/modules/eventbus/module.go b/modules/eventbus/module.go
@@ -664,6 +664,13 @@ func (m *EventBusModule) EmitEvent(ctx context.Context, event cloudevents.Event)
 	}
 	// Use a goroutine to prevent blocking eventbus operations with event emission
 	go func(s modular.Subject, e cloudevents.Event) {
+		defer func() {
+			if r := recover(); r != nil {
+				if m.logger != nil {
+					m.logger.Error("panic recovered in eventbus observer notification", "error", r, "event_type", e.Type())
+				}
+			}
+		}()
 		if err := s.NotifyObservers(ctx, e); err != nil {
 			if m.logger != nil {
 				m.logger.Debug("Failed to notify observers", "error", err, "event_type", e.Type())
@@ -678,6 +685,13 @@ func (m *EventBusModule) EmitEvent(ctx context.Context, event cloudevents.Event)
 // If no subject is available for event emission, it silently skips the event emission
 // to avoid noisy error messages in tests and non-observable applications.
 func (m *EventBusModule) emitEvent(ctx context.Context, eventType string, data map[string]interface{}) {
+	defer func() {
+		if r := recover(); r != nil {
+			if m.logger != nil {
+				m.logger.Error("panic recovered in eventbus event emission", "error", r, "event_type", eventType)
+			}
+		}
+	}()
 	// Skip event emission if no subject is available (non-observable application)
 	m.mutex.RLock()
 	subj := m.subject
diff --git a/modules/eventbus/nats.go b/modules/eventbus/nats.go
@@ -209,6 +209,11 @@ func (n *NatsEventBus) Stop(ctx context.Context) error {
 	// Wait for all workers to finish
 	done := make(chan struct{})
 	go func() {
+		defer func() {
+			if r := recover(); r != nil {
+				slog.Error("panic recovered in NATS eventbus shutdown waiter", "error", r)
+			}
+		}()
 		n.wg.Wait()
 		close(done)
 	}()
diff --git a/modules/eventbus/redis.go b/modules/eventbus/redis.go
@@ -172,6 +172,11 @@ func (r *RedisEventBus) Stop(ctx context.Context) error {
 	// Wait for all workers to finish
 	done := make(chan struct{})
 	go func() {
+		defer func() {
+			if r := recover(); r != nil {
+				slog.Error("panic recovered in Redis eventbus shutdown waiter", "error", r)
+			}
+		}()
 		r.wg.Wait()
 		close(done)
 	}()
@@ -328,6 +333,11 @@ func (r *RedisEventBus) SubscriberCount(topic string) int {
 // handleMessages processes messages for a Redis subscription
 func (r *RedisEventBus) handleMessages(sub *redisSubscription) {
 	defer r.wg.Done()
+	defer func() {
+		if r := recover(); r != nil {
+			slog.Error("panic recovered in Redis message handler", "error", r, "topic", sub.topic)
+		}
+	}()
 
 	ch := sub.pubsub.Channel()
 
@@ -374,5 +384,10 @@ func (r *RedisEventBus) processEvent(sub *redisSubscription, event Event) {
 
 // processEventAsync processes an event asynchronously
 func (r *RedisEventBus) processEventAsync(sub *redisSubscription, event Event) {
+	defer func() {
+		if r := recover(); r != nil {
+			slog.Error("panic recovered in Redis async event handler", "error", r, "topic", event.Type())
+		}
+	}()
 	r.processEvent(sub, event)
 }
diff --git a/modules/letsencrypt/module.go b/modules/letsencrypt/module.go
@@ -711,6 +711,11 @@ func (m *LetsEncryptModule) startRenewalTimer(ctx context.Context) {
 	m.renewalWg.Add(1)
 	go func() {
 		defer m.renewalWg.Done()
+		defer func() {
+			if r := recover(); r != nil {
+				m.logger.Error("panic recovered in certificate renewal goroutine", "error", r)
+			}
+		}()
 		for {
 			select {
 			case <-m.renewalTicker.C:
diff --git a/modules/reverseproxy/dryrun.go b/modules/reverseproxy/dryrun.go
@@ -139,6 +139,12 @@ func (d *DryRunHandler) ProcessDryRun(ctx context.Context, req *http.Request, pr
 
 	// Send request to primary backend
 	go func() {
+		defer func() {
+			if r := recover(); r != nil {
+				d.logger.Error("panic recovered in dry-run primary request", "error", r)
+				primaryChan <- ResponseInfo{Error: fmt.Sprintf("panic: %v", r)}
+			}
+		}()
 		primaryStart := time.Now()
 		response := d.sendRequest(ctx, req, primaryBackend, requestBody)
 		response.ResponseTime = time.Since(primaryStart)
@@ -147,6 +153,12 @@ func (d *DryRunHandler) ProcessDryRun(ctx context.Context, req *http.Request, pr
 
 	// Send request to secondary backend
 	go func() {
+		defer func() {
+			if r := recover(); r != nil {
+				d.logger.Error("panic recovered in dry-run secondary request", "error", r)
+				secondaryChan <- ResponseInfo{Error: fmt.Sprintf("panic: %v", r)}
+			}
+		}()
 		secondaryStart := time.Now()
 		response := d.sendRequest(ctx, req, secondaryBackend, requestBody)
 		response.ResponseTime = time.Since(secondaryStart)
diff --git a/modules/reverseproxy/health_checker.go b/modules/reverseproxy/health_checker.go
diff --git a/modules/reverseproxy/module.go b/modules/reverseproxy/module.go
diff --git a/reload_orchestrator.go b/reload_orchestrator.go
diff --git a/scheduler/scheduler.go b/scheduler/scheduler.go
