xlog_decoder/core/
protocol.rs

1use std::sync::atomic::{AtomicU16, Ordering};
2
3use thiserror::Error;
4
5/// Byte length of one encoded xlog header.
6pub const HEADER_LEN: usize = 1 + 2 + 1 + 1 + 4 + 64;
7/// Byte offset of the encoded end-hour field.
8pub const HEADER_END_HOUR_OFFSET: usize = 4;
9/// Byte offset of the encoded payload-length field.
10pub const HEADER_LEN_OFFSET: usize = 5;
11/// Byte offset of the encoded client public key field.
12pub const HEADER_CLIENT_PUBKEY_OFFSET: usize = 9;
13/// Byte length of the trailing end marker.
14pub const TAILER_LEN: usize = 1;
15/// Tail marker terminating a complete xlog block.
16pub const MAGIC_END: u8 = 0x00;
17
18/// Magic byte for sync + zlib + encrypted blocks.
19pub const MAGIC_SYNC_ZLIB_START: u8 = 0x06;
20/// Magic byte for sync + zlib + plaintext blocks.
21pub const MAGIC_SYNC_NO_CRYPT_ZLIB_START: u8 = 0x08;
22/// Magic byte for async + zlib + encrypted blocks.
23pub const MAGIC_ASYNC_ZLIB_START: u8 = 0x07;
24/// Magic byte for async + zlib + plaintext blocks.
25pub const MAGIC_ASYNC_NO_CRYPT_ZLIB_START: u8 = 0x09;
26
27/// Magic byte for sync + zstd + encrypted blocks.
28pub const MAGIC_SYNC_ZSTD_START: u8 = 0x0A;
29/// Magic byte for sync + zstd + plaintext blocks.
30pub const MAGIC_SYNC_NO_CRYPT_ZSTD_START: u8 = 0x0B;
31/// Magic byte for async + zstd + encrypted blocks.
32pub const MAGIC_ASYNC_ZSTD_START: u8 = 0x0C;
33/// Magic byte for async + zstd + plaintext blocks.
34pub const MAGIC_ASYNC_NO_CRYPT_ZSTD_START: u8 = 0x0D;
35
36#[derive(Debug, Copy, Clone, PartialEq, Eq)]
37/// Compression family encoded into the xlog magic byte.
38pub enum CompressionKind {
39    /// zlib-framed payloads.
40    Zlib,
41    /// zstd-framed payloads.
42    Zstd,
43}
44
45#[derive(Debug, Copy, Clone, PartialEq, Eq)]
46/// Append path encoded into the xlog magic byte.
47pub enum AppendMode {
48    /// Synchronous write path.
49    Sync,
50    /// Asynchronous buffered write path.
51    Async,
52}
53
54#[derive(Debug, Copy, Clone, PartialEq, Eq)]
55/// Decoded header for one xlog block.
56pub struct LogHeader {
57    /// Magic byte describing compression, append mode, and crypto usage.
58    pub magic: u8,
59    /// Sequence number. Async blocks increment and skip zero; sync blocks use zero.
60    pub seq: u16,
61    /// Begin hour in local time.
62    pub begin_hour: u8,
63    /// End hour in local time.
64    pub end_hour: u8,
65    /// Payload byte length, excluding header and tail marker.
66    pub len: u32,
67    /// Optional client public key bytes used by encrypted logs.
68    pub client_pubkey: [u8; 64],
69}
70
71#[derive(Debug, Error)]
72/// Errors returned by header decode/mutation helpers.
73pub enum ProtocolError {
74    #[error("invalid header length")]
75    /// The provided buffer is shorter than [`HEADER_LEN`].
76    InvalidHeaderLen,
77    #[error("invalid magic byte: {0:#x}")]
78    /// The header used a start magic byte outside the supported xlog set.
79    InvalidMagic(u8),
80}
81
82impl LogHeader {
83    /// Encode this header into the on-disk xlog header layout.
84    pub fn encode(self) -> [u8; HEADER_LEN] {
85        let mut out = [0u8; HEADER_LEN];
86        out[0] = self.magic;
87        out[1..3].copy_from_slice(&self.seq.to_le_bytes());
88        out[3] = self.begin_hour;
89        out[HEADER_END_HOUR_OFFSET] = self.end_hour;
90        out[HEADER_LEN_OFFSET..HEADER_LEN_OFFSET + 4].copy_from_slice(&self.len.to_le_bytes());
91        out[HEADER_CLIENT_PUBKEY_OFFSET..HEADER_CLIENT_PUBKEY_OFFSET + 64]
92            .copy_from_slice(&self.client_pubkey);
93        out
94    }
95
96    /// Decode one xlog header from the start of `buf`.
97    pub fn decode(buf: &[u8]) -> Result<Self, ProtocolError> {
98        if buf.len() < HEADER_LEN {
99            return Err(ProtocolError::InvalidHeaderLen);
100        }
101        let magic = buf[0];
102        if !magic_start_is_valid(magic) {
103            return Err(ProtocolError::InvalidMagic(magic));
104        }
105
106        let mut key = [0u8; 64];
107        key.copy_from_slice(&buf[HEADER_CLIENT_PUBKEY_OFFSET..HEADER_CLIENT_PUBKEY_OFFSET + 64]);
108        Ok(Self {
109            magic,
110            seq: u16::from_le_bytes([buf[1], buf[2]]),
111            begin_hour: buf[3],
112            end_hour: buf[HEADER_END_HOUR_OFFSET],
113            len: u32::from_le_bytes([
114                buf[HEADER_LEN_OFFSET],
115                buf[HEADER_LEN_OFFSET + 1],
116                buf[HEADER_LEN_OFFSET + 2],
117                buf[HEADER_LEN_OFFSET + 3],
118            ]),
119            client_pubkey: key,
120        })
121    }
122}
123
124/// Select the xlog magic byte for a `(compression, append mode, crypto)` tuple.
125pub fn select_magic(compress: CompressionKind, mode: AppendMode, crypt: bool) -> u8 {
126    match (compress, mode, crypt) {
127        (CompressionKind::Zlib, AppendMode::Sync, true) => MAGIC_SYNC_ZLIB_START,
128        (CompressionKind::Zlib, AppendMode::Sync, false) => MAGIC_SYNC_NO_CRYPT_ZLIB_START,
129        (CompressionKind::Zlib, AppendMode::Async, true) => MAGIC_ASYNC_ZLIB_START,
130        (CompressionKind::Zlib, AppendMode::Async, false) => MAGIC_ASYNC_NO_CRYPT_ZLIB_START,
131        (CompressionKind::Zstd, AppendMode::Sync, true) => MAGIC_SYNC_ZSTD_START,
132        (CompressionKind::Zstd, AppendMode::Sync, false) => MAGIC_SYNC_NO_CRYPT_ZSTD_START,
133        (CompressionKind::Zstd, AppendMode::Async, true) => MAGIC_ASYNC_ZSTD_START,
134        (CompressionKind::Zstd, AppendMode::Async, false) => MAGIC_ASYNC_NO_CRYPT_ZSTD_START,
135    }
136}
137
138/// Return whether `magic` is one of the supported xlog start markers.
139pub fn magic_start_is_valid(magic: u8) -> bool {
140    matches!(
141        magic,
142        MAGIC_SYNC_ZLIB_START
143            | MAGIC_SYNC_NO_CRYPT_ZLIB_START
144            | MAGIC_ASYNC_ZLIB_START
145            | MAGIC_ASYNC_NO_CRYPT_ZLIB_START
146            | MAGIC_SYNC_ZSTD_START
147            | MAGIC_SYNC_NO_CRYPT_ZSTD_START
148            | MAGIC_ASYNC_ZSTD_START
149            | MAGIC_ASYNC_NO_CRYPT_ZSTD_START
150    )
151}
152
153/// Add `add_len` to the payload length stored in an encoded header.
154///
155/// Returns the new payload length.
156pub fn update_log_len_in_place(buf: &mut [u8], add_len: u32) -> Result<u32, ProtocolError> {
157    if buf.len() < HEADER_LEN {
158        return Err(ProtocolError::InvalidHeaderLen);
159    }
160    let current = u32::from_le_bytes([
161        buf[HEADER_LEN_OFFSET],
162        buf[HEADER_LEN_OFFSET + 1],
163        buf[HEADER_LEN_OFFSET + 2],
164        buf[HEADER_LEN_OFFSET + 3],
165    ]);
166    let next = current.saturating_add(add_len);
167    buf[HEADER_LEN_OFFSET..HEADER_LEN_OFFSET + 4].copy_from_slice(&next.to_le_bytes());
168    Ok(next)
169}
170
171/// Update the encoded end-hour field inside an existing header.
172pub fn update_end_hour_in_place(buf: &mut [u8], hour: u8) -> Result<(), ProtocolError> {
173    if buf.len() < HEADER_LEN {
174        return Err(ProtocolError::InvalidHeaderLen);
175    }
176    buf[HEADER_END_HOUR_OFFSET] = hour;
177    Ok(())
178}
179
180/// Async sequence generator matching Mars xlog semantics.
181pub struct SeqGenerator {
182    seq: AtomicU16,
183}
184
185impl Default for SeqGenerator {
186    fn default() -> Self {
187        Self {
188            seq: AtomicU16::new(0),
189        }
190    }
191}
192
193impl SeqGenerator {
194    /// Create a generator with a fixed initial sequence state.
195    pub fn with_seed(seed: u16) -> Self {
196        Self {
197            seq: AtomicU16::new(seed),
198        }
199    }
200
201    /// Generate the next async sequence number.
202    ///
203    /// Matches the historical C++ behavior: increment first, then skip `0`.
204    /// Under wraparound races concurrent callers may observe gaps, but `0` is
205    /// never returned and values are not reused.
206    pub fn next_async(&self) -> u16 {
207        let mut next = self.seq.fetch_add(1, Ordering::Relaxed).wrapping_add(1);
208        if next == 0 {
209            next = self.seq.fetch_add(1, Ordering::Relaxed).wrapping_add(1);
210            if next == 0 {
211                next = 1;
212                self.seq.store(1, Ordering::Relaxed);
213            }
214        }
215        next
216    }
217
218    /// Return the fixed sync sequence number used by Mars xlog.
219    pub fn sync_seq() -> u16 {
220        0
221    }
222}
223
224#[cfg(test)]
225mod tests {
226    use super::{
227        update_end_hour_in_place, update_log_len_in_place, LogHeader, ProtocolError, SeqGenerator,
228        HEADER_LEN, MAGIC_ASYNC_NO_CRYPT_ZLIB_START,
229    };
230
231    #[test]
232    fn decode_rejects_short_buffers_and_invalid_magic() {
233        assert!(matches!(
234            LogHeader::decode(&[0u8; HEADER_LEN - 1]),
235            Err(ProtocolError::InvalidHeaderLen)
236        ));
237
238        let mut bytes = [0u8; HEADER_LEN];
239        bytes[0] = 0x05;
240        assert!(matches!(
241            LogHeader::decode(&bytes),
242            Err(ProtocolError::InvalidMagic(0x05))
243        ));
244    }
245
246    #[test]
247    fn in_place_updates_reject_short_buffers_and_saturate_length() {
248        let mut short = [0u8; HEADER_LEN - 1];
249        assert!(matches!(
250            update_log_len_in_place(&mut short, 1),
251            Err(ProtocolError::InvalidHeaderLen)
252        ));
253        assert!(matches!(
254            update_end_hour_in_place(&mut short, 1),
255            Err(ProtocolError::InvalidHeaderLen)
256        ));
257
258        let mut bytes = LogHeader {
259            magic: MAGIC_ASYNC_NO_CRYPT_ZLIB_START,
260            seq: 1,
261            begin_hour: 2,
262            end_hour: 2,
263            len: u32::MAX - 1,
264            client_pubkey: [0; 64],
265        }
266        .encode();
267        assert_eq!(update_log_len_in_place(&mut bytes, 10).unwrap(), u32::MAX);
268        update_end_hour_in_place(&mut bytes, 9).unwrap();
269        let decoded = LogHeader::decode(&bytes).unwrap();
270        assert_eq!(decoded.len, u32::MAX);
271        assert_eq!(decoded.end_hour, 9);
272    }
273
274    #[test]
275    fn seq_generator_wraps_without_emitting_zero() {
276        let seq = SeqGenerator::with_seed(u16::MAX - 1);
277        assert_eq!(seq.next_async(), u16::MAX);
278        assert_eq!(seq.next_async(), 1);
279        assert_eq!(seq.next_async(), 2);
280        assert_eq!(SeqGenerator::sync_seq(), 0);
281    }
282}
xlog_decoder/core/protocol.rs

xlog_decoder/core/
protocol.rs
