sva_engine/cache/
persist.rs

1// Concern: node values on a backend beneath memory, staged then committed: version, budget, eviction | Non-concern: the bytes' medium (a Backend) | IO: (key) -> Header; samples -> staged; persist()
2
3use std::collections::BTreeMap;
4use std::sync::{Arc, Mutex, MutexGuard};
5
6use sva_formula::{FOURIER_DUAL_RULES_VERSION, Hash};
7use sva_samples::{Buffer, Extent};
8
9use super::codec::{self, STORE_FORMAT};
10use super::index::{Index, key_of, name_of};
11use super::joined;
12use super::stored::{Header, Samples};
13
14pub const DEFAULT_STORE_BYTES: u64 = 2 << 30;
15
16fn version() -> String {
17    format!("sva store format {STORE_FORMAT} table {FOURIER_DUAL_RULES_VERSION}")
18}
19
20pub const INDEX_NAME: &str = "index";
21
22const META: &str = "meta";
23
24/// Named bytes. A `put` is whole or absent: no reader sees half of one.
25pub trait Backend: Sized {
26    type Lock;
27    /// The one exclusive lock over these names, shared by every store over them.
28    fn lock(&self) -> impl Future<Output = Result<Self::Lock, String>>;
29    fn get(&self, name: &str) -> impl Future<Output = Result<Option<Vec<u8>>, String>>;
30    /// At most `len` bytes from `from` on, fewer where the name ends sooner.
31    fn get_range(
32        &self,
33        name: &str,
34        from: u64,
35        len: u64,
36    ) -> impl Future<Output = Result<Option<Vec<u8>>, String>>;
37    fn put(&self, name: &str, bytes: &[u8]) -> impl Future<Output = Result<(), String>>;
38    /// False where another holder has `name` open, so it stays.
39    fn delete(&self, name: &str) -> impl Future<Output = Result<bool, String>>;
40    /// Each name, with its size in bytes.
41    fn list(&self) -> impl Future<Output = Result<Vec<(String, u64)>, String>>;
42    /// An area beside these names that no other store writes and `list` never names.
43    fn staging(&self) -> impl Future<Output = Result<Self, String>>;
44    /// `name` moved from here into `to` in one step, over whatever `to` held under it; as
45    /// `delete`, false where either is open.
46    fn rename(&self, name: &str, to: &Self) -> impl Future<Output = Result<bool, String>>;
47}
48
49#[derive(Clone, Default)]
50struct Staged {
51    chunks: Vec<(String, Extent)>,
52    meta: bool,
53}
54
55/// Node values under their keys, beneath the memory tier alone: it stages, and only `persist`
56/// commits each value whole by rename, evicting the least recently used past the budget. Every
57/// change to the backend's names runs under its lock; lookups and reads never wait on it.
58pub struct Store<B> {
59    backend: B,
60    staging: B,
61    max_bytes: u64,
62    index: Mutex<Index>,
63    staged: Mutex<BTreeMap<Hash, Staged>>,
64    written: Mutex<u64>,
65}
66
67#[derive(Clone, Copy, Debug, PartialEq, Eq)]
68pub enum Opening {
69    Locking,
70    ReadingIndex,
71    /// The entries it names and their bytes: none of an index another format wrote.
72    IndexRead {
73        index_bytes: u64,
74        entries: usize,
75        bytes: u64,
76    },
77}
78
79enum Prepared {
80    Ready { bytes: u64 },
81    Dropped,
82    Refused,
83}
84
85#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
86pub struct Persisted {
87    pub written: usize,
88    pub evicted: usize,
89    /// Values larger than the whole budget.
90    pub refused: usize,
91}
92
93fn locked<T>(held: &Mutex<T>) -> MutexGuard<'_, T> {
94    held.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
95}
96
97impl<B: Backend> Store<B> {
98    /// Over another format's index, a fresh one: what that format left is neither read nor
99    /// removed, nor counted against the budget.
100    pub async fn open(backend: B, max_bytes: u64) -> Result<Store<B>, String> {
101        Store::open_told(backend, max_bytes, &|_| ()).await
102    }
103
104    pub async fn open_told(
105        backend: B,
106        max_bytes: u64,
107        told: &dyn Fn(Opening),
108    ) -> Result<Store<B>, String> {
109        let version = version();
110        told(Opening::Locking);
111        let held = backend.lock().await?;
112        told(Opening::ReadingIndex);
113        let found = backend.get(INDEX_NAME).await?;
114        let index_bytes = found.as_ref().map_or(0, |text| text.len() as u64);
115        let index = match found.and_then(|text| Index::read(&text, &version)) {
116            Some(index) => index,
117            None => {
118                let empty = Index::default();
119                backend
120                    .put(INDEX_NAME, empty.text(&version).as_bytes())
121                    .await?;
122                empty
123            }
124        };
125        told(Opening::IndexRead {
126            index_bytes,
127            entries: index.held.len(),
128            bytes: index.bytes,
129        });
130        let staging = backend.staging().await?;
131        drop(held);
132        Ok(Store {
133            backend,
134            staging,
135            max_bytes,
136            index: Mutex::new(index),
137            staged: Mutex::new(BTreeMap::new()),
138            written: Mutex::new(0),
139        })
140    }
141
142    pub fn max_bytes(&self) -> u64 {
143        self.max_bytes
144    }
145
146    pub fn bytes(&self) -> u64 {
147        locked(&self.index).bytes
148    }
149
150    pub fn holds(&self, key: Hash) -> bool {
151        locked(&self.index).held.contains_key(&key)
152    }
153
154    pub(crate) fn used(&self, key: Hash) {
155        locked(&self.index).used(key);
156    }
157
158    /// Reads headers, never a sample; one standing for another's samples reads that one's.
159    pub(crate) async fn lookup(&self, key: Hash) -> Option<Header> {
160        let found = self.written(key).await?;
161        let &Samples::Of { key: of, by } = found.samples() else {
162            return Some(found);
163        };
164        let samples = match self.written(of).await?.into_parts().1 {
165            Samples::Entry { file, runs, .. } => Samples::Entry {
166                file,
167                runs,
168                shift: -by,
169            },
170            Samples::Staged { file, chunks, .. } => Samples::Staged {
171                file,
172                chunks,
173                shift: -by,
174            },
175            _ => return None,
176        };
177        Some(Header::new(found.into_parts().0, samples))
178    }
179
180    /// Staged first; past the index, the backend, for another store's commits.
181    async fn written(&self, key: Hash) -> Option<Header> {
182        if let Some(staged) = self.staged_value(key).await {
183            return Some(staged);
184        }
185        let name = name_of(key);
186        let Some(mut bytes) = self.backend.get_range(&name, 0, 8).await.ok()? else {
187            locked(&self.index).forget(key);
188            return None;
189        };
190        let rest = codec::head_len(&bytes)? - 8;
191        bytes.extend(self.backend.get_range(&name, 8, rest as u64).await.ok()??);
192        let (found, len) =
193            codec::read_head(&bytes, key).filter(|(found, _)| found.stored().key == key)?;
194        let mut index = locked(&self.index);
195        match index.held.contains_key(&key) {
196            true => index.used(key),
197            false => index.touch(key, len),
198        }
199        Some(found)
200    }
201
202    async fn staged_value(&self, key: Hash) -> Option<Header> {
203        let chunks = {
204            let staged = locked(&self.staged);
205            let held = staged.get(&key).filter(|staged| staged.meta)?;
206            held.chunks.clone()
207        };
208        let meta = self.staging.get(&staged_name(key, META)).await.ok()??;
209        let (found, _) = codec::read_head(&meta, key)?;
210        Some(match found.refers() {
211            true => found,
212            false => {
213                let samples = Samples::Staged {
214                    file: key,
215                    chunks,
216                    shift: 0,
217                };
218                Header::new(found.into_parts().0, samples)
219            }
220        })
221    }
222
223    /// Whole chunks holding `over`; `None` where they are gone or corrupt.
224    pub(crate) async fn read(&self, head: &Header, over: Extent) -> Option<Vec<Buffer>> {
225        let mut out = Vec::new();
226        match head.samples() {
227            Samples::None | Samples::Of { .. } => {}
228            Samples::Entry { file, runs, shift } => {
229                for run in runs {
230                    let met = run.extent().intersect(over.shifted(-shift));
231                    if met.is_empty() {
232                        continue;
233                    }
234                    let (from, to) = codec::chunks_over(run, met);
235                    let (at, len) = codec::span_of(run, from, to);
236                    let bytes = self.backend.get_range(&name_of(*file), at, len).await;
237                    let mut samples = codec::read_chunks(&bytes.ok()??, run, from, to)?;
238                    samples.start += shift;
239                    out.push(samples);
240                }
241            }
242            Samples::Staged { chunks, shift, .. } => {
243                for (name, e) in chunks {
244                    if !e.intersect(over.shifted(-shift)).is_empty() {
245                        let bytes = self.staging.get(name).await.ok()??;
246                        let mut samples = codec::read_chunk(&bytes)?;
247                        samples.start += shift;
248                        out.push(samples);
249                    }
250                }
251            }
252        }
253        Some(out)
254    }
255
256    pub(crate) async fn stage(&self, key: Hash, samples: &Buffer) -> Result<(), String> {
257        let n = {
258            let mut written = locked(&self.written);
259            *written += 1;
260            *written
261        };
262        let name = staged_name(key, &n.to_string());
263        self.staging.put(&name, &codec::chunk(samples)).await?;
264        locked(&self.staged)
265            .entry(key)
266            .or_default()
267            .chunks
268            .push((name, samples.extent()));
269        Ok(())
270    }
271
272    pub(crate) async fn stage_meta(&self, key: Hash, head: &Header) -> Result<(), String> {
273        let name = staged_name(key, META);
274        self.staging.put(&name, &codec::entry(head, &[])).await?;
275        locked(&self.staged).entry(key).or_default().meta = true;
276        Ok(())
277    }
278
279    /// Free while nothing is staged or used since and the budget holds. The index file is
280    /// every store's account and names each value before it moves: one cut short leaves a
281    /// gone entry named, never one unnamed. A value staged without its meta is dropped; one
282    /// failed or over an open entry stays staged.
283    pub(crate) async fn persist(&self) -> Result<Persisted, String> {
284        let idle = {
285            let index = locked(&self.index);
286            index.bytes <= self.max_bytes && index.unsynced().is_empty()
287        };
288        if idle && locked(&self.staged).is_empty() {
289            return Ok(Persisted::default());
290        }
291        let _held = self.backend.lock().await?;
292        let (mut index, seen) = self.current().await?;
293        let mut done = Persisted::default();
294        let mut moving = Vec::new();
295        let keys: Vec<Hash> = locked(&self.staged).keys().copied().collect();
296        for key in keys {
297            let Some(held) = locked(&self.staged).get(&key).cloned() else {
298                continue;
299            };
300            match self.prepared(key, &held).await? {
301                Prepared::Ready { bytes } => {
302                    moving.push((key, held, bytes));
303                    continue;
304                }
305                Prepared::Dropped => {}
306                Prepared::Refused => done.refused += 1,
307            }
308            self.unstaged(key, &held).await?;
309        }
310        if !moving.is_empty() {
311            let mut naming = index.clone();
312            for (key, _, bytes) in &moving {
313                naming.touch(*key, *bytes);
314            }
315            let text = naming.text(&version());
316            self.backend.put(INDEX_NAME, text.as_bytes()).await?;
317        }
318        for (key, held, bytes) in moving {
319            if !self.staging.rename(&name_of(key), &self.backend).await? {
320                continue;
321            }
322            index.touch(key, bytes);
323            done.written += 1;
324            self.unstaged(key, &held).await?;
325        }
326        for key in index.oldest_first() {
327            if index.bytes <= self.max_bytes {
328                break;
329            }
330            if self.backend.delete(&name_of(key)).await? {
331                index.forget(key);
332                done.evicted += 1;
333            }
334        }
335        self.backend
336            .put(INDEX_NAME, index.text(&version()).as_bytes())
337            .await?;
338        index.synced();
339        let mut local = locked(&self.index);
340        for key in local.used_since(seen) {
341            index.used(key);
342        }
343        *local = index;
344        Ok(done)
345    }
346
347    /// The index file, or else the directory listed, with this store's uses since; its clock.
348    async fn current(&self) -> Result<(Index, u64), String> {
349        let found = self.backend.get(INDEX_NAME).await?;
350        let mut index = match found.and_then(|text| Index::read(&text, &version())) {
351            Some(index) => index,
352            None => {
353                let listed = self.backend.list().await?;
354                let listed = listed
355                    .into_iter()
356                    .filter_map(|(name, bytes)| Some((key_of(&name)?, bytes)));
357                Index::listed(listed.collect(), &locked(&self.index))
358            }
359        };
360        let local = locked(&self.index);
361        for key in local.unsynced() {
362            index.used(key);
363        }
364        Ok((index, local.clock()))
365    }
366
367    async fn unstaged(&self, key: Hash, held: &Staged) -> Result<(), String> {
368        for (chunk, _) in &held.chunks {
369            self.staging.delete(chunk).await?;
370        }
371        self.staging.delete(&staged_name(key, META)).await?;
372        locked(&self.staged).remove(&key);
373        Ok(())
374    }
375
376    /// `key`'s staged value joined with its entry's, whole in staging under the entry's name.
377    async fn prepared(&self, key: Hash, held: &Staged) -> Result<Prepared, String> {
378        let whole = match held.meta {
379            true => self.staged_value_of(key, held).await?,
380            false => None,
381        };
382        let whole = whole.filter(|(head, runs)| !runs.is_empty() || head.refers());
383        let Some((head, mut runs)) = whole else {
384            return Ok(Prepared::Dropped);
385        };
386        let name = name_of(key);
387        if !head.refers() {
388            let held = self.backend.get(&name).await?;
389            if let Some(held) = held.and_then(|bytes| codec::read_runs(&bytes, key)) {
390                runs = joined_runs(runs, held);
391            }
392        }
393        let bytes = codec::entry(&head, &runs);
394        if bytes.len() as u64 > self.max_bytes {
395            return Ok(Prepared::Refused);
396        }
397        self.staging.put(&name, &bytes).await?;
398        Ok(Prepared::Ready {
399            bytes: bytes.len() as u64,
400        })
401    }
402
403    async fn staged_value_of(
404        &self,
405        key: Hash,
406        held: &Staged,
407    ) -> Result<Option<(Header, Vec<Buffer>)>, String> {
408        let Some(meta) = self.staging.get(&staged_name(key, META)).await? else {
409            return Ok(None);
410        };
411        let Some((head, _)) = codec::read_head(&meta, key) else {
412            return Ok(None);
413        };
414        let mut runs = Vec::new();
415        for (chunk, _) in &held.chunks {
416            let bytes = self.staging.get(chunk).await?;
417            match bytes.as_deref().and_then(codec::read_chunk) {
418                Some(samples) => joined(&mut runs, vec![Arc::new(samples)]),
419                None => return Ok(None),
420            }
421        }
422        let runs = runs.into_iter().map(Arc::unwrap_or_clone).collect();
423        Ok(Some((head, runs)))
424    }
425}
426
427fn joined_runs(staged: Vec<Buffer>, held: Vec<Buffer>) -> Vec<Buffer> {
428    let mut runs: Vec<Arc<Buffer>> = held.into_iter().map(Arc::new).collect();
429    joined(&mut runs, staged.into_iter().map(Arc::new).collect());
430    runs.into_iter().map(Arc::unwrap_or_clone).collect()
431}
432
433fn staged_name(key: Hash, part: &str) -> String {
434    format!("{}.{part}", name_of(key))
435}
sva_engine/cache/persist.rs

sva_engine/cache/
persist.rs
