[FIXED] Stream peers drift after server peer-remove (#6720)

If a server is peer-removed, a new peer will be selected if the server
hosted a stream. That new peer would be given the new peer set by the
meta layer, but would then reset back to the old peer set as part of the
stream's log.

This is a short-term fix, proposing the new peer set through the
stream's log to ensure all replicas agree on it.

Later we'll need to rework some parts to ensure the meta layer only
'suggests' new changes, but the proposing actually goes through the
stream/consumer's log instead of meta's log.

Signed-off-by: Maurice van Veen <[email protected]>
Co-authored-by: Neil Twigg <[email protected]>

Author: neilalexander

server/jetstream_cluster.go

@@ -2037,7 +2037,24 @@ retry:
 			samePeers = slices.Equal(groupPeerIDs, nodePeerIDs)
 		}
 		if !samePeers {
+			// At this point we have no way of knowing:
+			// 1. Whether the group has lost enough nodes to cause a quorum
+			//    loss, in which case a proposal may fail, therefore we will
+			//    force a peerstate write;
+			// 2. Whether nodes in the group have other applies queued up
+			//    that could change the peerstate again, therefore the leader
+			//    should send out a new proposal anyway too just to make sure
+			//    that this change gets captured in the log.
 			node.UpdateKnownPeers(groupPeerIDs)
+
+			// If the peers changed as a result of an update by the meta layer, we must reflect that in the log of
+			// this group. Otherwise, a new peer would come up and instantly reset the peer state back to whatever is
+			// in the log at that time, overwriting what the meta layer told it.
+			// Will need to address this properly later on, by for example having the meta layer decide the new
+			// placement, but have the leader of this group propose it through its own log instead.
+			if node.Leader() {
+				node.ProposeKnownPeers(groupPeerIDs)
+			}
 		}
 		rg.node = node
 		js.mu.Unlock()

server/jetstream_cluster_4_test.go

@@ -5088,3 +5088,109 @@ func TestJetStreamClusterTTLAndDedupe(t *testing.T) {
 	_, err = js.PublishMsg(m)
 	require_NoError(t, err)
 }
+
+func TestJetStreamClusterServerPeerRemovePeersDrift(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R4S", 4)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:      "TEST",
+		Retention: nats.LimitsPolicy,
+		Subjects:  []string{"foo"},
+		Replicas:  3,
+	})
+	require_NoError(t, err)
+
+	var acc *Account
+	var mset *stream
+
+	// Wait for 3 of the 4 servers to have created the stream.
+	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
+		count := 0
+		for _, s := range c.servers {
+			acc, err = s.lookupAccount(globalAccountName)
+			if err != nil {
+				return err
+			}
+			_, err = acc.lookupStream("TEST")
+			if err != nil {
+				continue
+			}
+			count++
+		}
+		if count != 3 {
+			return fmt.Errorf("expected 3 streams, got: %d", count)
+		}
+		return nil
+	})
+
+	sl := c.streamLeader(globalAccountName, "TEST")
+
+	// Get a random server that:
+	// - is not stream leader
+	// - is not meta leader (peer-removing the meta leader has other issues)
+	// - already hosts the stream so a peer-remove results in changing the stream peer set
+	var rs *Server
+	for _, s := range c.servers {
+		acc, err = s.lookupAccount(globalAccountName)
+		require_NoError(t, err)
+		_, err = acc.lookupStream("TEST")
+		if s == sl || s.isMetaLeader.Load() || err != nil {
+			continue
+		}
+		rs = s
+		break
+	}
+	if rs == nil {
+		t.Fatal("No server found that's not either stream or meta leader.")
+	}
+	rs.Shutdown()
+
+	// Peer-remove the selected server so the stream moves to the remaining empty server.
+	sc, err := nats.Connect(sl.ClientURL(), nats.UserInfo("admin", "s3cr3t!"))
+	require_NoError(t, err)
+	req := &JSApiMetaServerRemoveRequest{Server: rs.Name()}
+	jsreq, err := json.Marshal(req)
+	require_NoError(t, err)
+	_, err = sc.Request(JSApiRemoveServer, jsreq, time.Second)
+	require_NoError(t, err)
+
+	// Eventually there should again be a R3 stream and everyone should agree on the peers.
+	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
+		count := 0
+		var ps []string
+		for _, s := range c.servers {
+			if s == rs {
+				continue
+			}
+			acc, err = s.lookupAccount(globalAccountName)
+			if err != nil {
+				return err
+			}
+			mset, err = acc.lookupStream("TEST")
+			if err != nil {
+				return err
+			}
+			rn := mset.raftNode().(*raft)
+			rn.RLock()
+			peerNames := rn.peerNames()
+			rn.RUnlock()
+			slices.Sort(peerNames)
+			if count == 0 {
+				ps = peerNames
+			} else if !slices.Equal(ps, peerNames) {
+				rsid := rs.NodeName()
+				containsOld := slices.Contains(ps, rsid) || slices.Contains(peerNames, rsid)
+				return fmt.Errorf("no equal peers, expected: %v, got: %v, contains old peer (%s): %v", ps, peerNames, rsid, containsOld)
+			}
+			count++
+		}
+		if count != 3 {
+			return fmt.Errorf("expected 3 servers hosting stream, got: %d", count)
+		}
+		return nil
+	})
+}