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 {
127            vault: Arc::new(vault),
128            kinds: kinds.into_iter().collect(),
129        }
130    }
131}
132
133impl<A: Application> Evidence<A> {
134    pub(crate) fn wired(
135        state: StateManager<A>, declaration: Option<Declaration>, recovered: &[Arc<Entry<Recorded<A>>>],
136    ) -> Result<Self, String> {
137        let Some(Declaration { vault, kinds }) = declaration else {
138            return Ok(Self { inner: None });
139        };
140        let mut frozen = BTreeMap::new();
141        for (id, version) in kinds {
142            if frozen.insert(id.clone(), version).is_some() {
143                return Err(format!("evidence kind `{id}` is declared more than once"));
144            }
145        }
146        let mut kept: Held<EvidenceRecord> = BTreeMap::new();
147        for entry in recovered {
148            let Recorded::System(fact) = &entry.value else {
149                continue;
150            };
151            let Some(record) = EvidenceRecord::folded(fact, entry.position) else {
152                continue;
153            };
154            kept.entry(record.scope.clone()).or_default().insert((record.kind.clone(), record.key.clone()), record);
155        }
156
157        Ok(Self {
158            inner: Some(Arc::new(Inner {
159                state,
160                vault,
161                kinds: frozen,
162                drafts: Mutex::new(BTreeMap::new()),
163                kept: Mutex::new(kept),
164            })),
165        })
166    }
167
168    pub async fn open<K: EvidenceKind>(
169        &self, scope: impl Into<String>, key: impl Into<String>, details: K::Details,
170    ) -> Result<Draft, EvidenceError> {
171        self.opening::<K>(scope.into(), key.into(), details, false).await
172    }
173
174    pub async fn external<K: EvidenceKind>(
175        &self, scope: impl Into<String>, key: impl Into<String>, details: K::Details,
176    ) -> Result<Draft, EvidenceError> {
177        self.opening::<K>(scope.into(), key.into(), details, true).await
178    }
179
180    pub async fn read<K: EvidenceKind>(&self, scope: &str, key: &str, offset: u64, limit: usize) -> Result<Excerpt, EvidenceError> {
181        let inner = self.inner()?;
182        let kind = inner.declared::<K>()?;
183        let ceiling = inner.vault.ceiling();
184        if limit == 0 || limit > ceiling {
185            return Err(EvidenceError::Bound { requested: limit, ceiling });
186        }
187
188        let staged = inner.draft(&kind, scope, key);
189        let (bytes, length) = match staged {
190            Some(open) => {
191                let read = inner.vault.read_draft(&open.name, offset, limit).await?;
192                open.observed(read.1);
193                read
194            }
195            None => {
196                let name = encode(&kind, scope, key);
197                if !inner.is_kept(&kind, scope, key) {
198                    return Err(EvidenceError::Unknown);
199                }
200                inner.vault.read(&name, offset, limit).await?
201            }
202        };
203
204        Ok(Excerpt {
205            requested: offset,
206            offset: offset.min(length),
207            bytes,
208            length,
209        })
210    }
211
212    pub fn list(&self, scope: &str) -> Vec<EvidenceRecord> {
213        let Ok(inner) = self.inner() else {
214            return Vec::new();
215        };
216        let mut listing: BTreeMap<(DefinitionId, String), EvidenceRecord> = BTreeMap::new();
217
218        for (at, open) in inner.scope(&inner.drafts, scope) {
219            listing.insert(
220                at,
221                EvidenceRecord::new(
222                    open.kind.clone(),
223                    open.version,
224                    open.scope.clone(),
225                    open.key.clone(),
226                    open.length(),
227                    open.described.clone(),
228                ),
229            );
230        }
231        for (at, record) in inner.scope(&inner.kept, scope) {
232            listing.insert(at, record);
233        }
234
235        listing.into_values().collect()
236    }
237
238    pub async fn finalize(&self, draft: &Draft) -> Result<EvidenceRecord, EvidenceError> {
239        let inner = self.inner()?;
240        let open = &draft.open;
241        inner.begin(open)?;
242
243        let finalized = inner.finalizing(open).await;
244        inner.end(open, finalized.is_ok());
245
246        finalized
247    }
248
249    pub async fn abandon(&self, draft: &Draft) -> Result<(), EvidenceError> {
250        let inner = self.inner()?;
251        let open = &draft.open;
252        inner.begin(open)?;
253
254        let quarantined = inner.vault.quarantine(&open.name).await;
255        inner.forget(open);
256        open.release(open.length());
257
258        quarantined
259    }
260
261    pub async fn finalize_scope(&self, scope: &str) -> Result<Vec<EvidenceRecord>, EvidenceError> {
262        let inner = self.inner()?;
263        let mut finalized = Vec::new();
264
265        for (_, open) in inner.scope(&inner.drafts, scope) {
266            finalized.push(self.finalize(&Draft { open }).await?);
267        }
268
269        Ok(finalized)
270    }
271
272    pub async fn abandon_scope(&self, scope: &str) -> Result<(), EvidenceError> {
273        let inner = self.inner()?;
274
275        for (_, open) in inner.scope(&inner.drafts, scope) {
276            self.abandon(&Draft { open }).await?;
277        }
278
279        Ok(())
280    }
281
282    pub fn watch(&self, draft: &Draft) -> Result<watch::Receiver<u64>, EvidenceError> {
283        let inner = self.inner()?;
284        let open = &draft.open;
285        if open.released() {
286            return Err(EvidenceError::Unknown);
287        }
288        let (receiver, first) = open.watch();
289        if first && open.external.is_some() {
290            let cadence = inner.vault.cadence();
291            executor::task(observing(Arc::downgrade(open), cadence));
292        }
293
294        Ok(receiver)
295    }
296
297    async fn opening<K: EvidenceKind>(
298        &self, scope: String, key: String, details: K::Details, foreign: bool,
299    ) -> Result<Draft, EvidenceError> {
300        let inner = self.inner()?;
301        let kind = inner.declared::<K>()?;
302        let version = inner.kinds[&kind];
303        let described = serde_json::to_vec(&details).map_err(|error| EvidenceError::Malformed {
304            kind: kind.clone(),
305            cause: error.to_string(),
306        })?;
307        if inner.draft(&kind, &scope, &key).is_some() || inner.is_kept(&kind, &scope, &key) {
308            return Err(EvidenceError::Conflict { kind, scope, key });
309        }
310        let name = encode(&kind, &scope, &key);
311        let length = inner.vault.open(&name).await?;
312        let external = match foreign {
313            true => Some(inner.vault.external(&name).await?.ok_or(EvidenceError::Unsupported)?),
314            false => None,
315        };
316        let open = Arc::new(Open::new(
317            Staged {
318                kind: kind.clone(),
319                version,
320                scope: scope.clone(),
321                key: key.clone(),
322                name,
323                described,
324                external,
325                length,
326            },
327            Arc::clone(&inner.vault),
328        ));
329        inner.drafts.lock().expect("the evidence index is healthy").entry(scope).or_default().insert((kind, key), Arc::clone(&open));
330
331        Ok(Draft { open })
332    }
333
334    fn inner(&self) -> Result<&Arc<Inner<A>>, EvidenceError> {
335        self.inner.as_ref().ok_or(EvidenceError::Unserved)
336    }
337}
338
339impl<A: Application> Clone for Evidence<A> {
340    fn clone(&self) -> Self {
341        Self { inner: self.inner.clone() }
342    }
343}
344
345impl<A: Application> Inner<A> {
346    fn declared<K: EvidenceKind>(&self) -> Result<DefinitionId, EvidenceError> {
347        let (kind, _) = K::definition();
348        match self.kinds.contains_key(&kind) {
349            true => Ok(kind),
350            false => Err(EvidenceError::Undeclared { kind }),
351        }
352    }
353
354    fn draft(&self, kind: &DefinitionId, scope: &str, key: &str) -> Option<Arc<Open>> {
355        self.drafts.lock().expect("the evidence index is healthy").get(scope)?.get(&(kind.clone(), key.to_owned())).map(Arc::clone)
356    }
357
358    fn is_kept(&self, kind: &DefinitionId, scope: &str, key: &str) -> bool {
359        self.kept
360            .lock()
361            .expect("the evidence index is healthy")
362            .get(scope)
363            .is_some_and(|held| held.contains_key(&(kind.clone(), key.to_owned())))
364    }
365
366    fn scope<T: Clone>(&self, index: &Mutex<Held<T>>, scope: &str) -> Vec<((DefinitionId, String), T)> {
367        index
368            .lock()
369            .expect("the evidence index is healthy")
370            .get(scope)
371            .map(|held| held.iter().map(|(at, value)| (at.clone(), value.clone())).collect())
372            .unwrap_or_default()
373    }
374
375    fn begin(&self, open: &Arc<Open>) -> Result<(), EvidenceError> {
376        match self.draft(&open.kind, &open.scope, &open.key) {
377            Some(held) if Arc::ptr_eq(&held, open) => match open.begin_finalize() {
378                true => Ok(()),
379                false => Err(EvidenceError::Finalizing),
380            },
381            Some(_) | None => Err(EvidenceError::Unknown),
382        }
383    }
384
385    fn end(&self, open: &Arc<Open>, ended: bool) {
386        if !ended {
387            open.end_finalize();
388        }
389    }
390
391    async fn finalizing(&self, open: &Arc<Open>) -> Result<EvidenceRecord, EvidenceError> {
392        // Durable first, then read back: a digest folded from what a producer
393        // said it wrote proves nothing about what survived.
394        let (digest, length) = self.vault.verify(&open.name).await?;
395        let written = open.length();
396        if open.external.is_none() && written != length {
397            return Err(EvidenceError::Corrupt { written, found: length });
398        }
399
400        let mut record =
401            EvidenceRecord::new(open.kind.clone(), open.version, open.scope.clone(), open.key.clone(), length, open.described.clone());
402        let receipt = crate::records::RecordManager::new(self.state.clone())
403            .publish_system(record.recorded(digest))
404            .await
405            .map_err(|_| EvidenceError::Unavailable)?;
406        self.state.wait_until_stored(receipt.position).await;
407        record.finalized = Some(At {
408            value: digest,
409            position: receipt.position,
410        });
411
412        // Visible before removed: the fact is already history, so the
413        // evidence enters the finalized index before its draft leaves the
414        // active one and a concurrent listing never falls through the gap
415        // between them.
416        self.kept
417            .lock()
418            .expect("the evidence index is healthy")
419            .entry(record.scope.clone())
420            .or_default()
421            .insert((record.kind.clone(), record.key.clone()), record.clone());
422        self.vault.claim(&open.name).await?;
423        self.forget(open);
424        open.release(length);
425
426        Ok(record)
427    }
428
429    fn forget(&self, open: &Arc<Open>) {
430        let mut drafts = self.drafts.lock().expect("the evidence index is healthy");
431        let Some(held) = drafts.get_mut(&open.scope) else {
432            return;
433        };
434        held.remove(&(open.kind.clone(), open.key.clone()));
435        if held.is_empty() {
436            drafts.remove(&open.scope);
437        }
438    }
439}
440
441/* === Public Functions === */
442
443/* === Private Functions === */
444
445async fn observing(draft: Weak<Open>, cadence: std::time::Duration) {
446    loop {
447        executor::sleep(cadence).await;
448        let Some(open) = draft.upgrade() else {
449            return;
450        };
451        if open.released() || open.idle() {
452            return;
453        }
454        if open.observe().await.is_err() {
455            return;
456        }
457    }
458}
459
460fn encode(kind: &DefinitionId, scope: &str, key: &str) -> String {
461    let mut name = String::new();
462    for (index, part) in [kind.as_str(), scope, key].into_iter().enumerate() {
463        if index > 0 {
464            name.push('~');
465        }
466        for byte in part.bytes() {
467            match byte {
468                b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' => name.push(byte as char),
469                _ => name.push_str(&format!("%{byte:02x}")),
470            }
471        }
472    }
473
474    name
475}
476
477/* === Tests === */
harmos_runtime/evidence/mod.rs

harmos_runtime/evidence/
mod.rs
