krabka_protocol/krabka/freeze/
set_topic_freeze.rs

1//! `SetTopicFreeze`, api key 1015.
2//!
3//! One request sets a write freeze on a scope, or it removes one. A scope is a
4//! literal topic name or a topic-name prefix. The controller writes the outcome
5//! to the metadata log, so the freeze holds after a restart.
6//!
7//! A freeze refuses every produce to a topic that the scope covers. A thaw
8//! gives the topic back to its producers. The two directions are not
9//! symmetrical. A freeze is the safe direction and one command sets it. A thaw
10//! is the dangerous direction, so it names the break-glass proposal that
11//! approved it in `proposal_id`.
12//!
13//! # Signature
14//!
15//! `key_id` and `signature` carry a detached Ed25519 signature that the
16//! operator makes on their own machine. The private key never reaches a broker.
17//! Both fields are empty for an unsigned request. The broker decides whether it
18//! accepts an unsigned request, and it always refuses an unsigned thaw.
19
20use bytes::{Buf, BufMut};
21
22use super::PATTERN_TYPE_LITERAL;
23use crate::{
24    Decode, Encode, ProtocolError, ProtocolRequest, UnknownTaggedFields,
25    krabka::common::{
26        BOOL_LEN, I8_LEN, I16_LEN, I32_LEN, I64_LEN, UUID_LEN, check_version,
27        read_unknown_tagged_fields, unknown_tagged_fields_len, write_unknown_tagged_fields,
28    },
29    primitives::{
30        fixed::{
31            get_bool, get_i8, get_i16, get_i32, get_i64, put_bool, put_i8, put_i16, put_i32,
32            put_i64,
33        },
34        string_bytes::{
35            compact_bytes_len, compact_nullable_string_len, compact_string_len,
36            get_compact_bytes_owned, get_compact_nullable_string_owned, get_compact_string_owned,
37            put_compact_bytes, put_compact_nullable_string, put_compact_string,
38        },
39        uuid::{Uuid, get_uuid, put_uuid},
40    },
41};
42
43/// Api key of `SetTopicFreeze`.
44pub const API_KEY: i16 = 1015;
45/// Lowest version of `SetTopicFreeze` that this build speaks.
46pub const MIN_VERSION: i16 = 0;
47/// Highest version of `SetTopicFreeze` that this build speaks.
48pub const MAX_VERSION: i16 = 0;
49/// First version with flexible framing from KIP-482.
50pub const FLEXIBLE_MIN: i16 = 0;
51
52/// Reports whether `version` uses flexible framing from KIP-482.
53///
54/// The request header takes the same framing as the request body, so a caller
55/// that writes the header needs this answer.
56#[inline]
57#[must_use]
58pub fn is_flexible(version: i16) -> bool {
59    version >= FLEXIBLE_MIN
60}
61
62/// Sets a write freeze on one scope, or removes one.
63#[derive(Debug, Clone, PartialEq, Eq)]
64pub struct SetTopicFreezeRequest {
65    /// A literal topic name, or a topic-name prefix. `pattern_type` says which.
66    pub scope: String,
67    /// [`PATTERN_TYPE_LITERAL`] or
68    /// [`PATTERN_TYPE_PREFIXED`](super::PATTERN_TYPE_PREFIXED).
69    pub pattern_type: i8,
70    /// `true` sets the freeze. `false` removes it, and that is the thaw.
71    pub frozen: bool,
72    /// Text that says why the operator set or removed the freeze. The broker
73    /// keeps it in the metadata log and in the audit event.
74    pub reason: String,
75    /// The break-glass proposal that approved a thaw. It is nil for a freeze,
76    /// because a freeze needs no approval.
77    pub proposal_id: Uuid,
78    /// Milliseconds since the Unix epoch, at the moment the operator signed the
79    /// request. The broker refuses a value outside its skew window, and it
80    /// refuses a value that is not newer than the entry that this one replaces.
81    pub set_at_ms: i64,
82    /// The operator key that made `signature`. It is empty for an unsigned
83    /// request.
84    pub key_id: String,
85    /// Detached Ed25519 signature over the canonical bytes of the freeze. It is
86    /// empty for an unsigned request.
87    pub signature: Vec<u8>,
88    /// Tagged fields that this build does not know.
89    pub unknown_tagged_fields: UnknownTaggedFields,
90}
91
92impl Default for SetTopicFreezeRequest {
93    fn default() -> Self {
94        Self {
95            scope: String::new(),
96            pattern_type: PATTERN_TYPE_LITERAL,
97            frozen: true,
98            reason: String::new(),
99            proposal_id: Uuid::ZERO,
100            set_at_ms: 0,
101            key_id: String::new(),
102            signature: Vec::new(),
103            unknown_tagged_fields: UnknownTaggedFields::default(),
104        }
105    }
106}
107
108impl Encode for SetTopicFreezeRequest {
109    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
110        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
111        put_compact_string(buf, &self.scope);
112        put_i8(buf, self.pattern_type);
113        put_bool(buf, self.frozen);
114        put_compact_string(buf, &self.reason);
115        put_uuid(buf, self.proposal_id);
116        put_i64(buf, self.set_at_ms);
117        put_compact_string(buf, &self.key_id);
118        put_compact_bytes(buf, &self.signature);
119        write_unknown_tagged_fields(buf, &self.unknown_tagged_fields);
120        Ok(())
121    }
122
123    fn encoded_len(&self, _version: i16) -> usize {
124        compact_string_len(&self.scope)
125            + I8_LEN
126            + BOOL_LEN
127            + compact_string_len(&self.reason)
128            + UUID_LEN
129            + I64_LEN
130            + compact_string_len(&self.key_id)
131            + compact_bytes_len(&self.signature)
132            + unknown_tagged_fields_len(&self.unknown_tagged_fields)
133    }
134}
135
136impl Decode<'_> for SetTopicFreezeRequest {
137    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
138        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
139        Ok(Self {
140            scope: get_compact_string_owned(buf)?,
141            pattern_type: get_i8(buf)?,
142            frozen: get_bool(buf)?,
143            reason: get_compact_string_owned(buf)?,
144            proposal_id: get_uuid(buf)?,
145            set_at_ms: get_i64(buf)?,
146            key_id: get_compact_string_owned(buf)?,
147            signature: get_compact_bytes_owned(buf)?.to_vec(),
148            unknown_tagged_fields: read_unknown_tagged_fields(buf)?,
149        })
150    }
151}
152
153impl ProtocolRequest for SetTopicFreezeRequest {
154    const API_KEY: i16 = API_KEY;
155    const MIN_VERSION: i16 = MIN_VERSION;
156    const MAX_VERSION: i16 = MAX_VERSION;
157    const LATEST_STABLE_VERSION: i16 = MAX_VERSION;
158    const FLEXIBLE_MIN: i16 = FLEXIBLE_MIN;
159
160    type Response = SetTopicFreezeResponse;
161}
162
163/// Outcome of a [`SetTopicFreezeRequest`].
164#[derive(Debug, Clone, PartialEq, Eq, Default)]
165pub struct SetTopicFreezeResponse {
166    /// Milliseconds that the broker held the request back for a quota.
167    pub throttle_time_ms: i32,
168    /// Kafka error code. `0` means the controller applied the request.
169    pub error_code: i16,
170    /// Text that explains a non-zero `error_code`. `None` when the broker has
171    /// nothing to add to the code.
172    pub error_message: Option<String>,
173    /// Tagged fields that this build does not know.
174    pub unknown_tagged_fields: UnknownTaggedFields,
175}
176
177impl Encode for SetTopicFreezeResponse {
178    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
179        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
180        put_i32(buf, self.throttle_time_ms);
181        put_i16(buf, self.error_code);
182        put_compact_nullable_string(buf, self.error_message.as_deref());
183        write_unknown_tagged_fields(buf, &self.unknown_tagged_fields);
184        Ok(())
185    }
186
187    fn encoded_len(&self, _version: i16) -> usize {
188        I32_LEN
189            + I16_LEN
190            + compact_nullable_string_len(self.error_message.as_deref())
191            + unknown_tagged_fields_len(&self.unknown_tagged_fields)
192    }
193}
194
195impl Decode<'_> for SetTopicFreezeResponse {
196    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
197        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
198        Ok(Self {
199            throttle_time_ms: get_i32(buf)?,
200            error_code: get_i16(buf)?,
201            error_message: get_compact_nullable_string_owned(buf)?,
202            unknown_tagged_fields: read_unknown_tagged_fields(buf)?,
203        })
204    }
205}
206
207#[cfg(test)]
208impl SetTopicFreezeRequest {
209    /// Builds a signed freeze with every field set, for the round-trip tests.
210    fn populated() -> Self {
211        Self {
212            scope: "orders".to_string(),
213            pattern_type: PATTERN_TYPE_LITERAL,
214            frozen: true,
215            reason: "DR cutover".to_string(),
216            proposal_id: Uuid::ZERO,
217            set_at_ms: 1_770_000_000_000,
218            key_id: "alice-yubi".to_string(),
219            signature: (0..64).collect(),
220            unknown_tagged_fields: crate::krabka::test_support::sample_tagged_fields(),
221        }
222    }
223}
224
225#[cfg(test)]
226impl SetTopicFreezeResponse {
227    /// Builds a response with every field set, for the round-trip tests.
228    fn populated() -> Self {
229        Self {
230            throttle_time_ms: 21,
231            error_code: 1011,
232            error_message: Some("freeze scope names an internal topic".to_string()),
233            unknown_tagged_fields: crate::krabka::test_support::sample_tagged_fields(),
234        }
235    }
236}
237
238#[cfg(test)]
239mod tests {
240    use assert2::assert;
241
242    use super::*;
243    use crate::krabka::{
244        freeze::{PATTERN_TYPE_ANY, PATTERN_TYPE_PREFIXED},
245        test_support::{reject_truncated, roundtrip, roundtrip_case, unsupported_versions},
246    };
247
248    #[test]
249    fn requests_roundtrip() {
250        let cases = [
251            ("default", SetTopicFreezeRequest::default()),
252            ("signed literal freeze", SetTopicFreezeRequest::populated()),
253            (
254                "unsigned prefixed freeze",
255                SetTopicFreezeRequest {
256                    scope: "tenant-a.".to_string(),
257                    pattern_type: PATTERN_TYPE_PREFIXED,
258                    reason: "tenant offboarding".to_string(),
259                    ..SetTopicFreezeRequest::default()
260                },
261            ),
262            (
263                "signed thaw naming a proposal",
264                SetTopicFreezeRequest {
265                    frozen: false,
266                    proposal_id: Uuid([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15]),
267                    ..SetTopicFreezeRequest::populated()
268                },
269            ),
270        ];
271        for (label, request) in &cases {
272            for version in MIN_VERSION..=MAX_VERSION {
273                roundtrip_case(label, request, version);
274            }
275        }
276    }
277
278    #[test]
279    fn responses_roundtrip() {
280        let cases = [
281            ("default", SetTopicFreezeResponse::default()),
282            (
283                "refusal with a message",
284                SetTopicFreezeResponse::populated(),
285            ),
286            (
287                "accepted with no message",
288                SetTopicFreezeResponse {
289                    throttle_time_ms: 7,
290                    error_code: 0,
291                    error_message: None,
292                    unknown_tagged_fields: UnknownTaggedFields::default(),
293                },
294            ),
295        ];
296        for (label, response) in &cases {
297            for version in MIN_VERSION..=MAX_VERSION {
298                roundtrip_case(label, response, version);
299            }
300        }
301    }
302
303    #[test]
304    fn default_is_an_unsigned_literal_freeze() {
305        assert!(
306            SetTopicFreezeRequest::default()
307                == SetTopicFreezeRequest {
308                    scope: String::new(),
309                    pattern_type: PATTERN_TYPE_LITERAL,
310                    frozen: true,
311                    reason: String::new(),
312                    proposal_id: Uuid::ZERO,
313                    set_at_ms: 0,
314                    key_id: String::new(),
315                    signature: Vec::new(),
316                    unknown_tagged_fields: UnknownTaggedFields::default(),
317                }
318        );
319    }
320
321    #[test]
322    fn pattern_types_match_the_kafka_acl_values() {
323        assert!(
324            (
325                PATTERN_TYPE_ANY,
326                PATTERN_TYPE_LITERAL,
327                PATTERN_TYPE_PREFIXED
328            ) == (1, 3, 4)
329        );
330    }
331
332    #[test]
333    fn truncated_input_is_rejected() {
334        reject_truncated(&SetTopicFreezeRequest::populated(), MIN_VERSION);
335        reject_truncated(&SetTopicFreezeResponse::populated(), MIN_VERSION);
336    }
337
338    #[test]
339    fn signature_survives_a_roundtrip() {
340        let request = SetTopicFreezeRequest::populated();
341        assert!(request.signature.len() == 64);
342        roundtrip(&request, MIN_VERSION);
343    }
344
345    #[test]
346    fn rejects_unsupported_versions() {
347        for version in unsupported_versions(MIN_VERSION, MAX_VERSION) {
348            let request = SetTopicFreezeRequest::populated();
349            assert!(matches!(
350                request.encode(&mut Vec::<u8>::new(), version),
351                Err(ProtocolError::UnsupportedVersion {
352                    api_key: API_KEY,
353                    ..
354                })
355            ));
356            let response = SetTopicFreezeResponse::populated();
357            assert!(matches!(
358                response.encode(&mut Vec::<u8>::new(), version),
359                Err(ProtocolError::UnsupportedVersion {
360                    api_key: API_KEY,
361                    ..
362                })
363            ));
364        }
365    }
366}
krabka_protocol/krabka/freeze/set_topic_freeze.rs

krabka_protocol/krabka/freeze/
set_topic_freeze.rs
