harmos_sidecar/kit/
observer.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// An observer is a resident body with the shared lifecycle verbs: it
11// initializes (retrying each cadence until it succeeds), observes once per
12// declared interval for as long as its activation runs, and stops and cleans
13// up when that activation is cancelled. `Activity` is that body for one
14// activation; whoever runs the activation owns its lifecycle.
15
16/* === Modules === */
17
18mod attachment;
19
20/* === Re-Exports === */
21
22pub use attachment::ObserverAttachment;
23pub use harmos_domain::inventory::observers::{Activation, DemandKey};
24
25/* === Imports === */
26
27// Standard Library Imports
28use std::any::Any;
29use std::future::Future;
30use std::panic::AssertUnwindSafe;
31use std::pin::Pin;
32use std::sync::Arc;
33use std::time::Duration;
34// Third Party Imports
35use futures_util::FutureExt;
36use tokio::time::{Instant, MissedTickBehavior, interval};
37// Local Imports
38use super::identity::{ActivationId, Identity};
39use super::{ObserverCaller, ObserverHost, ObserverPublisher, ObserverStreamPublisher, PublicationError, Supervisor};
40use harmos_domain::inventory::publications;
41
42/* === Constants === */
43
44/* === Macros === */
45
46/* === Type Aliases === */
47
48type ContextFactory<H> = Arc<dyn Fn(&H) -> Arc<dyn Any + Send + Sync> + Send + Sync>;
49type Open<R, H> = fn(Arc<R>, H, Option<ContextFactory<H>>, Identity) -> Result<Activity, ObserverError>;
50type Step<'a> = Pin<Box<dyn Future<Output = Result<(), ObserverError>> + Send + 'a>>;
51
52/* === Structs === */
53
54pub struct Declarations<R, H = Supervisor>(std::marker::PhantomData<(R, H)>);
55
56pub struct Inventory<H = Supervisor>(std::marker::PhantomData<H>);
57
58pub struct Outputs<H = Supervisor>(std::marker::PhantomData<H>);
59
60#[derive(Clone, Copy, Debug, PartialEq, Eq)]
61pub struct Tick {
62    ordinal: u64,
63    elapsed: Duration,
64}
65
66pub struct ObserverContext<'a, R, H = Supervisor> {
67    identity: Identity,
68    resources: &'a R,
69    tick: &'a Tick,
70    host: &'a H,
71    streams: &'static [publications::Entry],
72    extension: Option<&'a (dyn Any + Send + Sync)>,
73}
74
75pub struct Instance<R, H = Supervisor> {
76    entry: harmos_domain::inventory::observers::Entry,
77    open: Open<R, H>,
78    context: Option<ContextFactory<H>>,
79}
80
81// One activation of one observer: its resources, bound host and, once
82// initialized, the observer itself. Dropping an `observe` future interrupts the
83// activation; `stop` and `cleanup` then release whatever it acquired.
84pub struct Activity {
85    identity: Identity,
86    body: Box<dyn Body>,
87}
88
89struct Bound<S: Observer<H>, H: Send + Sync + 'static> {
90    identity: Identity,
91    resources: Arc<S::Resources>,
92    host: H,
93    extension: Option<Arc<dyn Any + Send + Sync>>,
94    observer: Option<S>,
95    stopped: bool,
96}
97
98/* === Enums === */
99
100#[derive(Debug, thiserror::Error)]
101pub enum ObserverError {
102    #[error("the observer host or context factory panicked")]
103    BindingPanicked,
104
105    #[error("the observer activation identity could not be drawn: {0}")]
106    Identity(#[source] std::io::Error),
107
108    #[error("the observer callback panicked")]
109    CallbackPanicked,
110
111    #[error("the observer tick ordinal is exhausted")]
112    TickExhausted,
113
114    #[error("the observer refused to stop: {0}")]
115    StopFailed(#[source] Arc<dyn std::error::Error + Send + Sync>),
116
117    #[error("the observer panicked while stopping")]
118    StopPanicked,
119
120    #[error("the observer refused to clean up: {0}")]
121    CleanupFailed(#[source] Arc<dyn std::error::Error + Send + Sync>),
122
123    #[error("the observer panicked while cleaning up")]
124    CleanupPanicked,
125}
126
127/* === Traits === */
128
129trait Body: Send {
130    fn observe<'a>(&'a mut self, ready: &'a (dyn Fn() + Send + Sync)) -> Step<'a>;
131    fn stop(&mut self) -> Step<'_>;
132    fn cleanup(&mut self) -> Step<'_>;
133    fn is_initialized(&self) -> bool;
134}
135
136pub trait Observer<H: Send + Sync + 'static = Supervisor>: Sized + Send + 'static {
137    type Resources: Send + Sync + 'static;
138
139    type Error: std::error::Error + Send + Sync + 'static;
140
141    const ID: &'static str;
142
143    const DESCRIPTION: &'static str = "";
144
145    const STREAMS: &'static [publications::Entry] = &[];
146
147    const DEMAND_KEYS: &'static [DemandKey] = &[];
148
149    const ACTIVATION: Activation = Activation::Plugin;
150
151    const INTERVAL: Duration;
152
153    fn initialize(resources: &Self::Resources) -> impl Future<Output = Result<Self, Self::Error>> + Send;
154
155    fn observe(&mut self, ctx: &ObserverContext<'_, Self::Resources, H>) -> impl Future<Output = Result<(), Self::Error>> + Send;
156
157    fn stop(&mut self) -> impl Future<Output = Result<(), Self::Error>> + Send {
158        std::future::ready(Ok(()))
159    }
160
161    fn cleanup(&mut self) -> impl Future<Output = Result<(), Self::Error>> + Send {
162        std::future::ready(Ok(()))
163    }
164}
165
166/* === Implementations === */
167
168impl<R: Send + Sync + 'static, H: Send + Sync + 'static> harmos_domain::catalog::Visitor for Declarations<R, H> {
169    type Entry = fn() -> Instance<R, H>;
170}
171
172impl<R: Send + Sync + 'static, H: Send + Sync + 'static, S: Observer<H, Resources = R>> harmos_domain::catalog::Member<S>
173    for Declarations<R, H>
174{
175    const ENTRY: fn() -> Instance<R, H> = Instance::of::<S>;
176}
177
178impl<H: Send + Sync + 'static> harmos_domain::catalog::Visitor for Outputs<H> {
179    type Entry = &'static [publications::Entry];
180}
181
182impl<H: Send + Sync + 'static, S: Observer<H>> harmos_domain::catalog::Member<S> for Outputs<H> {
183    const ENTRY: &'static [publications::Entry] = S::STREAMS;
184}
185
186impl<H: Send + Sync + 'static> harmos_domain::catalog::Visitor for Inventory<H> {
187    type Entry = harmos_domain::inventory::observers::Entry;
188}
189
190impl<H: Send + Sync + 'static, S: Observer<H>> harmos_domain::catalog::Member<S> for Inventory<H> {
191    const ENTRY: harmos_domain::inventory::observers::Entry = harmos_domain::inventory::observers::Entry {
192        id: S::ID,
193        description: S::DESCRIPTION,
194        interval: S::INTERVAL,
195        activation: S::ACTIVATION,
196        streams: S::STREAMS,
197        demand_keys: S::DEMAND_KEYS,
198    };
199}
200
201impl Tick {
202    fn advance(&mut self, elapsed: Duration) -> Result<(), ObserverError> {
203        let ordinal = self.ordinal.checked_add(1).ok_or(ObserverError::TickExhausted)?;
204        self.ordinal = ordinal;
205        self.elapsed = elapsed;
206        Ok(())
207    }
208
209    pub fn ordinal(&self) -> u64 {
210        self.ordinal
211    }
212
213    pub fn elapsed(&self) -> Duration {
214        self.elapsed
215    }
216}
217
218impl<R, H> ObserverContext<'_, R, H> {
219    pub fn observer_id(&self) -> &'static str {
220        self.identity.observer
221    }
222
223    pub fn activation_id(&self) -> ActivationId {
224        self.identity.activation
225    }
226
227    pub fn resources(&self) -> &R {
228        self.resources
229    }
230
231    pub fn tick(&self) -> &Tick {
232        self.tick
233    }
234
235    pub fn host(&self) -> &H {
236        self.host
237    }
238
239    pub fn extension<T: Send + Sync + 'static>(&self) -> Option<&T> {
240        self.extension.and_then(|extension| extension.downcast_ref())
241    }
242}
243
244impl<R> ObserverContext<'_, R, Supervisor> {
245    pub fn supervisor(&self) -> &Supervisor {
246        self.host
247    }
248}
249
250impl<R: Sync, H> ObserverContext<'_, R, H> {
251    pub async fn publish<V: Send>(&self, value: V) -> Result<(), H::Error>
252    where
253        H: ObserverPublisher<R, V>,
254    {
255        if let Some(stream) = H::STREAM {
256            self.check_stream(stream.id, stream.schema)?;
257        }
258        self.host.publish(self, value).await
259    }
260
261    pub async fn publish_on<S: harmos_domain::StreamSpec>(&self, row: &S::Row) -> Result<(), H::Error>
262    where
263        S::Row: Sync,
264        H: ObserverStreamPublisher<R, S>,
265    {
266        self.check_stream(S::ID, S::VERSION)?;
267        self.host.publish_on(self, row).await
268    }
269
270    pub async fn call<C>(&self, request: H::Request) -> Result<H::Response, H::Error>
271    where
272        H: ObserverCaller<R, C>,
273    {
274        self.host.call(self, request).await
275    }
276
277    fn check_stream(&self, id: &'static str, version: u32) -> Result<(), PublicationError> {
278        if self.streams.iter().any(|entry| entry.id == id && entry.schema == version) {
279            Ok(())
280        } else {
281            Err(PublicationError::Undeclared { stream: id })
282        }
283    }
284}
285
286impl<R: Send + Sync + 'static, H: Send + Sync + 'static> Instance<R, H> {
287    pub fn of<S: Observer<H, Resources = R>>() -> Self {
288        assert!(
289            !S::ID.trim().is_empty() && !S::ID.chars().any(char::is_control),
290            "an observer identity must be nonempty and contain no control characters"
291        );
292        assert!(!S::INTERVAL.is_zero(), "an observer interval must be nonzero");
293        assert!(
294            S::ACTIVATION != Activation::Demand || !S::STREAMS.is_empty() || !S::DEMAND_KEYS.is_empty(),
295            "a demand observer must declare streams or demand keys"
296        );
297
298        for key in S::DEMAND_KEYS {
299            assert!(!S::STREAMS.iter().any(|entry| entry.id == key.name()), "demand keys must differ from publication identities");
300        }
301
302        Self { entry: <Inventory<H> as harmos_domain::catalog::Member<S>>::ENTRY, open: Activity::open::<S, H>, context: None }
303    }
304
305    pub fn with_context<T: Send + Sync + 'static>(mut self, factory: impl Fn(&H) -> T + Send + Sync + 'static) -> Self {
306        self.context = Some(Arc::new(move |host| Arc::new(factory(host))));
307        self
308    }
309
310    pub fn declaration(&self) -> harmos_domain::inventory::observers::Declaration {
311        self.entry.declaration()
312    }
313
314    pub fn id(&self) -> &'static str {
315        self.entry.id
316    }
317
318    pub fn streams(&self) -> &'static [publications::Entry] {
319        self.entry.streams
320    }
321
322    pub fn demand_keys(&self) -> &'static [DemandKey] {
323        self.entry.demand_keys
324    }
325
326    pub fn activation(&self) -> Activation {
327        self.entry.activation
328    }
329
330    pub fn interval(&self) -> Duration {
331        self.entry.interval
332    }
333
334    // Open one activation over its resources and bound host. Nothing runs
335    // until the caller drives `observe`.
336    pub fn activate(&self, resources: Arc<R>, host: H) -> Result<Activity, ObserverError> {
337        let identity = Identity::new(self.entry.id).map_err(ObserverError::Identity)?;
338        (self.open)(resources, host, self.context.clone(), identity)
339    }
340}
341
342impl Activity {
343    fn open<S: Observer<H>, H: Send + Sync + 'static>(
344        resources: Arc<S::Resources>, host: H, context: Option<ContextFactory<H>>, identity: Identity,
345    ) -> Result<Self, ObserverError> {
346        let extension = std::panic::catch_unwind(AssertUnwindSafe(|| context.map(|factory| factory(&host)))).map_err(|_| {
347            tracing::error!(
348                observer = identity.observer, activation_id = %identity.activation,
349                "observer context factory panicked; activation refused"
350            );
351            ObserverError::BindingPanicked
352        })?;
353        let bound = Bound::<S, H> { identity, resources, host, extension, observer: None, stopped: false };
354        Ok(Self { identity, body: Box::new(bound) })
355    }
356
357    pub fn observer_id(&self) -> &'static str {
358        self.identity.observer
359    }
360
361    pub fn activation_id(&self) -> ActivationId {
362        self.identity.activation
363    }
364
365    pub fn is_initialized(&self) -> bool {
366        self.body.is_initialized()
367    }
368
369    // Initialize at the declared cadence until it succeeds, report readiness,
370    // then observe once per interval. Returns only when the activation can no
371    // longer continue; dropping the future interrupts it.
372    pub async fn observe(&mut self, ready: impl Fn() + Send + Sync) -> ObserverError {
373        match self.body.observe(&ready).await {
374            Ok(()) => unreachable!("observation runs until it is interrupted or fails"),
375            Err(failure) => failure,
376        }
377    }
378
379    pub async fn stop(&mut self) -> Result<(), ObserverError> {
380        self.body.stop().await
381    }
382
383    pub async fn cleanup(&mut self) -> Result<(), ObserverError> {
384        self.body.cleanup().await
385    }
386}
387
388impl<S: Observer<H>, H: Send + Sync + 'static> Body for Bound<S, H> {
389    fn observe<'a>(&'a mut self, ready: &'a (dyn Fn() + Send + Sync)) -> Step<'a> {
390        Box::pin(async move {
391            let name = self.identity.observer;
392            let activation = self.identity.activation;
393            let mut clock = interval(S::INTERVAL);
394            clock.set_missed_tick_behavior(MissedTickBehavior::Skip);
395
396            // Initialization Owns Its Partial Acquisitions; Only Success Is Cleaned Up
397            while self.observer.is_none() {
398                clock.tick().await;
399                match AssertUnwindSafe(S::initialize(&self.resources)).catch_unwind().await {
400                    Ok(Ok(observer)) => self.observer = Some(observer),
401                    Ok(Err(error)) => tracing::warn!(
402                        observer = name, activation_id = %activation, error = %harmos_domain::failure::Report(&error),
403                        "observer initialization failed; retrying next cadence"
404                    ),
405                    Err(_) => tracing::error!(
406                        observer = name, activation_id = %activation,
407                        "observer initialization panicked; retrying next cadence"
408                    ),
409                }
410            }
411            ready();
412
413            // Observe Once Per Declared Interval
414            let opened = Instant::now();
415            clock.reset();
416            let mut tick = Tick { ordinal: 0, elapsed: Duration::ZERO };
417            let observer = self.observer.as_mut().expect("initialized observer");
418            loop {
419                clock.tick().await;
420                if let Err(exhausted) = tick.advance(opened.elapsed()) {
421                    tracing::error!(observer = name, activation_id = %activation,
422                        "observer tick ordinal exhausted; activation stopping");
423                    return Err(exhausted);
424                }
425                let context = ObserverContext {
426                    identity: self.identity,
427                    resources: self.resources.as_ref(),
428                    tick: &tick,
429                    host: &self.host,
430                    extension: self.extension.as_deref(),
431                    streams: S::STREAMS,
432                };
433                let observed = AssertUnwindSafe(async { observer.observe(&context).await }).catch_unwind().await;
434                match observed {
435                    Ok(Ok(())) => {}
436                    Ok(Err(error)) => tracing::warn!(
437                        observer = name, activation_id = %activation, tick = tick.ordinal(), error = %harmos_domain::failure::Report(&error),
438                        "observer callback refused"
439                    ),
440                    Err(_) => {
441                        tracing::error!(
442                            observer = name, activation_id = %activation, tick = tick.ordinal(),
443                            "observer callback panicked; activation stopping"
444                        );
445                        return Err(ObserverError::CallbackPanicked);
446                    }
447                }
448            }
449        })
450    }
451
452    fn stop(&mut self) -> Step<'_> {
453        Box::pin(async move {
454            let Some(observer) = self.observer.as_mut() else {
455                return Ok(());
456            };
457            if self.stopped {
458                return Ok(());
459            }
460            guarded(async { observer.stop().await }, ObserverError::StopFailed, ObserverError::StopPanicked).await?;
461            self.stopped = true;
462            Ok(())
463        })
464    }
465
466    fn cleanup(&mut self) -> Step<'_> {
467        Box::pin(async move {
468            let Some(observer) = self.observer.as_mut() else {
469                return Ok(());
470            };
471            guarded(async { observer.cleanup().await }, ObserverError::CleanupFailed, ObserverError::CleanupPanicked).await?;
472            self.observer = None;
473            Ok(())
474        })
475    }
476
477    fn is_initialized(&self) -> bool {
478        self.observer.is_some()
479    }
480}
481
482/* === Public Functions === */
483
484pub(super) fn validate_outputs<R, H>(instances: &[Instance<R, H>], grants: &[publications::Entry]) -> Result<(), String> {
485    for output in instances.iter().flat_map(|instance| instance.entry.streams) {
486        if !grants
487            .iter()
488            .any(|grant| grant.id == output.id && grant.schema == output.schema && grant.window == output.window && grant.row == output.row)
489        {
490            return Err(format!("observer output `{}` differs from the artifact publication grant", output.id));
491        }
492    }
493    Ok(())
494}
495
496/* === Private Functions === */
497
498async fn guarded<E: std::error::Error + Send + Sync + 'static>(
499    work: impl Future<Output = Result<(), E>>, refused: fn(Arc<dyn std::error::Error + Send + Sync>) -> ObserverError,
500    panicked: ObserverError,
501) -> Result<(), ObserverError> {
502    match AssertUnwindSafe(work).catch_unwind().await {
503        Ok(Ok(())) => Ok(()),
504        Ok(Err(source)) => Err(refused(Arc::new(source))),
505        Err(_) => Err(panicked),
506    }
507}
508
509/* === Tests === */
harmos_sidecar/kit/observer.rs

harmos_sidecar/kit/
observer.rs
