rocketmq_protocol/
contract.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 rocketmq_error::CanonicalCondition;
16
17use crate::protocol::header_codec::HeaderValueKind;
18use crate::protocol::remoting_command_defaults::RemotingCommandDefaultsConflict;
19use crate::protocol::remoting_command_facade::InvalidRemotingSerializeType;
20
21/// A deterministic RocketMQ protocol contract violation.
22///
23/// Header variants contain only static schema metadata and never retain raw
24/// wire values. Typed configuration sources remain available through
25/// [`std::error::Error::source`], while their text is excluded from this
26/// error's fixed display messages.
27#[derive(Clone, Debug, Eq, PartialEq, thiserror::Error)]
28#[non_exhaustive]
29pub enum ProtocolContractViolation {
30    /// A checkpoint uses an unsupported manifest schema version.
31    #[error("checkpoint schema version {actual} does not match {expected}")]
32    CheckpointSchemaVersion {
33        /// Supported schema version.
34        expected: u16,
35        /// Version carried by the manifest.
36        actual: u16,
37    },
38    /// A checkpoint field violates a fixed validation rule.
39    #[error("invalid checkpoint field {field}: {reason}")]
40    InvalidCheckpointField {
41        /// Stable field identifier.
42        field: &'static str,
43        /// Stable validation reason.
44        reason: &'static str,
45    },
46    /// Checkpoint offsets violate a fixed ordering rule.
47    #[error("invalid checkpoint offsets: {reason}")]
48    InvalidCheckpointOffsets {
49        /// Stable offset-ordering reason.
50        reason: &'static str,
51    },
52    /// A checkpoint set does not contain a Store member.
53    #[error("checkpoint set has no Store member")]
54    MissingStoreMembers,
55    /// A checkpoint set repeats a Store member.
56    #[error("checkpoint set repeats a Store member")]
57    DuplicateStoreMember,
58    /// A checkpoint artifact does not match its set barrier.
59    #[error("checkpoint member does not match the set barrier")]
60    CheckpointSetBindingMismatch {
61        /// Static checkpoint member category.
62        member: &'static str,
63    },
64    /// A checkpoint permits destructive rollback behavior.
65    #[error("checkpoint permits destructive WAL or persistent-volume replacement")]
66    DestructiveRollback,
67    /// Restore verification does not prove every required invariant.
68    #[error("checkpoint restore verification is incomplete")]
69    RestoreVerificationIncomplete,
70    /// A field declared as required was absent.
71    #[error("missing required header field {header}.{key}")]
72    Missing {
73        /// Stable header type identifier.
74        header: &'static str,
75        /// Canonical wire key.
76        key: &'static str,
77    },
78    /// A present field could not be decoded as its declared wire type.
79    #[error("invalid header field {header}.{key}: expected {expected:?}")]
80    InvalidValue {
81        /// Stable header type identifier.
82        header: &'static str,
83        /// Canonical wire key.
84        key: &'static str,
85        /// Expected protocol value category.
86        expected: HeaderValueKind,
87    },
88    /// Canonical and alias keys carried different values.
89    #[error("conflicting canonical and alias values for {header}.{key}")]
90    Conflict {
91        /// Stable header type identifier.
92        header: &'static str,
93        /// Canonical wire key.
94        key: &'static str,
95    },
96    /// A typed value and dynamic extension field disagreed.
97    #[error("typed header and dynamic field conflict for {header}.{key}")]
98    DynamicFieldConflict {
99        /// Stable header type identifier.
100        header: &'static str,
101        /// Canonical wire key.
102        key: &'static str,
103    },
104    /// A legacy custom-header validation hook failed.
105    #[error("legacy header validation failed for {header}")]
106    LegacyValidation {
107        /// Static Rust type name for the legacy header.
108        header: &'static str,
109    },
110    /// A legacy header could not provide its compatibility map.
111    #[error("legacy header map conversion failed for {header}")]
112    LegacyMapConversionFailed {
113        /// Static Rust type name for the legacy header.
114        header: &'static str,
115    },
116    /// A header-specific validation rule failed.
117    #[error("header validation failed for {header}: {rule}")]
118    Validation {
119        /// Stable header type identifier.
120        header: &'static str,
121        /// Static, non-sensitive validation rule identifier.
122        rule: &'static str,
123    },
124    /// An unsigned Rust value exceeded its declared signed Java range.
125    #[error("header value is outside Java range for {header}.{key}")]
126    JavaRange {
127        /// Stable header type identifier.
128        header: &'static str,
129        /// Canonical wire key.
130        key: &'static str,
131    },
132    /// A wire key cannot be represented by the ROCKETMQ binary format.
133    #[error("header key length exceeds the ROCKETMQ limit for {header}.{key}")]
134    KeyLengthOverflow {
135        /// Stable header type identifier.
136        header: &'static str,
137        /// Canonical wire key.
138        key: &'static str,
139    },
140    /// A wire value cannot be represented by the ROCKETMQ binary format.
141    #[error("header value length exceeds the ROCKETMQ limit for {header}.{key}")]
142    ValueLengthOverflow {
143        /// Stable header type identifier.
144        header: &'static str,
145        /// Canonical wire key.
146        key: &'static str,
147    },
148    /// The complete extension-field payload exceeded its signed length field.
149    #[error("ROCKETMQ extension-field payload exceeds the signed 32-bit wire limit")]
150    ExtensionFieldsLengthOverflow,
151    /// A dynamic extension-field key exceeded Java's signed 16-bit length.
152    #[error("dynamic header key length exceeds the ROCKETMQ wire limit")]
153    DynamicKeyLengthOverflow,
154    /// A dynamic extension-field value exceeded its signed 32-bit length.
155    #[error("dynamic header value length exceeds the ROCKETMQ wire limit")]
156    DynamicValueLengthOverflow,
157    /// Direct binary encoding was requested for a header without that capability.
158    #[error("direct binary codec is unavailable for {header}")]
159    FastCodecUnavailable {
160        /// Stable header type identifier.
161        header: &'static str,
162    },
163    /// Process remoting configuration selects an unsupported serialization type.
164    #[error("invalid remoting serialization type")]
165    InvalidSerializeType(
166        #[from]
167        #[source]
168        InvalidRemotingSerializeType,
169    ),
170    /// Process remoting defaults conflict with an initialized value.
171    #[error("remoting command defaults conflict with the initialized value")]
172    RemotingDefaultsConflict(
173        #[from]
174        #[source]
175        RemotingCommandDefaultsConflict,
176    ),
177    /// A subscription-group name is blank.
178    #[error("The specified group is blank.")]
179    BlankSubscriptionGroup,
180    /// A subscription-group name exceeds the supported length.
181    #[error("The specified group is longer than group max length: {max_length}")]
182    SubscriptionGroupTooLong {
183        /// Maximum accepted UTF-8 byte length.
184        max_length: usize,
185    },
186    /// A subscription-group name contains unsupported characters.
187    #[error("The specified group contains illegal characters, allowing only ^[%|a-zA-Z0-9_-]+$")]
188    SubscriptionGroupIllegalCharacters,
189    /// A batch contains the same subscription-group name more than once.
190    #[error("The specified group list contains a duplicate group.")]
191    DuplicateSubscriptionGroup,
192}
193
194impl ProtocolContractViolation {
195    /// Returns the protocol-independent condition for this violation.
196    #[must_use]
197    pub const fn condition(&self) -> CanonicalCondition {
198        match self {
199            Self::DestructiveRollback | Self::RestoreVerificationIncomplete | Self::RemotingDefaultsConflict(_) => {
200                CanonicalCondition::FailedPrecondition
201            }
202            Self::CheckpointSchemaVersion { .. }
203            | Self::InvalidCheckpointField { .. }
204            | Self::InvalidCheckpointOffsets { .. }
205            | Self::MissingStoreMembers
206            | Self::DuplicateStoreMember
207            | Self::CheckpointSetBindingMismatch { .. }
208            | Self::Missing { .. }
209            | Self::InvalidValue { .. }
210            | Self::Conflict { .. }
211            | Self::DynamicFieldConflict { .. }
212            | Self::LegacyValidation { .. }
213            | Self::LegacyMapConversionFailed { .. }
214            | Self::Validation { .. }
215            | Self::JavaRange { .. }
216            | Self::KeyLengthOverflow { .. }
217            | Self::ValueLengthOverflow { .. }
218            | Self::ExtensionFieldsLengthOverflow
219            | Self::DynamicKeyLengthOverflow
220            | Self::DynamicValueLengthOverflow
221            | Self::FastCodecUnavailable { .. }
222            | Self::InvalidSerializeType(_)
223            | Self::BlankSubscriptionGroup
224            | Self::SubscriptionGroupTooLong { .. }
225            | Self::SubscriptionGroupIllegalCharacters
226            | Self::DuplicateSubscriptionGroup => CanonicalCondition::InvalidArgument,
227        }
228    }
229}
230
231#[cfg(test)]
232mod tests {
233    use std::error::Error as _;
234
235    use super::*;
236    use crate::protocol::remoting_command::SERIALIZE_TYPE_PROPERTY;
237    use crate::protocol::remoting_command_facade::resolve_remoting_serialize_type;
238
239    #[test]
240    fn invalid_configuration_keeps_a_typed_source_behind_fixed_text() {
241        const SENTINEL: &str = "unsupported-serialize-type-secret";
242
243        let error = resolve_remoting_serialize_type(Some(SENTINEL), None).unwrap_err();
244
245        assert_eq!(error.condition(), CanonicalCondition::InvalidArgument);
246        assert_eq!(error.to_string(), "invalid remoting serialization type");
247        assert!(!format!("{error:?}").contains(SENTINEL));
248        let source = error
249            .source()
250            .and_then(|source| source.downcast_ref::<InvalidRemotingSerializeType>())
251            .expect("typed remoting serialization source");
252        assert_eq!(source.key(), SERIALIZE_TYPE_PROPERTY);
253        assert_eq!(source.value(), SENTINEL);
254    }
255
256    #[test]
257    fn rollback_contracts_are_failed_preconditions() {
258        assert_eq!(
259            ProtocolContractViolation::DestructiveRollback.condition(),
260            CanonicalCondition::FailedPrecondition
261        );
262        assert_eq!(
263            ProtocolContractViolation::RestoreVerificationIncomplete.condition(),
264            CanonicalCondition::FailedPrecondition
265        );
266    }
267}
rocketmq_protocol/contract.rs

rocketmq_protocol/
contract.rs
