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<H> = fn(Arc<harmos_sdk::ExecutionContext>, 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<H = Supervisor>(std::marker::PhantomData<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, H = Supervisor> {
67    identity: Identity,
68    resources: &'a harmos_sdk::ExecutionContext,
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<H = Supervisor> {
76    entry: harmos_domain::inventory::observers::Entry,
77    open: Open<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<harmos_sdk::ExecutionContext>,
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 Error: std::error::Error + Send + Sync + 'static;
138
139    const ID: &'static str;
140
141    const DESCRIPTION: &'static str = "";
142
143    const STREAMS: &'static [publications::Entry] = &[];
144
145    const DEMAND_KEYS: &'static [DemandKey] = &[];
146
147    const ACTIVATION: Activation = Activation::Plugin;
148
149    const INTERVAL: Duration;
150
151    fn initialize(resources: &harmos_sdk::ExecutionContext) -> impl Future<Output = Result<Self, Self::Error>> + Send;
152
153    fn observe(&mut self, ctx: &ObserverContext<'_, H>) -> impl Future<Output = Result<(), Self::Error>> + Send;
154
155    fn stop(&mut self) -> impl Future<Output = Result<(), Self::Error>> + Send {
156        std::future::ready(Ok(()))
157    }
158
159    fn cleanup(&mut self) -> impl Future<Output = Result<(), Self::Error>> + Send {
160        std::future::ready(Ok(()))
161    }
162}
163
164/* === Implementations === */
165
166impl<H: Send + Sync + 'static> harmos_domain::catalog::Visitor for Declarations<H> {
167    type Entry = fn() -> Instance<H>;
168}
169
170impl<H: Send + Sync + 'static, S: Observer<H>> harmos_domain::catalog::Member<S> for Declarations<H> {
171    const ENTRY: fn() -> Instance<H> = Instance::of::<S>;
172}
173
174impl<H: Send + Sync + 'static> harmos_domain::catalog::Visitor for Outputs<H> {
175    type Entry = &'static [publications::Entry];
176}
177
178impl<H: Send + Sync + 'static, S: Observer<H>> harmos_domain::catalog::Member<S> for Outputs<H> {
179    const ENTRY: &'static [publications::Entry] = S::STREAMS;
180}
181
182impl<H: Send + Sync + 'static> harmos_domain::catalog::Visitor for Inventory<H> {
183    type Entry = harmos_domain::inventory::observers::Entry;
184}
185
186impl<H: Send + Sync + 'static, S: Observer<H>> harmos_domain::catalog::Member<S> for Inventory<H> {
187    const ENTRY: harmos_domain::inventory::observers::Entry = harmos_domain::inventory::observers::Entry {
188        id: S::ID,
189        description: S::DESCRIPTION,
190        interval: S::INTERVAL,
191        activation: S::ACTIVATION,
192        streams: S::STREAMS,
193        demand_keys: S::DEMAND_KEYS,
194    };
195}
196
197impl Tick {
198    fn advance(&mut self, elapsed: Duration) -> Result<(), ObserverError> {
199        let ordinal = self.ordinal.checked_add(1).ok_or(ObserverError::TickExhausted)?;
200        self.ordinal = ordinal;
201        self.elapsed = elapsed;
202        Ok(())
203    }
204
205    pub fn ordinal(&self) -> u64 {
206        self.ordinal
207    }
208
209    pub fn elapsed(&self) -> Duration {
210        self.elapsed
211    }
212}
213
214impl<H> ObserverContext<'_, H> {
215    pub fn observer_id(&self) -> &'static str {
216        self.identity.observer
217    }
218
219    pub fn activation_id(&self) -> ActivationId {
220        self.identity.activation
221    }
222
223    pub fn resource<T: Send + Sync + 'static>(&self) -> Result<Arc<T>, harmos_sdk::operations::ResourceError> {
224        self.resources.resource::<T>()
225    }
226
227    pub fn tick(&self) -> &Tick {
228        self.tick
229    }
230
231    pub fn host(&self) -> &H {
232        self.host
233    }
234
235    pub fn extension<T: Send + Sync + 'static>(&self) -> Option<&T> {
236        let extension = self.extension?;
237        extension.downcast_ref()
238    }
239}
240
241impl ObserverContext<'_, Supervisor> {
242    pub fn supervisor(&self) -> &Supervisor {
243        self.host
244    }
245}
246
247impl<H> ObserverContext<'_, H> {
248    pub async fn publish<V: Send>(&self, value: V) -> Result<(), H::Error>
249    where
250        H: ObserverPublisher<V>,
251    {
252        if let Some(stream) = H::STREAM {
253            self.check_stream(stream.id, stream.schema)?;
254        }
255        self.host.publish(self, value).await
256    }
257
258    pub async fn publish_on<S: harmos_domain::StreamSpec>(&self, row: &S::Row) -> Result<(), H::Error>
259    where
260        S::Row: Sync,
261        H: ObserverStreamPublisher<S>,
262    {
263        self.check_stream(S::ID, S::VERSION)?;
264        self.host.publish_on(self, row).await
265    }
266
267    pub async fn call<C>(&self, request: H::Request) -> Result<H::Response, H::Error>
268    where
269        H: ObserverCaller<C>,
270    {
271        self.host.call(self, request).await
272    }
273
274    fn check_stream(&self, id: &'static str, version: u32) -> Result<(), PublicationError> {
275        if self.streams.iter().any(|entry| entry.id == id && entry.schema == version) {
276            Ok(())
277        } else {
278            Err(PublicationError::Undeclared { stream: id })
279        }
280    }
281}
282
283impl<H: Send + Sync + 'static> Instance<H> {
284    pub fn of<S: Observer<H>>() -> Self {
285        assert!(
286            !S::ID.trim().is_empty() && !S::ID.chars().any(char::is_control),
287            "an observer identity must be nonempty and contain no control characters"
288        );
289        assert!(!S::INTERVAL.is_zero(), "an observer interval must be nonzero");
290        assert!(
291            S::ACTIVATION != Activation::Demand || !S::STREAMS.is_empty() || !S::DEMAND_KEYS.is_empty(),
292            "a demand observer must declare streams or demand keys"
293        );
294
295        for key in S::DEMAND_KEYS {
296            assert!(!S::STREAMS.iter().any(|entry| entry.id == key.name()), "demand keys must differ from publication identities");
297        }
298
299        Self {
300            entry: <Inventory<H> as harmos_domain::catalog::Member<S>>::ENTRY,
301            open: Activity::open::<S, H>,
302            context: None,
303        }
304    }
305
306    pub fn with_context<T: Send + Sync + 'static>(mut self, factory: impl Fn(&H) -> T + Send + Sync + 'static) -> Self {
307        let erased = move |host: &H| -> Arc<dyn Any + Send + Sync> {
308            let extension = factory(host);
309            Arc::new(extension)
310        };
311        self.context = Some(Arc::new(erased));
312        self
313    }
314
315    pub fn declaration(&self) -> harmos_domain::inventory::observers::Declaration {
316        self.entry.declaration()
317    }
318
319    pub fn id(&self) -> &'static str {
320        self.entry.id
321    }
322
323    pub fn streams(&self) -> &'static [publications::Entry] {
324        self.entry.streams
325    }
326
327    pub fn demand_keys(&self) -> &'static [DemandKey] {
328        self.entry.demand_keys
329    }
330
331    pub fn activation(&self) -> Activation {
332        self.entry.activation
333    }
334
335    pub fn interval(&self) -> Duration {
336        self.entry.interval
337    }
338
339    // Open one activation over its resources and bound host. Nothing runs
340    // until the caller drives `observe`.
341    pub fn activate(&self, resources: Arc<harmos_sdk::ExecutionContext>, host: H) -> Result<Activity, ObserverError> {
342        let identity = Identity::new(self.entry.id).map_err(ObserverError::Identity)?;
343        (self.open)(resources, host, self.context.clone(), identity)
344    }
345}
346
347impl Activity {
348    fn open<S: Observer<H>, H: Send + Sync + 'static>(
349        resources: Arc<harmos_sdk::ExecutionContext>, host: H, context: Option<ContextFactory<H>>, identity: Identity,
350    ) -> Result<Self, ObserverError> {
351        let create_extension = || match context {
352            Some(factory) => {
353                let extension = factory(&host);
354                Some(extension)
355            }
356            None => None,
357        };
358        let guarded_extension = AssertUnwindSafe(create_extension);
359        let extension = std::panic::catch_unwind(guarded_extension).map_err(|_| {
360            tracing::error!(
361                observer = identity.observer, activation_id = %identity.activation,
362                "observer context factory panicked; activation refused"
363            );
364            ObserverError::BindingPanicked
365        })?;
366        let bound = Bound::<S, H> {
367            identity,
368            resources,
369            host,
370            extension,
371            observer: None,
372            stopped: false,
373        };
374        Ok(Self { identity, body: Box::new(bound) })
375    }
376
377    pub fn observer_id(&self) -> &'static str {
378        self.identity.observer
379    }
380
381    pub fn activation_id(&self) -> ActivationId {
382        self.identity.activation
383    }
384
385    pub fn is_initialized(&self) -> bool {
386        self.body.is_initialized()
387    }
388
389    // Initialize at the declared cadence until it succeeds, report readiness,
390    // then observe once per interval. Returns only when the activation can no
391    // longer continue; dropping the future interrupts it.
392    pub async fn observe(&mut self, ready: impl Fn() + Send + Sync) -> ObserverError {
393        match self.body.observe(&ready).await {
394            Ok(()) => unreachable!("observation runs until it is interrupted or fails"),
395            Err(failure) => failure,
396        }
397    }
398
399    pub async fn stop(&mut self) -> Result<(), ObserverError> {
400        self.body.stop().await
401    }
402
403    pub async fn cleanup(&mut self) -> Result<(), ObserverError> {
404        self.body.cleanup().await
405    }
406}
407
408impl<S: Observer<H>, H: Send + Sync + 'static> Body for Bound<S, H> {
409    fn observe<'a>(&'a mut self, ready: &'a (dyn Fn() + Send + Sync)) -> Step<'a> {
410        Box::pin(async move {
411            let name = self.identity.observer;
412            let activation = self.identity.activation;
413            let mut clock = interval(S::INTERVAL);
414            clock.set_missed_tick_behavior(MissedTickBehavior::Skip);
415
416            // Initialization Owns Its Partial Acquisitions; Only Success Is Cleaned Up
417            while self.observer.is_none() {
418                clock.tick().await;
419                match AssertUnwindSafe(S::initialize(&self.resources)).catch_unwind().await {
420                    Ok(Ok(observer)) => self.observer = Some(observer),
421                    Ok(Err(error)) => tracing::warn!(
422                        observer = name, activation_id = %activation, error = %harmos_domain::failure::Report(&error),
423                        "observer initialization failed; retrying next cadence"
424                    ),
425                    Err(_) => tracing::error!(
426                        observer = name, activation_id = %activation,
427                        "observer initialization panicked; retrying next cadence"
428                    ),
429                }
430            }
431            ready();
432
433            let opened = Instant::now();
434            clock.reset();
435            let mut tick = Tick { ordinal: 0, elapsed: Duration::ZERO };
436            let observer = self.observer.as_mut().expect("initialized observer");
437            loop {
438                clock.tick().await;
439                advance_tick(&mut tick, opened.elapsed(), self.identity)?;
440                let context = ObserverContext {
441                    identity: self.identity,
442                    resources: self.resources.as_ref(),
443                    tick: &tick,
444                    host: &self.host,
445                    extension: self.extension.as_deref(),
446                    streams: S::STREAMS,
447                };
448                let observed = AssertUnwindSafe(async { observer.observe(&context).await }).catch_unwind().await;
449                observed_result(observed, self.identity, tick.ordinal())?;
450            }
451        })
452    }
453
454    fn stop(&mut self) -> Step<'_> {
455        Box::pin(async move {
456            let Some(observer) = self.observer.as_mut() else {
457                return Ok(());
458            };
459            if self.stopped {
460                return Ok(());
461            }
462            guarded(async { observer.stop().await }, ObserverError::StopFailed, ObserverError::StopPanicked).await?;
463            self.stopped = true;
464            Ok(())
465        })
466    }
467
468    fn cleanup(&mut self) -> Step<'_> {
469        Box::pin(async move {
470            let Some(observer) = self.observer.as_mut() else {
471                return Ok(());
472            };
473            guarded(async { observer.cleanup().await }, ObserverError::CleanupFailed, ObserverError::CleanupPanicked).await?;
474            self.observer = None;
475            Ok(())
476        })
477    }
478
479    fn is_initialized(&self) -> bool {
480        self.observer.is_some()
481    }
482}
483
484/* === Public Functions === */
485
486pub(super) fn validate_outputs<H>(instances: &[Instance<H>], grants: &[publications::Entry]) -> Result<(), String> {
487    let outputs = instances.iter().flat_map(|instance| instance.entry.streams);
488    let missing = |output: &&publications::Entry| !matches_output(output, grants);
489    let mut invalid = outputs.filter(missing);
490    match invalid.next() {
491        Some(output) => {
492            let message = format!("observer output `{}` differs from the artifact publication grant", output.id);
493            Err(message)
494        }
495        None => Ok(()),
496    }
497}
498
499/* === Private Functions === */
500
501fn observed_result<E: std::error::Error + 'static>(
502    observed: Result<Result<(), E>, Box<dyn Any + Send>>, identity: Identity, tick: u64,
503) -> Result<(), ObserverError> {
504    match observed {
505        Ok(Ok(())) => Ok(()),
506        Ok(Err(error)) => {
507            tracing::warn!(observer = identity.observer, activation_id = %identity.activation,
508                tick, error = %harmos_domain::failure::Report(&error), "observer callback refused");
509            Ok(())
510        }
511        Err(_) => {
512            tracing::error!(observer = identity.observer, activation_id = %identity.activation,
513                tick, "observer callback panicked; activation stopping");
514            Err(ObserverError::CallbackPanicked)
515        }
516    }
517}
518
519fn matches_output(output: &publications::Entry, grants: &[publications::Entry]) -> bool {
520    grants
521        .iter()
522        .any(|grant| grant.id == output.id && grant.schema == output.schema && grant.window == output.window && grant.row == output.row)
523}
524
525fn advance_tick(tick: &mut Tick, elapsed: Duration, identity: Identity) -> Result<(), ObserverError> {
526    let result = tick.advance(elapsed);
527    if result.is_err() {
528        tracing::error!(observer = identity.observer, activation_id = %identity.activation,
529            "observer tick ordinal exhausted; activation stopping");
530    }
531    result
532}
533
534async fn guarded<E: std::error::Error + Send + Sync + 'static>(
535    work: impl Future<Output = Result<(), E>>, refused: fn(Arc<dyn std::error::Error + Send + Sync>) -> ObserverError,
536    panicked: ObserverError,
537) -> Result<(), ObserverError> {
538    match AssertUnwindSafe(work).catch_unwind().await {
539        Ok(Ok(())) => Ok(()),
540        Ok(Err(source)) => {
541            let source = Arc::new(source);
542            let error = refused(source);
543            Err(error)
544        }
545        Err(_) => Err(panicked),
546    }
547}
548
549/* === Tests === */
550
551#[cfg(test)]
552mod tests {
553    use super::{Identity, ObserverContext, Tick};
554    use std::sync::Arc;
555    use std::time::Duration;
556
557    #[test]
558    fn context_resource_resolution() {
559        let value = Arc::new(17_u32);
560        let mut table = harmos_sdk::__private::WireTable::new("observer").unwrap();
561        table.provide_resource(Arc::clone(&value)).unwrap();
562        let resources = table.context();
563        let tick = Tick { ordinal: 1, elapsed: Duration::ZERO };
564        let context = ObserverContext {
565            identity: Identity::new("resource_observer").unwrap(),
566            resources: &resources,
567            tick: &tick,
568            host: &(),
569            streams: &[],
570            extension: None,
571        };
572
573        let resolved = context.resource::<u32>().unwrap();
574        assert!(Arc::ptr_eq(&value, &resolved));
575        assert!(context.resource::<u64>().is_err());
576    }
577}
harmos_sidecar/kit/observer.rs

harmos_sidecar/kit/
observer.rs
