From 84080a60c0770a05015fc79ae49684480363382c Mon Sep 17 00:00:00 2001 From: Damilola Edwards Date: Sat, 11 Jul 2026 09:16:34 +0100 Subject: [PATCH] Remove dead code in dispatcher Unsubscribe so subscriptions are freed The subscription dispatcher's Unsubscribe returned early on the wrong condition (dispatcher != nil), which is always true for a live subscription, so the removal loop below it never ran. As a result every block, slot, epoch and finalized event subscription created by the check and generate tasks was never removed. Over a long running instance the dispatcher slice grows without bound, Fire iterates every dead entry on each event, and the leaked channels pin block objects in memory, trending to memory and cpu exhaustion. Invert the guard so an already removed subscription is the early return and a live one is actually removed. The same defect and fix apply to both the execution and consensus dispatchers. Add tests covering removal, repeated unsubscribe, and delivery after removal. --- pkg/clients/consensus/subscriptions.go | 2 +- pkg/clients/consensus/subscriptions_test.go | 56 +++++++++++++++++++++ pkg/clients/execution/subscriptions.go | 2 +- pkg/clients/execution/subscriptions_test.go | 56 +++++++++++++++++++++ 4 files changed, 114 insertions(+), 2 deletions(-) create mode 100644 pkg/clients/consensus/subscriptions_test.go create mode 100644 pkg/clients/execution/subscriptions_test.go diff --git a/pkg/clients/consensus/subscriptions.go b/pkg/clients/consensus/subscriptions.go index ecc556aa..fd066335 100644 --- a/pkg/clients/consensus/subscriptions.go +++ b/pkg/clients/consensus/subscriptions.go @@ -41,7 +41,7 @@ func (d *Dispatcher[T]) Unsubscribe(subscription *Subscription[T]) { d.mutex.Lock() defer d.mutex.Unlock() - if subscription.dispatcher != nil { + if subscription.dispatcher == nil { return } diff --git a/pkg/clients/consensus/subscriptions_test.go b/pkg/clients/consensus/subscriptions_test.go new file mode 100644 index 00000000..4e8ea4ac --- /dev/null +++ b/pkg/clients/consensus/subscriptions_test.go @@ -0,0 +1,56 @@ +package consensus + +import "testing" + +// TestDispatcherUnsubscribeRemovesSubscription verifies that Unsubscribe actually +// removes the subscription from the dispatcher. A regression here means every +// subscription leaks and Fire keeps delivering to and iterating over dead entries. +func TestDispatcherUnsubscribeRemovesSubscription(t *testing.T) { + d := &Dispatcher[int]{} + + subs := make([]*Subscription[int], 5) + for i := range subs { + subs[i] = d.Subscribe(1) + } + + if got := len(d.subscriptions); got != 5 { + t.Fatalf("after subscribing 5: got %d subscriptions, want 5", got) + } + + // remove two, including one that is not the last element + subs[1].Unsubscribe() + subs[3].Unsubscribe() + + if got := len(d.subscriptions); got != 3 { + t.Fatalf("after unsubscribing 2: got %d subscriptions, want 3", got) + } + + // unsubscribing again must be a safe no-op + subs[1].Unsubscribe() + + if got := len(d.subscriptions); got != 3 { + t.Fatalf("after double unsubscribe: got %d subscriptions, want 3", got) + } + + // Fire must only reach the still-subscribed channels + d.Fire(42) + + for i, s := range subs { + want := 0 + if i == 0 || i == 2 || i == 4 { + want = 1 + } + + if got := len(s.channel); got != want { + t.Fatalf("subscription %d: got %d buffered events, want %d", i, got, want) + } + } + + subs[0].Unsubscribe() + subs[2].Unsubscribe() + subs[4].Unsubscribe() + + if got := len(d.subscriptions); got != 0 { + t.Fatalf("after unsubscribing all: got %d subscriptions, want 0", got) + } +} diff --git a/pkg/clients/execution/subscriptions.go b/pkg/clients/execution/subscriptions.go index 66da7155..1f41f718 100644 --- a/pkg/clients/execution/subscriptions.go +++ b/pkg/clients/execution/subscriptions.go @@ -41,7 +41,7 @@ func (d *Dispatcher[T]) Unsubscribe(subscription *Subscription[T]) { d.mutex.Lock() defer d.mutex.Unlock() - if subscription.dispatcher != nil { + if subscription.dispatcher == nil { return } diff --git a/pkg/clients/execution/subscriptions_test.go b/pkg/clients/execution/subscriptions_test.go new file mode 100644 index 00000000..b0e78812 --- /dev/null +++ b/pkg/clients/execution/subscriptions_test.go @@ -0,0 +1,56 @@ +package execution + +import "testing" + +// TestDispatcherUnsubscribeRemovesSubscription verifies that Unsubscribe actually +// removes the subscription from the dispatcher. A regression here means every +// subscription leaks and Fire keeps delivering to and iterating over dead entries. +func TestDispatcherUnsubscribeRemovesSubscription(t *testing.T) { + d := &Dispatcher[int]{} + + subs := make([]*Subscription[int], 5) + for i := range subs { + subs[i] = d.Subscribe(1) + } + + if got := len(d.subscriptions); got != 5 { + t.Fatalf("after subscribing 5: got %d subscriptions, want 5", got) + } + + // remove two, including one that is not the last element + subs[1].Unsubscribe() + subs[3].Unsubscribe() + + if got := len(d.subscriptions); got != 3 { + t.Fatalf("after unsubscribing 2: got %d subscriptions, want 3", got) + } + + // unsubscribing again must be a safe no-op + subs[1].Unsubscribe() + + if got := len(d.subscriptions); got != 3 { + t.Fatalf("after double unsubscribe: got %d subscriptions, want 3", got) + } + + // Fire must only reach the still-subscribed channels + d.Fire(42) + + for i, s := range subs { + want := 0 + if i == 0 || i == 2 || i == 4 { + want = 1 + } + + if got := len(s.channel); got != want { + t.Fatalf("subscription %d: got %d buffered events, want %d", i, got, want) + } + } + + subs[0].Unsubscribe() + subs[2].Unsubscribe() + subs[4].Unsubscribe() + + if got := len(d.subscriptions); got != 0 { + t.Fatalf("after unsubscribing all: got %d subscriptions, want 0", got) + } +}