harmos_runtime/evidence/
mod.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12mod draft;
13mod fact;
14mod vault;
15
16/* === Re-Exports === */
17
18pub use draft::Draft;
19pub(crate) use fact::SOURCE;
20pub use fact::{EvidenceKind, EvidenceRecord};
21pub use vault::{Excerpt, Vault};
22
23/* === Imports === */
24
25// Standard Library Imports
26use std::collections::BTreeMap;
27use std::sync::{Arc, Mutex, Weak};
28// Third Party Imports
29use tokio::sync::watch;
30// Local Imports
31use crate::supervision::executor;
32use crate::{Application, DefinitionId, state::At, state::Entry, state::Recorded, state::StateManager};
33use draft::{Open, Staged};
34use vault::Keeping;
35
36/* === Constants === */
37
38/* === Macros === */
39
40/* === Type Aliases === */
41
42type Held<T> = BTreeMap<String, BTreeMap<(DefinitionId, String), T>>;
43
44/* === Structs === */
45
46pub(crate) struct Declaration {
47    vault: Arc<dyn Keeping>,
48    kinds: Vec<(DefinitionId, u32)>,
49}
50
51pub struct Evidence<A: Application> {
52    inner: Option<Arc<Inner<A>>>,
53}
54
55struct Inner<A: Application> {
56    state: StateManager<A>,
57    vault: Arc<dyn Keeping>,
58    kinds: BTreeMap<DefinitionId, u32>,
59    drafts: Mutex<Held<Arc<Open>>>,
60    kept: Mutex<Held<EvidenceRecord>>,
61}
62
63/* === Enums === */
64
65#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
66pub enum EvidenceError {
67    #[error("this runtime was assembled without an evidence vault")]
68    Unserved,
69
70    #[error("`{kind}` is not a declared evidence kind")]
71    Undeclared { kind: DefinitionId },
72
73    #[error("this evidence vault hands out no external draft paths")]
74    Unsupported,
75
76    #[error("this draft is written by a process outside the runtime")]
77    Foreign,
78
79    #[error("`{scope}`/`{key}` already names evidence of kind `{kind}`")]
80    Conflict { kind: DefinitionId, scope: String, key: String },
81
82    #[error("no evidence is open or finalized under this key")]
83    Unknown,
84
85    #[error("`{named}` does not declare the details of `{kind}` evidence")]
86    Mismatched { kind: DefinitionId, named: DefinitionId },
87
88    #[error("the details of `{kind}` do not match its declaration: {cause}")]
89    Malformed { kind: DefinitionId, cause: String },
90
91    #[error("this evidence is already being finalized")]
92    Finalizing,
93
94    #[error("a read of {requested} bytes is past this vault's {ceiling}-byte bound")]
95    Bound { requested: usize, ceiling: usize },
96
97    #[error("the stored evidence is {found} bytes where {written} were written")]
98    Corrupt { written: u64, found: u64 },
99
100    #[error("the evidence vault could not be read: {0}")]
101    Unreadable(String),
102
103    #[error("the evidence vault could not be written: {0}")]
104    Unwritable(String),
105
106    #[error("the state is unavailable, so nothing was finalized")]
107    Unavailable,
108}
109
110/* === Traits === */
111
112/* === Implementations === */
113
114impl EvidenceError {
115    pub fn unreadable(cause: impl std::fmt::Display) -> Self {
116        Self::Unreadable(cause.to_string())
117    }
118
119    pub fn unwritable(cause: impl std::fmt::Display) -> Self {
120        Self::Unwritable(cause.to_string())
121    }
122}
123
124impl Declaration {
125    pub(crate) fn new(vault: impl Vault, kinds: impl IntoIterator<Item = (DefinitionId, u32)>) -> Self {
126        Self { vault: Arc::new(vault), kinds: kinds.into_iter().collect() }
127    }
128}
129
130impl<A: Application> Evidence<A> {
131    pub(crate) fn wired(
132        state: StateManager<A>, declaration: Option<Declaration>, recovered: &[Arc<Entry<Recorded<A>>>],
133    ) -> Result<Self, String> {
134        let Some(Declaration { vault, kinds }) = declaration else {
135            return Ok(Self { inner: None });
136        };
137        let mut frozen = BTreeMap::new();
138        for (id, version) in kinds {
139            if frozen.insert(id.clone(), version).is_some() {
140                return Err(format!("evidence kind `{id}` is declared more than once"));
141            }
142        }
143        let mut kept: Held<EvidenceRecord> = BTreeMap::new();
144        for entry in recovered {
145            let Recorded::System(fact) = &entry.value else {
146                continue;
147            };
148            let Some(record) = EvidenceRecord::folded(fact, entry.position) else {
149                continue;
150            };
151            kept.entry(record.scope.clone()).or_default().insert((record.kind.clone(), record.key.clone()), record);
152        }
153
154        Ok(Self {
155            inner: Some(Arc::new(Inner { state, vault, kinds: frozen, drafts: Mutex::new(BTreeMap::new()), kept: Mutex::new(kept) })),
156        })
157    }
158
159    pub async fn open<K: EvidenceKind>(
160        &self, scope: impl Into<String>, key: impl Into<String>, details: K::Details,
161    ) -> Result<Draft, EvidenceError> {
162        self.opening::<K>(scope.into(), key.into(), details, false).await
163    }
164
165    pub async fn external<K: EvidenceKind>(
166        &self, scope: impl Into<String>, key: impl Into<String>, details: K::Details,
167    ) -> Result<Draft, EvidenceError> {
168        self.opening::<K>(scope.into(), key.into(), details, true).await
169    }
170
171    pub async fn read<K: EvidenceKind>(&self, scope: &str, key: &str, offset: u64, limit: usize) -> Result<Excerpt, EvidenceError> {
172        let inner = self.inner()?;
173        let kind = inner.declared::<K>()?;
174        let ceiling = inner.vault.ceiling();
175        if limit == 0 || limit > ceiling {
176            return Err(EvidenceError::Bound { requested: limit, ceiling });
177        }
178
179        let staged = inner.draft(&kind, scope, key);
180        let (bytes, length) = match staged {
181            Some(open) => {
182                let read = inner.vault.read_draft(&open.name, offset, limit).await?;
183                open.observed(read.1);
184                read
185            }
186            None => {
187                let name = encode(&kind, scope, key);
188                if !inner.is_kept(&kind, scope, key) {
189                    return Err(EvidenceError::Unknown);
190                }
191                inner.vault.read(&name, offset, limit).await?
192            }
193        };
194
195        Ok(Excerpt { requested: offset, offset: offset.min(length), bytes, length })
196    }
197
198    pub fn list(&self, scope: &str) -> Vec<EvidenceRecord> {
199        let Ok(inner) = self.inner() else {
200            return Vec::new();
201        };
202        let mut listing: BTreeMap<(DefinitionId, String), EvidenceRecord> = BTreeMap::new();
203
204        for (at, open) in inner.scope(&inner.drafts, scope) {
205            listing.insert(
206                at,
207                EvidenceRecord::new(
208                    open.kind.clone(),
209                    open.version,
210                    open.scope.clone(),
211                    open.key.clone(),
212                    open.length(),
213                    open.described.clone(),
214                ),
215            );
216        }
217        for (at, record) in inner.scope(&inner.kept, scope) {
218            listing.insert(at, record);
219        }
220
221        listing.into_values().collect()
222    }
223
224    pub async fn finalize(&self, draft: &Draft) -> Result<EvidenceRecord, EvidenceError> {
225        let inner = self.inner()?;
226        let open = &draft.open;
227        inner.begin(open)?;
228
229        let finalized = inner.finalizing(open).await;
230        inner.end(open, finalized.is_ok());
231
232        finalized
233    }
234
235    pub async fn abandon(&self, draft: &Draft) -> Result<(), EvidenceError> {
236        let inner = self.inner()?;
237        let open = &draft.open;
238        inner.begin(open)?;
239
240        let quarantined = inner.vault.quarantine(&open.name).await;
241        inner.forget(open);
242        open.release(open.length());
243
244        quarantined
245    }
246
247    pub async fn finalize_scope(&self, scope: &str) -> Result<Vec<EvidenceRecord>, EvidenceError> {
248        let inner = self.inner()?;
249        let mut finalized = Vec::new();
250
251        for (_, open) in inner.scope(&inner.drafts, scope) {
252            finalized.push(self.finalize(&Draft { open }).await?);
253        }
254
255        Ok(finalized)
256    }
257
258    pub async fn abandon_scope(&self, scope: &str) -> Result<(), EvidenceError> {
259        let inner = self.inner()?;
260
261        for (_, open) in inner.scope(&inner.drafts, scope) {
262            self.abandon(&Draft { open }).await?;
263        }
264
265        Ok(())
266    }
267
268    pub fn watch(&self, draft: &Draft) -> Result<watch::Receiver<u64>, EvidenceError> {
269        let inner = self.inner()?;
270        let open = &draft.open;
271        if open.released() {
272            return Err(EvidenceError::Unknown);
273        }
274        let (receiver, first) = open.watch();
275        if first && open.external.is_some() {
276            let cadence = inner.vault.cadence();
277            executor::task(observing(Arc::downgrade(open), cadence));
278        }
279
280        Ok(receiver)
281    }
282
283    async fn opening<K: EvidenceKind>(
284        &self, scope: String, key: String, details: K::Details, foreign: bool,
285    ) -> Result<Draft, EvidenceError> {
286        let inner = self.inner()?;
287        let kind = inner.declared::<K>()?;
288        let version = inner.kinds[&kind];
289        let described =
290            serde_json::to_vec(&details).map_err(|error| EvidenceError::Malformed { kind: kind.clone(), cause: error.to_string() })?;
291        if inner.draft(&kind, &scope, &key).is_some() || inner.is_kept(&kind, &scope, &key) {
292            return Err(EvidenceError::Conflict { kind, scope, key });
293        }
294        let name = encode(&kind, &scope, &key);
295        let length = inner.vault.open(&name).await?;
296        let external = match foreign {
297            true => Some(inner.vault.external(&name).await?.ok_or(EvidenceError::Unsupported)?),
298            false => None,
299        };
300        let open = Arc::new(Open::new(
301            Staged { kind: kind.clone(), version, scope: scope.clone(), key: key.clone(), name, described, external, length },
302            Arc::clone(&inner.vault),
303        ));
304        inner.drafts.lock().expect("the evidence index is healthy").entry(scope).or_default().insert((kind, key), Arc::clone(&open));
305
306        Ok(Draft { open })
307    }
308
309    fn inner(&self) -> Result<&Arc<Inner<A>>, EvidenceError> {
310        self.inner.as_ref().ok_or(EvidenceError::Unserved)
311    }
312}
313
314impl<A: Application> Clone for Evidence<A> {
315    fn clone(&self) -> Self {
316        Self { inner: self.inner.clone() }
317    }
318}
319
320impl<A: Application> Inner<A> {
321    fn declared<K: EvidenceKind>(&self) -> Result<DefinitionId, EvidenceError> {
322        let (kind, _) = K::definition();
323        match self.kinds.contains_key(&kind) {
324            true => Ok(kind),
325            false => Err(EvidenceError::Undeclared { kind }),
326        }
327    }
328
329    fn draft(&self, kind: &DefinitionId, scope: &str, key: &str) -> Option<Arc<Open>> {
330        self.drafts.lock().expect("the evidence index is healthy").get(scope)?.get(&(kind.clone(), key.to_owned())).map(Arc::clone)
331    }
332
333    fn is_kept(&self, kind: &DefinitionId, scope: &str, key: &str) -> bool {
334        self.kept
335            .lock()
336            .expect("the evidence index is healthy")
337            .get(scope)
338            .is_some_and(|held| held.contains_key(&(kind.clone(), key.to_owned())))
339    }
340
341    fn scope<T: Clone>(&self, index: &Mutex<Held<T>>, scope: &str) -> Vec<((DefinitionId, String), T)> {
342        index
343            .lock()
344            .expect("the evidence index is healthy")
345            .get(scope)
346            .map(|held| held.iter().map(|(at, value)| (at.clone(), value.clone())).collect())
347            .unwrap_or_default()
348    }
349
350    fn begin(&self, open: &Arc<Open>) -> Result<(), EvidenceError> {
351        match self.draft(&open.kind, &open.scope, &open.key) {
352            Some(held) if Arc::ptr_eq(&held, open) => match open.begin_finalize() {
353                true => Ok(()),
354                false => Err(EvidenceError::Finalizing),
355            },
356            Some(_) | None => Err(EvidenceError::Unknown),
357        }
358    }
359
360    fn end(&self, open: &Arc<Open>, ended: bool) {
361        if !ended {
362            open.end_finalize();
363        }
364    }
365
366    async fn finalizing(&self, open: &Arc<Open>) -> Result<EvidenceRecord, EvidenceError> {
367        // Durable first, then read back: a digest folded from what a producer
368        // said it wrote proves nothing about what survived.
369        let (digest, length) = self.vault.verify(&open.name).await?;
370        let written = open.length();
371        if open.external.is_none() && written != length {
372            return Err(EvidenceError::Corrupt { written, found: length });
373        }
374
375        let mut record =
376            EvidenceRecord::new(open.kind.clone(), open.version, open.scope.clone(), open.key.clone(), length, open.described.clone());
377        let receipt = crate::records::RecordManager::new(self.state.clone())
378            .publish_system(record.recorded(digest))
379            .await
380            .map_err(|_| EvidenceError::Unavailable)?;
381        self.state.wait_until_stored(receipt.position).await;
382        record.finalized = Some(At { value: digest, position: receipt.position });
383
384        // Visible before removed: the fact is already history, so the
385        // evidence enters the finalized index before its draft leaves the
386        // active one and a concurrent listing never falls through the gap
387        // between them.
388        self.kept
389            .lock()
390            .expect("the evidence index is healthy")
391            .entry(record.scope.clone())
392            .or_default()
393            .insert((record.kind.clone(), record.key.clone()), record.clone());
394        self.vault.claim(&open.name).await?;
395        self.forget(open);
396        open.release(length);
397
398        Ok(record)
399    }
400
401    fn forget(&self, open: &Arc<Open>) {
402        let mut drafts = self.drafts.lock().expect("the evidence index is healthy");
403        let Some(held) = drafts.get_mut(&open.scope) else {
404            return;
405        };
406        held.remove(&(open.kind.clone(), open.key.clone()));
407        if held.is_empty() {
408            drafts.remove(&open.scope);
409        }
410    }
411}
412
413/* === Public Functions === */
414
415/* === Private Functions === */
416
417async fn observing(draft: Weak<Open>, cadence: std::time::Duration) {
418    loop {
419        executor::sleep(cadence).await;
420        let Some(open) = draft.upgrade() else {
421            return;
422        };
423        if open.released() || open.idle() {
424            return;
425        }
426        if open.observe().await.is_err() {
427            return;
428        }
429    }
430}
431
432fn encode(kind: &DefinitionId, scope: &str, key: &str) -> String {
433    let mut name = String::new();
434    for (index, part) in [kind.as_str(), scope, key].into_iter().enumerate() {
435        if index > 0 {
436            name.push('~');
437        }
438        for byte in part.bytes() {
439            match byte {
440                b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' => name.push(byte as char),
441                _ => name.push_str(&format!("%{byte:02x}")),
442            }
443        }
444    }
445
446    name
447}
448
449/* === Tests === */
harmos_runtime/evidence/mod.rs

harmos_runtime/evidence/
mod.rs
