krabka_raft/
wire.rs

1//! Krabka-private controller RPCs over Kafka TCP framing.
2//!
3//! These bodies are NOT part of `krabka-protocol`'s codegen. They are
4//! controller-only and Krabka-specific, with hand-written `encode_v0` and
5//! `decode_v0` methods. The KIP-595 quorum RPCs (Fetch, Vote, Begin, End) ride
6//! the generated codecs instead. See [`crate::kraft::transport::wire`]. The two
7//! types here back the observer metadata-fetch and the follower-to-leader
8//! submit-change forward.
9//!
10//! Api keys: `1003` `SubmitChange` (forward), `1004` `MetadataFetch`
11//! (observer), and `1005` `DelegationTokenMutation` (guarded forward). They
12//! come from [`PrivateRpc::api_key`], so this module and `krabka-metadata`
13//! cannot disagree.
14//!
15//! # Version negotiation
16//!
17//! The request header's `api_version` versions each body. No `ApiVersions`
18//! response advertises these keys. The finalized `krabka.version` feature
19//! negotiates them instead, as `metadata.version` negotiates Kafka's
20//! inter-broker protocol versions. The controller finalizes a `krabka.version`
21//! level only once every registered node supports it, so the level tells a
22//! sender which request versions every peer serves:
23//!
24//! - A **sender** sends [`private_request_version`]: the version that
25//!   `krabka_metadata::private_rpc_version` gives for the `krabka.version`
26//!   finalized in its current metadata image. With no image, or an image that
27//!   has not finalized the feature, that is v0, the krabka-broker 1.0.0
28//!   baseline that every 1.x node serves.
29//! - A **receiver** serves every version from 0 up to
30//!   [`private_api_highest_version`], the table's version at this build's
31//!   `KRABKA_VERSION_MAX`. It answers any other version with
32//!   [`PRIVATE_UNSUPPORTED_VERSION`] in the v0 response shape, which every 1.x
33//!   sender decodes.
34//!
35//! At `krabka.version` 0 and 1 every RPC is v0. The first level that raises
36//! one adds a `max_request_version` arm in `krabka-metadata`, and the matching
37//! encode and decode methods here.
38
39use {
40    crate::kraft::snapshot_fetch::SnapshotId,
41    bytes::{Buf, BufMut, Bytes},
42    krabka_metadata::{
43        KrabkaVersion, MetadataImage, PrivateRpc,
44        krabka_version::{KRABKA_VERSION_MAX, KRABKA_VERSION_MIN},
45        private_rpc_version,
46    },
47    krabka_protocol::ProtocolError,
48};
49
50const I32_LEN: usize = 4;
51const SUBMIT_CHANGE_RESPONSE_FIXED_LEN: usize = 10;
52const METADATA_FETCH_REQUEST_LEN: usize = 32;
53const METADATA_FETCH_RESPONSE_FIXED_LEN: usize = 54;
54/// `snapshot_id.end_offset` sentinel meaning "no snapshot": the observer's
55/// fetch offset is still inside the responder's retained log.
56const NO_SNAPSHOT: i64 = -1;
57
58/// Forwards a `Controller::submit_change` from a follower to the leader.
59///
60/// The body is the wincode-encoded `Vec<MetadataRecord>`. The response carries
61/// a single `error_code`, where 0 means applied and any non-zero value means
62/// not-leader or metadata-validation.
63pub const API_KEY_SUBMIT_CHANGE: i16 = PrivateRpc::SubmitChange.api_key();
64
65/// `KrabkaSubmitChangeResponse::error_code`: the leader refused a
66/// compare-and-set until its uncommitted tail commits. The forwarding node
67/// turns it back into [`crate::RaftError::UncommittedTail`].
68pub const SUBMIT_CHANGE_UNCOMMITTED_TAIL: i16 = 4;
69
70/// The `error_code` of a krabka-private response when the connection
71/// principal lacks `ClusterAction` on the cluster. It is Kafka's
72/// `CLUSTER_AUTHORIZATION_FAILED`.
73pub const PRIVATE_CLUSTER_AUTHORIZATION_FAILED: i16 = 31;
74
75/// Observer metadata fetch.
76///
77/// The body carries a `fetch_offset`, which is a `KraftLog` offset, and
78/// `max_bytes`. The response carries committed `__cluster_metadata` entries
79/// encoded as Kafka record batches, plus `log_start_offset`, `high_watermark`,
80/// `quorum_high_watermark`, a `leader_hint`, the responder's `leader_epoch`,
81/// and the KIP-630 `snapshot_id` that replaces the records when the fetch
82/// offset has been pruned away.
83pub const API_KEY_METADATA_FETCH: i16 = PrivateRpc::MetadataFetch.api_key();
84
85/// Generation-bound delegation-token mutation forwarded to the leader.
86pub const API_KEY_DELEGATION_TOKEN_MUTATION: i16 = PrivateRpc::DelegationTokenMutation.api_key();
87
88/// The version of every krabka-private API that a sender uses when it has no
89/// metadata image to read `krabka.version` from: v0, the krabka-broker 1.0.0
90/// formats, which every 1.x receiver serves.
91///
92/// It is also what an image that has not finalized `krabka.version` negotiates,
93/// since that image reads as level 0.
94pub const PRIVATE_BASELINE_VERSION: i16 = 0;
95
96/// The lowest version of every krabka-private API that this build serves.
97const PRIVATE_LOWEST_VERSION: i16 = PRIVATE_BASELINE_VERSION;
98
99/// The `error_code` of a krabka-private response to a request at a version
100/// the receiver does not implement. It is Kafka's `UNSUPPORTED_VERSION`.
101pub const PRIVATE_UNSUPPORTED_VERSION: i16 = 35;
102
103/// The highest version of the krabka-private `api_key` this build serves, or
104/// `None` when `api_key` is not one of them.
105///
106/// It is the version the `krabka.version` table gives at this build's
107/// `KRABKA_VERSION_MAX`. A receiver serves every version from 0 up to it,
108/// because a sender may send any version that a level up to that maximum
109/// gives, and a level never lowers a version.
110#[must_use]
111pub const fn private_api_highest_version(api_key: i16) -> Option<i16> {
112    private_rpc_version(api_key, KRABKA_VERSION_MAX)
113}
114
115/// The version of `rpc` a sender puts in the request header, negotiated from
116/// `image`, the sender's current metadata image.
117///
118/// It is `KrabkaVersion::finalized(image).max_request_version(rpc)`. The
119/// fallbacks all land on a version every peer serves:
120///
121/// - With no image (`None`), it is [`PRIVATE_BASELINE_VERSION`], v0.
122/// - An image that has not finalized `krabka.version` reads as level 0, which
123///   gives v0 too.
124/// - A level above this build's `KRABKA_VERSION_MAX` sends this build's
125///   highest version. The controller finalized that level because every
126///   registered node supports it, and a node that supports a level serves
127///   every version below the one it gives.
128#[must_use]
129pub fn private_request_version(rpc: PrivateRpc, image: Option<&MetadataImage>) -> i16 {
130    let Some(image) = image else {
131        return PRIVATE_BASELINE_VERSION;
132    };
133    let level = KrabkaVersion::finalized(image)
134        .level()
135        .clamp(KRABKA_VERSION_MIN, KRABKA_VERSION_MAX);
136    KrabkaVersion::new(level)
137        .max_request_version(rpc)
138        .unwrap_or(PRIVATE_BASELINE_VERSION)
139}
140
141/// The answer to a krabka-private request at `version` when this build does
142/// not serve that version: the v0 response of `api_key` with
143/// [`PRIVATE_UNSUPPORTED_VERSION`]. `None` when it serves `version`, or when
144/// `api_key` is not a krabka-private API.
145///
146/// # Errors
147/// Never in practice: the v0 error responses carry no payload.
148pub fn unsupported_version_response(
149    api_key: i16,
150    version: i16,
151) -> Result<Option<Bytes>, ProtocolError> {
152    let Some(highest) = private_api_highest_version(api_key) else {
153        return Ok(None);
154    };
155    if (PRIVATE_LOWEST_VERSION..=highest).contains(&version) {
156        return Ok(None);
157    }
158    let mut out = Vec::new();
159    if api_key == API_KEY_METADATA_FETCH {
160        KrabkaMetadataFetchResponse::unsupported_version().encode_v0(&mut out)?;
161    } else {
162        KrabkaSubmitChangeResponse {
163            error_code: PRIVATE_UNSUPPORTED_VERSION,
164            leader_hint: -1,
165            result: Bytes::new(),
166        }
167        .encode_v0(&mut out)?;
168    }
169    Ok(Some(Bytes::from(out)))
170}
171
172fn require_remaining(buf: &[u8], required: usize) -> Result<(), ProtocolError> {
173    match required.checked_sub(buf.remaining()) {
174        Some(0) | None => Ok(()),
175        Some(needed) => Err(ProtocolError::UnexpectedEof { needed }),
176    }
177}
178
179fn put_i32_len_prefixed_bytes(
180    out: &mut Vec<u8>,
181    bytes: &Bytes,
182    too_long: &'static str,
183) -> Result<(), ProtocolError> {
184    out.put_i32(i32::try_from(bytes.len()).map_err(|_| ProtocolError::InvalidValue(too_long))?);
185    out.put_slice(bytes);
186    Ok(())
187}
188
189fn get_i32_len_prefixed_bytes(
190    buf: &mut &[u8],
191    negative_len: &'static str,
192) -> Result<Bytes, ProtocolError> {
193    require_remaining(buf, I32_LEN)?;
194    let len = buf.get_i32();
195    let len = usize::try_from(len).map_err(|_| ProtocolError::InvalidValue(negative_len))?;
196    require_remaining(buf, len)?;
197    let (bytes, rest) = buf
198        .split_at_checked(len)
199        .ok_or(ProtocolError::UnexpectedEof { needed: len })?;
200    let bytes = Bytes::copy_from_slice(bytes);
201    *buf = rest;
202    Ok(bytes)
203}
204
205/// Forward-to-leader payload.
206///
207/// The body is opaque wincode bytes that represent the `Vec<MetadataRecord>` to
208/// apply. The controller layer owns the serde details, so the wire module stays
209/// metadata-agnostic.
210#[derive(Debug, Clone, PartialEq, Eq)]
211pub struct KrabkaSubmitChangeRequest {
212    pub records: Bytes,
213}
214
215impl KrabkaSubmitChangeRequest {
216    /// # Errors
217    /// Returns an error if the record payload is too large for the wire format.
218    pub fn encode_v0(&self, out: &mut Vec<u8>) -> Result<(), ProtocolError> {
219        put_i32_len_prefixed_bytes(out, &self.records, "records length exceeds i32::MAX")
220    }
221
222    /// # Errors
223    /// Returns an error if the payload is truncated or has an invalid length.
224    pub fn decode_v0(buf: &mut &[u8]) -> Result<Self, ProtocolError> {
225        let records = get_i32_len_prefixed_bytes(buf, "negative records length")?;
226        Ok(Self { records })
227    }
228}
229
230#[derive(Debug, Clone, PartialEq, Eq)]
231pub struct KrabkaSubmitChangeResponse {
232    /// 0 means success. Any other value is an opaque transport-level error
233    /// code: 1 is not leader, 2 is metadata validation, 3 is other, and 4
234    /// ([`SUBMIT_CHANGE_UNCOMMITTED_TAIL`]) is an uncommitted leader tail.
235    pub error_code: i16,
236    /// The leader id the responder believes is current, when the responder
237    /// cannot apply the change itself. -1 means "unknown".
238    pub leader_hint: i64,
239    /// Wincode-encoded [`crate::SubmitChangeResult`] on success.
240    pub result: Bytes,
241}
242
243impl KrabkaSubmitChangeResponse {
244    /// Encodes this response with wire version zero.
245    ///
246    /// # Errors
247    ///
248    /// Returns an error when the response payload exceeds the protocol's
249    /// signed 32-bit length field.
250    pub fn encode_v0(&self, out: &mut Vec<u8>) -> Result<(), ProtocolError> {
251        out.put_i16(self.error_code);
252        out.put_i64(self.leader_hint);
253        put_i32_len_prefixed_bytes(out, &self.result, "result length exceeds i32::MAX")?;
254        Ok(())
255    }
256
257    /// # Errors
258    /// Returns an error if the response payload is truncated.
259    pub fn decode_v0(buf: &mut &[u8]) -> Result<Self, ProtocolError> {
260        require_remaining(buf, SUBMIT_CHANGE_RESPONSE_FIXED_LEN)?;
261        Ok(Self {
262            error_code: buf.get_i16(),
263            leader_hint: buf.get_i64(),
264            result: get_i32_len_prefixed_bytes(buf, "negative result length")?,
265        })
266    }
267}
268
269#[derive(Debug, Clone, PartialEq, Eq)]
270pub struct KrabkaMetadataFetchRequest {
271    /// Next `KraftLog` offset the observer wants.
272    pub fetch_offset: i64,
273    /// Soft cap on the encoded record-batch payload.
274    pub max_bytes: i32,
275    /// The observer's node id, as Kafka's `FetchRequest.ReplicaId`. The leader
276    /// tracks the observer's progress under it for `DescribeQuorum`. A
277    /// negative id is a fetcher that is not a replica and is not tracked.
278    pub replica_id: i32,
279    /// The observer's directory id, as `FetchRequest.ReplicaDirectoryId`. Nil
280    /// when the observer has none.
281    pub replica_directory_id: uuid::Uuid,
282}
283
284impl KrabkaMetadataFetchRequest {
285    pub fn encode_v0(&self, out: &mut Vec<u8>) {
286        out.put_i64(self.fetch_offset);
287        out.put_i32(self.max_bytes);
288        out.put_i32(self.replica_id);
289        out.put_slice(self.replica_directory_id.as_bytes());
290    }
291
292    /// # Errors
293    /// Returns an error if the request payload is truncated.
294    pub fn decode_v0(buf: &mut &[u8]) -> Result<Self, ProtocolError> {
295        require_remaining(buf, METADATA_FETCH_REQUEST_LEN)?;
296        let fetch_offset = buf.get_i64();
297        let max_bytes = buf.get_i32();
298        let replica_id = buf.get_i32();
299        let mut directory = [0u8; 16];
300        buf.copy_to_slice(&mut directory);
301        Ok(Self {
302            fetch_offset,
303            max_bytes,
304            replica_id,
305            replica_directory_id: uuid::Uuid::from_bytes(directory),
306        })
307    }
308}
309
310#[derive(Debug, Clone, PartialEq, Eq)]
311pub struct KrabkaMetadataFetchResponse {
312    /// 0 means success. 1 means this node cannot serve, so read
313    /// `leader_hint`.
314    pub error_code: i16,
315    /// Leader id the responder believes is current. -1 means unknown.
316    pub leader_hint: i64,
317    /// The responder's quorum epoch. An observer sends it as the
318    /// `CurrentLeaderEpoch` of a KIP-595 `FetchSnapshot`, which the leader
319    /// checks against its own epoch.
320    pub leader_epoch: i32,
321    /// Lowest retained log offset on the responder.
322    pub log_start_offset: i64,
323    /// Highest committed and applied log offset on the responder. This bounds
324    /// the records it can serve, and on a follower it is clamped to that
325    /// follower's own log end.
326    pub high_watermark: i64,
327    /// Highest offset the *quorum* has committed, as the responder last heard
328    /// it from the leader.
329    ///
330    /// Every controller serves this API, not only the leader, so an observer
331    /// cannot read `high_watermark` as the quorum's progress: a follower that
332    /// is itself catching up reports its own clamped watermark, and an
333    /// observer that has drawn level with that follower would measure zero lag
334    /// against a quorum thousands of records ahead. This field is what the
335    /// readiness probe measures against.
336    pub quorum_high_watermark: i64,
337    /// The KIP-630 snapshot `(end_offset, epoch)` the observer must install
338    /// before it can resume fetching, set when `fetch_offset` is below
339    /// `log_start_offset` because the responder has pruned those records away.
340    ///
341    /// This is the observer's half of what `FetchResponse.SnapshotId` does for
342    /// a controller follower (KIP-595). Without it a restarted observer fetches
343    /// an offset the leader no longer holds, is served an empty slice, and
344    /// never makes progress.
345    pub snapshot_id: Option<SnapshotId>,
346    /// Concatenated Kafka `RecordBatch`es, one for each committed log batch.
347    /// Empty whenever `snapshot_id` is set: the records the observer asked for
348    /// are gone, so there is nothing to serve from the log.
349    pub records: Bytes,
350}
351
352impl KrabkaMetadataFetchResponse {
353    /// An unsupported-version refusal in the v0 response shape.
354    #[must_use]
355    pub fn unsupported_version() -> Self {
356        Self {
357            error_code: PRIVATE_UNSUPPORTED_VERSION,
358            leader_hint: -1,
359            leader_epoch: -1,
360            log_start_offset: -1,
361            high_watermark: -1,
362            quorum_high_watermark: -1,
363            snapshot_id: None,
364            records: Bytes::new(),
365        }
366    }
367
368    /// # Errors
369    /// Returns an error if the record payload is too large for the wire format.
370    pub fn encode_v0(&self, out: &mut Vec<u8>) -> Result<(), ProtocolError> {
371        out.put_i16(self.error_code);
372        out.put_i64(self.leader_hint);
373        out.put_i32(self.leader_epoch);
374        out.put_i64(self.log_start_offset);
375        out.put_i64(self.high_watermark);
376        out.put_i64(self.quorum_high_watermark);
377        let (snapshot_end_offset, snapshot_epoch) = self.snapshot_id.unwrap_or((NO_SNAPSHOT, -1));
378        out.put_i64(snapshot_end_offset);
379        out.put_i32(snapshot_epoch);
380        put_i32_len_prefixed_bytes(out, &self.records, "records length exceeds i32::MAX")
381    }
382
383    /// # Errors
384    /// Returns an error if the payload is truncated or has an invalid length.
385    pub fn decode_v0(buf: &mut &[u8]) -> Result<Self, ProtocolError> {
386        require_remaining(buf, METADATA_FETCH_RESPONSE_FIXED_LEN)?;
387        let error_code = buf.get_i16();
388        let leader_hint = buf.get_i64();
389        let leader_epoch = buf.get_i32();
390        let log_start_offset = buf.get_i64();
391        let high_watermark = buf.get_i64();
392        let quorum_high_watermark = buf.get_i64();
393        let snapshot_end_offset = buf.get_i64();
394        let snapshot_epoch = buf.get_i32();
395        let records = get_i32_len_prefixed_bytes(buf, "negative records length")?;
396        Ok(Self {
397            error_code,
398            leader_hint,
399            leader_epoch,
400            log_start_offset,
401            high_watermark,
402            quorum_high_watermark,
403            snapshot_id: (snapshot_end_offset != NO_SNAPSHOT)
404                .then_some((snapshot_end_offset, snapshot_epoch)),
405            records,
406        })
407    }
408}
409
410#[cfg(test)]
411mod tests {
412    use {super::*, assert2::check, std::fmt};
413
414    /// The v0 `SubmitChange` request body that [`submit_request`] encodes to.
415    const SUBMIT_REQUEST_V0: &[u8] = &[0, 0, 0, 3, 1, 2, 3];
416
417    /// The v0 `SubmitChange` response body that [`submit_response`] encodes to.
418    const SUBMIT_RESPONSE_V0: &[u8] = &[0, 4, 0, 0, 0, 0, 0, 0, 0, 3, 0, 0, 0, 2, b'o', b'k'];
419
420    /// The v0 `MetadataFetch` request body that [`fetch_request`] encodes to.
421    const FETCH_REQUEST_V0: &[u8] = &[
422        0, 0, 0, 0, 0, 0, 0, 42, 0, 0x10, 0, 0, 0, 0, 0, 7, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12,
423        13, 14, 15, 16,
424    ];
425
426    /// The v0 `MetadataFetch` response body that [`fetch_response`] encodes to.
427    const FETCH_RESPONSE_V0: &[u8] = &[
428        0, 0, // error_code
429        0, 0, 0, 0, 0, 0, 0, 3, // leader_hint
430        0, 0, 0, 4, // leader_epoch
431        0, 0, 0, 0, 0, 0, 0, 1, // log_start_offset
432        0, 0, 0, 0, 0, 0, 0, 99, // high_watermark
433        0, 0, 0, 0, 0, 0, 2, 0, // quorum_high_watermark
434        0, 0, 0, 0, 0, 0, 0, 64, // snapshot end offset
435        0, 0, 0, 2, // snapshot epoch
436        0, 0, 0, 1, 0xaa, // records
437    ];
438
439    fn submit_request() -> KrabkaSubmitChangeRequest {
440        KrabkaSubmitChangeRequest {
441            records: Bytes::from_static(b"\x01\x02\x03"),
442        }
443    }
444
445    fn submit_response() -> KrabkaSubmitChangeResponse {
446        KrabkaSubmitChangeResponse {
447            error_code: SUBMIT_CHANGE_UNCOMMITTED_TAIL,
448            leader_hint: 3,
449            result: Bytes::from_static(b"ok"),
450        }
451    }
452
453    fn fetch_request() -> KrabkaMetadataFetchRequest {
454        KrabkaMetadataFetchRequest {
455            fetch_offset: 42,
456            max_bytes: 1_048_576,
457            replica_id: 7,
458            replica_directory_id: uuid::Uuid::from_u128(0x0102_0304_0506_0708_090a_0b0c_0d0e_0f10),
459        }
460    }
461
462    fn fetch_response() -> KrabkaMetadataFetchResponse {
463        KrabkaMetadataFetchResponse {
464            error_code: 0,
465            leader_hint: 3,
466            leader_epoch: 4,
467            log_start_offset: 1,
468            high_watermark: 99,
469            quorum_high_watermark: 512,
470            snapshot_id: Some((64, 2)),
471            records: Bytes::from_static(b"\xaa"),
472        }
473    }
474
475    /// The v0 bodies of the krabka-private RPCs, byte for byte. A change to
476    /// one of them is a change to the 1.x rolling-upgrade contract.
477    #[test]
478    fn v0_bodies_match_their_golden_bytes() {
479        struct Scenario {
480            case: &'static str,
481            encoded: Vec<u8>,
482            golden: &'static [u8],
483        }
484
485        let mut submit_request_bytes = Vec::new();
486        submit_request()
487            .encode_v0(&mut submit_request_bytes)
488            .unwrap();
489        let mut submit_response_bytes = Vec::new();
490        submit_response()
491            .encode_v0(&mut submit_response_bytes)
492            .unwrap();
493        let mut fetch_request_bytes = Vec::new();
494        fetch_request().encode_v0(&mut fetch_request_bytes);
495        let mut fetch_response_bytes = Vec::new();
496        fetch_response()
497            .encode_v0(&mut fetch_response_bytes)
498            .unwrap();
499
500        let cases = [
501            Scenario {
502                case: "SubmitChange request",
503                encoded: submit_request_bytes,
504                golden: SUBMIT_REQUEST_V0,
505            },
506            Scenario {
507                case: "SubmitChange response",
508                encoded: submit_response_bytes,
509                golden: SUBMIT_RESPONSE_V0,
510            },
511            Scenario {
512                case: "MetadataFetch request",
513                encoded: fetch_request_bytes,
514                golden: FETCH_REQUEST_V0,
515            },
516            Scenario {
517                case: "MetadataFetch response",
518                encoded: fetch_response_bytes,
519                golden: FETCH_RESPONSE_V0,
520            },
521        ];
522        for Scenario {
523            case,
524            encoded,
525            golden,
526        } in cases
527        {
528            check!(encoded == golden, "{case}");
529        }
530    }
531
532    /// The golden v0 bodies decode back to the values they were made from.
533    #[test]
534    fn golden_v0_bodies_decode_back() {
535        check!(
536            KrabkaSubmitChangeRequest::decode_v0(&mut &*SUBMIT_REQUEST_V0).unwrap()
537                == submit_request()
538        );
539        check!(
540            KrabkaSubmitChangeResponse::decode_v0(&mut &*SUBMIT_RESPONSE_V0).unwrap()
541                == submit_response()
542        );
543        check!(
544            KrabkaMetadataFetchRequest::decode_v0(&mut &*FETCH_REQUEST_V0).unwrap()
545                == fetch_request()
546        );
547        check!(
548            KrabkaMetadataFetchResponse::decode_v0(&mut &*FETCH_RESPONSE_V0).unwrap()
549                == fetch_response()
550        );
551    }
552
553    /// A receiver serves every version from 0 up to the highest its
554    /// `krabka.version` table gives at `KRABKA_VERSION_MAX`, and answers any
555    /// other version with `UNSUPPORTED_VERSION` in that API's v0 response
556    /// shape.
557    #[test]
558    fn unsupported_versions_are_answered_in_the_v0_shape() {
559        let submit_refusal = KrabkaSubmitChangeResponse {
560            error_code: PRIVATE_UNSUPPORTED_VERSION,
561            leader_hint: -1,
562            result: Bytes::new(),
563        };
564        let mut submit_refusal_bytes = Vec::new();
565        submit_refusal.encode_v0(&mut submit_refusal_bytes).unwrap();
566        let fetch_refusal = KrabkaMetadataFetchResponse::unsupported_version();
567        let mut fetch_refusal_bytes = Vec::new();
568        fetch_refusal.encode_v0(&mut fetch_refusal_bytes).unwrap();
569        let submit = Some(Bytes::from(submit_refusal_bytes));
570        let fetch = Some(Bytes::from(fetch_refusal_bytes));
571        let max = |api_key| private_api_highest_version(api_key).expect("a private api");
572        let cases = [
573            (API_KEY_SUBMIT_CHANGE, 0, None),
574            (API_KEY_SUBMIT_CHANGE, max(API_KEY_SUBMIT_CHANGE), None),
575            (
576                API_KEY_SUBMIT_CHANGE,
577                max(API_KEY_SUBMIT_CHANGE) + 1,
578                submit.clone(),
579            ),
580            (API_KEY_SUBMIT_CHANGE, -1, submit.clone()),
581            (API_KEY_DELEGATION_TOKEN_MUTATION, 0, None),
582            (
583                API_KEY_DELEGATION_TOKEN_MUTATION,
584                max(API_KEY_DELEGATION_TOKEN_MUTATION),
585                None,
586            ),
587            (
588                API_KEY_DELEGATION_TOKEN_MUTATION,
589                max(API_KEY_DELEGATION_TOKEN_MUTATION) + 1,
590                submit,
591            ),
592            (API_KEY_METADATA_FETCH, 0, None),
593            (API_KEY_METADATA_FETCH, max(API_KEY_METADATA_FETCH), None),
594            (
595                API_KEY_METADATA_FETCH,
596                max(API_KEY_METADATA_FETCH) + 1,
597                fetch,
598            ),
599            (1, 99, None),
600        ];
601        for (api_key, version, want) in cases {
602            check!(
603                unsupported_version_response(api_key, version).unwrap() == want,
604                "api {api_key} v{version}"
605            );
606        }
607    }
608
609    /// The api keys are `PrivateRpc`'s, and this build serves exactly v0 of
610    /// each: the only bodies this module has codecs for. A `krabka.version`
611    /// level that raises a version fails this row until the module gains the
612    /// matching encode and decode methods.
613    #[test]
614    fn private_apis_come_from_the_krabka_version_table() {
615        let actual: Vec<_> = PrivateRpc::ALL
616            .iter()
617            .map(|&rpc| {
618                (
619                    rpc,
620                    rpc.api_key(),
621                    private_api_highest_version(rpc.api_key()),
622                )
623            })
624            .collect();
625        let expected = vec![
626            (PrivateRpc::SubmitChange, API_KEY_SUBMIT_CHANGE, Some(0)),
627            (PrivateRpc::MetadataFetch, API_KEY_METADATA_FETCH, Some(0)),
628            (
629                PrivateRpc::DelegationTokenMutation,
630                API_KEY_DELEGATION_TOKEN_MUTATION,
631                Some(0),
632            ),
633        ];
634        check!(actual == expected);
635        check!(
636            (
637                API_KEY_SUBMIT_CHANGE,
638                API_KEY_METADATA_FETCH,
639                API_KEY_DELEGATION_TOKEN_MUTATION
640            ) == (1003, 1004, 1005)
641        );
642        check!(private_api_highest_version(1002).is_none());
643    }
644
645    /// A sender's request version follows the `krabka.version` finalized in
646    /// its image: v0 without an image, with the feature absent, and at levels
647    /// 0 and 1, which are all the krabka-broker 1.0.0 formats. A level above
648    /// this build's maximum sends this build's highest version.
649    #[test]
650    fn sender_version_follows_the_finalized_krabka_version() {
651        struct Row {
652            case: &'static str,
653            image: Option<MetadataImage>,
654            want: [i16; 3],
655        }
656
657        use krabka_metadata::{
658            FeatureLevelRecord, MetadataRecord, krabka_version::KRABKA_VERSION_FEATURE,
659        };
660
661        let image_at = |level: Option<i16>| {
662            let mut image = MetadataImage::new(uuid::Uuid::nil());
663            if let Some(level) = level {
664                image.apply(&MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
665                    name: KRABKA_VERSION_FEATURE.into(),
666                    level,
667                }));
668            }
669            image
670        };
671        let rows = [
672            Row {
673                case: "no image",
674                image: None,
675                want: [0, 0, 0],
676            },
677            Row {
678                case: "feature absent",
679                image: Some(image_at(None)),
680                want: [0, 0, 0],
681            },
682            Row {
683                case: "level 0",
684                image: Some(image_at(Some(0))),
685                want: [0, 0, 0],
686            },
687            Row {
688                case: "level 1",
689                image: Some(image_at(Some(1))),
690                want: [0, 0, 0],
691            },
692            Row {
693                case: "level above this build's maximum",
694                image: Some(image_at(Some(KRABKA_VERSION_MAX + 1))),
695                want: [0, 0, 0],
696            },
697        ];
698        let actual: Vec<_> = rows
699            .iter()
700            .map(|row| {
701                (
702                    row.case,
703                    PrivateRpc::ALL.map(|rpc| private_request_version(rpc, row.image.as_ref())),
704                )
705            })
706            .collect();
707        let expected: Vec<_> = rows.iter().map(|row| (row.case, row.want)).collect();
708        check!(actual == expected);
709        check!(private_request_version(PrivateRpc::SubmitChange, None) == PRIVATE_BASELINE_VERSION);
710    }
711
712    fn assert_unexpected_eof<T: fmt::Debug>(result: Result<T, ProtocolError>, want: usize) {
713        match result {
714            Err(ProtocolError::UnexpectedEof { needed }) => assert2::assert!(needed == want),
715            other => panic!("expected UnexpectedEof {{ needed: {want} }}, got {other:?}"),
716        }
717    }
718
719    fn assert_invalid_value<T: fmt::Debug>(result: Result<T, ProtocolError>) {
720        match result {
721            Err(ProtocolError::InvalidValue(_)) => {}
722            other => panic!("expected InvalidValue, got {other:?}"),
723        }
724    }
725
726    #[test]
727    fn submit_change_round_trips() {
728        let req = KrabkaSubmitChangeRequest {
729            records: Bytes::from_static(b"\x01\x02\x03"),
730        };
731        let mut out = Vec::new();
732        req.encode_v0(&mut out).unwrap();
733        let mut cur: &[u8] = &out;
734        assert2::assert!(KrabkaSubmitChangeRequest::decode_v0(&mut cur).unwrap() == req);
735
736        let resp = KrabkaSubmitChangeResponse {
737            error_code: 1,
738            leader_hint: 3,
739            result: Bytes::from_static(b"result"),
740        };
741        let mut out = Vec::new();
742        resp.encode_v0(&mut out).unwrap();
743        let mut cur: &[u8] = &out;
744        assert2::assert!(KrabkaSubmitChangeResponse::decode_v0(&mut cur).unwrap() == resp);
745    }
746
747    #[test]
748    fn submit_change_request_decode_checks_prefix_and_payload_lengths() {
749        let mut short_prefix: &[u8] = &[0, 0, 0];
750        assert_unexpected_eof(KrabkaSubmitChangeRequest::decode_v0(&mut short_prefix), 1);
751
752        let mut negative_len: &[u8] = &(-1_i32).to_be_bytes();
753        assert_invalid_value(KrabkaSubmitChangeRequest::decode_v0(&mut negative_len));
754
755        let mut exact_empty: &[u8] = &[0, 0, 0, 0];
756        let decoded = KrabkaSubmitChangeRequest::decode_v0(&mut exact_empty).unwrap();
757        assert2::assert!(decoded.records.is_empty());
758        assert2::assert!(exact_empty.is_empty());
759
760        let mut short_payload: &[u8] = &[0, 0, 0, 4, 0xaa];
761        assert_unexpected_eof(KrabkaSubmitChangeRequest::decode_v0(&mut short_payload), 3);
762    }
763
764    #[test]
765    fn submit_change_response_decode_checks_fixed_length() {
766        let mut short: &[u8] = &[0, 1, 2];
767        assert_unexpected_eof(KrabkaSubmitChangeResponse::decode_v0(&mut short), 7);
768    }
769
770    #[test]
771    fn metadata_fetch_round_trips() {
772        let req = KrabkaMetadataFetchRequest {
773            fetch_offset: 42,
774            max_bytes: 1_048_576,
775            replica_id: 7,
776            replica_directory_id: uuid::Uuid::from_u128(0x0102_0304_0506_0708_090a_0b0c_0d0e_0f10),
777        };
778        let mut out = Vec::new();
779        req.encode_v0(&mut out);
780        let mut cur: &[u8] = &out;
781        assert2::assert!(KrabkaMetadataFetchRequest::decode_v0(&mut cur).unwrap() == req);
782
783        let resp = KrabkaMetadataFetchResponse {
784            error_code: 0,
785            leader_hint: 3,
786            leader_epoch: 4,
787            log_start_offset: 1,
788            high_watermark: 99,
789            quorum_high_watermark: 512,
790            snapshot_id: None,
791            records: Bytes::from_static(b"\x01\x02\x03"),
792        };
793        let mut out = Vec::new();
794        resp.encode_v0(&mut out).unwrap();
795        assert2::assert!(NO_SNAPSHOT == -1);
796        let snapshot_end_offset = i64::from_be_bytes(out[38..46].try_into().unwrap());
797        let snapshot_epoch = i32::from_be_bytes(out[46..50].try_into().unwrap());
798        assert2::assert!(snapshot_end_offset == -1);
799        assert2::assert!(snapshot_epoch == -1);
800        let mut cur: &[u8] = &out;
801        assert2::assert!(KrabkaMetadataFetchResponse::decode_v0(&mut cur).unwrap() == resp);
802    }
803
804    #[test]
805    fn metadata_fetch_request_decode_checks_fixed_length() {
806        let mut short: &[u8] = &[0, 1, 2, 3, 4];
807        assert_unexpected_eof(KrabkaMetadataFetchRequest::decode_v0(&mut short), 27);
808
809        let request = KrabkaMetadataFetchRequest {
810            fetch_offset: 9,
811            max_bytes: 512,
812            replica_id: -1,
813            replica_directory_id: uuid::Uuid::nil(),
814        };
815        let mut exact = Vec::new();
816        request.encode_v0(&mut exact);
817        assert2::assert!(exact.len() == METADATA_FETCH_REQUEST_LEN);
818        let mut cur: &[u8] = &exact;
819        let decoded = KrabkaMetadataFetchRequest::decode_v0(&mut cur).unwrap();
820        assert2::assert!(decoded == request);
821        assert2::assert!(cur.is_empty());
822    }
823
824    #[test]
825    fn metadata_fetch_response_decode_checks_fixed_and_payload_lengths() {
826        let mut short_fixed: &[u8] = &[0, 1, 2, 3, 4, 5, 6, 7, 8];
827        assert_unexpected_eof(KrabkaMetadataFetchResponse::decode_v0(&mut short_fixed), 45);
828
829        let resp = KrabkaMetadataFetchResponse {
830            error_code: 0,
831            leader_hint: -1,
832            leader_epoch: 4,
833            log_start_offset: 4,
834            high_watermark: 4,
835            quorum_high_watermark: 6,
836            snapshot_id: None,
837            records: Bytes::new(),
838        };
839        let mut exact = Vec::new();
840        resp.encode_v0(&mut exact).unwrap();
841        let mut cur: &[u8] = &exact;
842        assert2::assert!(KrabkaMetadataFetchResponse::decode_v0(&mut cur).unwrap() == resp);
843        assert2::assert!(cur.is_empty());
844
845        let mut short_payload = Vec::new();
846        short_payload.extend_from_slice(&0_i16.to_be_bytes());
847        short_payload.extend_from_slice(&1_i64.to_be_bytes());
848        short_payload.extend_from_slice(&5_i32.to_be_bytes());
849        short_payload.extend_from_slice(&2_i64.to_be_bytes());
850        short_payload.extend_from_slice(&3_i64.to_be_bytes());
851        short_payload.extend_from_slice(&3_i64.to_be_bytes());
852        short_payload.extend_from_slice(&(-1_i64).to_be_bytes());
853        short_payload.extend_from_slice(&(-1_i32).to_be_bytes());
854        short_payload.extend_from_slice(&4_i32.to_be_bytes());
855        short_payload.push(0xaa);
856        let mut cur: &[u8] = &short_payload;
857        assert_unexpected_eof(KrabkaMetadataFetchResponse::decode_v0(&mut cur), 3);
858    }
859
860    /// A pruned observer fetch answers with the snapshot id instead of
861    /// records, so the id has to survive the round trip as a `Some` and the
862    /// `-1` sentinel has to decode back to `None` rather than to `(-1, -1)`.
863    #[yare::parameterized(
864        nonzero_snapshot = { Some((4_096_i64, 7_i32)) },
865        zero_snapshot = { Some((0, 0)) },
866        no_snapshot = { None },
867    )]
868    fn metadata_fetch_response_round_trips_a_snapshot_id(want: Option<(i64, i32)>) {
869        let resp = KrabkaMetadataFetchResponse {
870            error_code: 0,
871            leader_hint: 1,
872            leader_epoch: 4,
873            log_start_offset: 4_096,
874            high_watermark: 5_000,
875            quorum_high_watermark: 5_000,
876            snapshot_id: want,
877            records: Bytes::new(),
878        };
879        let mut out = Vec::new();
880        resp.encode_v0(&mut out).unwrap();
881        let mut cur: &[u8] = &out;
882        let decoded = KrabkaMetadataFetchResponse::decode_v0(&mut cur).unwrap();
883        assert2::assert!(decoded == resp, "snapshot id {want:?}");
884        assert2::assert!(cur.is_empty());
885    }
886}
krabka_raft/wire.rs

krabka_raft/
wire.rs
