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    pub async fn join(self) -> Result<crate::operations::Response<O::Response>> {
153        self.await
154    }
155
156    pub async fn cancel_and_join(self) -> Result<crate::operations::Response<O::Response>> {
157        self.cancel();
158        self.await
159    }
160
161    // Keep the operation running without a handle until its submitting
162    // attempt ends, which cancels and settles it like any other child. A
163    // detached resident lease is held until its resources retire.
164    pub fn detach(mut self) -> OperationInstanceId {
165        self.released = true;
166        if let Outcome::Resident { lease, .. } = &mut self.outcome
167            && let Some(lease) = lease.take()
168        {
169            lease.detach();
170        }
171        self.id
172    }
173
174    // Release this handle. A resident lease that was the last one stops its
175    // instance and waits for it; a request-scoped instance is cancelled and
176    // settled.
177    pub async fn release(mut self) -> Result<()> {
178        self.released = true;
179        let lease = match &mut self.outcome {
180            Outcome::Resident { lease, .. } => lease.take(),
181            Outcome::Request(_) => None,
182        };
183        match lease {
184            Some(lease) => lease.release().await,
185            None => {
186                self.control.cancel();
187                self.settle().await
188            }
189        }
190    }
191
192    // Wait for the instance to settle without consuming the handle. A
193    // resident reports its outcome and retries a retained cleanup.
194    pub async fn settle(&self) -> Result<()> {
195        if let Outcome::Resident { lease: Some(lease), .. } = &self.outcome {
196            return lease.settlement().await;
197        }
198        self.control.settle().await.map_err(|source| Error::OperationSettlement { source })
199    }
200}
201
202impl<O: OperationDeclaration> OperationInstance<O> {
203    // Encoded invocations keep their own contract: dropping one never
204    // cancels, so the typed instance they await must not cancel either.
205    pub(super) fn unguarded(mut self) -> Self {
206        self.released = true;
207        self
208    }
209}
210
211impl<O: OperationDeclaration> Future for OperationInstance<O> {
212    type Output = Result<crate::operations::Response<O::Response>>;
213    fn poll(self: Pin<&mut Self>, cx: &mut TaskContext<'_>) -> Poll<Self::Output> {
214        let instance = self.get_mut();
215        let answer = match &mut instance.outcome {
216            Outcome::Request(outcome) => match Pin::new(outcome).poll(cx) {
217                Poll::Ready(Ok(answer)) => answer,
218                Poll::Ready(Err(_)) => Err(crate::Refusal::Unavailable.into()),
219                Poll::Pending => return Poll::Pending,
220            },
221            Outcome::Resident { settled, .. } => match settled.get_mut().expect("resident settlement is healthy").as_mut().poll(cx) {
222                Poll::Ready(answer) => answer,
223                Poll::Pending => return Poll::Pending,
224            },
225        };
226        instance.released = true;
227        Poll::Ready(answer)
228    }
229}
230
231impl<O: OperationDeclaration> Drop for OperationInstance<O> {
232    fn drop(&mut self) {
233        // A resident lease releases itself when dropped.
234        if matches!(self.outcome, Outcome::Request(_)) && !self.released && !self.control.is_settled() {
235            self.control.cancel();
236        }
237    }
238}
239
240impl<O> harmos_sdk::operations::Instance<O> for OperationInstance<O>
241where
242    O: OperationDeclaration,
243{
244    type Error = Error;
245
246    fn id(&self) -> OperationInstanceId {
247        self.id
248    }
249
250    fn status(&self) -> impl Future<Output = Result<harmos_domain::operations::instance::Status>> + Send {
251        std::future::ready(Ok(self.control.status().portable()))
252    }
253
254    fn cancel(&self) -> impl Future<Output = Result<()>> + Send {
255        OperationInstance::cancel(self);
256        std::future::ready(Ok(()))
257    }
258
259    fn join(self) -> impl Future<Output = Result<crate::operations::Response<O::Response>>> + Send {
260        self
261    }
262
263    fn detach(self) -> OperationInstanceId {
264        OperationInstance::detach(self)
265    }
266}
267
268impl Context<'_> {
269    pub fn context(&self) -> &super::CallContext {
270        self.operations.call_context()
271    }
272
273    // Derive a child caller while retaining this attempt's identity, admission,
274    // cancellation and deadline. Only the selected portable value changes.
275    pub fn with_value<T: super::ContextValue>(&self, value: T) -> Result<Self> {
276        let context = self.context().with_value(value)?;
277        let mut child = self.clone();
278        child.operations = child.operations.with_context(context);
279        Ok(child)
280    }
281
282    pub fn id(&self) -> OperationInstanceId {
283        self.id
284    }
285
286    // An active instance this attempt's caller submitted; others stay hidden.
287    pub(crate) fn owned_instance(&self, id: OperationInstanceId) -> Option<Arc<crate::supervision::instance::Instance>> {
288        self.operations.instances.owned(id, &self.origin.identity)
289    }
290
291    pub fn attempt(&self) -> usize {
292        self.attempt
293    }
294
295    pub fn scope(&self) -> &Scope<'_> {
296        &self.scope
297    }
298
299    pub fn is_host_call(&self) -> bool {
300        self.origin.identity == super::caller::Identity::Host
301    }
302
303    // Revalidate The Caller Before Every Streamed Destination
304    pub(crate) fn admit_stream_grant(&self, grant: &harmos_domain::plugin::Grant) -> Result<()> {
305        self.source.check()?;
306        match &self.origin.grants {
307            Some(authority) => authority.admit(Some(grant)),
308            None if self.is_host_call() => Ok(()),
309            None => Err(Error::PermissionDenied),
310        }
311    }
312
313    // Refuse a provider-prepared grant the admitted caller does not hold.
314    #[cfg(any(all(feature = "plugin-guest", not(target_arch = "wasm32")), feature = "plugin-sidecar"))]
315    pub(super) fn admit_grant(&self, grant: Option<&harmos_domain::plugin::Grant>) -> Result<()> {
316        match (&self.origin.grants, grant) {
317            (Some(authority), grant) => authority.admit(grant),
318            (None, None) => Ok(()),
319            (None, Some(_)) => Err(Error::PermissionDenied),
320        }
321    }
322
323    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
324    pub fn caller_scope(&self) -> Option<&crate::plugin::PluginScope> {
325        self.origin.scope.as_ref()
326    }
327
328    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
329    pub(crate) fn retains_publication(&self, scope: &crate::plugin::PluginScope) -> bool {
330        self.source.check().is_ok_and(|origin| {
331            (origin.has_lifecycle_loan() || origin.resident) && origin.scope.as_ref().is_some_and(|caller| caller.same_activation(scope))
332        })
333    }
334
335    pub fn submit_at<O: OperationDeclaration>(&self, id: &str, request: O) -> Result<OperationInstance<O>> {
336        self.operations.admit_at::<O, ()>(id, (request.into(), ()), Some(&self.scope), self.admission.held(), &self.source, None)
337    }
338
339    pub fn submit<O: OperationDeclaration>(&self, request: O) -> Result<OperationInstance<O>> {
340        self.operations.submit_scoped::<O, ()>((request.into(), ()), &self.scope, &self.admission, &self.source, None)
341    }
342
343    pub fn submit_endpoints<O: OperationDeclaration, P: Send + 'static>(&self, request: O, endpoints: P) -> Result<OperationInstance<O>> {
344        self.operations.submit_scoped::<O, P>((request.into(), endpoints), &self.scope, &self.admission, &self.source, None)
345    }
346
347    pub fn submit_charged<O: OperationDeclaration, P: Send + 'static>(
348        &self, request: super::StreamItem<O>, endpoints: P,
349    ) -> Result<OperationInstance<O>> {
350        self.operations.submit_scoped::<O, P>(
351            (super::binding::Request::charged(request)?, endpoints),
352            &self.scope,
353            &self.admission,
354            &self.source,
355            None,
356        )
357    }
358
359    pub fn submit_caused<O: OperationDeclaration>(&self, request: O, cause: Position) -> Result<OperationInstance<O>> {
360        self.operations.submit_scoped::<O, ()>((request.into(), ()), &self.scope, &self.admission, &self.source, Some(cause))
361    }
362
363    pub async fn call_caused<O: OperationDeclaration>(
364        &self, request: O, cause: Position,
365    ) -> Result<crate::operations::Response<O::Response>> {
366        self.submit_caused::<O>(request, cause)?.await
367    }
368
369    pub async fn call<O: OperationDeclaration>(&self, request: O) -> Result<crate::operations::Response<O::Response>> {
370        self.submit(request)?.await
371    }
372}
373
374/* === Public Functions === */
375
376pub(crate) fn fresh_instance() -> Result<OperationInstanceId> {
377    NEXT_INSTANCE
378        .try_update(Ordering::Relaxed, Ordering::Relaxed, |id| id.checked_add(1))
379        .map(OperationInstanceId::new)
380        .map_err(|_| Error::ResourceExhausted { resource: "operation instance identities" })
381}
382
383/* === Private Functions === */
384
385/* === Tests === */
harmos_runtime/operations/handler.rs

harmos_runtime/operations/
handler.rs
