eventbus: dont panic on closing Subscription twice (#3034)

libp2p · Nov 13, 2024 · 7268c98 · 7268c98
1 parent 88e2f7a
commit 7268c98
Show file tree

Hide file tree

Showing 2 changed files with 32 additions and 19 deletions.
diff --git a/p2p/host/eventbus/basic.go b/p2p/host/eventbus/basic.go
@@ -155,6 +155,7 @@ type sub struct {
 	dropper       func(reflect.Type)
 	metricsTracer MetricsTracer
 	name          string
+	closeOnce     sync.Once
 }
 
 func (s *sub) Name() string {
@@ -172,31 +173,32 @@ func (s *sub) Close() error {
 		for range s.ch {
 		}
 	}()
-
-	for _, n := range s.nodes {
-		n.lk.Lock()
-
-		for i := 0; i < len(n.sinks); i++ {
-			if n.sinks[i].ch == s.ch {
-				n.sinks[i], n.sinks[len(n.sinks)-1] = n.sinks[len(n.sinks)-1], nil
-				n.sinks = n.sinks[:len(n.sinks)-1]
-
-				if s.metricsTracer != nil {
-					s.metricsTracer.RemoveSubscriber(n.typ)
+	s.closeOnce.Do(func() {
+		for _, n := range s.nodes {
+			n.lk.Lock()
+
+			for i := 0; i < len(n.sinks); i++ {
+				if n.sinks[i].ch == s.ch {
+					n.sinks[i], n.sinks[len(n.sinks)-1] = n.sinks[len(n.sinks)-1], nil
+					n.sinks = n.sinks[:len(n.sinks)-1]
+
+					if s.metricsTracer != nil {
+						s.metricsTracer.RemoveSubscriber(n.typ)
+					}
+					break
 				}
-				break
 			}
-		}
 
-		tryDrop := len(n.sinks) == 0 && n.nEmitters.Load() == 0
+			tryDrop := len(n.sinks) == 0 && n.nEmitters.Load() == 0
 
-		n.lk.Unlock()
+			n.lk.Unlock()
 
-		if tryDrop {
-			s.dropper(n.typ)
+			if tryDrop {
+				s.dropper(n.typ)
+			}
 		}
-	}
-	close(s.ch)
+		close(s.ch)
+	})
 	return nil
 }
 

diff --git a/p2p/host/eventbus/basic_test.go b/p2p/host/eventbus/basic_test.go
@@ -562,6 +562,17 @@ func TestSubFailFully(t *testing.T) {
 	}
 }
 
+func TestSubCloseMultiple(t *testing.T) {
+	bus := NewBus()
+
+	sub, err := bus.Subscribe([]interface{}{new(EventB)})
+	require.NoError(t, err)
+	err = sub.Close()
+	require.NoError(t, err)
+	err = sub.Close()
+	require.NoError(t, err)
+}
+
 func testMany(t testing.TB, subs, emits, msgs int, stateful bool) {
 	if race.WithRace() && subs+emits > 5000 {
 		t.SkipNow()