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

harmos_runtime/operations/
surface.rs
