harmos_sidecar/kit/observer/
attachment.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// Serves a plugin's observers through its one observe control. Each host
11// submission names an observer and runs it for as long as the invocation
12// lasts: the attempt binds, initializes and ticks the observer on a thread of
13// its own, and the invocation's cleanup stops and cleans it up once the host
14// cancels it.
15
16/* === Modules === */
17
18/* === Re-Exports === */
19
20/* === Imports === */
21
22// Standard Library Imports
23use std::collections::BTreeMap;
24use std::future::Future;
25use std::pin::Pin;
26use std::sync::Arc;
27use std::sync::atomic::{AtomicBool, Ordering};
28// Third Party Imports
29use tokio::sync::oneshot;
30// Local Imports
31use super::{Activity, Instance, ObserverError, ObserverHost};
32use crate::{Refusal, Supervisor};
33use harmos_domain::observers::Control;
34use harmos_sdk::operations::{Operation, ServingError, WireTable};
35use harmos_sdk::plugin::Attachment;
36
37/* === Constants === */
38
39/* === Macros === */
40
41/* === Type Aliases === */
42
43/* === Structs === */
44
45pub struct ObserverAttachment<R, H = Supervisor> {
46    resources: Arc<R>,
47    declarations: Vec<harmos_domain::inventory::observers::Declaration>,
48    instances: BTreeMap<&'static str, Instance<R, H>>,
49    closing: AtomicBool,
50}
51
52struct Observe<R, H> {
53    attachment: Arc<ObserverAttachment<R, H>>,
54    running: Option<Running>,
55    activity: Option<Activity>,
56}
57
58struct Interrupt<'a>(&'a mut Option<oneshot::Sender<()>>);
59
60// One activation observing on its own thread until told to stop; joining it
61// hands the activity back for stopping and cleanup.
62struct Running {
63    stop: Option<oneshot::Sender<()>>,
64    failed: Option<oneshot::Receiver<ObserverError>>,
65    returned: Option<oneshot::Receiver<Activity>>,
66}
67
68/* === Enums === */
69
70/* === Traits === */
71
72/* === Implementations === */
73
74impl<R: Send + Sync + 'static, H: ObserverHost<R>> ObserverAttachment<R, H> {
75    pub fn new(
76        instances: Vec<Instance<R, H>>, resources: Arc<R>, grants: &[harmos_domain::inventory::publications::Entry],
77    ) -> Result<Arc<Self>, Refusal> {
78        super::validate_outputs(&instances, grants).map_err(Refusal::of)?;
79        let declarations: Vec<_> = instances.iter().map(Instance::declaration).collect();
80        let mut selected = BTreeMap::new();
81        for instance in instances {
82            if instance.id().is_empty() || selected.contains_key(instance.id()) {
83                return Err(Refusal::of("observer identities must be nonempty and unique"));
84            }
85            selected.insert(instance.id(), instance);
86        }
87        Ok(Arc::new(Self { resources, declarations, instances: selected, closing: AtomicBool::new(false) }))
88    }
89
90    pub fn from_catalog<C>(resources: Arc<R>, grants: &[harmos_domain::inventory::publications::Entry]) -> Result<Arc<Self>, Refusal>
91    where
92        C: harmos_domain::catalog::Catalog<super::Declarations<R, H>>,
93    {
94        use harmos_domain::catalog::Projection;
95        let mut instances = Vec::new();
96        C::ENTRIES
97            .try_for_each(|prepare| {
98                instances.push(prepare());
99                Ok::<(), std::convert::Infallible>(())
100            })
101            .unwrap();
102        Self::new(instances, resources, grants)
103    }
104
105    // Bind the built-in observe control the host submits to run an observer.
106    pub fn bind(self: &Arc<Self>, table: &mut WireTable<Supervisor>) -> Result<(), ServingError> {
107        table.bind_observe_control::<Observe<R, H>>(Arc::clone(self))
108    }
109}
110
111impl<R: Send + Sync + 'static, H: ObserverHost<R>> Attachment<Supervisor> for ObserverAttachment<R, H> {
112    fn observers(&self) -> Vec<harmos_domain::inventory::observers::Declaration> {
113        self.declarations.clone()
114    }
115
116    fn resources(&self) -> Arc<dyn std::any::Any + Send + Sync> {
117        self.resources.clone()
118    }
119
120    // Running activations are invocations the host cancels and cleans up;
121    // quiescing only refuses new ones.
122    fn quiesce(&self) -> Pin<Box<dyn Future<Output = Result<(), ServingError>> + Send + '_>> {
123        self.closing.store(true, Ordering::Release);
124        Box::pin(std::future::ready(Ok(())))
125    }
126}
127
128impl<R, H> Operation for Observe<R, H>
129where
130    R: Send + Sync + 'static,
131    H: ObserverHost<R>,
132{
133    type Declaration = Control;
134    type Resources = ObserverAttachment<R, H>;
135    type Error = Refusal;
136    type Endpoints = ();
137    type Context<'a> = &'a Supervisor;
138
139    async fn initialize<'ctx>(attachment: Arc<Self::Resources>, _: Self::Context<'ctx>) -> Result<Self, Refusal> {
140        Ok(Self { attachment, running: None, activity: None })
141    }
142
143    async fn run<'ctx>(
144        &mut self, host: Self::Context<'ctx>, control: &mut Control, _endpoints: &mut Self::Endpoints,
145    ) -> Result<crate::operations::Response<()>, Refusal> {
146        if self.running.is_none() {
147            if self.attachment.closing.load(Ordering::Acquire) {
148                return Err(Refusal::of("observer admission is closing"));
149            }
150            let name = self.attachment.instances.get(control.observer.as_str()).ok_or_else(|| Refusal::of("unknown observer"))?.id();
151            self.running = Some(Running::spawn(Arc::clone(&self.attachment), name, host.clone())?);
152        }
153        let running = self.running.as_mut().expect("spawned activation");
154        let failed = running.failed.as_mut();
155        // Cancelling This Attempt Interrupts The Observing Thread
156        let _interrupt = Interrupt(&mut running.stop);
157        let failure = match failed {
158            Some(failed) => failed.await.ok(),
159            None => None,
160        };
161        running.failed.take();
162        match failure {
163            Some(failure) => Err(Refusal::of(failure)),
164            None => Err(Refusal::of("observer activation ended")),
165        }
166    }
167
168    async fn stop(&mut self) -> Result<(), Refusal> {
169        let failure = if let Some(mut running) = self.running.take() {
170            let (activity, failure) = running.join().await;
171            self.activity = activity;
172            failure
173        } else {
174            None
175        };
176        let stopped = match self.activity.as_mut() {
177            Some(activity) => activity.stop().await.map_err(Refusal::of),
178            None => Ok(()),
179        };
180        match failure {
181            Some(failure) => Err(Refusal::of(failure)),
182            None => stopped,
183        }
184    }
185
186    async fn cleanup(&mut self) -> Result<(), Refusal> {
187        match self.activity.as_mut() {
188            Some(activity) => activity.cleanup().await.map_err(Refusal::of),
189            None => Ok(()),
190        }
191    }
192}
193
194impl Drop for Interrupt<'_> {
195    fn drop(&mut self) {
196        if let Some(stop) = self.0.take() {
197            let _ = stop.send(());
198        }
199    }
200}
201
202impl Running {
203    fn spawn<R: Send + Sync + 'static, H: ObserverHost<R>>(
204        attachment: Arc<ObserverAttachment<R, H>>, observer: &'static str, host: Supervisor,
205    ) -> Result<Self, Refusal> {
206        let (stop, stopped) = oneshot::channel();
207        let (failing, failed) = oneshot::channel();
208        let (returning, returned) = oneshot::channel();
209        crate::kit::task::spawn(observer.to_owned(), async move {
210            let Some(instance) = attachment.instances.get(observer) else {
211                return;
212            };
213            let resources = Arc::clone(&attachment.resources);
214            // Binding Happens On The Activation's Own Thread
215            let bound = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| H::bind(Arc::clone(&resources), host)));
216            let activated = bound.map_err(|_| ObserverError::BindingPanicked).and_then(|bound| instance.activate(resources, bound));
217            // The Activity Owns The Activation Now. Release this thread's
218            // extra attachment before waking either completion receiver.
219            drop(attachment);
220            let mut activity = match activated {
221                Ok(activity) => activity,
222                Err(failure) => {
223                    let _ = failing.send(failure);
224                    return;
225                }
226            };
227            tokio::select! {
228                failure = activity.observe(|| {}) => {
229                    let _ = failing.send(failure);
230                }
231                _ = stopped => {}
232            }
233            let _ = returning.send(activity);
234        })
235        .map_err(Refusal::of)?;
236        Ok(Self { stop: Some(stop), failed: Some(failed), returned: Some(returned) })
237    }
238
239    // Interrupt the activation and take its activity back.
240    async fn join(&mut self) -> (Option<Activity>, Option<ObserverError>) {
241        if let Some(stop) = self.stop.take() {
242            let _ = stop.send(());
243        }
244        let activity = match self.returned.take() {
245            Some(returned) => returned.await.ok(),
246            None => None,
247        };
248        // Joining the activation also settles its failure sender. Cancellation
249        // of run must not discard an error that shutdown still needs to report.
250        let failure = self.failed.take().and_then(|mut failed| failed.try_recv().ok());
251        (activity, failure)
252    }
253}
254
255/* === Public Functions === */
256
257/* === Private Functions === */
258
259/* === Tests === */
harmos_sidecar/kit/observer/attachment.rs

harmos_sidecar/kit/observer/
attachment.rs
