krabka_raft/kraft/controller/
control_state.rs

1//! The KIP-853 control state: the uncommitted-through-committed history of
2//! voter sets and `kraft.version` levels, together with the conversions between
3//! that history and the `VotersRecord` wire shape it is written to the log in.
4
5use {
6    super::{KraftControlState, records::decode_control_record},
7    crate::{
8        error::{MetadataReplayError, RaftError},
9        kraft::types::NodeId,
10        voter_wire::{from_wire, record, to_wire},
11    },
12    krabka_ids::Offset,
13    krabka_metadata::{
14        KRaftVersionRecord, MetadataRecord, VoterSet, VotersRecord,
15        voters::{KRaftVersionRange, Voter},
16    },
17    krabka_protocol::{
18        owned::voters_record::VotersRecord as WireVotersRecord,
19        records::{RecordBatch, metadata::control::ControlRecord},
20    },
21    krabka_verified::raft,
22    std::collections::{BTreeMap, BTreeSet},
23    uuid::Uuid,
24};
25
26fn history_value_before<T>(history: &BTreeMap<i64, T>, frontier: i64) -> Option<&T> {
27    let offsets: Vec<i64> = history.keys().copied().collect();
28    let prefix_len = raft::control_history_frontier(&offsets, frontier);
29    let last = offsets.get(prefix_len.checked_sub(1)?)?;
30    history.get(last)
31}
32
33fn truncate_history<T>(history: &mut BTreeMap<i64, T>, frontier: i64) {
34    let offsets: Vec<i64> = history.keys().copied().collect();
35    let prefix_len = raft::control_history_frontier(&offsets, frontier);
36    if let Some(first_dropped) = offsets.get(prefix_len) {
37        drop(history.split_off(first_dropped));
38    }
39}
40
41impl KraftControlState {
42    pub fn new(voters: VoterSet, version: u16) -> Self {
43        Self {
44            voter_history: maplit::btreemap! {-1 => voters.clone()},
45            version_history: maplit::btreemap! {-1 => version},
46            committed_voters: voters,
47            committed_version: version,
48        }
49    }
50
51    pub fn latest_voters(&self) -> &VoterSet {
52        self.voter_history
53            .last_key_value()
54            .map_or(&self.committed_voters, |(_, voters)| voters)
55    }
56
57    pub fn latest_version(&self) -> u16 {
58        self.version_history
59            .last_key_value()
60            .map_or(self.committed_version, |(_, version)| *version)
61    }
62
63    pub fn voters_at(&self, end_offset: Offset) -> VoterSet {
64        history_value_before(&self.voter_history, end_offset.0)
65            .map_or_else(|| self.committed_voters.clone(), Clone::clone)
66    }
67
68    pub fn version_at(&self, end_offset: Offset) -> u16 {
69        history_value_before(&self.version_history, end_offset.0)
70            .map_or(self.committed_version, |version| *version)
71    }
72
73    pub fn apply(&mut self, offset: i64, record: &ControlRecord) -> Result<(), RaftError> {
74        match record {
75            ControlRecord::KRaftVersion(record) => {
76                let version = u16::try_from(record.k_raft_version).map_err(|_| {
77                    RaftError::ChangeRejected("negative kraft.version control record".into())
78                })?;
79                self.version_history.insert(offset, version);
80            }
81            ControlRecord::Voters(record) => {
82                self.voter_history
83                    .insert(offset, voter_set_from_wire(record)?);
84            }
85            _ => {}
86        }
87        Ok(())
88    }
89
90    pub fn truncate_to(&mut self, offset: i64) {
91        truncate_history(&mut self.voter_history, offset);
92        truncate_history(&mut self.version_history, offset);
93    }
94
95    pub fn commit_to(&mut self, high_watermark: i64) -> bool {
96        let voters = history_value_before(&self.voter_history, high_watermark)
97            .map_or_else(|| self.committed_voters.clone(), Clone::clone);
98        let version = history_value_before(&self.version_history, high_watermark)
99            .map_or(self.committed_version, |version| *version);
100        let changed = voters != self.committed_voters || version != self.committed_version;
101        self.committed_voters = voters;
102        self.committed_version = version;
103        changed
104    }
105}
106
107pub fn voter_supports_version(voter: &Voter, version: u16) -> bool {
108    voter.kraft_version.min <= version && version <= voter.kraft_version.max
109}
110
111pub fn voter_set_to_wire(voters: &VoterSet) -> WireVotersRecord {
112    let voters = voters
113        .iter()
114        .map(|voter| {
115            to_wire(
116                voter,
117                i32::try_from(voter.id.0).unwrap_or(i32::MAX),
118                i16::try_from(voter.kraft_version.min).unwrap_or(i16::MAX),
119                i16::try_from(voter.kraft_version.max).unwrap_or(i16::MAX),
120            )
121        })
122        .collect();
123    record(voters)
124}
125
126pub fn voter_set_from_wire(record: &WireVotersRecord) -> Result<VoterSet, RaftError> {
127    let mut voter_ids = BTreeSet::new();
128    let voter_ids_unique = record
129        .voters
130        .iter()
131        .all(|voter| voter_ids.insert(voter.voter_id));
132    match krabka_verified::voter_set_wire_decision(
133        record.version,
134        record.voters.len(),
135        voter_ids_unique,
136    ) {
137        krabka_verified::VoterSetWireDecision::UnsupportedRecordVersion => {
138            return Err(RaftError::InvalidVoterUpdate(format!(
139                "unsupported VotersRecord version {}",
140                record.version
141            )));
142        }
143        krabka_verified::VoterSetWireDecision::Empty => {
144            return Err(RaftError::InvalidVoterUpdate(
145                "empty voter set in VotersRecord".into(),
146            ));
147        }
148        krabka_verified::VoterSetWireDecision::DuplicateId => {
149            return Err(RaftError::InvalidVoterUpdate(
150                "duplicate voter id in VotersRecord".into(),
151            ));
152        }
153        krabka_verified::VoterSetWireDecision::Accept => {}
154    }
155    let voters = record
156        .voters
157        .iter()
158        .map(|voter| {
159            let directory_id = Uuid::from_bytes(voter.voter_directory_id.0);
160            let directory_id_exact = directory_id.as_bytes() == &voter.voter_directory_id.0;
161            let mut endpoint_names = BTreeSet::new();
162            let endpoints_valid = voter.endpoints.iter().all(|endpoint| {
163                !endpoint.name.is_empty()
164                    && !endpoint.host.is_empty()
165                    && endpoint.port != 0
166                    && endpoint_names.insert(endpoint.name.as_str())
167            });
168            match krabka_verified::voter_wire_decision(
169                voter.voter_id,
170                directory_id_exact,
171                voter.endpoints.len(),
172                endpoints_valid,
173                voter.k_raft_version_feature.min_supported_version,
174                voter.k_raft_version_feature.max_supported_version,
175            ) {
176                krabka_verified::VoterWireDecision::NegativeId => {
177                    return Err(RaftError::InvalidVoterUpdate(
178                        "negative voter id in VotersRecord".into(),
179                    ));
180                }
181                krabka_verified::VoterWireDecision::DirectoryMismatch => {
182                    return Err(RaftError::InvalidVoterUpdate(
183                        "voter directory id changed during decoding".into(),
184                    ));
185                }
186                krabka_verified::VoterWireDecision::InvalidEndpoint => {
187                    return Err(RaftError::InvalidVoterUpdate(
188                        "voter endpoints must be nonempty, uniquely named, and have a name, host, and nonzero port".into(),
189                    ));
190                }
191                krabka_verified::VoterWireDecision::InvalidVersionRange => {
192                    return Err(RaftError::InvalidVoterUpdate(
193                        "invalid kraft.version range in VotersRecord".into(),
194                    ));
195                }
196                krabka_verified::VoterWireDecision::Accept => {}
197            }
198            let id = u64::try_from(voter.voter_id).map_err(|_| {
199                RaftError::InvalidVoterUpdate("voter id must be nonnegative".into())
200            })?;
201            let invalid_range =
202                |_| RaftError::InvalidVoterUpdate("invalid kraft.version range in VotersRecord".into());
203            let min = u16::try_from(voter.k_raft_version_feature.min_supported_version)
204                .map_err(invalid_range)?;
205            let max = u16::try_from(voter.k_raft_version_feature.max_supported_version)
206                .map_err(invalid_range)?;
207            Ok(from_wire(
208                voter,
209                NodeId(id),
210                directory_id,
211                KRaftVersionRange { min, max },
212            ))
213        })
214        .collect::<Result<Vec<_>, RaftError>>()?;
215    Ok(VoterSet::from_voters(voters))
216}
217
218/// The image records one committed control batch carries, in log order: each
219/// KIP-853 `KRaftVersionRecord` as a `V1KRaftVersion` and each `VotersRecord`
220/// as a `V1Voters`.
221///
222/// The KIP-595 `LeaderChange` marker and the snapshot framing records change no
223/// image, so they yield nothing, and so does a batch that is not a control
224/// batch.
225///
226/// A controller folds these into its image when it commits them. A broker-only
227/// observer reads only committed records, and folds them in as it reads them,
228/// so its image names the same voters, and the endpoints that reach them.
229///
230/// # Errors
231/// [`MetadataReplayError::InvalidControlRecord`] for a control record that
232/// does not decode, or that names a `kraft.version` or voter set Kafka would
233/// refuse, as restart recovery refuses it. Kafka halts on such a record on
234/// every role.
235pub fn control_batch_image_records(
236    batch: &RecordBatch,
237) -> Result<Vec<MetadataRecord>, MetadataReplayError> {
238    if !batch.attributes.is_control_batch() {
239        return Ok(Vec::new());
240    }
241    let mut records = Vec::new();
242    for record in &batch.records {
243        let offset = batch
244            .base_offset
245            .saturating_add(i64::from(record.offset_delta));
246        let invalid = |reason: String| MetadataReplayError::InvalidControlRecord { offset, reason };
247        match decode_control_record(record).map_err(|error| invalid(error.to_string()))? {
248            ControlRecord::KRaftVersion(record) => {
249                let kraft_version = u16::try_from(record.k_raft_version).map_err(|_| {
250                    invalid(format!("negative kraft.version {}", record.k_raft_version))
251                })?;
252                records.push(MetadataRecord::V1KRaftVersion(KRaftVersionRecord {
253                    kraft_version,
254                }));
255            }
256            ControlRecord::Voters(record) => {
257                records.push(MetadataRecord::V1Voters(VotersRecord {
258                    voters: voter_set_from_wire(&record)
259                        .map_err(|error| invalid(error.to_string()))?,
260                }));
261            }
262            _ => {}
263        }
264    }
265    Ok(records)
266}
267
268#[cfg(test)]
269mod tests {
270    use {
271        super::*,
272        krabka_metadata::voters::{KRaftVersionRange, Voter, VoterEndpoint},
273    };
274
275    fn versioned_voter() -> Voter {
276        let ep = VoterEndpoint {
277            name: "CONTROLLER".into(),
278            host: "127.0.0.1".into(),
279            port: 9092,
280        };
281        Voter {
282            id: NodeId(1),
283            directory_id: uuid::Uuid::nil(),
284            endpoints: vec![ep],
285            kraft_version: KRaftVersionRange { min: 2, max: 4 },
286        }
287    }
288
289    #[test]
290    fn voter_supports_version_checks_both_bounds() {
291        let voter = versioned_voter();
292        assert2::check!(!voter_supports_version(&voter, 1));
293        assert2::check!(voter_supports_version(&voter, 2));
294        assert2::check!(voter_supports_version(&voter, 3));
295        assert2::check!(voter_supports_version(&voter, 4));
296        assert2::check!(!voter_supports_version(&voter, 5));
297    }
298
299    #[test]
300    fn voter_set_to_wire_encodes_version_and_features() {
301        let voter = versioned_voter();
302        let set = VoterSet::from_voters([voter]);
303        let wire = voter_set_to_wire(&set);
304        assert2::assert!(wire.version == 0);
305        assert2::assert!(wire.voters[0].k_raft_version_feature.min_supported_version == 2);
306        assert2::assert!(wire.voters[0].k_raft_version_feature.max_supported_version == 4);
307    }
308}
krabka_raft/kraft/controller/control_state.rs

krabka_raft/kraft/controller/
control_state.rs
