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 {
117            kind,
118            version,
119            scope,
120            key,
121            name,
122            described,
123            external,
124            length,
125        } = staged;
126
127        Self {
128            kind,
129            version,
130            scope,
131            key,
132            name,
133            described,
134            external,
135            vault,
136            writing: tokio::sync::Mutex::new(()),
137            growth: Mutex::new(Growth {
138                length,
139                watchers: None,
140                finalizing: false,
141                released: false,
142            }),
143        }
144    }
145
146    pub(crate) fn length(&self) -> u64 {
147        self.held().length
148    }
149
150    pub(crate) fn released(&self) -> bool {
151        self.held().released
152    }
153
154    async fn append(&self, bytes: &[u8]) -> Result<u64, EvidenceError> {
155        if self.external.is_some() {
156            return Err(EvidenceError::Foreign);
157        }
158        let _writing = self.writing.lock().await;
159        if self.released() {
160            return Err(EvidenceError::Unknown);
161        }
162        let length = self.vault.append(&self.name, bytes).await?;
163        self.observed(length);
164
165        Ok(length)
166    }
167
168    pub(crate) fn observed(&self, length: u64) {
169        let mut held = self.held();
170        if held.length == length {
171            return;
172        }
173        held.length = length;
174        if let Some(watchers) = &held.watchers {
175            let _ = watchers.send(length);
176        }
177    }
178
179    pub(crate) fn watch(&self) -> (watch::Receiver<u64>, bool) {
180        let mut held = self.held();
181        match &held.watchers {
182            Some(watchers) => (watchers.subscribe(), false),
183            None => {
184                let (sender, receiver) = watch::channel(held.length);
185                held.watchers = Some(sender);
186
187                (receiver, true)
188            }
189        }
190    }
191
192    pub(crate) fn idle(&self) -> bool {
193        let mut held = self.held();
194        let gone = held.watchers.as_ref().is_some_and(|watchers| watchers.receiver_count() == 0);
195        if gone {
196            held.watchers = None;
197        }
198
199        gone
200    }
201
202    pub(crate) fn begin_finalize(&self) -> bool {
203        let mut held = self.held();
204        if held.finalizing || held.released {
205            return false;
206        }
207        held.finalizing = true;
208
209        true
210    }
211
212    pub(crate) fn end_finalize(&self) {
213        self.held().finalizing = false;
214    }
215
216    pub(crate) async fn observe(&self) -> Result<(), EvidenceError> {
217        let length = self.vault.length(&self.name).await?;
218        self.observed(length);
219
220        Ok(())
221    }
222
223    pub(crate) fn release(&self, length: u64) {
224        let mut held = self.held();
225        held.released = true;
226        held.length = length;
227        if let Some(watchers) = held.watchers.take() {
228            let _ = watchers.send(length);
229        }
230    }
231
232    fn held(&self) -> std::sync::MutexGuard<'_, Growth> {
233        self.growth.lock().expect("the draft lock is healthy")
234    }
235}
236
237/* === Public Functions === */
238
239/* === Private Functions === */
240
241/* === Tests === */
harmos_runtime/evidence/draft.rs

harmos_runtime/evidence/
draft.rs
