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, evidence::Evidence, 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    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
76    Plugin(crate::plugin::PluginScope),
77}
78
79pub(crate) fn authored_runner<A: Application>(
80    definition: ServiceDefinition, factory: harmos_sdk::operations::ServiceFactory, owner: ServiceOwner,
81) -> Arc<dyn crate::operations::resident::Runner> {
82    Arc::new(Typed::<A, (), _>::new(definition, move |_| AuthoredService {
83        factory: factory.clone(),
84        instance: None,
85        owner: owner.clone(),
86    }))
87}
88
89/* === Structs === */
90
91#[derive(Clone, Debug)]
92pub struct ServiceDefinition {
93    pub id: String,
94    pub policy: ServicePolicy,
95}
96
97// The runtime managers a service context lends as typed values.
98pub(crate) struct Handles<A: Application> {
99    pub(crate) storage: Arc<std::sync::OnceLock<crate::storage::StorageManager<crate::storage::ErasedCatalog>>>,
100    pub(crate) state: StateManager<A>,
101    pub(crate) records: crate::records::RecordManager<A>,
102    pub(crate) streams: StreamManager<A>,
103    pub(crate) evidence: Evidence<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) evidence: Evidence<A>,
124    pub(crate) stopping: watch::Receiver<bool>,
125    pub(crate) ready: watch::Sender<bool>,
126
127    #[cfg(feature = "plugin-sidecar")]
128    pub(crate) provider: Arc<crate::supervision::instance::Instance>,
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(&self, batch: harmos_sdk::RegistrationBatch) -> Result<()> {
198        let (staged, ids) = self.stage(batch)?;
199        staged.commit()?;
200        self.activate(&ids).await
201    }
202
203    pub(crate) fn stage(&self, batch: harmos_sdk::RegistrationBatch) -> Result<(crate::operations::resident::Staged, Vec<String>)> {
204        if !batch.operations.is_empty() {
205            return Err(Error::InvalidInput {
206                field: "service registration",
207                reason: "operations belong to the operation manager",
208            });
209        }
210        let mut entries = Vec::with_capacity(batch.services.len());
211        let mut ids = Vec::with_capacity(batch.services.len());
212        for factory in batch.services {
213            let id = crate::QualifiedId::qualify(self.owner(), &factory.local_id).map_err(|_| Error::InvalidInput {
214                field: "service identity",
215                reason: "invalid local service identifier",
216            })?;
217            let definition = ServiceDefinition {
218                id: id.as_str().to_owned(),
219                policy: factory.policy,
220            };
221            validate(&definition)?;
222            ids.push(definition.id.clone());
223            let runner = (self.operations.registry.service_runner)(definition.clone(), factory, self.owner.clone());
224            entries.push((crate::operations::resident::Definition::Service(definition), runner));
225        }
226        let staged = self.operations.stage_services(
227            entries,
228            #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
229            match &self.owner {
230                ServiceOwner::Host => None,
231                ServiceOwner::Plugin(scope) => Some(scope.clone()),
232            },
233        )?;
234        Ok((staged, ids))
235    }
236
237    pub(crate) async fn activate(&self, ids: &[String]) -> Result<()> {
238        for id in ids {
239            if let Err(error) = self.operations.start_registered_service(id, &self.owner).await {
240                for id in ids {
241                    self.operations.resident.unregister(id).await?;
242                }
243                return Err(error);
244            }
245        }
246        Ok(())
247    }
248
249    pub async fn unregister(&self, id: &crate::QualifiedId) -> Result<()> {
250        if id.as_str().split_once('.').map(|(owner, _)| owner) != Some(self.owner()) {
251            return Err(Error::PermissionDenied);
252        }
253        self.operations.resident.unregister(id.as_str()).await
254    }
255}
256
257impl<A: Application> Resident<A> for AuthoredService {
258    type Error = harmos_sdk::ServingError;
259
260    async fn run(&mut self, context: ServiceContext<A>) -> std::result::Result<(), Self::Error> {
261        let mut stopping = context.stopping.clone();
262        let context = Arc::new(context);
263        let mut dependencies = harmos_sdk::Dependencies::new();
264        match &self.owner {
265            ServiceOwner::Host => dependencies.insert(context.clone()).map_err(|_| harmos_sdk::ServingError::Catalog)?,
266            #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
267            ServiceOwner::Plugin(scope) => dependencies.insert(Arc::new(scope.clone())).map_err(|_| harmos_sdk::ServingError::Catalog)?,
268        }
269        let call = harmos_sdk::ExecutionContext::new(dependencies, harmos_sdk::operations::Host::context(&context.operations).clone());
270        self.instance = Some(self.factory.initialize(call.clone()).await?);
271        context.ready();
272        tokio::select! {
273            biased;
274            _ = stopping.wait_for(|stopping| *stopping) => Ok(()),
275            result = self.instance.as_mut().expect("initialized service").run(call) => result,
276        }
277    }
278
279    async fn stop(&mut self) -> std::result::Result<(), Self::Error> {
280        if let Some(instance) = &mut self.instance {
281            instance.stop().await?;
282        }
283        Ok(())
284    }
285
286    async fn cleanup(&mut self) -> std::result::Result<(), Self::Error> {
287        if let Some(instance) = &mut self.instance {
288            instance.cleanup().await?;
289        }
290        self.instance = None;
291        Ok(())
292    }
293}
294
295impl ServiceDefinition {
296    pub fn new(id: impl Into<String>) -> Self {
297        Self { id: id.into(), policy: POLICY }
298    }
299}
300
301impl<A: Application> Handles<A> {
302    pub(crate) fn context(
303        &self, operations: Manager, stopping: watch::Receiver<bool>, ready: watch::Sender<bool>,
304        #[cfg(feature = "plugin-sidecar")] provider: Arc<crate::supervision::instance::Instance>,
305    ) -> ServiceContext<A> {
306        ServiceContext {
307            state: self.state.clone(),
308            records: self.records.clone(),
309            operations,
310            streams: self.streams.clone(),
311            evidence: self.evidence.clone(),
312            stopping,
313            ready,
314            #[cfg(feature = "plugin-sidecar")]
315            provider,
316            #[cfg(feature = "plugin-sidecar")]
317            deactivation: Arc::default(),
318        }
319    }
320}
321
322impl<A: Application> Clone for Handles<A> {
323    fn clone(&self) -> Self {
324        Self {
325            storage: self.storage.clone(),
326            state: self.state.clone(),
327            records: self.records.clone(),
328            streams: self.streams.clone(),
329            evidence: self.evidence.clone(),
330        }
331    }
332}
333
334impl<A: Application> ServiceContext<A> {
335    pub fn ready(&self) {
336        if !self.is_stopping() {
337            self.ready.send_replace(true);
338        }
339    }
340
341    // An owned host handle for calls made while this context is borrowed,
342    // such as inside `until_stopping`.
343    pub fn host(&self) -> crate::operations::OperationManager<A::Lanes> {
344        crate::operations::OperationManager::new(self.operations.clone())
345    }
346
347    pub fn records(&self) -> &crate::records::RecordManager<A> {
348        &self.records
349    }
350
351    pub fn state(&self) -> &StateManager<A> {
352        &self.state
353    }
354
355    pub fn streams(&self) -> &StreamManager<A> {
356        &self.streams
357    }
358
359    pub fn evidence(&self) -> &Evidence<A> {
360        &self.evidence
361    }
362
363    pub fn is_stopping(&self) -> bool {
364        *self.stopping.borrow()
365    }
366
367    pub async fn stopping(&mut self) {
368        while !*self.stopping.borrow_and_update() {
369            if self.stopping.changed().await.is_err() {
370                break;
371            }
372        }
373    }
374
375    pub async fn rest(&mut self, duration: Duration) -> bool {
376        self.until_stopping(executor::sleep(duration)).await.is_none()
377    }
378
379    pub async fn until_stopping<T>(&mut self, future: impl Future<Output = T>) -> Option<T> {
380        if self.is_stopping() {
381            return None;
382        }
383        let mut future = Box::pin(future);
384        let mut stop = Box::pin(self.stopping.changed());
385
386        std::future::poll_fn(|context| {
387            if let Poll::Ready(Ok(()) | Err(_)) = stop.as_mut().poll(context) {
388                return Poll::Ready(None);
389            }
390            future.as_mut().poll(context).map(Some)
391        })
392        .await
393    }
394}
395
396impl<A: Application> harmos_sdk::operations::Host for ServiceContext<A> {
397    fn publish_on<S: harmos_domain::StreamSpec>(
398        &self, row: &S::Row,
399    ) -> impl std::future::Future<Output = std::result::Result<(), Self::Error>> + Send
400    where
401        S::Row: harmos_domain::Message,
402    {
403        use harmos_domain::Message;
404        let published = self.call::<harmos_operations::streams::Publish>(harmos_operations::streams::Publish {
405            stream: S::ID.to_owned(),
406            schema: S::VERSION,
407            sequence: None,
408            published_unix_micros: None,
409            context: None,
410            payload: row.encode(),
411            following: Vec::new(),
412        });
413        async move {
414            published.await?;
415            Ok(())
416        }
417    }
418
419    type Error = Error;
420    type Instance<O: OperationDeclaration> = Running<O>;
421
422    fn context(&self) -> &CallContext {
423        harmos_sdk::operations::Host::context(&self.operations)
424    }
425
426    fn with_value<T: ContextValue>(&self, value: T) -> std::result::Result<Self, ContextError> {
427        let operations = harmos_sdk::operations::Host::with_value(&self.operations, value)?;
428        let mut child = self.clone();
429        child.operations = operations;
430        Ok(child)
431    }
432
433    fn submit<O: OperationDeclaration>(&self, request: O::Request) -> impl Future<Output = Result<Running<O>>> + Send {
434        harmos_sdk::operations::Host::submit::<O>(&self.operations, request)
435    }
436
437    fn submit_owned<O: OperationDeclaration>(
438        &self, request: O::Request, claims: crate::operations::RequestClaims,
439    ) -> impl Future<Output = Result<Running<O>>> + Send {
440        harmos_sdk::operations::Host::submit_owned::<O>(&self.operations, request, claims)
441    }
442
443    fn call<O: OperationDeclaration>(&self, request: O::Request) -> impl Future<Output = Result<Response<O::Response>>> + Send {
444        harmos_sdk::operations::Host::call::<O>(&self.operations, request)
445    }
446}
447
448impl<A: Application> Clone for ServiceContext<A> {
449    fn clone(&self) -> Self {
450        Self {
451            state: self.state.clone(),
452            records: self.records.clone(),
453            operations: self.operations.clone(),
454            streams: self.streams.clone(),
455            evidence: self.evidence.clone(),
456            stopping: self.stopping.clone(),
457            ready: self.ready.clone(),
458            #[cfg(feature = "plugin-sidecar")]
459            provider: self.provider.clone(),
460            #[cfg(feature = "plugin-sidecar")]
461            deactivation: self.deactivation.clone(),
462        }
463    }
464}
465
466impl<A, F, Fut, E> Resident<A> for F
467where
468    A: Application,
469    F: FnMut(ServiceContext<A>) -> Fut + Send + 'static,
470    Fut: Future<Output = std::result::Result<(), E>> + Send + 'static,
471    E: std::error::Error + Send + Sync + 'static,
472{
473    type Error = E;
474
475    fn run(&mut self, context: ServiceContext<A>) -> impl Future<Output = std::result::Result<(), Self::Error>> + Send {
476        self(context)
477    }
478
479    // Closures must not own asynchronous resources beyond their run future.
480    // Resource-owning services implement Resident and retain their cleanup state.
481    async fn cleanup(&mut self) -> std::result::Result<(), Self::Error> {
482        Ok(())
483    }
484}
485
486/* === Public Functions === */
487
488/* === Private Functions === */
489
490pub(crate) fn validate(definition: &ServiceDefinition) -> Result<()> {
491    if definition.id.is_empty() {
492        return Err(Error::InvalidInput {
493            field: "service name",
494            reason: "must not be empty",
495        });
496    }
497    if definition.id == crate::evidence::SOURCE {
498        return Err(Error::InvalidInput {
499            field: "service name",
500            reason: "artifact is reserved for evidence facts",
501        });
502    }
503    definition.policy.validate().map_err(|source| Error::OperationDeclaration { source: source.into() })
504}
505
506/* === Tests === */
507
508#[cfg(test)]
509mod tests {
510    use super::*;
511    use std::sync::atomic::{AtomicUsize, Ordering};
512
513    #[crate::lanes]
514    enum Lanes {
515        #[default]
516        Default,
517    }
518
519    struct Model;
520    impl Application for Model {
521        type Lanes = Lanes;
522        type State = ();
523        type Transaction = crate::state::Never;
524        type Record = ();
525        type Streams = ();
526        type Resources = ();
527    }
528
529    #[derive(Default)]
530    struct Counts {
531        initialized: AtomicUsize,
532        stopped: AtomicUsize,
533        cleaned: AtomicUsize,
534        running: tokio::sync::Notify,
535    }
536
537    struct Worker {
538        counts: Arc<Counts>,
539        attempt: usize,
540    }
541
542    impl harmos_sdk::Service for Worker {
543        type Error = crate::Error;
544
545        async fn initialize(context: harmos_sdk::ExecutionContext) -> Result<Self> {
546            let counts = context.resource::<Counts>()?;
547            let attempt = counts.initialized.fetch_add(1, Ordering::SeqCst);
548            Ok(Self { counts, attempt })
549        }
550
551        async fn run(&mut self, context: harmos_sdk::ExecutionContext) -> Result<()> {
552            if self.attempt == 0 {
553                return Err(Error::InvalidInput {
554                    field: "fixture",
555                    reason: "first attempt fails",
556                });
557            }
558            let context = context.resource::<ServiceContext<Model>>()?;
559            self.counts.running.notify_one();
560            let mut context = (*context).clone();
561            context.stopping().await;
562            Ok(())
563        }
564
565        async fn stop(&mut self) -> Result<()> {
566            self.counts.stopped.fetch_add(1, Ordering::SeqCst);
567            Ok(())
568        }
569
570        async fn cleanup(&mut self) -> Result<()> {
571            self.counts.cleaned.fetch_add(1, Ordering::SeqCst);
572            Ok(())
573        }
574    }
575
576    #[tokio::test]
577    async fn authored_service_reconstructs_after_cleanup_and_stops_on_withdrawal() {
578        use crate::RegistrationExt;
579
580        let runtime = crate::Runtime::<Model>::builder((), ()).register(Lanes::catalog()).runtime().await.unwrap();
581        let counts = Arc::new(Counts::default());
582        let mut dependencies = harmos_sdk::Dependencies::new();
583        dependencies.insert(counts.clone()).unwrap();
584        let policy = ServicePolicy {
585            stop_timeout: Duration::from_secs(1),
586            restart: RestartPolicy::OnFailure {
587                credit: 1,
588                sustained_health: Duration::from_secs(1),
589                backoff: Backoff::Fixed {
590                    delay: Duration::from_millis(1),
591                    jitter: Jitter::None,
592                },
593            },
594        };
595        let batch = harmos_sdk::RegistrationBatch::new(dependencies).service::<Worker>("worker", policy);
596        runtime.register(batch.defaults()).await.unwrap();
597        executor::timeout(Duration::from_secs(2), counts.running.notified()).await.unwrap();
598        assert_eq!(counts.initialized.load(Ordering::SeqCst), 2);
599        assert_eq!(counts.cleaned.load(Ordering::SeqCst), 1);
600
601        runtime.unregister(&crate::QualifiedId::parse("host.worker").unwrap()).await.unwrap();
602        assert_eq!(counts.stopped.load(Ordering::SeqCst), 2);
603        assert_eq!(counts.cleaned.load(Ordering::SeqCst), 2);
604        runtime.stop().await.unwrap();
605        assert_eq!(counts.initialized.load(Ordering::SeqCst), 2);
606    }
607}
harmos_runtime/services/mod.rs

harmos_runtime/services/
mod.rs
