CASSGO-41 Fix deadlock in refresh debouncer stop

host_source.go

@@ -747,25 +747,27 @@ const (
 // debounces requests to call a refresh function (currently used for ring refresh). It also supports triggering a refresh immediately.
 type refreshDebouncer struct {
 	mu           sync.Mutex
-	stopped      bool
 	broadcaster  *errorBroadcaster
 	interval     time.Duration
 	timer        *time.Timer
 	refreshNowCh chan struct{}
-	quit         chan struct{}
 	refreshFn    func() error
+	quitCtxFn    context.CancelFunc
+	quitCtx      context.Context
+	stoppedCtxFn context.CancelFunc
+	stoppedCtx   context.Context
 }
 
 func newRefreshDebouncer(interval time.Duration, refreshFn func() error) *refreshDebouncer {
 	d := &refreshDebouncer{
-		stopped:      false,
 		broadcaster:  nil,
 		refreshNowCh: make(chan struct{}, 1),
-		quit:         make(chan struct{}),
 		interval:     interval,
 		timer:        time.NewTimer(interval),
 		refreshFn:    refreshFn,
 	}
+	d.stoppedCtx, d.stoppedCtxFn = context.WithCancel(context.Background())
+	d.quitCtx, d.quitCtxFn = context.WithCancel(context.Background())
 	d.timer.Stop()
 	go d.flusher()
 	return d
@@ -775,7 +777,7 @@ func newRefreshDebouncer(interval time.Duration, refreshFn func() error) *refres
 func (d *refreshDebouncer) debounce() {
 	d.mu.Lock()
 	defer d.mu.Unlock()
-	if d.stopped {
+	if d.isStopped() {
 		return
 	}
 	d.timer.Reset(d.interval)
@@ -786,6 +788,11 @@ func (d *refreshDebouncer) refreshNow() <-chan error {
 	d.mu.Lock()
 	defer d.mu.Unlock()
 	if d.broadcaster == nil {
+		if d.isStopped() {
+			ch := make(chan error)
+			close(ch)
+			return ch
+		}
 		d.broadcaster = newErrorBroadcaster()
 		select {
 		case d.refreshNowCh <- struct{}{}:
@@ -797,14 +804,15 @@ func (d *refreshDebouncer) refreshNow() <-chan error {
 }
 
 func (d *refreshDebouncer) flusher() {
+	defer d.stoppedCtxFn()
 	for {
 		select {
 		case <-d.refreshNowCh:
 		case <-d.timer.C:
-		case <-d.quit:
+		case <-d.quitCtx.Done():
 		}
 		d.mu.Lock()
-		if d.stopped {
+		if d.isStopped() {
 			if d.broadcaster != nil {
 				d.broadcaster.stop()
 				d.broadcaster = nil
@@ -838,15 +846,12 @@ func (d *refreshDebouncer) flusher() {
 }
 
 func (d *refreshDebouncer) stop() {
-	d.mu.Lock()
-	if d.stopped {
-		d.mu.Unlock()
-		return
-	}
-	d.stopped = true
-	d.mu.Unlock()
-	d.quit <- struct{}{} // sync with flusher
-	close(d.quit)
+	d.quitCtxFn()         // wake up flusher
+	<-d.stoppedCtx.Done() // wait for flusher to exit
+}
+
+func (d *refreshDebouncer) isStopped() bool {
+	return d.quitCtx.Err() != nil
 }
 
 // broadcasts an error to multiple channels (listeners)

host_source_test.go

@@ -274,6 +274,32 @@ func TestRefreshDebouncer_EventsAfterRefreshNow(t *testing.T) {
 	}
 }
 
+// https://github.com/gocql/gocql/issues/1752
+func TestRefreshDebouncer_DeadlockOnStop(t *testing.T) {
+	// there's no way to guarantee this bug manifests because it depends on which `case` is picked from the `select`
+	// with 4 iterations of this test the deadlock would be hit pretty consistently
+	const iterations = 4
+	for i := 0; i < iterations; i++ {
+		refreshCalledCh := make(chan int, 5)
+		refreshDuration := 500 * time.Millisecond
+		fn := func() error {
+			refreshCalledCh <- 0
+			time.Sleep(refreshDuration)
+			return nil
+		}
+		d := newRefreshDebouncer(50*time.Millisecond, fn)
+		timeBeforeRefresh := time.Now()
+		_ = d.refreshNow()
+		<-refreshCalledCh
+		d.debounce()
+		d.stop()
+		timeAfterRefresh := time.Now()
+		if timeAfterRefresh.Sub(timeBeforeRefresh) < refreshDuration {
+			t.Errorf("refresh debouncer stop() didn't wait until flusher stopped")
+		}
+	}
+}
+
 func TestErrorBroadcaster_MultipleListeners(t *testing.T) {
 	b := newErrorBroadcaster()
 	defer b.stop()