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    owner: Identifier,
128    definition: OperationDefinition,
129    pub(super) contract: Option<crate::operations::Contract>,
130
131    #[cfg(feature = "grpc")]
132    pub(super) grpc: Option<crate::grpc::routing::Handler>,
133
134    pub(super) generation: OperationGeneration,
135    lane: &'static str,
136
137    // A resident control runs until cancelled, even while its plugin retires.
138    #[cfg(feature = "plugin-sidecar")]
139    resident: bool,
140
141    active: AtomicBool,
142    publication: Arc<AtomicBool>,
143    retirement: AtomicU8,
144    pending: Mutex<BTreeMap<OperationInstanceId, Arc<Instance>>>,
145    pub(in crate::operations) binding: Arc<dyn Any + Send + Sync>,
146
147    #[cfg(feature = "plugin-sidecar")]
148    pub(super) native_binding: Option<Arc<dyn Any + Send + Sync>>,
149
150    pub(super) wire: Option<super::wire::Dispatch>,
151
152    #[cfg(feature = "plugin-sidecar")]
153    pub(super) native: Option<Arc<super::native::Binding>>,
154
155    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
156    plugin: Option<PluginScope>,
157
158    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
159    pub(super) sdk: Option<harmos_operations::lifecycle::Action>,
160}
161
162pub(super) struct Submission {
163    pub(super) submitted: tokio::time::Instant,
164    pub(super) entry: Arc<Entry>,
165    pub(super) cause: Option<Position>,
166    pub(super) preparation: Option<OwnedSemaphorePermit>,
167}
168
169// The caller's grant authority for one admitted request. Typed bindings check
170// the request grant at admission; encoded bindings learn it from the provider
171// while preparing and check it before any attempt runs.
172#[derive(Clone, Debug)]
173pub(super) struct GrantAuthority {
174    owner: String,
175    operation: String,
176
177    // A cross-owner plugin caller and its declared requirement for this operation.
178    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
179    plugin: Option<(String, Option<harmos_domain::inventory::contracts::Required>)>,
180
181    // External Authority Never Replaces A Plugin Caller's Own Requirement.
182    external: Option<crate::operations::authorization::Authorization>,
183}
184
185struct Cached {
186    target: Weak<Entry>,
187    key: CacheKey,
188    identity: Identity,
189    response: CachedResponse,
190}
191
192/* === Enums === */
193
194#[repr(u8)]
195enum Retirement {
196    Retained,
197    Abandoned,
198    Explicit,
199    Runtime,
200}
201
202/* === Traits === */
203
204/* === Implementations === */
205
206#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
207impl std::fmt::Debug for WeakOperationManager {
208    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
209        formatter.debug_struct("WeakOperationManager").finish_non_exhaustive()
210    }
211}
212
213#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
214impl WeakOperationManager {
215    pub(crate) fn upgrade(&self) -> Option<Manager> {
216        Some(Manager {
217            schema_settings: self.schema_settings.upgrade()?,
218            registry: self.registry.upgrade()?,
219            instances: self.instances.upgrade()?,
220            lanes: self.lanes.upgrade()?,
221            preparation: match &self.preparation {
222                Some(budget) => Some(budget.upgrade()?),
223                None => None,
224            },
225            context: self.context.upgrade()?,
226            resident: self.resident.upgrade()?,
227        })
228    }
229}
230
231#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
232impl Manager {
233    pub(crate) fn downgrade(&self) -> WeakOperationManager {
234        WeakOperationManager {
235            schema_settings: Arc::downgrade(&self.schema_settings),
236            registry: Arc::downgrade(&self.registry),
237            instances: Arc::downgrade(&self.instances),
238            lanes: Arc::downgrade(&self.lanes),
239            preparation: self.preparation.as_ref().map(Arc::downgrade),
240            context: Arc::downgrade(&self.context),
241            resident: self.resident.downgrade(),
242        }
243    }
244}
245
246impl<L: Lane> OperationManager<L> {
247    pub(crate) fn new(inner: Manager) -> Self {
248        Self { inner, lane: std::marker::PhantomData }
249    }
250
251    pub(crate) fn execution(&self) -> Manager {
252        self.inner.clone()
253    }
254
255    pub async fn register(&self, registration: crate::operations::Registration<L>) -> Result<()> {
256        self.inner.register_native(registration)
257    }
258
259    pub async fn unregister(&self, id: &harmos_domain::QualifiedId) -> Result<()> {
260        self.inner.unregister(id).await
261    }
262
263    pub fn with_context(&self, context: super::CallContext) -> Self {
264        Self::new(self.inner.with_context(context))
265    }
266}
267
268impl<L: Lane> std::ops::Deref for OperationManager<L> {
269    type Target = Manager;
270    fn deref(&self) -> &Self::Target {
271        &self.inner
272    }
273}
274
275/* === Public Functions === */
276
277/* === Private Functions === */
278
279#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
280fn grant_allowed(required: Option<&harmos_domain::inventory::contracts::Required>, grant: Option<&harmos_domain::plugin::Grant>) -> bool {
281    match grant {
282        None => true,
283        Some(grant) => required.is_some_and(|required| required.grants(grant)),
284    }
285}
286
287async fn settle(entry: &Entry) -> Result<()> {
288    let pending = entry.pending.lock().expect("operation settlement lock is healthy").clone();
289    for settled in pending.into_values() {
290        settled.settle().await.map_err(|source| Error::OperationSettlement { source })?;
291    }
292    Ok(())
293}
294
295/* === Tests === */
harmos_runtime/operations/manager/mod.rs

harmos_runtime/operations/manager/
mod.rs
