harmos_runtime/operations/
handler.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14pub use harmos_domain::operations::instance::Id as OperationInstanceId;
15
16/* === Imports === */
17
18// Standard Library Imports
19use std::future::Future;
20use std::pin::Pin;
21use std::sync::{
22    Arc,
23    atomic::{AtomicU64, Ordering},
24};
25use std::task::{Context as TaskContext, Poll};
26// Third Party Imports
27use tokio::sync::oneshot;
28
29// Local Imports
30use super::OperationManager;
31use super::admission::Admission;
32use super::caller::{Origin, Source};
33use crate::operations::OperationDeclaration;
34use crate::{Error, Result, operations::Scope, state::Position};
35
36/* === Constants === */
37
38static NEXT_INSTANCE: AtomicU64 = AtomicU64::new(1);
39static NEXT_GENERATION: AtomicU64 = AtomicU64::new(1);
40
41/* === Macros === */
42
43/* === Type Aliases === */
44
45/* === Structs === */
46
47#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
48pub struct OperationGeneration(u64);
49
50// Dropping an instance before its outcome is delivered requests cancellation;
51// `detach` keeps the operation running without a handle. A resident handle
52// (an observer or service activation) holds one lease on its shared instance:
53// dropping it releases that lease and the last release stops the instance.
54#[must_use = "dropping an unsettled operation instance requests its cancellation"]
55pub struct OperationInstance<O: OperationDeclaration> {
56    pub(crate) id: OperationInstanceId,
57    pub(super) generation: OperationGeneration,
58    pub(crate) outcome: Outcome<O::Response>,
59    pub(super) control: Arc<crate::supervision::instance::Instance>,
60    pub(super) released: bool,
61}
62
63pub(crate) enum Outcome<T: Send + Sync + 'static> {
64    Request(oneshot::Receiver<Result<crate::operations::Response<T>>>),
65    Resident {
66        lease: Option<super::resident::Lease>,
67        // Polled only through `&mut`; the lock makes a shared handle `Sync`.
68        settled: std::sync::Mutex<Settling<T>>,
69    },
70}
71
72pub(crate) type Settling<T> = Pin<Box<dyn Future<Output = Result<crate::operations::Response<T>>> + Send>>;
73
74#[derive(Clone)]
75pub struct Context<'a> {
76    pub(crate) id: OperationInstanceId,
77    pub(crate) attempt: usize,
78    pub(crate) scope: Scope<'a>,
79    pub(crate) operations: OperationManager,
80    pub(super) admission: Admission,
81    pub(super) source: Source,
82    pub(super) origin: Origin,
83
84    #[cfg(feature = "plugin-sidecar")]
85    pub(super) retirement: bool,
86}
87
88/* === Enums === */
89
90/* === Traits === */
91
92/* === Implementations === */
93
94impl OperationGeneration {
95    pub(super) fn fresh() -> Result<Self> {
96        NEXT_GENERATION
97            .try_update(Ordering::Relaxed, Ordering::Relaxed, |id| id.checked_add(1))
98            .map(Self)
99            .map_err(|_| Error::ResourceExhausted { resource: "operation provider generations" })
100    }
101
102    pub fn value(self) -> u64 {
103        self.0
104    }
105}
106
107impl<O: OperationDeclaration> OperationInstance<O> {
108    pub fn id(&self) -> OperationInstanceId {
109        self.id
110    }
111
112    pub fn generation(&self) -> OperationGeneration {
113        self.generation
114    }
115
116    // A resident instance stops for every lease holder.
117    pub fn cancel(&self) {
118        match &self.outcome {
119            Outcome::Resident { lease: Some(lease), .. } => lease.stop(),
120            _ => self.control.cancel(),
121        }
122    }
123
124    pub fn status(&self) -> super::OperationStatus {
125        self.control.status()
126    }
127
128    pub fn subscribe(&self) -> tokio::sync::watch::Receiver<super::OperationStatus> {
129        self.control.settlement()
130    }
131
132    // Wait until the instance runs; a resident that stops or fails first
133    // reports why once its cleanup settles.
134    pub async fn ready(&self) -> Result<()> {
135        let mut status = self.control.settlement();
136        loop {
137            let current = status.borrow_and_update().clone();
138            match current.phase() {
139                super::OperationPhase::Running => return Ok(()),
140                phase if phase.is_terminal() || current.cancellation_requested() || current.outcome().is_some() => {
141                    self.settle().await?;
142                    return Err(Error::NotReady { domain: "operation activation" });
143                }
144                _ => {}
145            }
146            if status.changed().await.is_err() {
147                return Err(Error::NotReady { domain: "operation activation" });
148            }
149        }
150    }
151
152    #[tracing::instrument(target = "harmos::operations", name = "operations.join", level = "debug", skip_all)]
153    pub async fn join(self) -> Result<crate::operations::Response<O::Response>> {
154        self.await
155    }
156
157    pub async fn cancel_and_join(self) -> Result<crate::operations::Response<O::Response>> {
158        self.cancel();
159        self.await
160    }
161
162    // Keep the operation running without a handle until its submitting
163    // attempt ends, which cancels and settles it like any other child. A
164    // detached resident lease is held until its resources retire.
165    pub fn detach(mut self) -> OperationInstanceId {
166        self.released = true;
167        if let Outcome::Resident { lease, .. } = &mut self.outcome
168            && let Some(lease) = lease.take()
169        {
170            lease.detach();
171        }
172        self.id
173    }
174
175    // Release this handle. A resident lease that was the last one stops its
176    // instance and waits for it; a request-scoped instance is cancelled and
177    // settled.
178    pub async fn release(mut self) -> Result<()> {
179        self.released = true;
180        let lease = match &mut self.outcome {
181            Outcome::Resident { lease, .. } => lease.take(),
182            Outcome::Request(_) => None,
183        };
184        match lease {
185            Some(lease) => lease.release().await,
186            None => {
187                self.control.cancel();
188                self.settle().await
189            }
190        }
191    }
192
193    // Wait for the instance to settle without consuming the handle. A
194    // resident reports its outcome and retries a retained cleanup.
195    #[tracing::instrument(target = "harmos::operations", name = "operations.settle", level = "debug", skip_all)]
196    pub async fn settle(&self) -> Result<()> {
197        if let Outcome::Resident { lease: Some(lease), .. } = &self.outcome {
198            return lease.settlement().await;
199        }
200        self.control.settle().await.map_err(|source| Error::OperationSettlement { source })
201    }
202}
203
204impl<O: OperationDeclaration> OperationInstance<O> {
205    // Encoded invocations keep their own contract: dropping one never
206    // cancels, so the typed instance they await must not cancel either.
207    pub(super) fn unguarded(mut self) -> Self {
208        self.released = true;
209        self
210    }
211}
212
213impl<O: OperationDeclaration> Future for OperationInstance<O> {
214    type Output = Result<crate::operations::Response<O::Response>>;
215    fn poll(self: Pin<&mut Self>, cx: &mut TaskContext<'_>) -> Poll<Self::Output> {
216        let instance = self.get_mut();
217        let answer = match &mut instance.outcome {
218            Outcome::Request(outcome) => match Pin::new(outcome).poll(cx) {
219                Poll::Ready(Ok(answer)) => answer,
220                Poll::Ready(Err(_)) => Err(crate::Refusal::Unavailable.into()),
221                Poll::Pending => return Poll::Pending,
222            },
223            Outcome::Resident { settled, .. } => match settled.get_mut().expect("resident settlement is healthy").as_mut().poll(cx) {
224                Poll::Ready(answer) => answer,
225                Poll::Pending => return Poll::Pending,
226            },
227        };
228        instance.released = true;
229        Poll::Ready(answer)
230    }
231}
232
233impl<O: OperationDeclaration> Drop for OperationInstance<O> {
234    fn drop(&mut self) {
235        // A resident lease releases itself when dropped.
236        if matches!(self.outcome, Outcome::Request(_)) && !self.released && !self.control.is_settled() {
237            self.control.cancel();
238        }
239    }
240}
241
242impl<O> harmos_sdk::operations::Instance<O> for OperationInstance<O>
243where
244    O: OperationDeclaration,
245{
246    type Error = Error;
247
248    fn id(&self) -> OperationInstanceId {
249        self.id
250    }
251
252    fn status(&self) -> impl Future<Output = Result<harmos_domain::operations::instance::Status>> + Send {
253        std::future::ready(Ok(self.control.status().portable()))
254    }
255
256    fn cancel(&self) -> impl Future<Output = Result<()>> + Send {
257        OperationInstance::cancel(self);
258        std::future::ready(Ok(()))
259    }
260
261    fn join(self) -> impl Future<Output = Result<crate::operations::Response<O::Response>>> + Send {
262        self
263    }
264
265    fn detach(self) -> OperationInstanceId {
266        OperationInstance::detach(self)
267    }
268}
269
270impl Context<'_> {
271    pub fn context(&self) -> &super::CallContext {
272        self.operations.call_context()
273    }
274
275    // Derive a child caller while retaining this attempt's identity, admission,
276    // cancellation and deadline. Only the selected portable value changes.
277    pub fn with_value<T: super::ContextValue>(&self, value: T) -> Result<Self> {
278        let context = self.context().with_value(value)?;
279        let mut child = self.clone();
280        child.operations = child.operations.with_context(context);
281        Ok(child)
282    }
283
284    pub fn id(&self) -> OperationInstanceId {
285        self.id
286    }
287
288    // An active instance this attempt's caller submitted; others stay hidden.
289    pub(crate) fn owned_instance(&self, id: OperationInstanceId) -> Option<Arc<crate::supervision::instance::Instance>> {
290        self.operations.instances.owned(id, &self.origin.identity)
291    }
292
293    pub fn attempt(&self) -> usize {
294        self.attempt
295    }
296
297    pub fn scope(&self) -> &Scope<'_> {
298        &self.scope
299    }
300
301    pub fn is_host_call(&self) -> bool {
302        self.origin.identity == super::caller::Identity::Host
303    }
304
305    // Revalidate The Caller Before Every Streamed Destination
306    pub(crate) fn admit_stream_grant(&self, grant: &harmos_domain::plugin::Grant) -> Result<()> {
307        self.source.check()?;
308        match &self.origin.grants {
309            Some(authority) => authority.admit(Some(grant)),
310            None if self.is_host_call() => Ok(()),
311            None => Err(Error::PermissionDenied),
312        }
313    }
314
315    // Refuse a provider-prepared grant the admitted caller does not hold.
316    #[cfg(any(all(feature = "plugin-guest", not(target_arch = "wasm32")), feature = "plugin-sidecar"))]
317    pub(super) fn admit_grant(&self, grant: Option<&harmos_domain::plugin::Grant>) -> Result<()> {
318        match (&self.origin.grants, grant) {
319            (Some(authority), grant) => authority.admit(grant),
320            (None, None) => Ok(()),
321            (None, Some(_)) => Err(Error::PermissionDenied),
322        }
323    }
324
325    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
326    pub fn caller_scope(&self) -> Option<&crate::plugin::PluginScope> {
327        self.origin.scope.as_ref()
328    }
329
330    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
331    pub(crate) fn retains_publication(&self, scope: &crate::plugin::PluginScope) -> bool {
332        self.source.check().is_ok_and(|origin| {
333            (origin.has_lifecycle_loan() || origin.resident) && origin.scope.as_ref().is_some_and(|caller| caller.same_activation(scope))
334        })
335    }
336
337    pub fn submit_at<O: OperationDeclaration>(&self, id: &str, request: O) -> Result<OperationInstance<O>> {
338        self.operations.admit_at::<O, ()>(id, (request.into(), ()), Some(&self.scope), self.admission.held(), &self.source, None)
339    }
340
341    pub fn submit<O: OperationDeclaration>(&self, request: O) -> Result<OperationInstance<O>> {
342        self.operations.submit_scoped::<O, ()>((request.into(), ()), &self.scope, &self.admission, &self.source, None)
343    }
344
345    pub fn submit_endpoints<O: OperationDeclaration, P: Send + 'static>(&self, request: O, endpoints: P) -> Result<OperationInstance<O>> {
346        self.operations.submit_scoped::<O, P>((request.into(), endpoints), &self.scope, &self.admission, &self.source, None)
347    }
348
349    pub fn submit_charged<O: OperationDeclaration, P: Send + 'static>(
350        &self, request: super::StreamItem<O>, endpoints: P,
351    ) -> Result<OperationInstance<O>> {
352        self.operations.submit_scoped::<O, P>(
353            (super::binding::Request::charged(request)?, endpoints),
354            &self.scope,
355            &self.admission,
356            &self.source,
357            None,
358        )
359    }
360
361    pub fn submit_caused<O: OperationDeclaration>(&self, request: O, cause: Position) -> Result<OperationInstance<O>> {
362        self.operations.submit_scoped::<O, ()>((request.into(), ()), &self.scope, &self.admission, &self.source, Some(cause))
363    }
364
365    pub async fn call_caused<O: OperationDeclaration>(
366        &self, request: O, cause: Position,
367    ) -> Result<crate::operations::Response<O::Response>> {
368        self.submit_caused::<O>(request, cause)?.await
369    }
370
371    pub async fn call<O: OperationDeclaration>(&self, request: O) -> Result<crate::operations::Response<O::Response>> {
372        self.submit(request)?.await
373    }
374}
375
376/* === Public Functions === */
377
378pub(crate) fn fresh_instance() -> Result<OperationInstanceId> {
379    NEXT_INSTANCE
380        .try_update(Ordering::Relaxed, Ordering::Relaxed, |id| id.checked_add(1))
381        .map(OperationInstanceId::new)
382        .map_err(|_| Error::ResourceExhausted { resource: "operation instance identities" })
383}
384
385/* === Private Functions === */
386
387/* === Tests === */
harmos_runtime/operations/handler.rs

harmos_runtime/operations/
handler.rs
