fix: recover from panics that occur during envoy gateway's reconciliation

internal/message/metrics.go

@@ -13,6 +13,11 @@ var (
 		"Current depth of watchable queue.",
 	)
 
+	panicCounter = metrics.NewCounter(
+		"watchable_panics_recovered_total",
+		"Total number of panics recovered while handling items in queue.",
+	)
+
 	watchableSubscribeDurationSeconds = metrics.NewHistogram(
 		"watchable_subscribe_duration_seconds",
 		"How long in seconds a subscribed watchable queue is handled.",

internal/message/watchutil.go

@@ -6,6 +6,8 @@
 package message
 
 import (
+	"fmt"
+	"runtime/debug"
 	"time"
 
 	"github.com/telepresenceio/watchable"
@@ -36,6 +38,28 @@ func (m Metadata) LabelValues() []metrics.LabelValue {
 	return labels
 }
 
+// handleWithCrashRecovery calls the provided handle function and gracefully recovers from any panics
+// that might occur when the handle function is called.
+func handleWithCrashRecovery[K comparable, V any](
+	handle func(updateFunc Update[K, V], errChans chan error),
+	update Update[K, V],
+	meta Metadata,
+	errChans chan error,
+) {
+	defer func() {
+		if r := recover(); r != nil {
+			logger.WithValues("runner", meta.Runner).Error(fmt.Errorf("%+v", r), "observed a panic",
+				"stackTrace", string(debug.Stack()))
+			watchableSubscribeTotal.WithFailure(metrics.ReasonError, meta.LabelValues()...).Increment()
+			panicCounter.WithFailure(metrics.ReasonError, meta.LabelValues()...).Increment()
+		}
+	}()
+	startHandleTime := time.Now()
+	handle(update, errChans)
+	watchableSubscribeTotal.WithSuccess(meta.LabelValues()...).Increment()
+	watchableSubscribeDurationSeconds.With(meta.LabelValues()...).Record(time.Since(startHandleTime).Seconds())
+}
+
 // HandleSubscription takes a channel returned by
 // watchable.Map.Subscribe() (or .SubscribeSubset()), and calls the
 // given function for each initial value in the map, and for any
@@ -57,25 +81,20 @@ func HandleSubscription[K comparable, V any](
 			watchableSubscribeTotal.WithFailure(metrics.ReasonError, meta.LabelValues()...).Increment()
 		}
 	}()
+	defer close(errChans)
 
 	if snapshot, ok := <-subscription; ok {
 		for k, v := range snapshot.State {
-			startHandleTime := time.Now()
-			handle(Update[K, V]{
+			handleWithCrashRecovery(handle, Update[K, V]{
 				Key:   k,
 				Value: v,
-			}, errChans)
-			watchableSubscribeTotal.WithSuccess(meta.LabelValues()...).Increment()
-			watchableSubscribeDurationSeconds.With(meta.LabelValues()...).Record(time.Since(startHandleTime).Seconds())
+			}, meta, errChans)
 		}
 	}
 	for snapshot := range subscription {
 		watchableDepth.With(meta.LabelValues()...).Record(float64(len(subscription)))
 		for _, update := range snapshot.Updates {
-			startHandleTime := time.Now()
-			handle(Update[K, V](update), errChans)
-			watchableSubscribeTotal.WithSuccess(meta.LabelValues()...).Increment()
-			watchableSubscribeDurationSeconds.With(meta.LabelValues()...).Record(time.Since(startHandleTime).Seconds())
+			handleWithCrashRecovery(handle, Update[K, V](update), meta, errChans)
 		}
 	}
 }

internal/message/watchutil_test.go

@@ -30,6 +30,34 @@ func TestHandleSubscriptionAlreadyClosed(t *testing.T) {
 	assert.Equal(t, 0, calls)
 }
 
+func TestPanicInSubscriptionHandler(t *testing.T) {
+	defer func() {
+		if r := recover(); r != nil {
+			assert.Fail(t, "recovered from an unexpected panic")
+		}
+	}()
+	var m watchable.Map[string, any]
+	m.Store("foo", "bar")
+
+	go func() {
+		time.Sleep(100 * time.Millisecond)
+		m.Store("baz", "qux")
+		time.Sleep(100 * time.Millisecond)
+		m.Close()
+	}()
+
+	numCalls := 0
+	message.HandleSubscription[string, any](
+		message.Metadata{Runner: "demo", Message: "demo"},
+		m.Subscribe(context.Background()),
+		func(update message.Update[string, any], errChans chan error) {
+			numCalls++
+			panic("oops " + update.Key)
+		},
+	)
+	assert.Equal(t, 2, numCalls)
+}
+
 func TestHandleSubscriptionAlreadyInitialized(t *testing.T) {
 	var m watchable.Map[string, any]
 	m.Store("foo", "bar")