rocketmq_protocol/protocol/body/
controller_rollout.rs

1// Copyright 2026 The RocketMQ Rust Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::collections::{BTreeMap, BTreeSet};
16
17use serde::{Deserialize, Serialize};
18
19/// Fresh voter observations returned only for an explicit Controller rollout check.
20/// Operators must recheck before each restart; this observation reserves no resources.
21#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
22#[serde(rename_all = "camelCase")]
23pub struct RolloutQuorumStatus {
24    pub schema_version: u32,
25    pub target_node_id: u64,
26    pub leader_id: u64,
27    pub voters: Vec<u64>,
28    /// Advertised Raft endpoints of current voters, used to identify the deployment.
29    pub peer_endpoints: BTreeMap<u64, String>,
30    pub remaining_ready_voters: Vec<u64>,
31    pub allowed: bool,
32}
33
34impl RolloutQuorumStatus {
35    /// Checks the response version, target, and remaining majority before an operator acts.
36    pub fn permits(&self, target: u64) -> bool {
37        let voters = self.voters.iter().copied().collect::<BTreeSet<_>>();
38        let remaining = self.remaining_ready_voters.iter().copied().collect::<BTreeSet<_>>();
39        self.schema_version == 1
40            && self.allowed
41            && target != 0
42            && self.target_node_id == target
43            && voters.len() >= 3
44            && voters.len() == self.voters.len()
45            && !voters.contains(&0)
46            && voters.contains(&target)
47            && voters.contains(&self.leader_id)
48            && self.peer_endpoints.keys().copied().collect::<BTreeSet<_>>() == voters
49            && self.peer_endpoints.values().all(|endpoint| !endpoint.is_empty())
50            && remaining.len() == self.remaining_ready_voters.len()
51            && !remaining.contains(&target)
52            && remaining.is_subset(&voters)
53            && remaining.len() > voters.len() / 2
54    }
55}
56
57#[cfg(test)]
58mod tests {
59    use super::*;
60
61    #[test]
62    fn rollout_observation_rejects_unsupported_malformed_or_insufficient_majorities() {
63        let good = RolloutQuorumStatus {
64            schema_version: 1,
65            target_node_id: 2,
66            leader_id: 1,
67            voters: vec![1, 2, 3],
68            peer_endpoints: (1..=3).map(|id| (id, format!("node-{id}:9879"))).collect(),
69            remaining_ready_voters: vec![1, 3],
70            allowed: true,
71        };
72        assert!(good.permits(2));
73        assert!(!good.permits(1));
74        let mut bad = good.clone();
75        bad.schema_version = 2;
76        assert!(!bad.permits(2));
77        bad = good.clone();
78        bad.remaining_ready_voters = vec![1, 1];
79        assert!(!bad.permits(2));
80        bad = good.clone();
81        bad.remaining_ready_voters = vec![1, 4];
82        assert!(!bad.permits(2));
83        bad = good.clone();
84        bad.remaining_ready_voters = vec![1, 2];
85        assert!(!bad.permits(2));
86        bad = good.clone();
87        bad.allowed = false;
88        assert!(!bad.permits(2));
89        bad = good.clone();
90        bad.voters.push(3);
91        assert!(!bad.permits(2));
92        assert!(serde_json::from_str::<RolloutQuorumStatus>("{}").is_err());
93        assert_eq!(
94            serde_json::from_slice::<RolloutQuorumStatus>(&serde_json::to_vec(&good).unwrap()).unwrap(),
95            good
96        );
97    }
98}
rocketmq_protocol/protocol/body/controller_rollout.rs

rocketmq_protocol/protocol/body/
controller_rollout.rs
