sema_core/runtime/
completion.rs

1use std::any::{type_name, Any};
2
3use crate::{SemaError, Value};
4
5use super::{
6    CompletionKind, NativeCallContext, OperationId, RuntimeId, Trace, WaitGeneration, WaitId,
7};
8
9pub type SendPayload = Box<dyn Any + Send>;
10pub type DecodedCompletion = Result<Value, SemaError>;
11
12pub trait CompletionDecoder: Trace {
13    fn decode(
14        self: Box<Self>,
15        context: &mut NativeCallContext<'_>,
16        result: Result<SendPayload, ExternalFailure>,
17    ) -> DecodedCompletion;
18}
19
20pub struct ExternalCompletion {
21    pub runtime_id: RuntimeId,
22    pub wait_id: WaitId,
23    pub generation: WaitGeneration,
24    pub operation_id: OperationId,
25    pub kind: CompletionKind,
26    pub result: Result<SendPayload, ExternalFailure>,
27}
28
29#[derive(Clone, Copy, Debug, Eq, PartialEq)]
30pub enum ExternalFailureCode {
31    Rejected,
32    Cancelled,
33    DeadlineExceeded,
34    BoundExceeded,
35    WorkerPanic,
36    Decode,
37}
38
39#[derive(Clone, Debug, Eq, PartialEq)]
40pub struct ExternalFailure {
41    code: ExternalFailureCode,
42    message: String,
43    operation: Option<&'static str>,
44    expected_type: Option<&'static str>,
45}
46
47impl ExternalFailure {
48    fn new(code: ExternalFailureCode, message: impl Into<String>) -> Self {
49        Self {
50            code,
51            message: message.into(),
52            operation: None,
53            expected_type: None,
54        }
55    }
56
57    pub fn code(&self) -> ExternalFailureCode {
58        self.code
59    }
60
61    pub fn message(&self) -> &str {
62        &self.message
63    }
64
65    pub fn operation(&self) -> Option<&'static str> {
66        self.operation
67    }
68
69    pub fn expected_type(&self) -> Option<&'static str> {
70        self.expected_type
71    }
72
73    pub fn deadline_exceeded(message: impl Into<String>) -> Self {
74        Self::new(ExternalFailureCode::DeadlineExceeded, message)
75    }
76
77    pub fn bound_exceeded(message: impl Into<String>) -> Self {
78        Self::new(ExternalFailureCode::BoundExceeded, message)
79    }
80
81    /// The runtime-side failure used after an executor rejects an unadmitted submission.
82    pub fn rejected() -> Self {
83        Self::new(ExternalFailureCode::Rejected, "external operation rejected")
84    }
85
86    pub(crate) fn decode(
87        message: String,
88        operation: &'static str,
89        expected_type: &'static str,
90    ) -> Self {
91        Self {
92            code: ExternalFailureCode::Decode,
93            message,
94            operation: Some(operation),
95            expected_type: Some(expected_type),
96        }
97    }
98
99    pub(crate) fn cancelled() -> Self {
100        Self::new(
101            ExternalFailureCode::Cancelled,
102            "external operation cancelled",
103        )
104    }
105
106    pub(crate) fn worker_panic() -> Self {
107        Self::new(ExternalFailureCode::WorkerPanic, "external worker panicked")
108    }
109}
110
111pub fn downcast_send_payload<T: Any + Send>(
112    payload: SendPayload,
113    operation: &'static str,
114) -> Result<T, ExternalFailure> {
115    payload.downcast::<T>().map(|value| *value).map_err(|_| {
116        let expected = type_name::<T>();
117        ExternalFailure::decode(
118            format!("{operation} returned an unexpected payload; expected {expected}"),
119            operation,
120            expected,
121        )
122    })
123}
124
125#[derive(Clone, Copy, Debug, Eq, PartialEq)]
126pub enum CompletionDelivery {
127    Delivered,
128    InboxClosed,
129}
130
131pub trait CompletionSender: Send + Sync + 'static {
132    fn send(&self, completion: ExternalCompletion) -> CompletionDelivery;
133}
sema_core/runtime/completion.rs

sema_core/runtime/
completion.rs
