[raft] Keep URL of removed nodes and reject duplicates

pkg/raftstore/README.md

@@ -68,11 +68,11 @@ The members of a Raft group are bootstrapped from the `--raft_peers` flag on the
 
 - **Joining:** Starting a node that joins a running group is not supported yet, see [#1802](https://github.com/interuss/dss/issues/1802).
 - **Validation:** A configuration change is validated when it is proposed, to fail fast, and again by every node when it is applied. The following changes are rejected:
-  - adding a node that is already a member or with an invalid peer URL.
+  - adding a node that is already a member, with an invalid peer URL, or with the peer URL of another member.
   - removing a node that is not a member, or a voter if the group would be left with fewer than 3 voters.
   - adding or removing a node ID that was previously removed from the group.
 - **Removal:** A node stops when it applies its own removal.
-- **Persistence:** The peer URLs of the members and the removed node IDs are replicated state: they are rebuilt from the configuration changes of the log and stored in snapshots alongside the application state.
+- **Persistence:** The peer URLs of the members and of the removed nodes are replicated state: they are rebuilt from the configuration changes of the log and stored in snapshots alongside the application state.
 
 ## Future Work
 

pkg/raftstore/consensus/consensus.go

@@ -102,6 +102,7 @@ func NewConsensus(ctx context.Context, logger *zap.Logger, connectParams params.
 		membership: membership{
 			nodeID:  connectParams.NodeID,
 			members: map[uint64]member{},
+			removed: map[uint64]member{},
 		},
 	}
 
@@ -124,7 +125,7 @@ func NewConsensus(ctx context.Context, logger *zap.Logger, connectParams params.
 		if err != nil {
 			return nil, stacktrace.Propagate(err, "failed to decode snapshot")
 		}
-		err = consensus.membership.restore(initSnapshot.Members, initSnapshot.RemovedIDs)
+		err = consensus.membership.restore(initSnapshot.Members, initSnapshot.Removed)
 		if err != nil {
 			return nil, stacktrace.Propagate(err, "failed to restore membership")
 		}
@@ -356,7 +357,7 @@ func (c *Consensus) startRaftUpdatesConsumer(tickInterval time.Duration, snapsho
 						return
 					}
 
-					err = c.membership.restore(env.Members, env.RemovedIDs)
+					err = c.membership.restore(env.Members, env.Removed)
 					if err != nil {
 						c.logger.Error("failed to restore membership", zap.Error(err))
 						return
@@ -527,28 +528,25 @@ type confChangeContext struct {
 // AddNode proposes adding nodeID, reachable by its peers at peerURL, as a voter of the Raft group.
 // It blocks until the change is applied or ctx is cancelled.
 func (c *Consensus) AddNode(ctx context.Context, nodeID uint64, peerURL string) error {
-	return c.proposeConfChange(ctx, &raftpb.ConfChangeSingle{Type: raftpb.ConfChangeAddNode.Enum(), NodeId: new(nodeID)}, peerURL)
+	return c.proposeConfChange(ctx, &raftpb.ConfChangeSingle{Type: raftpb.ConfChangeAddNode.Enum(), NodeId: new(nodeID)}, confChangeContext{member: member{URL: peerURL}})
 }
 
 // RemoveNode proposes removing nodeID from the Raft group. Its ID is retired and can never be reused.
 // It blocks until the change is applied or ctx is cancelled.
 func (c *Consensus) RemoveNode(ctx context.Context, nodeID uint64) error {
-	return c.proposeConfChange(ctx, &raftpb.ConfChangeSingle{Type: raftpb.ConfChangeRemoveNode.Enum(), NodeId: new(nodeID)}, "")
+	return c.proposeConfChange(ctx, &raftpb.ConfChangeSingle{Type: raftpb.ConfChangeRemoveNode.Enum(), NodeId: new(nodeID)}, confChangeContext{})
 }
 
 // proposeConfChange proposes a single membership change and blocks until it is applied or ctx is cancelled.
 // Note that Raft only allows one config change in flight: a change proposed while another one is pending
 // is silently dropped by the leader, in which case this blocks until ctx is cancelled.
-func (c *Consensus) proposeConfChange(ctx context.Context, change *raftpb.ConfChangeSingle, peerURL string) error {
-	err := c.membership.validateConfChange(change, peerURL)
+func (c *Consensus) proposeConfChange(ctx context.Context, change *raftpb.ConfChangeSingle, cctx confChangeContext) error {
+	err := c.membership.validateConfChange(change, cctx.member)
 	if err != nil {
 		return err
 	}
 
-	cctx := confChangeContext{
-		ID:  uuid.New().String(),
-		URL: peerURL,
-	}
+	cctx.ID = uuid.New().String()
 
 	ctxBuf, err := json.Marshal(cctx)
 	if err != nil {
@@ -598,7 +596,7 @@ func (c *Consensus) applyConfigChangeV2(data []byte) error {
 	if len(cc.GetChanges()) != 1 {
 		err = stacktrace.NewErrorWithCode(dsserr.BadRequest, "expected exactly one change, got %d", len(cc.GetChanges()))
 	} else {
-		err = c.membership.validateConfChange(cc.GetChanges()[0], cctx.URL)
+		err = c.membership.validateConfChange(cc.GetChanges()[0], cctx.member)
 	}
 	if err != nil {
 		c.logger.Warn("rejecting config change", zap.String("id", cctx.ID), zap.Error(err))
@@ -642,10 +640,10 @@ func (c *Consensus) updateSnapshotConfState(msgs []*raftpb.Message) {
 // so that a node catching up via a snapshot fully recovers it, including which IDs must never be reused.
 // TODO: Use protobuf instead of gob as a follow-up to https://github.com/interuss/dss/pull/1765
 type snapshot struct {
-	// Members maps the ID of every member of the group to its description.
+	// Members maps the ID of every member of the group to its membership properties.
 	Members map[uint64]member
-	// RemovedIDs are the IDs of the nodes permanently removed from the group.
-	RemovedIDs []uint64
+	// Removed maps the ID of every node permanently removed from the group to its membership properties when it was removed.
+	Removed map[uint64]member
 	// AppData is the application snapshot data.
 	AppData []byte
 }
@@ -659,8 +657,8 @@ func (c *Consensus) encodeSnapshot() ([]byte, error) {
 	}
 
 	env := snapshot{
-		Members:    c.membership.members,
-		RemovedIDs: c.membership.removedIDs,
+		Members: c.membership.members,
+		Removed: c.membership.removed,
 		// TODO: Avoid encoding the data twice (to be addressed as part of https://github.com/interuss/dss/pull/1765)
 		AppData: appData,
 	}

pkg/raftstore/consensus/membership.go

@@ -1,7 +1,6 @@
 package consensus
 
 import (
-	"slices"
 	"sync"
 
 	dsserr "github.com/interuss/dss/pkg/errors"
@@ -26,15 +25,16 @@ type membership struct {
 
 	// members maps the ID of every member of the group to its description.
 	members map[uint64]member
-	// removedIDs are the IDs of the nodes removed from the group
-	// removed IDs can never be reused.
-	removedIDs []uint64
+	// removed maps the ID of every node removed from the group to its description when it was removed.
+	// Removed IDs can never be reused.
+	removed map[uint64]member
 }
 
 func (m *membership) isIDRemoved(id uint64) bool {
 	m.RLock()
 	defer m.RUnlock()
-	return slices.Contains(m.removedIDs, id)
+	_, ok := m.removed[id]
+	return ok
 }
 
 // addNode records nodeID as a member described by mem and adds it as a peer of the transport.
@@ -48,38 +48,42 @@ func (m *membership) addNode(nodeID uint64, mem member) {
 	}
 }
 
-// removeNode removes nodeID from the members, retires its ID and removes it from the peers of the transport.
+// removeNode moves nodeID from the members to the removed nodes, which retires its ID, and removes it from the peers
+// of the transport.
 func (m *membership) removeNode(nodeID uint64) {
 	m.Lock()
+	m.removed[nodeID] = m.members[nodeID]
 	delete(m.members, nodeID)
-	m.removedIDs = append(m.removedIDs, nodeID)
 	m.Unlock()
 
 	if nodeID != m.nodeID {
 		m.transport.RemovePeer(types.ID(nodeID))
 	}
 }
 
-// restore replaces the members and the removed IDs and updates the transport peers accordingly.
-func (m *membership) restore(members map[uint64]member, removedIDs []uint64) error {
-	for _, id := range removedIDs {
+// restore replaces the members and the removed nodes and updates the transport peers accordingly.
+func (m *membership) restore(members map[uint64]member, removed map[uint64]member) error {
+	for id := range removed {
 		if _, ok := members[id]; ok {
 			return stacktrace.NewError("removed ID %d is still present in the members map", id)
 		}
 	}
 	if members == nil {
 		members = map[uint64]member{}
 	}
+	if removed == nil {
+		removed = map[uint64]member{}
+	}
 
 	m.Lock()
 	m.members = members
-	m.removedIDs = removedIDs
+	m.removed = removed
 	m.Unlock()
 
 	// It is safe to do this outside the lock because the transport methods as well as addNode and removeNode
 	// are only called by the consumer goroutine. So all transport updates happen one at a time. The lock only protects
 	// the readers on other goroutine (isIDRemoved, validateConfChange etc.) who never call the transport methods.
-	for _, id := range removedIDs {
+	for id := range removed {
 		m.transport.RemovePeer(types.ID(id))
 	}
 	for id, mem := range members {
@@ -91,19 +95,21 @@ func (m *membership) restore(members map[uint64]member, removedIDs []uint64) err
 }
 
 // validateConfChange checks that applying change keeps the membership valid.
+// When adding a node, mem describes it.
 // NB: This method must remain deterministic as it is called by every node when the change is applied.
 // It uses members and not the ConfState, which can be more recent than the members after restoring
 // a snapshot received from the leader (see Consensus.updateSnapshotConfState).
-// A rejected change is reported as a dsserr.BadRequest error.
-func (m *membership) validateConfChange(change *raftpb.ConfChangeSingle, peerURL string) error {
+// A rejected change is reported as a dsserr.BadRequest error, or a dsserr.AlreadyExists error if the URL of the node
+// is already used by a member.
+func (m *membership) validateConfChange(change *raftpb.ConfChangeSingle, mem member) error {
 	m.RLock()
 	defer m.RUnlock()
 
 	if change.GetNodeId() == raft.None {
 		return stacktrace.NewErrorWithCode(dsserr.BadRequest, "node ID %d is invalid", change.GetNodeId())
 	}
 
-	if slices.Contains(m.removedIDs, change.GetNodeId()) {
+	if _, ok := m.removed[change.GetNodeId()]; ok {
 		return stacktrace.NewErrorWithCode(dsserr.BadRequest, "node ID %d was removed from the group and cannot be reused", change.GetNodeId())
 	}
 
@@ -113,8 +119,13 @@ func (m *membership) validateConfChange(change *raftpb.ConfChangeSingle, peerURL
 		if isMember {
 			return stacktrace.NewErrorWithCode(dsserr.BadRequest, "node %d is already a member of the group", change.GetNodeId())
 		}
-		if _, err := types.NewURLs([]string{peerURL}); err != nil {
-			return stacktrace.NewErrorWithCode(dsserr.BadRequest, "invalid URL %q for node %d: %v", peerURL, change.GetNodeId(), err)
+		if _, err := types.NewURLs([]string{mem.URL}); err != nil {
+			return stacktrace.NewErrorWithCode(dsserr.BadRequest, "invalid URL %q for node %d: %v", mem.URL, change.GetNodeId(), err)
+		}
+		for id, other := range m.members {
+			if other.URL == mem.URL {
+				return stacktrace.NewErrorWithCode(dsserr.AlreadyExists, "the URL %q of node %d is already used by member %d", mem.URL, change.GetNodeId(), id)
+			}
 		}
 
 	case raftpb.ConfChangeRemoveNode:

pkg/raftstore/consensus/membership_test.go

@@ -13,27 +13,32 @@ import (
 
 func TestValidateConfChange(t *testing.T) {
 	voters := []uint64{1, 2, 3, 4}
-	removed := []uint64{5}
+	removed := map[uint64]member{5: {URL: "http://node5:9011"}}
 	add := func(id uint64) *raftpb.ConfChangeSingle {
 		return &raftpb.ConfChangeSingle{Type: raftpb.ConfChangeAddNode.Enum(), NodeId: new(id)}
 	}
 	remove := func(id uint64) *raftpb.ConfChangeSingle {
 		return &raftpb.ConfChangeSingle{Type: raftpb.ConfChangeRemoveNode.Enum(), NodeId: new(id)}
 	}
 
+	node6 := member{URL: "http://node6:9011"}
+
 	tests := []struct {
-		name    string
-		change  *raftpb.ConfChangeSingle
-		url     string
-		voters  []uint64
-		wantErr bool
+		name     string
+		change   *raftpb.ConfChangeSingle
+		node     member
+		voters   []uint64
+		wantErr  bool
+		wantCode stacktrace.ErrorCode
 	}{
-		{name: "add new node", change: add(6), url: "http://node6:9011", voters: voters},
-		{name: "add invalid ID", change: add(raft.None), url: "http://node6:9011", voters: voters, wantErr: true},
-		{name: "add removed ID", change: add(5), url: "http://node5:9011", voters: voters, wantErr: true},
-		{name: "add existing member", change: add(2), url: "http://node2:9011", voters: voters, wantErr: true},
-		{name: "add without URL", change: add(6), voters: voters, wantErr: true},
-		{name: "add with URL without port", change: add(6), url: "http://node6", voters: voters, wantErr: true},
+		{name: "add new node", change: add(6), node: node6, voters: voters},
+		{name: "add invalid ID", change: add(raft.None), node: node6, voters: voters, wantErr: true},
+		{name: "add removed ID", change: add(5), node: member{URL: "http://node5:9011"}, voters: voters, wantErr: true},
+		{name: "add existing member", change: add(2), node: member{URL: "http://node2:9011"}, voters: voters, wantErr: true},
+		{name: "add without URL", change: add(6), node: member{}, voters: voters, wantErr: true},
+		{name: "add with URL without port", change: add(6), node: member{URL: "http://node6"}, voters: voters, wantErr: true},
+		{name: "add with the URL of a member", change: add(6), node: member{URL: "http://node2:9011"}, voters: voters, wantErr: true, wantCode: dsserr.AlreadyExists},
+		{name: "add with the URL of a removed node", change: add(6), node: member{URL: "http://node5:9011"}, voters: voters},
 		{name: "remove member", change: remove(4), voters: voters},
 		{name: "remove removed ID", change: remove(5), voters: voters, wantErr: true},
 		{name: "remove non member", change: remove(6), voters: voters, wantErr: true},
@@ -46,13 +51,17 @@ func TestValidateConfChange(t *testing.T) {
 			for _, id := range tt.voters {
 				members[id] = member{URL: fmt.Sprintf("http://node%d:9011", id)}
 			}
-			m := &membership{members: members, removedIDs: removed}
-			err := m.validateConfChange(tt.change, tt.url)
+			m := &membership{members: members, removed: removed}
+			err := m.validateConfChange(tt.change, tt.node)
 			if !tt.wantErr {
 				require.NoError(t, err)
 				return
 			}
-			require.Equal(t, dsserr.BadRequest, stacktrace.GetCode(err))
+			wantCode := dsserr.BadRequest
+			if tt.wantCode != 0 {
+				wantCode = tt.wantCode
+			}
+			require.Equal(t, wantCode, stacktrace.GetCode(err))
 		})
 	}
 }