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

harmos_sidecar/kit/observer/
attachment.rs
