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

@@ -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
 	}
 }
 

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