Fix deadlock in refresh debouncer stop

Deadlock could happen when closing the refresh debouncer (i.e. when closing the session).
This would result in either a panic or the goroutine calling session.Close() to hang.

patch by João Reis; reviewed by James Hartig, Kevin Kyyro for CASSGO-41

Author: joao-r-reis

host_source.go

@@ -402,10 +402,10 @@ func (h *HostInfo) HostnameAndPort() string {
 }
 
 func (h *HostInfo) ConnectAddressAndPort() string {
-        h.mu.Lock()
-        defer h.mu.Unlock()
-        addr, _ := h.connectAddressLocked()
-        return net.JoinHostPort(addr.String(), strconv.Itoa(h.port))
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	addr, _ := h.connectAddressLocked()
+	return net.JoinHostPort(addr.String(), strconv.Itoa(h.port))
 }
 
 func (h *HostInfo) String() string {
@@ -747,27 +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{}
+	done         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{}),
+		done:         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
@@ -777,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.isStopped() {
+	if d.stopped {
 		return
 	}
 	d.timer.Reset(d.interval)
@@ -788,11 +788,6 @@ 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{}{}:
@@ -804,15 +799,15 @@ func (d *refreshDebouncer) refreshNow() <-chan error {
 }
 
 func (d *refreshDebouncer) flusher() {
-	defer d.stoppedCtxFn()
+	defer close(d.done)
 	for {
 		select {
 		case <-d.refreshNowCh:
 		case <-d.timer.C:
-		case <-d.quitCtx.Done():
+		case <-d.quit:
 		}
 		d.mu.Lock()
-		if d.isStopped() {
+		if d.stopped {
 			if d.broadcaster != nil {
 				d.broadcaster.stop()
 				d.broadcaster = nil
@@ -846,12 +841,15 @@ func (d *refreshDebouncer) flusher() {
 }
 
 func (d *refreshDebouncer) stop() {
-	d.quitCtxFn()         // wake up flusher
-	<-d.stoppedCtx.Done() // wait for flusher to exit
-}
-
-func (d *refreshDebouncer) isStopped() bool {
-	return d.quitCtx.Err() != nil
+	d.mu.Lock()
+	if d.stopped {
+		d.mu.Unlock()
+		return
+	}
+	d.stopped = true
+	d.mu.Unlock()
+	close(d.quit)
+	<-d.done
 }
 
 // broadcasts an error to multiple channels (listeners)