harmos_runtime/operations/manager/
retirement.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
17
18// Third Party Imports
19
20// Local Imports
21use super::*;
22
23/* === Constants === */
24
25/* === Macros === */
26
27/* === Type Aliases === */
28
29/* === Structs === */
30
31/* === Enums === */
32
33/* === Traits === */
34
35/* === Implementations === */
36
37impl OperationManager {
38    pub async fn unregister(&self, id: &str) -> Result<()> {
39        self.unregister_selected(id, None).await
40    }
41
42    pub async fn unregister_from(&self, owner: &str, id: &str) -> Result<()> {
43        self.unregister_selected(id, Some(owner)).await
44    }
45
46    pub(super) async fn unregister_selected(&self, id: &str, owner: Option<&str>) -> Result<()> {
47        let entry = {
48            let state = self.registry.state.lock().expect("operation registry lock is healthy");
49            Arc::clone(state.select(id, owner)?)
50        };
51        #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
52        if let Some(scope) = &entry.plugin {
53            // Notify the staging gate before revocation, outside registry guards.
54            scope.unbind_operations();
55        }
56        let retired = {
57            let mut state = self.registry.state.lock().expect("operation registry lock is healthy");
58            if !state.owned(id, entry.owner.as_str()).is_some_and(|current| Arc::ptr_eq(current, &entry)) {
59                return Err(Error::NotRegistered { domain: "operation", name: id.to_owned() });
60            }
61            let _pending = entry.pending.lock().expect("operation settlement lock is healthy");
62            entry.retirement.store(Retirement::Explicit as u8, Ordering::Release);
63            entry.active.store(false, Ordering::Release);
64            let retired = state.remove_cache(|cached| !cached.target.upgrade().is_some_and(|target| !Arc::ptr_eq(&target, &entry)));
65            drop(_pending);
66            retired
67        };
68        drop(retired);
69        settle(&entry).await?;
70        let mut state = self.registry.state.lock().expect("operation registry lock is healthy");
71        if state.owned(id, entry.owner.as_str()).is_some_and(|current| Arc::ptr_eq(current, &entry)) {
72            state.remove_owned(id, entry.owner.as_str());
73        }
74        Ok(())
75    }
76
77    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
78    pub(in crate::operations) fn fence(&self, entries: &[Arc<Entry>], abandoned: bool) {
79        let retired = {
80            let mut state = self.registry.state.lock().expect("operation registry lock is healthy");
81            for entry in entries {
82                let _pending = entry.pending.lock().expect("operation settlement lock is healthy");
83                // A by-name unregister retains its reservation for retry even
84                // when the plugin receipt is abandoned during that wait.
85                let _ = entry.retirement.try_update(Ordering::AcqRel, Ordering::Acquire, |state| {
86                    (state != Retirement::Explicit as u8).then_some(if abandoned {
87                        Retirement::Abandoned as u8
88                    } else {
89                        Retirement::Retained as u8
90                    })
91                });
92                entry.active.store(false, Ordering::Release);
93            }
94            state.remove_cache(|cached| {
95                cached.target.upgrade().is_some_and(|target| entries.iter().any(|entry| Arc::ptr_eq(&target, entry)))
96            })
97        };
98        drop(retired);
99    }
100
101    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
102    pub(in crate::operations) fn remove_settled(&self, entries: &[Arc<Entry>]) {
103        let mut state = self.registry.state.lock().expect("operation registry lock is healthy");
104        for entry in entries {
105            if entry.settled() && state.owned(entry.id(), entry.owner.as_str()).is_some_and(|current| Arc::ptr_eq(current, entry)) {
106                state.remove_owned(entry.id(), entry.owner.as_str());
107            }
108        }
109        // The caller owns every entry until after this guard is released.
110    }
111
112    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
113    pub(in crate::operations) async fn settle_entries(&self, entries: &[Arc<Entry>]) -> Result<()> {
114        self.settle_owned_entries(entries).await?;
115        self.remove_settled(entries);
116        Ok(())
117    }
118
119    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
120    pub(in crate::operations) async fn settle_owned_entries(&self, entries: &[Arc<Entry>]) -> Result<()> {
121        for entry in entries {
122            settle(entry).await?;
123        }
124        Ok(())
125    }
126}
127
128/* === Public Functions === */
129
130/* === Private Functions === */
131
132/* === Tests === */
harmos_runtime/operations/manager/retirement.rs

harmos_runtime/operations/manager/
retirement.rs
