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) + } +}