avoid race conditions on nodes

Turns out we sometimes modify returned nodes. Handle this better, but
also fix up some cases where we were generating node lists we didn't really
need to answer simple questions.

api.go

@@ -605,8 +605,8 @@ func (api *API) ExportCSV(ctx context.Context, indexName string, fieldName strin
 	snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN)
 
 	// Validate that this handler owns the shard.
-	if !snap.OwnsShard(api.Node().ID, indexName, shard) {
-		api.server.logger.Printf("node %s does not own shard %d of index %s", api.Node().ID, shard, indexName)
+	if !snap.OwnsShard(api.NodeID(), indexName, shard) {
+		api.server.logger.Printf("node %s does not own shard %d of index %s", api.NodeID(), shard, indexName)
 		return ErrClusterDoesNotOwnShard
 	}
 
@@ -807,6 +807,12 @@ func (api *API) Node() *topology.Node {
 	return api.server.node()
 }
 
+// NodeID gets the ID alone, so it doesn't have to do a complete lookup
+// of the node, searching by its ID, to return the ID it searched for.
+func (api *API) NodeID() string {
+	return api.server.nodeID
+}
+
 // PrimaryNode returns the coordinator node for the cluster.
 func (api *API) PrimaryNode() *topology.Node {
 	// Create a snapshot of the cluster to use for node/partition calculations.
@@ -1735,8 +1741,8 @@ func (api *API) validateShardOwnership(indexName string, shard uint64) error {
 	// Create a snapshot of the cluster to use for node/partition calculations.
 	snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN)
 	// Validate that this handler owns the shard.
-	if !snap.OwnsShard(api.Node().ID, indexName, shard) {
-		api.server.logger.Printf("node %s does not own shard %d of index %s", api.Node().ID, shard, indexName)
+	if !snap.OwnsShard(api.NodeID(), indexName, shard) {
+		api.server.logger.Printf("node %s does not own shard %d of index %s", api.NodeID(), shard, indexName)
 		return ErrClusterDoesNotOwnShard
 	}
 	return nil
@@ -2026,7 +2032,7 @@ func (api *API) PrimaryReplicaNodeURL() url.URL {
 	// Create a snapshot of the cluster to use for node/partition calculations.
 	snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN)
 
-	node := snap.PrimaryReplicaNode(api.Node().ID)
+	node := snap.PrimaryReplicaNode(api.NodeID())
 	if node == nil {
 		return url.URL{}
 	}
@@ -2137,7 +2143,7 @@ func (api *API) ReserveIDs(key IDAllocKey, session [32]byte, offset uint64, coun
 	// Create a snapshot of the cluster to use for node/partition calculations.
 	snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN)
 
-	if !snap.IsPrimaryFieldTranslationNode(api.Node().ID) {
+	if !snap.IsPrimaryFieldTranslationNode(api.NodeID()) {
 		return api.holder.ida.reserve(key, session, offset, count)
 	}
 
@@ -2152,7 +2158,7 @@ func (api *API) CommitIDs(key IDAllocKey, session [32]byte, count uint64) error
 	// Create a snapshot of the cluster to use for node/partition calculations.
 	snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN)
 
-	if !snap.IsPrimaryFieldTranslationNode(api.Node().ID) {
+	if !snap.IsPrimaryFieldTranslationNode(api.NodeID()) {
 		return api.holder.ida.commit(key, session, count)
 	}
 
@@ -2167,7 +2173,7 @@ func (api *API) ResetIDAlloc(index string) error {
 	// Create a snapshot of the cluster to use for node/partition calculations.
 	snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN)
 
-	if !snap.IsPrimaryFieldTranslationNode(api.Node().ID) {
+	if !snap.IsPrimaryFieldTranslationNode(api.NodeID()) {
 		return api.holder.ida.reset(index)
 	}
 

cluster.go

@@ -709,25 +709,29 @@ func (c *cluster) addNodeBasicSorted(node *topology.Node) bool {
 // concurrent use, result may be modified.
 func (c *cluster) Nodes() []*topology.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))
 
 	// Create a snapshot of the cluster to use for node/partition calculations.
-	snap := topology.NewClusterSnapshot(topology.NewLocalNoder(nodes), c.Hasher, c.ReplicaN)
-	primaryNode := snap.PrimaryFieldTranslationNode()
+	primary := topology.PrimaryNode(nodes, c.Hasher)
 
 	// Set node states and IsPrimary.
-	for _, node := range nodes {
-		node.IsPrimary = node.ID == primaryNode.ID
-
+	for i, node := range nodes {
+		copiedNodes[i] = *node
+		result[i] = &copiedNodes[i]
+		if node == primary {
+			copiedNodes[i].IsPrimary = true
+		}
 		s, err := c.stator.NodeState(context.Background(), node.ID)
 		if err != nil {
 			// TODO should we delete this?
-			node.State = string(disco.NodeStateUnknown)
+			copiedNodes[i].State = string(disco.NodeStateUnknown)
 			continue
 		}
-		node.State = string(s)
+		copiedNodes[i].State = string(s)
 	}
-
-	return nodes
+	return result
 }
 
 // removeNodeBasicSorted removes a node from the cluster, maintaining the sort

etcd/embed.go

@@ -1127,9 +1127,11 @@ func (e *Etcd) RemoveShard(ctx context.Context, index, field string, shard uint6
 // based on the etcd peers.
 func (e *Etcd) Nodes() []*topology.Node {
 	peers := e.Peers()
+	// For N>1, this might actually reduce GC load. Maybe.
+	nodeData := make([]topology.Node, len(peers))
 	nodes := make([]*topology.Node, len(peers))
 	for i, peer := range peers {
-		node := &topology.Node{}
+		node := &nodeData[i]
 
 		if meta, err := e.Metadata(context.Background(), peer.ID); err != nil {
 			log.Println(err, "getting metadata") // TODO: handle this with a logger

topology/hasher.go

@@ -39,3 +39,13 @@ func (h *Jmphasher) Hash(key uint64, n int) int {
 func (h *Jmphasher) Name() string {
 	return "jump-hash"
 }
+
+// PrimaryNode yields the node that would be selected as the primary from
+// a list, for a given ID. It assumes the list is already in the
+// expected order, as from Noder.Nodes().
+func PrimaryNode(nodes []*Node, hasher Hasher) *Node {
+	if len(nodes) == 0 {
+		return nil
+	}
+	return nodes[hasher.Hash(0, len(nodes))]
+}

topology/snapshot.go

@@ -107,8 +107,14 @@ func (c *ClusterSnapshot) ShardNodes(index string, shard uint64) []*Node {
 }
 
 // OwnsShard returns true if a host owns a fragment.
-func (c *ClusterSnapshot) OwnsShard(nodeID string, index string, shard uint64) bool {
-	return Nodes(c.ShardNodes(index, shard)).ContainsID(nodeID)
+func (c *ClusterSnapshot) OwnsShard(nodeID string, index string, shard uint64) (ret bool) {
+	idx := c.Hasher.Hash(uint64(c.ShardToShardPartition(index, shard)), len(c.Nodes))
+	for i := 0; i < c.ReplicaN; i++ {
+		if c.Nodes[(idx+i)%len(c.Nodes)].ID == nodeID {
+			return true
+		}
+	}
+	return false
 }
 
 // KeyNodes returns a list of nodes that own a key.
@@ -147,11 +153,14 @@ func (c *ClusterSnapshot) IsPrimaryFieldTranslationNode(nodeID string) bool {
 }
 
 // PrimaryPartitionNode returns the primary node of the given partition.
-func (c *ClusterSnapshot) PrimaryPartitionNode(partition int) *Node {
-	if nodes := c.PartitionNodes(partition); len(nodes) > 0 {
-		return nodes[0]
+func (c *ClusterSnapshot) PrimaryPartitionNode(partitionID int) *Node {
+	// Determine primary owner node.
+	nodeIndex := c.PrimaryNodeIndex(partitionID)
+	if nodeIndex < 0 {
+		// no nodes anyway
+		return nil
 	}
-	return nil
+	return c.Nodes[nodeIndex]
 }
 
 // IsPrimary returns true if the given node is the primary for the given