harmos_runtime/services/
mod.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// Resident Services Use Independent Restart Credit And Bounded Stop Deadlines.
11
12/* === Modules === */
13
14mod restart;
15mod runner;
16
17#[cfg(feature = "plugin-sidecar")]
18mod sidecar;
19
20/* === Re-Exports === */
21
22pub(crate) use runner::{Once, Typed};
23
24#[cfg(feature = "plugin-sidecar")]
25pub(crate) use sidecar::Endpoint as NativeEndpoint;
26#[cfg(feature = "plugin-sidecar")]
27pub(crate) use sidecar::RuntimeDriver;
28
29#[cfg(feature = "plugin-sidecar")]
30pub use sidecar::{DemandLease, DemandSource, Managed, ManagedError, Notice, SidecarError, Supervised};
31
32/* === Imports === */
33
34// Standard Library Imports
35use std::future::Future;
36use std::sync::Arc;
37use std::task::Poll;
38use std::time::Duration;
39// Third Party Imports
40use tokio::sync::watch;
41// Local Imports
42use crate::ServicePolicy;
43use crate::operations::{
44    Backoff, CallContext, ContextError, ContextValue, Jitter, Manager, OperationDeclaration, Response, RestartPolicy, Running,
45};
46use crate::supervision::executor;
47use crate::{Application, Error, Result, state::StateManager, streams::StreamManager};
48
49/* === Constants === */
50
51// Three Restart Credits Refill After Sustained Health.
52pub const POLICY: ServicePolicy = ServicePolicy {
53    stop_timeout: Duration::from_secs(5),
54    restart: RestartPolicy::OnFailure {
55        credit: 3,
56        sustained_health: Duration::from_millis(250),
57        backoff: Backoff::Exponential {
58            initial: Duration::from_millis(10),
59            maximum: Duration::from_secs(5),
60            jitter: Jitter::Deterministic,
61        },
62    },
63};
64
65/* === Macros === */
66
67/* === Type Aliases === */
68
69pub(crate) type ServiceRunner =
70    fn(ServiceDefinition, harmos_sdk::operations::ServiceFactory, ServiceOwner) -> Arc<dyn crate::operations::resident::Runner>;
71
72#[derive(Clone)]
73pub(crate) enum ServiceOwner {
74    Host,
75
76    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
77    Plugin(crate::plugin::PluginScope),
78}
79
80pub(crate) fn authored_runner<A: Application>(
81    definition: ServiceDefinition, factory: harmos_sdk::operations::ServiceFactory, owner: ServiceOwner,
82) -> Arc<dyn crate::operations::resident::Runner> {
83    Arc::new(Typed::<A, harmos_sdk::ExecutionContext, _>::new(definition, move |_| AuthoredService {
84        factory: factory.clone(),
85        instance: None,
86        owner: owner.clone(),
87    }))
88}
89
90/* === Structs === */
91
92#[derive(Clone, Debug)]
93pub struct ServiceDefinition {
94    pub id: String,
95    pub policy: ServicePolicy,
96}
97
98// The runtime managers a service context lends as typed values.
99pub(crate) struct Handles<A: Application> {
100    pub(crate) storage: Arc<std::sync::OnceLock<crate::storage::StorageManager<crate::storage::ErasedCatalog>>>,
101    pub(crate) state: StateManager<A>,
102    pub(crate) records: crate::records::RecordManager<A>,
103    pub(crate) streams: StreamManager<A>,
104}
105
106#[derive(Clone)]
107pub struct ServiceManager {
108    operations: Manager,
109    owner: ServiceOwner,
110}
111
112struct AuthoredService {
113    owner: ServiceOwner,
114    factory: harmos_sdk::operations::ServiceFactory,
115    instance: Option<Box<dyn harmos_sdk::operations::ServiceInstance>>,
116}
117
118pub struct ServiceContext<A: Application> {
119    pub(crate) state: StateManager<A>,
120    pub(crate) records: crate::records::RecordManager<A>,
121    pub(crate) operations: Manager,
122    pub(crate) streams: StreamManager<A>,
123    pub(crate) stopping: watch::Receiver<bool>,
124    pub(crate) ready: watch::Sender<bool>,
125
126    #[cfg(feature = "plugin-sidecar")]
127    pub(crate) provider: Arc<crate::supervision::instance::Instance>,
128
129    #[cfg(feature = "plugin-sidecar")]
130    deactivation: Arc<std::sync::atomic::AtomicBool>,
131}
132
133/* === Enums === */
134
135#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
136pub enum ServiceExit {
137    #[error("application: {0}")]
138    Application(String),
139
140    #[error("resident panicked")]
141    Panicked,
142
143    #[error("restart credit exhausted")]
144    CreditExhausted,
145
146    #[error("resident stop deadline exceeded")]
147    StopTimedOut,
148}
149
150#[derive(Clone, Debug, PartialEq, Eq)]
151pub enum ServiceStatus {
152    Starting,
153    Running { credit: usize },
154    Stopping,
155    BackingOff,
156    Stopped,
157    Inactive { cause: ServiceExit, history: Vec<String> },
158}
159
160/* === Traits === */
161
162// A resident runs until stopped. Each run is stopped and cleaned up before the
163// next restart or before the instance stops.
164pub trait Resident<A: Application>: Send + 'static {
165    type Error: std::error::Error + Send + Sync + 'static;
166
167    fn run(&mut self, context: ServiceContext<A>) -> impl Future<Output = std::result::Result<(), Self::Error>> + Send;
168
169    fn stop(&mut self) -> impl Future<Output = std::result::Result<(), Self::Error>> + Send {
170        async { Ok(()) }
171    }
172
173    fn cleanup(&mut self) -> impl Future<Output = std::result::Result<(), Self::Error>> + Send;
174}
175
176/* === Implementations === */
177
178impl ServiceManager {
179    pub(crate) fn new(operations: Manager) -> Self {
180        Self { operations, owner: ServiceOwner::Host }
181    }
182    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
183    pub(crate) fn scoped(operations: Manager, scope: crate::plugin::PluginScope) -> Self {
184        Self {
185            operations,
186            owner: ServiceOwner::Plugin(scope),
187        }
188    }
189    fn owner(&self) -> &str {
190        match &self.owner {
191            ServiceOwner::Host => "host",
192            #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
193            ServiceOwner::Plugin(scope) => scope.owner(),
194        }
195    }
196
197    pub async fn register<S: harmos_sdk::Service>(&self, id: &str, policy: crate::ServicePolicy) -> Result<()> {
198        let batch = harmos_sdk::__private::RegistrationBatch::new().service::<S>(id, policy);
199        self.register_batch(batch).await
200    }
201
202    pub(crate) async fn register_batch(&self, batch: harmos_sdk::__private::RegistrationBatch) -> Result<()> {
203        let (staged, ids) = self.stage(batch)?;
204        staged.commit()?;
205        self.activate(&ids).await
206    }
207
208    pub(crate) fn stage(
209        &self, batch: harmos_sdk::__private::RegistrationBatch,
210    ) -> Result<(crate::operations::resident::Staged, Vec<String>)> {
211        if !batch.operations.is_empty() {
212            return Err(Error::InvalidInput {
213                field: "service registration",
214                reason: "operations belong to the operation manager",
215            });
216        }
217        let mut entries = Vec::with_capacity(batch.services.len());
218        let mut ids = Vec::with_capacity(batch.services.len());
219        for factory in batch.services {
220            let id = crate::QualifiedId::qualify(self.owner(), &factory.local_id).map_err(|_| Error::InvalidInput {
221                field: "service identity",
222                reason: "invalid local service identifier",
223            })?;
224            let definition = ServiceDefinition {
225                id: id.as_str().to_owned(),
226                policy: factory.policy,
227            };
228            validate(&definition)?;
229            ids.push(definition.id.clone());
230            let runner = (self.operations.registry.service_runner)(definition.clone(), factory, self.owner.clone());
231            entries.push((crate::operations::resident::Definition::Service(definition), runner));
232        }
233        let staged = self.operations.stage_services(
234            entries,
235            #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
236            match &self.owner {
237                ServiceOwner::Host => None,
238                ServiceOwner::Plugin(scope) => Some(scope.clone()),
239            },
240        )?;
241        Ok((staged, ids))
242    }
243
244    pub(crate) async fn activate(&self, ids: &[String]) -> Result<()> {
245        for id in ids {
246            if let Err(error) = self.operations.start_registered_service(id, &self.owner).await {
247                for id in ids {
248                    self.operations.resident.unregister(id).await?;
249                }
250                return Err(error);
251            }
252        }
253        Ok(())
254    }
255
256    pub async fn unregister(&self, id: &crate::QualifiedId) -> Result<()> {
257        if id.as_str().split_once('.').map(|(owner, _)| owner) != Some(self.owner()) {
258            return Err(Error::PermissionDenied);
259        }
260        self.operations.resident.unregister(id.as_str()).await
261    }
262}
263
264impl<A: Application> Resident<A> for AuthoredService {
265    type Error = harmos_sdk::ServingError;
266
267    async fn run(&mut self, context: ServiceContext<A>) -> std::result::Result<(), Self::Error> {
268        let mut stopping = context.stopping.clone();
269        let context = Arc::new(context);
270        let mut dependencies = harmos_sdk::__private::ResourceSet::new();
271        match &self.owner {
272            ServiceOwner::Host => dependencies.insert(context.clone()).map_err(|_| harmos_sdk::ServingError::Catalog)?,
273            #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
274            ServiceOwner::Plugin(scope) => dependencies.insert(Arc::new(scope.clone())).map_err(|_| harmos_sdk::ServingError::Catalog)?,
275        }
276        let call = harmos_sdk::ExecutionContext::new(dependencies, harmos_sdk::operations::Host::context(&context.operations).clone())
277            .with_resources(context.operations.resources().clone());
278        self.instance = Some(self.factory.initialize(call.clone()).await?);
279        context.ready();
280        tokio::select! {
281            biased;
282            _ = stopping.wait_for(|stopping| *stopping) => Ok(()),
283            result = self.instance.as_mut().expect("initialized service").run(call) => result,
284        }
285    }
286
287    async fn stop(&mut self) -> std::result::Result<(), Self::Error> {
288        if let Some(instance) = &mut self.instance {
289            instance.stop().await?;
290        }
291        Ok(())
292    }
293
294    async fn cleanup(&mut self) -> std::result::Result<(), Self::Error> {
295        if let Some(instance) = &mut self.instance {
296            instance.cleanup().await?;
297        }
298        self.instance = None;
299        Ok(())
300    }
301}
302
303impl ServiceDefinition {
304    pub fn new(id: impl Into<String>) -> Self {
305        Self { id: id.into(), policy: POLICY }
306    }
307}
308
309impl<A: Application> Handles<A> {
310    pub(crate) fn context(
311        &self, operations: Manager, stopping: watch::Receiver<bool>, ready: watch::Sender<bool>,
312        #[cfg(feature = "plugin-sidecar")] provider: Arc<crate::supervision::instance::Instance>,
313    ) -> ServiceContext<A> {
314        ServiceContext {
315            state: self.state.clone(),
316            records: self.records.clone(),
317            operations,
318            streams: self.streams.clone(),
319            stopping,
320            ready,
321            #[cfg(feature = "plugin-sidecar")]
322            provider,
323            #[cfg(feature = "plugin-sidecar")]
324            deactivation: Arc::default(),
325        }
326    }
327}
328
329impl<A: Application> Clone for Handles<A> {
330    fn clone(&self) -> Self {
331        Self {
332            storage: self.storage.clone(),
333            state: self.state.clone(),
334            records: self.records.clone(),
335            streams: self.streams.clone(),
336        }
337    }
338}
339
340impl<A: Application> ServiceContext<A> {
341    pub fn resource<T: Send + Sync + 'static>(&self) -> std::result::Result<Arc<T>, harmos_sdk::ResourceError> {
342        self.operations.resources().resource::<T>()
343    }
344
345    pub fn ready(&self) {
346        if !self.is_stopping() {
347            self.ready.send_replace(true);
348        }
349    }
350
351    // An owned host handle for calls made while this context is borrowed,
352    // such as inside `until_stopping`.
353    pub fn host(&self) -> crate::operations::OperationManager<A::Lanes> {
354        crate::operations::OperationManager::new(self.operations.clone())
355    }
356
357    pub fn records(&self) -> &crate::records::RecordManager<A> {
358        &self.records
359    }
360
361    pub fn state(&self) -> &StateManager<A> {
362        &self.state
363    }
364
365    pub fn streams(&self) -> &StreamManager<A> {
366        &self.streams
367    }
368
369    pub fn is_stopping(&self) -> bool {
370        *self.stopping.borrow()
371    }
372
373    pub async fn stopping(&mut self) {
374        while !*self.stopping.borrow_and_update() {
375            if self.stopping.changed().await.is_err() {
376                break;
377            }
378        }
379    }
380
381    pub async fn rest(&mut self, duration: Duration) -> bool {
382        self.until_stopping(executor::sleep(duration)).await.is_none()
383    }
384
385    pub async fn until_stopping<T>(&mut self, future: impl Future<Output = T>) -> Option<T> {
386        if self.is_stopping() {
387            return None;
388        }
389        let mut future = Box::pin(future);
390        let mut stop = Box::pin(self.stopping.changed());
391
392        std::future::poll_fn(|context| {
393            if let Poll::Ready(Ok(()) | Err(_)) = stop.as_mut().poll(context) {
394                return Poll::Ready(None);
395            }
396            future.as_mut().poll(context).map(Some)
397        })
398        .await
399    }
400}
401
402impl<A: Application> harmos_sdk::operations::Host for ServiceContext<A> {
403    fn publish_on<S: harmos_domain::StreamSpec>(
404        &self, row: &S::Row,
405    ) -> impl std::future::Future<Output = std::result::Result<(), Self::Error>> + Send
406    where
407        S::Row: harmos_domain::Message,
408    {
409        use harmos_domain::Message;
410        let published = self.call::<harmos_operations::streams::Publish>(harmos_operations::streams::Publish {
411            stream: S::ID.to_owned(),
412            schema: S::VERSION,
413            sequence: None,
414            published_unix_micros: None,
415            context: None,
416            payload: row.encode(),
417            following: Vec::new(),
418        });
419        async move {
420            published.await?;
421            Ok(())
422        }
423    }
424
425    type Error = Error;
426    type Instance<O: OperationDeclaration> = Running<O>;
427
428    fn context(&self) -> &CallContext {
429        harmos_sdk::operations::Host::context(&self.operations)
430    }
431
432    fn with_value<T: ContextValue>(&self, value: T) -> std::result::Result<Self, ContextError> {
433        let operations = harmos_sdk::operations::Host::with_value(&self.operations, value)?;
434        let mut child = self.clone();
435        child.operations = operations;
436        Ok(child)
437    }
438
439    fn submit<O: OperationDeclaration>(&self, request: O::Request) -> impl Future<Output = Result<Running<O>>> + Send {
440        harmos_sdk::operations::Host::submit::<O>(&self.operations, request)
441    }
442
443    fn submit_owned<O: OperationDeclaration>(
444        &self, request: O::Request, claims: crate::operations::RequestClaims,
445    ) -> impl Future<Output = Result<Running<O>>> + Send {
446        harmos_sdk::operations::Host::submit_owned::<O>(&self.operations, request, claims)
447    }
448
449    fn call<O: OperationDeclaration>(&self, request: O::Request) -> impl Future<Output = Result<Response<O::Response>>> + Send {
450        harmos_sdk::operations::Host::call::<O>(&self.operations, request)
451    }
452}
453
454impl<A: Application> Clone for ServiceContext<A> {
455    fn clone(&self) -> Self {
456        Self {
457            state: self.state.clone(),
458            records: self.records.clone(),
459            operations: self.operations.clone(),
460            streams: self.streams.clone(),
461            stopping: self.stopping.clone(),
462            ready: self.ready.clone(),
463            #[cfg(feature = "plugin-sidecar")]
464            provider: self.provider.clone(),
465            #[cfg(feature = "plugin-sidecar")]
466            deactivation: self.deactivation.clone(),
467        }
468    }
469}
470
471impl<A, F, Fut, E> Resident<A> for F
472where
473    A: Application,
474    F: FnMut(ServiceContext<A>) -> Fut + Send + 'static,
475    Fut: Future<Output = std::result::Result<(), E>> + Send + 'static,
476    E: std::error::Error + Send + Sync + 'static,
477{
478    type Error = E;
479
480    fn run(&mut self, context: ServiceContext<A>) -> impl Future<Output = std::result::Result<(), Self::Error>> + Send {
481        self(context)
482    }
483
484    // Closures must not own asynchronous resources beyond their run future.
485    // Resource-owning services implement Resident and retain their cleanup state.
486    async fn cleanup(&mut self) -> std::result::Result<(), Self::Error> {
487        Ok(())
488    }
489}
490
491/* === Public Functions === */
492
493/* === Private Functions === */
494
495pub(crate) fn validate(definition: &ServiceDefinition) -> Result<()> {
496    if definition.id.is_empty() {
497        return Err(Error::InvalidInput {
498            field: "service name",
499            reason: "must not be empty",
500        });
501    }
502    definition.policy.validate().map_err(|source| Error::OperationDeclaration { source: source.into() })
503}
504
505/* === Tests === */
506
507#[cfg(test)]
508mod tests {
509    use super::*;
510    use std::sync::atomic::{AtomicUsize, Ordering};
511
512    #[crate::lanes]
513    enum Lanes {
514        #[default]
515        Default,
516    }
517
518    struct Model;
519    impl crate::Lifecycle for Model {
520        type Config = ();
521        type Error = std::convert::Infallible;
522        type Context = crate::Startup<Self>;
523
524        async fn prepare(_: (), _: &mut crate::Setup<Self>) -> std::result::Result<Self, Self::Error> {
525            Ok(Self)
526        }
527
528        async fn initialize(&mut self, _: &Self::Context) -> std::result::Result<(), Self::Error> {
529            Ok(())
530        }
531
532        async fn stop(&mut self, _: &Self::Context) -> std::result::Result<(), Self::Error> {
533            Ok(())
534        }
535
536        async fn cleanup(&mut self) -> std::result::Result<(), Self::Error> {
537            Ok(())
538        }
539    }
540
541    impl Application for Model {
542        type Lanes = Lanes;
543        type State = ();
544        type Transaction = crate::state::Never;
545        type Record = ();
546        type Streams = ();
547    }
548
549    #[derive(Default)]
550    struct Counts {
551        initialized: AtomicUsize,
552        stopped: AtomicUsize,
553        cleaned: AtomicUsize,
554        running: tokio::sync::Notify,
555    }
556
557    struct Worker {
558        counts: Arc<Counts>,
559        attempt: usize,
560    }
561
562    impl harmos_sdk::Service for Worker {
563        type Error = crate::Error;
564
565        async fn initialize(context: harmos_sdk::ExecutionContext) -> Result<Self> {
566            let counts = context.resource::<Counts>()?;
567            let attempt = counts.initialized.fetch_add(1, Ordering::SeqCst);
568            Ok(Self { counts, attempt })
569        }
570
571        async fn run(&mut self, context: harmos_sdk::ExecutionContext) -> Result<()> {
572            if self.attempt == 0 {
573                return Err(Error::InvalidInput {
574                    field: "fixture",
575                    reason: "first attempt fails",
576                });
577            }
578            let context = context.resource::<ServiceContext<Model>>()?;
579            self.counts.running.notify_one();
580            let mut context = (*context).clone();
581            context.stopping().await;
582            Ok(())
583        }
584
585        async fn stop(&mut self) -> Result<()> {
586            self.counts.stopped.fetch_add(1, Ordering::SeqCst);
587            Ok(())
588        }
589
590        async fn cleanup(&mut self) -> Result<()> {
591            self.counts.cleaned.fetch_add(1, Ordering::SeqCst);
592            Ok(())
593        }
594    }
595
596    #[tokio::test]
597    async fn authored_service_reconstructs_after_cleanup_and_stops_on_withdrawal() {
598        use crate::operations::RegistrationExt;
599
600        let counts = Arc::new(Counts::default());
601        let runtime = crate::Runtime::<Model>::launch_with((), (), |setup| {
602            setup.resource::<Counts>(counts.clone());
603        })
604        .await
605        .unwrap();
606        let policy = ServicePolicy {
607            stop_timeout: Duration::from_secs(1),
608            restart: RestartPolicy::OnFailure {
609                credit: 1,
610                sustained_health: Duration::from_secs(1),
611                backoff: Backoff::Fixed {
612                    delay: Duration::from_millis(1),
613                    jitter: Jitter::None,
614                },
615            },
616        };
617        let batch = harmos_sdk::__private::RegistrationBatch::new().service::<Worker>("worker", policy);
618        runtime.register_batch(batch.lane(<Lanes as crate::operations::Lane>::default_lane())).await.unwrap();
619        executor::timeout(Duration::from_secs(2), counts.running.notified()).await.unwrap();
620        assert_eq!(counts.initialized.load(Ordering::SeqCst), 2);
621        assert_eq!(counts.cleaned.load(Ordering::SeqCst), 1);
622
623        runtime.unregister(&crate::QualifiedId::parse("host.worker").unwrap()).await.unwrap();
624        assert_eq!(counts.stopped.load(Ordering::SeqCst), 2);
625        assert_eq!(counts.cleaned.load(Ordering::SeqCst), 2);
626        runtime.shutdown().await.unwrap();
627        assert_eq!(counts.initialized.load(Ordering::SeqCst), 2);
628    }
629}
harmos_runtime/services/mod.rs

harmos_runtime/services/
mod.rs
