harmos_runtime/operations/manager/
mod.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12mod admission;
13mod cache;
14mod construction;
15mod entry;
16mod grants;
17mod inspection;
18mod lookup;
19mod registration;
20mod registry;
21mod retirement;
22mod shutdown;
23mod staging;
24mod submission;
25
26mod discovery;
27
28mod residents;
29
30/* === Re-Exports === */
31
32/* === Imports === */
33
34// Standard Library Imports
35use std::any::Any;
36use std::collections::{BTreeMap, VecDeque};
37use std::sync::{
38    Arc, Mutex, Weak,
39    atomic::{AtomicBool, AtomicU8, Ordering},
40};
41// Third Party Imports
42use tokio::sync::{OwnedSemaphorePermit, Semaphore};
43// Local Imports
44use super::admission::{Admission, Lanes};
45use super::binding::Typed;
46use super::caller::{Identity, Origin, Source};
47use super::{OperationGeneration, OperationInstanceId, Running};
48
49use super::instances::Instances;
50use crate::operations::{CacheKey, OperationDeclaration, OperationDefinition};
51#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
52use crate::plugin::PluginScope;
53use crate::supervision::executor;
54use crate::supervision::instance::Instance;
55use crate::{Error, Identifier, Result, operations::Lane, operations::Scope, state::Position};
56
57/* === Constants === */
58
59const DEFAULT_CACHE_CAPACITY: usize = 64;
60
61/* === Macros === */
62
63/* === Type Aliases === */
64
65type CachedResponse = harmos_sdk::__private::ErasedResponse;
66
67/* === Structs === */
68
69#[derive(Clone)]
70pub struct OperationManager<L: Lane> {
71    inner: Manager,
72    lane: std::marker::PhantomData<fn() -> L>,
73}
74
75#[derive(Clone)]
76pub struct Manager {
77    pub(crate) registry: Arc<Registry>,
78    pub(super) schema_settings: Arc<crate::config::SchemaSettings>,
79    pub(super) instances: Arc<Instances>,
80    pub(super) lanes: Arc<Lanes>,
81    pub(crate) preparation: Option<Arc<Semaphore>>,
82    pub(super) context: Arc<std::sync::OnceLock<super::CallContext>>,
83    pub(crate) resident: super::resident::Registry,
84}
85
86#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
87#[derive(Clone)]
88pub(crate) struct WeakOperationManager {
89    schema_settings: std::sync::Weak<crate::config::SchemaSettings>,
90    registry: Weak<Registry>,
91    instances: Weak<Instances>,
92    lanes: Weak<Lanes>,
93    preparation: Option<Weak<Semaphore>>,
94    context: Weak<std::sync::OnceLock<super::CallContext>>,
95    resident: super::resident::WeakRegistry,
96}
97
98pub(crate) struct Registry {
99    pub(crate) service_runner: crate::services::ServiceRunner,
100    middleware: super::middleware::Configuration,
101    state: Mutex<State>,
102}
103
104#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
105pub(crate) struct GenerationCommit<'a> {
106    _state: std::sync::MutexGuard<'a, State>,
107}
108
109struct State {
110    closing: bool,
111    entries: BTreeMap<String, BTreeMap<String, Arc<Entry>>>,
112
113    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
114    staging: Vec<std::sync::Weak<PluginScope>>,
115
116    #[cfg(feature = "grpc")]
117    grpc: BTreeMap<String, Weak<Entry>>,
118
119    #[cfg(feature = "grpc")]
120    grpc_services: std::collections::BTreeSet<String>,
121
122    cache: VecDeque<Cached>,
123    cache_capacity: usize,
124}
125
126pub(super) struct Entry {
127    pub(super) carrier: harmos_sdk::__private::ResourceSet,
128    owner: Identifier,
129    definition: OperationDefinition,
130    pub(super) contract: Option<crate::operations::Contract>,
131
132    #[cfg(feature = "grpc")]
133    pub(super) grpc: Option<crate::grpc::routing::Handler>,
134
135    pub(super) generation: OperationGeneration,
136    lane: &'static str,
137
138    // A resident control runs until cancelled, even while its plugin retires.
139    #[cfg(feature = "plugin-sidecar")]
140    resident: bool,
141
142    active: AtomicBool,
143    publication: Arc<AtomicBool>,
144    retirement: AtomicU8,
145    pending: Mutex<BTreeMap<OperationInstanceId, Arc<Instance>>>,
146    pub(in crate::operations) binding: Arc<dyn Any + Send + Sync>,
147
148    #[cfg(feature = "plugin-sidecar")]
149    pub(super) native_binding: Option<Arc<dyn Any + Send + Sync>>,
150
151    pub(super) wire: Option<super::wire::Dispatch>,
152
153    #[cfg(feature = "plugin-sidecar")]
154    pub(super) native: Option<Arc<super::native::Binding>>,
155
156    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
157    plugin: Option<PluginScope>,
158
159    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
160    pub(super) sdk: Option<harmos_operations::lifecycle::Action>,
161}
162
163pub(super) struct Submission {
164    pub(super) submitted: tokio::time::Instant,
165    pub(super) entry: Arc<Entry>,
166    pub(super) cause: Option<Position>,
167    pub(super) preparation: Option<OwnedSemaphorePermit>,
168}
169
170// The caller's grant authority for one admitted request. Typed bindings check
171// the request grant at admission; encoded bindings learn it from the provider
172// while preparing and check it before any attempt runs.
173#[derive(Clone, Debug)]
174pub(super) struct GrantAuthority {
175    owner: String,
176    operation: String,
177
178    // A cross-owner plugin caller and its declared requirement for this operation.
179    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
180    plugin: Option<(String, Option<harmos_domain::inventory::contracts::Required>)>,
181
182    // External Authority Never Replaces A Plugin Caller's Own Requirement.
183    external: Option<crate::operations::authorization::Authorization>,
184}
185
186struct Cached {
187    target: Weak<Entry>,
188    key: CacheKey,
189    identity: Identity,
190    response: CachedResponse,
191}
192
193/* === Enums === */
194
195#[repr(u8)]
196enum Retirement {
197    Retained,
198    Abandoned,
199    Explicit,
200    Runtime,
201}
202
203/* === Traits === */
204
205/* === Implementations === */
206
207impl Manager {
208    pub(crate) fn resources(&self) -> &harmos_sdk::__private::ResourceSet {
209        &self.instances.resources
210    }
211
212    pub(super) fn resources_for(&self, entry: &Entry) -> harmos_sdk::__private::ResourceSet {
213        harmos_sdk::ExecutionContext::new(entry.carrier.clone(), super::CallContext::default())
214            .with_resources(self.resources().clone())
215            .resources()
216            .clone()
217    }
218}
219
220#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
221impl std::fmt::Debug for WeakOperationManager {
222    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
223        formatter.debug_struct("WeakOperationManager").finish_non_exhaustive()
224    }
225}
226
227#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
228impl WeakOperationManager {
229    pub(crate) fn upgrade(&self) -> Option<Manager> {
230        Some(Manager {
231            schema_settings: self.schema_settings.upgrade()?,
232            registry: self.registry.upgrade()?,
233            instances: self.instances.upgrade()?,
234            lanes: self.lanes.upgrade()?,
235            preparation: match &self.preparation {
236                Some(budget) => Some(budget.upgrade()?),
237                None => None,
238            },
239            context: self.context.upgrade()?,
240            resident: self.resident.upgrade()?,
241        })
242    }
243}
244
245#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
246impl Manager {
247    pub(crate) fn downgrade(&self) -> WeakOperationManager {
248        WeakOperationManager {
249            schema_settings: Arc::downgrade(&self.schema_settings),
250            registry: Arc::downgrade(&self.registry),
251            instances: Arc::downgrade(&self.instances),
252            lanes: Arc::downgrade(&self.lanes),
253            preparation: self.preparation.as_ref().map(Arc::downgrade),
254            context: Arc::downgrade(&self.context),
255            resident: self.resident.downgrade(),
256        }
257    }
258}
259
260impl<L: Lane> OperationManager<L> {
261    pub(crate) fn new(inner: Manager) -> Self {
262        Self { inner, lane: std::marker::PhantomData }
263    }
264
265    pub(crate) fn execution(&self) -> Manager {
266        self.inner.clone()
267    }
268
269    pub async fn register<C: harmos_sdk::operations::OperationCatalog>(&self, lane: L) -> Result<()> {
270        use crate::operations::RegistrationExt;
271        self.register_batch(C::batch().lane(lane)).await
272    }
273
274    pub(crate) async fn register_batch(&self, registration: crate::operations::Registration<L>) -> Result<()> {
275        self.inner.register_native(registration)
276    }
277
278    pub async fn unregister(&self, id: &harmos_domain::QualifiedId) -> Result<()> {
279        self.inner.unregister(id).await
280    }
281
282    pub fn with_context(&self, context: super::CallContext) -> Self {
283        Self::new(self.inner.with_context(context))
284    }
285}
286
287impl<L: Lane> std::ops::Deref for OperationManager<L> {
288    type Target = Manager;
289    fn deref(&self) -> &Self::Target {
290        &self.inner
291    }
292}
293
294/* === Public Functions === */
295
296/* === Private Functions === */
297
298#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
299fn grant_allowed(required: Option<&harmos_domain::inventory::contracts::Required>, grant: Option<&harmos_domain::plugin::Grant>) -> bool {
300    match grant {
301        None => true,
302        Some(grant) => required.is_some_and(|required| required.grants(grant)),
303    }
304}
305
306async fn settle(entry: &Entry) -> Result<()> {
307    let pending = entry.pending.lock().expect("operation settlement lock is healthy").clone();
308    for settled in pending.into_values() {
309        settled.settle().await.map_err(|source| Error::OperationSettlement { source })?;
310    }
311    Ok(())
312}
313
314/* === Tests === */
harmos_runtime/operations/manager/mod.rs

harmos_runtime/operations/manager/
mod.rs
