harmos-runtime 0.1.6

Harmos runtime: managed state, undo, recovery, background work and typed streams.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121

/* ***
 *
 * Harmos
 *
 * SPDX-License-Identifier: Apache-2.0
 *
 * ***
 */

/* === Modules === */

/* === Re-Exports === */

/* === Imports === */

// Standard Library Imports
use std::collections::BTreeSet;
// Third Party Imports

// Local Imports
use super::{Error, Host, Registration};
use crate::{Application, Runtime, state::Never};

/* === Constants === */

/* === Macros === */

/* === Type Aliases === */

/* === Structs === */

struct Collection;

struct Cleanup {
    runtime: Option<Runtime<Collection>>,
    executor: tokio::runtime::Handle,
}

/* === Enums === */

/* === Traits === */

/* === Implementations === */

impl Application for Collection {
    type State = ();
    type Transaction = Never;
    type Record = ();
    type Streams = ();
    type Resources = ();
}

impl Drop for Cleanup {
    fn drop(&mut self) {
        if let Some(runtime) = self.runtime.take() {
            self.executor.spawn(async move {
                if let Err(error) = runtime.stop().await {
                    let error = harmos_domain::failure::Report(&error);
                    tracing::error!(%error, "catalog runtime shutdown failed after cancellation");
                }
            });
        }
    }
}

/* === Public Functions === */

// Validate a guest's catalog registrations, sorted by namespace: at most 256
// distinct artifact namespaces with nonzero schemas, 16 MiB in total.
pub fn validate(mut entries: Vec<Registration>) -> Result<Vec<Registration>, Error> {
    if entries.len() > 256 {
        return Err(invalid("catalog exceeds 256 namespaces"));
    }
    let mut namespaces = BTreeSet::new();
    let mut size = 0usize;
    for entry in &entries {
        if !harmos_domain::artifact_subset(&entry.namespace) || entry.schema == 0 || !namespaces.insert(entry.namespace.clone()) {
            return Err(invalid("catalog has an invalid or duplicate namespace/schema"));
        }
        size = size
            .checked_add(entry.namespace.len())
            .and_then(|size| size.checked_add(entry.payload.len()))
            .filter(|size| *size <= 16 * 1024 * 1024)
            .ok_or_else(|| invalid("catalog exceeds 16 MiB"))?;
    }
    entries.sort_by(|left, right| left.namespace.cmp(&right.namespace));
    Ok(entries)
}

pub async fn registrations(bytes: &[u8]) -> Result<Vec<Registration>, Error> {
    let runtime = Runtime::<Collection>::builder((), ())
        .runtime()
        .await
        .map_err(|error| Error::Guest { message: format!("catalog runtime could not start: {error}") })?;
    let mut cleanup = Cleanup { runtime: Some(runtime), executor: tokio::runtime::Handle::current() };
    let result = async {
        let host = Host::new(cleanup.runtime.as_ref().expect("runtime is owned until cleanup"))?;
        let guest =
            host.load(bytes)?.register_with_limits(Some(super::host::store_limits(crate::config::PluginGuestSettings::default()))).await?;
        validate(guest.registrations().to_vec())
    }
    .await;
    let runtime = cleanup.runtime.take().expect("runtime has not been stopped");
    // The task retains shutdown ownership even if this final await is cancelled,
    // so the runtime is moved into it rather than borrowed across the spawn.
    cleanup
        .executor
        .spawn(async move { runtime.stop().await })
        .await
        .map_err(|error| invalid(&format!("catalog runtime shutdown task failed: {error}")))?
        .map_err(|error| invalid(&format!("catalog runtime shutdown failed: {error}")))?;
    result
}

/* === Private Functions === */

fn invalid(message: &str) -> Error {
    Error::Guest { message: message.to_owned() }
}

/* === Tests === */