From 4964f1953ed7d0e013e492d488869eb5867e69d6 Mon Sep 17 00:00:00 2001
From: inphi <mlaw2501@gmail.com>
Date: Mon, 16 Oct 2023 23:02:28 -0400
Subject: [PATCH] event: fix deadlock during Unsubscribe

A goroutine is used to manage the lifetime of subscriptions managed by
resubscriptions. When the subscription ends with no error, the resub
goroutine ends as well. However, the resub goroutine needs to live
long enough to read from the unsub channel. Otheriwse, an Unsubscribe
call deadlocks when writing to the unsub channel.
---
 event/subscription.go      | 17 ++++++++++++-----
 event/subscription_test.go | 20 ++++++++++++++++++++
 2 files changed, 32 insertions(+), 5 deletions(-)

diff --git a/event/subscription.go b/event/subscription.go
index 6c62874719..1c1626b165 100644
--- a/event/subscription.go
+++ b/event/subscription.go
@@ -153,14 +153,21 @@ func (s *resubscribeSub) Err() <-chan error {
 }
 
 func (s *resubscribeSub) loop() {
-	defer close(s.err)
 	var done bool
+	var unsubbed bool
+	defer func() {
+		close(s.err)
+		// Read unsub chan to avoid blocking Unsubscribe
+		if !unsubbed {
+			<-s.unsub
+		}
+	}()
 	for !done {
 		sub := s.subscribe()
 		if sub == nil {
 			break
 		}
-		done = s.waitForError(sub)
+		done, unsubbed = s.waitForError(sub)
 		sub.Unsubscribe()
 	}
 }
@@ -197,14 +204,14 @@ func (s *resubscribeSub) subscribe() Subscription {
 	}
 }
 
-func (s *resubscribeSub) waitForError(sub Subscription) bool {
+func (s *resubscribeSub) waitForError(sub Subscription) (bool, bool) {
 	defer sub.Unsubscribe()
 	select {
 	case err := <-sub.Err():
 		s.lastSubErr = err
-		return err == nil
+		return err == nil, false
 	case <-s.unsub:
-		return true
+		return true, true
 	}
 }
 
diff --git a/event/subscription_test.go b/event/subscription_test.go
index ba081705c4..06b4a4941b 100644
--- a/event/subscription_test.go
+++ b/event/subscription_test.go
@@ -154,3 +154,23 @@ func TestResubscribeWithErrorHandler(t *testing.T) {
 		t.Fatalf("unexpected subscription errors %v, want %v", subErrs, expectedSubErrs)
 	}
 }
+
+func TestResubscribeWithCompletedSubscription(t *testing.T) {
+	t.Parallel()
+
+	innerSubDone := make(chan struct{}, 1)
+	sub := ResubscribeErr(100*time.Millisecond, func(ctx context.Context, lastErr error) (Subscription, error) {
+		return NewSubscription(func(unsubscribed <-chan struct{}) error {
+			select {
+			case <-time.After(2 * time.Second):
+				innerSubDone <- struct{}{}
+				return nil
+			case <-unsubscribed:
+				return nil
+			}
+		}), nil
+	})
+
+	<-innerSubDone
+	sub.Unsubscribe()
+}