krabka_remote_storage/diskless/
index.rs

1//! Compacted diskless index-topic keys and values.
2
3use {
4    super::WalIndexEntry,
5    bytes::Bytes,
6    serde::{Deserialize, Serialize},
7    uuid::Uuid,
8};
9
10/// Stable keyed replay fence ignored by every diskless index projection.
11pub const REPLAY_FENCE_KEY: &[u8] = b"__krabka_diskless_replay_fence";
12
13/// Compaction key for one logical range.
14///
15/// Every `__diskless_wal_index` key starts with a big-endian `i16` key
16/// version. As in Kafka's coordinator records, the key version names the
17/// record type the key and its value belong to, so a reader dispatches on it
18/// before it looks at anything else and a key type it does not know is
19/// refused instead of being decoded as another type.
20#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
21pub struct WalIndexKey {
22    pub topic_id: Uuid,
23    pub partition: i32,
24    pub first_offset: i64,
25}
26
27impl WalIndexKey {
28    /// Key version of a range key, whose value is a [`WalFlushRecord`].
29    ///
30    /// Part of the 1.x on-disk contract: a 1.x broker reads every range key
31    /// that any earlier 1.x broker wrote.
32    pub const KEY_VERSION: i16 = 0;
33    const LEN: usize = PARTITION_KEY_PREFIX_LEN + 8;
34
35    #[must_use]
36    pub fn to_bytes(self) -> Bytes {
37        let mut out = encode_partition_key_prefix::<{ Self::LEN }>(
38            Self::KEY_VERSION,
39            self.topic_id,
40            self.partition,
41        );
42        out.extend_from_slice(&self.first_offset.to_be_bytes());
43        out.into()
44    }
45
46    /// Decode a range key, or `None` when `bytes` are not one: another key
47    /// version, or the wrong length.
48    #[must_use]
49    pub fn from_bytes(bytes: &[u8]) -> Option<Self> {
50        let PartitionKeyPrefix {
51            topic_id,
52            partition,
53            rest,
54        } = decode_partition_key_prefix::<{ Self::LEN }>(bytes, Self::KEY_VERSION)?;
55        Some(Self {
56            topic_id,
57            partition,
58            first_offset: i64::from_be_bytes(rest.try_into().ok()?),
59        })
60    }
61}
62
63/// The leading key version of a `__diskless_wal_index` key, or `None` for a
64/// key shorter than the version itself.
65#[must_use]
66pub fn wal_index_key_version(key: &[u8]) -> Option<i16> {
67    Some(i16::from_be_bytes(key.get(..2)?.try_into().ok()?))
68}
69
70/// Length of the big-endian `i16` key version, topic id and big-endian `i32`
71/// partition that every partition-scoped `__diskless_wal_index` key starts
72/// with.
73const PARTITION_KEY_PREFIX_LEN: usize = 22;
74
75/// Start a `KEY_LEN`-byte partition-scoped key with its version, topic id and
76/// partition.
77///
78/// The caller appends the rest of the key, if any.
79fn encode_partition_key_prefix<const KEY_LEN: usize>(
80    version: i16,
81    topic_id: Uuid,
82    partition: i32,
83) -> Vec<u8> {
84    let mut out = Vec::with_capacity(KEY_LEN);
85    out.extend_from_slice(&version.to_be_bytes());
86    out.extend_from_slice(topic_id.as_bytes());
87    out.extend_from_slice(&partition.to_be_bytes());
88    out
89}
90
91/// A partition-scoped key split into its topic id, partition and trailing bytes.
92struct PartitionKeyPrefix<'a> {
93    topic_id: Uuid,
94    partition: i32,
95    rest: &'a [u8],
96}
97
98/// Split a `KEY_LEN`-byte partition-scoped key into its topic id, partition and
99/// the bytes after them, or `None` when `bytes` are the wrong length or carry a
100/// key version other than `version`.
101fn decode_partition_key_prefix<const KEY_LEN: usize>(
102    bytes: &[u8],
103    version: i16,
104) -> Option<PartitionKeyPrefix<'_>> {
105    let bytes: &[u8; KEY_LEN] = bytes.try_into().ok()?;
106    if wal_index_key_version(bytes) != Some(version) {
107        return None;
108    }
109    let topic_id = Uuid::from_bytes(bytes.get(2..18)?.try_into().ok()?);
110    let partition = i32::from_be_bytes(bytes.get(18..PARTITION_KEY_PREFIX_LEN)?.try_into().ok()?);
111    Some(PartitionKeyPrefix {
112        topic_id,
113        partition,
114        rest: bytes.get(PARTITION_KEY_PREFIX_LEN..)?,
115    })
116}
117
118impl From<&WalIndexEntry> for WalIndexKey {
119    fn from(value: &WalIndexEntry) -> Self {
120        Self {
121            topic_id: value.topic_id,
122            partition: value.partition,
123            first_offset: value.first_offset,
124        }
125    }
126}
127
128/// Compaction key for a partition delete floor.
129#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
130pub struct WalDeleteFloorKey {
131    pub topic_id: Uuid,
132    pub partition: i32,
133}
134
135/// Reserved delete-floor partition used to protect an uploaded object until
136/// its range records are committed.
137///
138/// Its key/value remain readable by 1.x
139/// brokers: older readers project it as floor zero on a negative partition,
140/// which no data partition can address.
141pub const WAL_OBJECT_RESERVATION_PARTITION: i32 = i32::MIN;
142
143/// Object UUID encoded into the existing delete-floor key space.
144///
145/// Diskless object names use `UUIDv4`. Changing only the version nibble to v8
146/// creates a reversible namespace that cannot collide with generated diskless
147/// topic IDs, while the negative partition sentinel keeps the record inert to
148/// older delete-floor projections.
149#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
150pub struct WalObjectReservationKey {
151    object_id: Uuid,
152}
153
154impl WalObjectReservationKey {
155    const KEY_VERSION: i16 = WalDeleteFloorKey::KEY_VERSION;
156
157    #[must_use]
158    pub fn new(object_id: Uuid) -> Self { Self { object_id } }
159
160    #[must_use]
161    pub fn object_id(self) -> Uuid { self.object_id }
162
163    #[must_use]
164    pub fn to_bytes(self) -> Bytes {
165        WalDeleteFloorKey {
166            topic_id: reservation_topic_id(self.object_id),
167            partition: WAL_OBJECT_RESERVATION_PARTITION,
168        }
169        .to_bytes()
170    }
171
172    #[must_use]
173    pub fn from_bytes(bytes: &[u8]) -> Option<Self> {
174        let key = WalDeleteFloorKey::from_bytes(bytes)?;
175        if key.partition != WAL_OBJECT_RESERVATION_PARTITION || key.topic_id.get_version_num() != 8
176        {
177            return None;
178        }
179        Some(Self {
180            object_id: object_id_from_reservation_topic(key.topic_id),
181        })
182    }
183
184    /// This record deliberately shares the v1 delete-floor key/value
185    /// versions, so persisted 1.x readers can still replay it.
186    #[must_use]
187    pub const fn key_version() -> i16 { Self::KEY_VERSION }
188}
189
190fn reservation_topic_id(object_id: Uuid) -> Uuid {
191    let mut bytes = *object_id.as_bytes();
192    bytes[6] = (bytes[6] & 0x0f) | 0x80;
193    Uuid::from_bytes(bytes)
194}
195
196fn object_id_from_reservation_topic(topic_id: Uuid) -> Uuid {
197    let mut bytes = *topic_id.as_bytes();
198    bytes[6] = (bytes[6] & 0x0f) | 0x40;
199    Uuid::from_bytes(bytes)
200}
201
202impl WalDeleteFloorKey {
203    /// Key version of a delete-floor key, whose value is a
204    /// [`WalDeleteFloorRecord`].
205    ///
206    /// Part of the 1.x on-disk contract: a 1.x broker reads every delete-floor
207    /// key that any earlier 1.x broker wrote.
208    pub const KEY_VERSION: i16 = 1;
209    const LEN: usize = PARTITION_KEY_PREFIX_LEN;
210
211    #[must_use]
212    pub fn to_bytes(self) -> Bytes {
213        encode_partition_key_prefix::<{ Self::LEN }>(
214            Self::KEY_VERSION,
215            self.topic_id,
216            self.partition,
217        )
218        .into()
219    }
220
221    /// Decode a delete-floor key, or `None` when `bytes` are not one: another
222    /// key version, or the wrong length.
223    #[must_use]
224    pub fn from_bytes(bytes: &[u8]) -> Option<Self> {
225        let PartitionKeyPrefix {
226            topic_id,
227            partition,
228            ..
229        } = decode_partition_key_prefix::<{ Self::LEN }>(bytes, Self::KEY_VERSION)?;
230        Some(Self {
231            topic_id,
232            partition,
233        })
234    }
235}
236
237/// Split the leading big-endian `i16` version off a `__diskless_wal_index`
238/// value, refusing any version other than `expected`.
239fn split_value_version<'a>(
240    bytes: &'a [u8],
241    record: &str,
242    expected: i16,
243) -> Result<&'a [u8], String> {
244    let (version, body) = bytes
245        .split_first_chunk::<2>()
246        .ok_or_else(|| format!("truncated {record}: {} bytes hold no version", bytes.len()))?;
247    let version = i16::from_be_bytes(*version);
248    if version != expected {
249        return Err(format!(
250            "unsupported {record} version {version}: this build reads version {expected}. \
251             A record written before krabka 1.0 has no version prefix and is not readable; \
252             reformat the cluster"
253        ));
254    }
255    Ok(body)
256}
257
258fn encode_value_version(version: i16, body: &[u8]) -> Bytes {
259    let mut out = Vec::with_capacity(2 + body.len());
260    out.extend_from_slice(&version.to_be_bytes());
261    out.extend_from_slice(body);
262    out.into()
263}
264
265/// Durable `DeleteRecords` floor.
266///
267/// The value is a big-endian `i16` [`Self::VERSION`] followed by the wincode
268/// body.
269#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
270pub struct WalDeleteFloorRecord {
271    pub topic_id: Uuid,
272    pub partition: i32,
273    pub floor: i64,
274}
275
276impl WalDeleteFloorRecord {
277    /// Value version of a delete-floor record.
278    ///
279    /// Part of the 1.x on-disk contract: a 1.x broker reads every delete-floor
280    /// record that any earlier 1.x broker wrote.
281    pub const VERSION: i16 = 0;
282
283    /// Encode with the index topic codec.
284    ///
285    /// # Errors
286    /// Returns the codec error when the record cannot be encoded.
287    pub fn to_bytes(&self) -> Result<Bytes, String> {
288        let body = <serde_wincode::SerdeCompat<Self> as wincode::Serialize>::serialize(self)
289            .map_err(|error| error.to_string())?;
290        Ok(encode_value_version(Self::VERSION, &body))
291    }
292
293    /// Decode the index topic codec.
294    ///
295    /// # Errors
296    /// Returns an error for a missing or unsupported version, or when the body
297    /// is malformed.
298    pub fn from_bytes(bytes: &[u8]) -> Result<Self, String> {
299        let body = split_value_version(bytes, "diskless WAL delete-floor record", Self::VERSION)?;
300        <serde_wincode::SerdeCompat<Self> as wincode::Deserialize>::deserialize(body)
301            .map_err(|error| error.to_string())
302    }
303}
304
305/// Durable index value for a flushed object.
306///
307/// The value is a big-endian `i16` `format_version` followed by the wincode
308/// body of `object_key` and `entries`, so a reader checks the version before it
309/// decodes anything else.
310#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
311pub struct WalFlushRecord {
312    pub object_key: String,
313    pub format_version: u16,
314    pub entries: Vec<WalIndexEntry>,
315}
316
317/// The wincode body of a [`WalFlushRecord`], as it is encoded.
318#[derive(Serialize)]
319struct WalFlushBodyRef<'a> {
320    object_key: &'a str,
321    entries: &'a [WalIndexEntry],
322}
323
324/// The wincode body of a [`WalFlushRecord`], as it is decoded.
325#[derive(Deserialize)]
326struct WalFlushBody {
327    object_key: String,
328    entries: Vec<WalIndexEntry>,
329}
330
331impl WalFlushRecord {
332    /// Value version of a flush record.
333    ///
334    /// Part of the 1.x on-disk contract: a 1.x broker reads every flush record
335    /// that any earlier 1.x broker wrote. It is encoded as a big-endian `i16`.
336    pub const FORMAT_VERSION: u16 = 2;
337
338    /// Encode one keyed WAL range in the format a flusher publishes.
339    ///
340    /// # Errors
341    /// Returns the index codec's serialization error.
342    pub fn keyed_entry(object_key: &str, entry: WalIndexEntry) -> Result<(Bytes, Bytes), String> {
343        let key = WalIndexKey::from(&entry).to_bytes();
344        let value = Self {
345            object_key: object_key.into(),
346            format_version: Self::FORMAT_VERSION,
347            entries: vec![entry],
348        }
349        .to_bytes()?;
350        Ok((key, value))
351    }
352
353    /// Encode with the index topic codec.
354    ///
355    /// # Errors
356    /// Returns an error for an unsupported version or codec failure.
357    pub fn to_bytes(&self) -> Result<Bytes, String> {
358        if self.format_version != Self::FORMAT_VERSION {
359            return Err(format!(
360                "unsupported diskless WAL index format version {}",
361                self.format_version
362            ));
363        }
364        let version = Self::wire_version();
365        let body =
366            <serde_wincode::SerdeCompat<WalFlushBodyRef<'_>> as wincode::Serialize>::serialize(
367                &WalFlushBodyRef {
368                    object_key: &self.object_key,
369                    entries: &self.entries,
370                },
371            )
372            .map_err(|error| error.to_string())?;
373        Ok(encode_value_version(version, &body))
374    }
375
376    /// Decode with strict format-version checking.
377    ///
378    /// # Errors
379    /// Returns an error for a missing or unsupported version, or malformed
380    /// bytes.
381    pub fn from_bytes(bytes: &[u8]) -> Result<Self, String> {
382        let body = split_value_version(bytes, "diskless WAL index format", Self::wire_version())?;
383        let body =
384            <serde_wincode::SerdeCompat<WalFlushBody> as wincode::Deserialize>::deserialize(body)
385                .map_err(|error| error.to_string())?;
386        Ok(Self {
387            object_key: body.object_key,
388            format_version: Self::FORMAT_VERSION,
389            entries: body.entries,
390        })
391    }
392
393    /// [`Self::FORMAT_VERSION`] as the `i16` the value leads with. The
394    /// version is far below `i16::MAX`, so the conversion keeps its value.
395    const fn wire_version() -> i16 { Self::FORMAT_VERSION.cast_signed() }
396}
krabka_remote_storage/diskless/index.rs

krabka_remote_storage/diskless/
index.rs
