Repository: gocql/gocql
Pull request #1680 (2023-03-13 18:55:16)

Always refresh ring on topology changes and reconnections
Currently, there are some cases where gocql does not refresh the ring during a control connection reconnection. If topology changes are happening on the cluster during the reconnection, some `NEW_NODE` and `REMOVED_NODE` protocol events might be missed. 

To fix this, every successful reconnection on the control connection should trigger a ring refresh operation. This PR adds that.

Also, this PR fixes an issue where `NEW_NODE` events can be overwritten by `UP` events and cause certain nodes to not be added to the ring. This is an issue already described in #1669 but the fix proposed by this PR is a bit different than one outlined on that PR. This PR causes any topology event to trigger a ring refresh regardless of how many nodes were removed or added. This eliminates the need to have duplicated logic around adding and removing nodes in both `ringDescriber` and `Session`.

**Note**: this is not an implementation of #1681.

martin-sucha: Hi! Thanks for the pull request.

Did you mean to put removal of `addNewNode` to the [remove unused code](https://github.com/gocql/gocql/pull/1680/commits/fff05239a6593369c4da5cd90ec6970067587e6a) commit instead of `handleNewNode` and `handleRemovedNode`?

Could you please update the description of commit https://github.com/gocql/gocql/pull/1680/commits/326072be085e740a48f545521301691f99261e2d to what the description of the merge request says? `refresh ring improvements` is a bit vague. The commit title could be something like  `Always refresh ring on topology change`. That should make it easier to find the commit in commit log and see the more details with `git blame`.

It seems gocql currently does not de-bounce the ring refresh, so always refreshing the ring could theoretically cause a lot of loads if events come in quick succession. I suspect this is not a problem in practice, but we can still add the de-bounce (maybe in a separate pull request?) to guard against that.

Another thing that I noticed while reviewing the code is that `refreshRing` loads data from `system.peers`, but seems to not use data from `system.local`. That happens only in `addNewNode` resp. `ringDescriber.getHostInfo`, which is not used in current master. We should read from system.local in ringDescriber.GetHosts as well.

Otherwise this looks good to me.

It would be great if you could add a test to make sure that we don't introduce the same issues in the future.
joao-r-reis: Thanks for the quick review. I've changed the commits a bit so they should be clearer now.

> Did you mean to put removal of addNewNode to the [remove unused code](https://github.com/gocql/gocql/pull/1680/commits/fff05239a6593369c4da5cd90ec6970067587e6a) commit instead of handleNewNode and handleRemovedNode?

I removed `addNewNode` initially and then I found out that `handleNewNode` and `handleRemovedNode` were also unused so I removed these 2 later as well (now they are in the same commit).

> It seems gocql currently does not de-bounce the ring refresh, so always refreshing the ring could theoretically cause a lot of loads if events come in quick succession. I suspect this is not a problem in practice, but we can still add the de-bounce (maybe in a separate pull request?) to guard against that.

The topology change events are being debounced but I agree that it would probably be a good idea to debounce the ring refresh operations themselves. I'm happy to address this in a subsequent PR or I can just create a new issue so another contributor can pick it up, let me know what you prefer.

> Another thing that I noticed while reviewing the code is that refreshRing loads data from system.peers, but seems to not use data from system.local. That happens only in addNewNode resp. ringDescriber.getHostInfo, which is not used in current master. We should read from system.local in ringDescriber.GetHosts as well.

I'll start working on this and I'll also look into writing a test for this.
martin-sucha: 
> > Did you mean to put removal of addNewNode to the [remove unused code](https://github.com/gocql/gocql/pull/1680/commits/fff05239a6593369c4da5cd90ec6970067587e6a) commit instead of handleNewNode and handleRemovedNode?
> 
> I removed `addNewNode` initially and then I found out that `handleNewNode` and `handleRemovedNode` were also unused so I removed these 2 later as well (now they are in the same commit).

Techically, `handleNewNode` and `handleRemovedNode` are used until https://github.com/gocql/gocql/pull/1680/commits/7b506bcb0421440ad23c02b53d830b35c7509b60 so the removal should be part of that commit.

> 
> > It seems gocql currently does not de-bounce the ring refresh, so always refreshing the ring could theoretically cause a lot of loads if events come in quick succession. I suspect this is not a problem in practice, but we can still add the de-bounce (maybe in a separate pull request?) to guard against that.
> 
> The topology change events are being debounced but I agree that it would probably be a good idea to debounce the ring refresh operations themselves. I'm happy to address this in a subsequent PR or I can just create a new issue so another contributor can pick it up, let me know what you prefer.

Feel free to send a pull request, that would be great.

> > Another thing that I noticed while reviewing the code is that `refreshRing` loads data from `system.peers`, but seems to not use data from `system.local`. That happens only in `addNewNode` resp. `ringDescriber.getHostInfo`, which is not used in current master. We should read from system.local in ringDescriber.GetHosts as well.
> 
> I'll start working on this and I'll also look into writing a test for this.

Thanks!
joao-r-reis: I've rebased my branch and pushed commits to make `GetHosts()` retrieve the local host info from `system.local` and to remove `getHostInfo()` which was only used by `addNewNode()` and this function was removed in one of my earlier commits because it was also unused.

For the debouncer, I would think having an event debouncer would be enough and I see that `gocql` already has one, is there a need to have two debouncers? What would be the goal of the ring refresh debouncer?
StevenLacerda: @martin-sucha Can you take a look at the merge request? This should help tremendously with K8's environments. Thanks!
joao-r-reis: I've implemented a debouncer, can you take a look and see if this is aligned with what you were thinking? I'm going to look into tests now.
martin-sucha: > I've added unit tests but I don't see a way to simulate a topology change for an integration test, does the current test code support writing an integration test that could test whether the control connection updates the topology when it reconnects?

Yeah, the current integration tests use ccm and it would probably be hard to write an integration test that changes topology. We should probably use Go code directly instead of ccm to setup the cluster. That is a bigger chunk of work though. I've opened https://github.com/gocql/gocql/issues/1686 to track that.

> Oh I've also changed the host_source_test.go tests to run with the unit flag instead of cassandra flag since this file seems to contain unit tests only.

Nice catch. Thanks!
joao-r-reis: Managed to add a ccm test that tests whether the control connection refreshes the ring on reconnect. I used `DisableTopologyEvents` config and a custom `HostFilter` to avoid using real topology changes.

I'll take a look at your review feedback now, thanks.

Patch:
diff --git a/AUTHORS b/AUTHORS
index e73541f..750c4c0 100644
--- a/AUTHORS
+++ b/AUTHORS
@@ -134,4 +134,5 @@ Valerii Ponomarov <kiparis.kh@gmail.com>
 Neal Turett <neal.turett@datadoghq.com>
 Doug Schaapveld <djschaap@gmail.com>
 Steven Seidman <steven.seidman@datadoghq.com>
-Wojciech Przytuła <wojciech.przytula@scylladb.com>
\ No newline at end of file
+Wojciech Przytuła <wojciech.przytula@scylladb.com>
+João Reis <joao.reis@datastax.com>
\ No newline at end of file
diff --git a/conn.go b/conn.go
index e67c0f3..9fb40c4 100644
--- a/conn.go
+++ b/conn.go
@@ -1651,6 +1651,10 @@ func (c *Conn) querySystemPeers(ctx context.Context, version cassVersion) *Iter
 	}
 }
 
+func (c *Conn) querySystemLocal(ctx context.Context) *Iter {
+	return c.query(ctx, "SELECT * FROM system.local WHERE key='local'")
+}
+
 func (c *Conn) awaitSchemaAgreement(ctx context.Context) (err error) {
 	const localSchemas = "SELECT schema_version FROM system.local WHERE key='local'"
 
@@ -1721,23 +1725,6 @@ func (c *Conn) awaitSchemaAgreement(ctx context.Context) (err error) {
 	return fmt.Errorf("gocql: cluster schema versions not consistent: %+v", schemas)
 }
 
-func (c *Conn) localHostInfo(ctx context.Context) (*HostInfo, error) {
-	row, err := c.query(ctx, "SELECT * FROM system.local WHERE key='local'").rowMap()
-	if err != nil {
-		return nil, err
-	}
-
-	port := c.conn.RemoteAddr().(*net.TCPAddr).Port
-
-	// TODO(zariel): avoid doing this here
-	host, err := c.session.hostInfoFromMap(row, &HostInfo{connectAddress: c.host.connectAddress, port: port})
-	if err != nil {
-		return nil, err
-	}
-
-	return c.session.ring.addOrUpdate(host), nil
-}
-
 var (
 	ErrQueryArgLength    = errors.New("gocql: query argument length mismatch")
 	ErrTimeoutNoResponse = errors.New("gocql: no response received from cassandra within timeout period")
diff --git a/control.go b/control.go
index aed09a0..bda84eb 100644
--- a/control.go
+++ b/control.go
@@ -98,7 +98,7 @@ func (c *controlConn) heartBeat() {
 	reconn:
 		// try to connect a bit faster
 		sleepTime = 1 * time.Second
-		c.reconnect(true)
+		c.reconnect()
 		continue
 	}
 }
@@ -269,11 +269,14 @@ type connHost struct {
 
 func (c *controlConn) setupConn(conn *Conn) error {
 	// we need up-to-date host info for the filterHost call below
-	host, err := conn.localHostInfo(context.TODO())
+	iter := conn.querySystemLocal(context.TODO())
+	host, err := c.session.hostInfoFromIter(iter, conn.host.connectAddress, conn.conn.RemoteAddr().(*net.TCPAddr).Port)
 	if err != nil {
 		return err
 	}
 
+	host = c.session.ring.addOrUpdate(host)
+
 	if c.session.cfg.filterHost(host) {
 		return fmt.Errorf("host was filtered: %v", host.ConnectAddress())
 	}
@@ -335,7 +338,7 @@ func (c *controlConn) registerEvents(conn *Conn) error {
 	return nil
 }
 
-func (c *controlConn) reconnect(refreshring bool) {
+func (c *controlConn) reconnect() {
 	if atomic.LoadInt32(&c.state) == controlConnClosing {
 		return
 	}
@@ -381,8 +384,9 @@ func (c *controlConn) reconnect(refreshring bool) {
 		return
 	}
 
-	if refreshring {
-		c.session.hostSource.refreshRing()
+	err = c.session.refreshRing()
+	if err != nil {
+		c.session.logger.Printf("gocql: unable to refresh ring: %v\n", err)
 	}
 }
 
@@ -399,7 +403,7 @@ func (c *controlConn) HandleError(conn *Conn, err error, closed bool) {
 		return
 	}
 
-	c.reconnect(false)
+	c.reconnect()
 }
 
 func (c *controlConn) getConn() *connHost {
@@ -433,7 +437,7 @@ func (c *controlConn) withConnHost(fn func(*connHost) *Iter) *Iter {
 
 			connectAttempts++
 
-			c.reconnect(false)
+			c.reconnect()
 			continue
 		}
 
diff --git a/events.go b/events.go
index 35ab99a..73461f6 100644
--- a/events.go
+++ b/events.go
@@ -131,8 +131,10 @@ func (s *Session) handleKeyspaceChange(keyspace, change string) {
 
 // handleNodeEvent handles inbound status and topology change events.
 //
-// Within each category (topology vs status), events are debounced by
-// host IP; only the latest event is processed.
+// Status events are debounced by host IP; only the latest event is processed.
+//
+// Topology events are debounced by performing a single full topology refresh
+// whenever any topology event comes in.
 //
 // Processing topology change events before status change events ensures
 // that a NEW_NODE event is not dropped in favor of a newer UP event (which
@@ -144,22 +146,14 @@ func (s *Session) handleNodeEvent(frames []frame) {
 		port   int
 	}
 
-	// topology change events
-	tEvents := make(map[string]*nodeEvent)
+	topologyEventReceived := false
 	// status change events
 	sEvents := make(map[string]*nodeEvent)
 
 	for _, frame := range frames {
-		// TODO: can we be sure the order of events in the buffer is correct?
 		switch f := frame.(type) {
 		case *topologyChangeEventFrame:
-			event, ok := tEvents[f.host.String()]
-			if !ok {
-				event = &nodeEvent{change: f.change, host: f.host, port: f.port}
-				tEvents[f.host.String()] = event
-			}
-			event.change = f.change
-
+			topologyEventReceived = true
 		case *statusChangeEventFrame:
 			event, ok := sEvents[f.host.String()]
 			if !ok {
@@ -170,26 +164,8 @@ func (s *Session) handleNodeEvent(frames []frame) {
 		}
 	}
 
-	for _, f := range tEvents {
-		if gocqlDebug {
-			s.logger.Printf("gocql: dispatching topology change event: %+v\n", f)
-		}
-
-		// ignore events we received if they were disabled
-		// see https://github.com/gocql/gocql/issues/1591
-		switch f.change {
-		case "NEW_NODE":
-			if !s.cfg.Events.DisableTopologyEvents {
-				s.handleNewNode(f.host, f.port)
-			}
-		case "REMOVED_NODE":
-			if !s.cfg.Events.DisableTopologyEvents {
-				s.handleRemovedNode(f.host, f.port)
-			}
-		case "MOVED_NODE":
-			// java-driver handles this, not mentioned in the spec
-			// TODO(zariel): refresh token map
-		}
+	if topologyEventReceived && !s.cfg.Events.DisableTopologyEvents {
+		s.debounceRingRefresh()
 	}
 
 	for _, f := range sEvents {
@@ -212,73 +188,6 @@ func (s *Session) handleNodeEvent(frames []frame) {
 	}
 }
 
-func (s *Session) addNewNode(hostID UUID) {
-	// Get host info and apply any filters to the host
-	hostInfo, err := s.hostSource.getHostInfo(hostID)
-	if err != nil {
-		s.logger.Printf("gocql: events: unable to fetch host info for hostID: %q: %v\n", hostID, err)
-		return
-	} else if hostInfo == nil {
-		// ignore if it's null because we couldn't find it
-		return
-	}
-
-	if t := hostInfo.Version().nodeUpDelay(); t > 0 {
-		time.Sleep(t)
-	}
-
-	// should this handle token moving?
-	hostInfo = s.ring.addOrUpdate(hostInfo)
-
-	if !s.cfg.filterHost(hostInfo) {
-		s.startPoolFill(hostInfo)
-	}
-
-	if s.control != nil && !s.cfg.IgnorePeerAddr {
-		// TODO(zariel): debounce ring refresh
-		s.hostSource.refreshRing()
-	}
-}
-
-func (s *Session) handleNewNode(ip net.IP, port int) {
-	if gocqlDebug {
-		s.logger.Printf("gocql: Session.handleNewNode: %s:%d\n", ip.String(), port)
-	}
-
-	host, ok := s.ring.getHostByIP(ip.String())
-	if ok && host.IsUp() {
-		return
-	}
-
-	if err := s.hostSource.refreshRing(); err != nil && gocqlDebug {
-		s.logger.Printf("gocql: Session.handleNewNode: failed to refresh ring: %w\n", err.Error())
-	}
-}
-
-func (s *Session) handleRemovedNode(ip net.IP, port int) {
-	if gocqlDebug {
-		s.logger.Printf("gocql: Session.handleRemovedNode: %s:%d\n", ip.String(), port)
-	}
-
-	// we remove all nodes but only add ones which pass the filter
-	host, ok := s.ring.getHostByIP(ip.String())
-	if ok {
-		hostID := host.HostID()
-		s.ring.removeHost(hostID)
-
-		host.setState(NodeDown)
-		if !s.cfg.filterHost(host) {
-			s.policy.RemoveHost(host)
-			s.pool.removeHost(hostID)
-		}
-
-	}
-
-	if err := s.hostSource.refreshRing(); err != nil && gocqlDebug {
-		s.logger.Println("failed to refresh ring:", err)
-	}
-}
-
 func (s *Session) handleNodeUp(eventIp net.IP, eventPort int) {
 	if gocqlDebug {
 		s.logger.Printf("gocql: Session.handleNodeUp: %s:%d\n", eventIp.String(), eventPort)
@@ -286,9 +195,7 @@ func (s *Session) handleNodeUp(eventIp net.IP, eventPort int) {
 
 	host, ok := s.ring.getHostByIP(eventIp.String())
 	if !ok {
-		if err := s.hostSource.refreshRing(); err != nil && gocqlDebug {
-			s.logger.Printf("gocql: Session.handleNodeUp: failed to refresh ring: %w\n", err.Error())
-		}
+		s.debounceRingRefresh()
 		return
 	}
 
diff --git a/host_source.go b/host_source.go
index bb97271..a5efd5f 100644
--- a/host_source.go
+++ b/host_source.go
@@ -559,12 +559,54 @@ func (s *Session) hostInfoFromMap(row map[string]interface{}, host *HostInfo) (*
 	return host, nil
 }
 
+func (s *Session) hostInfoFromIter(iter *Iter, connectAddress net.IP, defaultPort int) (*HostInfo, error) {
+	rows, err := iter.SliceMap()
+	if err != nil {
+		// TODO(zariel): make typed error
+		return nil, err
+	}
+
+	if len(rows) == 0 {
+		return nil, errors.New("query returned 0 rows")
+	}
+
+	host, err := s.hostInfoFromMap(rows[0], &HostInfo{connectAddress: connectAddress, port: defaultPort})
+	if err != nil {
+		return nil, err
+	}
+	return host, nil
+}
+
+// Ask the control node for the local host info
+func (r *ringDescriber) getLocalHostInfo() (*HostInfo, error) {
+	if r.session.control == nil {
+		return nil, errNoControl
+	}
+
+	iter := r.session.control.withConnHost(func(ch *connHost) *Iter {
+		return ch.conn.querySystemLocal(context.TODO())
+	})
+
+	if iter == nil {
+		return nil, errNoControl
+	}
+
+	host, err := r.session.hostInfoFromIter(iter, nil, r.session.cfg.Port)
+	if err != nil {
+		return nil, fmt.Errorf("could not retrieve local host info: %w", err)
+	}
+	return host, nil
+}
+
 // Ask the control node for host info on all it's known peers
-func (r *ringDescriber) getClusterPeerInfo() ([]*HostInfo, error) {
-	var hosts []*HostInfo
+func (r *ringDescriber) getClusterPeerInfo(localHost *HostInfo) ([]*HostInfo, error) {
+	if r.session.control == nil {
+		return nil, errNoControl
+	}
+
+	var peers []*HostInfo
 	iter := r.session.control.withConnHost(func(ch *connHost) *Iter {
-		hosts = append(hosts, ch.host)
-		return ch.conn.querySystemPeers(context.TODO(), ch.host.version)
+		return ch.conn.querySystemPeers(context.TODO(), localHost.version)
 	})
 
 	if iter == nil {
@@ -589,10 +631,10 @@ func (r *ringDescriber) getClusterPeerInfo() ([]*HostInfo, error) {
 			continue
 		}
 
-		hosts = append(hosts, host)
+		peers = append(peers, host)
 	}
 
-	return hosts, nil
+	return peers, nil
 }
 
 // Return true if the host is a valid peer
@@ -604,16 +646,22 @@ func isValidPeer(host *HostInfo) bool {
 		len(host.tokens) == 0)
 }
 
-// Return a list of hosts the cluster knows about
+// GetHosts returns a list of hosts found via queries to system.local and system.peers
 func (r *ringDescriber) GetHosts() ([]*HostInfo, string, error) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 
-	hosts, err := r.getClusterPeerInfo()
+	localHost, err := r.getLocalHostInfo()
 	if err != nil {
 		return r.prevHosts, r.prevPartitioner, err
 	}
 
+	peerHosts, err := r.getClusterPeerInfo(localHost)
+	if err != nil {
+		return r.prevHosts, r.prevPartitioner, err
+	}
+
+	hosts := append([]*HostInfo{localHost}, peerHosts...)
 	var partitioner string
 	if len(hosts) > 0 {
 		partitioner = hosts[0].Partitioner()
@@ -622,55 +670,22 @@ func (r *ringDescriber) GetHosts() ([]*HostInfo, string, error) {
 	return hosts, partitioner, nil
 }
 
-// Given an ip/port return HostInfo for the specified ip/port
-func (r *ringDescriber) getHostInfo(hostID UUID) (*HostInfo, error) {
-	var host *HostInfo
-	for _, table := range []string{"system.peers", "system.local"} {
-		iter := r.session.control.withConnHost(func(ch *connHost) *Iter {
-			if ch.host.HostID() == hostID.String() {
-				host = ch.host
-				return nil
-			}
-
-			if table == "system.peers" {
-				return ch.conn.querySystemPeers(context.TODO(), ch.host.version)
-			} else {
-				return ch.conn.query(context.TODO(), fmt.Sprintf("SELECT * FROM %s", table))
-			}
-		})
-
-		if iter != nil {
-			rows, err := iter.SliceMap()
-			if err != nil {
-				return nil, err
-			}
-
-			for _, row := range rows {
-				h, err := r.session.hostInfoFromMap(row, &HostInfo{port: r.session.cfg.Port})
-				if err != nil {
-					return nil, err
-				}
-
-				if h.HostID() == hostID.String() {
-					host = h
-					break
-				}
-			}
-		}
-	}
+// debounceRingRefresh submits a ring refresh request to the ring refresh debouncer.
+func (s *Session) debounceRingRefresh() {
+	s.ringRefresher.debounce()
+}
 
-	if host == nil {
-		return nil, errors.New("unable to fetch host info: invalid control connection")
+// refreshRing executes a ring refresh immediately and cancels pending debounce ring refresh requests.
+func (s *Session) refreshRing() error {
+	err, ok := <-s.ringRefresher.refreshNow()
+	if !ok {
+		return errors.New("could not refresh ring because stop was requested")
 	}
 
-	return host, nil
+	return err
 }
 
-func (r *ringDescriber) refreshRing() error {
-	// if we have 0 hosts this will return the previous list of hosts to
-	// attempt to reconnect to the cluster otherwise we would never find
-	// downed hosts again, could possibly have an optimisation to only
-	// try to add new hosts if GetHosts didn't error and the hosts didn't change.
+func refreshRing(r *ringDescriber) error {
 	hosts, partitioner, err := r.GetHosts()
 	if err != nil {
 		return err
@@ -678,7 +693,6 @@ func (r *ringDescriber) refreshRing() error {
 
 	prevHosts := r.session.ring.currentHosts()
 
-	// TODO: move this to session
 	for _, h := range hosts {
 		if r.session.cfg.filterHost(h) {
 			continue
@@ -710,9 +724,6 @@ func (r *ringDescriber) refreshRing() error {
 		delete(prevHosts, h.HostID())
 	}
 
-	// TODO(zariel): it may be worth having a mutex covering the overall ring state
-	// in a session so that everything sees a consistent state. Becuase as is today
-	// events can come in and due to ordering an UP host could be removed from the cluster
 	for _, host := range prevHosts {
 		r.session.removeHost(host)
 	}
@@ -721,3 +732,161 @@ func (r *ringDescriber) refreshRing() error {
 	r.session.policy.SetPartitioner(partitioner)
 	return nil
 }
+
+const (
+	ringRefreshDebounceTime = 1 * time.Second
+)
+
+// 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
+}
+
+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.timer.Stop()
+	go d.flusher()
+	return d
+}
+
+// debounces a request to call the refresh function
+func (d *refreshDebouncer) debounce() {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	if d.stopped {
+		return
+	}
+	d.timer.Reset(d.interval)
+}
+
+// requests an immediate refresh which will cancel pending refresh requests
+func (d *refreshDebouncer) refreshNow() <-chan error {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	if d.broadcaster == nil {
+		d.broadcaster = newErrorBroadcaster()
+		select {
+		case d.refreshNowCh <- struct{}{}:
+		default:
+			// already a refresh pending
+		}
+	}
+	return d.broadcaster.newListener()
+}
+
+func (d *refreshDebouncer) flusher() {
+	for {
+		select {
+		case <-d.refreshNowCh:
+		case <-d.timer.C:
+		case <-d.quit:
+		}
+		d.mu.Lock()
+		if d.stopped {
+			if d.broadcaster != nil {
+				d.broadcaster.stop()
+				d.broadcaster = nil
+			}
+			d.timer.Stop()
+			d.mu.Unlock()
+			return
+		}
+
+		// make sure both request channels are cleared before we refresh
+		select {
+		case <-d.refreshNowCh:
+		default:
+		}
+
+		d.timer.Stop()
+		select {
+		case <-d.timer.C:
+		default:
+		}
+
+		curBroadcaster := d.broadcaster
+		d.broadcaster = nil
+		d.mu.Unlock()
+
+		err := d.refreshFn()
+		if curBroadcaster != nil {
+			curBroadcaster.broadcast(err)
+		}
+	}
+}
+
+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)
+}
+
+// broadcasts an error to multiple channels (listeners)
+type errorBroadcaster struct {
+	listeners []chan<- error
+	mu        sync.Mutex
+}
+
+func newErrorBroadcaster() *errorBroadcaster {
+	return &errorBroadcaster{
+		listeners: nil,
+		mu:        sync.Mutex{},
+	}
+}
+
+func (b *errorBroadcaster) newListener() <-chan error {
+	ch := make(chan error, 1)
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.listeners = append(b.listeners, ch)
+	return ch
+}
+
+func (b *errorBroadcaster) broadcast(err error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	curListeners := b.listeners
+	if len(curListeners) > 0 {
+		b.listeners = nil
+	} else {
+		return
+	}
+
+	for _, listener := range curListeners {
+		listener <- err
+		close(listener)
+	}
+}
+
+func (b *errorBroadcaster) stop() {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if len(b.listeners) == 0 {
+		return
+	}
+	for _, listener := range b.listeners {
+		close(listener)
+	}
+	b.listeners = nil
+}
diff --git a/internal/ccm/ccm.go b/internal/ccm/ccm.go
index df48175..a663806 100644
--- a/internal/ccm/ccm.go
+++ b/internal/ccm/ccm.go
@@ -1,3 +1,4 @@
+//go:build ccm
 // +build ccm
 
 package ccm
@@ -8,11 +9,17 @@ import (
 	"errors"
 	"fmt"
 	"os/exec"
+	"runtime"
 	"strings"
 )
 
 func execCmd(args ...string) (*bytes.Buffer, error) {
-	cmd := exec.Command("ccm", args...)
+	execName := "ccm"
+	if runtime.GOOS == "windows" {
+		args = append([]string{"/c", execName}, args...)
+		execName = "cmd.exe"
+	}
+	cmd := exec.Command(execName, args...)
 	stdout := &bytes.Buffer{}
 	cmd.Stdout = stdout
 	cmd.Stderr = &bytes.Buffer{}
@@ -41,7 +48,11 @@ func AllUp() error {
 }
 
 func NodeUp(node string) error {
-	_, err := execCmd(node, "start", "--wait-for-binary-proto", "--wait-other-notice")
+	args := []string{node, "start", "--wait-for-binary-proto"}
+	if runtime.GOOS == "windows" {
+		args = append(args, "--quiet-windows")
+	}
+	_, err := execCmd(args...)
 	return err
 }
 
diff --git a/session.go b/session.go
index 7696e6c..3b0d71c 100644
--- a/session.go
+++ b/session.go
@@ -43,6 +43,7 @@ type Session struct {
 	frameObserver       FrameHeaderObserver
 	streamObserver      StreamObserver
 	hostSource          *ringDescriber
+	ringRefresher       *refreshDebouncer
 	stmtsLRU            *preparedLRU
 
 	connCfg *ConnConfig
@@ -146,6 +147,7 @@ func NewSession(cfg ClusterConfig) (*Session, error) {
 	s.routingKeyInfoCache.lru = lru.New(cfg.MaxRoutingKeyInfo)
 
 	s.hostSource = &ringDescriber{session: s}
+	s.ringRefresher = newRefreshDebouncer(ringRefreshDebounceTime, func() error { return refreshRing(s.hostSource) })
 
 	if cfg.PoolConfig.HostSelectionPolicy == nil {
 		cfg.PoolConfig.HostSelectionPolicy = RoundRobinHostPolicy()
@@ -488,6 +490,10 @@ func (s *Session) Close() {
 		s.schemaEvents.stop()
 	}
 
+	if s.ringRefresher != nil {
+		s.ringRefresher.stop()
+	}
+
 	if s.cancel != nil {
 		s.cancel()
 	}


Test patch:
diff --git a/control_ccm_test.go b/control_ccm_test.go
new file mode 100644
index 0000000..48ed154
--- /dev/null
+++ b/control_ccm_test.go
@@ -0,0 +1,173 @@
+//go:build ccm
+// +build ccm
+
+package gocql
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/gocql/gocql/internal/ccm"
+)
+
+type TestHostFilter struct {
+	mu           sync.Mutex
+	allowedHosts map[string]ccm.Host
+}
+
+func (f *TestHostFilter) Accept(h *HostInfo) bool {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	_, ok := f.allowedHosts[h.ConnectAddress().String()]
+	return ok
+}
+
+func (f *TestHostFilter) SetAllowedHosts(hosts map[string]ccm.Host) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.allowedHosts = hosts
+}
+
+func TestControlConn_ReconnectRefreshesRing(t *testing.T) {
+	if err := ccm.AllUp(); err != nil {
+		t.Fatal(err)
+	}
+
+	allCcmHosts, err := ccm.Status()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if len(allCcmHosts) < 2 {
+		t.Skip("this test requires at least 2 nodes")
+	}
+
+	allAllowedHosts := map[string]ccm.Host{}
+	var firstNode *ccm.Host
+	for _, node := range allCcmHosts {
+		if firstNode == nil {
+			firstNode = &node
+		}
+		allAllowedHosts[node.Addr] = node
+	}
+
+	allowedHosts := map[string]ccm.Host{
+		firstNode.Addr: *firstNode,
+	}
+
+	testFilter := &TestHostFilter{allowedHosts: allowedHosts}
+
+	session := createSession(t, func(config *ClusterConfig) {
+		config.Hosts = []string{firstNode.Addr}
+		config.Events.DisableTopologyEvents = true
+		config.Events.DisableNodeStatusEvents = true
+		config.HostFilter = testFilter
+	})
+	defer session.Close()
+
+	if session.control == nil || session.control.conn.Load() == nil {
+		t.Fatal("control conn is nil")
+	}
+
+	controlConnection := session.control.getConn()
+	ccHost := controlConnection.host
+
+	var ccHostName string
+	for _, node := range allCcmHosts {
+		if node.Addr == ccHost.ConnectAddress().String() {
+			ccHostName = node.Name
+			break
+		}
+	}
+
+	if ccHostName == "" {
+		t.Fatal("could not find name of control host")
+	}
+
+	if err := ccm.NodeDown(ccHostName); err != nil {
+		t.Fatal()
+	}
+
+	defer func() {
+		ccmStatus, err := ccm.Status()
+		if err != nil {
+			t.Logf("could not bring nodes back up after test: %v", err)
+			return
+		}
+		for _, node := range ccmStatus {
+			if node.State == ccm.NodeStateDown {
+				err = ccm.NodeUp(node.Name)
+				if err != nil {
+					t.Logf("could not bring node %v back up after test: %v", node.Name, err)
+				}
+			}
+		}
+	}()
+
+	assertNodeDown := func() error {
+		hosts := session.ring.currentHosts()
+		if len(hosts) != 1 {
+			return fmt.Errorf("expected 1 host in ring but there were %v", len(hosts))
+		}
+		for _, host := range hosts {
+			if host.IsUp() {
+				return fmt.Errorf("expected host to be DOWN but %v isn't", host.String())
+			}
+		}
+
+		pools := session.pool.hostConnPools
+		if len(pools) != 0 {
+			return fmt.Errorf("expected 0 connection pool but there were %v", len(pools))
+		}
+		return nil
+	}
+
+	maxAttempts := 5
+	delayPerAttempt := 1 * time.Second
+	assertErr := assertNodeDown()
+	for i := 0; i < maxAttempts && assertErr != nil; i++ {
+		time.Sleep(delayPerAttempt)
+		assertErr = assertNodeDown()
+	}
+
+	if assertErr != nil {
+		t.Fatal(err)
+	}
+
+	testFilter.SetAllowedHosts(allAllowedHosts)
+
+	if err = ccm.NodeUp(ccHostName); err != nil {
+		t.Fatal(err)
+	}
+
+	assertNodeUp := func() error {
+		hosts := session.ring.currentHosts()
+		if len(hosts) != len(allCcmHosts) {
+			return fmt.Errorf("expected %v hosts in ring but there were %v", len(allCcmHosts), len(hosts))
+		}
+		for _, host := range hosts {
+			if !host.IsUp() {
+				return fmt.Errorf("expected all hosts to be UP but %v isn't", host.String())
+			}
+		}
+		pools := session.pool.hostConnPools
+		if len(pools) != len(allCcmHosts) {
+			return fmt.Errorf("expected %v connection pool but there were %v", len(allCcmHosts), len(pools))
+		}
+		return nil
+	}
+
+	maxAttempts = 30
+	delayPerAttempt = 1 * time.Second
+	assertErr = assertNodeUp()
+	for i := 0; i < maxAttempts && assertErr != nil; i++ {
+		time.Sleep(delayPerAttempt)
+		assertErr = assertNodeUp()
+	}
+
+	if assertErr != nil {
+		t.Fatal(err)
+	}
+}
diff --git a/host_source_test.go b/host_source_test.go
index e78c272..bd45e75 100644
--- a/host_source_test.go
+++ b/host_source_test.go
@@ -1,11 +1,15 @@
-//go:build all || cassandra
-// +build all cassandra
+//go:build all || unit
+// +build all unit
 
 package gocql
 
 import (
+	"errors"
 	"net"
+	"sync"
+	"sync/atomic"
 	"testing"
+	"time"
 )
 
 func TestUnmarshalCassVersion(t *testing.T) {
@@ -99,3 +103,245 @@ func TestHostInfo_ConnectAddress(t *testing.T) {
 		})
 	}
 }
+
+// This test sends debounce requests and waits until the refresh function is called (which should happen when the timer elapses).
+func TestRefreshDebouncer_MultipleEvents(t *testing.T) {
+	const numberOfEvents = 10
+	channel := make(chan int, numberOfEvents) // should never use more than 1 but allow for more to possibly detect bugs
+	fn := func() error {
+		channel <- 0
+		return nil
+	}
+	beforeEvents := time.Now()
+	wg := sync.WaitGroup{}
+	d := newRefreshDebouncer(2*time.Second, fn)
+	defer d.stop()
+	for i := 0; i < numberOfEvents; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			d.debounce()
+		}()
+	}
+	wg.Wait()
+	timeoutCh := time.After(2500 * time.Millisecond) // extra time to avoid flakiness
+	select {
+	case <-channel:
+	case <-timeoutCh:
+		t.Fatalf("timeout elapsed without flush function being called")
+	}
+	afterFunctionCall := time.Now()
+
+	// use 1.5 seconds instead of 2 seconds to avoid timer precision issues
+	if afterFunctionCall.Sub(beforeEvents) < 1500*time.Millisecond {
+		t.Fatalf("function was called after %v ms instead of ~2 seconds", afterFunctionCall.Sub(beforeEvents).Milliseconds())
+	}
+
+	// wait another 2 seconds and check if function was called again
+	time.Sleep(2500 * time.Millisecond)
+	if len(channel) > 0 {
+		t.Fatalf("function was called more than once")
+	}
+}
+
+// This test:
+//
+//	1 - Sends debounce requests when test starts
+//	2 - Calls refreshNow() before the timer elapsed (which stops the timer) about 1.5 seconds after test starts
+//
+// The end result should be 1 refresh function call when refreshNow() is called.
+func TestRefreshDebouncer_RefreshNow(t *testing.T) {
+	const numberOfEvents = 10
+	channel := make(chan int, numberOfEvents) // should never use more than 1 but allow for more to possibly detect bugs
+	fn := func() error {
+		channel <- 0
+		return nil
+	}
+	beforeEvents := time.Now()
+	eventsWg := sync.WaitGroup{}
+	d := newRefreshDebouncer(2*time.Second, fn)
+	defer d.stop()
+	for i := 0; i < numberOfEvents; i++ {
+		eventsWg.Add(1)
+		go func() {
+			defer eventsWg.Done()
+			d.debounce()
+		}()
+	}
+
+	refreshNowWg := sync.WaitGroup{}
+	refreshNowWg.Add(1)
+	go func() {
+		defer refreshNowWg.Done()
+		time.Sleep(1500 * time.Millisecond)
+		d.refreshNow()
+	}()
+
+	eventsWg.Wait()
+	select {
+	case <-channel:
+		t.Fatalf("function was called before the expected time")
+	default:
+	}
+
+	refreshNowWg.Wait()
+
+	timeoutCh := time.After(200 * time.Millisecond) // allow for 200ms of delay to prevent flakiness
+	select {
+	case <-channel:
+	case <-timeoutCh:
+		t.Fatalf("timeout elapsed without flush function being called")
+	}
+	afterFunctionCall := time.Now()
+
+	// use 1 second instead of 1.5s to avoid timer precision issues
+	if afterFunctionCall.Sub(beforeEvents) < 1000*time.Millisecond {
+		t.Fatalf("function was called after %v ms instead of ~1.5 seconds", afterFunctionCall.Sub(beforeEvents).Milliseconds())
+	}
+
+	// wait some time and check if function was called again
+	time.Sleep(2500 * time.Millisecond)
+	if len(channel) > 0 {
+		t.Fatalf("function was called more than once")
+	}
+}
+
+// This test:
+//
+//	1 - Sends debounce requests when test starts
+//	2 - Calls refreshNow() before the timer elapsed (which stops the timer) about 1 second after test starts
+//	3 - Sends more debounce requests (which resets the timer with a 3-second interval) about 2 seconds after test starts
+//
+// The end result should be 2 refresh function calls:
+//
+//	1 - When refreshNow() is called (1 second after the test starts)
+//	2 - When the timer elapses after the second "wave" of debounce requests (5 seconds after the test starts)
+func TestRefreshDebouncer_EventsAfterRefreshNow(t *testing.T) {
+	const numberOfEvents = 10
+	channel := make(chan int, numberOfEvents) // should never use more than 2 but allow for more to possibly detect bugs
+	fn := func() error {
+		channel <- 0
+		return nil
+	}
+	beforeEvents := time.Now()
+	wg := sync.WaitGroup{}
+	d := newRefreshDebouncer(3*time.Second, fn)
+	defer d.stop()
+	for i := 0; i < numberOfEvents; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			d.debounce()
+			time.Sleep(2000 * time.Millisecond)
+			d.debounce()
+		}()
+	}
+
+	go func() {
+		time.Sleep(1 * time.Second)
+		d.refreshNow()
+	}()
+
+	wg.Wait()
+	timeoutCh := time.After(1500 * time.Millisecond) // extra 500ms to prevent flakiness
+	select {
+	case <-channel:
+	case <-timeoutCh:
+		t.Fatalf("timeout elapsed without flush function being called after refreshNow()")
+	}
+	afterFunctionCall := time.Now()
+
+	// use 500ms instead of 1s to avoid timer precision issues
+	if afterFunctionCall.Sub(beforeEvents) < 500*time.Millisecond {
+		t.Fatalf("function was called after %v ms instead of ~1 second", afterFunctionCall.Sub(beforeEvents).Milliseconds())
+	}
+
+	timeoutCh = time.After(4 * time.Second) // extra 1s to prevent flakiness
+	select {
+	case <-channel:
+	case <-timeoutCh:
+		t.Fatalf("timeout elapsed without flush function being called after debounce requests")
+	}
+	afterSecondFunctionCall := time.Now()
+
+	// use 2.5s instead of 3s to avoid timer precision issues
+	if afterSecondFunctionCall.Sub(afterFunctionCall) < 2500*time.Millisecond {
+		t.Fatalf("function was called after %v ms instead of ~3 seconds", afterSecondFunctionCall.Sub(afterFunctionCall).Milliseconds())
+	}
+
+	if len(channel) > 0 {
+		t.Fatalf("function was called more than twice")
+	}
+}
+
+func TestErrorBroadcaster_MultipleListeners(t *testing.T) {
+	b := newErrorBroadcaster()
+	defer b.stop()
+	const numberOfListeners = 10
+	var listeners []<-chan error
+	for i := 0; i < numberOfListeners; i++ {
+		listeners = append(listeners, b.newListener())
+	}
+
+	err := errors.New("expected error")
+	wg := sync.WaitGroup{}
+	result := atomic.Value{}
+	for _, listener := range listeners {
+		currentListener := listener
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			receivedErr, ok := <-currentListener
+			if !ok {
+				result.Store(errors.New("listener was closed"))
+			} else if receivedErr != err {
+				result.Store(errors.New("expected received error to be the same as the one that was broadcasted"))
+			}
+		}()
+	}
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		b.broadcast(err)
+		b.stop()
+	}()
+	wg.Wait()
+	if loadedVal := result.Load(); loadedVal != nil {
+		t.Errorf(loadedVal.(error).Error())
+	}
+}
+
+func TestErrorBroadcaster_StopWithoutBroadcast(t *testing.T) {
+	var b = newErrorBroadcaster()
+	defer b.stop()
+	const numberOfListeners = 10
+	var listeners []<-chan error
+	for i := 0; i < numberOfListeners; i++ {
+		listeners = append(listeners, b.newListener())
+	}
+
+	wg := sync.WaitGroup{}
+	result := atomic.Value{}
+	for _, listener := range listeners {
+		currentListener := listener
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			// broadcaster stopped, expect listener to be closed
+			_, ok := <-currentListener
+			if ok {
+				result.Store(errors.New("expected listener to be closed"))
+			}
+		}()
+	}
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		// call stop without broadcasting anything to current listeners
+		b.stop()
+	}()
+	wg.Wait()
+	if loadedVal := result.Load(); loadedVal != nil {
+		t.Errorf(loadedVal.(error).Error())
+	}
+}