krabka_log/
barrier_marker.rs

1//! The barrier marker control record, shared by the broker and log readers.
2
3use {
4    crate::BARRIER_CONTROL_TYPE,
5    krabka_protocol::{
6        ProtocolError,
7        primitives::{
8            fixed::{get_i16, get_i64},
9            string_bytes::get_string_owned,
10        },
11        records::Record,
12    },
13};
14
15/// The contents of one barrier marker.
16#[derive(Debug, Clone, PartialEq, Eq)]
17pub struct BarrierMarker {
18    /// The barrier group this marker belongs to.
19    pub group: String,
20    /// The epoch this marker stamps.
21    pub epoch: i64,
22    /// When the coordinator started the injection, in milliseconds since the
23    /// Unix epoch.
24    pub triggered_at: i64,
25}
26
27/// Read the contents of a barrier marker back out of its record.
28///
29/// # Errors
30/// Returns [`ProtocolError`] when the record carries no key or no value,
31/// when the key is not a barrier control-record key, when either the key
32/// version or the value version is unsupported, or when the value is
33/// truncated, holds a non-UTF-8 group name, or has trailing bytes.
34pub fn parse_barrier_marker(record: &Record) -> Result<BarrierMarker, ProtocolError> {
35    let key = record
36        .key
37        .as_deref()
38        .ok_or(ProtocolError::InvalidValue("barrier marker has no key"))?;
39    let mut cur = key;
40    if get_i16(&mut cur)? != 0 {
41        return Err(ProtocolError::InvalidValue(
42            "unsupported control record key version",
43        ));
44    }
45    if get_i16(&mut cur)? != BARRIER_CONTROL_TYPE {
46        return Err(ProtocolError::InvalidValue(
47            "control record is not a barrier marker",
48        ));
49    }
50
51    let value = record
52        .value
53        .as_deref()
54        .ok_or(ProtocolError::InvalidValue("barrier marker has no value"))?;
55    let mut cur = value;
56    if get_i16(&mut cur)? != 0 {
57        return Err(ProtocolError::InvalidValue(
58            "unsupported barrier marker version",
59        ));
60    }
61    let group = get_string_owned(&mut cur)?;
62    let epoch = get_i64(&mut cur)?;
63    let triggered_at = get_i64(&mut cur)?;
64    if !cur.is_empty() {
65        return Err(ProtocolError::InvalidValue(
66            "trailing bytes after barrier marker",
67        ));
68    }
69    Ok(BarrierMarker {
70        group,
71        epoch,
72        triggered_at,
73    })
74}
75
76#[cfg(test)]
77mod tests {
78    use {super::*, assert2::assert, bytes::Bytes, krabka_protocol::records::Record};
79
80    fn record(group: &str, epoch: i64, triggered_at: i64) -> Record {
81        let mut key = Vec::with_capacity(4);
82        key.extend_from_slice(&0i16.to_be_bytes());
83        key.extend_from_slice(&BARRIER_CONTROL_TYPE.to_be_bytes());
84
85        let mut value = Vec::with_capacity(20 + group.len());
86        value.extend_from_slice(&0i16.to_be_bytes());
87        value.extend_from_slice(
88            &i16::try_from(group.len())
89                .expect("test group length fits marker")
90                .to_be_bytes(),
91        );
92        value.extend_from_slice(group.as_bytes());
93        value.extend_from_slice(&epoch.to_be_bytes());
94        value.extend_from_slice(&triggered_at.to_be_bytes());
95        Record {
96            key: Some(Bytes::from(key)),
97            value: Some(Bytes::from(value)),
98            ..Record::default()
99        }
100    }
101
102    #[test]
103    fn reads_a_barrier_marker_from_its_control_record() {
104        let marker = parse_barrier_marker(&record("orders-cut", 7, 1_724_500_000_000))
105            .expect("valid marker");
106        assert!(
107            marker
108                == BarrierMarker {
109                    group: "orders-cut".to_owned(),
110                    epoch: 7,
111                    triggered_at: 1_724_500_000_000,
112                }
113        );
114    }
115
116    #[test]
117    fn refuses_a_record_without_a_key() {
118        let error = parse_barrier_marker(&Record::default()).expect_err("key is required");
119        assert!(matches!(
120            error,
121            ProtocolError::InvalidValue("barrier marker has no key")
122        ));
123    }
124}
krabka_log/barrier_marker.rs

krabka_log/
barrier_marker.rs
