krabka_hlc/
stamp.rs

1//! The hybrid logical clock stamp and its fixed-width byte layout.
2
3use derive_more::{Display, From, Into};
4use krabka_ids::NodeId;
5
6use crate::{error::HlcError, headers::HLC_HEADER};
7
8/// The byte length of an encoded [`Hlc`].
9///
10/// The layout is fixed-width, so this length is the same for every stamp: 8
11/// bytes of wall time, 4 bytes of counter, and 4 bytes of node id.
12pub const HLC_ENCODED_LEN: usize = 16;
13
14/// A wall-clock instant, in microseconds since the Unix epoch.
15///
16/// This is a coordinate on the time line and not an extent of time, so it is a
17/// newtype here and not a [`krabka_units::Time`]. See the code style guide's
18/// "Dimensioned Values" section for the rule.
19///
20/// The unit is the microsecond. A Kafka record timestamp counts milliseconds,
21/// and one millisecond is too coarse to order the records inside one batch. A
22/// microsecond gives the logical counter far fewer events to separate.
23///
24/// The value is signed, and a value below zero names an instant before the
25/// Unix epoch. The clock accepts such a value and does not treat it as an
26/// error, because a host whose clock is unset reports one.
27#[derive(
28    Debug, Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord, Hash, Display, From, Into,
29)]
30pub struct WallMicros(pub i64);
31
32impl WallMicros {
33    /// The Unix epoch, `1970-01-01T00:00:00Z`.
34    pub const EPOCH: Self = Self(0);
35
36    /// The inner `i64`. Use it for arithmetic against other time counts.
37    #[must_use]
38    pub const fn get(self) -> i64 {
39        self.0
40    }
41}
42
43/// One hybrid logical clock stamp: a wall time, a logical counter, and the
44/// node that minted it.
45///
46/// The three fields give a total order over every event in the cluster.
47/// [`Ord`] compares the wall time first, then the counter, then the node id.
48/// The wall time keeps the order close to real time. The counter separates
49/// events that share a microsecond. The node id breaks the remaining ties, so
50/// two nodes that stamp the same wall time and the same counter still get one
51/// order, and every node computes that same order.
52///
53/// The type is `Copy`, and it is 16 bytes wide in memory and 16 bytes wide on
54/// the wire.
55#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Display)]
56#[display("{wall}.{counter}@{node}")]
57pub struct Hlc {
58    /// The wall time this stamp claims.
59    pub wall: WallMicros,
60
61    /// The logical counter that separates events inside one microsecond.
62    pub counter: u32,
63
64    /// The node that minted the stamp.
65    pub node: NodeId,
66}
67
68impl Hlc {
69    /// A stamp built from its three parts.
70    #[must_use]
71    pub const fn new(wall: WallMicros, counter: u32, node: NodeId) -> Self {
72        Self {
73            wall,
74            counter,
75            node,
76        }
77    }
78
79    /// The stamp as the exact bytes of the `krabka.hlc` header value.
80    ///
81    /// The layout is big-endian and fixed-width: 8 bytes of wall time as an
82    /// `i64`, 4 bytes of counter as a `u32`, and 4 bytes of node id as a
83    /// `u32`. There is no varint framing and no length prefix. Java and Go
84    /// clients read this header by hand, and a plain big-endian layout is what
85    /// `DataInputStream.readLong` and `binary.BigEndian.Uint32` already give
86    /// them.
87    ///
88    /// # Errors
89    ///
90    /// Returns [`HlcError::NodeIdTooLarge`] when the node id is above
91    /// `u32::MAX`, which is wider than the 4-byte field the layout gives it.
92    pub fn encode(self) -> Result<[u8; HLC_ENCODED_LEN], HlcError> {
93        let node = u32::try_from(self.node.get())
94            .map_err(|_| HlcError::NodeIdTooLarge(self.node.get()))?;
95        let wall = self.wall.get().to_be_bytes();
96        let counter = self.counter.to_be_bytes();
97        let node = node.to_be_bytes();
98        Ok([
99            wall[0], wall[1], wall[2], wall[3], wall[4], wall[5], wall[6], wall[7], counter[0],
100            counter[1], counter[2], counter[3], node[0], node[1], node[2], node[3],
101        ])
102    }
103
104    /// The stamp that `bytes` carries, with full validation.
105    ///
106    /// This function reads a peer-supplied header value, so it validates the
107    /// length before it reads a field. It never panics on malformed input, and
108    /// it never allocates. Every 16-byte input is a valid stamp, because each
109    /// of the three fields covers the whole range of its type.
110    ///
111    /// Use [`extract_from_headers`](crate::extract_from_headers) on an ingress
112    /// path, where a bad value must not fail the record that carried it. Use
113    /// this function where the caller wants to know why a value failed.
114    ///
115    /// # Errors
116    ///
117    /// Returns [`HlcError::Length`] when `bytes` is not exactly
118    /// [`HLC_ENCODED_LEN`] bytes long.
119    pub fn decode(bytes: &[u8]) -> Result<Self, HlcError> {
120        let Ok(value) = <&[u8; HLC_ENCODED_LEN]>::try_from(bytes) else {
121            return Err(HlcError::Length(bytes.len()));
122        };
123        // A slice pattern over a fixed-width array. It splits the value into
124        // its three fields with no indexing, no `copy_from_slice`, and no
125        // conversion that can fail, so no path here can panic.
126        let [wall @ .., c0, c1, c2, c3, n0, n1, n2, n3] = *value;
127        Ok(Self {
128            wall: WallMicros(i64::from_be_bytes(wall)),
129            counter: u32::from_be_bytes([c0, c1, c2, c3]),
130            node: NodeId(u64::from(u32::from_be_bytes([n0, n1, n2, n3]))),
131        })
132    }
133
134    /// The stamp as a Kafka record header: the [`HLC_HEADER`] key and the
135    /// encoded value.
136    ///
137    /// A producer attaches the pair to the record it writes. A record that
138    /// carries no stamp gets no header, so an unstamped record costs zero
139    /// bytes.
140    ///
141    /// # Errors
142    ///
143    /// Returns [`HlcError::NodeIdTooLarge`] when the node id is above
144    /// `u32::MAX`. See [`Hlc::encode`].
145    pub fn header(self) -> Result<(&'static str, [u8; HLC_ENCODED_LEN]), HlcError> {
146        Ok((HLC_HEADER, self.encode()?))
147    }
148}
krabka_hlc/stamp.rs

krabka_hlc/
stamp.rs
