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::{Manager, OperationInstanceId};
29use crate::operations::OperationDeclaration;
30use crate::supervision::executor;
31use crate::{Error, 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(&Manager, Submission, Source, Option<Parent<'_>>, EncodedInput) -> Result<EncodedInvocation>;
41
42/* === Structs === */
43
44struct Encoded;
45
46pub(super) struct EncodedInput {
47    pub(super) payload: Vec<u8>,
48    pub(super) preparation: Option<OwnedSemaphorePermit>,
49
50    #[cfg(feature = "plugin-sidecar")]
51    pub(super) native_capacity: Option<OwnedSemaphorePermit>,
52}
53
54#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
55#[derive(Clone)]
56pub struct PluginCallback {
57    operations: Manager,
58    entry: Arc<Entry>,
59    contract: crate::operations::Contract,
60}
61
62#[cfg(feature = "plugin-sidecar")]
63#[derive(Clone)]
64pub(crate) struct PluginControl(PluginCallback);
65
66pub(super) struct Parent<'a> {
67    pub(super) scope: &'a Scope<'a>,
68    pub(super) ancestors: &'a [&'static str],
69}
70
71pub struct EncodedInvocation {
72    pub(super) id: OperationInstanceId,
73    pub(super) generation: super::OperationGeneration,
74    pub(super) outcome: Outcome,
75    pub(crate) control: Arc<crate::supervision::instance::Instance>,
76}
77
78/* === Enums === */
79
80/* === Traits === */
81
82/* === Implementations === */
83
84impl OperationDeclaration for Encoded {
85    type Request = Vec<u8>;
86    type Response = Vec<u8>;
87    type Input = super::NoStream;
88    type Output = super::NoStream;
89    const ID: &'static str = "encoded";
90    const POLICY: super::OperationPolicy = super::OperationPolicy {
91        timeout: harmos_domain::operations::UNBOUNDED,
92        attempt_timeout: harmos_domain::operations::UNBOUNDED,
93        retries: super::RetryPolicy::Never,
94    };
95}
96
97impl EncodedInvocation {
98    pub fn id(&self) -> OperationInstanceId {
99        self.id
100    }
101
102    pub fn generation(&self) -> super::OperationGeneration {
103        self.generation
104    }
105
106    pub fn cancel(&self) {
107        self.control.cancel();
108    }
109
110    pub fn status(&self) -> super::OperationStatus {
111        self.control.status()
112    }
113
114    #[tracing::instrument(target = "harmos::operations", name = "operations.join", level = "debug", skip_all)]
115    pub async fn join(self) -> Result<Vec<u8>> {
116        self.await
117    }
118
119    #[tracing::instrument(target = "harmos::operations", name = "operations.settle", level = "debug", skip_all)]
120    pub async fn settle(&self) -> Result<()> {
121        self.control.settle().await.map_err(|source| Error::OperationSettlement { source })
122    }
123}
124
125impl Future for EncodedInvocation {
126    type Output = Result<Vec<u8>>;
127
128    fn poll(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
129        self.get_mut().outcome.as_mut().poll(context)
130    }
131}
132
133impl Manager {
134    pub fn submit_encoded(&self, operation: &str, payload: Vec<u8>) -> Result<EncodedInvocation> {
135        self.submit_encoded_as(operation, payload, Source::Host, None)
136    }
137
138    pub async fn call_encoded(&self, operation: &str, payload: Vec<u8>) -> Result<Vec<u8>> {
139        self.submit_encoded(operation, payload)?.await
140    }
141
142    pub(super) fn submit_encoded_as(
143        &self, operation: &str, payload: Vec<u8>, source: Source, parent: Option<Parent<'_>>,
144    ) -> Result<EncodedInvocation> {
145        self.submit_encoded_at_as(operation, payload, source, parent, executor::now(), None)
146    }
147
148    #[tracing::instrument(target = "harmos::operations", name = "operations.submit_encoded_at_as", level = "debug", skip_all)]
149    fn submit_encoded_at_as(
150        &self, operation: &str, payload: Vec<u8>, source: Source, parent: Option<Parent<'_>>, submitted: tokio::time::Instant,
151        expected: Option<([u8; 32], u64)>,
152    ) -> Result<EncodedInvocation> {
153        let origin = source.check()?;
154        let qualified = crate::QualifiedId::parse(operation).map_err(|_| Error::InvalidInput {
155            field: "encoded operation",
156            reason: "routing requires an owner-qualified operation identity",
157        })?;
158        let (owner, local) = qualified.as_str().split_once('.').expect("validated qualified identity");
159        let entry = self.capture_routed(local, owner, &origin).map_err(|error| match error {
160            Error::NotRegistered { .. } => Error::NotRegistered {
161                domain: "operation",
162                name: operation.to_owned(),
163            },
164            error => error,
165        })?;
166        entry.available_to(&origin)?;
167        entry.authorize(&origin)?;
168        if expected.is_some_and(|(fingerprint, generation)| {
169            entry.declared().fingerprint.0 != fingerprint || entry.generation.value() != generation
170        }) {
171            return Err(Error::InvalidInput {
172                field: "encoded operation identity",
173                reason: "fingerprint or generation differs from the assigned target",
174            });
175        }
176        let dispatch = entry.wire.ok_or(Error::InvalidInput {
177            field: "encoded operation",
178            reason: "operation has no cross-boundary codec binding",
179        })?;
180
181        dispatch(
182            self,
183            Submission {
184                submitted,
185                entry,
186                cause: None,
187                preparation: None,
188            },
189            source,
190            parent,
191            EncodedInput {
192                payload,
193                preparation: None,
194                #[cfg(feature = "plugin-sidecar")]
195                native_capacity: None,
196            },
197        )
198    }
199}
200
201#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
202impl Manager {
203    #[cfg(feature = "plugin-sidecar")]
204    pub(crate) fn capture_host_control(
205        &self, scope: &crate::plugin::PluginScope, contract: &crate::operations::Contract,
206    ) -> Result<PluginControl> {
207        if contract.operation().visibility != crate::operations::Visibility::Host || scope.observe_contract() != Some(contract) {
208            return Err(Error::InvalidInput {
209                field: "observer control",
210                reason: "control differs from the frozen Host contract",
211            });
212        }
213        self.capture_callback(scope, contract).map(PluginControl)
214    }
215
216    pub fn capture_callback(&self, scope: &crate::plugin::PluginScope, contract: &crate::operations::Contract) -> Result<PluginCallback> {
217        if scope.is_closed() {
218            return Err(Error::Closing { domain: "callback scope" });
219        }
220        let entry = self.capture_provider(contract.operation().id.as_str(), Some(scope.owner()))?;
221        let selected = entry.plugin().ok_or(Error::InvalidInput {
222            field: "callback scope",
223            reason: "callback target has no original plugin generation",
224        })?;
225        if !selected.same_activation(scope) {
226            return Err(Error::InvalidInput {
227                field: "callback scope",
228                reason: "callback target belongs to a different plugin generation",
229            });
230        }
231        PluginCallback::from_entry(self.clone(), entry, contract)
232    }
233}
234
235#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
236impl PluginCallback {
237    pub(super) fn from_entry(operations: Manager, entry: Arc<Entry>, contract: &crate::operations::Contract) -> Result<Self> {
238        if entry.declared() != contract.operation() || entry.contract.as_ref() != Some(contract) {
239            return Err(Error::InvalidInput {
240                field: "callback contract",
241                reason: "callback requires the exact retained canonical contract",
242            });
243        }
244        if entry.wire.is_none() {
245            return Err(Error::InvalidInput {
246                field: "callback codec",
247                reason: "callback target has no original codec binding",
248            });
249        }
250        Ok(Self {
251            operations,
252            entry,
253            contract: contract.clone(),
254        })
255    }
256
257    pub fn contract(&self) -> &crate::operations::Contract {
258        &self.contract
259    }
260
261    pub fn submit_with(
262        &self, encoding_memory_bytes: std::num::NonZeroUsize, encode: impl FnOnce(usize) -> Result<Vec<u8>>,
263    ) -> Result<EncodedInvocation> {
264        self.submit_with_as(Source::for_callback(&self.entry), encoding_memory_bytes, encode, false)
265    }
266
267    // A resident control runs until cancelled; it holds no native call slot.
268    fn submit_with_as(
269        &self, source: Source, encoding_memory_bytes: std::num::NonZeroUsize, encode: impl FnOnce(usize) -> Result<Vec<u8>>, resident: bool,
270    ) -> Result<EncodedInvocation> {
271        #[cfg(not(feature = "plugin-sidecar"))]
272        let _ = resident;
273        let submitted = executor::now();
274        let origin = source.check()?;
275        self.entry.available()?;
276        self.entry.authorize(&origin)?;
277        let timeout = self.entry.declared().policy.timeout;
278        budget(timeout, submitted)?;
279        let preparation = self.operations.claim_preparation(encoding_memory_bytes)?;
280        #[cfg(feature = "plugin-sidecar")]
281        let native_capacity = self
282            .entry
283            .native
284            .as_ref()
285            .map(|binding| if resident { Ok(super::native::Binding::resident_slot()) } else { binding.reserve() })
286            .transpose()
287            .map_err(|source| Error::OperationFailed {
288                operation: self.entry.id().to_owned(),
289                source: Arc::new(source),
290            })?;
291        let payload = catch_unwind(AssertUnwindSafe(|| encode(encoding_memory_bytes.get())))
292            .map_err(|_| Error::OperationPanicked { operation: self.entry.id().to_owned() })??;
293        if payload.capacity() > encoding_memory_bytes.get() {
294            return Err(Error::ResourceExhausted { resource: "callback encoding claim" });
295        }
296        #[cfg(feature = "plugin-sidecar")]
297        if native_capacity.is_some() && payload.capacity() > harmos_sidecar::Frame::LIMIT {
298            return Err(Error::ResourceExhausted { resource: "native callback frame" });
299        }
300        source.check()?;
301        self.entry.available()?;
302        budget(timeout, submitted)?;
303        let dispatch = self.entry.wire.ok_or(Error::InvalidInput {
304            field: "callback codec",
305            reason: "callback target has no original codec binding",
306        })?;
307        dispatch(
308            &self.operations,
309            Submission {
310                submitted,
311                entry: Arc::clone(&self.entry),
312                cause: None,
313                preparation: None,
314            },
315            source,
316            None,
317            EncodedInput {
318                payload,
319                preparation: Some(preparation),
320                #[cfg(feature = "plugin-sidecar")]
321                native_capacity,
322            },
323        )
324    }
325}
326
327#[cfg(feature = "plugin-sidecar")]
328impl PluginControl {
329    pub(crate) fn submit_with(
330        &self, bytes: std::num::NonZeroUsize, encode: impl FnOnce(usize) -> Result<Vec<u8>>,
331    ) -> Result<EncodedInvocation> {
332        self.0.submit_with_as(Source::Host, bytes, encode, true)
333    }
334}
335
336#[cfg(any(feature = "plugin-sidecar", feature = "plugin-guest"))]
337impl super::Caller {
338    pub(crate) fn submit_encoded_at(
339        &self, operation: &str, fingerprint: [u8; 32], generation: u64, payload: Vec<u8>, submitted: tokio::time::Instant,
340        values: Vec<super::CallValue>,
341    ) -> Result<EncodedInvocation> {
342        let context = self.operations.call_context().with_values(values)?;
343        self.operations.with_context(context).submit_encoded_at_as(
344            operation,
345            payload,
346            Source::Plugin(self.source.clone()),
347            None,
348            submitted,
349            Some((fingerprint, generation)),
350        )
351    }
352}
353
354impl super::Context<'_> {
355    pub fn submit_encoded(&self, operation: &str, payload: Vec<u8>) -> Result<EncodedInvocation> {
356        self.operations.submit_encoded_as(
357            operation,
358            payload,
359            self.source.clone(),
360            Some(Parent {
361                scope: &self.scope,
362                ancestors: self.admission.held(),
363            }),
364        )
365    }
366
367    #[cfg(any(feature = "plugin-sidecar", all(feature = "plugin-guest", not(target_arch = "wasm32"))))]
368    pub(crate) fn submit_encoded_at(
369        &self, operation: &str, fingerprint: [u8; 32], generation: u64, payload: Vec<u8>, submitted: tokio::time::Instant,
370        values: Vec<super::CallValue>,
371    ) -> Result<EncodedInvocation> {
372        let context = self.operations.call_context().with_values(values)?;
373        self.operations.with_context(context).submit_encoded_at_as(
374            operation,
375            payload,
376            self.source.clone(),
377            Some(Parent {
378                scope: &self.scope,
379                ancestors: self.admission.held(),
380            }),
381            submitted,
382            Some((fingerprint, generation)),
383        )
384    }
385
386    pub async fn call_encoded(&self, operation: &str, payload: Vec<u8>) -> Result<Vec<u8>> {
387        self.submit_encoded(operation, payload)?.await
388    }
389}
390
391/* === Public Functions === */
392
393pub(super) fn dispatch_native(
394    operations: &Manager, mut submission: Submission, source: Source, parent: Option<Parent<'_>>, input: EncodedInput,
395) -> Result<EncodedInvocation> {
396    let submitted = submission.submitted;
397    let timeout = submission.entry.declared().policy.timeout;
398    budget(timeout, submitted)?;
399    let factory = submission.entry.binding.downcast_ref::<harmos_sdk::operations::OperationFactory>().ok_or(Error::InvalidInput {
400        field: "native operation",
401        reason: "entry does not own a native factory",
402    })?;
403    let name = submission.entry.id().to_owned();
404    submission.preparation = match input.preparation {
405        Some(original) => Some(original),
406        None => Some(operations.claim_preparation(std::num::NonZeroUsize::new(input.payload.capacity().max(1)).unwrap())?),
407    };
408    let decoding_bytes = catch_unwind(AssertUnwindSafe(|| factory.decoding_memory(&input.payload)))
409        .map_err(|_| Error::OperationPanicked { operation: name.clone() })?
410        .map_err(|source| Error::OperationPayload { operation: name.clone(), source })?;
411    if let Some(bytes) = std::num::NonZeroUsize::new(decoding_bytes) {
412        let decoded = operations.claim_preparation(bytes)?;
413        if let Some(original) = &mut submission.preparation {
414            original.merge(decoded);
415        } else {
416            submission.preparation = Some(decoded);
417        }
418    }
419    let prepared = catch_unwind(AssertUnwindSafe(|| factory.prepare_encoded(&input.payload)))
420        .map_err(|_| Error::OperationPanicked { operation: name.clone() })?
421        .map_err(|source| Error::OperationFailed {
422            operation: name,
423            source: Arc::new(source),
424        })?;
425    drop(input.payload);
426    dispatch_prepared(operations, submission, source, parent, prepared, super::binding::Request::from(Vec::new()))
427}
428
429pub(super) fn dispatch_prepared(
430    operations: &Manager, submission: Submission, source: Source, parent: Option<Parent<'_>>,
431    prepared: harmos_sdk::operations::registration::PreparedNative, mut request: super::binding::Request<Vec<u8>>,
432) -> Result<EncodedInvocation> {
433    let submitted = submission.submitted;
434    let timeout = submission.entry.declared().policy.timeout;
435    let harmos_sdk::operations::registration::PreparedNative { instance, grant, key, .. } = prepared;
436    let instance = std::sync::Mutex::new(Some(instance));
437    let binding = super::authored::factory_with::<Encoded, ()>(operations, move |_, ()| {
438        instance.lock().expect("native preparation lock is healthy").take().ok_or(harmos_sdk::ServingError::Uninitialized)
439    });
440    request.binding = Some(Arc::new(binding));
441    request.grant = grant;
442    request.key = key;
443    let target = Arc::downgrade(&submission.entry);
444    budget(timeout, submitted)?;
445    let (scope, ancestors) = parent.map_or((None, &[][..]), |parent| (Some(parent.scope), parent.ancestors));
446    let invocation = operations.admit_entry::<Encoded, ()>((request, ()), scope, ancestors, &source, submission)?;
447    let id = invocation.id();
448    let generation = invocation.generation();
449    let control = Arc::clone(&invocation.control);
450    let invocation = invocation.unguarded();
451    let outcome = Box::pin(async move {
452        let response = invocation.await?;
453        available(&target, &source)?;
454        budget(timeout, submitted)?;
455        Ok(response.try_unwrap().unwrap_or_else(|response| response.to_vec()))
456    });
457    Ok(EncodedInvocation { id, generation, outcome, control })
458}
459
460pub(super) fn available(target: &Weak<Entry>, source: &Source) -> Result<()> {
461    let origin = source.check()?;
462    target.upgrade().ok_or(Error::Closing { domain: "operation" })?.available_to(&origin)
463}
464
465pub(super) fn budget(timeout: Duration, submitted: tokio::time::Instant) -> Result<()> {
466    if executor::now().saturating_duration_since(submitted) >= timeout {
467        return Err(Refusal::TimedOut.into());
468    }
469    Ok(())
470}
471
472/* === Private Functions === */
473
474/* === Tests === */
475
476#[cfg(test)]
477mod tests {
478    use super::*;
479    use crate::RegistrationExt;
480
481    #[crate::lanes]
482    enum Lanes {
483        #[default]
484        Default,
485    }
486
487    struct Model;
488    impl crate::Application for Model {
489        type Lanes = Lanes;
490        type State = ();
491        type Transaction = crate::state::Never;
492        type Record = ();
493        type Streams = ();
494        type Resources = ();
495    }
496
497    #[crate::operation(harmos_path = crate, resources = [], id = "fenced.probe", response = u64,
498        policy = crate::OperationPolicy {
499            timeout: Duration::from_secs(1),
500            attempt_timeout: Duration::from_secs(1),
501            retries: crate::RetryPolicy::Never,
502        },
503    )]
504    struct Probe {
505        value: u64,
506    }
507
508    impl crate::Operation for Probe {
509        type Error = std::io::Error;
510
511        async fn run(self, _: crate::Context<'_, Self>) -> std::io::Result<crate::Response<u64>> {
512            Ok(self.value.into())
513        }
514    }
515
516    #[tokio::test]
517    async fn assigned_identity_fences_before_decoding() {
518        let builder = crate::Runtime::<Model>::builder((), ()).register(Lanes::catalog());
519        let runtime = builder.runtime().await.unwrap();
520        let batch = harmos_sdk::RegistrationBatch::new(harmos_sdk::Dependencies::new()).operation::<Probe>();
521        runtime.register(batch.clone().defaults()).await.unwrap();
522        let entry = runtime.operations.entry(Probe::ID).unwrap();
523        let fingerprint = entry.declared().fingerprint.0;
524        let generation = entry.generation.value();
525        let mut wrong = fingerprint;
526        wrong[0] ^= 1;
527
528        let submit = |fingerprint, generation| {
529            runtime.operations.submit_encoded_at_as(
530                "host.fenced.probe",
531                Vec::new(),
532                Source::Host,
533                None,
534                executor::now(),
535                Some((fingerprint, generation)),
536            )
537        };
538        assert!(matches!(submit(wrong, generation), Err(Error::InvalidInput { field: "encoded operation identity", .. })));
539        assert!(matches!(submit(fingerprint, generation), Err(Error::InvalidInput { field: "operation preparation bytes", .. })));
540
541        let id = crate::QualifiedId::parse("host.fenced.probe").unwrap();
542        runtime.unregister(&id).await.unwrap();
543        runtime.register(batch.defaults()).await.unwrap();
544        assert!(matches!(submit(fingerprint, generation), Err(Error::InvalidInput { field: "encoded operation identity", .. })));
545        runtime.stop().await.unwrap();
546    }
547}
harmos_runtime/operations/wire.rs

harmos_runtime/operations/
wire.rs
