Repository: nats-io/nats-server
Pull request #3539 (2022-10-10 19:32:23)

[FIXED] JetStream: return error on negative replicas count
If a stream is created or updated with a negative replicas count, and error is now returned. Same for consumers.

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>



Patch:
diff --git a/server/consumer.go b/server/consumer.go
index a123035a..e91a8429 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -369,6 +369,10 @@ func checkConsumerCfg(
 	if config.Replicas > 0 && config.Replicas > cfg.Replicas {
 		return NewJSConsumerReplicasExceedsStreamError()
 	}
+	// Check that it is not negative
+	if config.Replicas < 0 {
+		return NewJSReplicasCountCannotBeNegativeError()
+	}
 
 	// Check if we have a BackOff defined that MaxDeliver is within range etc.
 	if lbo := len(config.BackOff); lbo > 0 && config.MaxDeliver <= lbo {
diff --git a/server/errors.json b/server/errors.json
index 2df7d6bf..2413531b 100644
--- a/server/errors.json
+++ b/server/errors.json
@@ -1308,5 +1308,15 @@
     "help": "",
     "url": "",
     "deprecates": ""
+  },
+  {
+    "constant": "JSReplicasCountCannotBeNegative",
+    "code": 400,
+    "error_code": 10133,
+    "description": "replicas count cannot be negative",
+    "comment": "",
+    "help": "",
+    "url": "",
+    "deprecates": ""
   }
 ]
\ No newline at end of file
diff --git a/server/jetstream_errors_generated.go b/server/jetstream_errors_generated.go
index 86fa5bf5..8cf409c3 100644
--- a/server/jetstream_errors_generated.go
+++ b/server/jetstream_errors_generated.go
@@ -245,6 +245,9 @@ const (
 	// JSRaftGeneralErrF General RAFT error string ({err})
 	JSRaftGeneralErrF ErrorIdentifier = 10041
 
+	// JSReplicasCountCannotBeNegative replicas count cannot be negative
+	JSReplicasCountCannotBeNegative ErrorIdentifier = 10133
+
 	// JSRestoreSubscribeFailedErrF JetStream unable to subscribe to restore snapshot {subject}: {err}
 	JSRestoreSubscribeFailedErrF ErrorIdentifier = 10042
 
@@ -481,6 +484,7 @@ var (
 		JSNotEnabledForAccountErr:                  {Code: 503, ErrCode: 10039, Description: "JetStream not enabled for account"},
 		JSPeerRemapErr:                             {Code: 503, ErrCode: 10075, Description: "peer remap failed"},
 		JSRaftGeneralErrF:                          {Code: 500, ErrCode: 10041, Description: "{err}"},
+		JSReplicasCountCannotBeNegative:            {Code: 400, ErrCode: 10133, Description: "replicas count cannot be negative"},
 		JSRestoreSubscribeFailedErrF:               {Code: 500, ErrCode: 10042, Description: "JetStream unable to subscribe to restore snapshot {subject}: {err}"},
 		JSSequenceNotFoundErrF:                     {Code: 400, ErrCode: 10043, Description: "sequence {seq} not found"},
 		JSSnapshotDeliverSubjectInvalidErr:         {Code: 400, ErrCode: 10015, Description: "deliver subject not valid"},
@@ -1423,6 +1427,16 @@ func NewJSRaftGeneralError(err error, opts ...ErrorOption) *ApiError {
 	}
 }
 
+// NewJSReplicasCountCannotBeNegativeError creates a new JSReplicasCountCannotBeNegative error: "replicas count cannot be negative"
+func NewJSReplicasCountCannotBeNegativeError(opts ...ErrorOption) *ApiError {
+	eopts := parseOpts(opts)
+	if ae, ok := eopts.err.(*ApiError); ok {
+		return ae
+	}
+
+	return ApiErrors[JSReplicasCountCannotBeNegative]
+}
+
 // NewJSRestoreSubscribeFailedError creates a new JSRestoreSubscribeFailedErrF error: "JetStream unable to subscribe to restore snapshot {subject}: {err}"
 func NewJSRestoreSubscribeFailedError(err error, subject interface{}, opts ...ErrorOption) *ApiError {
 	eopts := parseOpts(opts)
diff --git a/server/stream.go b/server/stream.go
index d84de8fd..07eb29c0 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -947,6 +947,9 @@ func (s *Server) checkStreamCfg(config *StreamConfig, acc *Account) (StreamConfi
 	if cfg.Replicas > StreamMaxReplicas {
 		return cfg, NewJSStreamInvalidConfigError(fmt.Errorf("maximum replicas is %d", StreamMaxReplicas))
 	}
+	if cfg.Replicas < 0 {
+		return cfg, NewJSReplicasCountCannotBeNegativeError()
+	}
 	if cfg.MaxMsgs == 0 {
 		cfg.MaxMsgs = -1
 	}


Test patch:
diff --git a/server/jetstream_cluster_3_test.go b/server/jetstream_cluster_3_test.go
index 7f5dc38f..334aa9bb 100644
--- a/server/jetstream_cluster_3_test.go
+++ b/server/jetstream_cluster_3_test.go
@@ -460,3 +460,62 @@ func TestJetStreamClusterDeleteConsumerWhileServerDown(t *testing.T) {
 		t.Fatalf("Expected to not find consumer, but did")
 	}
 }
+
+func TestJetStreamClusterNegativeReplicas(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	if config := s.JetStreamConfig(); config != nil {
+		defer removeDir(t, config.StoreDir)
+	}
+	defer s.Shutdown()
+
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	testBadReplicas := func(t *testing.T, s *Server, name string) {
+		nc, js := jsClientConnect(t, s)
+		defer nc.Close()
+
+		_, err := js.AddStream(&nats.StreamConfig{
+			Name:     name,
+			Replicas: -1,
+		})
+		require_Error(t, err, NewJSReplicasCountCannotBeNegativeError())
+
+		_, err = js.AddStream(&nats.StreamConfig{
+			Name:     name,
+			Replicas: 1,
+		})
+		require_NoError(t, err)
+
+		// Check upadte now.
+		_, err = js.UpdateStream(&nats.StreamConfig{
+			Name:     name,
+			Replicas: -11,
+		})
+		require_Error(t, err, NewJSReplicasCountCannotBeNegativeError())
+
+		// Now same for consumers
+		durName := fmt.Sprintf("%s_dur", name)
+		_, err = js.AddConsumer(name, &nats.ConsumerConfig{
+			Durable:  durName,
+			Replicas: -1,
+		})
+		require_Error(t, err, NewJSReplicasCountCannotBeNegativeError())
+
+		_, err = js.AddConsumer(name, &nats.ConsumerConfig{
+			Durable:  durName,
+			Replicas: 1,
+		})
+		require_NoError(t, err)
+
+		// Check update now
+		_, err = js.UpdateConsumer(name, &nats.ConsumerConfig{
+			Durable:  durName,
+			Replicas: -11,
+		})
+		require_Error(t, err, NewJSReplicasCountCannotBeNegativeError())
+	}
+
+	t.Run("Standalone", func(t *testing.T) { testBadReplicas(t, s, "TEST1") })
+	t.Run("Clustered", func(t *testing.T) { testBadReplicas(t, c.randomServer(), "TEST2") })
+}