harmos-runtime 0.1.5

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
122
123
124
125
126
127
128
129
130
131
132

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

/* === Modules === */

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

/* === Imports === */

// Standard Library Imports

// Third Party Imports

// Local Imports
use super::*;

/* === Constants === */

/* === Macros === */

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

/* === Structs === */

/* === Enums === */

/* === Traits === */

/* === Implementations === */

impl OperationManager {
    pub async fn unregister(&self, id: &str) -> Result<()> {
        self.unregister_selected(id, None).await
    }

    pub async fn unregister_from(&self, owner: &str, id: &str) -> Result<()> {
        self.unregister_selected(id, Some(owner)).await
    }

    pub(super) async fn unregister_selected(&self, id: &str, owner: Option<&str>) -> Result<()> {
        let entry = {
            let state = self.registry.state.lock().expect("operation registry lock is healthy");
            Arc::clone(state.select(id, owner)?)
        };
        #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
        if let Some(scope) = &entry.plugin {
            // Notify the staging gate before revocation, outside registry guards.
            scope.unbind_operations();
        }
        let retired = {
            let mut state = self.registry.state.lock().expect("operation registry lock is healthy");
            if !state.owned(id, entry.owner.as_str()).is_some_and(|current| Arc::ptr_eq(current, &entry)) {
                return Err(Error::NotRegistered { domain: "operation", name: id.to_owned() });
            }
            let _pending = entry.pending.lock().expect("operation settlement lock is healthy");
            entry.retirement.store(Retirement::Explicit as u8, Ordering::Release);
            entry.active.store(false, Ordering::Release);
            let retired = state.remove_cache(|cached| !cached.target.upgrade().is_some_and(|target| !Arc::ptr_eq(&target, &entry)));
            drop(_pending);
            retired
        };
        drop(retired);
        settle(&entry).await?;
        let mut state = self.registry.state.lock().expect("operation registry lock is healthy");
        if state.owned(id, entry.owner.as_str()).is_some_and(|current| Arc::ptr_eq(current, &entry)) {
            state.remove_owned(id, entry.owner.as_str());
        }
        Ok(())
    }

    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
    pub(in crate::operations) fn fence(&self, entries: &[Arc<Entry>], abandoned: bool) {
        let retired = {
            let mut state = self.registry.state.lock().expect("operation registry lock is healthy");
            for entry in entries {
                let _pending = entry.pending.lock().expect("operation settlement lock is healthy");
                // A by-name unregister retains its reservation for retry even
                // when the plugin receipt is abandoned during that wait.
                let _ = entry.retirement.try_update(Ordering::AcqRel, Ordering::Acquire, |state| {
                    (state != Retirement::Explicit as u8).then_some(if abandoned {
                        Retirement::Abandoned as u8
                    } else {
                        Retirement::Retained as u8
                    })
                });
                entry.active.store(false, Ordering::Release);
            }
            state.remove_cache(|cached| {
                cached.target.upgrade().is_some_and(|target| entries.iter().any(|entry| Arc::ptr_eq(&target, entry)))
            })
        };
        drop(retired);
    }

    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
    pub(in crate::operations) fn remove_settled(&self, entries: &[Arc<Entry>]) {
        let mut state = self.registry.state.lock().expect("operation registry lock is healthy");
        for entry in entries {
            if entry.settled() && state.owned(entry.id(), entry.owner.as_str()).is_some_and(|current| Arc::ptr_eq(current, entry)) {
                state.remove_owned(entry.id(), entry.owner.as_str());
            }
        }
        // The caller owns every entry until after this guard is released.
    }

    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
    pub(in crate::operations) async fn settle_entries(&self, entries: &[Arc<Entry>]) -> Result<()> {
        self.settle_owned_entries(entries).await?;
        self.remove_settled(entries);
        Ok(())
    }

    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
    pub(in crate::operations) async fn settle_owned_entries(&self, entries: &[Arc<Entry>]) -> Result<()> {
        for entry in entries {
            settle(entry).await?;
        }
        Ok(())
    }
}

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

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

/* === Tests === */