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

krabka_protocol/krabka/freeze/
set_topic_freeze.rs
