From cbb13af57887400fe10b4c015acf71635ad691aa Mon Sep 17 00:00:00 2001 From: Damilola Edwards Date: Tue, 18 Aug 2026 17:41:55 +0100 Subject: [PATCH] Fix duplicate event subscriptions on retry subscribeToBeaconEvents subscribed topics one at a time and gave up on the first failure, leaving earlier topics subscribed. The retry loop then resubscribed the whole list every two seconds, so topics before a bad one built up duplicate live subscriptions forever while topics after it never subscribed at all. Now already-subscribed topics are tracked and skipped on retry, and a topic that fails is logged and skipped instead of aborting the rest of the list. --- pkg/beacon/subscriptions.go | 37 +++++++-- pkg/beacon/subscriptions_test.go | 131 +++++++++++++++++++++++++++++++ 2 files changed, 163 insertions(+), 5 deletions(-) create mode 100644 pkg/beacon/subscriptions_test.go diff --git a/pkg/beacon/subscriptions.go b/pkg/beacon/subscriptions.go index 538b3c4..6cc94a3 100644 --- a/pkg/beacon/subscriptions.go +++ b/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 diff --git a/pkg/beacon/subscriptions_test.go b/pkg/beacon/subscriptions_test.go new file mode 100644 index 0000000..2671086 --- /dev/null +++ b/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) + } +}