harmos_runtime/evidence/
draft.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17use std::path::{Path, PathBuf};
18use std::sync::{Arc, Mutex};
19// Third Party Imports
20use tokio::sync::watch;
21// Local Imports
22use super::EvidenceError;
23use super::vault::Keeping;
24use crate::DefinitionId;
25
26/* === Constants === */
27
28/* === Macros === */
29
30/* === Type Aliases === */
31
32/* === Structs === */
33
34#[derive(Clone)]
35pub struct Draft {
36    pub(crate) open: Arc<Open>,
37}
38
39pub(crate) struct Open {
40    pub(crate) kind: DefinitionId,
41    pub(crate) version: u32,
42    pub(crate) scope: String,
43    pub(crate) key: String,
44    pub(crate) name: String,
45    pub(crate) described: Vec<u8>,
46    pub(crate) external: Option<PathBuf>,
47    vault: Arc<dyn Keeping>,
48    writing: tokio::sync::Mutex<()>,
49    growth: Mutex<Growth>,
50}
51
52struct Growth {
53    length: u64,
54    watchers: Option<watch::Sender<u64>>,
55    finalizing: bool,
56    released: bool,
57}
58
59pub(crate) struct Staged {
60    pub(crate) kind: DefinitionId,
61    pub(crate) version: u32,
62    pub(crate) scope: String,
63    pub(crate) key: String,
64    pub(crate) name: String,
65    pub(crate) described: Vec<u8>,
66    pub(crate) external: Option<PathBuf>,
67    pub(crate) length: u64,
68}
69
70/* === Enums === */
71
72/* === Traits === */
73
74/* === Implementations === */
75
76impl Draft {
77    pub fn kind(&self) -> &DefinitionId {
78        &self.open.kind
79    }
80
81    pub fn scope(&self) -> &str {
82        &self.open.scope
83    }
84
85    pub fn key(&self) -> &str {
86        &self.open.key
87    }
88
89    pub fn length(&self) -> u64 {
90        self.open.length()
91    }
92
93    pub fn path(&self) -> Option<&Path> {
94        self.open.external.as_deref()
95    }
96
97    pub async fn write(&self, bytes: &[u8]) -> Result<u64, EvidenceError> {
98        self.open.append(bytes).await
99    }
100}
101
102impl std::fmt::Debug for Draft {
103    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
104        formatter
105            .debug_struct("Draft")
106            .field("kind", &self.open.kind)
107            .field("scope", &self.open.scope)
108            .field("key", &self.open.key)
109            .field("length", &self.open.length())
110            .finish()
111    }
112}
113
114impl Open {
115    pub(crate) fn new(staged: Staged, vault: Arc<dyn Keeping>) -> Self {
116        let Staged { kind, version, scope, key, name, described, external, length } = staged;
117
118        Self {
119            kind,
120            version,
121            scope,
122            key,
123            name,
124            described,
125            external,
126            vault,
127            writing: tokio::sync::Mutex::new(()),
128            growth: Mutex::new(Growth { length, watchers: None, finalizing: false, released: false }),
129        }
130    }
131
132    pub(crate) fn length(&self) -> u64 {
133        self.held().length
134    }
135
136    pub(crate) fn released(&self) -> bool {
137        self.held().released
138    }
139
140    async fn append(&self, bytes: &[u8]) -> Result<u64, EvidenceError> {
141        if self.external.is_some() {
142            return Err(EvidenceError::Foreign);
143        }
144        let _writing = self.writing.lock().await;
145        if self.released() {
146            return Err(EvidenceError::Unknown);
147        }
148        let length = self.vault.append(&self.name, bytes).await?;
149        self.observed(length);
150
151        Ok(length)
152    }
153
154    pub(crate) fn observed(&self, length: u64) {
155        let mut held = self.held();
156        if held.length == length {
157            return;
158        }
159        held.length = length;
160        if let Some(watchers) = &held.watchers {
161            let _ = watchers.send(length);
162        }
163    }
164
165    pub(crate) fn watch(&self) -> (watch::Receiver<u64>, bool) {
166        let mut held = self.held();
167        match &held.watchers {
168            Some(watchers) => (watchers.subscribe(), false),
169            None => {
170                let (sender, receiver) = watch::channel(held.length);
171                held.watchers = Some(sender);
172
173                (receiver, true)
174            }
175        }
176    }
177
178    pub(crate) fn idle(&self) -> bool {
179        let mut held = self.held();
180        let gone = held.watchers.as_ref().is_some_and(|watchers| watchers.receiver_count() == 0);
181        if gone {
182            held.watchers = None;
183        }
184
185        gone
186    }
187
188    pub(crate) fn begin_finalize(&self) -> bool {
189        let mut held = self.held();
190        if held.finalizing || held.released {
191            return false;
192        }
193        held.finalizing = true;
194
195        true
196    }
197
198    pub(crate) fn end_finalize(&self) {
199        self.held().finalizing = false;
200    }
201
202    pub(crate) async fn observe(&self) -> Result<(), EvidenceError> {
203        let length = self.vault.length(&self.name).await?;
204        self.observed(length);
205
206        Ok(())
207    }
208
209    pub(crate) fn release(&self, length: u64) {
210        let mut held = self.held();
211        held.released = true;
212        held.length = length;
213        if let Some(watchers) = held.watchers.take() {
214            let _ = watchers.send(length);
215        }
216    }
217
218    fn held(&self) -> std::sync::MutexGuard<'_, Growth> {
219        self.growth.lock().expect("the draft lock is healthy")
220    }
221}
222
223/* === Public Functions === */
224
225/* === Private Functions === */
226
227/* === Tests === */
harmos_runtime/evidence/draft.rs

harmos_runtime/evidence/
draft.rs
