Repository: nats-io/nats-server

Issue:
JetStream: allow replicas > current cluster size as long as there is quorum
## Feature Request
Allow replicas number greater than current cluster size as long as there is quorum

#### Use Case:
Say there is a cluster of 3 nodes, if one creates a stream with a replicas of 3, it works.
Then one server is shutdown, there is still quorum and this stream can be used.
However, trying to create another stream with replicas(3) would fail with "insufficient resources".
As shown above a stream with R(3) still can function as long as there is quorum, so it would be nice to be able to create the stream and use it even if there is currently not enough servers to match the replicas count, but there is still quorum.

#### Proposed Change:
Allow creation of assets with a replicas > current cluster size as long that there is quorum.

#### Who Benefits From The Change(s)?
MQTT feature inside the NATS Server creates a stream for every new session (or clean session). If it is determined that the cluster is of size 3 for instance, the assets are created with this replicas number. Which means that it does not work if a server is not running. Reducing the replicas count to quorum size is an issue too, for instance in this case we would create with R(2), but then if one of the server where the asset is stored goes down, then MQTT is unusable.

#### Alternative Approaches
When creating assets, if we get "insufficient resources", we can reduce the replicas count, but then that brings other issues that now this asset may be unusable if the other server is restarted but one of the server hosting the R(2) asset goes down.


Discussion:
@ripienaar: This sounds like a bug to me rather than freature request.  Running with 2 servers in a previously 3 node cluster should not impact anything.@kozlovic: What I mean is that if one tries to *create* a stream with R3 when only 2 servers are running (but there is quorum because it was originally a 3 nodes cluster). Are you saying that in that situation it should still be working? I would like it to work this way for sure, which that is why I describe that it makes a difference in behavior just based on when the stream is created.@ripienaar: Yeah, I am 100% saying that should be working today and if it does not, thats a bug not a new feature.@derekcollison: We should be a bit careful here. I generally agree we can do better, but I am also concerned around scenarios like a 3 node cluster and being asked to create a R=5 stream. 

What I think might work is that if we have seen in the past that the cluster size was N, then anything up to N (sans limit of 5) is ok. But if you ask to create a stream with R=5 on a properly constructed cluster of just three nodes that should fail IMO.

@kozlovic: > What I think might work is that if we have seen in the past that the cluster size was N, then anything up to N (sans limit of 5) is ok. But if you ask to create a stream with R=5 on a properly constructed cluster of just three nodes that should fail IMO.

Would agree on that, but in our past conversation, I suggested it, but you said that we can't know what the cluster size was. I guess I miscommunicated the intent. But yes, I would think that this is a fair limitation.

Solution patch:
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 53c24d18..64b92b8a 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -5853,6 +5853,7 @@ func (cc *jetStreamCluster) selectPeerGroup(r int, cluster string, cfg *StreamCo
 	type wn struct {
 		id    string
 		avail uint64
+		off   bool
 		ha    int
 		ns    int
 	}
@@ -5941,6 +5942,8 @@ func (cc *jetStreamCluster) selectPeerGroup(r int, cluster string, cfg *StreamCo
 	// Which is why we keep taps on how often which one happened.
 	err := selectPeerError{}
 
+	var onlinePeers int
+
 	// Shuffle them up.
 	rand.Shuffle(len(peers), func(i, j int) { peers[i], peers[j] = peers[j], peers[i] })
 	for _, p := range peers {
@@ -5956,8 +5959,8 @@ func (cc *jetStreamCluster) selectPeerGroup(r int, cluster string, cfg *StreamCo
 			continue
 		}
 
-		// If we know its offline or we do not have config or err don't consider.
-		if ni.offline || ni.cfg == nil || ni.stats == nil {
+		// If we've never heard from a server, don't consider.
+		if ni.cfg == nil || ni.stats == nil {
 			s.Debugf("Peer selection: discard %s@%s reason: offline", ni.name, ni.cluster)
 			err.offline = true
 			continue
@@ -6054,20 +6057,35 @@ func (cc *jetStreamCluster) selectPeerGroup(r int, cluster string, cfg *StreamCo
 			}
 		}
 		// Add to our list of potential nodes.
-		nodes = append(nodes, wn{p.ID, available, peerHA[p.ID], peerStreams[p.ID]})
+		nodes = append(nodes, wn{p.ID, available, ni.offline, peerHA[p.ID], peerStreams[p.ID]})
+		if !ni.offline {
+			onlinePeers++
+		}
 	}
 
 	// If we could not select enough peers, fail.
-	if len(nodes) < (r - len(existing)) {
-		s.Debugf("Peer selection: required %d nodes but found %d (cluster: %s replica: %d existing: %v/%d peers: %d result-peers: %d err: %+v)",
-			(r - len(existing)), len(nodes), cluster, r, existing, replaceFirstExisting, len(peers), len(nodes), err)
+	quorum := r/2 + 1
+	missingQuorum := onlinePeers+len(existing) < quorum
+	if missingNodes := len(nodes) < (r - len(existing)); missingNodes || missingQuorum {
 		if len(peers) == 0 {
 			err.noJsClust = true
+		} else if !missingNodes && missingQuorum {
+			err.offline = true
 		}
+		s.Debugf("Peer selection: required %d nodes but found %d (cluster: %s replica: %d existing: %v/%d peers: %d result-peers: %d err: %+v)",
+			r-len(existing), len(nodes), cluster, r, existing, replaceFirstExisting, len(peers), len(nodes), err)
 		return nil, &err
 	}
 	// Sort based on available from most to least, breaking ties by number of total streams assigned to the peer.
 	slices.SortFunc(nodes, func(i, j wn) int {
+		// Prefer online servers to offline ones.
+		if i.off != j.off {
+			if i.off {
+				return 1
+			} else {
+				return -1
+			}
+		}
 		if i.avail == j.avail {
 			return cmp.Compare(i.ns, j.ns)
 		}
@@ -6075,7 +6093,17 @@ func (cc *jetStreamCluster) selectPeerGroup(r int, cluster string, cfg *StreamCo
 	})
 	// If we are placing a replicated stream, let's sort based on HAAssets, as that is more important to balance.
 	if cfg.Replicas > 1 {
-		slices.SortStableFunc(nodes, func(i, j wn) int { return cmp.Compare(i.ha, j.ha) })
+		slices.SortStableFunc(nodes, func(i, j wn) int {
+			// Prefer online servers to offline ones.
+			if i.off != j.off {
+				if i.off {
+					return 1
+				} else {
+					return -1
+				}
+			}
+			return cmp.Compare(i.ha, j.ha)
+		})
 	}
 
 	var results []string



Test patch:
diff --git a/server/jetstream_cluster_1_test.go b/server/jetstream_cluster_1_test.go
index fc666828..966cb79b 100644
--- a/server/jetstream_cluster_1_test.go
+++ b/server/jetstream_cluster_1_test.go
@@ -7032,6 +7032,9 @@ func TestJetStreamClusterConsumerInfoAfterCreate(t *testing.T) {
 		require_NoError(t, json.Unmarshal(resp.Data, &sdr))
 	}
 
+	// Wait some time for the leader to settle.
+	time.Sleep(500 * time.Millisecond)
+
 	// Scale down to ensure the consumer gets created on this server.
 	cfg.Replicas = 1
 	si, err = js.UpdateStream(cfg)
@@ -8924,6 +8927,53 @@ func TestJetStreamClusterDontReviveRemovedStream(t *testing.T) {
 	require_Error(t, err, ErrJetStreamStreamNotFound)
 }
 
+func TestJetStreamClusterCreateR3StreamWithOfflineNodes(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	ml := c.leader()
+	nc, js := jsClientConnect(t, ml)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{Name: "FOO", Replicas: 3})
+	require_NoError(t, err)
+
+	// Shutdown one server, we'll still have quorum.
+	rs1 := c.randomNonLeader()
+	rs1.Shutdown()
+	_, err = js.AddStream(&nats.StreamConfig{Name: "BAR", Replicas: 3})
+	require_NoError(t, err)
+
+	// While we still have quorum, publish a few messages to the streams.
+	// The offline server will need to both assign the stream and catchup when it comes back.
+	for _, subj := range []string{"FOO", "BAR"} {
+		_, err = js.Publish(subj, nil)
+		require_NoError(t, err)
+	}
+
+	// Shutdown one more server, only one server left that can't reach quorum on its own.
+	rs2 := c.randomNonLeader()
+	rs2.Shutdown()
+	_, err = js.AddStream(&nats.StreamConfig{Name: "BAZ", Replicas: 3})
+	require_Contains(t, err.Error(), "no suitable peers for placement", "peer offline")
+
+	// Restart offline servers and ensure all comes back up.
+	c.restartServer(rs1)
+	c.restartServer(rs2)
+	c.waitOnLeader()
+	c.waitOnStreamLeader(globalAccountName, "FOO")
+	c.waitOnStreamLeader(globalAccountName, "BAR")
+	checkFor(t, 2*time.Second, 200*time.Millisecond, func() error {
+		if err := checkState(t, c, globalAccountName, "FOO"); err != nil {
+			return err
+		}
+		if err := checkState(t, c, globalAccountName, "BAR"); err != nil {
+			return err
+		}
+		return nil
+	})
+}
+
 //
 // DO NOT ADD NEW TESTS IN THIS FILE (unless to balance test times)
 // Add at the end of jetstream_cluster_<n>_test.go, with <n> being the highest value.