diff --git a/Makefile b/Makefile
index 58d8180ee..0d3554a65 100644
--- a/Makefile
+++ b/Makefile
@@ -229,7 +229,7 @@ docker-test:
 # The \-\-\- FAIL avoids counting the extra two FAIL strings at then bottom of log.topt.
 topt:
 	mv log.topt.roar log.topt.roar.prev || true
-	$(eval SHELL:=/bin/bash) set -o pipefail; go test -v -tags='$(BUILD_TAGS) $(TEST_TAGS)' $(TESTFLAGS) 2>&1 | tee log.topt.roar
+	$(eval SHELL:=/bin/bash) set -o pipefail; go test -v -timeout 60m -tags='$(BUILD_TAGS) $(TEST_TAGS)' $(TESTFLAGS) 2>&1 | tee log.topt.roar
 	@echo "   log.topt.roar green: \c"; cat log.topt.roar | grep PASS |wc -l
 	@echo "   log.topt.roar   red: \c"; cat log.topt.roar | grep '\-\-\- FAIL' | wc -l
 
diff --git a/api.go b/api.go
index 2902fde2a..741d48894 100644
--- a/api.go
+++ b/api.go
@@ -130,7 +130,10 @@ func appendMap(a, b map[apiMethod]struct{}) map[apiMethod]struct{} {
 }
 
 func (api *API) validate(f apiMethod) error {
-	state := api.cluster.State()
+	state, err := api.cluster.State()
+	if err != nil {
+		return errors.Wrap(err, "getting cluster state")
+	}
 	if _, ok := validAPIMethods[state][f]; ok {
 		return nil
 	}
@@ -207,7 +210,11 @@ func (api *API) CreateIndex(ctx context.Context, indexName string, options Index
 		return nil, errors.Wrap(err, "validating api method")
 	}
 
-	if !api.holder.isCoordinator() {
+	// Create a snapshot of the cluster to use for node/partition calculations.
+	snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN)
+
+	if !snap.IsPrimaryFieldTranslationNode(api.Node().ID) {
+		fmt.Println("--- DEBUG: forward to coordinator")
 		if err := api.server.defaultClient.CreateIndex(ctx, indexName, options); err != nil {
 			return nil, errors.Wrap(err, "forwarding CreateIndex to coordinator")
 		}
@@ -303,7 +310,10 @@ func (api *API) CreateField(ctx context.Context, indexName string, fieldName str
 		}
 	}
 
-	if !api.holder.isCoordinator() {
+	// Create a snapshot of the cluster to use for node/partition calculations.
+	snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN)
+
+	if !snap.IsPrimaryFieldTranslationNode(api.Node().ID) {
 		if err := api.server.defaultClient.CreateFieldWithOptions(ctx, indexName, fieldName, fo); err != nil {
 			return nil, errors.Wrap(err, "forwarding CreateField to coordinator")
 		}
@@ -834,6 +844,13 @@ func (api *API) Node() *topology.Node {
 	return api.server.node()
 }
 
+// CoordinatorNode returns the coordinator node for the cluster.
+func (api *API) CoordinatorNode() *topology.Node {
+	// Create a snapshot of the cluster to use for node/partition calculations.
+	snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN)
+	return snap.PrimaryFieldTranslationNode()
+}
+
 // NodeUsage represents all usage measurements for one node.
 type NodeUsage struct {
 	Disk DiskUsage `json:"bytesOnDisk"`
@@ -1791,7 +1808,7 @@ func (api *API) ResizeAbort() error {
 // State returns the cluster state which is usually "NORMAL", but could be
 // "STARTING", "RESIZING", or potentially others. See cluster.go for more
 // details.
-func (api *API) State() string {
+func (api *API) State() (string, error) {
 	return api.cluster.State()
 }
 
@@ -2125,7 +2142,10 @@ func (api *API) ReserveIDs(key IDAllocKey, session [32]byte, offset uint64, coun
 		return nil, errors.Wrap(err, "validating api method")
 	}
 
-	if api.holder.isCoordinator() {
+	// Create a snapshot of the cluster to use for node/partition calculations.
+	snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN)
+
+	if !snap.IsPrimaryFieldTranslationNode(api.Node().ID) {
 		return api.holder.ida.reserve(key, session, offset, count)
 	}
 
@@ -2137,7 +2157,10 @@ func (api *API) CommitIDs(key IDAllocKey, session [32]byte, count uint64) error
 		return errors.Wrap(err, "validating api method")
 	}
 
-	if api.holder.isCoordinator() {
+	// Create a snapshot of the cluster to use for node/partition calculations.
+	snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN)
+
+	if !snap.IsPrimaryFieldTranslationNode(api.Node().ID) {
 		return api.holder.ida.commit(key, session, count)
 	}
 
@@ -2149,7 +2172,10 @@ func (api *API) ResetIDAlloc(index string) error {
 		return errors.Wrap(err, "validating api method")
 	}
 
-	if api.holder.isCoordinator() {
+	// Create a snapshot of the cluster to use for node/partition calculations.
+	snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN)
+
+	if !snap.IsPrimaryFieldTranslationNode(api.Node().ID) {
 		return api.holder.ida.reset(index)
 	}
 
diff --git a/api_test.go b/api_test.go
index b07e337fc..3d452ed97 100644
--- a/api_test.go
+++ b/api_test.go
@@ -161,7 +161,6 @@ func TestAPI_ImportColumnAttrs(t *testing.T) {
 				t.Fatal(err)
 			}
 		}
-
 	})
 }
 
diff --git a/cluster.go b/cluster.go
index 2046f9f46..a3c9b5d6a 100644
--- a/cluster.go
+++ b/cluster.go
@@ -77,9 +77,8 @@ type cluster struct { // nolint: maligned
 	noder            topology.Noder
 	unprotectedNoder topology.Noder
 
-	id    string
-	Node  *topology.Node
-	nodes []*topology.Node
+	id   string
+	Node *topology.Node
 
 	// Hashing algorithm used to assign partitions to nodes.
 	Hasher topology.Hasher
@@ -143,7 +142,7 @@ type cluster struct { // nolint: maligned
 
 // newCluster returns a new instance of Cluster with defaults.
 func newCluster() *cluster {
-	c := &cluster{
+	return &cluster{
 		Hasher:     &topology.Jmphasher{},
 		partitionN: topology.DefaultPartitionN,
 		ReplicaN:   1,
@@ -161,40 +160,10 @@ func newCluster() *cluster {
 
 		confirmDownRetries: defaultConfirmDownRetries,
 		confirmDownSleep:   defaultConfirmDownSleep,
+
+		noder:  topology.NewEmptyLocalNoder(),
+		stator: disco.NopStator,
 	}
-
-	// TODO: these are temporary until etcd fully implements noder
-	c.noder = c
-	c.unprotectedNoder = &unprotectedCluster{
-		c: c,
-	}
-
-	return c
-}
-
-// unprotectedCluster is a temporary struct used in cases of NewClusterSnapshot
-// which are inside of a c.mu.Lock(). These cases can't use the normal c.noder
-// (which is also temporary), because c.Nodes() aquires c.mu.Lock() as well.
-type unprotectedCluster struct {
-	c *cluster
-}
-
-// Nodes returns a copy of the slice of nodes in the cluster.
-func (uc *unprotectedCluster) Nodes() []*topology.Node {
-	ret := make([]*topology.Node, len(uc.c.nodes))
-	copy(ret, uc.c.nodes)
-	return ret
-}
-
-// SetNodes implements the Noder interface.
-func (uc *unprotectedCluster) SetNodes(nodes []*topology.Node) {}
-
-// AppendNode implements the Noder interface.
-func (uc *unprotectedCluster) AppendNode(node *topology.Node) {}
-
-// RemoveNode implements the Noder interface.
-func (uc *unprotectedCluster) RemoveNode(nodeID string) bool {
-	return false
 }
 
 // initializeAntiEntropy is called by the anti entropy routine when it starts.
@@ -225,25 +194,25 @@ func (c *cluster) abortAntiEntropy() {
 }
 
 func (c *cluster) coordinatorNode() *topology.Node {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
 	return c.unprotectedCoordinatorNode()
 }
 
 // unprotectedCoordinatorNode returns the coordinator node.
 func (c *cluster) unprotectedCoordinatorNode() *topology.Node {
-	return c.unprotectedNodeByID(c.Coordinator)
+	// Create a snapshot of the cluster to use for node/partition calculations.
+	snap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN)
+	return snap.PrimaryFieldTranslationNode()
 }
 
 // isCoordinator is true if this node is the coordinator.
 func (c *cluster) isCoordinator() bool {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
 	return c.unprotectedIsCoordinator()
 }
 
 func (c *cluster) unprotectedIsCoordinator() bool {
-	return c.Coordinator == c.Node.ID
+	// Create a snapshot of the cluster to use for node/partition calculations.
+	snap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN)
+	return snap.PrimaryFieldTranslationNode().ID == c.Node.ID
 }
 
 // setCoordinator tells the current node to become the
@@ -282,7 +251,7 @@ func (c *cluster) setCoordinator(n *topology.Node) error {
 // and should be refactored.
 func (c *cluster) unprotectedSendSync(m Message) error {
 	var eg errgroup.Group
-	for _, node := range c.nodes {
+	for _, node := range c.noder.Nodes() {
 		node := node
 		// Don't send to myself.
 		if node.ID == c.Node.ID {
@@ -309,7 +278,7 @@ func (c *cluster) unprotectedUpdateCoordinator(n *topology.Node) bool {
 		c.Coordinator = n.ID
 		changed = true
 	}
-	for _, node := range c.nodes {
+	for _, node := range c.noder.Nodes() {
 		if node.ID == n.ID {
 			node.IsCoordinator = true
 		} else {
@@ -365,7 +334,7 @@ func (c *cluster) removeNode(nodeID string) error {
 
 // nodeIDs returns the list of IDs in the cluster.
 func (c *cluster) nodeIDs() []string {
-	return topology.Nodes(c.nodes).IDs()
+	return topology.Nodes(c.Nodes()).IDs()
 }
 
 func (c *cluster) unprotectedSetID(id string) {
@@ -379,10 +348,12 @@ func (c *cluster) unprotectedSetID(id string) {
 	c.Topology.clusterID = c.id
 }
 
-func (c *cluster) State() string {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-	return c.state
+func (c *cluster) State() (string, error) {
+	state, err := c.stator.ClusterState(context.Background())
+	if err != nil {
+		return string(disco.ClusterStateUnknown), err
+	}
+	return string(state), nil
 }
 
 func (c *cluster) SetState(state string) {
@@ -456,33 +427,14 @@ func (c *cluster) setMyNodeState(state string) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	c.Node.State = state
-	for i, n := range c.nodes {
+	nodes := c.noder.Nodes()
+	for i, n := range nodes {
 		if n.ID == c.Node.ID {
-			c.nodes[i].State = state
+			nodes[i].State = state
 		}
 	}
 }
 
-func (c *cluster) setNodeState(state string) error { // nolint: unparam
-	c.setMyNodeState(state)
-	if c.isCoordinator() {
-		return c.receiveNodeState(c.Node.ID, state)
-	}
-
-	// Send node state to coordinator.
-	ns := &NodeStateMessage{
-		NodeID: c.Node.ID,
-		State:  state,
-	}
-
-	c.logger.Printf("sending state %s (%s)", state, c.Coordinator)
-	if err := c.sendTo(c.coordinatorNode(), ns); err != nil {
-		return fmt.Errorf("sending node state error: err=%s", err)
-	}
-
-	return nil
-}
-
 // receiveNodeState sets node state in Topology in order for the
 // Coordinator to keep track of, during startup, which nodes have
 // finished opening their Holder.
@@ -498,11 +450,12 @@ func (c *cluster) receiveNodeState(nodeID string, state string) error {
 	if c.Topology.nodeStates[nodeID] != state {
 		changed = true
 		c.Topology.nodeStates[nodeID] = state
-		for i, n := range c.nodes {
+		nodes := c.noder.Nodes()
+		for i, n := range nodes {
 			if n.ID == nodeID {
-				c.nodes[i].Mu.Lock()
-				c.nodes[i].State = state
-				c.nodes[i].Mu.Unlock()
+				nodes[i].Mu.Lock()
+				nodes[i].State = state
+				nodes[i].Mu.Unlock()
 			}
 		}
 	}
@@ -547,7 +500,7 @@ func (c *cluster) unprotectedStatus() *ClusterStatus {
 	return &ClusterStatus{
 		ClusterID: c.id,
 		State:     c.state,
-		Nodes:     c.nodes,
+		Nodes:     c.noder.Nodes(),
 		Schema:    &Schema{Indexes: c.holder.Schema()},
 	}
 }
@@ -560,7 +513,7 @@ func (c *cluster) nodeByID(id string) *topology.Node {
 
 // unprotectedNodeByID returns a node reference by ID.
 func (c *cluster) unprotectedNodeByID(id string) *topology.Node {
-	for _, n := range c.nodes {
+	for _, n := range c.noder.Nodes() {
 		if n.ID == id {
 			return n
 		}
@@ -581,7 +534,7 @@ func (c *cluster) topologyContainsNode(id string) bool {
 
 // nodePositionByID returns the position of the node in slice c.Nodes.
 func (c *cluster) nodePositionByID(nodeID string) int {
-	for i, n := range c.nodes {
+	for i, n := range c.noder.Nodes() {
 		if n.ID == nodeID {
 			return i
 		}
@@ -609,10 +562,10 @@ func (c *cluster) addNodeBasicSorted(node *topology.Node) bool {
 		return false
 	}
 
-	c.nodes = append(c.nodes, node)
+	c.noder.AppendNode(node)
 
 	// All hosts must be merged in the same order on all nodes in the cluster.
-	sort.Sort(topology.ByID(c.nodes))
+	// sort.Sort(topology.ByID(c.nodes)) // TODO: this should no longer apply
 
 	return true
 }
@@ -620,11 +573,25 @@ func (c *cluster) addNodeBasicSorted(node *topology.Node) bool {
 // Nodes returns a copy of the slice of nodes in the cluster. Safe for
 // concurrent use, result may be modified.
 func (c *cluster) Nodes() []*topology.Node {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-	ret := make([]*topology.Node, len(c.nodes))
-	copy(ret, c.nodes)
-	return ret
+	nodes := c.noder.Nodes()
+
+	// Create a snapshot of the cluster to use for node/partition calculations.
+	snap := topology.NewClusterSnapshot(topology.NewLocalNoder(nodes), c.Hasher, c.ReplicaN)
+	primaryNode := snap.PrimaryFieldTranslationNode()
+
+	// Set node states and IsPrimary.
+	for _, node := range nodes {
+		node.IsCoordinator = node.ID == primaryNode.ID
+		// s, err := c.stator.NodeState(context.Background(), node.ID)
+		// if err != nil {
+		// 	node.State = nodeStateDown
+		// 	continue
+		// }
+		// node.State = string(s)
+
+	}
+
+	return nodes
 }
 
 func (c *cluster) AllNodeStates() map[string]string {
@@ -636,16 +603,7 @@ func (c *cluster) AllNodeStates() map[string]string {
 // removeNodeBasicSorted removes a node from the cluster, maintaining the sort
 // order. Returns true if the node was removed. unprotected.
 func (c *cluster) removeNodeBasicSorted(nodeID string) bool {
-	i := c.nodePositionByID(nodeID)
-	if i < 0 {
-		return false
-	}
-
-	copy(c.nodes[i:], c.nodes[i+1:])
-	c.nodes[len(c.nodes)-1] = nil
-	c.nodes = c.nodes[:len(c.nodes)-1]
-
-	return true
+	return c.noder.RemoveNode(nodeID)
 }
 
 // frag is a struct of basic fragment information.
@@ -699,7 +657,7 @@ func (c *cluster) fragsByHost(idx *Index) fragsByHost {
 // for the given set of shards with data.
 func (c *cluster) fragCombos(idx string, availableShards *roaring.Bitmap, fieldViews viewsByField) fragsByHost {
 	// Create a snapshot of the cluster to use for node/partition calculations.
-	snap := topology.NewClusterSnapshot(c.unprotectedNoder, c.Hasher, c.ReplicaN)
+	snap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN)
 
 	t := make(fragsByHost)
 	_ = availableShards.ForEach(func(i uint64) error {
@@ -721,8 +679,10 @@ func (c *cluster) fragCombos(idx string, availableShards *roaring.Bitmap, fieldV
 // added or removed. An error is returned for any case other than where
 // exactly one node is added or removed. unprotected.
 func (c *cluster) diff(other *cluster) (action string, nodeID string, err error) {
-	lenFrom := len(c.nodes)
-	lenTo := len(other.nodes)
+	cNodes := c.noder.Nodes()
+	otherNodes := other.noder.Nodes()
+	lenFrom := len(cNodes)
+	lenTo := len(otherNodes)
 	// Determine if a node is being added or removed.
 	if lenFrom == lenTo {
 		return "", "", errors.New("clusters are the same size")
@@ -734,7 +694,7 @@ func (c *cluster) diff(other *cluster) (action string, nodeID string, err error)
 		}
 		action = resizeJobActionAdd
 		// Determine the node ID that is being added.
-		for _, n := range other.nodes {
+		for _, n := range otherNodes {
 			if c.unprotectedNodeByID(n.ID) == nil {
 				nodeID = n.ID
 				break
@@ -747,7 +707,7 @@ func (c *cluster) diff(other *cluster) (action string, nodeID string, err error)
 		}
 		action = resizeJobActionRemove
 		// Determine the node ID that is being removed.
-		for _, n := range c.nodes {
+		for _, n := range cNodes {
 			if other.unprotectedNodeByID(n.ID) == nil {
 				nodeID = n.ID
 				break
@@ -769,7 +729,7 @@ func (c *cluster) fragSources(to *cluster, idx *Index) (map[string][]*ResizeSour
 	}
 
 	// Initialize the map with all the nodes in `to`.
-	for _, n := range to.nodes {
+	for _, n := range to.noder.Nodes() {
 		m[n.ID] = nil
 	}
 
@@ -782,7 +742,7 @@ func (c *cluster) fragSources(to *cluster, idx *Index) (map[string][]*ResizeSour
 	srcCluster := c
 	if action == resizeJobActionAdd && c.ReplicaN > 1 {
 		srcCluster = newCluster()
-		srcCluster.nodes = topology.Nodes(c.nodes).Clone()
+		srcCluster.noder.SetNodes(topology.Nodes(c.noder.Nodes()).Clone())
 		srcCluster.Hasher = c.Hasher
 		srcCluster.partitionN = c.partitionN
 		srcCluster.ReplicaN = 1
@@ -859,13 +819,13 @@ func (c *cluster) translationNodes(to *cluster) (map[string][]*translationResize
 	}
 
 	// Initialize the map with all the nodes in `to`.
-	for _, n := range to.nodes {
+	for _, n := range to.noder.Nodes() {
 		m[n.ID] = nil
 	}
 
 	// Create a snapshot of the cluster to use for node/partition calculations.
-	fSnap := topology.NewClusterSnapshot(c.unprotectedNoder, c.Hasher, c.ReplicaN)
-	toSnap := topology.NewClusterSnapshot(to.unprotectedNoder, c.Hasher, to.ReplicaN)
+	fSnap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN)
+	toSnap := topology.NewClusterSnapshot(to.noder, c.Hasher, to.ReplicaN)
 
 	for pid := 0; pid < c.partitionN; pid++ {
 		fNodes := fSnap.PartitionNodes(pid)
@@ -914,7 +874,7 @@ func (c *cluster) translationNodes(to *cluster) (map[string][]*translationResize
 func (c *cluster) shardDistributionByIndex(indexName string) map[string]map[string][]uint64 {
 	dist := make(map[string]map[string][]uint64)
 
-	for _, node := range c.nodes {
+	for _, node := range c.noder.Nodes() {
 		nodeDist := make(map[string][]uint64)
 		nodeDist["primary-shards"] = make([]uint64, 0)
 		nodeDist["replica-shards"] = make([]uint64, 0)
@@ -1017,12 +977,14 @@ func (c *cluster) partitionNodes(partitionID int) []*topology.Node {
 		useTopology = true
 	}
 
+	cNodes := c.noder.Nodes()
+
 	replicaN := c.ReplicaN
 	var nodeN int
 	if useTopology {
 		nodeN = len(c.Topology.nodeIDs)
 	} else {
-		nodeN = len(c.nodes)
+		nodeN = len(cNodes)
 	}
 	if replicaN > nodeN {
 		replicaN = nodeN
@@ -1044,11 +1006,11 @@ func (c *cluster) partitionNodes(partitionID int) []*topology.Node {
 	for i := 0; i < replicaN; i++ {
 		if useTopology {
 			maybeNodeID := c.Topology.nodeIDs[(nodeIndex+i)%nodeN]
-			if node := topology.Nodes(c.nodes).NodeByID(maybeNodeID); node != nil {
+			if node := topology.Nodes(cNodes).NodeByID(maybeNodeID); node != nil {
 				nodes = append(nodes, node)
 			}
 		} else {
-			nodes = append(nodes, c.nodes[(nodeIndex+i)%len(c.nodes)])
+			nodes = append(nodes, cNodes[(nodeIndex+i)%len(cNodes)])
 		}
 	}
 
@@ -1078,7 +1040,7 @@ func (t *Topology) PrimaryNodeIndex(partitionID int) (nodeIndex int) {
 	n := len(t.nodeIDs)
 	if n == 0 {
 		if t.cluster != nil {
-			n = len(t.cluster.nodes)
+			n = len(t.cluster.noder.Nodes())
 		}
 	}
 	nodeIndex = t.Hasher.Hash(uint64(partitionID), n)
@@ -1178,28 +1140,6 @@ func (c *cluster) open() error {
 }
 
 func (c *cluster) waitForStarted() error {
-	// If not coordinator then wait for ClusterStatus from coordinator.
-	if !c.isCoordinator() {
-		// In the case where a node has been restarted and memberlist has
-		// not had enough time to determine the node went down/up, then
-		// the coordinator needs to be alerted that this node is back up
-		// (and now in a state of STARTING) so that it can be put to the correct
-		// cluster state.
-		// TODO: Because the normal code path already sends a NodeJoin event (via
-		// memberlist), this is a bit redundant in most cases. Perhaps determine
-		// that the node has been restarted and don't do this step.
-		msg := &NodeEvent{
-			Event: NodeJoin,
-			Node:  c.Node,
-		}
-		if err := c.broadcaster.SendSync(msg); err != nil {
-			return fmt.Errorf("sending restart NodeJoin: %v", err)
-		}
-
-		c.logger.Printf("%v wait for joining to complete", c.Node.ID)
-		<-c.joining
-		c.logger.Printf("joining has completed. I am NodeID '%v'", c.Node.ID)
-	}
 	return nil
 }
 
@@ -1220,7 +1160,7 @@ func (c *cluster) markAsJoined() {
 
 // needTopologyAgreement is unprotected.
 func (c *cluster) needTopologyAgreement() bool {
-	return (c.state == ClusterStateStarting || c.state == ClusterStateDegraded) && !stringSlicesAreEqual(c.Topology.nodeIDs, c.nodeIDs())
+	return false
 }
 
 // haveTopologyAgreement is unprotected.
@@ -1405,7 +1345,7 @@ func (c *cluster) unprotectedGenerateResizeJob(nodeAction nodeAction) (*resizeJo
 // Broadcaster is associated to the resizeJob here for use in broadcasting
 // the resize instructions to other nodes in the cluster.
 func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (*resizeJob, error) {
-	j := newResizeJob(c.nodes, nodeAction.node, nodeAction.action)
+	j := newResizeJob(c.noder.Nodes(), nodeAction.node, nodeAction.action)
 	// A *new* node which is being added needs a schema update even if
 	// there's no data to send it.
 	var sendSchemaToNewNode string
@@ -1413,7 +1353,7 @@ func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (*
 
 	// toCluster is a clone of Cluster with the new node added/removed for comparison.
 	toCluster := newCluster()
-	toCluster.nodes = topology.Nodes(c.nodes).Clone()
+	toCluster.noder.SetNodes(topology.Nodes(c.noder.Nodes()).Clone())
 	toCluster.Hasher = c.Hasher
 	toCluster.partitionN = c.partitionN
 	toCluster.ReplicaN = c.ReplicaN
@@ -1429,7 +1369,7 @@ func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (*
 	// fragmentSourcesByNode is a map of Node.ID to sources of fragment data.
 	// It is initialized with all the nodes in toCluster.
 	fragmentSourcesByNode := make(map[string][]*ResizeSource)
-	for _, n := range toCluster.nodes {
+	for _, n := range toCluster.noder.Nodes() {
 		fragmentSourcesByNode[n.ID] = nil
 	}
 
@@ -1449,7 +1389,7 @@ func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (*
 	// key translation data for indexes.
 	// It is initialized with all the nodes in toCluster.
 	translationSourcesByNode := make(map[string][]*TranslationResizeSource)
-	for _, n := range toCluster.nodes {
+	for _, n := range toCluster.noder.Nodes() {
 		translationSourcesByNode[n.ID] = nil
 	}
 
@@ -1486,7 +1426,7 @@ func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (*
 		}
 	}
 
-	for _, node := range toCluster.nodes {
+	for _, node := range toCluster.noder.Nodes() {
 		dataToSend := len(fragmentSourcesByNode[node.ID]) != 0 || len(translationSourcesByNode[node.ID]) != 0
 		// If we're adding a new node, that node needs to get a resize
 		// instruction even if there's no data it needs to read.
@@ -1498,7 +1438,7 @@ func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (*
 		}
 
 		// Create a snapshot of the cluster to use for node/partition calculations.
-		snap := topology.NewClusterSnapshot(c.unprotectedNoder, c.Hasher, c.ReplicaN)
+		snap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN)
 
 		instr := &ResizeInstruction{
 			JobID:              j.ID,
@@ -1525,7 +1465,7 @@ func (c *cluster) completeCurrentJob(state string) error {
 
 func (c *cluster) unprotectedCompleteCurrentJob(state string) error {
 	// Create a snapshot of the cluster to use for node/partition calculations.
-	snap := topology.NewClusterSnapshot(c.unprotectedNoder, c.Hasher, c.ReplicaN)
+	snap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN)
 	if !snap.IsPrimaryFieldTranslationNode(c.Node.ID) {
 		return ErrNodeNotCoordinator
 	}
@@ -2373,15 +2313,6 @@ func (c *cluster) mergeClusterStatus(cs *ClusterStatus) error {
 
 	// Add all nodes from the coordinator.
 	for _, node := range officialNodes {
-		if node.ID == c.Node.ID && node.State != c.Node.State {
-			c.logger.Printf("mismatched state in mergeClusterStatus got %v have %v", node.State, c.Node.State)
-			go func(fromState, toState string) {
-				err := c.setNodeState(toState)
-				if err != nil {
-					c.logger.Printf("error setting node state from %v to %v: %v", fromState, toState, err)
-				}
-			}(node.State, c.Node.State)
-		}
 		if err := c.addNode(node); err != nil {
 			return errors.Wrap(err, "adding node")
 		}
@@ -2391,7 +2322,7 @@ func (c *cluster) mergeClusterStatus(cs *ClusterStatus) error {
 	// except for self. Generate a list to remove first
 	// so that nodes aren't removed mid-loop.
 	nodeIDsToRemove := []string{}
-	for _, node := range c.nodes {
+	for _, node := range c.noder.Nodes() {
 		// Don't remove this node.
 		if node.ID == c.Node.ID {
 			continue
@@ -2419,7 +2350,8 @@ func (c *cluster) mergeClusterStatus(cs *ClusterStatus) error {
 // If there is only one node in the cluster, returns nil.
 // If the current node is the first node in the list, returns the last node.
 func (c *cluster) unprotectedPreviousNode() *topology.Node {
-	if len(c.nodes) <= 1 {
+	cNodes := c.noder.Nodes()
+	if len(cNodes) <= 1 {
 		return nil
 	}
 
@@ -2427,9 +2359,9 @@ func (c *cluster) unprotectedPreviousNode() *topology.Node {
 	if pos == -1 {
 		return nil
 	} else if pos == 0 {
-		return c.nodes[len(c.nodes)-1]
+		return cNodes[len(cNodes)-1]
 	} else {
-		return c.nodes[pos-1]
+		return cNodes[pos-1]
 	}
 }
 
@@ -2446,7 +2378,8 @@ func (c *cluster) unprotectedPrimaryReplicaNode() *topology.Node {
 	if pos <= 0 {
 		return nil
 	}
-	return c.nodes[pos-1]
+	cNodes := c.noder.Nodes()
+	return cNodes[pos-1]
 }
 
 // translateFieldKeys is basically a wrapper around
@@ -2484,7 +2417,7 @@ func (c *cluster) findFieldKeys(ctx context.Context, field *Field, keys ...strin
 	}
 
 	if !field.Keys() {
-		return nil, errors.Wrap(ErrTranslatingKeyNotFound, "field is not keyed 1")
+		return nil, errors.Wrap(ErrTranslatingKeyNotFound, "field is not keyed")
 	}
 
 	// Attempt to find the keys locally.
@@ -2547,7 +2480,7 @@ func (c *cluster) createFieldKeys(ctx context.Context, field *Field, keys ...str
 	}
 
 	if !field.Keys() {
-		return nil, errors.Wrap(ErrTranslatingKeyNotFound, "field is not keyed 2")
+		return nil, errors.Wrap(ErrTranslatingKeyNotFound, "field is not keyed")
 	}
 
 	// The coordinator is the only node that can create field keys, since it owns the authoritative copy.
diff --git a/cluster_internal_test.go b/cluster_internal_test.go
index fe3edc894..09119b864 100644
--- a/cluster_internal_test.go
+++ b/cluster_internal_test.go
@@ -419,22 +419,24 @@ func TestResizeJob(t *testing.T) {
 // Ensure the cluster can fairly distribute partitions across the nodes.
 func TestCluster_Owners(t *testing.T) {
 	c := cluster{
-		nodes: []*topology.Node{
+		noder: topology.NewLocalNoder([]*topology.Node{
 			{URI: NewTestURIFromHostPort("serverA", 1000)},
 			{URI: NewTestURIFromHostPort("serverB", 1000)},
 			{URI: NewTestURIFromHostPort("serverC", 1000)},
-		},
+		}),
 		Hasher:   NewTestModHasher(),
 		ReplicaN: 2,
 	}
 
+	cNodes := c.noder.Nodes()
+
 	// Verify nodes are distributed.
-	if a := c.partitionNodes(0); !reflect.DeepEqual(a, []*topology.Node{c.nodes[0], c.nodes[1]}) {
+	if a := c.partitionNodes(0); !reflect.DeepEqual(a, []*topology.Node{cNodes[0], cNodes[1]}) {
 		t.Fatalf("unexpected owners: %s", spew.Sdump(a))
 	}
 
 	// Verify nodes go around the ring.
-	if a := c.partitionNodes(2); !reflect.DeepEqual(a, []*topology.Node{c.nodes[2], c.nodes[0]}) {
+	if a := c.partitionNodes(2); !reflect.DeepEqual(a, []*topology.Node{cNodes[2], cNodes[0]}) {
 		t.Fatalf("unexpected owners: %s", spew.Sdump(a))
 	}
 }
@@ -487,7 +489,8 @@ func TestHasher(t *testing.T) {
 func TestCluster_ContainsShards(t *testing.T) {
 	c := NewTestCluster(t, 5)
 	c.ReplicaN = 3
-	shards := c.containsShards("test", roaring.NewBitmap(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10), c.nodes[2])
+	cNodes := c.noder.Nodes()
+	shards := c.containsShards("test", roaring.NewBitmap(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10), cNodes[2])
 
 	if !reflect.DeepEqual(shards, []uint64{0, 2, 3, 5, 6, 9, 10}) {
 		t.Fatalf("unexpected shars for node's index: %v", shards)
@@ -627,13 +630,16 @@ func TestCluster_Coordinator(t *testing.T) {
 
 	node1 := &topology.Node{ID: "node1", URI: uris[0]}
 	node2 := &topology.Node{ID: "node2", URI: uris[1]}
+	noder := topology.NewLocalNoder([]*topology.Node{node1, node2})
 
 	c1 := *newCluster()
 	c1.Node = node1
 	c1.Coordinator = node1.ID
+	c1.noder = noder
 	c2 := *newCluster()
 	c2.Node = node2
 	c2.Coordinator = node1.ID
+	c2.noder = noder
 
 	t.Run("IsCoordinator", func(t *testing.T) {
 		if !c1.isCoordinator() {
@@ -697,7 +703,7 @@ func TestCluster_Topology(t *testing.T) {
 
 // Ensure that general cluster functionality works as expected.
 func TestCluster_ResizeStates(t *testing.T) {
-
+	t.Skip("these tests don't really apply anymore; they were meant to tests the cluster startup process using memberlist and a topology file")
 	t.Run("Single node, no data", func(t *testing.T) {
 		tc := NewClusterCluster(t, 1)
 
@@ -708,9 +714,14 @@ func TestCluster_ResizeStates(t *testing.T) {
 
 		node := tc.Clusters[0]
 
+		state, err := node.State()
+		if err != nil {
+			t.Fatal(err)
+		}
+
 		// Ensure that node comes up in state NORMAL.
-		if node.State() != ClusterStateNormal {
-			t.Errorf("expected state: %v, but got: %v", ClusterStateNormal, node.State())
+		if state != ClusterStateNormal {
+			t.Errorf("expected state: %v, but got: %v", ClusterStateNormal, state)
 		}
 
 		expectedTop := &Topology{
@@ -749,9 +760,14 @@ func TestCluster_ResizeStates(t *testing.T) {
 			t.Fatal(err)
 		}
 
+		state, err := node.State()
+		if err != nil {
+			t.Fatal(err)
+		}
+
 		// Ensure that node comes up in state NORMAL.
-		if node.State() != ClusterStateNormal {
-			t.Errorf("expected state: %v, but got: %v", ClusterStateNormal, node.State())
+		if state != ClusterStateNormal {
+			t.Errorf("expected state: %v, but got: %v", ClusterStateNormal, state)
 		}
 
 		// Close TestCluster.
@@ -805,13 +821,22 @@ func TestCluster_ResizeStates(t *testing.T) {
 		}
 
 		node0 := tc.Clusters[0]
+		state0, err := node0.State()
+		if err != nil {
+			t.Fatal(err)
+		}
+
 		node1 := tc.Clusters[1]
+		state1, err := node1.State()
+		if err != nil {
+			t.Fatal(err)
+		}
 
 		// Ensure that nodes comes up in state NORMAL.
-		if node0.State() != ClusterStateNormal {
-			t.Errorf("expected node0 state: %v, but got: %v", ClusterStateNormal, node0.State())
-		} else if node1.State() != ClusterStateNormal {
-			t.Errorf("expected node1 state: %v, but got: %v", ClusterStateNormal, node1.State())
+		if state0 != ClusterStateNormal {
+			t.Errorf("expected node0 state: %v, but got: %v", ClusterStateNormal, state0)
+		} else if state1 != ClusterStateNormal {
+			t.Errorf("expected node1 state: %v, but got: %v", ClusterStateNormal, state1)
 		}
 
 		expectedTop := &Topology{
@@ -851,27 +876,30 @@ func TestCluster_ResizeStates(t *testing.T) {
 			t.Fatalf("opening cluster: %v", err)
 		}
 
-		// Ensure that node is in state STARTING before the other node joins.
-		if node0.State() != ClusterStateStarting {
-			t.Errorf("expected node0 state: %v, but got: %v", ClusterStateStarting, node0.State())
+		state0, err := node0.State()
+		if err != nil {
+			t.Fatal(err)
 		}
 
-		// Expect an error by adding a node not in the topology.
-		expectedError := "host is not in topology: node1"
-		if err := tc.addNode(); err == nil || err.Error() != expectedError {
-			t.Errorf("did not receive expected error: %s", expectedError)
+		// Ensure that node is in state STARTING before the other node joins.
+		if state0 != ClusterStateStarting {
+			t.Errorf("expected node0 state: %v, but got: %v", ClusterStateStarting, state0)
 		}
 
 		if err := tc.addNode(); err != nil {
 			t.Fatalf("adding node: %v", err)
 		}
-		node2 := tc.Clusters[2]
+		node1 := tc.Clusters[1]
+		state1, err := node1.State()
+		if err != nil {
+			t.Fatal(err)
+		}
 
 		// Ensure that node comes up in state NORMAL.
-		if node0.State() != ClusterStateNormal {
-			t.Errorf("expected node0 state: %v, but got: %v", ClusterStateNormal, node0.State())
-		} else if node2.State() != ClusterStateNormal {
-			t.Errorf("expected node2 state: %v, but got: %v", ClusterStateNormal, node2.State())
+		if state0 != ClusterStateNormal {
+			t.Errorf("expected node0 state: %v, but got: %v", ClusterStateNormal, state0)
+		} else if state1 != ClusterStateNormal {
+			t.Errorf("expected node2 state: %v, but got: %v", ClusterStateNormal, state1)
 		}
 
 		// Close TestCluster.
@@ -933,11 +961,21 @@ func TestCluster_ResizeStates(t *testing.T) {
 
 		node1 := tc.Clusters[1]
 
+		state1, err := node1.State()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		state0, err := node0.State()
+		if err != nil {
+			t.Fatal(err)
+		}
+
 		// Ensure that nodes come up in state NORMAL.
-		if node0.State() != ClusterStateNormal {
-			t.Errorf("expected node0 state: %v, but got: %v", ClusterStateNormal, node0.State())
-		} else if node1.State() != ClusterStateNormal {
-			t.Errorf("expected node1 state: %v, but got: %v", ClusterStateNormal, node1.State())
+		if state0 != ClusterStateNormal {
+			t.Errorf("expected node0 state: %v, but got: %v", ClusterStateNormal, state0)
+		} else if state1 != ClusterStateNormal {
+			t.Errorf("expected node1 state: %v, but got: %v", ClusterStateNormal, state1)
 		}
 		// INVAR: after node1.State() is normal, the rebalancing should have been done.
 
@@ -1030,7 +1068,6 @@ func TestAE(t *testing.T) {
 			t.Fatalf("abort should not have blocked this long")
 		}
 	})
-
 }
 
 // Ensures that coordinator can be changed.
@@ -1038,8 +1075,10 @@ func TestCluster_UpdateCoordinator(t *testing.T) {
 	t.Run("UpdateCoordinator", func(t *testing.T) {
 		c := NewTestCluster(t, 2)
 
-		oldNode := c.nodes[0]
-		newNode := c.nodes[1]
+		cNodes := c.noder.Nodes()
+
+		oldNode := cNodes[0]
+		newNode := cNodes[1]
 
 		// Update coordinator to the same value.
 		if c.updateCoordinator(oldNode) {
@@ -1085,8 +1124,8 @@ func TestCluster_confirmNodeDownUp(t *testing.T) {
 	if c.confirmNodeDown(uri) {
 		t.Errorf("expected node to be up")
 	}
-
 }
+
 func TestCluster_confirmNodeDownTimeout(t *testing.T) {
 	t.Skip("does a listen on :0, skip for now. TODO(jea) restore this.")
 	sleep := 50 * time.Millisecond
@@ -1143,7 +1182,6 @@ func TestCluster_confirmNodeDownDown(t *testing.T) {
 }
 
 func TestCluster_GetNonPrimaryReplicas(t *testing.T) {
-
 	c := newCluster()
 	c.ReplicaN = 3
 	topo := NewTopology(c.Hasher, c.partitionN, c.ReplicaN, c)
@@ -1151,7 +1189,7 @@ func TestCluster_GetNonPrimaryReplicas(t *testing.T) {
 	nNodes := 4
 	for i := 0; i < nNodes; i++ {
 		nodeID := fmt.Sprintf("node%d", i)
-		c.nodes = append(c.nodes, &topology.Node{
+		c.noder.AppendNode(&topology.Node{
 			ID:  nodeID,
 			URI: NewTestURI("http", fmt.Sprintf("host%d", i), uint16(0)),
 		})
diff --git a/disco/disco.go b/disco/disco.go
index 7cf882eaf..03443d384 100644
--- a/disco/disco.go
+++ b/disco/disco.go
@@ -173,7 +173,7 @@ type nopStator struct{}
 
 // ClusterState is a no-op implementation of the Stator ClusterState method.
 func (n *nopStator) ClusterState(context.Context) (ClusterState, error) {
-	return "", nil
+	return ClusterStateUnknown, nil
 }
 
 func (n *nopStator) Started(ctx context.Context) error {
diff --git a/executor.go b/executor.go
index 80cd0de4f..a6f733714 100644
--- a/executor.go
+++ b/executor.go
@@ -5266,7 +5266,7 @@ func (e *executor) executeSetRowAttrs(ctx context.Context, qcx *Qcx, index strin
 	}
 
 	// Execute on remote nodes in parallel.
-	nodes := topology.Nodes(e.Cluster.nodes).FilterID(e.Node.ID)
+	nodes := topology.Nodes(e.Cluster.noder.Nodes()).FilterID(e.Node.ID)
 	resp := make(chan error, len(nodes))
 	for _, node := range nodes {
 		go func(node *topology.Node) {
@@ -5378,7 +5378,7 @@ func (e *executor) executeBulkSetRowAttrs(ctx context.Context, qcx *Qcx, index s
 	}
 
 	// Execute on remote nodes in parallel.
-	nodes := topology.Nodes(e.Cluster.nodes).FilterID(e.Node.ID)
+	nodes := topology.Nodes(e.Cluster.noder.Nodes()).FilterID(e.Node.ID)
 	resp := make(chan error, len(nodes))
 	for _, node := range nodes {
 		go func(node *topology.Node) {
@@ -5430,7 +5430,7 @@ func (e *executor) executeSetColumnAttrs(ctx context.Context, qcx *Qcx, index st
 	}
 
 	// Execute on remote nodes in parallel.
-	nodes := topology.Nodes(e.Cluster.nodes).FilterID(e.Node.ID)
+	nodes := topology.Nodes(e.Cluster.noder.Nodes()).FilterID(e.Node.ID)
 	resp := make(chan error, len(nodes))
 	for _, node := range nodes {
 		go func(node *topology.Node) {
@@ -5484,7 +5484,12 @@ func (e *executor) shardsByNode(nodes []*topology.Node, index string, shards []u
 loop:
 	for _, shard := range shards {
 		for _, node := range snap.ShardNodes(index, shard) {
-			if topology.Nodes(nodes).Contains(node) {
+			// If the node being considered is in any state other than STARTED,
+			// then exclude it from the map. This way, one of that node's
+			// healthy replicas will be included instead.
+			// TODO: check state once stator is implemented
+			//if topology.Nodes(nodes).ContainsID(node.ID) && node.State == disco.NodeStateStarted {
+			if topology.Nodes(nodes).ContainsID(node.ID) {
 				m[node] = append(m[node], shard)
 				continue loop
 			}
@@ -5537,7 +5542,7 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64,
 
 			if resp.err != nil {
 				// Filter out unavailable nodes.
-				nodes = topology.Nodes(nodes).Filter(resp.node)
+				nodes = topology.Nodes(nodes).FilterID(resp.node.ID)
 
 				// Begin mapper against secondary nodes.
 				if err := e.mapper(ctx, cancel, ch, nodes, index, resp.shards, c, opt, mapFn, reduceFn); errors.Cause(err) == errShardUnavailable {
diff --git a/holder.go b/holder.go
index 01e628398..647e69144 100644
--- a/holder.go
+++ b/holder.go
@@ -1200,6 +1200,7 @@ func (h *Holder) recalculateCaches() {
 	}
 }
 
+// TODO: this needs to be removed
 func (h *Holder) isCoordinator() bool {
 	if s, ok := h.broadcaster.(*Server); ok {
 		return s.isCoordinator
@@ -1426,7 +1427,7 @@ func (s *holderSyncer) syncIndex(index string) error {
 	s.Stats.CountWithCustomTags(MetricColumnAttrStoreBlocks, int64(len(blks)), 1.0, []string{indexTag})
 
 	// Sync with every other host.
-	for _, node := range topology.Nodes(s.Cluster.nodes).FilterID(s.Node.ID) {
+	for _, node := range topology.Nodes(s.Cluster.noder.Nodes()).FilterID(s.Node.ID) {
 		// Retrieve attributes from differing blocks.
 		// Skip update and recomputation if no attributes have changed.
 		m, err := s.Cluster.InternalClient.ColumnAttrDiff(ctx, &node.URI, index, blks)
@@ -1473,7 +1474,7 @@ func (s *holderSyncer) syncField(index, name string) error {
 	s.Stats.CountWithCustomTags(MetricRowAttrStoreBlocks, int64(len(blks)), 1.0, []string{indexTag, fieldTag})
 
 	// Sync with every other host.
-	for _, node := range topology.Nodes(s.Cluster.nodes).FilterID(s.Node.ID) {
+	for _, node := range topology.Nodes(s.Cluster.noder.Nodes()).FilterID(s.Node.ID) {
 		// Retrieve attributes from differing blocks.
 		// Skip update and recomputation if no attributes have changed.
 		m, err := s.Cluster.InternalClient.RowAttrDiff(ctx, &node.URI, index, name, blks)
@@ -1836,7 +1837,7 @@ func (c *holderCleaner) IsClosing() bool {
 // any unnecessary fragments and files.
 func (c *holderCleaner) CleanHolder() error {
 	// Create a snapshot of the cluster to use for node/partition calculations.
-	snap := topology.NewClusterSnapshot(c.Cluster.unprotectedNoder, c.Cluster.Hasher, c.Cluster.ReplicaN)
+	snap := topology.NewClusterSnapshot(c.Cluster.noder, c.Cluster.Hasher, c.Cluster.ReplicaN)
 
 	for _, index := range c.Holder.Indexes() {
 		// Verify cleaner has not closed.
diff --git a/http/handler.go b/http/handler.go
index 32631b3ad..f9a016540 100644
--- a/http/handler.go
+++ b/http/handler.go
@@ -736,8 +736,15 @@ func (h *Handler) handleGetStatus(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
 		return
 	}
+
+	state, err := h.api.State()
+	if err != nil {
+		http.Error(w, "getting cluster state error: "+err.Error(), http.StatusInternalServerError)
+		return
+	}
+
 	status := getStatusResponse{
-		State:       h.api.State(),
+		State:       state,
 		Nodes:       h.api.Hosts(r.Context()),
 		LocalID:     h.api.Node().ID,
 		ClusterName: h.api.ClusterName(),
diff --git a/server.go b/server.go
index f92ad9008..7ed582643 100644
--- a/server.go
+++ b/server.go
@@ -603,21 +603,6 @@ func (s *Server) Open() error {
 	s.syncer.Closing = s.closing
 	s.syncer.Stats = s.holder.Stats.WithTags("component:HolderSyncer")
 
-	// TODO disco
-	if false {
-		node.URI = s.uri
-		node.GRPCURI = s.grpcURI
-
-		// Set metadata for this node.
-		data, err := json.Marshal(node)
-		if err != nil {
-			return errors.Wrap(err, "marshaling json metadata")
-		}
-		if err := s.metadator.SetMetadata(context.Background(), data); err != nil {
-			return errors.Wrap(err, "setting metadata")
-		}
-	}
-
 	err = s.cluster.setup()
 	if err != nil {
 		return errors.Wrap(err, "setting up cluster")
@@ -642,9 +627,6 @@ func (s *Server) Open() error {
 	// bring up the background tasks for the holder.
 	s.holder.SnapshotQueue = s.snapshotQueue
 	s.holder.Activate()
-	if err := s.cluster.setNodeState(nodeStateReady); err != nil {
-		return errors.Wrap(err, "setting nodeState")
-	}
 
 	// Listen for joining nodes.
 	// This needs to start after the Holder has opened so that nodes can join
@@ -788,7 +770,14 @@ func (s *Server) monitorAntiEntropy() {
 			s.holder.Stats.Count(MetricAntiEntropy, 1, 1.0)
 		}
 		t := time.Now()
-		if s.cluster.State() == ClusterStateResizing {
+
+		state, err := s.cluster.State()
+		if err != nil {
+			s.logger.Printf("cluster state error: err=%s", err)
+			continue
+		}
+
+		if state == ClusterStateResizing {
 			continue // don't launch anti-entropy during resize.
 			// the cluster sets its state to resizing and *then* sends to
 			// abortAntiEntropyCh before starting to resize
@@ -1021,8 +1010,14 @@ func (s *Server) node() *topology.Node {
 
 // handleRemoteStatus receives incoming NodeStatus from remote nodes.
 func (s *Server) handleRemoteStatus(pb Message) {
+	state, err := s.cluster.State()
+	if err != nil {
+		s.logger.Printf("getting cluster state: %s", err)
+		return
+	}
+
 	// Ignore NodeStatus messages until the cluster is in a Normal state.
-	if s.cluster.State() != ClusterStateNormal {
+	if state != ClusterStateNormal {
 		return
 	}
 
@@ -1081,7 +1076,7 @@ func (s *Server) monitorDiagnostics() {
 	s.diagnostics.SetVersion(Version)
 	s.diagnostics.Set("Host", s.uri.Host)
 	s.diagnostics.Set("Cluster", strings.Join(s.cluster.nodeIDs(), ","))
-	s.diagnostics.Set("NumNodes", len(s.cluster.nodes))
+	s.diagnostics.Set("NumNodes", len(s.cluster.noder.Nodes()))
 	s.diagnostics.Set("NumCPU", runtime.NumCPU())
 	s.diagnostics.Set("NodeID", s.nodeID)
 	s.diagnostics.Set("ClusterID", s.cluster.id)
diff --git a/test/cluster.go b/test/cluster.go
index 3dcced632..d4e95a967 100644
--- a/test/cluster.go
+++ b/test/cluster.go
@@ -484,9 +484,9 @@ func (c *Cluster) AwaitCoordinatorState(expectedState string, timeout time.Durat
 // in the expected state.
 func (c *Cluster) ExceptionalState(expectedState string) error {
 	for _, node := range c.Nodes {
-		state := node.API.State()
-		if state != expectedState {
-			return fmt.Errorf("node %q: state %s", node.ID(), state)
+		state, err := node.API.State()
+		if err != nil || state != expectedState {
+			return fmt.Errorf("node %q: state %s: err %v", node.ID(), state, err)
 		}
 	}
 	return nil
@@ -534,7 +534,12 @@ func MustNewCluster(tb testing.TB, size int, opts ...[]server.CommandOption) *Cl
 // receives a matching state. It polls up to n times before returning.
 func CheckClusterState(m *Command, state string, n int) bool {
 	for i := 0; i < n; i++ {
-		if m.API.State() == state {
+
+		apiState, err := m.API.State()
+		if err != nil {
+			return false
+		}
+		if apiState == state {
 			return true
 		}
 		time.Sleep(10 * time.Millisecond)
diff --git a/test/pilosa.go b/test/pilosa.go
index 81538b679..1a6635151 100644
--- a/test/pilosa.go
+++ b/test/pilosa.go
@@ -192,7 +192,13 @@ func (m *Command) URL() string { return m.API.Node().URI.String() }
 func (m *Command) ID() string { return m.API.Node().ID }
 
 // IsCoordinator returns true if this is the coordinator.
-func (m *Command) IsCoordinator() bool { return m.API.Node().IsCoordinator }
+func (m *Command) IsCoordinator() bool {
+	coord := m.API.CoordinatorNode()
+	if coord == nil {
+		return false
+	}
+	return coord.ID == m.API.Node().ID
+}
 
 // Client returns a client to connect to the program.
 func (m *Command) Client() *http.InternalClient {
diff --git a/topology/node.go b/topology/node.go
index e5bf0a51e..cb5940983 100644
--- a/topology/node.go
+++ b/topology/node.go
@@ -52,7 +52,7 @@ func (n *Node) Clone() *Node {
 }
 
 func (n *Node) String() string {
-	return fmt.Sprintf("Node:%s:%s:%s", n.URI, n.State, n.ID)
+	return fmt.Sprintf("Node:%s:%s:%s(%v)", n.URI, n.State, n.ID, n.IsCoordinator)
 }
 
 // Nodes represents a list of nodes.
diff --git a/topology/noder.go b/topology/noder.go
index d6dff517a..f0499b997 100644
--- a/topology/noder.go
+++ b/topology/noder.go
@@ -41,6 +41,11 @@ func NewLocalNoder(nodes []*Node) *localNoder {
 	}
 }
 
+// NewEmptyLocalNoder is an empty Noder used for testing.
+func NewEmptyLocalNoder() *localNoder {
+	return &localNoder{}
+}
+
 // Nodes implements the Noder interface.
 func (n *localNoder) Nodes() []*Node {
 	return n.nodes
diff --git a/topology/snapshot.go b/topology/snapshot.go
index da87aa522..decccccfc 100644
--- a/topology/snapshot.go
+++ b/topology/snapshot.go
@@ -139,25 +139,13 @@ func (c *ClusterSnapshot) PartitionNodes(partitionID int) []*Node {
 // field keys. The primary could be any node in the cluster, but we arbitrarily
 // define it to be the node responsible for partition 0.
 func (c *ClusterSnapshot) PrimaryFieldTranslationNode() *Node {
-	// return c.PrimaryPartitionNode(0)
-	for _, n := range c.Nodes {
-		if n.IsCoordinator {
-			return n
-		}
-	}
-	return nil
+	return c.PrimaryPartitionNode(0)
 }
 
 // IsPrimaryFieldTranslationNode returns true if nodeID represents the primary
 // node responsible for field translation.
 func (c *ClusterSnapshot) IsPrimaryFieldTranslationNode(nodeID string) bool {
-	// return c.PrimaryFieldTranslationNode().ID == nodeID
-	for i := range c.Nodes {
-		if c.Nodes[i].ID == nodeID && c.Nodes[i].IsCoordinator {
-			return true
-		}
-	}
-	return false
+	return c.PrimaryFieldTranslationNode().ID == nodeID
 }
 
 // PrimaryPartitionNode returns the primary node of the given partition.
diff --git a/translator_test.go b/translator_test.go
index 7308d7fca..1e2448656 100644
--- a/translator_test.go
+++ b/translator_test.go
@@ -514,10 +514,14 @@ func TestTranslation_Replication(t *testing.T) {
 
 		exp := `{"results":[{"attrs":{},"columns":[],"keys":["x1","x2"]}]}`
 
-		if !test.CheckClusterState(coord, pilosa.ClusterStateNormal, 1000) {
-			t.Fatalf("unexpected coord cluster state: %s, got: %s", pilosa.ClusterStateNormal, coord.API.State())
-		} else if !test.CheckClusterState(other, pilosa.ClusterStateNormal, 1000) {
-			t.Fatalf("unexpected other cluster state: %s, got: %s", pilosa.ClusterStateNormal, other.API.State())
+		coordState, err := coord.API.State()
+		if err != nil || !test.CheckClusterState(coord, pilosa.ClusterStateNormal, 1000) {
+			t.Fatalf("unexpected coord cluster state: %s, got: %s, err: %v", pilosa.ClusterStateNormal, coordState, err)
+		}
+
+		otherState, err := other.API.State()
+		if err != nil || !test.CheckClusterState(other, pilosa.ClusterStateNormal, 1000) {
+			t.Fatalf("unexpected other cluster state: %s, got: %s, err: %v", pilosa.ClusterStateNormal, otherState, err)
 		}
 
 		// Verify the data exists
@@ -528,8 +532,9 @@ func TestTranslation_Replication(t *testing.T) {
 			t.Fatal(err)
 		}
 
-		if !test.CheckClusterState(coord, pilosa.ClusterStateDegraded, 1000) {
-			t.Fatalf("unexpected coord cluster state: %s, got: %s", pilosa.ClusterStateDegraded, coord.API.State())
+		coordState, err = coord.API.State()
+		if err != nil || !test.CheckClusterState(coord, pilosa.ClusterStateDegraded, 1000) {
+			t.Fatalf("unexpected coord cluster state: %s, got: %s", pilosa.ClusterStateDegraded, coordState)
 		}
 
 		// Verify the data exists with one node down
diff --git a/utils_internal_test.go b/utils_internal_test.go
index 8a99a7c7b..534cbbc79 100644
--- a/utils_internal_test.go
+++ b/utils_internal_test.go
@@ -75,14 +75,16 @@ func NewTestCluster(tb testing.TB, n int) *cluster {
 	c.Topology = NewTopology(c.Hasher, c.partitionN, c.ReplicaN, c)
 
 	for i := 0; i < n; i++ {
-		c.nodes = append(c.nodes, &topology.Node{
+		c.noder.AppendNode(&topology.Node{
 			ID:  fmt.Sprintf("node%d", i),
 			URI: NewTestURI("http", fmt.Sprintf("host%d", i), uint16(0)),
 		})
 	}
 
-	c.Node = c.nodes[0]
-	c.Coordinator = c.nodes[0].ID
+	cNodes := c.noder.Nodes()
+
+	c.Node = cNodes[0]
+	c.Coordinator = cNodes[0].ID
 	c.SetState(ClusterStateNormal)
 
 	return c
@@ -231,8 +233,13 @@ func (t *ClusterCluster) addNode() error {
 			return err
 		}
 
+		state, err := coord.State()
+		if err != nil {
+			return err
+		}
+
 		// Wait for the AddNode job to finish.
-		if c.State() != ClusterStateNormal {
+		if state != ClusterStateNormal {
 			t.resizeDone = make(chan struct{})
 			t.mu.Lock()
 			t.resizing = true
@@ -341,9 +348,6 @@ func (t *ClusterCluster) Open() error {
 		if err := c.holder.Open(); err != nil {
 			return err
 		}
-		if err := c.setNodeState(nodeStateReady); err != nil {
-			return err
-		}
 	}
 
 	// Start the listener on the coordinator.
@@ -553,15 +557,17 @@ func NewTestClusterWithReplication(tb testing.TB, nNodes, nReplicas, partitionN
 
 	for i := 0; i < nNodes; i++ {
 		nodeID := fmt.Sprintf("node%d", i)
-		c.nodes = append(c.nodes, &topology.Node{
+		c.noder.AppendNode(&topology.Node{
 			ID:  nodeID,
 			URI: NewTestURI("http", fmt.Sprintf("host%d", i), uint16(0)),
 		})
 		c.Topology.addID(nodeID)
 	}
 
-	c.Node = c.nodes[0]
-	c.Coordinator = c.nodes[0].ID
+	cNodes := c.noder.Nodes()
+
+	c.Node = cNodes[0]
+	c.Coordinator = cNodes[0].ID
 	c.SetState(ClusterStateNormal)
 
 	if err := c.holder.Open(); err != nil {