[IMPROVED] Meta peer-remove JS audit success response is not error

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

Author: MauriceVanVeen

server/jetstream_api.go

@@ -2432,9 +2432,10 @@ func (s *Server) jsLeaderServerRemoveRequest(sub *subscription, c *client, _ *Ac
 	// So we have a valid peer.
 	if err := meta.ProposeRemovePeer(found); err == nil {
 		if cc.peerRemoveReply == nil {
-			cc.peerRemoveReply = make(map[string]string, 1)
+			cc.peerRemoveReply = make(map[string]peerRemoveInfo, 1)
 		}
-		cc.peerRemoveReply[found] = reply
+		// Only copy the request, the subject and reply are already copied.
+		cc.peerRemoveReply[found] = peerRemoveInfo{ci: ci, subject: subject, reply: reply, request: string(msg)}
 	}
 }
 

server/jetstream_cluster.go

@@ -53,7 +53,7 @@ type jetStreamCluster struct {
 	inflight map[string]map[string]*inflightInfo
 	// Holds a map of a peer ID to the reply subject, to only respond after gaining
 	// quorum on the peer-remove action.
-	peerRemoveReply map[string]string
+	peerRemoveReply map[string]peerRemoveInfo
 	// Signals meta-leader should check the stream assignments.
 	streamsCheck bool
 	// Server.
@@ -87,6 +87,14 @@ type inflightInfo struct {
 	cfg  *StreamConfig
 }
 
+// Used to track inflight peer-remove info to respond 'success' after quorum.
+type peerRemoveInfo struct {
+	ci      *ClientInfo
+	subject string
+	reply   string
+	request string
+}
+
 // Used to guide placement of streams and meta controllers in clustered JetStream.
 type Placement struct {
 	Cluster   string   `json:"cluster,omitempty"`
@@ -2116,12 +2124,12 @@ func (js *jetStream) applyMetaEntries(entries []*Entry, ru *recoveryUpdates) (bo
 				s := js.srv
 				if s.JetStreamIsLeader() {
 					var (
-						reply string
-						ok    bool
+						info peerRemoveInfo
+						ok   bool
 					)
 					js.mu.Lock()
 					if cc := js.cluster; cc != nil && cc.peerRemoveReply != nil {
-						if reply, ok = cc.peerRemoveReply[peer]; ok {
+						if info, ok = cc.peerRemoveReply[peer]; ok {
 							delete(cc.peerRemoveReply, peer)
 						}
 						if len(cc.peerRemoveReply) == 0 {
@@ -2130,10 +2138,11 @@ func (js *jetStream) applyMetaEntries(entries []*Entry, ru *recoveryUpdates) (bo
 					}
 					js.mu.Unlock()
 
-					if reply != _EMPTY_ {
+					if info.reply != _EMPTY_ {
+						sysAcc := s.SystemAccount()
 						var resp = JSApiMetaServerRemoveResponse{ApiResponse: ApiResponse{Type: JSApiMetaServerRemoveResponseType}}
 						resp.Success = true
-						s.sendInternalAccountMsg(nil, reply, s.jsonResponse(&resp))
+						s.sendAPIResponse(info.ci, sysAcc, info.subject, info.reply, info.request, s.jsonResponse(&resp))
 					}
 				}
 			}

server/jetstream_cluster_1_test.go

@@ -11138,6 +11138,10 @@ func TestJetStreamClusterMetaPeerRemoveResponseAfterQuorum(t *testing.T) {
 	require_NoError(t, err)
 	defer sub.Drain()
 
+	auditSub, err := nc.SubscribeSync(JSAuditAdvisory)
+	require_NoError(t, err)
+	defer auditSub.Drain()
+
 	// Since a majority is down, we'll expect to time out since there's no quorum.
 	req := &JSApiMetaServerRemoveRequest{Server: remove1}
 	jsreq, err := json.Marshal(req)
@@ -11156,6 +11160,18 @@ func TestJetStreamClusterMetaPeerRemoveResponseAfterQuorum(t *testing.T) {
 	}
 	require_Error(t, resp.Error, NewJSClusterServerMemberChangeInflightError())
 
+	// Audit should reflect the same.
+	rmsg, err = auditSub.NextMsg(time.Second)
+	require_NoError(t, err)
+	var auditResp JSAPIAudit
+	require_NoError(t, json.Unmarshal(rmsg.Data, &auditResp))
+	resp = JSApiMetaServerRemoveResponse{}
+	require_NoError(t, json.Unmarshal([]byte(auditResp.Response), &resp))
+	if resp.Error == nil {
+		t.Fatalf("Expected an error, got none")
+	}
+	require_Error(t, resp.Error, NewJSClusterServerMemberChangeInflightError())
+
 	// Don't allow concurrent meta membership changes.
 	req = &JSApiMetaServerRemoveRequest{Server: remove2}
 	jsreq, err = json.Marshal(req)
@@ -11169,6 +11185,18 @@ func TestJetStreamClusterMetaPeerRemoveResponseAfterQuorum(t *testing.T) {
 	}
 	require_Error(t, resp.Error, NewJSClusterServerMemberChangeInflightError())
 
+	// Audit should reflect the same.
+	rmsg, err = auditSub.NextMsg(time.Second)
+	require_NoError(t, err)
+	auditResp = JSAPIAudit{}
+	require_NoError(t, json.Unmarshal(rmsg.Data, &auditResp))
+	resp = JSApiMetaServerRemoveResponse{}
+	require_NoError(t, json.Unmarshal([]byte(auditResp.Response), &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.
 	c.restartServer(s1)
@@ -11185,6 +11213,18 @@ func TestJetStreamClusterMetaPeerRemoveResponseAfterQuorum(t *testing.T) {
 	}
 	require_True(t, resp.Success)
 
+	// Audit should reflect the same.
+	rmsg, err = auditSub.NextMsg(time.Second)
+	require_NoError(t, err)
+	auditResp = JSAPIAudit{}
+	require_NoError(t, json.Unmarshal(rmsg.Data, &auditResp))
+	resp = JSApiMetaServerRemoveResponse{}
+	require_NoError(t, json.Unmarshal([]byte(auditResp.Response), &resp))
+	if resp.Error != nil {
+		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}
@@ -11198,6 +11238,18 @@ func TestJetStreamClusterMetaPeerRemoveResponseAfterQuorum(t *testing.T) {
 		t.Fatalf("Expected an error, got none")
 	}
 	require_Error(t, resp.Error, NewJSClusterServerNotMemberError())
+
+	// Audit should reflect the same.
+	rmsg, err = auditSub.NextMsg(time.Second)
+	require_NoError(t, err)
+	auditResp = JSAPIAudit{}
+	require_NoError(t, json.Unmarshal(rmsg.Data, &auditResp))
+	resp = JSApiMetaServerRemoveResponse{}
+	require_NoError(t, json.Unmarshal([]byte(auditResp.Response), &resp))
+	if resp.Error == nil {
+		t.Fatalf("Expected an error, got none")
+	}
+	require_Error(t, resp.Error, NewJSClusterServerNotMemberError())
 }
 
 //