krabka_remote_storage/diskless/
object.rs

1//! CKWL object framing, independent of capture selection and trust policy.
2
3use {
4    bytes::{BufMut, Bytes, BytesMut},
5    uuid::Uuid,
6};
7
8const MAGIC: [u8; 4] = *b"CKWL";
9const OBJECT_VERSION: u16 = 1;
10const HEADER_LEN: usize = 6;
11const TRAILER_LEN: usize = 8;
12const OBJECT_ENTRY_LEN: usize = 48;
13
14/// One run recorded by a CKWL footer.
15#[derive(Debug, Clone, PartialEq, Eq)]
16pub struct WalObjectEntry {
17    pub topic_id: Uuid,
18    pub partition: i32,
19    pub first_offset: i64,
20    pub last_offset: i64,
21    pub byte_start: u64,
22    pub byte_len: u32,
23}
24
25/// Invalid combined diskless-WAL object framing.
26#[derive(Debug, thiserror::Error)]
27pub enum WalObjectError {
28    /// Object does not contain the minimum header and trailer.
29    #[error("wal object too short")]
30    TooShort,
31    /// Header or trailer magic is invalid.
32    #[error("bad wal object magic")]
33    BadMagic,
34    /// Object uses an unsupported framing version.
35    #[error("unsupported wal object version {0}")]
36    BadVersion(u16),
37    /// Footer length, entry encoding, or byte ranges are invalid.
38    #[error("corrupt wal object manifest")]
39    BadManifest,
40    /// A run or the footer is too large for the CKWL `u32` length fields.
41    #[error("wal object exceeds the CKWL 4 GiB length limit")]
42    TooLarge,
43}
44
45/// Builder for combined CKWL objects.
46#[derive(Default)]
47pub struct WalObjectBuilder {
48    body: BytesMut,
49    entries: Vec<WalObjectEntry>,
50}
51
52impl WalObjectBuilder {
53    #[must_use]
54    pub fn new() -> Self { Self::default() }
55
56    #[must_use]
57    pub fn body_len(&self) -> usize { self.body.len() }
58
59    #[must_use]
60    pub fn is_empty(&self) -> bool { self.entries.is_empty() }
61
62    /// Append one contiguous partition run.
63    ///
64    /// # Errors
65    /// Returns [`WalObjectError::TooLarge`], and leaves the builder unchanged,
66    /// if the run does not fit the CKWL `u32` length field.
67    pub fn append_run(
68        &mut self,
69        topic_id: Uuid,
70        partition: i32,
71        first_offset: i64,
72        last_offset: i64,
73        run: &[u8],
74    ) -> Result<(), WalObjectError> {
75        let byte_start =
76            u64::try_from(HEADER_LEN + self.body.len()).map_err(|_| WalObjectError::TooLarge)?;
77        let byte_len = u32::try_from(run.len()).map_err(|_| WalObjectError::TooLarge)?;
78        self.body.extend_from_slice(run);
79        self.entries.push(WalObjectEntry {
80            topic_id,
81            partition,
82            first_offset,
83            last_offset,
84            byte_start,
85            byte_len,
86        });
87        Ok(())
88    }
89
90    /// Frame the appended runs as one CKWL object.
91    ///
92    /// # Errors
93    /// Returns [`WalObjectError::TooLarge`] if the footer does not fit the
94    /// CKWL `u32` length field. No partial object is returned.
95    pub fn finish(self) -> Result<Bytes, WalObjectError> {
96        let footer_len = self
97            .entries
98            .len()
99            .checked_mul(OBJECT_ENTRY_LEN)
100            .and_then(|len| u32::try_from(len).ok())
101            .ok_or(WalObjectError::TooLarge)?;
102        let mut out = BytesMut::with_capacity(
103            HEADER_LEN + self.body.len() + self.entries.len() * OBJECT_ENTRY_LEN + TRAILER_LEN,
104        );
105        out.extend_from_slice(&MAGIC);
106        out.put_u16_le(OBJECT_VERSION);
107        out.extend_from_slice(&self.body);
108        for entry in self.entries {
109            out.extend_from_slice(entry.topic_id.as_bytes());
110            out.put_i32_le(entry.partition);
111            out.put_i64_le(entry.first_offset);
112            out.put_i64_le(entry.last_offset);
113            out.put_u64_le(entry.byte_start);
114            out.put_u32_le(entry.byte_len);
115        }
116        out.put_u32_le(footer_len);
117        out.extend_from_slice(&MAGIC);
118        Ok(out.freeze())
119    }
120
121    /// Append one run and finish the object.
122    ///
123    /// # Errors
124    /// Returns [`WalObjectError::TooLarge`] if the run or footer does not fit
125    /// the CKWL `u32` length fields.
126    pub fn finish_with_run(
127        mut self,
128        topic_id: Uuid,
129        partition: i32,
130        first_offset: i64,
131        last_offset: i64,
132        run: &[u8],
133    ) -> Result<Bytes, WalObjectError> {
134        self.append_run(topic_id, partition, first_offset, last_offset, run)?;
135        self.finish()
136    }
137}
138
139/// Parse and validate a CKWL footer.
140///
141/// # Errors
142/// Returns an error for malformed framing, entries, versions, or byte ranges.
143pub fn parse_wal_object(object: &Bytes) -> Result<Vec<WalObjectEntry>, WalObjectError> {
144    if object.len() < HEADER_LEN + TRAILER_LEN {
145        return Err(WalObjectError::TooShort);
146    }
147    let (Some(&[m0, m1, m2, m3, v0, v1]), Some(&[l0, l1, l2, l3, t0, t1, t2, t3])) = (
148        object.first_chunk::<HEADER_LEN>(),
149        object.last_chunk::<TRAILER_LEN>(),
150    ) else {
151        return Err(WalObjectError::TooShort);
152    };
153    if [m0, m1, m2, m3] != MAGIC || [t0, t1, t2, t3] != MAGIC {
154        return Err(WalObjectError::BadMagic);
155    }
156    let version = u16::from_le_bytes([v0, v1]);
157    if version != OBJECT_VERSION {
158        return Err(WalObjectError::BadVersion(version));
159    }
160    let trailer = object.len() - TRAILER_LEN;
161    let footer_len = usize::try_from(u32::from_le_bytes([l0, l1, l2, l3]))
162        .map_err(|_| WalObjectError::BadManifest)?;
163    if footer_len % OBJECT_ENTRY_LEN != 0 {
164        return Err(WalObjectError::BadManifest);
165    }
166    let footer = trailer
167        .checked_sub(footer_len)
168        .filter(|start| *start >= HEADER_LEN)
169        .ok_or(WalObjectError::BadManifest)?;
170    let mut entries = Vec::with_capacity(footer_len / OBJECT_ENTRY_LEN);
171    let footer_bytes = object
172        .get(footer..trailer)
173        .ok_or(WalObjectError::BadManifest)?;
174    for raw in footer_bytes.as_chunks::<OBJECT_ENTRY_LEN>().0 {
175        let entry = WalObjectEntry {
176            topic_id: Uuid::from_slice(&raw[..16]).map_err(|_| WalObjectError::BadManifest)?,
177            partition: i32::from_le_bytes(
178                raw[16..20]
179                    .try_into()
180                    .map_err(|_| WalObjectError::BadManifest)?,
181            ),
182            first_offset: i64::from_le_bytes(
183                raw[20..28]
184                    .try_into()
185                    .map_err(|_| WalObjectError::BadManifest)?,
186            ),
187            last_offset: i64::from_le_bytes(
188                raw[28..36]
189                    .try_into()
190                    .map_err(|_| WalObjectError::BadManifest)?,
191            ),
192            byte_start: u64::from_le_bytes(
193                raw[36..44]
194                    .try_into()
195                    .map_err(|_| WalObjectError::BadManifest)?,
196            ),
197            byte_len: u32::from_le_bytes(
198                raw[44..48]
199                    .try_into()
200                    .map_err(|_| WalObjectError::BadManifest)?,
201            ),
202        };
203        let start = usize::try_from(entry.byte_start).map_err(|_| WalObjectError::BadManifest)?;
204        let end = start
205            .checked_add(usize::try_from(entry.byte_len).map_err(|_| WalObjectError::BadManifest)?)
206            .filter(|end| *end <= footer)
207            .ok_or(WalObjectError::BadManifest)?;
208        let _ = end;
209        if start < HEADER_LEN {
210            return Err(WalObjectError::BadManifest);
211        }
212        entries.push(entry);
213    }
214    Ok(entries)
215}
krabka_remote_storage/diskless/object.rs

krabka_remote_storage/diskless/
object.rs
