harmos_runtime/operations/
invocation.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// Native Declared Calls Preserve Typed Payload Owners Through Erased Admission.
11
12/* === Modules === */
13
14/* === Re-Exports === */
15
16/* === Imports === */
17
18use std::any::Any;
19use std::sync::{Arc, Mutex};
20
21// Third Party Imports
22
23use super::HostContext;
24use harmos_domain::{MemorySize, Message, MessageError, OperationDeclaration};
25use harmos_sdk::__private::ErasedResponse;
26use harmos_sdk::operations::{CallFuture, Caller, InvocationError, Target};
27
28/* === Constants === */
29
30/* === Macros === */
31
32macro_rules! native_payload {
33    ($ty:ty) => {
34        impl Message for $ty {
35            const CODEC: &'static str = "harmos.native";
36            fn encode(&self) -> Vec<u8> {
37                panic!("native payload cannot be encoded")
38            }
39            fn encoded_size(&self) -> usize {
40                0
41            }
42            fn decode(_: &[u8]) -> Result<Self, MessageError> {
43                Err(MessageError::borrowed("native payload cannot be decoded"))
44            }
45            fn decoding_memory(_: &[u8]) -> Result<usize, MessageError> {
46                Err(MessageError::borrowed("native payload cannot be decoded"))
47            }
48        }
49        impl harmos_domain::schema::Described for $ty {
50            const NAME: &'static str = "harmos::native";
51            const FIELDS: &'static [harmos_domain::schema::declaration::Field] = &[];
52            const MARKER: &'static [u8] = &[];
53        }
54    };
55}
56native_payload!(Request);
57native_payload!(Answer);
58
59/* === Type Aliases === */
60
61/* === Structs === */
62
63struct Request {
64    payload: Mutex<Option<Box<dyn Any + Send>>>,
65    memory: usize,
66}
67
68struct Answer(ErasedResponse);
69struct Invocation;
70
71/* === Enums === */
72
73/* === Traits === */
74
75/* === Implementations === */
76
77impl Caller for HostContext {
78    fn call(&self, target: Target, request: Box<dyn Any + Send>) -> CallFuture<'_> {
79        Box::pin(async move {
80            let result = async {
81                let context = self.borrow();
82                let origin = context.source.check()?;
83                let entry = context.operations.entry_as(target.id, &origin)?;
84                if entry.declared().fingerprint != (target.definition)().fingerprint {
85                    return Err(crate::Error::InvalidInput {
86                        field: "declared operation",
87                        reason: "callee fingerprint differs",
88                    });
89                }
90                let Some(factory) = entry.binding.downcast_ref::<harmos_sdk::operations::OperationFactory>().cloned() else {
91                    let payload = (target.encode)(request.as_ref()).ok_or(crate::Error::InvalidInput {
92                        field: "declared operation",
93                        reason: "callee payload differs",
94                    })?;
95                    let bytes = context.call_encoded(target.id, payload).await?;
96                    return (target.decode)(&bytes).map_err(|source| crate::Error::OperationFailed {
97                        operation: target.id.to_owned(),
98                        source: Arc::new(source),
99                    });
100                };
101                if !factory.accepts_target(&target) {
102                    return Err(crate::Error::InvalidInput {
103                        field: "declared operation",
104                        reason: "callee payload types differ",
105                    });
106                }
107                let grant = (target.grant)(request.as_ref());
108                let key = (target.key)(request.as_ref());
109                let memory = (target.memory)(request.as_ref());
110                let binding = super::authored::factory_projected::<Invocation, ()>(
111                    &context.operations,
112                    move |request, ()| {
113                        let payload = request
114                            .payload
115                            .into_inner()
116                            .map_err(|_| harmos_sdk::ServingError::Catalog)?
117                            .ok_or(harmos_sdk::ServingError::Catalog)?;
118                        factory.prepare_native(payload, Box::new(()))
119                    },
120                    |response| Some(harmos_sdk::Response::new(Answer(response))),
121                    |response| response.0.clone(),
122                );
123                let mut input = super::binding::Request::from(Request {
124                    payload: Mutex::new(Some(request)),
125                    memory,
126                });
127                input.binding = Some(Arc::new(binding));
128                input.grant = grant;
129                input.key = key;
130                let running = context.operations.admit_at::<Invocation, ()>(
131                    target.id,
132                    (input, ()),
133                    Some(&context.scope),
134                    context.admission.held(),
135                    &context.source,
136                    None,
137                )?;
138                let response = running.await?;
139                Ok(response.0.clone())
140            }
141            .await;
142            result.map_err(|error: crate::Error| InvocationError::failed(target.id, error))
143        })
144    }
145}
146
147impl OperationDeclaration for Invocation {
148    type Request = Request;
149    type Response = Answer;
150    type Input = harmos_domain::operations::NoStream;
151    type Output = harmos_domain::operations::NoStream;
152    const ID: &'static str = "native.invocation";
153    const POLICY: harmos_domain::OperationPolicy = harmos_operations::host::POLICY;
154}
155
156impl MemorySize for Request {
157    fn memory_size(&self) -> usize {
158        self.memory
159    }
160}
161impl MemorySize for Answer {
162    fn memory_size(&self) -> usize {
163        std::mem::size_of::<Self>()
164    }
165}
166
167/* === Public Functions === */
168
169/* === Private Functions === */
170
171/* === Tests === */
172
173#[cfg(test)]
174mod tests {
175    use super::*;
176    use crate::RegistrationExt;
177    use harmos_sdk::{Context, Operation, Response};
178
179    #[crate::lanes]
180    enum Lanes {
181        #[default]
182        Open,
183    }
184    struct Model;
185    impl crate::Application for Model {
186        type Lanes = Lanes;
187        type State = ();
188        type Transaction = crate::state::Never;
189        type Record = ();
190        type Streams = ();
191        type Resources = ();
192    }
193
194    const POLICY: crate::OperationPolicy = crate::OperationPolicy {
195        timeout: std::time::Duration::from_secs(2),
196        attempt_timeout: std::time::Duration::from_secs(2),
197        retries: crate::RetryPolicy::Never,
198    };
199
200    #[crate::operation(harmos_path = crate, resources = [u64], id = "test.leaf", response = u64, policy = POLICY,
201        key = |request: &Leaf| crate::operations::CacheKey::new(request.value.to_le_bytes()).ok())]
202    struct Leaf {
203        value: u64,
204    }
205    impl Operation for Leaf {
206        type Error = std::convert::Infallible;
207        async fn run(self, ctx: Context<'_, Self>) -> Result<Response<u64>, Self::Error> {
208            Ok((self.value + *ctx.resource::<u64>()).into())
209        }
210    }
211
212    #[crate::operation(harmos_path = crate, id = "test.parent", response = u64, policy = POLICY, operations = [Leaf])]
213    struct Parent {
214        value: u64,
215    }
216    impl Operation for Parent {
217        type Error = InvocationError;
218        async fn run(self, ctx: Context<'_, Self>) -> Result<Response<u64>, Self::Error> {
219            ctx.operation(Leaf { value: self.value }).await
220        }
221    }
222
223    #[tokio::test]
224    async fn declared_calls_verify_activation_and_run_with_resources() {
225        let runtime = crate::Runtime::<Model>::builder((), ()).register(Lanes::catalog()).runtime().await.unwrap();
226        let missing = harmos_sdk::RegistrationBatch::new(harmos_sdk::Dependencies::new()).operation::<Parent>();
227        assert!(runtime.register(missing.defaults()).await.is_err());
228        let missing = harmos_sdk::RegistrationBatch::new(harmos_sdk::Dependencies::new()).operation::<Leaf>();
229        assert!(runtime.register(missing.defaults()).await.is_err());
230        let mut dependencies = harmos_sdk::Dependencies::new();
231        dependencies.insert(Arc::new(7_u64)).unwrap();
232        let batch = harmos_sdk::RegistrationBatch::new(dependencies).operation::<Parent>().operation::<Leaf>();
233        runtime.register(batch.defaults()).await.unwrap();
234        runtime.operations.set_cache_capacity(2);
235        let direct = runtime.operation(Leaf { value: 5 }).run().await.unwrap();
236        let child = runtime.operation(Parent { value: 5 }).run().await.unwrap();
237        assert_eq!(*child, 12);
238        assert!(direct.ptr_eq(&child));
239        runtime.stop().await.unwrap();
240    }
241}
harmos_runtime/operations/invocation.rs

harmos_runtime/operations/
invocation.rs
