diff --git a/api.go b/api.go
index 66fdfb8f5..b78eb9760 100644
--- a/api.go
+++ b/api.go
@@ -30,7 +30,6 @@ import (
 	"github.com/molecula/featurebase/v3/pql"
 	"github.com/molecula/featurebase/v3/roaring"
 	"github.com/molecula/featurebase/v3/stats"
-	"github.com/molecula/featurebase/v3/topology"
 	"github.com/molecula/featurebase/v3/tracing"
 	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
@@ -763,7 +762,7 @@ func (api *API) ExportCSV(ctx context.Context, indexName string, fieldName strin
 }
 
 // ShardNodes returns the node and all replicas which should contain a shard's data.
-func (api *API) ShardNodes(ctx context.Context, indexName string, shard uint64) ([]*topology.Node, error) {
+func (api *API) ShardNodes(ctx context.Context, indexName string, shard uint64) ([]*disco.Node, error) {
 	span, _ := tracing.StartSpanFromContext(ctx, "API.ShardNodes")
 	defer span.Finish()
 
@@ -778,7 +777,7 @@ func (api *API) ShardNodes(ctx context.Context, indexName string, shard uint64)
 }
 
 // PartitionNodes returns the node and all replicas which should contain a partition key data.
-func (api *API) PartitionNodes(ctx context.Context, partitionID int) ([]*topology.Node, error) {
+func (api *API) PartitionNodes(ctx context.Context, partitionID int) ([]*disco.Node, error) {
 	span, _ := tracing.StartSpanFromContext(ctx, "API.PartitionNodes")
 	defer span.Finish()
 
@@ -896,7 +895,7 @@ func (api *API) TranslateData(ctx context.Context, indexName string, partition i
 	// Find the node that can service the request.
 	snap := api.cluster.NewSnapshot()
 	nodes := snap.PartitionNodes(partition)
-	var upNode *topology.Node
+	var upNode *disco.Node
 	for _, node := range nodes {
 		// we all UNKNOWN state here because we often mistakenly think
 		// a node is not up under heavy load, but prefer STARTED if we
@@ -963,14 +962,14 @@ func (api *API) FieldTranslateData(ctx context.Context, indexName, fieldName str
 
 // Hosts returns a list of the hosts in the cluster including their ID,
 // URL, and which is the primary.
-func (api *API) Hosts(ctx context.Context) []*topology.Node {
+func (api *API) Hosts(ctx context.Context) []*disco.Node {
 	span, _ := tracing.StartSpanFromContext(ctx, "API.Hosts")
 	defer span.Finish()
 	return api.cluster.Nodes()
 }
 
 // Node gets the ID, URI and primary status for this particular node.
-func (api *API) Node() *topology.Node {
+func (api *API) Node() *disco.Node {
 	return api.server.node()
 }
 
@@ -981,7 +980,7 @@ func (api *API) NodeID() string {
 }
 
 // PrimaryNode returns the primary node for the cluster.
-func (api *API) PrimaryNode() *topology.Node {
+func (api *API) PrimaryNode() *disco.Node {
 	// Create a snapshot of the cluster to use for node/partition calculations.
 	snap := api.cluster.NewSnapshot()
 	return snap.PrimaryFieldTranslationNode()
diff --git a/boltdb/translate_test.go b/boltdb/translate_test.go
index cd74244eb..4c7a79139 100644
--- a/boltdb/translate_test.go
+++ b/boltdb/translate_test.go
@@ -12,9 +12,9 @@ import (
 
 	pilosa "github.com/molecula/featurebase/v3"
 	"github.com/molecula/featurebase/v3/boltdb"
+	"github.com/molecula/featurebase/v3/disco"
 	"github.com/molecula/featurebase/v3/roaring"
 	"github.com/molecula/featurebase/v3/testhook"
-	"github.com/molecula/featurebase/v3/topology"
 )
 
 //var vv = pilosa.VV
@@ -382,7 +382,7 @@ func MustNewTranslateStore(tb testing.TB) *boltdb.TranslateStore {
 		panic(err)
 	}
 
-	s := boltdb.NewTranslateStore("I", "F", 0, topology.DefaultPartitionN, false)
+	s := boltdb.NewTranslateStore("I", "F", 0, disco.DefaultPartitionN, false)
 	s.Path = f.Name()
 	return s
 }
diff --git a/broadcast.go b/broadcast.go
index ecab7b946..c88453b9c 100644
--- a/broadcast.go
+++ b/broadcast.go
@@ -4,7 +4,7 @@ package pilosa
 import (
 	"fmt"
 
-	"github.com/molecula/featurebase/v3/topology"
+	"github.com/molecula/featurebase/v3/disco"
 	"github.com/pkg/errors"
 )
 
@@ -29,7 +29,7 @@ func (*nopSerializer) Unmarshal([]byte, Message) error { return nil }
 type broadcaster interface {
 	SendSync(Message) error
 	SendAsync(Message) error
-	SendTo(*topology.Node, Message) error
+	SendTo(*disco.Node, Message) error
 }
 
 // Message is the interface implemented by all core pilosa types which can be serialized to messages.
@@ -48,7 +48,7 @@ func (nopBroadcaster) SendSync(Message) error { return nil }
 func (nopBroadcaster) SendAsync(Message) error { return nil }
 
 // SendTo is a no-op implementation of Broadcaster SendTo method.
-func (nopBroadcaster) SendTo(*topology.Node, Message) error { return nil }
+func (nopBroadcaster) SendTo(*disco.Node, Message) error { return nil }
 
 // Broadcast message types.
 const (
diff --git a/cluster.go b/cluster.go
index b37d3eca0..7c60d6916 100644
--- a/cluster.go
+++ b/cluster.go
@@ -11,7 +11,6 @@ import (
 	"github.com/molecula/featurebase/v3/ingest"
 	"github.com/molecula/featurebase/v3/logger"
 	"github.com/molecula/featurebase/v3/roaring"
-	"github.com/molecula/featurebase/v3/topology"
 	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
 )
@@ -23,13 +22,13 @@ const (
 
 // cluster represents a collection of nodes.
 type cluster struct { // nolint: maligned
-	noder topology.Noder
+	noder disco.Noder
 
 	id   string
-	Node *topology.Node
+	Node *disco.Node
 
 	// Hashing algorithm used to assign partitions to nodes.
-	Hasher topology.Hasher
+	Hasher disco.Hasher
 
 	// The number of partitions in the cluster.
 	partitionN int
@@ -48,7 +47,6 @@ type cluster struct { // nolint: maligned
 
 	// Distributed Consensus
 	disCo   disco.DisCo
-	stator  disco.Stator
 	sharder disco.Sharder
 
 	holder      *Holder
@@ -78,8 +76,8 @@ type cluster struct { // nolint: maligned
 // newCluster returns a new instance of Cluster with defaults.
 func newCluster() *cluster {
 	return &cluster{
-		Hasher:     &topology.Jmphasher{},
-		partitionN: topology.DefaultPartitionN,
+		Hasher:     &disco.Jmphasher{},
+		partitionN: disco.DefaultPartitionN,
 		ReplicaN:   1,
 
 		closing: make(chan struct{}),
@@ -93,9 +91,8 @@ func newCluster() *cluster {
 		confirmDownRetries: defaultConfirmDownRetries,
 		confirmDownSleep:   defaultConfirmDownSleep,
 
-		disCo:  disco.NopDisCo,
-		noder:  topology.NewEmptyLocalNoder(),
-		stator: disco.NopStator,
+		disCo: disco.NopDisCo,
+		noder: disco.NewEmptyLocalNoder(),
 	}
 }
 
@@ -126,12 +123,12 @@ func (c *cluster) abortAntiEntropy() {
 	}
 }
 
-func (c *cluster) primaryNode() *topology.Node {
+func (c *cluster) primaryNode() *disco.Node {
 	return c.unprotectedPrimaryNode()
 }
 
 // unprotectedPrimaryNode returns the primary node.
-func (c *cluster) unprotectedPrimaryNode() *topology.Node {
+func (c *cluster) unprotectedPrimaryNode() *disco.Node {
 	// Create a snapshot of the cluster to use for node/partition calculations.
 	snap := c.NewSnapshot()
 	return snap.PrimaryFieldTranslationNode()
@@ -161,7 +158,7 @@ func (c *cluster) applySchemaWithNewShards(schema *Schema) error {
 
 // unprotectedStatus returns the the cluster's status including what nodes it contains, its ID, and current state.
 func (c *cluster) unprotectedStatus() (*ClusterStatus, error) {
-	state, err := c.stator.ClusterState(context.Background())
+	state, err := c.noder.ClusterState(context.Background())
 	if err != nil {
 		return nil, err
 	}
@@ -196,21 +193,21 @@ func (c *cluster) remoteSchema() (*Schema, error) {
 
 // nodeIDs returns the list of IDs in the cluster.
 func (c *cluster) nodeIDs() []string {
-	return topology.Nodes(c.Nodes()).IDs()
+	return disco.Nodes(c.Nodes()).IDs()
 }
 
 func (c *cluster) State() (disco.ClusterState, error) {
-	return c.stator.ClusterState(context.Background())
+	return c.noder.ClusterState(context.Background())
 }
 
-func (c *cluster) nodeByID(id string) *topology.Node {
+func (c *cluster) nodeByID(id string) *disco.Node {
 	c.mu.RLock()
 	defer c.mu.RUnlock()
 	return c.unprotectedNodeByID(id)
 }
 
 // unprotectedNodeByID returns a node reference by ID.
-func (c *cluster) unprotectedNodeByID(id string) *topology.Node {
+func (c *cluster) unprotectedNodeByID(id string) *disco.Node {
 	for _, n := range c.noder.Nodes() {
 		if n.ID == id {
 			return n
@@ -231,13 +228,13 @@ func (c *cluster) nodePositionByID(nodeID string) int {
 
 // 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 {
+func (c *cluster) Nodes() []*disco.Node {
 	nodes := c.noder.Nodes()
 	// duplicate the nodes since we're going to be altering them
-	copiedNodes := make([]topology.Node, len(nodes))
-	result := make([]*topology.Node, len(nodes))
+	copiedNodes := make([]disco.Node, len(nodes))
+	result := make([]*disco.Node, len(nodes))
 
-	primary := topology.PrimaryNode(nodes, c.Hasher)
+	primary := disco.PrimaryNode(nodes, c.Hasher)
 
 	// Set node states and IsPrimary.
 	for i, node := range nodes {
@@ -293,13 +290,13 @@ func (c *cluster) close() error {
 
 // PrimaryReplicaNode returns the node listed before the current node in c.Nodes.
 // This is different than "previous node" as the first node always returns nil.
-func (c *cluster) PrimaryReplicaNode() *topology.Node {
+func (c *cluster) PrimaryReplicaNode() *disco.Node {
 	c.mu.RLock()
 	defer c.mu.RUnlock()
 	return c.unprotectedPrimaryReplicaNode()
 }
 
-func (c *cluster) unprotectedPrimaryReplicaNode() *topology.Node {
+func (c *cluster) unprotectedPrimaryReplicaNode() *disco.Node {
 	pos := c.nodePositionByID(c.Node.ID)
 	if pos <= 0 {
 		return nil
@@ -642,7 +639,7 @@ func (c *cluster) findIndexKeys(ctx context.Context, indexName string, keys ...s
 	// TODO: use local replicas to short-circuit network traffic
 
 	// Group keys by node.
-	keysByNode := make(map[*topology.Node][]string)
+	keysByNode := make(map[*disco.Node][]string)
 	for partitionID, keys := range keysByPartition {
 		// Find the primary node for this partition.
 		primary := snap.PrimaryPartitionNode(partitionID)
@@ -752,7 +749,7 @@ func (c *cluster) createIndexKeys(ctx context.Context, indexName string, keys ..
 
 	// Group keys by node.
 	// Delete remote keys from the by-partition map so that it can be used for local translation.
-	keysByNode := make(map[*topology.Node][]string)
+	keysByNode := make(map[*disco.Node][]string)
 	for partitionID, keys := range keysByPartition {
 		// Find the primary node for this partition.
 		primary := snap.PrimaryPartitionNode(partitionID)
@@ -909,8 +906,8 @@ func (c *cluster) translateIndexIDSet(ctx context.Context, indexName string, idS
 	return idMap, nil
 }
 
-func (c *cluster) NewSnapshot() *topology.ClusterSnapshot {
-	return topology.NewClusterSnapshot(c.noder, c.Hasher, c.partitionAssigner, c.ReplicaN)
+func (c *cluster) NewSnapshot() *disco.ClusterSnapshot {
+	return disco.NewClusterSnapshot(c.noder, c.Hasher, c.partitionAssigner, c.ReplicaN)
 }
 
 // ClusterStatus describes the status of the cluster including its
@@ -918,7 +915,7 @@ func (c *cluster) NewSnapshot() *topology.ClusterSnapshot {
 type ClusterStatus struct {
 	ClusterID string
 	State     string
-	Nodes     []*topology.Node
+	Nodes     []*disco.Node
 	Schema    *Schema
 }
 
@@ -997,7 +994,7 @@ type NodeStateMessage struct {
 
 // NodeStatus is an internal message representing the contents of a node.
 type NodeStatus struct {
-	Node    *topology.Node
+	Node    *disco.Node
 	Indexes []*IndexStatus
 	Schema  *Schema
 }
diff --git a/cluster_internal_test.go b/cluster_internal_test.go
index ba70740bc..ccd53026e 100644
--- a/cluster_internal_test.go
+++ b/cluster_internal_test.go
@@ -10,10 +10,10 @@ import (
 	"time"
 
 	"github.com/davecgh/go-spew/spew"
+	"github.com/molecula/featurebase/v3/disco"
 	pnet "github.com/molecula/featurebase/v3/net"
 	"github.com/molecula/featurebase/v3/roaring"
 	"github.com/molecula/featurebase/v3/testhook"
-	"github.com/molecula/featurebase/v3/topology"
 	. "github.com/molecula/featurebase/v3/vprint" // nolint:staticcheck
 )
 
@@ -57,7 +57,7 @@ func newIndexWithTempPath(tb testing.TB, name string) *Index {
 // Ensure the cluster can fairly distribute partitions across the nodes.
 func TestCluster_Owners(t *testing.T) {
 	c := cluster{
-		noder: topology.NewLocalNoder([]*topology.Node{
+		noder: disco.NewLocalNoder([]*disco.Node{
 			{URI: NewTestURIFromHostPort("serverA", 1000)},
 			{URI: NewTestURIFromHostPort("serverB", 1000)},
 			{URI: NewTestURIFromHostPort("serverC", 1000)},
@@ -72,12 +72,12 @@ func TestCluster_Owners(t *testing.T) {
 	snap := c.NewSnapshot()
 
 	// Verify nodes are distributed.
-	if a := snap.PartitionNodes(0); !reflect.DeepEqual(a, []*topology.Node{cNodes[0], cNodes[1]}) {
+	if a := snap.PartitionNodes(0); !reflect.DeepEqual(a, []*disco.Node{cNodes[0], cNodes[1]}) {
 		t.Fatalf("unexpected owners: %s", spew.Sdump(a))
 	}
 
 	// Verify nodes go around the ring.
-	if a := snap.PartitionNodes(2); !reflect.DeepEqual(a, []*topology.Node{cNodes[2], cNodes[0]}) {
+	if a := snap.PartitionNodes(2); !reflect.DeepEqual(a, []*disco.Node{cNodes[2], cNodes[0]}) {
 		t.Fatalf("unexpected owners: %s", spew.Sdump(a))
 	}
 }
@@ -88,7 +88,7 @@ func TestCluster_Partition(t *testing.T) {
 		c := newCluster()
 		c.partitionN = partitionN
 
-		partitionID := topology.ShardToShardPartition(index, shard, partitionN)
+		partitionID := disco.ShardToShardPartition(index, shard, partitionN)
 		if partitionID < 0 || partitionID >= partitionN {
 			t.Errorf("partition out of range: shard=%d, p=%d, n=%d", shard, partitionID, partitionN)
 		}
@@ -118,7 +118,7 @@ func TestHasher(t *testing.T) {
 		{0x0ddc0ffeebadf00d, []int{0, 1, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 15, 15, 15, 15}},
 	} {
 		for i, v := range tt.bucket {
-			hasher := &topology.Jmphasher{}
+			hasher := &disco.Jmphasher{}
 			if got := hasher.Hash(tt.key, i+1); got != v {
 				t.Errorf("hash(%v,%v)=%v, want %v", tt.key, i+1, got, v)
 			}
@@ -150,15 +150,15 @@ func TestCluster_Nodes(t *testing.T) {
 		uris = append(uris, NewTestURIFromHostPort(fmt.Sprintf("node%d", i), uint16(arbitraryPorts[i])))
 	}
 
-	node0 := &topology.Node{ID: "node0", URI: uris[0]}
-	node1 := &topology.Node{ID: "node1", URI: uris[1]}
-	node2 := &topology.Node{ID: "node2", URI: uris[2]}
-	node3 := &topology.Node{ID: "node3", URI: uris[3]}
+	node0 := &disco.Node{ID: "node0", URI: uris[0]}
+	node1 := &disco.Node{ID: "node1", URI: uris[1]}
+	node2 := &disco.Node{ID: "node2", URI: uris[2]}
+	node3 := &disco.Node{ID: "node3", URI: uris[3]}
 
-	nodes := []*topology.Node{node0, node1, node2}
+	nodes := []*disco.Node{node0, node1, node2}
 
 	t.Run("NodeIDs", func(t *testing.T) {
-		actual := topology.Nodes(nodes).IDs()
+		actual := disco.Nodes(nodes).IDs()
 		expected := []string{node0.ID, node1.ID, node2.ID}
 		if !reflect.DeepEqual(actual, expected) {
 			t.Errorf("expected: %v, but got: %v", expected, actual)
@@ -166,7 +166,7 @@ func TestCluster_Nodes(t *testing.T) {
 	})
 
 	t.Run("Filter", func(t *testing.T) {
-		actual := topology.Nodes(topology.Nodes(nodes).Filter(nodes[1])).URIs()
+		actual := disco.Nodes(disco.Nodes(nodes).Filter(nodes[1])).URIs()
 		expected := []pnet.URI{uris[0], uris[2]}
 		if !reflect.DeepEqual(actual, expected) {
 			t.Errorf("expected: %v, but got: %v", expected, actual)
@@ -174,7 +174,7 @@ func TestCluster_Nodes(t *testing.T) {
 	})
 
 	t.Run("FilterURI", func(t *testing.T) {
-		actual := topology.Nodes(topology.Nodes(nodes).FilterURI(uris[1])).URIs()
+		actual := disco.Nodes(disco.Nodes(nodes).FilterURI(uris[1])).URIs()
 		expected := []pnet.URI{uris[0], uris[2]}
 		if !reflect.DeepEqual(actual, expected) {
 			t.Errorf("expected: %v, but got: %v", expected, actual)
@@ -182,8 +182,8 @@ func TestCluster_Nodes(t *testing.T) {
 	})
 
 	t.Run("Contains", func(t *testing.T) {
-		actualTrue := topology.Nodes(nodes).Contains(node1)
-		actualFalse := topology.Nodes(nodes).Contains(node3)
+		actualTrue := disco.Nodes(nodes).Contains(node1)
+		actualFalse := disco.Nodes(nodes).Contains(node3)
 		if !reflect.DeepEqual(actualTrue, true) {
 			t.Errorf("expected: %v, but got: %v", true, actualTrue)
 		}
@@ -193,8 +193,8 @@ func TestCluster_Nodes(t *testing.T) {
 	})
 
 	t.Run("Clone", func(t *testing.T) {
-		clone := topology.Nodes(nodes).Clone()
-		actual := topology.Nodes(clone).URIs()
+		clone := disco.Nodes(nodes).Clone()
+		actual := disco.Nodes(clone).URIs()
 		expected := []pnet.URI{uris[0], uris[1], uris[2]}
 		if !reflect.DeepEqual(actual, expected) {
 			t.Errorf("expected: %v, but got: %v", expected, actual)
diff --git a/ctl/backup.go b/ctl/backup.go
index 6bdd7a83a..920ba9492 100644
--- a/ctl/backup.go
+++ b/ctl/backup.go
@@ -14,9 +14,9 @@ import (
 
 	pilosa "github.com/molecula/featurebase/v3"
 	"github.com/molecula/featurebase/v3/authn"
+	"github.com/molecula/featurebase/v3/disco"
 	"github.com/molecula/featurebase/v3/encoding/proto"
 	"github.com/molecula/featurebase/v3/server"
-	"github.com/molecula/featurebase/v3/topology"
 	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
 )
@@ -300,7 +300,7 @@ func (cmd *BackupCommand) backupShard(ctx context.Context, indexName string, sha
 }
 
 // backupShardNode backs up a single shard from a single index on a specific node.
-func (cmd *BackupCommand) backupShardNode(ctx context.Context, indexName string, shard uint64, node *topology.Node) error {
+func (cmd *BackupCommand) backupShardNode(ctx context.Context, indexName string, shard uint64, node *disco.Node) error {
 	logger := cmd.Logger()
 	logger.Printf("backing up shard: index=%q id=%d", indexName, shard)
 
@@ -334,7 +334,7 @@ func (cmd *BackupCommand) backupShardNode(ctx context.Context, indexName string,
 }
 
 func (cmd *BackupCommand) backupIndexTranslateData(ctx context.Context, name string) error {
-	partitionN := topology.DefaultPartitionN
+	partitionN := disco.DefaultPartitionN
 
 	ch := make(chan int, partitionN)
 	for partitionID := 0; partitionID < partitionN; partitionID++ {
diff --git a/ctl/restore.go b/ctl/restore.go
index 59b7ca73f..8791f7dac 100644
--- a/ctl/restore.go
+++ b/ctl/restore.go
@@ -19,9 +19,9 @@ import (
 
 	pilosa "github.com/molecula/featurebase/v3"
 	"github.com/molecula/featurebase/v3/authn"
+	"github.com/molecula/featurebase/v3/disco"
 	"github.com/molecula/featurebase/v3/logger"
 	"github.com/molecula/featurebase/v3/server"
-	"github.com/molecula/featurebase/v3/topology"
 	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
 )
@@ -103,7 +103,7 @@ func (cmd *RestoreCommand) Run(ctx context.Context) (err error) {
 		return err
 	}
 
-	var primary *topology.Node
+	var primary *disco.Node
 	for _, node := range nodes {
 		if node.IsPrimary {
 			primary = node
@@ -138,7 +138,7 @@ func (cmd *RestoreCommand) Run(ctx context.Context) (err error) {
 	return nil
 }
 
-func (cmd *RestoreCommand) restoreSchema(ctx context.Context, primary *topology.Node) error {
+func (cmd *RestoreCommand) restoreSchema(ctx context.Context, primary *disco.Node) error {
 	f, err := os.Open(filepath.Join(cmd.Path, "schema"))
 	if err != nil {
 		return err
@@ -235,7 +235,7 @@ func (cmd *RestoreCommand) newClient() *retryablehttp.Client {
 	return client
 }
 
-func (cmd *RestoreCommand) restoreIDAlloc(ctx context.Context, primary *topology.Node) error {
+func (cmd *RestoreCommand) restoreIDAlloc(ctx context.Context, primary *disco.Node) error {
 	logger := cmd.Logger()
 
 	f, err := os.Open(filepath.Join(cmd.Path, "idalloc"))
@@ -411,7 +411,7 @@ func (cmd *RestoreCommand) restoreIndexTranslationFile(ctx context.Context, file
 	return nil
 }
 
-func (cmd *RestoreCommand) restoreFieldTranslation(ctx context.Context, nodes []*topology.Node) error {
+func (cmd *RestoreCommand) restoreFieldTranslation(ctx context.Context, nodes []*disco.Node) error {
 	filenames, err := filepath.Glob(filepath.Join(cmd.Path, "indexes", "*", "fields", "*", "translate"))
 	if err != nil {
 		return err
@@ -443,7 +443,7 @@ func (cmd *RestoreCommand) restoreFieldTranslation(ctx context.Context, nodes []
 	return g.Wait()
 }
 
-func (cmd *RestoreCommand) restoreFieldTranslationFile(ctx context.Context, nodes []*topology.Node, filename string) error {
+func (cmd *RestoreCommand) restoreFieldTranslationFile(ctx context.Context, nodes []*disco.Node, filename string) error {
 	logger := cmd.Logger()
 
 	rel, err := filepath.Rel(cmd.Path, filename)
diff --git a/disco/disco.go b/disco/disco.go
index 585c7a43a..e653c8f02 100644
--- a/disco/disco.go
+++ b/disco/disco.go
@@ -68,21 +68,6 @@ const (
 	NodeStateStarted  NodeState = "STARTED"
 )
 
-type Stator interface {
-	// Started will mark the actual node as already started.
-	// It must be called after all initialization processes
-	// are up and running.
-	Started(ctx context.Context) error
-
-	// ClusterState considers the state of all nodes and gives
-	// a general cluster state. The output calculation is as follows:
-	// - If any of the nodes are still starting: "STARTING"
-	// - If all nodes are up and running: "NORMAL"
-	// - If number of DOWN nodes is lower than number of replicas: "DEGRADED"
-	// - If number of unresponsive nodes is greater than (or equal to) the number of replicas: "DOWN"
-	ClusterState(context.Context) (ClusterState, error)
-}
-
 // Schema is a map of all indexes, each of those being a map of fields, then
 // views.
 type Schema map[string]*Index
@@ -123,13 +108,6 @@ type Schemator interface {
 	DeleteView(ctx context.Context, index, field, view string) error
 }
 
-// Metadator is in charge of storing specific metadata per node.
-// This metadata can be retrieved by any node using the specific peerID.
-type Metadator interface {
-	Metadata(ctx context.Context, peerID string) ([]byte, error)
-	SetMetadata(ctx context.Context, metadata []byte) error
-}
-
 // Sharder is an interface used to maintain the set of availableShards bitmaps
 // per field.
 type Sharder interface {
@@ -177,32 +155,6 @@ func (n *nopDisCo) DeleteNode(context.Context, string) error {
 	return nil
 }
 
-// NopStator represents a Stator that doesn't do anything.
-var NopStator Stator = &nopStator{}
-
-type nopStator struct{}
-
-// ClusterState is a no-op implementation of the Stator ClusterState method.
-func (n *nopStator) ClusterState(context.Context) (ClusterState, error) {
-	return ClusterStateUnknown, nil
-}
-
-func (n *nopStator) Started(ctx context.Context) error {
-	return nil
-}
-
-// NopMetadator represents a Metadator that doesn't do anything.
-var NopMetadator Metadator = &nopMetadator{}
-
-type nopMetadator struct{}
-
-func (*nopMetadator) Metadata(context.Context, string) ([]byte, error) {
-	return nil, nil
-}
-func (*nopMetadator) SetMetadata(context.Context, []byte) error {
-	return nil
-}
-
 // NopSharder represents a Sharder that doesn't do anything.
 var NopSharder Sharder = &nopSharder{}
 
diff --git a/topology/hasher.go b/disco/hasher.go
similarity index 98%
rename from topology/hasher.go
rename to disco/hasher.go
index 4f718c6bb..59eb26ba9 100644
--- a/topology/hasher.go
+++ b/disco/hasher.go
@@ -1,5 +1,5 @@
 // Copyright 2021 Molecula Corp. All rights reserved.
-package topology
+package disco
 
 // Hasher represents an interface to hash integers into buckets.
 type Hasher interface {
diff --git a/topology/node.go b/disco/node.go
similarity index 90%
rename from topology/node.go
rename to disco/node.go
index e5c33df5f..e52b00f4d 100644
--- a/topology/node.go
+++ b/disco/node.go
@@ -1,20 +1,19 @@
 // Copyright 2021 Molecula Corp. All rights reserved.
-package topology
+package disco
 
 import (
 	"fmt"
 
-	"github.com/molecula/featurebase/v3/disco"
 	"github.com/molecula/featurebase/v3/net"
 )
 
 // Node represents a node in the cluster.
 type Node struct {
-	ID        string          `json:"id"`
-	URI       net.URI         `json:"uri"`
-	GRPCURI   net.URI         `json:"grpc-uri"`
-	IsPrimary bool            `json:"isPrimary"`
-	State     disco.NodeState `json:"state"`
+	ID        string    `json:"id"`
+	URI       net.URI   `json:"uri"`
+	GRPCURI   net.URI   `json:"grpc-uri"`
+	IsPrimary bool      `json:"isPrimary"`
+	State     NodeState `json:"state"`
 }
 
 func (n *Node) Clone() *Node {
diff --git a/topology/noder.go b/disco/noder.go
similarity index 59%
rename from topology/noder.go
rename to disco/noder.go
index b1c3bccd4..2fb8b6052 100644
--- a/topology/noder.go
+++ b/disco/noder.go
@@ -1,7 +1,8 @@
 // Copyright 2021 Molecula Corp. All rights reserved.
-package topology
+package disco
 
 import (
+	"context"
 	"sort"
 )
 
@@ -10,6 +11,20 @@ import (
 type Noder interface {
 	Nodes() []*Node // Remember: this has to be sorted correctly!!
 	PrimaryNodeID(hasher Hasher) string
+
+	// SetMetadata records the local node's metadata.
+	SetMetadata(ctx context.Context, node *Node) error
+
+	// SetState changes a node to a given state.
+	SetState(ctx context.Context, state NodeState) error
+
+	// ClusterState considers the state of all nodes and gives
+	// a general cluster state. The output calculation is as follows:
+	// - If any of the nodes are still starting: "STARTING"
+	// - If all nodes are up and running: "NORMAL"
+	// - If number of DOWN nodes is lower than number of replicas: "DEGRADED"
+	// - If number of unresponsive nodes is greater than (or equal to) the number of replicas: "DOWN"
+	ClusterState(context.Context) (ClusterState, error)
 }
 
 // localNoder is a simple implementation of the Noder interface
@@ -64,3 +79,17 @@ func (n *localNoder) PrimaryNodeID(hasher Hasher) string {
 	}
 	return primaryNode.ID
 }
+
+// ClusterState is a no-op implementation of the Stator ClusterState method.
+func (n *localNoder) ClusterState(context.Context) (ClusterState, error) {
+	return ClusterStateUnknown, nil
+}
+
+func (n *localNoder) SetState(ctx context.Context, state NodeState) error {
+	return nil
+}
+
+// localNoder doesn't really implement metadata support.
+func (*localNoder) SetMetadata(context.Context, *Node) error {
+	return nil
+}
diff --git a/topology/snapshot.go b/disco/snapshot.go
similarity index 99%
rename from topology/snapshot.go
rename to disco/snapshot.go
index 313a11882..cb92e4fe6 100644
--- a/topology/snapshot.go
+++ b/disco/snapshot.go
@@ -1,5 +1,5 @@
 // Copyright 2021 Molecula Corp. All rights reserved.
-package topology
+package disco
 
 import (
 	"encoding/binary"
diff --git a/encoding/proto/proto.go b/encoding/proto/proto.go
index 8c165da1c..0f96c0457 100644
--- a/encoding/proto/proto.go
+++ b/encoding/proto/proto.go
@@ -14,7 +14,6 @@ import (
 	"github.com/molecula/featurebase/v3/pb"
 	"github.com/molecula/featurebase/v3/pql"
 	"github.com/molecula/featurebase/v3/roaring"
-	"github.com/molecula/featurebase/v3/topology"
 	"github.com/pkg/errors"
 )
 
@@ -159,7 +158,7 @@ func (s Serializer) Unmarshal(buf []byte, m pilosa.Message) error {
 		}
 		s.decodeNodeStatus(msg, mt)
 		return nil
-	case *topology.Node:
+	case *disco.Node:
 		msg := &pb.Node{}
 		err := proto.Unmarshal(buf, msg)
 		if err != nil {
@@ -345,7 +344,7 @@ func (s Serializer) encodeToProto(m pilosa.Message) proto.Message {
 		return s.encodeNodeEventMessage(mt)
 	case *pilosa.NodeStatus:
 		return s.encodeNodeStatus(mt)
-	case *topology.Node:
+	case *disco.Node:
 		return s.encodeNode(mt)
 	case *pilosa.QueryRequest:
 		return s.encodeQueryRequest(mt)
@@ -633,7 +632,7 @@ func (s Serializer) encodeFieldOptions(o *pilosa.FieldOptions) *pb.FieldOptions
 }
 
 // s.encodeNodes converts a slice of Nodes into its pb.representation.
-func (s Serializer) encodeNodes(a []*topology.Node) []*pb.Node {
+func (s Serializer) encodeNodes(a []*disco.Node) []*pb.Node {
 	other := make([]*pb.Node, len(a))
 	for i := range a {
 		other[i] = s.encodeNode(a[i])
@@ -642,7 +641,7 @@ func (s Serializer) encodeNodes(a []*topology.Node) []*pb.Node {
 }
 
 // s.encodeNode converts a Node into its pb.representation.
-func (s Serializer) encodeNode(m *topology.Node) *pb.Node {
+func (s Serializer) encodeNode(m *disco.Node) *pb.Node {
 	n := m.Clone()
 	return &pb.Node{
 		ID:      n.ID,
@@ -1019,9 +1018,9 @@ func (s Serializer) decodeDecimal(d *pb.Decimal, m *pql.Decimal) {
 	m.Scale = d.Scale
 }
 
-func (s Serializer) decodeNodes(a []*pb.Node, m []*topology.Node) {
+func (s Serializer) decodeNodes(a []*pb.Node, m []*disco.Node) {
 	for i := range a {
-		m[i] = &topology.Node{}
+		m[i] = &disco.Node{}
 		s.decodeNode(a[i], m[i])
 	}
 }
@@ -1029,13 +1028,13 @@ func (s Serializer) decodeNodes(a []*pb.Node, m []*topology.Node) {
 func (s Serializer) decodeClusterStatus(cs *pb.ClusterStatus, m *pilosa.ClusterStatus) {
 	m.State = cs.State
 	m.ClusterID = cs.ClusterID
-	m.Nodes = make([]*topology.Node, len(cs.Nodes))
+	m.Nodes = make([]*disco.Node, len(cs.Nodes))
 	s.decodeNodes(cs.Nodes, m.Nodes)
 	m.Schema = &pilosa.Schema{}
 	s.decodeSchema(cs.Schema, m.Schema)
 }
 
-func (s Serializer) decodeNode(node *pb.Node, m *topology.Node) {
+func (s Serializer) decodeNode(node *pb.Node, m *disco.Node) {
 	m.ID = node.ID
 	s.decodeURI(node.URI, &m.URI)
 	s.decodeURI(node.GRPCURI, &m.GRPCURI)
@@ -1120,12 +1119,12 @@ func (s Serializer) decodeNodeStateMessage(pb *pb.NodeStateMessage, m *pilosa.No
 
 func (s Serializer) decodeNodeEventMessage(pb *pb.NodeEventMessage, m *pilosa.NodeEvent) {
 	m.Event = pilosa.NodeEventType(pb.Event)
-	m.Node = &topology.Node{}
+	m.Node = &disco.Node{}
 	s.decodeNode(pb.Node, m.Node)
 }
 
 func (s Serializer) decodeNodeStatus(pb *pb.NodeStatus, m *pilosa.NodeStatus) {
-	m.Node = &topology.Node{}
+	m.Node = &disco.Node{}
 	m.Indexes = s.decodeIndexStatuses(pb.Indexes)
 	m.Schema = &pilosa.Schema{}
 	s.decodeSchema(pb.Schema, m.Schema)
diff --git a/etcd/config_gen.go b/etcd/config_gen.go
new file mode 100644
index 000000000..6f2c01c03
--- /dev/null
+++ b/etcd/config_gen.go
@@ -0,0 +1,71 @@
+package etcd
+
+import (
+	"fmt"
+	"os"
+	"strings"
+	"sync/atomic"
+)
+
+// Helper utilities to allow us to create etcd cluster configs
+// that can clean up after themselves.
+
+// DirCleaner represents the subset of the testing.TB interface
+// we care about, allowing us to take objects which behave like
+// that without importing all of testing to get them.
+type DirCleaner interface {
+	Cleanup(func())
+	Fatalf(string, ...interface{})
+	Logf(string, ...interface{})
+	Name() string
+	TempDir() string
+}
+
+var unixSocketCounter uint64
+
+// unixSocket returns a url for use in test etcd clusters, referring
+// to a file which will be cleaned up after a test completes.
+func unixSocket(dc DirCleaner) string {
+	count := atomic.AddUint64(&unixSocketCounter, 1)
+	addr := fmt.Sprintf("fake:%d", count)
+	dc.Cleanup(func() {
+		err := os.Remove(addr)
+
+		if err != nil && !os.IsNotExist(err) { // not an error if the socket is not present
+			dc.Logf("could not remove '%s', %v", addr, err)
+		}
+	})
+	return fmt.Sprintf("unix://%s", addr)
+}
+
+// GenEtcdConfigs generates a set of etcd configs, and an initial cluster
+// URL. It also makes temporary etcd dirs. Everything it creates
+// is cleaned up by `dc.Cleanup` after the test/benchmark completes. On
+// error, it calls `dc.Fatalf`, presumably terminating the test.
+func GenEtcdConfigs(dc DirCleaner, n int) (clusterName string, cfgs []Options) {
+	allPeerURLs := make([]string, n)
+	cfgs = make([]Options, n)
+	clusterName = fmt.Sprintf("cluster-%s", dc.Name())
+	for i := 0; i < n; i++ {
+		name := fmt.Sprintf("server%d", i)
+		discoDir := dc.TempDir()
+		clientURL := unixSocket(dc)
+		peerURL := unixSocket(dc)
+		cfgs[i] = Options{
+			Name:          name,
+			Dir:           discoDir,
+			LClientURL:    clientURL,
+			AClientURL:    clientURL,
+			LPeerURL:      peerURL,
+			APeerURL:      peerURL,
+			HeartbeatTTL:  60,
+			UnsafeNoFsync: true,
+		}
+		allPeerURLs[i] = fmt.Sprintf("%s=%s", name, peerURL)
+	}
+	peerURLs := strings.Join(allPeerURLs, ",")
+	for i := range cfgs {
+		cfgs[i].InitCluster = peerURLs
+	}
+	return clusterName, cfgs
+}
diff --git a/etcd/embed.go b/etcd/embed.go
index 7d5a0db11..6b5f5fd3d 100644
--- a/etcd/embed.go
+++ b/etcd/embed.go
@@ -16,7 +16,6 @@ import (
 	"github.com/molecula/featurebase/v3/disco"
 	"github.com/molecula/featurebase/v3/logger"
 	"github.com/molecula/featurebase/v3/monitor"
-	"github.com/molecula/featurebase/v3/topology"
 	"github.com/pkg/errors"
 	"go.etcd.io/etcd/api/v3/mvccpb"
 	"go.etcd.io/etcd/client/pkg/v3/transport"
@@ -55,8 +54,7 @@ type Options struct {
 var (
 	_ disco.DisCo     = &Etcd{}
 	_ disco.Schemator = &Etcd{}
-	_ disco.Stator    = &Etcd{}
-	_ disco.Metadator = &Etcd{}
+	_ disco.Noder     = &Etcd{}
 	_ disco.Sharder   = &Etcd{}
 )
 
@@ -71,7 +69,6 @@ const (
 )
 
 var (
-	etcdLeaderChanged   = etcdserver.ErrLeaderChanged.Error()
 	errEtcdShuttingDown = errors.New("etcd shutting down")
 )
 
@@ -79,7 +76,7 @@ var (
 // nodes in etcd, which we update from data collected either directly
 // from the KV, or via heartbeats.
 //
-// Any change to the topology.Node should create a new Node rather
+// Any change to the disco.Node should create a new Node rather
 // than reusing the old one, so we can return the structure and not worry
 // about data races.
 //
@@ -88,16 +85,9 @@ var (
 // but still handle cases where we get updates to several fields that
 // reach us out of order.
 type nodeData struct {
-	heartbeatState string
-	metadata       []byte
-	topologyNode   *topology.Node
-}
-
-func (n *nodeData) computedState() disco.NodeState {
-	if n.heartbeatState != "" {
-		return disco.NodeState(n.heartbeatState)
-	}
-	return disco.NodeStateUnknown
+	heartbeat disco.NodeState
+	metadata  []byte
+	node      *disco.Node
 }
 
 type Etcd struct {
@@ -110,20 +100,20 @@ type Etcd struct {
 
 	heartbeatLeasedKV *leasedKV
 
-	// We have a watcher running. watchCancel() cancels its context.
-	watchCancel func()
-	closeWatch  chan struct{}
+	// A context for our children (node watcher, lease keepalive)
+	childContext context.Context
+	// function to cancel the child contexts when we're done
+	childCancel func()
 
 	// knownNodes and sortedNodes get updated by data coming in from
-	// watchers. Any change to the contents of a *topology.Node here
+	// watchers. Any change to the contents of a *disco.Node here
 	// should be implemented by making a new one and replacing the pointer,
 	// so the old pointer stays valid and can be used.
-	nodeMu          sync.Mutex
-	nodeRev         int64
-	knownNodes      map[string]*nodeData
-	sortedNodes     []*topology.Node
-	nodeStates      map[string]disco.NodeState
-	nodeStatesDirty bool // do we need to remake the nodeStates map to use it?
+	nodeMu      sync.Mutex
+	nodeRev     int64
+	knownNodes  map[string]*nodeData
+	sortedNodes []*disco.Node // immutable nodes kept in sorted order
+	nodesDirty  bool          // do we need to recompute sortedNodes?
 
 	version string
 
@@ -137,7 +127,6 @@ func NewEtcd(opt Options, logger logger.Logger, replicas int, version string) *E
 		logger:     logger,
 		replicas:   replicas,
 		knownNodes: make(map[string]*nodeData),
-		nodeStates: make(map[string]disco.NodeState),
 		version:    version,
 	}
 
@@ -149,21 +138,25 @@ func NewEtcd(opt Options, logger logger.Logger, replicas int, version string) *E
 
 // Close implements io.Closer
 func (e *Etcd) Close() error {
-	if e.closeWatch != nil {
-		close(e.closeWatch)
+	// tell the heartbeat to stop. we do this before canceling
+	// the context because we want the heartbeat to get a chance
+	// to notify other nodes that it's down.
+	if e.heartbeatLeasedKV != nil {
+		e.heartbeatLeasedKV.Stop()
 	}
-	if e.watchCancel != nil {
-		e.watchCancel()
+	// cancel the contexts that heartbeat and watcher are using.
+	if e.childCancel != nil {
+		e.childCancel()
 	}
+	// shut down the server, if we have one.
 	if e.e != nil {
-		if e.heartbeatLeasedKV != nil {
-			e.heartbeatLeasedKV.Stop()
-		}
-
 		e.e.Close()
 		<-e.e.Server.StopNotify()
 	}
-
+	// shut down the client, if we have one. if something's still
+	// using it, we anticipate the client failing its current call,
+	// and any retry will be checking for the child context being
+	// cancelled, first, we hope.
 	if e.cli != nil {
 		e.cli.Close()
 	}
@@ -171,17 +164,6 @@ func (e *Etcd) Close() error {
 	return nil
 }
 
-// retryClient attempts to do a thing, but also tries to handle the
-// specific case where the client fails because of a leader election,
-// in which case we need to restart the client and retry the thing.
-//
-// We have to let go of the lock while calling `fn` because some fn are
-// long-lasting ones, like watchNodesOnce. So we grab a local copy of
-// the client object, then call things on that object. This should error
-// out sanely instead of panicing if we close the client while something
-// is running on it.
-//
-// New feature: retryClient can also retry on errTimeout.
 const etcdRetryTimes = 3
 
 // newClient requests a new client which is different from the one
@@ -200,6 +182,17 @@ func (e *Etcd) newClient(cli *clientv3.Client) *clientv3.Client {
 	return e.cli
 }
 
+// retryClient attempts to do a thing, but also tries to handle the
+// specific case where the client fails because of a leader election,
+// in which case we need to restart the client and retry the thing.
+//
+// We have to let go of the lock while calling `fn` because some fn are
+// long-lasting ones, like watchNodesOnce. So we grab a local copy of
+// the client object, then call things on that object. This should error
+// out sanely instead of panicing if we close the client while something
+// is running on it.
+//
+// New feature: retryClient can also retry on errTimeout.
 func (e *Etcd) retryClient(fn func(cli *clientv3.Client) error) (err error) {
 	e.cliMu.Lock()
 	cli := e.cli
@@ -210,7 +203,6 @@ func (e *Etcd) retryClient(fn func(cli *clientv3.Client) error) (err error) {
 		switch err {
 		case etcdserver.ErrLeaderChanged:
 			cli = e.newClient(cli)
-			break
 		case nil:
 			return nil
 		default:
@@ -242,7 +234,6 @@ func (e *Etcd) retryClient(fn func(cli *clientv3.Client) error) (err error) {
 			// timeout to give us a reasonable backoff period and keep us
 			// from spamming these.
 			time.Sleep(100 * time.Millisecond)
-			break
 		}
 	}
 	// if we got here, we got a total of three of some combination of
@@ -257,16 +248,29 @@ func (e *Etcd) parseOptions() (*embed.Config, error) {
 	cfg.Name = e.options.Name
 	cfg.Dir = e.options.Dir
 	cfg.InitialClusterToken = e.options.ClusterName
-	cfg.LCUrls = types.MustNewURLs([]string{e.options.LClientURL})
+	var err error
+	cfg.LCUrls, err = types.NewURLs([]string{e.options.LClientURL})
+	if err != nil {
+		return nil, fmt.Errorf("parsing listen client URL %q: %v", e.options.LClientURL, err)
+	}
 	cfg.UnsafeNoFsync = e.options.UnsafeNoFsync
 	if e.options.AClientURL != "" {
-		cfg.ACUrls = types.MustNewURLs([]string{e.options.AClientURL})
+		cfg.ACUrls, err = types.NewURLs([]string{e.options.AClientURL})
+		if err != nil {
+			return nil, fmt.Errorf("parsing advertise client URL %q: %v", e.options.AClientURL, err)
+		}
 	} else {
 		cfg.ACUrls = cfg.LCUrls
 	}
-	cfg.LPUrls = types.MustNewURLs([]string{e.options.LPeerURL})
+	cfg.LPUrls, err = types.NewURLs([]string{e.options.LPeerURL})
+	if err != nil {
+		return nil, fmt.Errorf("parsing listen peer URL %q: %v", e.options.LPeerURL, err)
+	}
 	if e.options.APeerURL != "" {
-		cfg.APUrls = types.MustNewURLs([]string{e.options.APeerURL})
+		cfg.APUrls, err = types.NewURLs([]string{e.options.APeerURL})
+		if err != nil {
+			return nil, fmt.Errorf("parsing advertise peer URL %q: %v", e.options.APeerURL, err)
+		}
 	} else {
 		cfg.APUrls = cfg.LPUrls
 	}
@@ -306,7 +310,9 @@ func (e *Etcd) parseOptions() (*embed.Config, error) {
 		KeyFile:       e.options.PeerKeyFile,
 	}
 
-	return cfg, nil
+	// We might get an error from Validate. etcd docs don't tell us what
+	// that error might be, though!
+	return cfg, cfg.Validate()
 }
 
 // Start starts etcd and hearbeat
@@ -316,6 +322,8 @@ func (e *Etcd) Start(ctx context.Context) (_ disco.InitialClusterState, err erro
 		return disco.InitialClusterStateNew, err
 	}
 	state := disco.InitialClusterState(opts.ClusterState)
+	// create a context that can be used for our watch processes, etcetera.
+	e.childContext, e.childCancel = context.WithCancel(context.Background())
 
 	e.e, err = embed.StartEtcd(opts)
 	if err != nil {
@@ -325,7 +333,8 @@ func (e *Etcd) Start(ctx context.Context) (_ disco.InitialClusterState, err erro
 	// later, so we have to stop the server ourselves.
 	defer func() {
 		if err != nil {
-			e.e.Server.Stop()
+			// shut down everything on our way out.
+			e.Close()
 		}
 	}()
 	e.cli = v3client.New(e.e.Server)
@@ -345,15 +354,9 @@ func (e *Etcd) Start(ctx context.Context) (_ disco.InitialClusterState, err erro
 		// heard from yet.
 		for _, member := range members {
 			peerID := member.ID.String()
-			e.knownNodes[peerID] = &nodeData{
-				topologyNode: &topology.Node{
-					ID:    peerID,
-					State: disco.NodeStateUnknown,
-				},
-			}
-			e.nodeStates[peerID] = disco.NodeStateUnknown
+			_ = e.seeNode(peerID)
 		}
-		e.nodeStatesDirty = true
+		e.nodesDirty = true
 		return state, e.startHeartbeatAndWatcher(ctx)
 	}
 }
@@ -362,22 +365,21 @@ func (e *Etcd) Start(ctx context.Context) (_ disco.InitialClusterState, err erro
 // watcher that watches for changes to events we care about.
 func (e *Etcd) startHeartbeatAndWatcher(ctx context.Context) error {
 	key := heartbeatPrefix + e.e.Server.ID().String()
-	e.heartbeatLeasedKV = newLeasedKV(e, key, e.options.HeartbeatTTL)
-	e.closeWatch = make(chan struct{})
+	e.heartbeatLeasedKV = newLeasedKV(e, e.childContext, key, e.options.HeartbeatTTL)
 
 	if err := e.heartbeatLeasedKV.Start(string(disco.NodeStateStarting)); err != nil {
 		return errors.Wrap(err, "startHeartbeat: starting a new heartbeat")
 	}
-	// WatchNodes does not check for an error, and will need to be shut
+	// watchNodes does not check for an error, and will need to be shut
 	// down later. We only get this far at a point where we're returning
 	// a nil error, and thus, the caller is expected to cleanly shut down
 	// the server later.
-	go e.WatchNodes()
+	go e.watchNodes()
 	return nil
 }
 
-func (e *Etcd) Started(ctx context.Context) (err error) {
-	return e.heartbeatLeasedKV.Set(ctx, string(disco.NodeStateStarted))
+func (e *Etcd) SetState(ctx context.Context, state disco.NodeState) (err error) {
+	return e.heartbeatLeasedKV.Set(ctx, string(state))
 }
 
 func (e *Etcd) ID() string {
@@ -423,15 +425,14 @@ func (e *Etcd) ClusterState(ctx context.Context) (out disco.ClusterState, err er
 		starting   bool
 	)
 	e.nodeMu.Lock()
-	err = e.populateNodeStates(ctx)
-	states := e.nodeStates
+	nodes := e.populateNodeStates(ctx)
 	e.nodeMu.Unlock()
 	if err != nil {
-		e.logger.Printf("ClusterState %q: getting node states: %v", e.options.Name, states)
+		e.logger.Errorf("requesting cluster state %q: getting node states: %v", e.options.Name, err)
 		return disco.ClusterStateUnknown, err
 	}
-	for _, state := range states {
-		switch state {
+	for _, node := range nodes {
+		switch node.State {
 		case disco.NodeStateStarting:
 			starting = true
 		case disco.NodeStateUnknown:
@@ -445,8 +446,8 @@ func (e *Etcd) ClusterState(ctx context.Context) (out disco.ClusterState, err er
 		return disco.ClusterStateStarting, nil
 	}
 
-	if heartbeats < len(states) {
-		if len(states)-heartbeats >= e.replicas {
+	if heartbeats < len(e.knownNodes) {
+		if len(e.knownNodes)-heartbeats >= e.replicas {
 			return disco.ClusterStateDown, nil
 		}
 
@@ -469,6 +470,24 @@ func parseNodeKey(key []byte) (prefix string, peerID string, err error) {
 	return string(key[:peerIndex+1]), string(key[peerIndex+1:]), nil
 }
 
+// seeNode encapsulates the practice of creating a new node, when needed,
+// and checking for duplicates.
+func (e *Etcd) seeNode(peerID string) *nodeData {
+	var node *nodeData
+	if node = e.knownNodes[peerID]; node == nil {
+		e.logger.Debugf("previously unseen node, peer ID %s", peerID)
+		node = &nodeData{
+			node: &disco.Node{
+				ID:    peerID,
+				State: disco.NodeStateUnknown,
+			},
+			heartbeat: disco.NodeStateUnknown,
+		}
+		e.knownNodes[peerID] = node
+	}
+	return node
+}
+
 // deleteNodeData is like putNodeData, but handles deletes rather than cases
 // where a value exists. you should call it with the node mutex locked.
 func (e *Etcd) deleteNodeData(key []byte, revision int64) error {
@@ -481,18 +500,14 @@ func (e *Etcd) deleteNodeData(key []byte, revision int64) error {
 	}
 	switch prefix {
 	case heartbeatPrefix:
-		if e.knownNodes[peerID] == nil {
-			e.knownNodes[peerID] = &nodeData{}
-		}
-		e.knownNodes[peerID].heartbeatState = ""
-		e.nodeStatesDirty = true
+		node := e.seeNode(peerID)
+		// mark state as unknown because we deleted the heartbeat.
+		node.heartbeat = disco.NodeStateUnknown
+		e.nodesDirty = true
 	case metadataPrefix:
-		if e.knownNodes[peerID] == nil {
-			e.knownNodes[peerID] = &nodeData{}
-		}
-		e.knownNodes[peerID].metadata = nil
-		e.knownNodes[peerID].topologyNode = &topology.Node{}
-		e.nodeStatesDirty = true
+		e.logger.Infof("deleting a previously-seen node, peer ID %q", peerID)
+		delete(e.knownNodes, peerID)
+		e.nodesDirty = true
 	default:
 		return fmt.Errorf("node watch: invalid prefix %q", prefix)
 	}
@@ -512,25 +527,21 @@ func (e *Etcd) putNodeData(key []byte, value []byte, revision int64) (err error)
 	}
 	switch prefix {
 	case heartbeatPrefix:
-		if e.knownNodes[peerID] == nil {
-			e.knownNodes[peerID] = &nodeData{}
-		}
-		e.knownNodes[peerID].heartbeatState = string(value)
-		e.nodeStatesDirty = true
+		node := e.seeNode(peerID)
+		node.heartbeat = disco.NodeState(value)
+		e.nodesDirty = true
 	case metadataPrefix:
-		if e.knownNodes[peerID] == nil {
-			e.knownNodes[peerID] = &nodeData{}
-		}
-		e.knownNodes[peerID].metadata = value
-		var newNode topology.Node
+		node := e.seeNode(peerID)
+		node.metadata = value
+		var newNode disco.Node
 		err := json.Unmarshal(value, &newNode)
 		if err != nil {
 			return fmt.Errorf("json unmarshal of node metadata: %v", err)
 		}
-		e.knownNodes[peerID].topologyNode = &newNode
-		// This saves us one remake of the node later, probably.
-		e.knownNodes[peerID].topologyNode.State = e.knownNodes[peerID].computedState()
-		e.nodeStatesDirty = true
+		// start with the heartbeat state
+		newNode.State = node.heartbeat
+		node.node = &newNode
+		e.nodesDirty = true
 	default:
 		return fmt.Errorf("node watch: invalid prefix %q", prefix)
 	}
@@ -540,51 +551,70 @@ func (e *Etcd) putNodeData(key []byte, value []byte, revision int64) (err error)
 // compute the states of all the nodes. we compute all of them because
 // we might have returned the old map in response to a query, so we want to
 // make a new one. You should have the node state lock held when you call this.
-func (e *Etcd) populateNodeStates(ctx context.Context) error {
-	if !e.nodeStatesDirty {
-		return nil
+// Returns an immutable sorted list of nodes; future updates will not
+// modify the slice or the nodes in it.
+func (e *Etcd) populateNodeStates(ctx context.Context) []*disco.Node {
+	if !e.nodesDirty {
+		return e.sortedNodes
 	}
-	e.nodeStates = make(map[string]disco.NodeState, len(e.knownNodes))
-	e.sortedNodes = make([]*topology.Node, 0, len(e.knownNodes))
-	for peerID, data := range e.knownNodes {
-		newState := data.computedState()
-		e.nodeStates[peerID] = newState
+	e.sortedNodes = make([]*disco.Node, 0, len(e.knownNodes))
+	for _, data := range e.knownNodes {
+		newState := data.heartbeat
 		// update the state with the current state, so we can
 		// reuse these nodes later. sortedNodes may end up shorter
 		// than the whole node list if we don't have all the nodes
 		// yet!
-		if data.topologyNode != nil {
-			if data.topologyNode.State != newState {
-				newNode := *data.topologyNode
+		if data.node != nil {
+			// The only part that should ever change is the state, which
+			// will be either "unknown" or the state from a heartbeat.
+			// If that computed state is different, we make a new node
+			// at this point. The reason is that, if we previously returned
+			// the sorted list of nodes, someone else could have a
+			// pointer to the existing node. We don't want to clone these
+			// every time anyone reads them, so instead we make them
+			// immutable and copy-on-write.
+			if data.node.State != newState {
+				newNode := *data.node
 				newNode.State = newState
-				data.topologyNode = &newNode
+				data.node = &newNode
 			}
-			e.sortedNodes = append(e.sortedNodes, data.topologyNode)
+			e.sortedNodes = append(e.sortedNodes, data.node)
 		}
 	}
 	// sort list by ID. list now contains sorted nodes which have their
 	// current states.
-	sort.Sort(topology.ByID(e.sortedNodes))
-	e.nodeStatesDirty = false
-	return nil
+	sort.Sort(disco.ByID(e.sortedNodes))
+	e.nodesDirty = false
+	return e.sortedNodes
 }
 
 // watchNodesOnce is a helper function to use with the retry logic
 // to let us restart the client if we need to.
-func (e *Etcd) watchNodesOnce(ctx context.Context, cli *clientv3.Client) (err error) {
+func (e *Etcd) watchNodesOnce(cli *clientv3.Client) (err error) {
 	e.nodeMu.Lock()
 	// we are looking for revisions HIGHER than the highest revision we've
 	// currently seen, we don't want one equal to it.
 	minRev := e.nodeRev + 1
+	done := e.childContext.Done()
 	e.nodeMu.Unlock()
-	watcher := cli.Watch(ctx, nodePrefix, clientv3.WithPrefix(), clientv3.WithRev(minRev))
+	watcher := cli.Watch(clientv3.WithRequireLeader(e.childContext), nodePrefix, clientv3.WithPrefix(), clientv3.WithRev(minRev))
 	for {
 		select {
-
-		case <-e.closeWatch:
-			return errEtcdShuttingDown
+		case <-done:
+			// we're done, this is not an error
+			return nil
 		case resp := <-watcher:
 			if err := resp.Err(); err != nil {
+				if resp.CompactRevision > minRev {
+					e.logger.Infof("watching node status, wanted rev %d, minimum now %d",
+						minRev, resp.CompactRevision)
+					// We've been told that any request with a revision under
+					// CompactRevision will always fail. Set nodeRev to one less than that,
+					// so we'll specify it as the minimum when we retry.
+					//
+					// We currently have no obvious way to verify that this will work.
+					e.nodeRev = resp.CompactRevision - 1
+				}
 				return err
 			}
 			// lock the node mutex for this whole process of updating so
@@ -596,41 +626,37 @@ func (e *Etcd) watchNodesOnce(ctx context.Context, cli *clientv3.Client) (err er
 				case mvccpb.PUT:
 					err := e.putNodeData(ev.Kv.Key, ev.Kv.Value, ev.Kv.ModRevision)
 					if err != nil {
-						e.logger.Printf("put event: %v", err)
+						e.logger.Warnf("put event: %v", err)
 					}
 				case mvccpb.DELETE:
 					err := e.deleteNodeData(ev.Kv.Key, ev.Kv.ModRevision)
 					if err != nil {
-						e.logger.Printf("delete event: %v", err)
+						e.logger.Warnf("delete event: %v", err)
 					}
 				default:
-					e.logger.Printf("watchp %q: unknown event %#v", e.options.Name, ev)
+					e.logger.Warnf("watchp %q: unknown event %#v", e.options.Name, ev)
 				}
 			}
 			e.nodeMu.Unlock()
-
 		}
 	}
 }
 
-// WatchNodes monitors changes to /heartbeat/ and /metadata/;
+// watchNodes monitors changes to /heartbeat/ and /metadata/;
 // basically, it catches changes to cluster state, but ignores the schema.
-func (e *Etcd) WatchNodes() {
-	ctx, cancel := context.WithCancel(context.Background())
-	e.watchCancel = cancel
-	watchInContext := func(cli *clientv3.Client) error {
-		return e.watchNodesOnce(ctx, cli)
-	}
+func (e *Etcd) watchNodes() {
 	// retryClient will retry on leader failure, but not for other failures
 	// such as ErrCompacted which can terminate a watch. But we want to resume
 	// watching again as long as our context isn't cancelled. The context
 	// should get cancelled when this Etcd gets shut down.
-	for ctx.Err() == nil {
-		err := e.retryClient(watchInContext)
+	for e.childContext.Err() == nil {
+		err := e.retryClient(func(cli *clientv3.Client) error {
+			return e.watchNodesOnce(cli)
+		})
 		if err != nil {
-			e.logger.Printf("WatchNodes: error from watch client: %v", err)
+			e.logger.Warnf("watchNodes: error from watch client: %v", err)
 		}
-		// delay slightly on error so we don't go completely crazy
+		// delay slightly on watch termination so we don't go completely crazy
 		time.Sleep(1 * time.Second)
 	}
 }
@@ -704,24 +730,15 @@ func (e *Etcd) Schema(ctx context.Context) (disco.Schema, error) {
 	return m, nil
 }
 
-func (e *Etcd) Metadata(ctx context.Context, peerID string) ([]byte, error) {
-	e.nodeMu.Lock()
-	defer e.nodeMu.Unlock()
-	err := e.populateNodeStates(ctx)
+func (e *Etcd) SetMetadata(ctx context.Context, node *disco.Node) error {
+	// Set metadata for this node.
+	data, err := json.Marshal(node)
 	if err != nil {
-		return nil, err
+		return errors.Wrap(err, "marshaling json metadata")
 	}
-	data, ok := e.knownNodes[peerID]
-	if !ok {
-		return nil, errors.New("node not found")
-	}
-	return data.metadata, nil
-}
-
-func (e *Etcd) SetMetadata(ctx context.Context, metadata []byte) error {
-	err := e.putKey(ctx, path.Join(metadataPrefix,
+	err = e.putKey(ctx, path.Join(metadataPrefix,
 		e.e.Server.ID().String()),
-		string(metadata),
+		string(data),
 	)
 	if err != nil {
 		return errors.Wrap(err, "SetMetadata")
@@ -979,31 +996,6 @@ func (e *Etcd) delKey(ctx context.Context, key string, withPrefix bool) (err err
 	return err
 }
 
-func memberList(cli *clientv3.Client) (ids []uint64, names []string, urls []string) {
-	ml, err := cli.MemberList(context.TODO())
-	if err != nil {
-		panic(err)
-	}
-	n := len(ml.Members)
-	ids = make([]uint64, n)
-	names = make([]string, n)
-	urls = make([]string, n)
-
-	for i, m := range ml.Members {
-		ids[i], names[i], urls[i] = m.ID, m.Name, m.PeerURLs[0]
-	}
-	return
-}
-
-func memberAdd(cli *clientv3.Client, peerURL string) (id uint64, name string) {
-	ma, err := cli.MemberAdd(context.TODO(), []string{peerURL})
-	if err != nil {
-		return 0, ""
-	}
-
-	return ma.Member.ID, ma.Member.Name
-}
-
 // Shards implements the Sharder interface.
 func (e *Etcd) Shards(ctx context.Context, index, field string) ([][]byte, error) {
 	key := path.Join(shardPrefix, index, field)
@@ -1031,19 +1023,15 @@ func (e *Etcd) SetShards(ctx context.Context, index, field string, shards []byte
 
 // Nodes implements the Noder interface. It returns the sorted list of nodes
 // based on the etcd peers.
-func (e *Etcd) Nodes() []*topology.Node {
+func (e *Etcd) Nodes() []*disco.Node {
 	e.nodeMu.Lock()
 	defer e.nodeMu.Unlock()
-	err := e.populateNodeStates(context.TODO())
-	if err != nil {
-		return nil
-	}
-	return e.sortedNodes
+	return e.populateNodeStates(context.TODO())
 }
 
 // PrimaryNodeID implements the Noder interface.
-func (e *Etcd) PrimaryNodeID(hasher topology.Hasher) string {
-	return topology.PrimaryNodeID(e.NodeIDs(), hasher)
+func (e *Etcd) PrimaryNodeID(hasher disco.Hasher) string {
+	return disco.PrimaryNodeID(e.NodeIDs(), hasher)
 }
 
 // NodeIDs returns the list of node IDs in the etcd cluster.
diff --git a/etcd/embed_test.go b/etcd/embed_test.go
index c737c62f9..ae2d5f8f2 100644
--- a/etcd/embed_test.go
+++ b/etcd/embed_test.go
@@ -6,7 +6,6 @@ import (
 	"testing"
 	"time"
 
-	pilosa "github.com/molecula/featurebase/v3"
 	"github.com/molecula/featurebase/v3/disco"
 	"github.com/molecula/featurebase/v3/logger"
 	"go.etcd.io/etcd/server/v3/embed"
@@ -15,10 +14,14 @@ import (
 func TestRestartEtcd(t *testing.T) {
 	cfg := embed.NewConfig()
 	cfg.Dir = "default.etcd"
-	curl, _ := url.Parse(pilosa.EtcdUnixSocket(t))
-	cfg.LPUrls = append(cfg.LPUrls, *curl)
-	curl, _ = url.Parse(pilosa.EtcdUnixSocket(t))
-	cfg.LCUrls = append(cfg.LCUrls, *curl)
+	curl, _ := url.Parse(unixSocket(t))
+	// append-in-place to replace any existing URLs with our new URL
+	cfg.LPUrls = append(cfg.LPUrls[:0], *curl)
+	cfg.APUrls = cfg.LPUrls
+	cfg.InitialCluster = "default=" + cfg.LPUrls[0].String()
+	curl, _ = url.Parse(unixSocket(t))
+	cfg.LCUrls = append(cfg.LCUrls[:0], *curl)
+	cfg.ACUrls = cfg.LCUrls
 	e, err := embed.StartEtcd(cfg)
 	if err != nil {
 		t.Fatal(err)
@@ -49,13 +52,19 @@ func TestRestartEtcd(t *testing.T) {
 
 func TestParseOptions(t *testing.T) {
 	var e = &Etcd{options: Options{ClusterURL: "http://foo"}, logger: logger.NewLogfLogger(t)}
-	curl, _ := url.Parse(pilosa.EtcdUnixSocket(t))
+	curl, _ := url.Parse(unixSocket(t))
 	e.options.LClientURL = curl.String()
-	curl, _ = url.Parse(pilosa.EtcdUnixSocket(t))
+
+	e.options.LPeerURL = "i'm a teapot"
+	_, err := e.parseOptions()
+	if err == nil {
+		t.Fatalf("invalid peer URL should be rejected")
+	}
+	curl, _ = url.Parse(unixSocket(t))
 	e.options.LPeerURL = curl.String()
 
 	e.options.ClusterURL = "http://foo"
-	_, err := e.parseOptions()
+	_, err = e.parseOptions()
 	if err == nil {
 		t.Fatalf("cluster URL should be rejected")
 	}
diff --git a/etcd/fake_test.go b/etcd/fake_test.go
new file mode 100644
index 000000000..42600e4d4
--- /dev/null
+++ b/etcd/fake_test.go
@@ -0,0 +1,126 @@
+package etcd
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/molecula/featurebase/v3/disco"
+	"github.com/molecula/featurebase/v3/logger"
+	"golang.org/x/sync/errgroup"
+)
+
+// fakeCluster is a cluster of just our local etcd wrappers, without the
+// rest of featurebase present. This code is expected to migrate into a
+// _test.go file later, but for now it's here so we can see code coverage.
+type fakeCluster struct {
+	tb    testing.TB
+	nodes []*Etcd
+}
+
+func (f *fakeCluster) Start() error {
+	eg, ctx := errgroup.WithContext(context.Background())
+	for _, node := range f.nodes {
+		node := node
+		eg.Go(func() error {
+			_, err := node.Start(ctx)
+			return err
+		})
+	}
+	return eg.Wait()
+}
+
+func (f *fakeCluster) BringUp() error {
+	eg, ctx := errgroup.WithContext(context.Background())
+	for _, node := range f.nodes {
+		node := node
+		eg.Go(func() error {
+			return node.SetState(ctx, disco.NodeStateStarted)
+		})
+	}
+	return eg.Wait()
+}
+
+// AwaitClusterState verifies that node 0 thinks the cluster is in the
+// requested state, or tells you why it failed.
+func (f *fakeCluster) AwaitClusterState(expected disco.ClusterState) (err error) {
+	state := disco.ClusterState("")
+	for state != expected {
+		state, err = f.nodes[0].ClusterState(context.Background())
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// MustAwaitClusterState verifies that node 0 thinks the cluster is in the
+// requested state, or fails a test.
+func (f *fakeCluster) MustAwaitClusterState(expected disco.ClusterState) {
+	err := f.AwaitClusterState(expected)
+	if err != nil {
+		f.tb.Fatalf("awaiting cluster state %s: %v", expected, err)
+	}
+}
+
+func (f *fakeCluster) Stop() error {
+	for i, node := range f.nodes {
+		err := node.Close()
+		if err != nil {
+			return fmt.Errorf("failure closing node %d: %v", i, err)
+		}
+	}
+	return nil
+}
+
+func (f *fakeCluster) MustNodeStates() []disco.NodeState {
+	nodes := f.nodes[0].Nodes()
+	states := make([]disco.NodeState, len(nodes))
+	for i, node := range nodes {
+		states[i] = node.State
+	}
+	return states
+}
+
+func (f *fakeCluster) MustClusterState() disco.ClusterState {
+	state, err := f.nodes[0].ClusterState(context.TODO())
+	if err != nil {
+		f.tb.Fatalf("getting cluster state: %v", err)
+	}
+	return state
+}
+
+// Elect tries to force a leader election by identifying a leader
+// and then stopping it. The cluster, if it has at least 3 members,
+// should still stay running.
+func (f *fakeCluster) Elect() (oldleader *Etcd, err error) {
+	var leader *Etcd
+	for i, node := range f.nodes {
+		if node.IsLeader() {
+			f.tb.Logf("leader is node %d, stopping it", i)
+			leader = node
+			copy(f.nodes[i:], f.nodes[i+1:])
+			f.nodes = f.nodes[:len(f.nodes)-1]
+		}
+	}
+	// close the leader
+	err = leader.Close()
+	if err != nil {
+		return leader, err
+	}
+	_, err = f.nodes[0].ClusterState(context.TODO())
+	return leader, err
+}
+
+// NewFakeCluster creates a fakeCluster of n nodes, providing only the
+// etcd objects, not the rest of a featurebase install.
+func NewFakeCluster(tb testing.TB, n int, replicas int) *fakeCluster {
+	logger := logger.NewLogfLogger(tb)
+	_, opts := GenEtcdConfigs(tb, n)
+	fc := &fakeCluster{nodes: make([]*Etcd, n)}
+	for i := range fc.nodes {
+		fc.nodes[i] = NewEtcd(opts[i], logger, replicas, "foo")
+	}
+	fc.tb = tb
+	return fc
+}
diff --git a/etcd/leasedkv.go b/etcd/leasedkv.go
index 70a592358..44cea686f 100644
--- a/etcd/leasedkv.go
+++ b/etcd/leasedkv.go
@@ -17,10 +17,11 @@ import (
 // It will try to renew the lease at any cost after losing it.
 // It will recreate the previous existing value for the key again.
 type leasedKV struct {
-	e       *Etcd
-	cancel  context.CancelFunc
-	done    <-chan struct{}
-	leaseID clientv3.LeaseID
+	e             *Etcd
+	parentContext context.Context
+	cancel        context.CancelFunc
+	done          <-chan struct{}
+	leaseID       clientv3.LeaseID
 
 	key        string
 	ttlSeconds int64
@@ -30,11 +31,12 @@ type leasedKV struct {
 	stopped bool   // protected by mu
 }
 
-func newLeasedKV(e *Etcd, key string, ttlSeconds int64) *leasedKV {
+func newLeasedKV(e *Etcd, ctx context.Context, key string, ttlSeconds int64) *leasedKV {
 	return &leasedKV{
-		e:          e,
-		key:        key,
-		ttlSeconds: ttlSeconds,
+		e:             e,
+		parentContext: ctx,
+		key:           key,
+		ttlSeconds:    ttlSeconds,
 	}
 }
 
@@ -53,8 +55,12 @@ func (l *leasedKV) Start(initValue string) error {
 	return nil
 }
 
+// create creates the lease and yields a KeepAlive channel. It also stashes a cancel
+// function for our local context that we use in case of internal issues, and the
+// done channel for the internal context, which will be readable as soon as either
+// our context or the parent context is done.
 func (l *leasedKV) create(initValue string) (<-chan *clientv3.LeaseKeepAliveResponse, error) {
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancel(l.parentContext)
 
 	if l.cancel != nil {
 		l.cancel()
@@ -99,8 +105,6 @@ func (l *leasedKV) create(initValue string) (<-chan *clientv3.LeaseKeepAliveResp
 func (l *leasedKV) consumeLease(ch <-chan *clientv3.LeaseKeepAliveResponse) {
 	for {
 		select {
-		case <-l.e.closeWatch:
-			return
 		case _, ok := <-ch:
 			if ok {
 				continue
@@ -146,7 +150,9 @@ func (l *leasedKV) Stop() {
 		l.cancel()
 	}
 	// low-effort attempt to cancel existing lease. if the cluster is
-	// shutting down, we don't want this to take long.
+	// shutting down, we don't want this to take long. Note that we don't
+	// use the parent context for this -- if we got cancelled, we still
+	// want this attempt to run.
 	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
 	err := l.e.retryClient(func(cli *clientv3.Client) (err error) {
 		_, err = cli.Revoke(ctx, l.leaseID)
@@ -213,6 +219,13 @@ func (l *leasedKV) Get(ctx context.Context) (string, error) {
 	return l.value, nil
 }
 
+// retry retries a function at a given interval until it succeeds, or until it
+// returns context.DeadlineExceeded, at which point we return the last other
+// error it returned, or DeadlineExceeded if we didn't have another previous
+// error. So other errors (connection failures, etcetera) get retried, but
+// DeadlineExceeded means we're done trying. But, if we failed due to a
+// connection error, then got a DeadlineExceeded on a retry, we want to report
+// the connection error, which is a lot more informative.
 func retry(desc string, sleep time.Duration, f func() error) (err error) {
 	for {
 		lastErr := f()
diff --git a/etcd/leasedkv_test.go b/etcd/leasedkv_test.go
index 6a9ad947c..f80791064 100644
--- a/etcd/leasedkv_test.go
+++ b/etcd/leasedkv_test.go
@@ -3,13 +3,10 @@ package etcd
 
 import (
 	"context"
-	"fmt"
-	"net"
 	"os"
 	"testing"
 	"time"
 
-	pilosa "github.com/molecula/featurebase/v3"
 	"github.com/molecula/featurebase/v3/disco"
 	"github.com/molecula/featurebase/v3/logger"
 	"github.com/molecula/featurebase/v3/testhook"
@@ -23,11 +20,46 @@ import (
 const initVal = "test"
 const newVal = "newValue"
 
+func TestClusterKv(t *testing.T) {
+	if !AllowCluster() {
+		t.Skip("only testing clusters when clustering is allowed")
+	}
+	c := NewFakeCluster(t, 3, 2)
+	err := c.Start()
+	if err != nil {
+		t.Fatalf("starting cluster: %v", err)
+	}
+	c.MustAwaitClusterState(disco.ClusterStateDown)
+	err = c.BringUp()
+	if err != nil {
+		t.Fatalf("bringing up cluster: %v", err)
+	}
+	c.MustAwaitClusterState(disco.ClusterStateNormal)
+	_, err = c.Elect()
+	if err != nil {
+		t.Fatalf("trying to cause election: %v", err)
+	}
+	ctx := context.TODO()
+	c.nodes[0].SetState(ctx, disco.NodeStateStarting)
+	c.nodes[1].SetState(ctx, disco.NodeStateStarting)
+	c.MustAwaitClusterState(disco.ClusterStateStarting)
+	c.nodes[0].SetState(ctx, disco.NodeStateStarted)
+	c.nodes[1].SetState(ctx, disco.NodeStateStarted)
+	// Two of three nodes are up, one is down, we have 2 replicas, so
+	// we should be able to handle reads but not writes, so we're in
+	// a Degraded state.
+	c.MustAwaitClusterState(disco.ClusterStateDegraded)
+	err = c.Stop()
+	if err != nil {
+		t.Fatalf("stopping cluster: %v", err)
+	}
+}
+
 func TestLeasedKv(t *testing.T) {
 	cfg := embed.NewConfig()
 
-	clientURL := pilosa.EtcdUnixSocket(t)
-	peerURL := pilosa.EtcdUnixSocket(t)
+	clientURL := unixSocket(t)
+	peerURL := unixSocket(t)
 	cfg.LPUrls = types.MustNewURLs([]string{peerURL})
 	cfg.APUrls = types.MustNewURLs([]string{peerURL})
 	cfg.LCUrls = types.MustNewURLs([]string{clientURL})
@@ -53,7 +85,7 @@ func TestLeasedKv(t *testing.T) {
 	}()
 	wrapper := &Etcd{e: etcd, cli: cli, logger: logger.NewLogfLogger(t)}
 
-	lkv := newLeasedKV(wrapper, "/test", 1)
+	lkv := newLeasedKV(wrapper, context.TODO(), "/test", 1)
 
 	ctx := context.Background()
 
@@ -105,16 +137,3 @@ func TestLeasedKv(t *testing.T) {
 		t.Fatal("expected error:", disco.ErrNoResults, "obtained:", err)
 	}
 }
-
-// listenerWithURL builds a TCP listener and corresponding http://localhost:%d
-// URL, and returns those.
-func listenerWithURL() (listener *net.TCPListener, url string, err error) {
-	l, err := net.Listen("tcp", "localhost:0")
-	if err != nil {
-		return listener, url, err
-	}
-	listener = l.(*net.TCPListener)
-	port := listener.Addr().(*net.TCPAddr).Port
-	url = fmt.Sprintf("http://localhost:%d", port)
-	return listener, url, err
-}
diff --git a/event.go b/event.go
index 811e6c7d9..5bebe82d7 100644
--- a/event.go
+++ b/event.go
@@ -1,7 +1,7 @@
 // Copyright 2021 Molecula Corp. All rights reserved.
 package pilosa
 
-import "github.com/molecula/featurebase/v3/topology"
+import "github.com/molecula/featurebase/v3/disco"
 
 // NodeEventType are the types of node events.
 type NodeEventType int
@@ -16,5 +16,5 @@ const (
 // NodeEvent is a single event related to node activity in the cluster.
 type NodeEvent struct {
 	Event NodeEventType
-	Node  *topology.Node
+	Node  *disco.Node
 }
diff --git a/executor.go b/executor.go
index 5ac0a38db..b3feb40aa 100644
--- a/executor.go
+++ b/executor.go
@@ -24,7 +24,6 @@ import (
 	"github.com/molecula/featurebase/v3/shardwidth"
 	"github.com/molecula/featurebase/v3/task"
 	"github.com/molecula/featurebase/v3/testhook"
-	"github.com/molecula/featurebase/v3/topology"
 	"github.com/molecula/featurebase/v3/tracing"
 	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
@@ -49,7 +48,7 @@ type executor struct {
 	Holder *Holder
 
 	// Local hostname & cluster configuration.
-	Node    *topology.Node
+	Node    *disco.Node
 	Cluster *cluster
 
 	// Client used for remote requests.
@@ -5921,7 +5920,7 @@ func (e *executor) executeClearValueField(ctx context.Context, qcx *Qcx, index s
 }
 
 // remoteExec executes a PQL query remotely for a set of shards on a node.
-func (e *executor) remoteExec(ctx context.Context, node *topology.Node, index string, q *pql.Query, shards []uint64, embed []*Row, maxMemory int64) (results []interface{}, err error) { // nolint: interfacer
+func (e *executor) remoteExec(ctx context.Context, node *disco.Node, index string, q *pql.Query, shards []uint64, embed []*Row, maxMemory int64) (results []interface{}, err error) { // nolint: interfacer
 	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeExec")
 	defer span.Finish()
 
@@ -5944,14 +5943,14 @@ func (e *executor) remoteExec(ctx context.Context, node *topology.Node, index st
 
 // shardsByNode returns a mapping of nodes to shards.
 // Returns errShardUnavailable if a shard cannot be allocated to a node.
-func (e *executor) shardsByNode(nodes []*topology.Node, index string, shards []uint64) (map[*topology.Node][]uint64, error) {
-	m := make(map[*topology.Node][]uint64)
+func (e *executor) shardsByNode(nodes []*disco.Node, index string, shards []uint64) (map[*disco.Node][]uint64, error) {
+	m := make(map[*disco.Node][]uint64)
 
 	// Create a snapshot of the cluster to use for node/partition calculations.
 	// We use e.Cluster.Nodes() here instead of e.Cluster.noder because we need
 	// the node states in order to ensure that we don't include an unavailable
 	// node in the map of nodes to which we distribute the query.
-	snap := topology.NewClusterSnapshot(topology.NewLocalNoder(e.Cluster.Nodes()), e.Cluster.Hasher, e.Cluster.partitionAssigner, e.Cluster.ReplicaN)
+	snap := disco.NewClusterSnapshot(disco.NewLocalNoder(e.Cluster.Nodes()), e.Cluster.Hasher, e.Cluster.partitionAssigner, e.Cluster.ReplicaN)
 
 loop:
 	for _, shard := range shards {
@@ -5959,7 +5958,7 @@ loop:
 			// 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.
-			if topology.Nodes(nodes).ContainsID(node.ID) && (node.State == disco.NodeStateStarted || node.State == disco.NodeStateUnknown) {
+			if disco.Nodes(nodes).ContainsID(node.ID) && (node.State == disco.NodeStateStarted || node.State == disco.NodeStateUnknown) {
 				m[node] = append(m[node], shard)
 				continue loop
 			}
@@ -6003,11 +6002,11 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64,
 	//
 	// However, if this request is being sent from the primary then all
 	// processing should be done locally so we start with just the local node.
-	var nodes []*topology.Node
+	var nodes []*disco.Node
 	if !opt.Remote {
-		nodes = topology.Nodes(e.Cluster.Nodes()).Clone()
+		nodes = disco.Nodes(e.Cluster.Nodes()).Clone()
 	} else {
-		nodes = []*topology.Node{e.Cluster.nodeByID(e.Node.ID)}
+		nodes = []*disco.Node{e.Cluster.nodeByID(e.Node.ID)}
 	}
 
 	// Start mapping across all primary owners.
@@ -6033,7 +6032,7 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64,
 			// the error from the healthy node and return that immediately.
 			if resp.err != nil && strings.Contains(resp.err.Error(), errConnectionRefused) {
 				// Filter out unavailable nodes.
-				nodes = topology.Nodes(nodes).FilterID(resp.node.ID)
+				nodes = disco.Nodes(nodes).FilterID(resp.node.ID)
 
 				// Begin mapper against secondary nodes.
 				if err := e.mapper(ctx, eg, ch, nodes, index, resp.shards, c, opt, true, mapFn, reduceFn); errors.Cause(err) == errShardUnavailable {
@@ -6107,7 +6106,7 @@ func makeEmbeddedDataForShards(allRows []*Row, shards []uint64) []*Row {
 	return newRows
 }
 
-func (e *executor) mapper(ctx context.Context, eg *errgroup.Group, ch chan mapResponse, nodes []*topology.Node, index string, shards []uint64, c *pql.Call, opt *execOptions, lastAttempt bool, mapFn mapFunc, reduceFn reduceFunc) (reterr error) {
+func (e *executor) mapper(ctx context.Context, eg *errgroup.Group, ch chan mapResponse, nodes []*disco.Node, index string, shards []uint64, c *pql.Call, opt *execOptions, lastAttempt bool, mapFn mapFunc, reduceFn reduceFunc) (reterr error) {
 	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.mapper")
 	defer span.Finish()
 
@@ -7702,7 +7701,7 @@ type mapOptions struct {
 type reduceFunc func(ctx context.Context, prev, v interface{}) interface{}
 
 type mapResponse struct {
-	node   *topology.Node
+	node   *disco.Node
 	shards []uint64
 
 	result interface{}
@@ -8774,7 +8773,7 @@ func (c *NopCommitor) Commit() error {
 
 func deleteKeyTranslation(ctx context.Context, idx *Index, shard uint64, records *roaring.Bitmap) (Commitor, error) {
 	// ShardToShardParition ...
-	paritionID := topology.ShardToShardPartition(idx.name, shard, idx.holder.partitionN)
+	paritionID := disco.ShardToShardPartition(idx.name, shard, idx.holder.partitionN)
 
 	return idx.TranslateStore(paritionID).Delete(records)
 }
diff --git a/fragment.go b/fragment.go
index 97dcf1be6..cca6ad568 100644
--- a/fragment.go
+++ b/fragment.go
@@ -23,6 +23,7 @@ import (
 
 	"github.com/cespare/xxhash"
 	"github.com/gogo/protobuf/proto"
+	"github.com/molecula/featurebase/v3/disco"
 	"github.com/molecula/featurebase/v3/logger"
 	pnet "github.com/molecula/featurebase/v3/net"
 	"github.com/molecula/featurebase/v3/pb"
@@ -31,7 +32,6 @@ import (
 	"github.com/molecula/featurebase/v3/shardwidth"
 	"github.com/molecula/featurebase/v3/stats"
 	"github.com/molecula/featurebase/v3/testhook"
-	"github.com/molecula/featurebase/v3/topology"
 	"github.com/molecula/featurebase/v3/tracing"
 	"github.com/molecula/featurebase/v3/vprint"
 	"github.com/pkg/errors"
@@ -3162,7 +3162,7 @@ func (h *blockHasher) WriteValue(v uint64) {
 type fragmentSyncer struct {
 	Fragment *fragment
 
-	Node    *topology.Node
+	Node    *disco.Node
 	Cluster *cluster
 
 	// FieldType helps determine which method of syncing to use.
diff --git a/holder.go b/holder.go
index bfa2fcb5d..f7dfe953f 100644
--- a/holder.go
+++ b/holder.go
@@ -19,7 +19,6 @@ import (
 	"github.com/molecula/featurebase/v3/stats"
 	"github.com/molecula/featurebase/v3/storage"
 	"github.com/molecula/featurebase/v3/testhook"
-	"github.com/molecula/featurebase/v3/topology"
 	"github.com/molecula/featurebase/v3/vprint"
 	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
@@ -211,7 +210,7 @@ type HolderConfig struct {
 
 func DefaultHolderConfig() *HolderConfig {
 	return &HolderConfig{
-		PartitionN:           topology.DefaultPartitionN,
+		PartitionN:           disco.DefaultPartitionN,
 		OpenTranslateStore:   OpenInMemTranslateStore,
 		OpenTranslateReader:  nil,
 		OpenTransactionStore: OpenInMemTransactionStore,
@@ -1197,7 +1196,7 @@ type holderSyncer struct {
 
 	Holder *Holder
 
-	Node    *topology.Node
+	Node    *disco.Node
 	Cluster *cluster
 
 	// Translation sync handling.
@@ -1447,7 +1446,7 @@ func (s *holderSyncer) stopTranslationSync() error {
 // setTranslateReadOnlyFlags updates all translation stores to enable or disable
 // writing new translation keys. Index stores are writable if the node owns the
 // partition. Field stores are writable if the node is the primary.
-func (s *holderSyncer) setTranslateReadOnlyFlags(snap *topology.ClusterSnapshot) {
+func (s *holderSyncer) setTranslateReadOnlyFlags(snap *disco.ClusterSnapshot) {
 	s.Cluster.mu.RLock()
 	isPrimaryFieldTranslator := snap.IsPrimaryFieldTranslationNode(s.Cluster.Node.ID)
 
@@ -1490,7 +1489,7 @@ func (s *holderSyncer) setTranslateReadOnlyFlags(snap *topology.ClusterSnapshot)
 // any key translation, whether that's field keys (every node replicates these
 // from the primary) or index keys (only the replica nodes for each partition
 // replicate these from whichever node is primary for that partition).
-func (s *holderSyncer) initializeReplication(snap *topology.ClusterSnapshot) error {
+func (s *holderSyncer) initializeReplication(snap *disco.ClusterSnapshot) error {
 	nodeMaps := make(map[string]TranslateOffsetMap)
 	if snap.ReplicaN > 1 {
 		if err := s.populateIndexReplication(nodeMaps, snap); err != nil {
@@ -1502,7 +1501,7 @@ func (s *holderSyncer) initializeReplication(snap *topology.ClusterSnapshot) err
 	}
 
 	// filter out empty nodes
-	nodes := make(map[*topology.Node]bool)
+	nodes := make(map[*disco.Node]bool)
 	for _, node := range snap.Nodes {
 		m := nodeMaps[node.ID]
 		if !m.Empty() {
@@ -1599,7 +1598,7 @@ func (s *holderSyncer) initializeReplication(snap *topology.ClusterSnapshot) err
 // populateFieldReplication populates a map from node IDs to TranslateOffsetMaps
 // to record that we need to translate fields which have key translation
 // from the primary node.
-func (s *holderSyncer) populateFieldReplication(nodeMaps map[string]TranslateOffsetMap, snap *topology.ClusterSnapshot) error {
+func (s *holderSyncer) populateFieldReplication(nodeMaps map[string]TranslateOffsetMap, snap *disco.ClusterSnapshot) error {
 	// Set up field translation
 	if !snap.IsPrimaryFieldTranslationNode(s.Cluster.Node.ID) {
 		primaryID := snap.PrimaryFieldTranslationNode().ID
@@ -1633,7 +1632,7 @@ func (s *holderSyncer) populateFieldReplication(nodeMaps map[string]TranslateOff
 // to record which nodes we need to replicate index key translation for.
 // That means nodes which are the primary for a partition that we're a
 // non-primary replica for.
-func (s *holderSyncer) populateIndexReplication(nodeMaps map[string]TranslateOffsetMap, snap *topology.ClusterSnapshot) error {
+func (s *holderSyncer) populateIndexReplication(nodeMaps map[string]TranslateOffsetMap, snap *disco.ClusterSnapshot) error {
 	for _, node := range snap.Nodes {
 		if node.ID == s.Node.ID {
 			continue
@@ -1647,8 +1646,8 @@ func (s *holderSyncer) populateIndexReplication(nodeMaps map[string]TranslateOff
 			}
 			for partitionID := 0; partitionID < snap.PartitionN; partitionID++ {
 				partitionNodes := snap.PartitionNodes(partitionID)
-				isPrimary := partitionNodes[0].ID == node.ID                          // remote is primary?
-				isReplica := topology.Nodes(partitionNodes[1:]).ContainsID(s.Node.ID) // local is replica?
+				isPrimary := partitionNodes[0].ID == node.ID                       // remote is primary?
+				isReplica := disco.Nodes(partitionNodes[1:]).ContainsID(s.Node.ID) // local is replica?
 				if !isPrimary || !isReplica {
 					continue
 				}
@@ -1677,7 +1676,7 @@ func (s *holderSyncer) populateIndexReplication(nodeMaps map[string]TranslateOff
 // readBothTranslateReader reads key translation for field keys or
 // index keys from a remote node. Both field and index keys may be sent,
 // the distinction is that field keys have a non-empty field name.
-func (s *holderSyncer) readBothTranslateReader(rd TranslateEntryReader, snap *topology.ClusterSnapshot) {
+func (s *holderSyncer) readBothTranslateReader(rd TranslateEntryReader, snap *disco.ClusterSnapshot) {
 	for {
 		var entry TranslateEntry
 		if err := rd.ReadEntry(&entry); err != nil {
diff --git a/http_handler.go b/http_handler.go
index 2f6af1288..a60ea5e25 100644
--- a/http_handler.go
+++ b/http_handler.go
@@ -31,13 +31,13 @@ import (
 	"github.com/gorilla/mux"
 	"github.com/molecula/featurebase/v3/authn"
 	"github.com/molecula/featurebase/v3/authz"
+	"github.com/molecula/featurebase/v3/disco"
 	"github.com/molecula/featurebase/v3/ingest"
 	"github.com/molecula/featurebase/v3/logger"
 	"github.com/molecula/featurebase/v3/monitor"
 	"github.com/molecula/featurebase/v3/pql"
 	"github.com/molecula/featurebase/v3/rbf"
 	"github.com/molecula/featurebase/v3/storage"
-	"github.com/molecula/featurebase/v3/topology"
 	"github.com/molecula/featurebase/v3/tracing"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
@@ -1227,10 +1227,10 @@ type getSchemaResponse struct {
 }
 
 type getStatusResponse struct {
-	State       string           `json:"state"`
-	Nodes       []*topology.Node `json:"nodes"`
-	LocalID     string           `json:"localID"`
-	ClusterName string           `json:"clusterName"`
+	State       string        `json:"state"`
+	Nodes       []*disco.Node `json:"nodes"`
+	LocalID     string        `json:"localID"`
+	ClusterName string        `json:"clusterName"`
 }
 
 func httpHash(s string) string {
diff --git a/internal/clustertests/pause_node_test.go b/internal/clustertests/pause_node_test.go
index f4b49078f..a88002fe1 100644
--- a/internal/clustertests/pause_node_test.go
+++ b/internal/clustertests/pause_node_test.go
@@ -20,7 +20,6 @@ import (
 	"github.com/molecula/featurebase/v3/disco"
 	"github.com/molecula/featurebase/v3/encoding/proto"
 	"github.com/molecula/featurebase/v3/net"
-	"github.com/molecula/featurebase/v3/topology"
 	"github.com/pkg/errors"
 )
 
@@ -136,7 +135,7 @@ func openTranslateStores(dirPath, index string) (map[int]pilosa.TranslateStore,
 	}
 
 	// in case of error, close any translateStore that has been opened
-	rollback := make([]pilosa.TranslateStore, 0, topology.DefaultPartitionN)
+	rollback := make([]pilosa.TranslateStore, 0, disco.DefaultPartitionN)
 	defer func() {
 		for _, ts := range rollback {
 			_ = ts.Close()
@@ -162,7 +161,7 @@ func openTranslateStores(dirPath, index string) (map[int]pilosa.TranslateStore,
 			return nil, err
 		}
 		// open bolt db
-		ts, err := boltdb.OpenTranslateStore(filePath, index, "", partition, topology.DefaultPartitionN, false)
+		ts, err := boltdb.OpenTranslateStore(filePath, index, "", partition, disco.DefaultPartitionN, false)
 		ts.SetReadOnly(true)
 		if err != nil {
 			return nil, err
@@ -193,7 +192,7 @@ func verifyNodeHasGivenKeys(ctx context.Context, node, index, dirPath string, ke
 	}
 
 	// read in all the translate stores for each partition
-	for partition := 0; partition < topology.DefaultPartitionN; partition++ {
+	for partition := 0; partition < disco.DefaultPartitionN; partition++ {
 		err := readIndexTranslateData(ctx, client, nodeDirPath, index, partition)
 		if err != nil {
 			return err
@@ -325,8 +324,8 @@ func TestPauseReplica(t *testing.T) {
 	}
 
 	// generate mapping from partition to primary node
-	partitionToNode := make([]string, topology.DefaultPartitionN)
-	for partition := 0; partition < topology.DefaultPartitionN; partition++ {
+	partitionToNode := make([]string, disco.DefaultPartitionN)
+	for partition := 0; partition < disco.DefaultPartitionN; partition++ {
 		nodes, err := cli.PartitionNodes(ctx, partition)
 		if err != nil {
 			t.Fatal(err)
@@ -345,7 +344,7 @@ func TestPauseReplica(t *testing.T) {
 		h := fnv.New64a()
 		_, _ = h.Write([]byte(index))
 		_, _ = h.Write([]byte(key))
-		partition := int(h.Sum64() % uint64(topology.DefaultPartitionN))
+		partition := int(h.Sum64() % uint64(disco.DefaultPartitionN))
 		// get node for this partition
 		return partitionToNode[partition]
 	}
diff --git a/internal_client.go b/internal_client.go
index dae937f9a..0056e4f31 100644
--- a/internal_client.go
+++ b/internal_client.go
@@ -21,10 +21,10 @@ import (
 
 	"github.com/hashicorp/go-retryablehttp"
 	"github.com/molecula/featurebase/v3/authn"
+	"github.com/molecula/featurebase/v3/disco"
 	"github.com/molecula/featurebase/v3/ingest"
 	"github.com/molecula/featurebase/v3/logger"
 	pnet "github.com/molecula/featurebase/v3/net"
-	"github.com/molecula/featurebase/v3/topology"
 	"github.com/molecula/featurebase/v3/tracing"
 	"github.com/pkg/errors"
 	"golang.org/x/oauth2"
@@ -541,7 +541,7 @@ func (c *InternalClient) CreateIndex(ctx context.Context, index string, opt Inde
 }
 
 // FragmentNodes returns a list of nodes that own a shard.
-func (c *InternalClient) FragmentNodes(ctx context.Context, index string, shard uint64) ([]*topology.Node, error) {
+func (c *InternalClient) FragmentNodes(ctx context.Context, index string, shard uint64) ([]*disco.Node, error) {
 	span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.FragmentNodes")
 	defer span.Finish()
 
@@ -566,7 +566,7 @@ func (c *InternalClient) FragmentNodes(ctx context.Context, index string, shard
 	}
 	defer resp.Body.Close()
 
-	var a []*topology.Node
+	var a []*disco.Node
 	if err := json.NewDecoder(resp.Body).Decode(&a); err != nil {
 		return nil, fmt.Errorf("json decode: %s", err)
 	}
@@ -574,7 +574,7 @@ func (c *InternalClient) FragmentNodes(ctx context.Context, index string, shard
 }
 
 // Nodes returns a list of all nodes.
-func (c *InternalClient) Nodes(ctx context.Context) ([]*topology.Node, error) {
+func (c *InternalClient) Nodes(ctx context.Context) ([]*disco.Node, error) {
 	span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Nodes")
 	defer span.Finish()
 
@@ -598,7 +598,7 @@ func (c *InternalClient) Nodes(ctx context.Context) ([]*topology.Node, error) {
 	}
 	defer resp.Body.Close()
 
-	var a []*topology.Node
+	var a []*disco.Node
 	if err := json.NewDecoder(resp.Body).Decode(&a); err != nil {
 		return nil, fmt.Errorf("json decode: %s", err)
 	}
@@ -665,7 +665,7 @@ func (c *InternalClient) QueryNode(ctx context.Context, uri *pnet.URI, index str
 	return qresp, nil
 }
 
-func getPrimaryNode(nodes []*topology.Node) *topology.Node {
+func getPrimaryNode(nodes []*disco.Node) *disco.Node {
 	for _, node := range nodes {
 		if node.IsPrimary {
 			return node
@@ -702,7 +702,7 @@ func (c *InternalClient) EnsureFieldWithOptions(ctx context.Context, indexName s
 }
 
 // importNode sends a pre-marshaled import request to a node.
-func (c *InternalClient) importNode(ctx context.Context, node *topology.Node, index, field string, buf []byte, opts *ImportOptions) error {
+func (c *InternalClient) importNode(ctx context.Context, node *disco.Node, index, field string, buf []byte, opts *ImportOptions) error {
 	span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.importNode")
 	defer span.Finish()
 
@@ -765,7 +765,7 @@ func (c *InternalClient) importHelper(ctx context.Context, req Message, process
 	// If we don't actually know what shards we're sending to, and we have
 	// a local API and a qcx, we'll have a process function that uses the local
 	// API. Otherwise, even if we have an API
-	var nodes []*topology.Node
+	var nodes []*disco.Node
 	var err error
 	if shard != ^uint64(0) {
 		// we need a list of nodes specific to this shard.
@@ -795,8 +795,8 @@ func (c *InternalClient) importHelper(ctx context.Context, req Message, process
 	// "us" is a usable local node if any, "them" is every node that we need
 	// to process which isn't that node. We start out with us == nil and
 	// them = the whole set of nodes.
-	var us *topology.Node
-	var them []*topology.Node = nodes
+	var us *disco.Node
+	var them []*disco.Node = nodes
 
 	// If we have an API, we know what node we are. Even if we don't have
 	// a Qcx, we still care, because looping back to the local node will
@@ -1005,7 +1005,7 @@ func (c *InternalClient) ExportCSV(ctx context.Context, index, field string, sha
 }
 
 // exportNode copies a CSV export from a node to w.
-func (c *InternalClient) exportNodeCSV(ctx context.Context, node *topology.Node, index, field string, shard uint64, w io.Writer) error {
+func (c *InternalClient) exportNodeCSV(ctx context.Context, node *disco.Node, index, field string, shard uint64, w io.Writer) error {
 	span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.exportNodeCSV")
 	defer span.Finish()
 
@@ -1048,7 +1048,7 @@ func (c *InternalClient) RetrieveShardFromURI(ctx context.Context, index, field,
 	span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.RetrieveShardFromURI")
 	defer span.Finish()
 
-	node := &topology.Node{
+	node := &disco.Node{
 		URI: uri,
 	}
 
@@ -2141,7 +2141,7 @@ func uriPathToURL(uri *pnet.URI, path string) url.URL {
 	}
 }
 
-func nodePathToURL(node *topology.Node, path string) url.URL {
+func nodePathToURL(node *disco.Node, path string) url.URL {
 	return url.URL{
 		Scheme: node.URI.Scheme,
 		Host:   node.URI.HostPort(),
@@ -2156,7 +2156,7 @@ func (c *InternalClient) RetrieveTranslatePartitionFromURI(ctx context.Context,
 	span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.RetrieveTranslatePartitionFromURI")
 	defer span.Finish()
 
-	node := &topology.Node{
+	node := &disco.Node{
 		URI: uri,
 	}
 
@@ -2303,7 +2303,7 @@ func (c *InternalClient) IDAllocDataReader(ctx context.Context) (io.ReadCloser,
 	return resp.Body, nil
 }
 
-func (c *InternalClient) IDAllocDataWriter(ctx context.Context, f io.Reader, primary *topology.Node) error {
+func (c *InternalClient) IDAllocDataWriter(ctx context.Context, f io.Reader, primary *disco.Node) error {
 	span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.IDAllocDataWriter")
 	defer span.Finish()
 
@@ -2419,7 +2419,7 @@ func (c *InternalClient) Status(ctx context.Context) (string, error) {
 	return rsp.State, nil
 }
 
-func (c *InternalClient) PartitionNodes(ctx context.Context, partitionID int) ([]*topology.Node, error) {
+func (c *InternalClient) PartitionNodes(ctx context.Context, partitionID int) ([]*disco.Node, error) {
 	span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.PartitionNodes")
 	defer span.Finish()
 
@@ -2444,7 +2444,7 @@ func (c *InternalClient) PartitionNodes(ctx context.Context, partitionID int) ([
 	}
 	defer resp.Body.Close()
 
-	var a []*topology.Node
+	var a []*disco.Node
 	if err := json.NewDecoder(resp.Body).Decode(&a); err != nil {
 		return nil, fmt.Errorf("json decode: %s", err)
 	}
diff --git a/internal_client_test.go b/internal_client_test.go
index c0f4b55c8..d768b465c 100644
--- a/internal_client_test.go
+++ b/internal_client_test.go
@@ -16,11 +16,11 @@ import (
 	"github.com/davecgh/go-spew/spew"
 	pilosa "github.com/molecula/featurebase/v3"
 	"github.com/molecula/featurebase/v3/authn"
+	"github.com/molecula/featurebase/v3/disco"
 	"github.com/molecula/featurebase/v3/encoding/proto"
 	"github.com/molecula/featurebase/v3/pql"
 	"github.com/molecula/featurebase/v3/server"
 	"github.com/molecula/featurebase/v3/test"
-	"github.com/molecula/featurebase/v3/topology"
 	"github.com/pkg/errors"
 )
 
@@ -285,7 +285,7 @@ func TestClient_Export(t *testing.T) {
 		bw := bufio.NewWriter(buf)
 
 		// Send export request for every partition.
-		for i := 0; i < topology.DefaultPartitionN; i++ {
+		for i := 0; i < disco.DefaultPartitionN; i++ {
 			if err := c.ExportCSV(context.Background(), "keyed", "unkeyedf", uint64(i), bw); err != nil {
 				t.Fatal(err)
 			}
@@ -326,7 +326,7 @@ func TestClient_Export(t *testing.T) {
 		bw := bufio.NewWriter(buf)
 
 		// Send export request.
-		for i := 0; i < topology.DefaultPartitionN; i++ {
+		for i := 0; i < disco.DefaultPartitionN; i++ {
 			if err := c.ExportCSV(context.Background(), "keyed", "keyedf", uint64(i), bw); err != nil {
 				t.Fatal(err)
 			}
diff --git a/server.go b/server.go
index a187133ca..ce2ce13d7 100644
--- a/server.go
+++ b/server.go
@@ -3,7 +3,6 @@ package pilosa
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"log"
 	"os"
@@ -25,7 +24,6 @@ import (
 	"github.com/molecula/featurebase/v3/sql2"
 	"github.com/molecula/featurebase/v3/stats"
 	"github.com/molecula/featurebase/v3/storage"
-	"github.com/molecula/featurebase/v3/topology"
 	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
 
@@ -57,9 +55,7 @@ type Server struct { // nolint: maligned
 
 	// Distributed Consensus
 	disCo     disco.DisCo
-	stator    disco.Stator
-	metadator disco.Metadator
-	noder     topology.Noder
+	noder     disco.Noder
 	sharder   disco.Sharder
 	schemator disco.Schemator
 
@@ -322,7 +318,7 @@ func OptServerNodeID(nodeID string) ServerOption {
 // OptServerClusterHasher is a functional option on Server
 // used to specify the consistent hash algorithm for data
 // location within the cluster.
-func OptServerClusterHasher(h topology.Hasher) ServerOption {
+func OptServerClusterHasher(h disco.Hasher) ServerOption {
 	return func(s *Server) error {
 		s.cluster.Hasher = h
 		return nil
@@ -399,16 +395,12 @@ func OptServerMaxQueryMemory(v int64) ServerOption {
 // OptServerDisCo is a functional option on Server
 // used to set the Distributed Consensus implementation.
 func OptServerDisCo(disCo disco.DisCo,
-	stator disco.Stator,
-	metadator disco.Metadator,
-	noder topology.Noder,
+	noder disco.Noder,
 	sharder disco.Sharder,
 	schemator disco.Schemator) ServerOption {
 
 	return func(s *Server) error {
 		s.disCo = disCo
-		s.stator = stator
-		s.metadator = metadator
 		s.noder = noder
 		s.sharder = sharder
 		s.schemator = schemator
@@ -450,9 +442,7 @@ func NewServer(opts ...ServerOption) (*Server, error) {
 		viewsRemovalInterval: time.Hour,
 
 		disCo:      disco.NopDisCo,
-		stator:     disco.NopStator,
-		metadator:  disco.NopMetadator,
-		noder:      topology.NewEmptyLocalNoder(),
+		noder:      disco.NewEmptyLocalNoder(),
 		sharder:    disco.NopSharder,
 		schemator:  disco.NopSchemator,
 		serializer: NopSerializer,
@@ -520,7 +510,6 @@ func NewServer(opts ...ServerOption) (*Server, error) {
 	s.cluster.logger = s.logger
 	s.cluster.holder = s.holder
 	s.cluster.disCo = s.disCo
-	s.cluster.stator = s.stator
 	s.cluster.noder = s.noder
 	s.cluster.sharder = s.sharder
 
@@ -600,7 +589,7 @@ func (s *Server) Open() error {
 	// Set node ID.
 	s.nodeID = s.disCo.ID()
 
-	node := &topology.Node{
+	node := &disco.Node{
 		ID:        s.nodeID,
 		URI:       s.uri,
 		GRPCURI:   s.grpcURI,
@@ -608,12 +597,7 @@ func (s *Server) Open() error {
 		IsPrimary: s.IsPrimary(),
 	}
 
-	// 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 {
+	if err := s.noder.SetMetadata(context.Background(), node); err != nil {
 		return errors.Wrap(err, "setting metadata")
 	}
 
@@ -649,7 +633,7 @@ func (s *Server) Open() error {
 	// bring up the background tasks for the holder.
 	s.holder.Activate()
 
-	if err := s.stator.Started(context.Background()); err != nil {
+	if err := s.noder.SetState(context.Background(), disco.NodeStateStarted); err != nil {
 		return errors.Wrap(err, "setting nodeState")
 	}
 
@@ -697,7 +681,7 @@ func (s *Server) Open() error {
 			<-timer.C
 		}
 		for {
-			state, err := s.stator.ClusterState(ctx)
+			state, err := s.noder.ClusterState(ctx)
 			if err != nil {
 				s.logger.Printf("failed to check cluster state: %v", err)
 				timer.Reset(time.Second)
@@ -1126,7 +1110,7 @@ func (s *Server) SendAsync(m Message) error {
 }
 
 // SendTo represents an implementation of Broadcaster.
-func (s *Server) SendTo(node *topology.Node, m Message) error {
+func (s *Server) SendTo(node *disco.Node, m Message) error {
 	msg, err := s.serializer.Marshal(m)
 	if err != nil {
 		return fmt.Errorf("marshaling message: %v", err)
@@ -1140,7 +1124,7 @@ func (s *Server) SendTo(node *topology.Node, m Message) error {
 
 // node returns the pilosa.node object. It is used by membership protocols to
 // get this node's name(ID), location(URI), and primary status.
-func (s *Server) node() *topology.Node {
+func (s *Server) node() *disco.Node {
 	return s.cluster.Node.Clone()
 }
 
diff --git a/server/server.go b/server/server.go
index c2972e272..52ae12d3b 100644
--- a/server/server.go
+++ b/server/server.go
@@ -459,7 +459,6 @@ func (m *Command) SetupServer() error {
 	}
 
 	e := petcd.NewEtcd(m.Config.Etcd, m.logger, m.Config.Cluster.ReplicaN, version)
-	discoOpt := pilosa.OptServerDisCo(e, e, e, e, e, e)
 
 	serverOptions := []pilosa.ServerOption{
 		pilosa.OptServerAntiEntropyInterval(time.Duration(m.Config.AntiEntropy.Interval)),
@@ -487,7 +486,7 @@ func (m *Command) SetupServer() error {
 		pilosa.OptServerMaxQueryMemory(m.Config.MaxQueryMemory),
 		pilosa.OptServerQueryHistoryLength(m.Config.QueryHistoryLength),
 		pilosa.OptServerPartitionAssigner(m.Config.Cluster.PartitionToNodeAssignment),
-		discoOpt,
+		pilosa.OptServerDisCo(e, e, e, e),
 	}
 
 	if m.Config.LookupDBDSN != "" {
diff --git a/test/disco.go b/test/disco.go
index 325a626eb..285c5b4f7 100644
--- a/test/disco.go
+++ b/test/disco.go
@@ -7,10 +7,8 @@ import (
 	"strings"
 	"testing"
 
-	pilosa "github.com/molecula/featurebase/v3"
 	"github.com/molecula/featurebase/v3/etcd"
 	"github.com/molecula/featurebase/v3/server"
-	"github.com/molecula/featurebase/v3/testhook"
 	"github.com/pkg/errors"
 )
 
@@ -51,22 +49,19 @@ func listenerWithURL() (listener *net.TCPListener, url string, err error) {
 	return listener, url, err
 }
 
-// GetPortsGenConfigs creates listener ports, and updates the configurations
-// of servers to match these created ports, including cross-references
-// like updating the InitCluster values in the Etcd configs.
+// GetPortsGenConfigs generates etcd configs for a number
+// of nodes, including cross-references so that every node gets
+// an initial cluster list pointing it at the other nodes,
+// and modifies the configs of the provided Command objects
+// to point to these etcd configs. It uses etcd.GenEtcdConfigs,
+// which in turn creates temporary directories and the like.
 func GetPortsGenConfigs(tb testing.TB, nodes []*Command) error {
-	peerUrls := make([]string, len(nodes))
+	clusterName, cfgs := etcd.GenEtcdConfigs(tb, len(nodes))
 	for i := range nodes {
 		if nodes[i].Config == nil {
 			nodes[i].Config = &server.Config{}
 		}
 		config := nodes[i].Config
-		name := fmt.Sprintf("server%d", i)
-		clusterName := fmt.Sprintf("cluster-%s", tb.Name())
-		discoDir, err := testhook.TempDir(tb, "disco.")
-		if err != nil {
-			return errors.Wrap(err, "creating temp directory")
-		}
 		grpcListener, grpcUrl, err := listenerWithURL()
 		if err != nil {
 			return errors.Wrap(err, "creating gRPC listener")
@@ -76,90 +71,11 @@ func GetPortsGenConfigs(tb testing.TB, nodes []*Command) error {
 		if colon != -1 {
 			grpcUrl = grpcUrl[colon:]
 		}
-		config.Name = name
+		config.Name = cfgs[i].Name
 		config.Cluster.Name = clusterName
 		config.BindGRPC = grpcUrl
 		config.GRPCListener = grpcListener
-		clientURL := pilosa.EtcdUnixSocket(tb)
-		peerURL := pilosa.EtcdUnixSocket(tb)
-		config.Etcd = etcd.Options{
-			Dir:           discoDir,
-			LClientURL:    clientURL,
-			AClientURL:    clientURL,
-			LPeerURL:      peerURL,
-			APeerURL:      peerURL,
-			HeartbeatTTL:  60,
-			UnsafeNoFsync: true,
-		}
-		peerUrls[i] = fmt.Sprintf("%s=%s", name, peerURL)
-	}
-	allPeerUrls := strings.Join(peerUrls, ",")
-	for i := range nodes {
-		nodes[i].Config.Etcd.InitCluster = allPeerUrls
+		config.Etcd = cfgs[i]
 	}
 	return nil
 }
-
-//GenPortsConfig creates specific configuration for etcd.
-func GenPortsConfig(tb testing.TB, ports []Ports) []*server.Config {
-	cfgs := make([]*server.Config, len(ports))
-	clusterURLs := make([]string, len(ports))
-	for i := range cfgs {
-		name := fmt.Sprintf("server%d", i)
-		clusterName := "cluster-abc123"
-
-		lClientURL := pilosa.EtcdUnixSocket(tb)
-		lPeerURL := pilosa.EtcdUnixSocket(tb)
-
-		discoDir := ""
-		if d, err := testhook.TempDir(tb, "disco."); err == nil {
-			discoDir = d
-		}
-
-		cfgs[i] = &server.Config{
-			Name:         name,
-			BindGRPC:     fmt.Sprintf(":%d", ports[i].Grpc),
-			GRPCListener: ports[i].LsnG,
-			Etcd: etcd.Options{
-				Dir:           discoDir,
-				LClientURL:    lClientURL,
-				AClientURL:    lClientURL,
-				LPeerURL:      lPeerURL,
-				APeerURL:      lPeerURL,
-				HeartbeatTTL:  5,
-				UnsafeNoFsync: true,
-			},
-		}
-		cfgs[i].Cluster.Name = clusterName
-
-		clusterURLs[i] = fmt.Sprintf("%s=%s", name, lPeerURL)
-	}
-	for i := range cfgs {
-		cfgs[i].Etcd.InitCluster = strings.Join(clusterURLs, ",")
-	}
-
-	return cfgs
-}
-
-func NewPorts(lsn []*net.TCPListener) []Ports {
-	var out []Ports
-
-	n := len(lsn)
-	ports := make([]int, n)
-	for i := 0; i < n; i++ {
-		ports[i] = lsn[i].Addr().(*net.TCPAddr).Port
-	}
-
-	for i := 0; i < n; i = i + 3 {
-		out = append(out, Ports{
-			LsnC:  lsn[i],
-			PortC: ports[i],
-			LsnP:  lsn[i+1],
-			PortP: ports[i+1],
-			Grpc:  ports[i+2],
-			LsnG:  lsn[i+2],
-		})
-	}
-
-	return out
-}
diff --git a/translate.go b/translate.go
index 9d5909f28..326562ba0 100644
--- a/translate.go
+++ b/translate.go
@@ -10,9 +10,9 @@ import (
 	"sort"
 	"sync"
 
+	"github.com/molecula/featurebase/v3/disco"
 	"github.com/molecula/featurebase/v3/ingest"
 	"github.com/molecula/featurebase/v3/roaring"
-	"github.com/molecula/featurebase/v3/topology"
 	"github.com/pkg/errors"
 )
 
@@ -199,7 +199,7 @@ func GenerateNextPartitionedID(index string, prev uint64, partitionID, partition
 	// Try to use the next ID if it is in the same partition.
 	// Otherwise find ID in next shard that has a matching partition.
 	for id := prev + 1; ; id += ShardWidth {
-		if topology.ShardToShardPartition(index, id/ShardWidth, partitionN) == partitionID {
+		if disco.ShardToShardPartition(index, id/ShardWidth, partitionN) == partitionID {
 			return id
 		}
 	}
diff --git a/translator_test.go b/translator_test.go
index 5df16b4a9..79bc55ca9 100644
--- a/translator_test.go
+++ b/translator_test.go
@@ -13,16 +13,16 @@ import (
 	"github.com/google/go-cmp/cmp"
 	pilosa "github.com/molecula/featurebase/v3"
 	"github.com/molecula/featurebase/v3/boltdb"
+	"github.com/molecula/featurebase/v3/disco"
 	"github.com/molecula/featurebase/v3/mock"
 	"github.com/molecula/featurebase/v3/server"
 	"github.com/molecula/featurebase/v3/test"
-	"github.com/molecula/featurebase/v3/topology"
 	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
 )
 
 func TestInMemTranslateStore_TranslateID(t *testing.T) {
-	s := pilosa.NewInMemTranslateStore("IDX", "FLD", 0, topology.DefaultPartitionN)
+	s := pilosa.NewInMemTranslateStore("IDX", "FLD", 0, disco.DefaultPartitionN)
 
 	// Setup initial keys.
 	if _, err := s.CreateKeys("foo"); err != nil {
@@ -272,7 +272,7 @@ func TestTranslation_KeyNotFound(t *testing.T) {
 }
 
 func TestInMemTranslateStore_ReadKey(t *testing.T) {
-	s := pilosa.NewInMemTranslateStore("IDX", "FLD", 0, topology.DefaultPartitionN)
+	s := pilosa.NewInMemTranslateStore("IDX", "FLD", 0, disco.DefaultPartitionN)
 
 	ids, err := s.FindKeys("foo")
 	if err != nil {
diff --git a/util.go b/util.go
index 35c5974b9..7a8bc52e9 100644
--- a/util.go
+++ b/util.go
@@ -9,16 +9,11 @@ import (
 	"path/filepath"
 	"reflect"
 	"strings"
-	"sync"
-	"testing"
 	"time"
 
 	"github.com/shirou/gopsutil/v3/mem"
 )
 
-var clientPort = os.Getpid()
-var muClientPort = sync.Mutex{}
-
 // LeftShifted16MaxContainerKey is 0xffffffffffff0000. It is similar
 // to the roaring.maxContainerKey  0x0000ffffffffffff, but
 // shifted 16 bits to the left so its domain is the full [0, 2^64) bit space.
@@ -107,24 +102,6 @@ func GetDiskUsage(path string) (DiskUsage, error) {
 	return DiskUsage{size}, err
 }
 
-// EtcdUnixSocket returns a url for use in test etcd clusters.
-func EtcdUnixSocket(tb testing.TB) string {
-	muClientPort.Lock()
-	defer func() {
-		clientPort++
-		muClientPort.Unlock()
-	}()
-	addr := fmt.Sprintf("fake:%d", clientPort)
-	tb.Cleanup(func() {
-		err := os.Remove(addr)
-
-		if err != nil && !os.IsNotExist(err) { //not an error if the socket is not present
-			tb.Logf("could not remove '%s', %v", addr, err)
-		}
-	})
-	return fmt.Sprintf("unix://%s", addr)
-}
-
 // Rev reverses a string
 func Rev(input string) string {
 	n := 0
diff --git a/utils_internal_test.go b/utils_internal_test.go
index 797253dcf..a7dcf4044 100644
--- a/utils_internal_test.go
+++ b/utils_internal_test.go
@@ -6,10 +6,10 @@ import (
 	"testing"
 	"time"
 
+	"github.com/molecula/featurebase/v3/disco"
 	"github.com/molecula/featurebase/v3/etcd"
 	pnet "github.com/molecula/featurebase/v3/net"
 	"github.com/molecula/featurebase/v3/testhook"
-	"github.com/molecula/featurebase/v3/topology"
 )
 
 // utilities used by tests
@@ -30,15 +30,15 @@ func NewTestCluster(tb testing.TB, n int) *cluster {
 	c.Hasher = NewTestModHasher()
 	c.Path = path
 
-	nodes := make([]*topology.Node, 0, n)
+	nodes := make([]*disco.Node, 0, n)
 
 	for i := 0; i < n; i++ {
-		nodes = append(nodes, &topology.Node{
+		nodes = append(nodes, &disco.Node{
 			ID:  fmt.Sprintf("node%d", i),
 			URI: NewTestURI("http", fmt.Sprintf("host%d", i), uint16(0)),
 		})
 	}
-	c.noder = topology.NewLocalNoder(nodes)
+	c.noder = disco.NewLocalNoder(nodes)
 
 	cNodes := c.noder.Nodes()