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    pub async fn join(self) -> Result<Vec<u8>> {
100        self.await
101    }
102
103    pub async fn settle(&self) -> Result<()> {
104        self.control.settle().await.map_err(|source| Error::OperationSettlement { source })
105    }
106}
107
108impl Future for EncodedInvocation {
109    type Output = Result<Vec<u8>>;
110
111    fn poll(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
112        self.get_mut().outcome.as_mut().poll(context)
113    }
114}
115
116impl OperationManager {
117    pub fn submit_encoded(&self, operation: &str, payload: Vec<u8>) -> Result<EncodedInvocation> {
118        self.submit_encoded_as(operation, payload, Source::Host, None)
119    }
120
121    pub async fn call_encoded(&self, operation: &str, payload: Vec<u8>) -> Result<Vec<u8>> {
122        self.submit_encoded(operation, payload)?.await
123    }
124
125    pub(super) fn submit_encoded_as(
126        &self, operation: &str, payload: Vec<u8>, source: Source, parent: Option<Parent<'_>>,
127    ) -> Result<EncodedInvocation> {
128        self.submit_encoded_at_as(operation, payload, source, parent, executor::now())
129    }
130
131    fn submit_encoded_at_as(
132        &self, operation: &str, payload: Vec<u8>, source: Source, parent: Option<Parent<'_>>, submitted: tokio::time::Instant,
133    ) -> Result<EncodedInvocation> {
134        let origin = source.check()?;
135        let entry = self.entry_as(operation, &origin)?;
136        entry.authorize(&origin)?;
137        let dispatch = entry
138            .wire
139            .ok_or(Error::InvalidInput { field: "encoded operation", reason: "operation has no cross-boundary codec binding" })?;
140
141        dispatch(
142            self,
143            Submission { submitted, entry, cause: None, preparation: None },
144            source,
145            parent,
146            EncodedInput {
147                payload,
148                preparation: None,
149                #[cfg(feature = "plugin-sidecar")]
150                native_capacity: None,
151            },
152        )
153    }
154}
155
156#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
157impl OperationManager {
158    #[cfg(feature = "plugin-sidecar")]
159    pub(crate) fn capture_host_control(
160        &self, scope: &crate::plugin::PluginScope, contract: &crate::operations::Contract,
161    ) -> Result<PluginControl> {
162        if contract.operation().visibility != crate::operations::Visibility::Host || scope.observe_contract() != Some(contract) {
163            return Err(Error::InvalidInput { field: "observer control", reason: "control differs from the frozen Host contract" });
164        }
165        self.capture_callback(scope, contract).map(PluginControl)
166    }
167
168    pub fn capture_callback(&self, scope: &crate::plugin::PluginScope, contract: &crate::operations::Contract) -> Result<PluginCallback> {
169        if scope.is_closed() {
170            return Err(Error::Closing { domain: "callback scope" });
171        }
172        let entry = self.capture_provider(contract.operation().id.as_str(), Some(scope.owner()))?;
173        let selected = entry
174            .plugin()
175            .ok_or(Error::InvalidInput { field: "callback scope", reason: "callback target has no original plugin generation" })?;
176        if !selected.same_activation(scope) {
177            return Err(Error::InvalidInput {
178                field: "callback scope",
179                reason: "callback target belongs to a different plugin generation",
180            });
181        }
182        PluginCallback::from_entry(self.clone(), entry, contract)
183    }
184}
185
186#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
187impl PluginCallback {
188    pub(super) fn from_entry(operations: OperationManager, entry: Arc<Entry>, contract: &crate::operations::Contract) -> Result<Self> {
189        if entry.declared() != contract.operation() || entry.contract.as_ref() != Some(contract) {
190            return Err(Error::InvalidInput {
191                field: "callback contract",
192                reason: "callback requires the exact retained canonical contract",
193            });
194        }
195        if entry.wire.is_none() {
196            return Err(Error::InvalidInput { field: "callback codec", reason: "callback target has no original codec binding" });
197        }
198        Ok(Self { operations, entry, contract: contract.clone() })
199    }
200
201    pub fn contract(&self) -> &crate::operations::Contract {
202        &self.contract
203    }
204
205    pub fn submit_with(
206        &self, encoding_memory_bytes: std::num::NonZeroUsize, encode: impl FnOnce(usize) -> Result<Vec<u8>>,
207    ) -> Result<EncodedInvocation> {
208        self.submit_with_as(Source::for_callback(&self.entry), encoding_memory_bytes, encode, false)
209    }
210
211    // A resident control runs until cancelled; it holds no native call slot.
212    fn submit_with_as(
213        &self, source: Source, encoding_memory_bytes: std::num::NonZeroUsize, encode: impl FnOnce(usize) -> Result<Vec<u8>>, resident: bool,
214    ) -> Result<EncodedInvocation> {
215        #[cfg(not(feature = "plugin-sidecar"))]
216        let _ = resident;
217        let submitted = executor::now();
218        let origin = source.check()?;
219        self.entry.available()?;
220        self.entry.authorize(&origin)?;
221        let timeout = self.entry.declared().policy.timeout;
222        budget(timeout, submitted)?;
223        let preparation = self.operations.claim_preparation(encoding_memory_bytes)?;
224        #[cfg(feature = "plugin-sidecar")]
225        let native_capacity = self
226            .entry
227            .native
228            .as_ref()
229            .map(|binding| if resident { Ok(super::native::Binding::resident_slot()) } else { binding.reserve() })
230            .transpose()
231            .map_err(|source| Error::OperationFailed { operation: self.entry.id().to_owned(), source: Arc::new(source) })?;
232        let payload = catch_unwind(AssertUnwindSafe(|| encode(encoding_memory_bytes.get())))
233            .map_err(|_| Error::OperationPanicked { operation: self.entry.id().to_owned() })??;
234        if payload.capacity() > encoding_memory_bytes.get() {
235            return Err(Error::ResourceExhausted { resource: "callback encoding claim" });
236        }
237        #[cfg(feature = "plugin-sidecar")]
238        if native_capacity.is_some() && payload.capacity() > harmos_sidecar::Frame::LIMIT {
239            return Err(Error::ResourceExhausted { resource: "native callback frame" });
240        }
241        source.check()?;
242        self.entry.available()?;
243        budget(timeout, submitted)?;
244        let dispatch = self
245            .entry
246            .wire
247            .ok_or(Error::InvalidInput { field: "callback codec", reason: "callback target has no original codec binding" })?;
248        dispatch(
249            &self.operations,
250            Submission { submitted, entry: Arc::clone(&self.entry), cause: None, preparation: None },
251            source,
252            None,
253            EncodedInput {
254                payload,
255                preparation: Some(preparation),
256                #[cfg(feature = "plugin-sidecar")]
257                native_capacity,
258            },
259        )
260    }
261}
262
263#[cfg(feature = "plugin-sidecar")]
264impl PluginControl {
265    pub(crate) fn submit_with(
266        &self, bytes: std::num::NonZeroUsize, encode: impl FnOnce(usize) -> Result<Vec<u8>>,
267    ) -> Result<EncodedInvocation> {
268        self.0.submit_with_as(Source::Host, bytes, encode, true)
269    }
270}
271
272#[cfg(any(feature = "plugin-sidecar", feature = "plugin-guest"))]
273impl super::Caller {
274    pub(crate) fn submit_encoded_at(
275        &self, operation: &str, payload: Vec<u8>, submitted: tokio::time::Instant, values: Vec<super::CallValue>,
276    ) -> Result<EncodedInvocation> {
277        let context = self.operations.call_context().with_values(values)?;
278        self.operations.with_context(context).submit_encoded_at_as(operation, payload, Source::Plugin(self.source.clone()), None, submitted)
279    }
280}
281
282impl super::Context<'_> {
283    pub fn submit_encoded(&self, operation: &str, payload: Vec<u8>) -> Result<EncodedInvocation> {
284        self.operations.submit_encoded_as(
285            operation,
286            payload,
287            self.source.clone(),
288            Some(Parent { scope: &self.scope, ancestors: self.admission.held() }),
289        )
290    }
291
292    #[cfg(any(feature = "plugin-sidecar", all(feature = "plugin-guest", not(target_arch = "wasm32"))))]
293    pub(crate) fn submit_encoded_at(
294        &self, operation: &str, payload: Vec<u8>, submitted: tokio::time::Instant, values: Vec<super::CallValue>,
295    ) -> Result<EncodedInvocation> {
296        let context = self.operations.call_context().with_values(values)?;
297        self.operations.with_context(context).submit_encoded_at_as(
298            operation,
299            payload,
300            self.source.clone(),
301            Some(Parent { scope: &self.scope, ancestors: self.admission.held() }),
302            submitted,
303        )
304    }
305
306    pub async fn call_encoded(&self, operation: &str, payload: Vec<u8>) -> Result<Vec<u8>> {
307        self.submit_encoded(operation, payload)?.await
308    }
309}
310
311/* === Public Functions === */
312
313pub(super) fn dispatch<O>(
314    operations: &OperationManager, mut submission: Submission, source: Source, parent: Option<Parent<'_>>, input: EncodedInput,
315) -> Result<EncodedInvocation>
316where
317    O: OperationDeclaration,
318{
319    let submitted = submission.submitted;
320    let timeout = submission.entry.declared().policy.timeout;
321    budget(timeout, submitted)?;
322    if operations.preparation.is_none() {
323        return Err(Error::InvalidInput {
324            field: "operation preparation bytes",
325            reason: "encoded preparation requires an explicit preparation budget",
326        });
327    }
328    submission.preparation = match input.preparation {
329        Some(original) => Some(original),
330        None => std::num::NonZeroUsize::new(input.payload.capacity()).map(|bytes| operations.claim_preparation(bytes)).transpose()?,
331    };
332    let decoding_bytes = codec::<O, _>(|| <O as crate::Message>::decoding_memory(&input.payload))?
333        .map_err(|source| Error::OperationPayload { operation: O::ID.to_owned(), source })?;
334    if let Some(bytes) = std::num::NonZeroUsize::new(decoding_bytes) {
335        let decoded = operations.claim_preparation(bytes)?;
336        if let Some(original) = &mut submission.preparation {
337            original.merge(decoded);
338        } else {
339            submission.preparation = Some(decoded);
340        }
341    }
342    budget(timeout, submitted)?;
343    let request = codec::<O, _>(|| <O as crate::Message>::decode(&input.payload))?
344        .map_err(|source| Error::OperationPayload { operation: O::ID.to_owned(), source })?;
345    let target = Arc::downgrade(&submission.entry);
346    budget(timeout, submitted)?;
347    let (scope, ancestors) = parent.map_or((None, &[][..]), |parent| (Some(parent.scope), parent.ancestors));
348    let invocation = operations.admit_entry::<O, ()>((request.into(), ()), scope, ancestors, &source, submission)?;
349    let id = invocation.id();
350    let generation = invocation.generation();
351    let control = Arc::clone(&invocation.control);
352    let invocation = invocation.unguarded();
353    let outcome = Box::pin(async move {
354        let response = invocation.await?;
355        available(&target, &source)?;
356        budget(timeout, submitted)?;
357        let payload = codec::<O, _>(|| response.encode_to_vec())?;
358        available(&target, &source)?;
359        budget(timeout, submitted)?;
360
361        Ok(payload)
362    });
363
364    Ok(EncodedInvocation { id, generation, outcome, control })
365}
366
367pub(super) fn available(target: &Weak<Entry>, source: &Source) -> Result<()> {
368    let origin = source.check()?;
369    target.upgrade().ok_or(Error::Closing { domain: "operation" })?.available_to(&origin)
370}
371
372pub(super) fn budget(timeout: Duration, submitted: tokio::time::Instant) -> Result<()> {
373    if executor::now().saturating_duration_since(submitted) >= timeout {
374        return Err(Refusal::TimedOut.into());
375    }
376    Ok(())
377}
378
379/* === Private Functions === */
380
381fn codec<O: OperationDeclaration, T>(run: impl FnOnce() -> T) -> Result<T> {
382    catch_unwind(AssertUnwindSafe(run)).map_err(|_| Error::OperationPanicked { operation: O::ID.to_owned() })
383}
384
385/* === Tests === */
harmos_runtime/operations/wire.rs

harmos_runtime/operations/
wire.rs
