From fb5cdfc7291f99edacf7043188b3cb085e54a38e Mon Sep 17 00:00:00 2001
From: Daniel Liu <139250065@qq.com>
Date: Sun, 21 Sep 2025 19:38:03 +0800
Subject: [PATCH] event, p2p/simulations/adapters: fix rare goroutine leaks
 #20657 (#1541)

Co-authored-by: Boqin Qin <bobbqqin@bupt.edu.cn>
Co-authored-by: Felix Lange <fjl@twurst.com>
---
 event/subscription.go            | 17 ++++++++---------
 event/subscription_test.go       |  2 +-
 p2p/simulations/adapters/exec.go |  2 +-
 3 files changed, 10 insertions(+), 11 deletions(-)

diff --git a/event/subscription.go b/event/subscription.go
index 3d272cdf1e..1fb4c460c5 100644
--- a/event/subscription.go
+++ b/event/subscription.go
@@ -145,7 +145,6 @@ func (s *resubscribeSub) loop() {
 func (s *resubscribeSub) subscribe() Subscription {
 	subscribed := make(chan error)
 	var sub Subscription
-retry:
 	for {
 		s.lastTry = mclock.Now()
 		ctx, cancel := context.WithCancel(context.Background())
@@ -157,19 +156,19 @@ retry:
 		select {
 		case err := <-subscribed:
 			cancel()
-			if err != nil {
-				// Subscribing failed, wait before launching the next try.
-				if s.backoffWait() {
-					return nil
+			if err == nil {
+				if sub == nil {
+					panic("event: ResubscribeFunc returned nil subscription and no error")
 				}
-				continue retry
+				return sub
 			}
-			if sub == nil {
-				panic("event: ResubscribeFunc returned nil subscription and no error")
+			// Subscribing failed, wait before launching the next try.
+			if s.backoffWait() {
+				return nil // unsubscribed during wait
 			}
-			return sub
 		case <-s.unsub:
 			cancel()
+			<-subscribed // avoid leaking the s.fn goroutine.
 			return nil
 		}
 	}
diff --git a/event/subscription_test.go b/event/subscription_test.go
index 5b8a2c8ede..c48be3aa30 100644
--- a/event/subscription_test.go
+++ b/event/subscription_test.go
@@ -102,7 +102,7 @@ func TestResubscribe(t *testing.T) {
 func TestResubscribeAbort(t *testing.T) {
 	t.Parallel()
 
-	done := make(chan error)
+	done := make(chan error, 1)
 	sub := Resubscribe(0, func(ctx context.Context) (Subscription, error) {
 		select {
 		case <-ctx.Done():
diff --git a/p2p/simulations/adapters/exec.go b/p2p/simulations/adapters/exec.go
index 1123a54727..6ef2e5b78b 100644
--- a/p2p/simulations/adapters/exec.go
+++ b/p2p/simulations/adapters/exec.go
@@ -262,7 +262,7 @@ func (n *ExecNode) Stop() error {
 	if err := n.Cmd.Process.Signal(syscall.SIGTERM); err != nil {
 		return n.Cmd.Process.Kill()
 	}
-	waitErr := make(chan error)
+	waitErr := make(chan error, 1)
 	go func() {
 		waitErr <- n.Cmd.Wait()
 	}()