haematite_cli/serve/storage/
embed_prune.rs

1//! Removes companion vectors of fields no longer declared at the branch head.
2//!
3//! Forks inherit their parent's vectors. Once the parent rewrites an undeclared
4//! field, the fork's copies are unreachable and need removal.
5//!
6//! Fields are found by skipping, never by reading every vector: the first
7//! vector key at or after a point gives a field, and a field the branch
8//! declares is passed over whole, so the reads are in proportion to the
9//! fields present, plus one page.
10//!
11//! A field with a position is never pruned, declared or not: a fork's
12//! inheriting removes the position of an undeclared field first (F1, step
13//! two), and until it has, the field is left alone, so the two never race.
14//!
15//! Durable commits: at most one a call, removing at most the page setting's
16//! vector keys of one field; none when nothing is found.
17//!
18//! EMB-E2: a field's windows (`vw\0`) are pruned as its vectors are, by the
19//! same rule, once no vector key is left to prune.
20
21use std::num::NonZeroUsize;
22
23use super::embedded::{self, Field};
24use super::{BranchRefRecord, Captured, Failure, Storage, branch_name, identifier, reads, version};
25
26/// What one call removed, and whether a later call may find more.
27#[derive(Debug, Clone, Copy, PartialEq, Eq)]
28pub struct Pruned {
29    pub removed: usize,
30    pub more: bool,
31}
32
33/// Where each family of vector keys starts and ends: one vector a record,
34/// then EMB-E2's windows.
35const FAMILIES: [(&[u8], &[u8]); 2] = [(b"v\0", b"v\x01"), (b"vw\0", b"vw\x01")];
36
37impl Storage {
38    /// Removes, in one commit, up to `page_records` vector keys of the first
39    /// field in `branch`'s companion that the branch does not declare
40    /// embedded at its head and that has no position (F7).
41    ///
42    /// # Errors
43    /// `branch_missing`; `embed_vector_key_corrupt` for a vector key that is
44    /// not UTF-8 or names no field; whatever a read or the commit answers.
45    pub fn embed_prune(
46        &mut self,
47        branch: &str,
48        page_records: NonZeroUsize,
49    ) -> Result<Pruned, Failure> {
50        identifier(branch)?;
51        let record = self
52            .refs
53            .get(&branch_name(branch))
54            .ok_or_else(|| Failure::new("branch_missing", branch))?;
55        let head = Captured {
56            branch: branch.to_owned(),
57            version: version(record),
58            shards: record.shards.clone(),
59        };
60        let name = embedded::companion_name(branch);
61        let nothing = Pruned {
62            removed: 0,
63            more: false,
64        };
65        let Some(companion) = self.refs.get(&name) else {
66            return Ok(nothing);
67        };
68        let positions = self.companion_positions(companion)?;
69        let kept = |collection: &str, field: &str| -> Result<bool, Failure> {
70            Ok(
71                positions.contains_key(&(collection.to_owned(), field.to_owned()))
72                    || self.declares(&head, collection, field)?,
73            )
74        };
75        let mut found = None;
76        for (index, family) in FAMILIES.into_iter().enumerate() {
77            let scan = Scan {
78                name: &name,
79                companion,
80                family,
81            };
82            if let Some(FamilyPage { removals, more }) =
83                family_page(self, &scan, page_records, &kept)?
84            {
85                // Windows are looked at once no vector is left to prune.
86                let (from, to) = FAMILIES[1];
87                let windows = index == 0 && first_key(self, companion, from, to)?.is_some();
88                found = Some((removals, more || windows));
89                break;
90            }
91        }
92        let Some((removals, more)) = found else {
93            return Ok(nothing);
94        };
95        let writes: Vec<(Vec<u8>, Option<Vec<u8>>)> =
96            removals.into_iter().map(|key| (key, None)).collect();
97        let removed = writes.len();
98        if removed == 0 {
99            return Ok(nothing);
100        }
101        self.commit_namespace(&name, &writes)?;
102        Ok(Pruned { removed, more })
103    }
104
105    /// Whether `head` declares `collection.field` embedded.
106    fn declares(&self, head: &Captured, collection: &str, field: &str) -> Result<bool, Failure> {
107        let key = super::embed_job::field_key(collection, field);
108        match reads::get_bytes(&self.reader, head, &key)? {
109            Some(bytes) => Ok(matches!(embedded::read_field(&bytes)?, Field::Embedded(_))),
110            None => Ok(false),
111        }
112    }
113}
114
115/// One family of a companion's vector keys, as [`FAMILIES`] names it.
116struct Scan<'a> {
117    name: &'a str,
118    companion: &'a BranchRefRecord,
119    family: (&'static [u8], &'static [u8]),
120}
121
122/// One page of a family's keys to remove.
123struct FamilyPage {
124    removals: Vec<Vec<u8>>,
125    /// Whether any key is left in the family after this page.
126    more: bool,
127}
128
129/// The first page of keys of the first field in `scan`'s family that `kept`
130/// does not keep, and whether any is left in the family after them; `None`
131/// when it keeps every field there.
132fn family_page(
133    storage: &Storage,
134    scan: &Scan<'_>,
135    page_records: NonZeroUsize,
136    kept: &impl Fn(&str, &str) -> Result<bool, Failure>,
137) -> Result<Option<FamilyPage>, Failure> {
138    let (family, family_end) = scan.family;
139    let mut from = family.to_vec();
140    loop {
141        let Some(key) = first_key(storage, scan.companion, &from, family_end)? else {
142            return Ok(None);
143        };
144        let (collection, field) = vector_field(scan.name, family, &key)?;
145        let mut prefix = family.to_vec();
146        prefix.extend_from_slice(format!("{collection}\0{field}\0").as_bytes());
147        let end = past(&prefix);
148        if kept(&collection, &field)? {
149            from = end;
150            continue;
151        }
152        let (removals, more) = field_page(storage, scan.companion, &prefix, &end, page_records)?;
153        let later = first_key(storage, scan.companion, &end, family_end)?.is_some();
154        return Ok(Some(FamilyPage {
155            removals,
156            more: more || later,
157        }));
158    }
159}
160
161/// The smallest key in `[from, to)` across the companion's shards: one read
162/// of one key on each.
163fn first_key(
164    storage: &Storage,
165    companion: &BranchRefRecord,
166    from: &[u8],
167    to: &[u8],
168) -> Result<Option<Vec<u8>>, Failure> {
169    let mut first: Option<Vec<u8>> = None;
170    for shard in &companion.shards {
171        let page = storage
172            .reader
173            .range_page_at(shard.shard_id, shard.head, from, to, NonZeroUsize::MIN)
174            .map_err(Failure::observer)?;
175        if let Some((key, _value)) = page.entries.into_iter().next()
176            && first.as_ref().is_none_or(|known| key < *known)
177        {
178            first = Some(key);
179        }
180    }
181    Ok(first)
182}
183
184/// The first `limit` keys in `[prefix, end)` across the companion's shards,
185/// and whether any is left after them.
186fn field_page(
187    storage: &Storage,
188    companion: &BranchRefRecord,
189    prefix: &[u8],
190    end: &[u8],
191    limit: NonZeroUsize,
192) -> Result<(Vec<Vec<u8>>, bool), Failure> {
193    let mut keys = Vec::new();
194    let mut more = false;
195    for shard in &companion.shards {
196        let page = storage
197            .reader
198            .range_page_at(shard.shard_id, shard.head, prefix, end, limit)
199            .map_err(Failure::observer)?;
200        more |= page.next_from.is_some();
201        keys.extend(page.entries.into_iter().map(|(key, _value)| key));
202    }
203    keys.sort();
204    if keys.len() > limit.get() {
205        keys.truncate(limit.get());
206        more = true;
207    }
208    Ok((keys, more))
209}
210
211/// The (collection, field) a key of `family` names, refused by name, never
212/// guessed, when they are not UTF-8 or it names no field and record. A
213/// window key's record part ends in its index, which need not be UTF-8. The
214/// key's bytes are never printed.
215fn vector_field(companion: &str, family: &[u8], key: &[u8]) -> Result<(String, String), Failure> {
216    let corrupt = || {
217        Failure::new(
218            "embed_vector_key_corrupt",
219            format!("{companion} holds a vector key that names no collection, field and record"),
220        )
221    };
222    let named = key.strip_prefix(family).ok_or_else(corrupt)?;
223    let mut parts = named.splitn(3, |byte| *byte == 0);
224    let (Some(collection), Some(field), Some(_record)) = (parts.next(), parts.next(), parts.next())
225    else {
226        return Err(corrupt());
227    };
228    let text = |part: &[u8]| {
229        std::str::from_utf8(part)
230            .map(str::to_owned)
231            .map_err(|_error| corrupt())
232    };
233    Ok((text(collection)?, text(field)?))
234}
235
236/// The first key after every key that starts with `prefix`, whose last byte
237/// is the NUL separator.
238fn past(prefix: &[u8]) -> Vec<u8> {
239    let mut end = prefix.to_vec();
240    if let Some(last) = end.last_mut() {
241        *last = 1;
242    }
243    end
244}
245
246#[cfg(test)]
247#[path = "embed_prune_tests.rs"]
248mod tests;
haematite_cli/serve/storage/embed_prune.rs

haematite_cli/serve/storage/
embed_prune.rs
