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
18// Standard Library Imports
19use std::any::Any;
20use std::sync::{Arc, Mutex};
21
22// Third Party Imports
23
24// Local Imports
25use super::HostContext;
26use harmos_domain::{MemorySize, Message, MessageError, OperationDeclaration};
27use harmos_sdk::__private::ErasedResponse;
28use harmos_sdk::operations::{CallFuture, Caller, InvocationError, Target};
29
30/* === Constants === */
31
32/* === Macros === */
33
34macro_rules! native_payload {
35    ($ty:ty) => {
36        impl Message for $ty {
37            const CODEC: &'static str = "harmos.native";
38            fn encode(&self) -> Vec<u8> {
39                panic!("native payload cannot be encoded")
40            }
41            fn encoded_size(&self) -> usize {
42                0
43            }
44            fn decode(_: &[u8]) -> Result<Self, MessageError> {
45                Err(MessageError::borrowed("native payload cannot be decoded"))
46            }
47            fn decoding_memory(_: &[u8]) -> Result<usize, MessageError> {
48                Err(MessageError::borrowed("native payload cannot be decoded"))
49            }
50        }
51        impl harmos_domain::schema::Described for $ty {
52            const NAME: &'static str = "harmos::native";
53            const FIELDS: &'static [harmos_domain::schema::declaration::Field] = &[];
54            const MARKER: &'static [u8] = &[];
55        }
56    };
57}
58
59native_payload!(Request);
60
61native_payload!(Answer);
62
63/* === Type Aliases === */
64
65/* === Structs === */
66
67struct Request {
68    payload: Mutex<Option<Box<dyn Any + Send>>>,
69    memory: usize,
70}
71
72struct Answer(ErasedResponse);
73
74struct Invocation;
75
76/* === Enums === */
77
78/* === Traits === */
79
80/* === Implementations === */
81
82impl Caller for HostContext {
83    fn call(&self, target: Target, request: Box<dyn Any + Send>) -> CallFuture<'_> {
84        Box::pin(async move {
85            let result = async {
86                let context = self.borrow();
87                let origin = context.source.check()?;
88                let entry = context.operations.entry_as(target.id, &origin)?;
89                if entry.declared().fingerprint != (target.definition)().fingerprint {
90                    return Err(crate::Error::InvalidInput {
91                        field: "declared operation",
92                        reason: "callee fingerprint differs",
93                    });
94                }
95                let Some(factory) = entry.binding.downcast_ref::<harmos_sdk::operations::OperationFactory>().cloned() else {
96                    let payload = (target.encode)(request.as_ref()).ok_or(crate::Error::InvalidInput {
97                        field: "declared operation",
98                        reason: "callee payload differs",
99                    })?;
100                    let bytes = context.call_encoded(target.id, payload).await?;
101                    return (target.decode)(&bytes).map_err(|source| crate::Error::OperationFailed {
102                        operation: target.id.to_owned(),
103                        source: Arc::new(source),
104                    });
105                };
106                if !factory.accepts_target(&target) {
107                    return Err(crate::Error::InvalidInput {
108                        field: "declared operation",
109                        reason: "callee payload types differ",
110                    });
111                }
112                let grant = (target.grant)(request.as_ref());
113                let key = (target.key)(request.as_ref());
114                let memory = (target.memory)(request.as_ref());
115                let resources = context.operations.resources_for(&entry);
116                let binding = super::authored::factory_projected::<Invocation, ()>(
117                    &context.operations,
118                    move |request, ()| {
119                        let payload = request
120                            .payload
121                            .into_inner()
122                            .map_err(|_| harmos_sdk::ServingError::Catalog)?
123                            .ok_or(harmos_sdk::ServingError::Catalog)?;
124                        factory.prepare_native(resources.clone(), payload, Box::new(()))
125                    },
126                    |response| Some(harmos_sdk::Response::new(Answer(response))),
127                    |response| response.0.clone(),
128                );
129                let mut input = super::binding::Request::from(Request {
130                    payload: Mutex::new(Some(request)),
131                    memory,
132                });
133                input.binding = Some(Arc::new(binding));
134                input.grant = grant;
135                input.key = key;
136                let running = context.operations.admit_at::<Invocation, ()>(
137                    target.id,
138                    (input, ()),
139                    Some(&context.scope),
140                    context.admission.held(),
141                    &context.source,
142                    None,
143                )?;
144                let response = running.await?;
145                Ok(response.0.clone())
146            }
147            .await;
148            result.map_err(|error: crate::Error| InvocationError::failed(target.id, error))
149        })
150    }
151}
152
153impl OperationDeclaration for Invocation {
154    type Request = Request;
155    type Response = Answer;
156    type Input = harmos_domain::operations::NoStream;
157    type Output = harmos_domain::operations::NoStream;
158    const ID: &'static str = "native.invocation";
159    const POLICY: harmos_domain::OperationPolicy = harmos_operations::host::POLICY;
160}
161
162impl MemorySize for Request {
163    fn memory_size(&self) -> usize {
164        self.memory
165    }
166}
167
168impl MemorySize for Answer {
169    fn memory_size(&self) -> usize {
170        std::mem::size_of::<Self>()
171    }
172}
173
174/* === Public Functions === */
175
176/* === Private Functions === */
177
178/* === Tests === */
179
180#[cfg(test)]
181mod tests {
182    use super::*;
183    use crate::operations::{Lane, RegistrationExt};
184    use harmos_sdk::{Context, Operation, Response};
185
186    #[crate::lanes]
187    enum Lanes {
188        #[default]
189        Open,
190    }
191    struct Model;
192    impl crate::Lifecycle for Model {
193        type Config = ();
194        type Error = std::convert::Infallible;
195        type Context = crate::Startup<Self>;
196
197        async fn prepare(_: (), _: &mut crate::Setup<Self>) -> std::result::Result<Self, Self::Error> {
198            Ok(Self)
199        }
200
201        async fn initialize(&mut self, _: &Self::Context) -> std::result::Result<(), Self::Error> {
202            Ok(())
203        }
204
205        async fn stop(&mut self, _: &Self::Context) -> std::result::Result<(), Self::Error> {
206            Ok(())
207        }
208
209        async fn cleanup(&mut self) -> std::result::Result<(), Self::Error> {
210            Ok(())
211        }
212    }
213
214    impl crate::Application for Model {
215        type Lanes = Lanes;
216        type State = ();
217        type Transaction = crate::state::Never;
218        type Record = ();
219        type Streams = ();
220    }
221
222    const POLICY: crate::OperationPolicy = crate::OperationPolicy {
223        timeout: std::time::Duration::from_secs(2),
224        attempt_timeout: std::time::Duration::from_secs(2),
225        retries: crate::RetryPolicy::Never,
226    };
227
228    #[crate::operation(harmos_path = crate, resources = [u64], id = "test.leaf", response = u64, policy = POLICY,
229        key = |request: &Leaf| crate::operations::CacheKey::new(request.value.to_le_bytes()).ok())]
230    struct Leaf {
231        value: u64,
232    }
233    impl Operation for Leaf {
234        type Error = std::convert::Infallible;
235        async fn run(self, ctx: Context<'_, Self>) -> Result<Response<u64>, Self::Error> {
236            Ok((self.value + *ctx.resource::<u64>()).into())
237        }
238    }
239
240    #[crate::operation(harmos_path = crate, id = "test.parent", response = u64, policy = POLICY, operations = [Leaf])]
241    struct Parent {
242        value: u64,
243    }
244    impl Operation for Parent {
245        type Error = InvocationError;
246        async fn run(self, ctx: Context<'_, Self>) -> Result<Response<u64>, Self::Error> {
247            ctx.operation(Leaf { value: self.value }).await
248        }
249    }
250
251    #[tokio::test]
252    async fn declared_calls_verify_activation_and_run_with_resources() {
253        let runtime = crate::Runtime::<Model>::launch((), ()).await.unwrap();
254        let missing = harmos_sdk::__private::RegistrationBatch::new().operation::<Parent>();
255        assert!(runtime.register_batch(missing.lane(<Lanes as crate::operations::Lane>::default_lane())).await.is_err());
256        let missing = harmos_sdk::__private::RegistrationBatch::new().operation::<Leaf>();
257        assert!(runtime.register_batch(missing.lane(<Lanes as crate::operations::Lane>::default_lane())).await.is_err());
258        runtime.shutdown().await.unwrap();
259        let runtime = crate::Runtime::<Model>::launch_with((), (), |setup| {
260            setup.resource(7_u64);
261        })
262        .await
263        .unwrap();
264        let batch = harmos_sdk::__private::RegistrationBatch::new().operation::<Parent>().operation::<Leaf>();
265        runtime.register_batch(batch.lane(<Lanes as crate::operations::Lane>::default_lane())).await.unwrap();
266        runtime.operations.set_cache_capacity(2);
267        let direct = runtime.operation(Leaf { value: 5 }).run().await.unwrap();
268        let child = runtime.operation(Parent { value: 5 }).run().await.unwrap();
269        assert_eq!(*child, 12);
270        assert!(direct.ptr_eq(&child));
271        runtime.shutdown().await.unwrap();
272    }
273
274    #[tokio::test]
275    async fn typed_calls_keep_carrier_and_runtime_resources() {
276        struct Carrier(u32);
277
278        impl harmos_sdk::operations::Carrier for Carrier {
279            type Declaration = Leaf;
280            type Error = harmos_sdk::ResourceError;
281            type StreamEndpoints = ();
282
283            async fn bind(context: harmos_sdk::ExecutionContext) -> Result<Self, Self::Error> {
284                let handle = context.resource::<u32>()?;
285                Ok(Self(*handle))
286            }
287
288            async fn attempt(
289                &mut self, context: harmos_sdk::ExecutionContext, request: &mut Leaf, _: &mut (),
290            ) -> Result<Response<u64>, Self::Error> {
291                let resource = context.resource::<u64>()?;
292                let response = request.value + u64::from(self.0) + *resource;
293                Ok(response.into())
294            }
295        }
296
297        let runtime = crate::Runtime::<Model>::launch_with((), (), |setup| {
298            setup.resource(7_u64);
299        })
300        .await
301        .unwrap();
302
303        let mut batch = harmos_sdk::__private::RegistrationBatch::new().carrier::<Leaf, Carrier>();
304        let factory = batch.operations.pop().unwrap();
305        let binding = Arc::new(factory);
306        let lane = <Lanes as Lane>::default_lane().name();
307        let definition = Leaf::definition();
308        let mut entry = runtime
309            .operations
310            .binding_erased(
311                "host",
312                lane,
313                binding,
314                definition,
315                None,
316                #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
317                None,
318            )
319            .unwrap();
320
321        let handle = Arc::new(11_u32);
322        let staged = Arc::get_mut(&mut entry).unwrap();
323        staged.carrier.insert(handle).unwrap();
324        runtime.operations.reserve(&[entry]).unwrap();
325
326        let response = runtime.operation(Leaf { value: 5 }).run().await.unwrap();
327        assert_eq!(*response, 23);
328        let resources = runtime.operations.resources();
329        assert!(resources.resource::<u32>().is_err());
330        let owner = resources.resource::<u64>().unwrap();
331        assert_eq!(*owner, 7);
332        runtime.shutdown().await.unwrap();
333    }
334}
harmos_runtime/operations/invocation.rs

harmos_runtime/operations/
invocation.rs
