harmos_runtime/operations/
wire.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17use std::future::Future;
18use std::panic::{AssertUnwindSafe, catch_unwind};
19use std::pin::Pin;
20use std::sync::{Arc, Weak};
21use std::task::{Context, Poll};
22use std::time::Duration;
23// Third Party Imports
24use tokio::sync::OwnedSemaphorePermit;
25// Local Imports
26use super::caller::Source;
27use super::manager::{Entry, Submission};
28use super::{OperationInstanceId, OperationManager};
29use crate::operations::OperationDeclaration;
30use crate::supervision::executor;
31use crate::{Error, Message, Refusal, Result, operations::Scope};
32
33/* === Constants === */
34
35/* === Macros === */
36
37/* === Type Aliases === */
38
39type Outcome = Pin<Box<dyn Future<Output = Result<Vec<u8>>> + Send>>;
40pub(super) type Dispatch = fn(&OperationManager, Submission, Source, Option<Parent<'_>>, EncodedInput) -> Result<EncodedInvocation>;
41
42/* === Structs === */
43
44pub(super) struct EncodedInput {
45    pub(super) payload: Vec<u8>,
46    pub(super) preparation: Option<OwnedSemaphorePermit>,
47
48    #[cfg(feature = "plugin-sidecar")]
49    pub(super) native_capacity: Option<OwnedSemaphorePermit>,
50}
51
52#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
53#[derive(Clone)]
54pub struct PluginCallback {
55    operations: OperationManager,
56    entry: Arc<Entry>,
57    contract: crate::operations::Contract,
58}
59
60#[cfg(feature = "plugin-sidecar")]
61#[derive(Clone)]
62pub(crate) struct PluginControl(PluginCallback);
63
64pub(super) struct Parent<'a> {
65    pub(super) scope: &'a Scope<'a>,
66    pub(super) ancestors: &'a [&'static str],
67}
68
69pub struct EncodedInvocation {
70    pub(super) id: OperationInstanceId,
71    pub(super) generation: super::OperationGeneration,
72    pub(super) outcome: Outcome,
73    pub(crate) control: Arc<crate::supervision::instance::Instance>,
74}
75
76/* === Enums === */
77
78/* === Traits === */
79
80/* === Implementations === */
81
82impl EncodedInvocation {
83    pub fn id(&self) -> OperationInstanceId {
84        self.id
85    }
86
87    pub fn generation(&self) -> super::OperationGeneration {
88        self.generation
89    }
90
91    pub fn cancel(&self) {
92        self.control.cancel();
93    }
94
95    pub fn status(&self) -> super::OperationStatus {
96        self.control.status()
97    }
98
99    #[tracing::instrument(target = "harmos::operations", name = "operations.join", level = "debug", skip_all)]
100    pub async fn join(self) -> Result<Vec<u8>> {
101        self.await
102    }
103
104    #[tracing::instrument(target = "harmos::operations", name = "operations.settle", level = "debug", skip_all)]
105    pub async fn settle(&self) -> Result<()> {
106        self.control.settle().await.map_err(|source| Error::OperationSettlement { source })
107    }
108}
109
110impl Future for EncodedInvocation {
111    type Output = Result<Vec<u8>>;
112
113    fn poll(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
114        self.get_mut().outcome.as_mut().poll(context)
115    }
116}
117
118impl OperationManager {
119    pub fn submit_encoded(&self, operation: &str, payload: Vec<u8>) -> Result<EncodedInvocation> {
120        self.submit_encoded_as(operation, payload, Source::Host, None)
121    }
122
123    pub async fn call_encoded(&self, operation: &str, payload: Vec<u8>) -> Result<Vec<u8>> {
124        self.submit_encoded(operation, payload)?.await
125    }
126
127    pub(super) fn submit_encoded_as(
128        &self, operation: &str, payload: Vec<u8>, source: Source, parent: Option<Parent<'_>>,
129    ) -> Result<EncodedInvocation> {
130        self.submit_encoded_at_as(operation, payload, source, parent, executor::now())
131    }
132
133    #[tracing::instrument(target = "harmos::operations", name = "operations.submit_encoded_at_as", level = "debug", skip_all)]
134    fn submit_encoded_at_as(
135        &self, operation: &str, payload: Vec<u8>, source: Source, parent: Option<Parent<'_>>, submitted: tokio::time::Instant,
136    ) -> Result<EncodedInvocation> {
137        let origin = source.check()?;
138        let entry = self.entry_as(operation, &origin)?;
139        entry.authorize(&origin)?;
140        let dispatch = entry
141            .wire
142            .ok_or(Error::InvalidInput { field: "encoded operation", reason: "operation has no cross-boundary codec binding" })?;
143
144        dispatch(
145            self,
146            Submission { submitted, entry, cause: None, preparation: None },
147            source,
148            parent,
149            EncodedInput {
150                payload,
151                preparation: None,
152                #[cfg(feature = "plugin-sidecar")]
153                native_capacity: None,
154            },
155        )
156    }
157}
158
159#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
160impl OperationManager {
161    #[cfg(feature = "plugin-sidecar")]
162    pub(crate) fn capture_host_control(
163        &self, scope: &crate::plugin::PluginScope, contract: &crate::operations::Contract,
164    ) -> Result<PluginControl> {
165        if contract.operation().visibility != crate::operations::Visibility::Host || scope.observe_contract() != Some(contract) {
166            return Err(Error::InvalidInput { field: "observer control", reason: "control differs from the frozen Host contract" });
167        }
168        self.capture_callback(scope, contract).map(PluginControl)
169    }
170
171    pub fn capture_callback(&self, scope: &crate::plugin::PluginScope, contract: &crate::operations::Contract) -> Result<PluginCallback> {
172        if scope.is_closed() {
173            return Err(Error::Closing { domain: "callback scope" });
174        }
175        let entry = self.capture_provider(contract.operation().id.as_str(), Some(scope.owner()))?;
176        let selected = entry
177            .plugin()
178            .ok_or(Error::InvalidInput { field: "callback scope", reason: "callback target has no original plugin generation" })?;
179        if !selected.same_activation(scope) {
180            return Err(Error::InvalidInput {
181                field: "callback scope",
182                reason: "callback target belongs to a different plugin generation",
183            });
184        }
185        PluginCallback::from_entry(self.clone(), entry, contract)
186    }
187}
188
189#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
190impl PluginCallback {
191    pub(super) fn from_entry(operations: OperationManager, entry: Arc<Entry>, contract: &crate::operations::Contract) -> Result<Self> {
192        if entry.declared() != contract.operation() || entry.contract.as_ref() != Some(contract) {
193            return Err(Error::InvalidInput {
194                field: "callback contract",
195                reason: "callback requires the exact retained canonical contract",
196            });
197        }
198        if entry.wire.is_none() {
199            return Err(Error::InvalidInput { field: "callback codec", reason: "callback target has no original codec binding" });
200        }
201        Ok(Self { operations, entry, contract: contract.clone() })
202    }
203
204    pub fn contract(&self) -> &crate::operations::Contract {
205        &self.contract
206    }
207
208    pub fn submit_with(
209        &self, encoding_memory_bytes: std::num::NonZeroUsize, encode: impl FnOnce(usize) -> Result<Vec<u8>>,
210    ) -> Result<EncodedInvocation> {
211        self.submit_with_as(Source::for_callback(&self.entry), encoding_memory_bytes, encode, false)
212    }
213
214    // A resident control runs until cancelled; it holds no native call slot.
215    fn submit_with_as(
216        &self, source: Source, encoding_memory_bytes: std::num::NonZeroUsize, encode: impl FnOnce(usize) -> Result<Vec<u8>>, resident: bool,
217    ) -> Result<EncodedInvocation> {
218        #[cfg(not(feature = "plugin-sidecar"))]
219        let _ = resident;
220        let submitted = executor::now();
221        let origin = source.check()?;
222        self.entry.available()?;
223        self.entry.authorize(&origin)?;
224        let timeout = self.entry.declared().policy.timeout;
225        budget(timeout, submitted)?;
226        let preparation = self.operations.claim_preparation(encoding_memory_bytes)?;
227        #[cfg(feature = "plugin-sidecar")]
228        let native_capacity = self
229            .entry
230            .native
231            .as_ref()
232            .map(|binding| if resident { Ok(super::native::Binding::resident_slot()) } else { binding.reserve() })
233            .transpose()
234            .map_err(|source| Error::OperationFailed { operation: self.entry.id().to_owned(), source: Arc::new(source) })?;
235        let payload = catch_unwind(AssertUnwindSafe(|| encode(encoding_memory_bytes.get())))
236            .map_err(|_| Error::OperationPanicked { operation: self.entry.id().to_owned() })??;
237        if payload.capacity() > encoding_memory_bytes.get() {
238            return Err(Error::ResourceExhausted { resource: "callback encoding claim" });
239        }
240        #[cfg(feature = "plugin-sidecar")]
241        if native_capacity.is_some() && payload.capacity() > harmos_sidecar::Frame::LIMIT {
242            return Err(Error::ResourceExhausted { resource: "native callback frame" });
243        }
244        source.check()?;
245        self.entry.available()?;
246        budget(timeout, submitted)?;
247        let dispatch = self
248            .entry
249            .wire
250            .ok_or(Error::InvalidInput { field: "callback codec", reason: "callback target has no original codec binding" })?;
251        dispatch(
252            &self.operations,
253            Submission { submitted, entry: Arc::clone(&self.entry), cause: None, preparation: None },
254            source,
255            None,
256            EncodedInput {
257                payload,
258                preparation: Some(preparation),
259                #[cfg(feature = "plugin-sidecar")]
260                native_capacity,
261            },
262        )
263    }
264}
265
266#[cfg(feature = "plugin-sidecar")]
267impl PluginControl {
268    pub(crate) fn submit_with(
269        &self, bytes: std::num::NonZeroUsize, encode: impl FnOnce(usize) -> Result<Vec<u8>>,
270    ) -> Result<EncodedInvocation> {
271        self.0.submit_with_as(Source::Host, bytes, encode, true)
272    }
273}
274
275#[cfg(any(feature = "plugin-sidecar", feature = "plugin-guest"))]
276impl super::Caller {
277    pub(crate) fn submit_encoded_at(
278        &self, operation: &str, payload: Vec<u8>, submitted: tokio::time::Instant, values: Vec<super::CallValue>,
279    ) -> Result<EncodedInvocation> {
280        let context = self.operations.call_context().with_values(values)?;
281        self.operations.with_context(context).submit_encoded_at_as(operation, payload, Source::Plugin(self.source.clone()), None, submitted)
282    }
283}
284
285impl super::Context<'_> {
286    pub fn submit_encoded(&self, operation: &str, payload: Vec<u8>) -> Result<EncodedInvocation> {
287        self.operations.submit_encoded_as(
288            operation,
289            payload,
290            self.source.clone(),
291            Some(Parent { scope: &self.scope, ancestors: self.admission.held() }),
292        )
293    }
294
295    #[cfg(any(feature = "plugin-sidecar", all(feature = "plugin-guest", not(target_arch = "wasm32"))))]
296    pub(crate) fn submit_encoded_at(
297        &self, operation: &str, payload: Vec<u8>, submitted: tokio::time::Instant, values: Vec<super::CallValue>,
298    ) -> Result<EncodedInvocation> {
299        let context = self.operations.call_context().with_values(values)?;
300        self.operations.with_context(context).submit_encoded_at_as(
301            operation,
302            payload,
303            self.source.clone(),
304            Some(Parent { scope: &self.scope, ancestors: self.admission.held() }),
305            submitted,
306        )
307    }
308
309    pub async fn call_encoded(&self, operation: &str, payload: Vec<u8>) -> Result<Vec<u8>> {
310        self.submit_encoded(operation, payload)?.await
311    }
312}
313
314/* === Public Functions === */
315
316pub(super) fn dispatch<O>(
317    operations: &OperationManager, mut submission: Submission, source: Source, parent: Option<Parent<'_>>, input: EncodedInput,
318) -> Result<EncodedInvocation>
319where
320    O: OperationDeclaration,
321{
322    let submitted = submission.submitted;
323    let timeout = submission.entry.declared().policy.timeout;
324    budget(timeout, submitted)?;
325    if operations.preparation.is_none() {
326        return Err(Error::InvalidInput {
327            field: "operation preparation bytes",
328            reason: "encoded preparation requires an explicit preparation budget",
329        });
330    }
331    submission.preparation = match input.preparation {
332        Some(original) => Some(original),
333        None => std::num::NonZeroUsize::new(input.payload.capacity()).map(|bytes| operations.claim_preparation(bytes)).transpose()?,
334    };
335    let decoding_bytes = codec::<O, _>(|| <O as crate::Message>::decoding_memory(&input.payload))?
336        .map_err(|source| Error::OperationPayload { operation: O::ID.to_owned(), source })?;
337    if let Some(bytes) = std::num::NonZeroUsize::new(decoding_bytes) {
338        let decoded = operations.claim_preparation(bytes)?;
339        if let Some(original) = &mut submission.preparation {
340            original.merge(decoded);
341        } else {
342            submission.preparation = Some(decoded);
343        }
344    }
345    budget(timeout, submitted)?;
346    let request = codec::<O, _>(|| <O as crate::Message>::decode(&input.payload))?
347        .map_err(|source| Error::OperationPayload { operation: O::ID.to_owned(), source })?;
348    let target = Arc::downgrade(&submission.entry);
349    budget(timeout, submitted)?;
350    let (scope, ancestors) = parent.map_or((None, &[][..]), |parent| (Some(parent.scope), parent.ancestors));
351    let invocation = operations.admit_entry::<O, ()>((request.into(), ()), scope, ancestors, &source, submission)?;
352    let id = invocation.id();
353    let generation = invocation.generation();
354    let control = Arc::clone(&invocation.control);
355    let invocation = invocation.unguarded();
356    let outcome = Box::pin(async move {
357        let response = invocation.await?;
358        available(&target, &source)?;
359        budget(timeout, submitted)?;
360        let payload = codec::<O, _>(|| response.encode_to_vec())?;
361        available(&target, &source)?;
362        budget(timeout, submitted)?;
363
364        Ok(payload)
365    });
366
367    Ok(EncodedInvocation { id, generation, outcome, control })
368}
369
370pub(super) fn available(target: &Weak<Entry>, source: &Source) -> Result<()> {
371    let origin = source.check()?;
372    target.upgrade().ok_or(Error::Closing { domain: "operation" })?.available_to(&origin)
373}
374
375pub(super) fn budget(timeout: Duration, submitted: tokio::time::Instant) -> Result<()> {
376    if executor::now().saturating_duration_since(submitted) >= timeout {
377        return Err(Refusal::TimedOut.into());
378    }
379    Ok(())
380}
381
382/* === Private Functions === */
383
384fn codec<O: OperationDeclaration, T>(run: impl FnOnce() -> T) -> Result<T> {
385    catch_unwind(AssertUnwindSafe(run)).map_err(|_| Error::OperationPanicked { operation: O::ID.to_owned() })
386}
387
388/* === Tests === */
harmos_runtime/operations/wire.rs

harmos_runtime/operations/
wire.rs
