rfc9162_tiles/
index.rs

1//! The index beside each sealed bundle, `tile/index/<N>`, and the manifest
2//! over them. Both derive from the bundles: a rebuild from the entries gives
3//! the same bytes, so losing them loses nothing.
4//!
5//! A sidecar holds a [`Bloom`] filter over every key of the bundle and the
6//! sorted (key, offset) pairs. The keys of an entry are its identity hash,
7//! `SHA-256(entry)`, and whatever the log's [`KeyFn`] derives from it.
8//!
9//! Sidecar bytes: `0x01`, big-endian u16 pair count, big-endian u32 filter
10//! length, the filter, then each pair as big-endian u16 key length, the key,
11//! and the one-byte offset of the entry in the bundle.
12
13use std::collections::BTreeSet;
14
15use serde::{Deserialize, Serialize};
16use sha2::{Digest, Sha256};
17
18use crate::bloom::{Bloom, BITS_PER_KEY};
19use crate::error::{Error, Result};
20use crate::layout::MANIFEST;
21use crate::store::TileStore;
22
23/// Derives the lookup keys of an entry from its bytes. The identity hash is
24/// always a key and need not be returned.
25pub type KeyFn = Box<dyn Fn(&[u8]) -> Vec<Vec<u8>> + Send + Sync>;
26
27/// A key function that adds no keys beyond the identity hash.
28#[must_use]
29pub fn no_keys() -> KeyFn {
30    Box::new(|_| Vec::new())
31}
32
33/// `SHA-256(entry)`, the key every entry has.
34#[must_use]
35pub fn identity(entry: &[u8]) -> [u8; 32] {
36    Sha256::digest(entry).into()
37}
38
39const VERSION: u8 = 1;
40
41/// The sidecar of one sealed bundle.
42#[derive(Debug, Clone, PartialEq, Eq)]
43pub struct Sidecar {
44    /// (key, offset in the bundle), sorted, without duplicates.
45    pub pairs: Vec<(Vec<u8>, u8)>,
46    /// The filter over every key in `pairs`.
47    pub filter: Bloom,
48}
49
50impl Sidecar {
51    /// The sidecar over a bundle's entries with the log's key function.
52    #[must_use]
53    pub fn build<E: AsRef<[u8]>>(entries: &[E], keys: &KeyFn) -> Self {
54        let mut set: BTreeSet<(Vec<u8>, u8)> = BTreeSet::new();
55        for (offset, entry) in entries.iter().enumerate() {
56            let entry = entry.as_ref();
57            set.insert((identity(entry).to_vec(), offset as u8));
58            for key in keys(entry) {
59                set.insert((key, offset as u8));
60            }
61        }
62        let pairs: Vec<(Vec<u8>, u8)> = set.into_iter().collect();
63        let distinct: BTreeSet<&[u8]> = pairs.iter().map(|(k, _)| k.as_slice()).collect();
64        let distinct: Vec<&[u8]> = distinct.into_iter().collect();
65        let filter = Bloom::build(&distinct, BITS_PER_KEY);
66        Self { pairs, filter }
67    }
68
69    /// Offsets of the entries carrying `key`, ascending.
70    #[must_use]
71    pub fn lookup(&self, key: &[u8]) -> Vec<u8> {
72        if !self.filter.contains(key) {
73            return Vec::new();
74        }
75        let start = self.pairs.partition_point(|(k, _)| k.as_slice() < key);
76        self.pairs[start..]
77            .iter()
78            .take_while(|(k, _)| k.as_slice() == key)
79            .map(|(_, offset)| *offset)
80            .collect()
81    }
82
83    /// The smallest and largest key, or nothing for an empty bundle.
84    #[must_use]
85    pub fn key_range(&self) -> Option<(&[u8], &[u8])> {
86        Some((
87            self.pairs.first()?.0.as_slice(),
88            self.pairs.last()?.0.as_slice(),
89        ))
90    }
91
92    /// The sidecar as bytes.
93    #[must_use]
94    pub fn to_bytes(&self) -> Vec<u8> {
95        let filter = self.filter.to_bytes();
96        let mut out = vec![VERSION];
97        out.extend_from_slice(&(self.pairs.len() as u16).to_be_bytes());
98        out.extend_from_slice(&(filter.len() as u32).to_be_bytes());
99        out.extend_from_slice(&filter);
100        for (key, offset) in &self.pairs {
101            out.extend_from_slice(&(key.len() as u16).to_be_bytes());
102            out.extend_from_slice(key);
103            out.push(*offset);
104        }
105        out
106    }
107
108    /// The sidecar `to_bytes` wrote.
109    pub fn from_bytes(path: &str, bytes: &[u8]) -> Result<Self> {
110        let corrupt = |reason: &str| Error::Corrupt {
111            path: path.to_string(),
112            reason: reason.to_string(),
113        };
114        if bytes.first() != Some(&VERSION) {
115            return Err(corrupt("not a version 1 sidecar"));
116        }
117        let count = usize::from(u16::from_be_bytes([
118            *bytes.get(1).ok_or_else(|| corrupt("cut"))?,
119            *bytes.get(2).ok_or_else(|| corrupt("cut"))?,
120        ]));
121        let filter_len = u32::from_be_bytes(
122            bytes
123                .get(3..7)
124                .ok_or_else(|| corrupt("cut"))?
125                .try_into()
126                .expect("four bytes"),
127        ) as usize;
128        let filter = Bloom::from_bytes(bytes.get(7..7 + filter_len).ok_or_else(|| corrupt("cut"))?)
129            .map_err(|e| corrupt(&e.to_string()))?;
130        let mut at = 7 + filter_len;
131        let mut pairs = Vec::with_capacity(count);
132        for _ in 0..count {
133            let len = usize::from(u16::from_be_bytes([
134                *bytes.get(at).ok_or_else(|| corrupt("cut"))?,
135                *bytes.get(at + 1).ok_or_else(|| corrupt("cut"))?,
136            ]));
137            at += 2;
138            let key = bytes
139                .get(at..at + len)
140                .ok_or_else(|| corrupt("cut"))?
141                .to_vec();
142            at += len;
143            let offset = *bytes.get(at).ok_or_else(|| corrupt("cut"))?;
144            at += 1;
145            pairs.push((key, offset));
146        }
147        if at != bytes.len() {
148            return Err(corrupt("bytes trail the last pair"));
149        }
150        if pairs.windows(2).any(|w| w[0] >= w[1]) {
151            return Err(corrupt("pairs are not sorted"));
152        }
153        Ok(Self { pairs, filter })
154    }
155}
156
157/// One manifest row: a sealed bundle, its key range and its sidecar's hash.
158#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
159pub struct ManifestRow {
160    /// The bundle index.
161    pub index: u64,
162    /// The smallest key, hex.
163    pub min: String,
164    /// The largest key, hex.
165    pub max: String,
166    /// `SHA-256` of the sidecar bytes, hex.
167    pub sidecar: String,
168    /// Pairs in the sidecar.
169    pub keys: usize,
170}
171
172/// The manifest, `manifest`: every sealed bundle's row, in order. JSON,
173/// rewritten whole on every change.
174#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
175pub struct Manifest {
176    /// Format version, 1.
177    pub version: u8,
178    /// One row per sealed bundle, by index.
179    pub bundles: Vec<ManifestRow>,
180}
181
182impl Manifest {
183    /// The manifest the store holds; empty for a store without one.
184    pub fn read(store: &impl TileStore) -> Result<Self> {
185        match store.get(MANIFEST)? {
186            None => Ok(Self {
187                version: 1,
188                bundles: Vec::new(),
189            }),
190            Some(bytes) => {
191                let manifest: Self =
192                    serde_json::from_slice(&bytes).map_err(|e| Error::Corrupt {
193                        path: MANIFEST.to_string(),
194                        reason: e.to_string(),
195                    })?;
196                if manifest.version != 1 {
197                    return Err(Error::Corrupt {
198                        path: MANIFEST.to_string(),
199                        reason: format!("version {}", manifest.version),
200                    });
201                }
202                if manifest
203                    .bundles
204                    .iter()
205                    .enumerate()
206                    .any(|(i, row)| row.index != i as u64)
207                {
208                    return Err(Error::Corrupt {
209                        path: MANIFEST.to_string(),
210                        reason: "rows are not the bundles in order".to_string(),
211                    });
212                }
213                Ok(manifest)
214            }
215        }
216    }
217
218    /// Writes the manifest, whole or not at all.
219    pub fn write(&self, store: &impl TileStore) -> Result<()> {
220        let bytes = serde_json::to_vec(self).map_err(|e| Error::Format(e.to_string()))?;
221        store.put(MANIFEST, &bytes)?;
222        Ok(())
223    }
224
225    /// The row for bundle `index` built from its sidecar bytes.
226    #[must_use]
227    pub fn row(index: u64, sidecar: &Sidecar, bytes: &[u8]) -> ManifestRow {
228        let (min, max) = sidecar
229            .key_range()
230            .map(|(min, max)| (rfc9162::hex(min), rfc9162::hex(max)))
231            .unwrap_or_default();
232        ManifestRow {
233            index,
234            min,
235            max,
236            sidecar: rfc9162::hex(&Sha256::digest(bytes)),
237            keys: sidecar.pairs.len(),
238        }
239    }
240
241    /// Rows whose key range holds `key`.
242    pub fn candidates<'a>(&'a self, key: &[u8]) -> impl Iterator<Item = &'a ManifestRow> + 'a {
243        let key = rfc9162::hex(key);
244        self.bundles
245            .iter()
246            .filter(move |row| row.min <= key && key <= row.max)
247    }
248}
249
250#[cfg(test)]
251mod tests {
252    use super::*;
253
254    #[test]
255    fn a_sidecar_round_trips_and_looks_up_every_key() {
256        let entries: Vec<Vec<u8>> = (0..256u32)
257            .map(|i| format!("{}:{i}", i % 7).into_bytes())
258            .collect();
259        let keys: KeyFn = Box::new(|e| vec![e.split(|b| *b == b':').next().unwrap().to_vec()]);
260        let sidecar = Sidecar::build(&entries, &keys);
261        // 256 identity keys and 256 caller keys, each pair once.
262        assert_eq!(sidecar.pairs.len(), 512);
263        let bytes = sidecar.to_bytes();
264        assert_eq!(Sidecar::from_bytes("s", &bytes).unwrap(), sidecar);
265        for (offset, entry) in entries.iter().enumerate() {
266            assert_eq!(sidecar.lookup(&identity(entry)), vec![offset as u8]);
267        }
268        // Entries 3, 10, 17, ... share the caller key "3".
269        let threes: Vec<u8> = (0..256u32)
270            .filter(|i| i % 7 == 3)
271            .map(|i| i as u8)
272            .collect();
273        assert_eq!(sidecar.lookup(b"3"), threes);
274        assert!(sidecar.lookup(b"9").is_empty());
275        assert!(Sidecar::from_bytes("s", &bytes[..bytes.len() - 1]).is_err());
276        assert!(Sidecar::from_bytes("s", &[2]).is_err());
277    }
278
279    #[test]
280    fn a_manifest_names_the_rows_holding_a_key() {
281        let sidecar = Sidecar::build(&[b"a".to_vec(), b"b".to_vec()], &no_keys());
282        let row = Manifest::row(0, &sidecar, &sidecar.to_bytes());
283        assert_eq!(row.keys, 2);
284        let manifest = Manifest {
285            version: 1,
286            bundles: vec![row],
287        };
288        let (min, max) = sidecar.key_range().unwrap();
289        assert_eq!(manifest.candidates(min).count(), 1);
290        assert_eq!(manifest.candidates(max).count(), 1);
291        assert_eq!(manifest.candidates(&[0u8; 32]).count(), 0);
292        assert_eq!(manifest.candidates(&[0xffu8; 32]).count(), 0);
293    }
294}
rfc9162_tiles/index.rs

rfc9162_tiles/
index.rs
