harmos_runtime/operations/
surface.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// In-process binding of the portable host: a thin layer over the
11// manager's own submission, so in-process and remote bodies share one shape.
12
13/* === Modules === */
14
15/* === Re-Exports === */
16
17/* === Imports === */
18
19// Standard Library Imports
20use std::future::Future;
21
22// Third Party Imports
23
24// Local Imports
25use super::caller::Source;
26use super::{CallContext, Context, ContextError, ContextValue, Manager, OperationDeclaration, Response, Running};
27use super::{EndpointSet, Lane, RetryPolicy};
28use crate::{Error, Result};
29
30/* === Constants === */
31
32/* === Macros === */
33
34/* === Type Aliases === */
35
36/* === Structs === */
37
38#[must_use = "an invocation runs only when start or run is awaited"]
39pub struct Invocation<O: OperationDeclaration, L: Lane, E: EndpointSet = ()> {
40    target: Target,
41    request: O::Request,
42    retry: Option<RetryPolicy>,
43    lane: Option<L>,
44    endpoints: E,
45}
46
47/* === Enums === */
48
49enum Target {
50    Host(Manager),
51
52    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
53    Plugin(crate::plugin::PluginScope),
54}
55
56/* === Traits === */
57
58/* === Implementations === */
59
60impl<O: OperationDeclaration, L: Lane> Invocation<O, L> {
61    pub(crate) fn new(operations: Manager, request: O::Request) -> Self {
62        Self {
63            target: Target::Host(operations),
64            request,
65            retry: None,
66            lane: None,
67            endpoints: (),
68        }
69    }
70}
71
72impl<O: OperationDeclaration, L: Lane, E: EndpointSet> Invocation<O, L, E> {
73    pub fn retry(mut self, policy: RetryPolicy) -> Self {
74        self.retry = Some(policy);
75        self
76    }
77
78    pub fn lane(mut self, lane: L) -> Self {
79        self.lane = Some(lane);
80        self
81    }
82
83    pub fn endpoints<F: EndpointSet>(self, endpoints: F) -> Invocation<O, L, F> {
84        Invocation {
85            target: self.target,
86            request: self.request,
87            retry: self.retry,
88            lane: self.lane,
89            endpoints,
90        }
91    }
92
93    pub async fn start(self) -> Result<Running<O>> {
94        if E::item_types() != super::item_types::<O>() {
95            return Err(Error::InvalidInput {
96                field: "operation endpoints",
97                reason: "endpoint items differ from the declaration",
98            });
99        }
100        let mut request = super::binding::Request::from(self.request);
101        request.retry = self.retry;
102        request.lane = self.lane.map(|lane| lane.name());
103        let (operations, source) = match self.target {
104            Target::Host(operations) => (operations, Source::Host),
105            #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
106            Target::Plugin(scope) => (scope.operation_manager()?, Source::Scope(scope)),
107        };
108        operations.admit::<O, E>((request, self.endpoints), None, &[], &source, None)
109    }
110
111    pub async fn run(self) -> Result<Response<O::Response>> {
112        self.start().await?.join().await
113    }
114}
115
116#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
117impl crate::plugin::PluginScope {
118    pub fn operation<O: OperationDeclaration<Request = O>, L: Lane>(&self, request: O) -> Invocation<O, L> {
119        Invocation {
120            target: Target::Plugin(self.clone()),
121            request,
122            retry: None,
123            lane: None,
124            endpoints: (),
125        }
126    }
127
128    pub async fn register<C: harmos_sdk::operations::OperationCatalog, L: Lane>(&self, lane: L) -> Result<()> {
129        use super::RegistrationExt;
130        self.register_batch(C::batch().lane(lane)).await
131    }
132
133    pub(crate) async fn register_batch<L: Lane>(&self, mut registration: super::Registration<L>) -> Result<()> {
134        if !self.is_ready() {
135            return Err(Error::Closing { domain: "plugin scope" });
136        }
137        let operations = self.operation_manager()?;
138        let services = std::mem::take(&mut registration.batch.services);
139        if services.is_empty() {
140            let entries = operations.native_entries(registration, self.owner(), Some(self))?;
141            return operations.reserve(&entries);
142        }
143        let mut batch = harmos_sdk::__private::RegistrationBatch::new();
144        batch.services = services;
145        let manager = crate::ServiceManager::scoped(operations.clone(), self.clone());
146        let (staged, ids) = manager.stage(batch)?;
147        operations.register_scoped_services(registration, staged, self.owner(), Some(self))?;
148        manager.activate(&ids).await
149    }
150
151    pub async fn unregister(&self, id: &harmos_domain::QualifiedId) -> Result<()> {
152        let (owner, _) = id.as_str().split_once('.').expect("validated qualified identity");
153        if owner != self.owner() {
154            return Err(Error::PermissionDenied);
155        }
156        let operations = self.operation_manager()?;
157        if matches!(operations.resident.definition(id.as_str()), Some(super::resident::Definition::Service(_))) {
158            return crate::ServiceManager::scoped(operations, self.clone()).unregister(id).await;
159        }
160        operations.unregister(id).await
161    }
162}
163
164impl<A: crate::Application> crate::Runtime<A> {
165    pub fn services(&self) -> crate::ServiceManager {
166        crate::ServiceManager::new(self.operations.execution())
167    }
168
169    pub async fn register<C: harmos_sdk::operations::OperationCatalog>(&self, lane: A::Lanes) -> Result<()> {
170        use super::RegistrationExt;
171        self.register_batch(C::batch().lane(lane)).await
172    }
173
174    pub(crate) async fn register_batch(&self, mut registration: super::Registration<A::Lanes>) -> Result<()> {
175        let services = std::mem::take(&mut registration.batch.services);
176        if services.is_empty() {
177            return self.operations.register_batch(registration).await;
178        }
179        let mut batch = harmos_sdk::__private::RegistrationBatch::new();
180        batch.services = services;
181        let manager = self.services();
182        let (staged, ids) = manager.stage(batch)?;
183        self.operations.execution().register_with_services(registration, staged)?;
184        manager.activate(&ids).await
185    }
186
187    pub async fn unregister(&self, id: &harmos_domain::QualifiedId) -> Result<()> {
188        if !id.as_str().starts_with("host.") {
189            return Err(Error::PermissionDenied);
190        }
191        if matches!(self.operations.resident.definition(id.as_str()), Some(super::resident::Definition::Service(_))) {
192            return self.services().unregister(id).await;
193        }
194        self.operations.unregister(id).await
195    }
196
197    pub fn operation<O: OperationDeclaration<Request = O>>(&self, request: O) -> Invocation<O, A::Lanes> {
198        Invocation::new(self.operations.execution(), request)
199    }
200}
201
202impl<A: crate::Application> crate::runtime::Handle<A> {
203    pub fn services(&self) -> crate::ServiceManager {
204        crate::ServiceManager::new(self.operations.execution())
205    }
206
207    pub async fn register<C: harmos_sdk::operations::OperationCatalog>(&self, lane: A::Lanes) -> Result<()> {
208        use super::RegistrationExt;
209        self.register_batch(C::batch().lane(lane)).await
210    }
211
212    pub(crate) async fn register_batch(&self, mut registration: super::Registration<A::Lanes>) -> Result<()> {
213        let services = std::mem::take(&mut registration.batch.services);
214        if services.is_empty() {
215            return self.operations.register_batch(registration).await;
216        }
217        let mut batch = harmos_sdk::__private::RegistrationBatch::new();
218        batch.services = services;
219        let manager = self.services();
220        let (staged, ids) = manager.stage(batch)?;
221        self.operations.execution().register_with_services(registration, staged)?;
222        manager.activate(&ids).await
223    }
224
225    pub async fn unregister(&self, id: &harmos_domain::QualifiedId) -> Result<()> {
226        if !id.as_str().starts_with("host.") {
227            return Err(Error::PermissionDenied);
228        }
229        if matches!(self.operations.resident.definition(id.as_str()), Some(super::resident::Definition::Service(_))) {
230            return self.services().unregister(id).await;
231        }
232        self.operations.unregister(id).await
233    }
234
235    pub fn operation<O: OperationDeclaration<Request = O>>(&self, request: O) -> Invocation<O, A::Lanes> {
236        Invocation::new(self.operations.execution(), request)
237    }
238}
239
240impl<L: Lane> harmos_sdk::operations::Host for super::OperationManager<L> {
241    fn publish_on<S: harmos_domain::StreamSpec>(
242        &self, row: &S::Row,
243    ) -> impl std::future::Future<Output = std::result::Result<(), Self::Error>> + Send
244    where
245        S::Row: harmos_domain::Message,
246    {
247        use harmos_domain::Message;
248        let published = self.call::<harmos_operations::streams::Publish>(harmos_operations::streams::Publish {
249            stream: S::ID.to_owned(),
250            schema: S::VERSION,
251            sequence: None,
252            published_unix_micros: None,
253            context: None,
254            payload: row.encode(),
255            following: Vec::new(),
256        });
257        async move {
258            published.await?;
259            Ok(())
260        }
261    }
262
263    type Error = Error;
264    type Instance<O: OperationDeclaration> = Running<O>;
265
266    fn context(&self) -> &CallContext {
267        self.call_context()
268    }
269
270    fn with_value<T: ContextValue>(&self, value: T) -> std::result::Result<Self, ContextError> {
271        let context = self.call_context().with_value(value)?;
272        Ok(Self::new(self.execution().with_context(context)))
273    }
274
275    fn submit<O: OperationDeclaration>(&self, request: O::Request) -> impl Future<Output = Result<Running<O>>> + Send {
276        std::future::ready(Manager::submit::<O>(self, request))
277    }
278
279    fn submit_owned<O: OperationDeclaration>(
280        &self, request: O::Request, claims: super::RequestClaims,
281    ) -> impl Future<Output = Result<Running<O>>> + Send {
282        std::future::ready(Manager::submit_owned::<O>(self, request, claims))
283    }
284
285    fn call<O: OperationDeclaration>(&self, request: O::Request) -> impl Future<Output = Result<Response<O::Response>>> + Send {
286        let submitted = Manager::submit::<O>(self, request);
287        async move { submitted?.await }
288    }
289}
290
291impl harmos_sdk::operations::Host for Manager {
292    fn publish_on<S: harmos_domain::StreamSpec>(
293        &self, row: &S::Row,
294    ) -> impl std::future::Future<Output = std::result::Result<(), Self::Error>> + Send
295    where
296        S::Row: harmos_domain::Message,
297    {
298        use harmos_domain::Message;
299        let published = self.call::<harmos_operations::streams::Publish>(harmos_operations::streams::Publish {
300            stream: S::ID.to_owned(),
301            schema: S::VERSION,
302            sequence: None,
303            published_unix_micros: None,
304            context: None,
305            payload: row.encode(),
306            following: Vec::new(),
307        });
308        async move {
309            published.await?;
310            Ok(())
311        }
312    }
313
314    type Error = Error;
315    type Instance<O: OperationDeclaration> = Running<O>;
316
317    fn context(&self) -> &CallContext {
318        self.call_context()
319    }
320
321    fn with_value<T: ContextValue>(&self, value: T) -> std::result::Result<Self, ContextError> {
322        let context = self.call_context().with_value(value)?;
323        Ok(self.with_context(context))
324    }
325
326    fn submit<O: OperationDeclaration>(&self, request: O::Request) -> impl Future<Output = Result<Running<O>>> + Send {
327        std::future::ready(Manager::submit::<O>(self, request))
328    }
329
330    fn submit_owned<O: OperationDeclaration>(
331        &self, request: O::Request, claims: super::RequestClaims,
332    ) -> impl Future<Output = Result<Running<O>>> + Send {
333        std::future::ready(Manager::submit_owned::<O>(self, request, claims))
334    }
335
336    fn call<O: OperationDeclaration>(&self, request: O::Request) -> impl Future<Output = Result<Response<O::Response>>> + Send {
337        let submitted = Manager::submit::<O>(self, request);
338        async move { submitted?.await }
339    }
340}
341
342impl harmos_sdk::operations::Host for Context<'_> {
343    fn publish_on<S: harmos_domain::StreamSpec>(
344        &self, row: &S::Row,
345    ) -> impl std::future::Future<Output = std::result::Result<(), Self::Error>> + Send
346    where
347        S::Row: harmos_domain::Message,
348    {
349        use harmos_domain::Message;
350        let published = self.call::<harmos_operations::streams::Publish>(harmos_operations::streams::Publish {
351            stream: S::ID.to_owned(),
352            schema: S::VERSION,
353            sequence: None,
354            published_unix_micros: None,
355            context: None,
356            payload: row.encode(),
357            following: Vec::new(),
358        });
359        async move {
360            published.await?;
361            Ok(())
362        }
363    }
364
365    type Error = Error;
366    type Instance<O: OperationDeclaration> = Running<O>;
367
368    fn context(&self) -> &CallContext {
369        Context::context(self)
370    }
371
372    fn with_value<T: ContextValue>(&self, value: T) -> std::result::Result<Self, ContextError> {
373        let context = Context::context(self).with_value(value)?;
374        let mut child = self.clone();
375        child.operations = child.operations.with_context(context);
376        Ok(child)
377    }
378
379    fn submit<O: OperationDeclaration>(&self, request: O::Request) -> impl Future<Output = Result<Running<O>>> + Send {
380        std::future::ready(Context::submit::<O>(self, request))
381    }
382
383    fn submit_owned<O: OperationDeclaration>(
384        &self, request: O::Request, claims: super::RequestClaims,
385    ) -> impl Future<Output = Result<Running<O>>> + Send {
386        std::future::ready(Context::submit_owned::<O>(self, request, claims))
387    }
388
389    fn call<O: OperationDeclaration>(&self, request: O::Request) -> impl Future<Output = Result<Response<O::Response>>> + Send {
390        let submitted = Context::submit::<O>(self, request);
391        async move { submitted?.await }
392    }
393
394    // A resident outlives the attempt that activated it; its lease names the
395    // attempt's caller.
396}
397
398/* === Public Functions === */
399
400/* === Private Functions === */
401
402/* === Tests === */
403
404#[cfg(test)]
405mod tests {
406    use super::*;
407    use crate::Application;
408    use crate::operations::RegistrationExt;
409    use std::sync::{
410        Arc,
411        atomic::{AtomicUsize, Ordering},
412    };
413
414    #[crate::lanes]
415    enum Lanes {
416        #[default]
417        Default,
418    }
419
420    struct Model;
421    impl crate::Lifecycle for Model {
422        type Config = ();
423        type Error = std::convert::Infallible;
424        type Context = crate::Startup<Self>;
425
426        async fn prepare(_: (), _: &mut crate::Setup<Self>) -> std::result::Result<Self, Self::Error> {
427            Ok(Self)
428        }
429
430        async fn initialize(&mut self, _: &Self::Context) -> std::result::Result<(), Self::Error> {
431            Ok(())
432        }
433
434        async fn stop(&mut self, _: &Self::Context) -> std::result::Result<(), Self::Error> {
435            Ok(())
436        }
437
438        async fn cleanup(&mut self) -> std::result::Result<(), Self::Error> {
439            Ok(())
440        }
441    }
442
443    impl Application for Model {
444        type Lanes = Lanes;
445        type State = ();
446        type Transaction = crate::state::Never;
447        type Record = ();
448        type Streams = ();
449    }
450
451    #[crate::operation(harmos_path = crate, resources = [AtomicUsize, crate::operations::HostContext],
452        id = "test.probe", response = u64,
453        policy = crate::OperationPolicy {
454            timeout: std::time::Duration::from_secs(1),
455            attempt_timeout: std::time::Duration::from_secs(1),
456            retries: RetryPolicy::Never,
457        },
458    )]
459    struct Probe {
460        value: u64,
461    }
462
463    impl crate::Operation for Probe {
464        type Error = std::io::Error;
465
466        async fn run(self, ctx: crate::Context<'_, Self>) -> std::io::Result<Response<u64>> {
467            let host = ctx.resource::<crate::operations::HostContext>();
468            let attempt = host.borrow().attempt();
469            if attempt == 0 {
470                ctx.resource::<AtomicUsize>().fetch_add(1, Ordering::SeqCst);
471            }
472            if u64::try_from(attempt).unwrap() < self.value {
473                return Err(std::io::Error::other("retry this request"));
474            }
475            Ok(0.into())
476        }
477
478        fn classify(_: &std::io::Error) -> super::super::FailureKind {
479            super::super::FailureKind::Transient
480        }
481    }
482
483    struct Idle;
484
485    impl harmos_sdk::Service for Idle {
486        type Error = std::io::Error;
487        async fn initialize(_: harmos_sdk::ExecutionContext) -> std::io::Result<Self> {
488            Ok(Self)
489        }
490        async fn run(&mut self, _: harmos_sdk::ExecutionContext) -> std::io::Result<()> {
491            std::future::pending().await
492        }
493    }
494
495    #[tokio::test]
496    async fn mixed_registration_releases_invalid_batch_reservations() {
497        let runtime = crate::Runtime::<Model>::launch_with((), (), |setup| {
498            setup.resource::<AtomicUsize>(Arc::new(AtomicUsize::new(0)));
499        })
500        .await
501        .unwrap();
502        let batch = harmos_sdk::__private::RegistrationBatch::new().operation::<Probe>().service::<Idle>(
503            "idle",
504            crate::ServicePolicy {
505                stop_timeout: std::time::Duration::from_secs(1),
506                restart: crate::RestartPolicy::Never,
507            },
508        );
509        assert!(
510            runtime
511                .register_batch(batch.clone().operation::<Probe>().lane(<Lanes as crate::operations::Lane>::default_lane()))
512                .await
513                .is_err()
514        );
515        assert!(runtime.operations.definition(Probe::ID).is_err());
516        assert!(runtime.operations.resident.definition("host.idle").is_none());
517        runtime.register_batch(batch.lane(<Lanes as crate::operations::Lane>::default_lane())).await.unwrap();
518        assert!(runtime.operations.definition(Probe::ID).is_ok());
519        assert!(runtime.operations.resident.definition("host.idle").is_some());
520        runtime.unregister(&crate::QualifiedId::parse("host.idle").unwrap()).await.unwrap();
521        runtime.unregister(&crate::QualifiedId::parse("host.test.probe").unwrap()).await.unwrap();
522        runtime.shutdown().await.unwrap();
523    }
524
525    #[tokio::test]
526    async fn operation_and_service_share_one_identity_namespace() {
527        let runtime = crate::Runtime::<Model>::launch_with((), (), |setup| {
528            setup.resource::<AtomicUsize>(Arc::new(AtomicUsize::new(0)));
529        })
530        .await
531        .unwrap();
532        let operation = harmos_sdk::__private::RegistrationBatch::new().operation::<Probe>();
533        let policy = crate::ServicePolicy {
534            stop_timeout: std::time::Duration::from_secs(1),
535            restart: crate::RestartPolicy::Never,
536        };
537        let service = harmos_sdk::__private::RegistrationBatch::new().service::<Idle>("test.probe", policy);
538        assert!(
539            runtime
540                .register_batch(
541                    operation.clone().service::<Idle>("test.probe", policy).lane(<Lanes as crate::operations::Lane>::default_lane())
542                )
543                .await
544                .is_err()
545        );
546        assert!(runtime.operations.definition(Probe::ID).is_err());
547        assert!(!runtime.operations.resident.contains("host.test.probe"));
548        runtime.operations.register_batch(operation.clone().lane(<Lanes as crate::operations::Lane>::default_lane())).await.unwrap();
549        assert!(runtime.services().register_batch(service.clone()).await.is_err());
550        runtime.unregister(&crate::QualifiedId::parse("host.test.probe").unwrap()).await.unwrap();
551        runtime.services().register_batch(service).await.unwrap();
552        assert!(runtime.operations.register_batch(operation.lane(<Lanes as crate::operations::Lane>::default_lane())).await.is_err());
553        runtime.unregister(&crate::QualifiedId::parse("host.test.probe").unwrap()).await.unwrap();
554        runtime.shutdown().await.unwrap();
555    }
556
557    #[tokio::test]
558    async fn native_registration_is_atomic_and_invocation_policy_is_local() {
559        let initialized = Arc::new(AtomicUsize::new(0));
560        let runtime = crate::Runtime::<Model>::launch_with((), (), |setup| {
561            setup.resource::<AtomicUsize>(initialized.clone());
562        })
563        .await
564        .unwrap();
565        let batch = harmos_sdk::__private::RegistrationBatch::new().operation::<Probe>();
566        assert!(
567            runtime
568                .register_batch(batch.clone().operation::<Probe>().lane(<Lanes as crate::operations::Lane>::default_lane()))
569                .await
570                .is_err()
571        );
572        assert!(runtime.operations.definition(Probe::ID).is_err());
573        let mut mismatched = batch.clone();
574        mismatched.operations[0].definition.id = crate::Identifier::new("different").unwrap();
575        assert!(runtime.register_batch(mismatched.lane(<Lanes as crate::operations::Lane>::default_lane())).await.is_err());
576        assert!(runtime.operations.definition("different").is_err());
577        runtime.operations.register_batch(batch.lane(<Lanes as crate::operations::Lane>::default_lane())).await.unwrap();
578        assert!(matches!(
579            runtime.operations.call_encoded("host.test.probe", Vec::new()).await,
580            Err(Error::InvalidInput { field: "operation preparation bytes", .. })
581        ));
582        let invocation = runtime.operation(Probe { value: 1 });
583        assert_eq!(initialized.load(Ordering::SeqCst), 0);
584        assert!(invocation.run().await.is_err());
585        assert_eq!(initialized.load(Ordering::SeqCst), 1);
586
587        let retry = RetryPolicy::OnTransient {
588            max_retries: 1,
589            backoff: crate::Backoff::Fixed {
590                delay: std::time::Duration::from_millis(1),
591                jitter: crate::Jitter::None,
592            },
593        };
594        assert_eq!(*runtime.operation(Probe { value: 1 }).retry(retry).run().await.unwrap(), 0);
595        assert_eq!(initialized.load(Ordering::SeqCst), 2);
596        assert!(runtime.operation(Probe { value: 1 }).run().await.is_err());
597        assert_eq!(initialized.load(Ordering::SeqCst), 3);
598        assert_eq!(runtime.operations.definition(Probe::ID).unwrap().policy.retries, RetryPolicy::Never);
599        runtime.operations.unregister(&crate::QualifiedId::parse("host.test.probe").unwrap()).await.unwrap();
600        assert!(runtime.operation(Probe { value: 0 }).start().await.is_err());
601        runtime.shutdown().await.unwrap();
602    }
603}
harmos_runtime/operations/surface.rs

harmos_runtime/operations/
surface.rs
