harmos_runtime/operations/
middleware.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
17use std::future::Future;
18use std::pin::Pin;
19use std::sync::{Arc, Mutex, OnceLock};
20
21// Third Party Imports
22
23// Local Imports
24use super::{CallContext, OperationGeneration, OperationInstanceId, OperationOutcome};
25use crate::operations::Definition;
26use crate::{Error, Result};
27
28/* === Constants === */
29
30/* === Macros === */
31
32/* === Type Aliases === */
33
34pub type Enter<'a> = Pin<Box<dyn Future<Output = Result<()>> + Send + 'a>>;
35
36/* === Structs === */
37
38pub struct Invocation {
39    operation: Definition,
40    id: OperationInstanceId,
41    generation: OperationGeneration,
42    context: CallContext,
43    origin: super::caller::Origin,
44}
45
46#[derive(Clone, Copy, Debug)]
47pub struct Observation {
48    pub outcome: OperationOutcome,
49    pub settled: bool,
50}
51
52#[derive(Default)]
53pub(super) struct Configuration {
54    state: Mutex<State>,
55}
56
57#[derive(Default)]
58struct State {
59    frozen: bool,
60    entries: Vec<Arc<dyn Middleware>>,
61    authorizer: Option<Arc<dyn super::authorization::Authorizer>>,
62    authorization_frozen: bool,
63}
64
65pub(super) struct Session {
66    call: Invocation,
67    entries: Vec<Arc<dyn Middleware>>,
68    entered: usize,
69    observed: bool,
70    context: Arc<OnceLock<CallContext>>,
71}
72
73/* === Enums === */
74
75/* === Traits === */
76
77pub trait Middleware: Send + Sync + 'static {
78    fn enter<'a>(&'a self, call: &'a mut Invocation) -> Enter<'a>;
79
80    fn observe(&self, _call: &Invocation, _observation: Observation) {}
81}
82
83/* === Implementations === */
84
85impl Invocation {
86    pub fn is_host_call(&self) -> bool {
87        self.origin.identity == super::caller::Identity::Host
88    }
89
90    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
91    pub fn caller_scope(&self) -> Option<&crate::plugin::PluginScope> {
92        self.origin.scope.as_ref()
93    }
94
95    pub fn operation(&self) -> &Definition {
96        &self.operation
97    }
98
99    pub fn id(&self) -> OperationInstanceId {
100        self.id
101    }
102
103    pub fn generation(&self) -> OperationGeneration {
104        self.generation
105    }
106
107    pub fn context(&self) -> &CallContext {
108        &self.context
109    }
110
111    pub fn context_mut(&mut self) -> crate::operations::ContextMut<'_> {
112        super::ContextMut::new(&mut self.context)
113    }
114}
115
116impl Configuration {
117    pub(super) fn register(&self, middleware: Arc<dyn Middleware>) -> Result<()> {
118        let mut state = self.state.lock().expect("middleware configuration lock is healthy");
119        if state.frozen {
120            return Err(Error::Closing { domain: "operation middleware configuration" });
121        }
122        state.entries.push(middleware);
123        Ok(())
124    }
125
126    pub(super) fn register_authorizer(&self, authorizer: Arc<dyn super::authorization::Authorizer>) -> Result<()> {
127        let mut state = self.state.lock().expect("middleware configuration lock is healthy");
128        if state.authorization_frozen {
129            return Err(Error::Closing { domain: "operation authorization configuration" });
130        }
131        if state.authorizer.is_some() {
132            return Err(Error::InvalidInput { field: "operation authorizer", reason: "authorization already has an owner" });
133        }
134        state.authorizer = Some(authorizer);
135        Ok(())
136    }
137
138    #[cfg(feature = "grpc")]
139    pub(super) fn authorizer(&self) -> Option<Arc<dyn super::authorization::Authorizer>> {
140        let mut state = self.state.lock().expect("middleware configuration lock is healthy");
141        state.authorization_frozen = true;
142        state.authorizer.clone()
143    }
144
145    pub(super) fn capture(&self) -> Vec<Arc<dyn Middleware>> {
146        let mut state = self.state.lock().expect("middleware configuration lock is healthy");
147        state.frozen = true;
148        state.entries.clone()
149    }
150}
151
152impl Session {
153    pub(super) fn new(
154        entries: Vec<Arc<dyn Middleware>>, operation: Definition, id: OperationInstanceId, generation: OperationGeneration,
155        inherited: CallContext, context: Arc<OnceLock<CallContext>>, origin: super::caller::Origin,
156    ) -> Self {
157        Self { call: Invocation { operation, id, generation, context: inherited, origin }, entries, entered: 0, observed: false, context }
158    }
159
160    pub(super) async fn enter(&mut self) -> Result<()> {
161        for entry in &self.entries {
162            self.entered += 1;
163            let result = guarded(Box::pin(async { entry.enter(&mut self.call).await })).await;
164            if let Err(error) = result {
165                self.freeze();
166                return Err(error);
167            }
168        }
169        self.freeze();
170        Ok(())
171    }
172
173    fn freeze(&self) {
174        let _ = self.context.set(self.call.context.clone());
175    }
176
177    pub(super) fn observe(&mut self, observation: Observation) {
178        if self.observed {
179            return;
180        }
181        self.observed = true;
182        for entry in self.entries[..self.entered].iter().rev() {
183            if std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
184                entry.observe(&self.call, observation);
185            }))
186            .is_err()
187            {
188                tracing::error!(operation = %self.call.operation.id, "operation middleware observation panicked");
189            }
190        }
191    }
192}
193
194impl Drop for Session {
195    fn drop(&mut self) {
196        self.observe(Observation { outcome: OperationOutcome::Unavailable, settled: false });
197    }
198}
199
200/* === Public Functions === */
201
202/* === Private Functions === */
203
204async fn guarded(mut future: Enter<'_>) -> Result<()> {
205    std::future::poll_fn(|context| match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| future.as_mut().poll(context))) {
206        Ok(result) => result,
207        Err(_) => std::task::Poll::Ready(Err(Error::InvalidInput { field: "operation middleware", reason: "middleware entry panicked" })),
208    })
209    .await
210}
211
212/* === Tests === */
harmos_runtime/operations/middleware.rs

harmos_runtime/operations/
middleware.rs
