rocketmq_protocol/protocol/body/
supervised_mutation.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 serde::de::Error as _;
16use serde::Deserialize;
17use serde::Deserializer;
18use serde::Serialize;
19
20/// Closed state condition shared by supervised Topic and Subscription Group mutations.
21#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
22#[serde(rename_all = "snake_case", tag = "kind")]
23pub enum ExpectedState {
24    Absent,
25    Present { version: u64 },
26}
27
28impl<'de> Deserialize<'de> for ExpectedState {
29    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
30    where
31        D: Deserializer<'de>,
32    {
33        #[derive(Deserialize)]
34        #[serde(deny_unknown_fields)]
35        struct RawExpectedState {
36            kind: String,
37            version: Option<u64>,
38        }
39        let raw = RawExpectedState::deserialize(deserializer)?;
40        match (raw.kind.as_str(), raw.version) {
41            ("absent", None) => Ok(Self::Absent),
42            ("present", Some(version)) => Ok(Self::Present { version }),
43            ("absent" | "present", _) => Err(D::Error::custom("expected state has invalid version presence")),
44            _ => Err(D::Error::custom("unknown expected state kind")),
45        }
46    }
47}
48
49/// Complete allowlisted Topic state accepted by supervised replacement.
50#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
51#[serde(deny_unknown_fields, rename_all = "camelCase")]
52pub struct SupervisedTopicConfig {
53    pub read_queue_nums: u32,
54    pub write_queue_nums: u32,
55    pub perm: u32,
56    pub order: bool,
57    pub message_type: String,
58}
59
60#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
61#[serde(deny_unknown_fields, rename_all = "camelCase")]
62pub struct SupervisedTopicConfigCasRequestBody {
63    pub expected_state: ExpectedState,
64    pub replacement: SupervisedTopicConfig,
65}
66
67/// Complete public Subscription Group state accepted by supervised replacement.
68#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
69#[serde(deny_unknown_fields, rename_all = "camelCase")]
70pub struct SupervisedSubscriptionGroupConfig {
71    pub consume_enable: bool,
72    pub consume_from_min_enable: bool,
73    pub consume_broadcast_enable: bool,
74    pub consume_message_orderly: bool,
75    pub retry_queue_nums: i32,
76    pub retry_max_times: i32,
77    pub broker_id: u64,
78    pub which_broker_when_consume_slowly: u64,
79    pub notify_consumer_ids_changed_enable: bool,
80    pub group_sys_flag: i32,
81    pub consume_timeout_minute: i32,
82}
83
84#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
85#[serde(deny_unknown_fields, rename_all = "camelCase")]
86pub struct SupervisedSubscriptionGroupConfigCasRequestBody {
87    pub expected_state: ExpectedState,
88    pub replacement: SupervisedSubscriptionGroupConfig,
89}
90
91/// Broker mutation preflight contains no arbitrary property map.
92#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
93#[serde(deny_unknown_fields, rename_all = "camelCase")]
94pub struct BrokerMutationConfigState {
95    pub generation: u64,
96    pub auto_create_topic_enable: bool,
97    pub auto_create_subscription_group: bool,
98    pub broker_permission: u32,
99    pub default_topic_queue_nums: u32,
100    pub message_index_enable: bool,
101    pub trace_topic_enable: bool,
102}
103
104/// Exact request-mode value used for conditional inspection and replacement.
105#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
106#[serde(deny_unknown_fields, rename_all = "camelCase")]
107pub struct SupervisedMessageRequestMode {
108    pub mode: String,
109    pub pop_share_queue_num: i32,
110}
111
112#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
113#[serde(rename_all = "snake_case", tag = "kind")]
114pub enum ExpectedMessageRequestMode {
115    Absent,
116    Present { mode: String, pop_share_queue_num: i32 },
117}
118
119impl<'de> Deserialize<'de> for ExpectedMessageRequestMode {
120    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
121    where
122        D: Deserializer<'de>,
123    {
124        #[derive(Deserialize)]
125        #[serde(deny_unknown_fields, rename_all = "snake_case")]
126        struct RawExpectedMode {
127            kind: String,
128            mode: Option<String>,
129            pop_share_queue_num: Option<i32>,
130        }
131        let raw = RawExpectedMode::deserialize(deserializer)?;
132        match (raw.kind.as_str(), raw.mode, raw.pop_share_queue_num) {
133            ("absent", None, None) => Ok(Self::Absent),
134            ("present", Some(mode), Some(pop_share_queue_num)) => Ok(Self::Present {
135                mode,
136                pop_share_queue_num,
137            }),
138            ("absent" | "present", _, _) => Err(D::Error::custom("expected request mode has invalid value presence")),
139            _ => Err(D::Error::custom("unknown expected request mode kind")),
140        }
141    }
142}
143
144#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
145#[serde(deny_unknown_fields, rename_all = "camelCase")]
146pub struct GetMessageRequestModeRequestBody {
147    pub topic: String,
148    pub consumer_group: String,
149}
150
151#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
152#[serde(deny_unknown_fields, rename_all = "camelCase")]
153pub struct MessageRequestModeStateBody {
154    pub current: Option<SupervisedMessageRequestMode>,
155}
156
157/// Durable state of an accepted supervised mutation.
158#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
159#[serde(rename_all = "snake_case")]
160pub enum MutationPersistenceState {
161    NotRequired,
162    Persisted,
163    /// The write is confirmed not to have reached the target file.
164    Failed,
165    /// The write may already have reached the target file; the Broker ended
166    /// its observation without confirming either durability or failure.
167    Unconfirmed,
168}
169
170/// Conditional request-mode outcome, including applied and durability truth.
171#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
172#[serde(deny_unknown_fields, rename_all = "camelCase")]
173pub struct MessageRequestModeMutationResultBody {
174    pub applied: bool,
175    pub changed: bool,
176    pub current: Option<SupervisedMessageRequestMode>,
177    pub persistence: MutationPersistenceState,
178}
179
180#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
181#[serde(deny_unknown_fields, rename_all = "camelCase")]
182pub struct SetMessageRequestModeCasRequestBody {
183    pub topic: String,
184    pub consumer_group: String,
185    pub expected_state: ExpectedMessageRequestMode,
186    pub replacement: SupervisedMessageRequestMode,
187}
188
189#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
190#[serde(deny_unknown_fields, rename_all = "camelCase")]
191pub struct StateCasResultBody {
192    pub applied: bool,
193    pub changed: bool,
194    pub state: ExpectedState,
195    pub persistence: MutationPersistenceState,
196}
197
198#[cfg(test)]
199mod tests {
200    use super::*;
201
202    #[test]
203    fn expected_state_rejects_unknown_variants_and_fields() {
204        assert!(serde_json::from_str::<ExpectedState>(r#"{"kind":"missing"}"#).is_err());
205        assert!(serde_json::from_str::<ExpectedState>(r#"{"kind":"absent","version":1}"#).is_err());
206        assert_eq!(
207            serde_json::from_str::<ExpectedState>(r#"{"kind":"present","version":7}"#)
208                .expect("present state should decode"),
209            ExpectedState::Present { version: 7 }
210        );
211    }
212
213    #[test]
214    fn expected_message_request_mode_is_closed_and_requires_complete_present_state() {
215        for invalid in [
216            r#"{"kind":"absent","mode":"PULL","pop_share_queue_num":0}"#,
217            r#"{"kind":"present","mode":"PULL"}"#,
218            r#"{"kind":"present","pop_share_queue_num":0}"#,
219            r#"{"kind":"present","mode":"PULL","pop_share_queue_num":0,"extra":true}"#,
220            r#"{"kind":"unknown"}"#,
221        ] {
222            assert!(
223                serde_json::from_str::<ExpectedMessageRequestMode>(invalid).is_err(),
224                "{invalid} should be rejected"
225            );
226        }
227        assert_eq!(
228            serde_json::from_str::<ExpectedMessageRequestMode>(
229                r#"{"kind":"present","mode":"POP","pop_share_queue_num":4}"#
230            )
231            .expect("complete present state should decode"),
232            ExpectedMessageRequestMode::Present {
233                mode: "POP".to_owned(),
234                pop_share_queue_num: 4,
235            }
236        );
237    }
238}
rocketmq_protocol/protocol/body/supervised_mutation.rs

rocketmq_protocol/protocol/body/
supervised_mutation.rs
