harmos_runtime/services/
mod.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// The service kind of resident operation: a resident that runs until stopped
11// under the resident half of its `OperationPolicy`. Environmental faults spend
12// restart credit and restart after backoff; application faults deactivate it.
13// The operation manager activates a service through `submit_with_resources`,
14// and its context is the same `Host` every other body uses, lending the
15// runtime's typed managers as context values.
16
17/* === Modules === */
18
19mod restart;
20mod runner;
21
22#[cfg(feature = "plugin-sidecar")]
23mod sidecar;
24
25/* === Re-Exports === */
26
27pub(crate) use runner::{Once, Typed};
28
29#[cfg(feature = "plugin-sidecar")]
30pub(crate) use sidecar::Endpoint as NativeEndpoint;
31#[cfg(feature = "plugin-sidecar")]
32pub(crate) use sidecar::RuntimeDriver;
33
34#[cfg(feature = "plugin-sidecar")]
35pub(crate) use sidecar::HostPublish;
36#[cfg(feature = "plugin-sidecar")]
37pub use sidecar::{DemandLease, DemandSource, Managed, ManagedError, Notice, Publish, SidecarError, Supervised};
38
39/* === Imports === */
40
41// Standard Library Imports
42use std::fmt;
43use std::future::Future;
44use std::sync::Arc;
45use std::task::Poll;
46use std::time::Duration;
47// Third Party Imports
48use tokio::sync::watch;
49// Local Imports
50use crate::operations::{
51    Backoff, CallContext, ContextError, ContextValue, Jitter, OperationDeclaration, OperationInstance, OperationManager, OperationPolicy,
52    Response, RestartPolicy, RetryPolicy,
53};
54use crate::supervision::executor;
55use crate::{Application, Error, Result, evidence::Evidence, state::StateManager, streams::StreamManager};
56
57/* === Constants === */
58
59// Three restarts after environmental faults, refilled after a quarter second of
60// health; `timeout` is the stop deadline.
61pub const POLICY: OperationPolicy = OperationPolicy {
62    timeout: Duration::from_secs(5),
63    attempt_timeout: Duration::from_secs(5),
64    retries: RetryPolicy::Never,
65    restart: RestartPolicy::OnTransient {
66        credit: 3,
67        sustained_health: Duration::from_millis(250),
68        backoff: Backoff::Exponential {
69            initial: Duration::from_millis(10),
70            maximum: Duration::from_secs(5),
71            jitter: Jitter::Deterministic,
72        },
73    },
74};
75
76/* === Macros === */
77
78/* === Type Aliases === */
79
80/* === Structs === */
81
82#[derive(Clone, Debug)]
83pub struct ServiceDefinition {
84    pub id: String,
85    pub policy: OperationPolicy,
86}
87
88// The runtime managers a service context lends as typed values.
89pub(crate) struct Handles<A: Application> {
90    pub(crate) storage: Arc<std::sync::OnceLock<crate::storage::StorageManager<crate::storage::ErasedCatalog>>>,
91    pub(crate) state: StateManager<A>,
92    pub(crate) records: crate::records::RecordManager<A>,
93    pub(crate) streams: StreamManager<A>,
94    pub(crate) evidence: Evidence<A>,
95}
96
97pub struct Service<A: Application> {
98    pub(crate) state: StateManager<A>,
99    pub(crate) records: crate::records::RecordManager<A>,
100    pub(crate) operations: OperationManager,
101    pub(crate) streams: StreamManager<A>,
102    pub(crate) evidence: Evidence<A>,
103    pub(crate) stopping: watch::Receiver<bool>,
104    pub(crate) ready: watch::Sender<bool>,
105
106    #[cfg(feature = "plugin-sidecar")]
107    pub(crate) provider: Arc<crate::supervision::instance::Instance>,
108}
109
110/* === Enums === */
111
112#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
113pub enum Fault<E> {
114    #[error("environmental fault: {0}")]
115    Environmental(E),
116
117    #[error("application fault: {0}")]
118    Application(E),
119}
120
121#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
122pub enum ServiceExit {
123    #[error("application: {0}")]
124    Application(String),
125
126    #[error("resident panicked")]
127    Panicked,
128
129    #[error("restart credit exhausted")]
130    CreditExhausted,
131
132    #[error("resident stop deadline exceeded")]
133    StopTimedOut,
134}
135
136#[derive(Clone, Debug, PartialEq, Eq)]
137pub enum ServiceStatus {
138    Starting,
139    Running { credit: usize },
140    Stopping,
141    BackingOff,
142    Stopped,
143    Inactive { cause: ServiceExit, history: Vec<String> },
144}
145
146/* === Traits === */
147
148// A resident runs until stopped. Each run is stopped and cleaned up before the
149// next restart or before the instance stops.
150pub trait Resident<A: Application>: Send + 'static {
151    type Error: fmt::Debug + fmt::Display + Send + Sync + 'static;
152
153    fn run(&mut self, context: Service<A>) -> impl Future<Output = std::result::Result<(), Fault<Self::Error>>> + Send;
154
155    fn stop(&mut self) -> impl Future<Output = std::result::Result<(), Self::Error>> + Send {
156        async { Ok(()) }
157    }
158
159    fn cleanup(&mut self) -> impl Future<Output = std::result::Result<(), Self::Error>> + Send;
160}
161
162/* === Implementations === */
163
164impl ServiceDefinition {
165    pub fn new(id: impl Into<String>) -> Self {
166        Self { id: id.into(), policy: POLICY }
167    }
168}
169
170impl<A: Application> Handles<A> {
171    pub(crate) fn context(
172        &self, operations: OperationManager, stopping: watch::Receiver<bool>, ready: watch::Sender<bool>,
173        #[cfg(feature = "plugin-sidecar")] provider: Arc<crate::supervision::instance::Instance>,
174    ) -> Service<A> {
175        Service {
176            state: self.state.clone(),
177            records: self.records.clone(),
178            operations,
179            streams: self.streams.clone(),
180            evidence: self.evidence.clone(),
181            stopping,
182            ready,
183            #[cfg(feature = "plugin-sidecar")]
184            provider,
185        }
186    }
187}
188
189impl<A: Application> Clone for Handles<A> {
190    fn clone(&self) -> Self {
191        Self {
192            storage: self.storage.clone(),
193            state: self.state.clone(),
194            records: self.records.clone(),
195            streams: self.streams.clone(),
196            evidence: self.evidence.clone(),
197        }
198    }
199}
200
201impl<A: Application> Service<A> {
202    #[cfg(feature = "plugin-sidecar")]
203    pub(crate) fn shared_attempt(&self) -> Self {
204        Self {
205            state: self.state.clone(),
206            records: self.records.clone(),
207            operations: self.operations.clone(),
208            streams: self.streams.clone(),
209            evidence: self.evidence.clone(),
210            stopping: self.stopping.clone(),
211            ready: self.ready.clone(),
212            provider: self.provider.clone(),
213        }
214    }
215
216    pub fn ready(&self) {
217        if !self.is_stopping() {
218            self.ready.send_replace(true);
219        }
220    }
221
222    // An owned host handle for calls made while this context is borrowed,
223    // such as inside `until_stopping`.
224    pub fn host(&self) -> OperationManager {
225        self.operations.clone()
226    }
227
228    pub fn records(&self) -> &crate::records::RecordManager<A> {
229        &self.records
230    }
231
232    pub fn state(&self) -> &StateManager<A> {
233        &self.state
234    }
235
236    pub fn streams(&self) -> &StreamManager<A> {
237        &self.streams
238    }
239
240    pub fn evidence(&self) -> &Evidence<A> {
241        &self.evidence
242    }
243
244    pub fn is_stopping(&self) -> bool {
245        *self.stopping.borrow()
246    }
247
248    pub async fn stopping(&mut self) {
249        while !*self.stopping.borrow_and_update() {
250            if self.stopping.changed().await.is_err() {
251                break;
252            }
253        }
254    }
255
256    pub async fn rest(&mut self, duration: Duration) -> bool {
257        self.until_stopping(executor::sleep(duration)).await.is_none()
258    }
259
260    pub async fn until_stopping<T>(&mut self, future: impl Future<Output = T>) -> Option<T> {
261        if self.is_stopping() {
262            return None;
263        }
264        let mut future = Box::pin(future);
265        let mut stop = Box::pin(self.stopping.changed());
266
267        std::future::poll_fn(|context| {
268            if let Poll::Ready(Ok(()) | Err(_)) = stop.as_mut().poll(context) {
269                return Poll::Ready(None);
270            }
271            future.as_mut().poll(context).map(Some)
272        })
273        .await
274    }
275}
276
277impl<A: Application> harmos_sdk::operations::Host for Service<A> {
278    type Error = Error;
279    type Instance<O: OperationDeclaration> = OperationInstance<O>;
280
281    fn context(&self) -> &CallContext {
282        harmos_sdk::operations::Host::context(&self.operations)
283    }
284
285    fn with_value<T: ContextValue>(&self, value: T) -> std::result::Result<Self, ContextError> {
286        let operations = harmos_sdk::operations::Host::with_value(&self.operations, value)?;
287        let mut child = self.clone_context();
288        child.operations = operations;
289        Ok(child)
290    }
291
292    fn submit<O: OperationDeclaration>(&self, request: O) -> impl Future<Output = Result<OperationInstance<O>>> + Send {
293        harmos_sdk::operations::Host::submit(&self.operations, request)
294    }
295
296    fn call<O: OperationDeclaration>(&self, request: O) -> impl Future<Output = Result<Response<O::Response>>> + Send {
297        harmos_sdk::operations::Host::call(&self.operations, request)
298    }
299
300    fn submit_with_resources<O: OperationDeclaration<Response = ()>, R: Send + Sync + 'static>(
301        &self, request: O, resources: Arc<R>,
302    ) -> impl Future<Output = Result<OperationInstance<O>>> + Send {
303        harmos_sdk::operations::Host::submit_with_resources::<O, R>(&self.operations, request, resources)
304    }
305}
306
307impl<A: Application> Service<A> {
308    fn clone_context(&self) -> Self {
309        Self {
310            state: self.state.clone(),
311            records: self.records.clone(),
312            operations: self.operations.clone(),
313            streams: self.streams.clone(),
314            evidence: self.evidence.clone(),
315            stopping: self.stopping.clone(),
316            ready: self.ready.clone(),
317            #[cfg(feature = "plugin-sidecar")]
318            provider: self.provider.clone(),
319        }
320    }
321}
322
323impl<A, F, Fut, E> Resident<A> for F
324where
325    A: Application,
326    F: FnMut(Service<A>) -> Fut + Send + 'static,
327    Fut: Future<Output = std::result::Result<(), Fault<E>>> + Send + 'static,
328    E: fmt::Debug + fmt::Display + Send + Sync + 'static,
329{
330    type Error = E;
331
332    fn run(&mut self, context: Service<A>) -> impl Future<Output = std::result::Result<(), Fault<Self::Error>>> + Send {
333        self(context)
334    }
335
336    // Closures must not own asynchronous resources beyond their run future.
337    // Resource-owning services implement Resident and retain their cleanup state.
338    async fn cleanup(&mut self) -> std::result::Result<(), Self::Error> {
339        Ok(())
340    }
341}
342
343/* === Public Functions === */
344
345/* === Private Functions === */
346
347pub(crate) fn validate(definition: &ServiceDefinition) -> Result<()> {
348    if definition.id.is_empty() {
349        return Err(Error::InvalidInput { field: "service name", reason: "must not be empty" });
350    }
351    if definition.id == crate::evidence::SOURCE {
352        return Err(Error::InvalidInput { field: "service name", reason: "artifact is reserved for evidence facts" });
353    }
354    definition.policy.validate().map_err(|source| Error::OperationDeclaration { source: source.into() })
355}
356
357/* === Tests === */
harmos_runtime/services/mod.rs

harmos_runtime/services/
mod.rs
