[FIXED] Disallow concurrent meta member changes

Signed-off-by: Maurice van Veen <[email protected]>

Author: MauriceVanVeen

server/errors.json

@@ -1998,5 +1998,15 @@
     "help": "",
     "url": "",
     "deprecates": ""
+  },
+  {
+    "constant": "JSClusterServerMemberChangeInflightErr",
+    "code": 400,
+    "error_code": 10202,
+    "description": "cluster member change is in progress",
+    "comment": "",
+    "help": "",
+    "url": "",
+    "deprecates": ""
   }
 ]

server/jetstream_api.go

@@ -2399,6 +2399,13 @@ func (s *Server) jsLeaderServerRemoveRequest(sub *subscription, c *client, _ *Ac
 	js.mu.Lock()
 	defer js.mu.Unlock()
 
+	// Another peer-remove is already in progress, don't allow multiple concurrent changes.
+	if cc.peerRemoveReply != nil {
+		resp.Error = NewJSClusterServerMemberChangeInflightError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
 	var found string
 	for _, p := range meta.Peers() {
 		// If Peer is specified, it takes precedence

server/jetstream_cluster_1_test.go

@@ -11127,7 +11127,8 @@ func TestJetStreamClusterMetaPeerRemoveResponseAfterQuorum(t *testing.T) {
 	}
 
 	// Shutdown a majority.
-	remove := s3.Name()
+	remove1 := s3.Name()
+	remove2 := s2.Name()
 	s1.Shutdown()
 	s2.Shutdown()
 	s3.Shutdown()
@@ -11138,21 +11139,35 @@ func TestJetStreamClusterMetaPeerRemoveResponseAfterQuorum(t *testing.T) {
 	defer sub.Drain()
 
 	// Since a majority is down, we'll expect to time out since there's no quorum.
-	req := &JSApiMetaServerRemoveRequest{Server: remove}
+	req := &JSApiMetaServerRemoveRequest{Server: remove1}
 	jsreq, err := json.Marshal(req)
 	require_NoError(t, err)
 	require_NoError(t, nc.PublishRequest(JSApiRemoveServer, reply, jsreq))
 	_, err = sub.NextMsg(time.Second)
 	require_Error(t, err, nats.ErrTimeout)
 
+	// Retrying the request should fail, as the leader has already removed it as its peer.
 	rmsg, err := nc.Request(JSApiRemoveServer, jsreq, time.Second)
 	require_NoError(t, err)
 	var resp JSApiMetaServerRemoveResponse
 	require_NoError(t, json.Unmarshal(rmsg.Data, &resp))
 	if resp.Error == nil {
 		t.Fatalf("Expected an error, got none")
 	}
-	require_Error(t, resp.Error, NewJSClusterServerNotMemberError())
+	require_Error(t, resp.Error, NewJSClusterServerMemberChangeInflightError())
+
+	// Don't allow concurrent meta membership changes.
+	req = &JSApiMetaServerRemoveRequest{Server: remove2}
+	jsreq, err = json.Marshal(req)
+	require_NoError(t, err)
+	rmsg, err = nc.Request(JSApiRemoveServer, jsreq, time.Second)
+	require_NoError(t, err)
+	resp = JSApiMetaServerRemoveResponse{}
+	require_NoError(t, json.Unmarshal(rmsg.Data, &resp))
+	if resp.Error == nil {
+		t.Fatalf("Expected an error, got none")
+	}
+	require_Error(t, resp.Error, NewJSClusterServerMemberChangeInflightError())
 
 	// Bring back one server so that the peer-remove can get quorum now.
 	// The response should come in shortly after.
@@ -11169,6 +11184,20 @@ func TestJetStreamClusterMetaPeerRemoveResponseAfterQuorum(t *testing.T) {
 		require_NoError(t, resp.Error)
 	}
 	require_True(t, resp.Success)
+
+	// A retry of the first peer-remove should return an error that the peer is already removed.
+	// Both a "success" response and this error should be used to know the peer-remove was successful.
+	req = &JSApiMetaServerRemoveRequest{Server: remove1}
+	jsreq, err = json.Marshal(req)
+	require_NoError(t, err)
+	rmsg, err = nc.Request(JSApiRemoveServer, jsreq, time.Second)
+	require_NoError(t, err)
+	resp = JSApiMetaServerRemoveResponse{}
+	require_NoError(t, json.Unmarshal(rmsg.Data, &resp))
+	if resp.Error == nil {
+		t.Fatalf("Expected an error, got none")
+	}
+	require_Error(t, resp.Error, NewJSClusterServerNotMemberError())
 }
 
 //

server/jetstream_errors_generated.go

@@ -59,6 +59,9 @@ const (
 	// JSClusterRequiredErr JetStream clustering support required
 	JSClusterRequiredErr ErrorIdentifier = 10010
 
+	// JSClusterServerMemberChangeInflightErr cluster member change is in progress
+	JSClusterServerMemberChangeInflightErr ErrorIdentifier = 10202
+
 	// JSClusterServerNotMemberErr server is not a member of the cluster
 	JSClusterServerNotMemberErr ErrorIdentifier = 10044
 
@@ -626,6 +629,7 @@ var (
 		JSClusterNotLeaderErr:                        {Code: 500, ErrCode: 10009, Description: "JetStream cluster can not handle request"},
 		JSClusterPeerNotMemberErr:                    {Code: 400, ErrCode: 10040, Description: "peer not a member"},
 		JSClusterRequiredErr:                         {Code: 503, ErrCode: 10010, Description: "JetStream clustering support required"},
+		JSClusterServerMemberChangeInflightErr:       {Code: 400, ErrCode: 10202, Description: "cluster member change is in progress"},
 		JSClusterServerNotMemberErr:                  {Code: 400, ErrCode: 10044, Description: "server is not a member of the cluster"},
 		JSClusterTagsErr:                             {Code: 400, ErrCode: 10011, Description: "tags placement not supported for operation"},
 		JSClusterUnSupportFeatureErr:                 {Code: 503, ErrCode: 10036, Description: "not currently supported in clustered mode"},
@@ -1031,6 +1035,16 @@ func NewJSClusterRequiredError(opts ...ErrorOption) *ApiError {
 	return ApiErrors[JSClusterRequiredErr]
 }
 
+// NewJSClusterServerMemberChangeInflightError creates a new JSClusterServerMemberChangeInflightErr error: "cluster member change is in progress"
+func NewJSClusterServerMemberChangeInflightError(opts ...ErrorOption) *ApiError {
+	eopts := parseOpts(opts)
+	if ae, ok := eopts.err.(*ApiError); ok {
+		return ae
+	}
+
+	return ApiErrors[JSClusterServerMemberChangeInflightErr]
+}
+
 // NewJSClusterServerNotMemberError creates a new JSClusterServerNotMemberErr error: "server is not a member of the cluster"
 func NewJSClusterServerNotMemberError(opts ...ErrorOption) *ApiError {
 	eopts := parseOpts(opts)