diff --git a/cluster.go b/cluster.go index 863093ef8..afd8dbd00 100644 --- a/cluster.go +++ b/cluster.go @@ -223,6 +223,7 @@ type cluster struct { // nolint: maligned joined bool abortAntiEntropyCh chan struct{} + muAntiEntropy sync.Mutex translationSyncer translationSyncer @@ -483,8 +484,6 @@ func (c *cluster) unprotectedSetState(state string) { if err := c.translationSyncer.Reset(); err != nil { c.logger.Printf("error resetting translation syncer: %s", err) } - case ClusterStateResizing: - c.abortAntiEntropy() } // TODO: consider NOT running cleanup on an active node that has @@ -1832,6 +1831,17 @@ func (c *cluster) ReceiveEvent(e *NodeEvent) (err error) { // nodeJoin should only be called by the coordinator. func (c *cluster) nodeJoin(node *Node) error { + c.abortAntiEntropy() + // Technically there is a race condition here which could + // allow the anti-entropy process to re-start (and acquire + // the lock) before this lock has time to succeed. In that + // case, the user would have to wait through an entire + // anti-entropy cycle. We decided it wasn't worth the + // complexity (of, for example, implementing this with + // channels) to avoid that rare case. + c.muAntiEntropy.Lock() + defer c.muAntiEntropy.Unlock() + c.mu.Lock() defer c.mu.Unlock() c.logger.Printf("node join event on coordinator, node: %s, id: %s", node.URI, node.ID) @@ -1902,6 +1912,17 @@ func (c *cluster) nodeJoin(node *Node) error { // nodeLeave initiates the removal of a node from the cluster. func (c *cluster) nodeLeave(nodeID string) error { + c.abortAntiEntropy() + // Technically there is a race condition here which could + // allow the anti-entropy process to re-start (and acquire + // the lock) before this lock has time to succeed. In that + // case, the user would have to wait through an entire + // anti-entropy cycle. We decided it wasn't worth the + // complexity (of, for example, implementing this with + // channels) to avoid that rare case. + c.muAntiEntropy.Lock() + defer c.muAntiEntropy.Unlock() + c.mu.Lock() defer c.mu.Unlock() // Refuse the request if this is not the coordinator. diff --git a/server.go b/server.go index 48ab7bab6..cecc520e4 100644 --- a/server.go +++ b/server.go @@ -664,10 +664,13 @@ func (s *Server) monitorAntiEntropy() { } // Sync holders. s.logger.Printf("holder sync beginning") + s.cluster.muAntiEntropy.Lock() if err := s.syncer.SyncHolder(); err != nil { + s.cluster.muAntiEntropy.Unlock() s.logger.Printf("holder sync error: err=%s", err) continue } + s.cluster.muAntiEntropy.Unlock() // Record successful sync in log. s.logger.Printf("holder sync complete")