krabka_raft/kraft/controller/
records.rs

1//! Construction, encoding and decoding of the record batches the engine writes
2//! to and reads from the metadata log: metadata value batches, KIP-853 typed
3//! control batches, and the KIP-595 `LeaderChange` marker.
4
5use {
6    crate::{
7        error::{MetadataReplayError, RaftError},
8        kraft::types::{Epoch, NodeId},
9    },
10    krabka_ids::Offset,
11    krabka_metadata::{MetadataImage, MetadataRecord, TranslateError, VoterSet, from_kraft_value},
12    krabka_protocol::{
13        Decode, Encode,
14        owned::no_op_record::NoOpRecord,
15        records::{
16            self, Record, RecordBatch,
17            metadata::{KraftMetadataRecord, control::ControlRecord, decode_value_header},
18        },
19    },
20};
21
22/// Decodes the committed metadata value at `offset` against `image`.
23///
24/// Every controller replay path reads it so: live apply, restart recovery, the
25/// downgrade-snapshot rebuild and the leader's walk of an earlier epoch's
26/// tail.
27///
28/// Bytes that are not a metadata record this build reads are a
29/// [`MetadataReplayError::UndecodableRecord`], and the caller stops the
30/// controller. Kafka's controller cannot replay such a record either:
31/// `MetadataRecordSerde` throws on it, and `QuorumController` and the
32/// controller-role `MetadataLoader` hand the throw to a fatal fault handler
33/// (`SharedServer.fatalQuorumControllerFaultHandler` and
34/// `SharedServer.metadataLoaderFaultHandler`, whose `fatal` is
35/// `processRoles.contains(ProcessRole.ControllerRole)`). The snapshot reader
36/// refuses the same bytes, so a snapshot and the log it replaces agree.
37///
38/// An empty KIP-835 `NoOpRecord` is `Ok(None)`: it changes nothing.
39///
40/// A record that decodes but names a topic, partition or ACL the image does
41/// not hold is `Ok(None)`: the caller skips it, as it skips a record that fails
42/// `MetadataImage::validate`. Kafka never commits such a record, because its
43/// active controller replays each record before it appends it. A krabka leader
44/// checks a write against its committed image, so a write can lose a race to an
45/// earlier one still in flight (a partition change behind the deletion of its
46/// topic). Every replica then decodes the same bytes against the same image
47/// and skips the same record, so the replicas agree. Stopping on it instead
48/// would stop every controller on a race a client can cause.
49///
50/// `TranslateError::InvalidReference` is one of those image lookups (an
51/// unknown partition, a directory list measured against the image's replicas)
52/// and is skipped too. `TranslateError::InvalidValue` is decided by the bytes
53/// alone (an unknown `fenced` or `leader_recovery_state` value, an integer out
54/// of range) and is undecodable, as are the krabka-private record errors
55/// (`UnknownPrivateTag`, `UnknownPrivateRecordVersion`, `PrivateTagMismatch`,
56/// `TrailingPrivateRecordBytes`): only the image-resolution errors below are
57/// skips.
58///
59/// # Errors
60/// [`MetadataReplayError::UndecodableRecord`] as above.
61pub fn decode_committed_value(
62    value: &[u8],
63    image: &MetadataImage,
64    offset: i64,
65) -> Result<Option<MetadataRecord>, MetadataReplayError> {
66    // A KIP-835 no-op changes nothing, and has no image record to become.
67    if is_kip835_noop(value) {
68        return Ok(None);
69    }
70    match from_kraft_value(value, image) {
71        Ok(record) => Ok(Some(record)),
72        Err(
73            error @ (TranslateError::UnknownTopicId(_)
74            | TranslateError::UnknownTopicName(_)
75            | TranslateError::UnknownAclId(_)
76            | TranslateError::InvalidReference { .. }),
77        ) => {
78            tracing::warn!(
79                offset,
80                %error,
81                "kraft: skipped a committed record that names state the image does not hold"
82            );
83            Ok(None)
84        }
85        Err(error) => Err(MetadataReplayError::UndecodableRecord { offset, error }),
86    }
87}
88
89/// The api key of Kafka's `NoOpRecord`.
90const NO_OP_RECORD_API_KEY: u32 = 20;
91
92/// The value bytes of the empty KIP-835 `NoOpRecord` at apiVersion 0, which
93/// Kafka's `QuorumController` appends every `metadata.max.idle.interval.ms`.
94///
95/// # Errors
96/// Returns the [`RaftError`] of the record encoder.
97pub fn noop_record_value() -> Result<bytes::Bytes, RaftError> {
98    Ok(KraftMetadataRecord::NoOp(NoOpRecord::default()).encode_value(0)?)
99}
100
101/// Whether `value` is an empty KIP-835 `NoOpRecord`: a record that changes
102/// nothing, which every replay skips. A `NoOpRecord` that carries a private
103/// tagged field is a krabka record carrier, and it is not one.
104#[must_use]
105pub fn is_kip835_noop(value: &[u8]) -> bool {
106    let mut header = value;
107    if decode_value_header(&mut header).ok().map(|h| h.api_key) != Some(NO_OP_RECORD_API_KEY) {
108        return false;
109    }
110    matches!(
111        KraftMetadataRecord::decode_value(value),
112        Ok((KraftMetadataRecord::NoOp(record), _)) if record.unknown_tagged_fields.0.is_empty()
113    )
114}
115
116pub fn metadata_record_batch(
117    leader_epoch: Epoch,
118    blobs: &[bytes::Bytes],
119) -> Result<RecordBatch, RaftError> {
120    if blobs.is_empty() {
121        return Ok(RecordBatch {
122            partition_leader_epoch: i32::try_from(leader_epoch).unwrap_or(i32::MAX),
123            ..Default::default()
124        });
125    }
126
127    let deltas = krabka_verified::metadata_record_offset_deltas(blobs.len()).ok_or_else(|| {
128        RaftError::ChangeRejected("metadata batch offset deltas exceed i32".to_owned())
129    })?;
130    // The kernel returns one delta per record, so the batch is non-empty.
131    let last_offset_delta = deltas.last().copied().unwrap_or_default();
132    let records: Vec<Record> = blobs
133        .iter()
134        .zip(deltas)
135        .map(|(blob, offset_delta)| Record {
136            offset_delta,
137            value: Some(blob.clone()),
138            ..Default::default()
139        })
140        .collect();
141
142    Ok(RecordBatch {
143        partition_leader_epoch: i32::try_from(leader_epoch).unwrap_or(i32::MAX),
144        last_offset_delta,
145        records,
146        ..Default::default()
147    })
148}
149
150pub fn typed_control_batch(
151    leader_epoch: Epoch,
152    controls: &[ControlRecord],
153) -> Result<RecordBatch, RaftError> {
154    let records = controls
155        .iter()
156        .enumerate()
157        .map(|(index, control)| {
158            let (key, value) = control.encode_key_value()?;
159            Ok(Record {
160                offset_delta: i32::try_from(index).unwrap_or(i32::MAX),
161                key: Some(key),
162                value: Some(value),
163                ..Default::default()
164            })
165        })
166        .collect::<Result<Vec<_>, krabka_protocol::ProtocolError>>()?;
167    Ok(RecordBatch {
168        partition_leader_epoch: i32::try_from(leader_epoch).unwrap_or(i32::MAX),
169        attributes: records::Attributes::default().with_control(true),
170        last_offset_delta: i32::try_from(controls.len().saturating_sub(1)).unwrap_or(i32::MAX),
171        records,
172        ..Default::default()
173    })
174}
175
176/// Decode one record of a control batch.
177///
178/// # Errors
179/// [`RaftError::MalformedControlRecord`] for a missing or empty key or value,
180/// as Kafka's `RecordsIterator.decodeControlRecord` throws, and the decoder's
181/// error for a key or value that does not decode.
182pub fn decode_control_record(record: &Record) -> Result<ControlRecord, RaftError> {
183    let key = match &record.key {
184        None => {
185            return Err(RaftError::MalformedControlRecord(
186                "Missing key in the record when a key was expected",
187            ));
188        }
189        Some(key) if key.is_empty() => {
190            return Err(RaftError::MalformedControlRecord(
191                "Got an unexpected empty key in the record",
192            ));
193        }
194        Some(key) => key,
195    };
196    let value = match &record.value {
197        None => {
198            return Err(RaftError::MalformedControlRecord(
199                "Missing value in the record when a value was expected",
200            ));
201        }
202        Some(value) if value.is_empty() => {
203            return Err(RaftError::MalformedControlRecord(
204                "Got an unexpected empty value in the record",
205            ));
206        }
207        Some(value) => value,
208    };
209    Ok(ControlRecord::decode(key, value)?)
210}
211
212/// Build the leader's `LeaderChange` control batch for `epoch`: a single
213/// KIP-595 `LeaderChange` control record (control-batch attribute set), naming
214/// the new leader and the current voter set. A real `KRaft` batch MUST contain at
215/// least one record — an empty batch crashes a JVM follower
216/// (`Batch must contain at least one record`) — so this carries the proper
217/// `LeaderChangeMessage` rather than zero records. Krabka readers skip it via
218/// `is_control_batch()`; it occupies exactly one log offset
219/// (`last_offset_delta = 0`), unchanged from the prior empty batch.
220///
221/// The message is always version 0, at every `kraft.version`. Kafka writes it
222/// at `ControlRecordUtils.LEADER_CHANGE_CURRENT_VERSION`, which is 0, and
223/// reads it at that version too, so a version 1 message, with its voter
224/// directory ids, is unreadable to `kafka-dump-log` and to a JVM replica.
225// cargo-mutants: the `version: 0` field equals `LeaderChangeMessage`'s `Default` (i16 -> 0), so
226// deleting it yields byte-identical encoding; it is not the wire schema version
227// (that is the `0` passed to `msg.encode`). Equivalent mutant.
228#[cfg_attr(test, mutants::skip)]
229pub fn leader_change_batch(epoch: Epoch, leader_id: NodeId, voter_set: &VoterSet) -> RecordBatch {
230    use krabka_protocol::{
231        Encode,
232        owned::{
233            common::leader_change_message::voter::Voter, leader_change_message::LeaderChangeMessage,
234        },
235        records::{
236            header::Attributes,
237            metadata::control::{ControlRecordType, control_record_key},
238        },
239    };
240
241    let voters: Vec<Voter> = voter_set
242        .iter()
243        .map(|voter| Voter {
244            voter_id: i32::try_from(voter.id.0).unwrap_or(i32::MAX),
245            ..Default::default()
246        })
247        .collect();
248    let msg = LeaderChangeMessage {
249        version: 0,
250        leader_id: i32::try_from(leader_id.0).unwrap_or(i32::MAX),
251        voters: voters.clone(),
252        granting_voters: voters,
253        ..Default::default()
254    };
255    let mut value = bytes::BytesMut::new();
256    // LeaderChangeMessage v0; encode is infallible for a well-formed message.
257    let _ = msg.encode(&mut value, 0);
258    let key = control_record_key(ControlRecordType::LeaderChange);
259    RecordBatch {
260        partition_leader_epoch: i32::try_from(epoch).unwrap_or(i32::MAX),
261        attributes: Attributes::default().with_control(true),
262        last_offset_delta: 0,
263        records: vec![Record {
264            offset_delta: 0,
265            key: Some(key),
266            value: Some(value.freeze()),
267            ..Default::default()
268        }],
269        ..Default::default()
270    }
271}
272
273/// Build the batch a leader starts `epoch` with: the [`leader_change_batch`]
274/// marker, then each of `controls` at the offsets that follow it.
275///
276/// # Errors
277/// Returns the [`RaftError`] of the control-record encoder.
278pub fn start_of_epoch_batch(
279    epoch: Epoch,
280    leader_id: NodeId,
281    voter_set: &VoterSet,
282    controls: &[ControlRecord],
283) -> Result<RecordBatch, RaftError> {
284    let mut batch = leader_change_batch(epoch, leader_id, voter_set);
285    for (offset_delta, control) in (1_i32..).zip(controls) {
286        let (key, value) = control.encode_key_value()?;
287        batch.records.push(Record {
288            offset_delta,
289            key: Some(key),
290            value: Some(value),
291            ..Default::default()
292        });
293        batch.last_offset_delta = offset_delta;
294    }
295    Ok(batch)
296}
297
298/// Encode a run of `RecordBatch`es into one contiguous `Bytes` blob (each batch
299/// is self-describing via its `batch_length` header, so they concatenate and
300/// decode back in order — see [`decode_batches`]). Used by the leader's Fetch
301/// serve path to ship replicated record bytes to a follower.
302pub fn encode_batches(batches: &[RecordBatch]) -> bytes::Bytes {
303    let mut out = bytes::BytesMut::new();
304    for batch in batches {
305        if let Err(e) = batch.encode(&mut out) {
306            tracing::error!(?e, "kraft: encode batch for fetch serve failed");
307        }
308    }
309    out.freeze()
310}
311
312/// Decode the contiguous `Bytes` blob produced by [`encode_batches`] back into a
313/// `Vec<RecordBatch>` (each batch's `base_offset` is preserved). Used by the
314/// follower's Fetch-response apply path.
315pub fn decode_batches(mut buf: &[u8]) -> Result<Vec<RecordBatch>, RaftError> {
316    let mut out = Vec::new();
317    while !buf.is_empty() {
318        match RecordBatch::decode(&mut buf) {
319            Ok(batch) => out.push(batch),
320            Err(e) => {
321                return Err(RaftError::ChangeRejected(format!(
322                    "decode replicated batch: {e}"
323                )));
324            }
325        }
326    }
327    Ok(out)
328}
329
330/// The offset of the last record `batch` contains: Kafka's
331/// `lastContainedOffset` for the image that applies it.
332#[must_use]
333pub fn batch_last_offset(batch: &RecordBatch) -> i64 {
334    batch
335        .base_offset
336        .saturating_add(i64::from(batch.last_offset_delta))
337}
338
339/// Whether applying `record` gives the image a features delta, which moves
340/// the finalized-features epoch of Kafka's `FeaturesPublisher` even when the
341/// level is unchanged.
342#[must_use]
343pub fn is_features_delta(record: &MetadataRecord) -> bool {
344    matches!(record, MetadataRecord::V1FeatureLevel(_))
345}
346
347/// The finalized-features epoch after loading a snapshot whose records are
348/// `records` and whose end offset is `end_offset`: the snapshot's last
349/// contained offset when it carries a `FeatureLevelRecord`, and `previous`
350/// otherwise.
351#[must_use]
352pub fn features_epoch_after_snapshot(
353    records: &[MetadataRecord],
354    end_offset: i64,
355    previous: i64,
356) -> i64 {
357    if records.iter().any(is_features_delta) {
358        end_offset.saturating_sub(1)
359    } else {
360        previous
361    }
362}
363
364pub fn next_batch_offset(batches: &[RecordBatch]) -> Option<Offset> {
365    batches.last().map(|batch| {
366        Offset(
367            batch
368                .base_offset
369                .saturating_add(i64::from(batch.last_offset_delta))
370                .saturating_add(1),
371        )
372    })
373}
374
375#[cfg(test)]
376mod metadata_record_batch_tests {
377    use {
378        super::*, assert2::check, krabka_protocol::owned::k_raft_version_record::KRaftVersionRecord,
379    };
380
381    #[yare::parameterized(
382        empty = { Vec::new(), Vec::new() },
383        single = { vec![bytes::Bytes::from_static(b"a")], vec![0] },
384        multiple = {
385            vec![
386                bytes::Bytes::from_static(b"a"),
387                bytes::Bytes::from_static(b"b"),
388                bytes::Bytes::from_static(b"c"),
389            ],
390            vec![0, 1, 2],
391        },
392    )]
393    fn metadata_record_offset_deltas_cover_empty_single_and_multiple_batches(
394        blobs: Vec<bytes::Bytes>,
395        expected_deltas: Vec<i32>,
396    ) {
397        let batch = metadata_record_batch(5, &blobs).expect("metadata batch");
398        check!(batch.partition_leader_epoch == 5);
399        check!(
400            batch
401                .records
402                .iter()
403                .map(|record| record.offset_delta)
404                .collect::<Vec<_>>()
405                == expected_deltas
406        );
407        check!(batch.last_offset_delta == expected_deltas.last().copied().unwrap_or(0));
408    }
409
410    #[test]
411    fn metadata_control_batch_keeps_the_leader_epoch() {
412        let control = typed_control_batch(7, &[ControlRecord::KRaftVersion(
413            KRaftVersionRecord::default(),
414        )])
415        .expect("control batch");
416        check!(control.partition_leader_epoch == 7);
417    }
418}
419
420#[cfg(test)]
421mod features_epoch_tests {
422    use {
423        super::*,
424        crate::test_support::single_partition_topic,
425        assert2::check,
426        krabka_metadata::{FeatureLevelRecord, metadata_version::METADATA_VERSION_FEATURE},
427    };
428
429    fn feature() -> MetadataRecord {
430        MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
431            name: METADATA_VERSION_FEATURE.into(),
432            level: 25,
433        })
434    }
435
436    fn topic() -> MetadataRecord {
437        MetadataRecord::V1Topic(single_partition_topic("a", uuid::Uuid::from_u128(1)))
438    }
439
440    /// A snapshot load is a features delta when it carries a
441    /// `FeatureLevelRecord`. The epoch is then the snapshot's last contained
442    /// offset, one below its end offset; otherwise it stays where it was.
443    #[yare::parameterized(
444        features_from_initial = { vec![feature(), topic()], 10, -1, 9 },
445        features_replace_earlier = { vec![topic(), feature()], 10, 4, 9 },
446        no_features_keep_initial = { vec![topic()], 10, -1, -1 },
447        no_features_keep_earlier = { vec![topic()], 10, 4, 4 },
448        empty_keeps_earlier = { Vec::new(), 10, 4, 4 },
449    )]
450    fn snapshot_load_moves_the_epoch_only_with_feature_records(
451        records: Vec<MetadataRecord>,
452        end_offset: i64,
453        previous: i64,
454        want: i64,
455    ) {
456        check!(features_epoch_after_snapshot(&records, end_offset, previous) == want);
457    }
458}
krabka_raft/kraft/controller/records.rs

krabka_raft/kraft/controller/
records.rs
