Fix duplicate event subscriptions on retry

pkg/beacon/subscriptions.go

@@ -16,6 +16,8 @@ import (
 )
 
 func (n *node) ensureBeaconSubscription(ctx context.Context) error {
+	subscribed := make(map[string]bool)
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -33,18 +35,36 @@ func (n *node) ensureBeaconSubscription(ctx context.Context) error {
 				continue
 			}
 
-			if err := n.subscribeToBeaconEvents(ctx); err != nil {
+			if err := n.subscribeToBeaconEvents(ctx, subscribed); err != nil {
 				n.log.WithError(err).Error("Failed to subscribe to beacon")
 
 				continue
 			}
 
-			return nil
+			if allTopicsSubscribed(n.options.BeaconSubscription.Topics, subscribed) {
+				return nil
+			}
+		}
+	}
+}
+
+func allTopicsSubscribed(topics EventTopics, subscribed map[string]bool) bool {
+	for _, topic := range topics {
+		if !subscribed[topic] {
+			return false
 		}
 	}
+
+	return true
 }
 
-func (n *node) subscribeToBeaconEvents(ctx context.Context) error {
+// subscribeToBeaconEvents subscribes to any topic in the configured topic
+// list that isn't already marked as subscribed in the subscribed map. A
+// topic that fails to subscribe (for example, one the client doesn't
+// support) is logged and skipped rather than aborting the remaining
+// topics, and a topic that already subscribed successfully on a previous
+// call is never subscribed again.
+func (n *node) subscribeToBeaconEvents(ctx context.Context, subscribed map[string]bool) error {
 	provider, isProvider := n.client.(eth2client.EventsProvider)
 	if !isProvider {
 		return errors.New("client does not implement eth2client.Subscriptions")
@@ -53,8 +73,11 @@ func (n *node) subscribeToBeaconEvents(ctx context.Context) error {
 	topics := n.options.BeaconSubscription.Topics
 	n.log.WithField("topics", topics).Info("Subscribing to events upstream")
 
-	// Open a new subscription for each topic.
 	for _, topic := range topics {
+		if subscribed[topic] {
+			continue
+		}
+
 		n.log.WithField("topic", topic).Info("Subscribing to event")
 
 		if err := provider.Events(ctx, &api.EventsOpts{
@@ -69,8 +92,12 @@ func (n *node) subscribeToBeaconEvents(ctx context.Context) error {
 				}
 			},
 		}); err != nil {
-			return err
+			n.log.WithError(err).WithField("topic", topic).Error("Failed to subscribe to event topic, skipping")
+
+			continue
 		}
+
+		subscribed[topic] = true
 	}
 
 	return nil

pkg/beacon/subscriptions_test.go

@@ -0,0 +1,131 @@
+package beacon
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	eapi "github.com/ethpandaops/go-eth2-client/api"
+	"github.com/sirupsen/logrus"
+)
+
+// eventsFakeClient records how many times each topic was subscribed and
+// fails deterministically for a configured bad topic, mirroring how
+// go-eth2-client validates topics client-side before opening a stream.
+type eventsFakeClient struct {
+	mu       sync.Mutex
+	calls    map[string]int
+	badTopic string
+}
+
+func (f *eventsFakeClient) Name() string    { return "fake" }
+func (f *eventsFakeClient) Address() string { return "fake://" }
+func (f *eventsFakeClient) IsActive() bool  { return true }
+func (f *eventsFakeClient) IsSynced() bool  { return true }
+
+func (f *eventsFakeClient) Events(_ context.Context, opts *eapi.EventsOpts) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+
+	for _, topic := range opts.Topics {
+		if topic == f.badTopic {
+			return errors.New("unsupported event topic " + topic)
+		}
+
+		f.calls[topic]++
+	}
+
+	return nil
+}
+
+func (f *eventsFakeClient) counts() map[string]int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+
+	out := make(map[string]int, len(f.calls))
+	for k, v := range f.calls {
+		out[k] = v
+	}
+
+	return out
+}
+
+func newSubscriptionTestNode(c *eventsFakeClient) *node {
+	log := logrus.New()
+	log.SetLevel(logrus.PanicLevel)
+
+	return &node{
+		log:     log,
+		options: DefaultOptions(),
+		config:  &Config{},
+		client:  c,
+	}
+}
+
+func TestEnsureBeaconSubscription_SkipsBadTopicAndStaysIdempotent(t *testing.T) {
+	c := &eventsFakeClient{
+		calls:    map[string]int{},
+		badTopic: "raw_event",
+	}
+
+	n := newSubscriptionTestNode(c)
+	n.options.BeaconSubscription.Enabled = true
+	n.options.BeaconSubscription.Topics = EventTopics{
+		topicBlock,
+		topicHead,
+		"raw_event", // never supported, should be skipped forever
+		topicChainReorg,
+	}
+
+	// Retries every 2s; give it enough time for several attempts.
+	ctx, cancel := context.WithTimeout(context.Background(), 7*time.Second)
+	defer cancel()
+
+	// ensureBeaconSubscription only returns once every topic subscribes,
+	// which never happens here because of the permanently bad topic, so it
+	// will return ctx.Err() when the deadline hits. That's expected.
+	_ = n.ensureBeaconSubscription(ctx)
+
+	counts := c.counts()
+
+	if counts[topicBlock] != 1 {
+		t.Fatalf("expected %q to be subscribed exactly once, got %d", topicBlock, counts[topicBlock])
+	}
+
+	if counts[topicHead] != 1 {
+		t.Fatalf("expected %q to be subscribed exactly once, got %d", topicHead, counts[topicHead])
+	}
+
+	if counts[topicChainReorg] != 1 {
+		t.Fatalf("expected %q (after the bad topic) to be subscribed exactly once, got %d",
+			topicChainReorg, counts[topicChainReorg])
+	}
+
+	if _, ok := counts["raw_event"]; ok {
+		t.Fatalf("expected the bad topic to never succeed, but it recorded a call")
+	}
+}
+
+func TestEnsureBeaconSubscription_ReturnsOnceAllTopicsSubscribe(t *testing.T) {
+	c := &eventsFakeClient{calls: map[string]int{}}
+
+	n := newSubscriptionTestNode(c)
+	n.options.BeaconSubscription.Enabled = true
+	n.options.BeaconSubscription.Topics = EventTopics{topicBlock, topicHead}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	err := n.ensureBeaconSubscription(ctx)
+	if err != nil {
+		t.Fatalf("expected ensureBeaconSubscription to return nil once all topics subscribe, got %v", err)
+	}
+
+	counts := c.counts()
+
+	if counts[topicBlock] != 1 || counts[topicHead] != 1 {
+		t.Fatalf("expected each topic subscribed exactly once, got %v", counts)
+	}
+}