restate_sdk_shared_core/vm/
errors.rs

1use crate::error::NotificationMetadata;
2use crate::fmt::{display_closed_error, DiffFormatter};
3use crate::service_protocol::messages::{CommandMessageHeaderDiff, ErrorBehavior, RestateMessage};
4use crate::service_protocol::{
5    CompletionId, ContentTypeError, DecodingError, MessageType, NotificationId,
6};
7use crate::{Error, Version};
8use std::borrow::Cow;
9use std::collections::{HashMap, HashSet};
10use std::fmt;
11// Error codes
12
13#[derive(Copy, Clone, PartialEq, Eq)]
14pub struct InvocationErrorCode(u16);
15
16impl InvocationErrorCode {
17    pub const fn new(code: u16) -> Self {
18        InvocationErrorCode(code)
19    }
20
21    pub const fn code(self) -> u16 {
22        self.0
23    }
24}
25
26impl fmt::Debug for InvocationErrorCode {
27    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
28        write!(f, "{}", self.0)
29    }
30}
31
32impl fmt::Display for InvocationErrorCode {
33    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
34        fmt::Debug::fmt(self, f)
35    }
36}
37
38impl From<u16> for InvocationErrorCode {
39    fn from(value: u16) -> Self {
40        InvocationErrorCode(value)
41    }
42}
43
44impl From<u32> for InvocationErrorCode {
45    fn from(value: u32) -> Self {
46        value
47            .try_into()
48            .map(InvocationErrorCode)
49            .unwrap_or(codes::INTERNAL)
50    }
51}
52
53impl From<InvocationErrorCode> for u16 {
54    fn from(value: InvocationErrorCode) -> Self {
55        value.0
56    }
57}
58
59impl From<InvocationErrorCode> for u32 {
60    fn from(value: InvocationErrorCode) -> Self {
61        value.0 as u32
62    }
63}
64
65pub mod codes {
66    use super::InvocationErrorCode;
67
68    pub const BAD_REQUEST: InvocationErrorCode = InvocationErrorCode(400);
69    pub const INTERNAL: InvocationErrorCode = InvocationErrorCode(500);
70    pub const UNSUPPORTED_MEDIA_TYPE: InvocationErrorCode = InvocationErrorCode(415);
71    pub const JOURNAL_MISMATCH: InvocationErrorCode = InvocationErrorCode(570);
72    pub const PROTOCOL_VIOLATION: InvocationErrorCode = InvocationErrorCode(571);
73    pub const AWAITING_TWO_ASYNC_RESULTS: InvocationErrorCode = InvocationErrorCode(572);
74    pub const UNSUPPORTED_FEATURE: InvocationErrorCode = InvocationErrorCode(573);
75    pub const CLOSED: InvocationErrorCode = InvocationErrorCode(598);
76    pub const SUSPENDED: InvocationErrorCode = InvocationErrorCode(599);
77}
78
79// Const errors
80
81impl Error {
82    const fn new_const(code: InvocationErrorCode, message: &'static str) -> Self {
83        Error {
84            code: code.0,
85            message: Cow::Borrowed(message),
86            stacktrace: String::new(),
87            related_command: None,
88            next_retry_delay: None,
89            behavior: ErrorBehavior::Retry,
90        }
91    }
92}
93
94pub const MISSING_CONTENT_TYPE: Error = Error::new_const(
95    codes::UNSUPPORTED_MEDIA_TYPE,
96    "Missing content type when invoking the service deployment",
97);
98
99pub const UNEXPECTED_INPUT_MESSAGE: Error = Error::new_const(
100    codes::PROTOCOL_VIOLATION,
101    "Expected incoming message to be an entry",
102);
103
104pub const KNOWN_ENTRIES_IS_ZERO: Error =
105    Error::new_const(codes::INTERNAL, "Known entries is zero, expected >= 1");
106
107pub const UNEXPECTED_ENTRY_MESSAGE: Error = Error::new_const(
108    codes::PROTOCOL_VIOLATION,
109    "Expected entry messages only when waiting replay entries",
110);
111
112pub const INPUT_CLOSED_WHILE_WAITING_ENTRIES: Error = Error::new_const(
113    codes::PROTOCOL_VIOLATION,
114    "The input was closed while still waiting to receive all journal to replay",
115);
116
117pub const EMPTY_IDEMPOTENCY_KEY: Error = Error::new_const(
118    codes::INTERNAL,
119    "Trying to execute an idempotent request with an empty idempotency key. The idempotency key must be non-empty.",
120);
121
122pub const EMPTY_LIMIT_KEY: Error = Error::new_const(
123    codes::INTERNAL,
124    "Trying to execute a request with an empty limit key. The limit key must be non-empty.",
125);
126
127pub const EMPTY_SCOPE: Error = Error::new_const(
128    codes::INTERNAL,
129    "Trying to execute a request with an empty scope. The scope must be non-empty.",
130);
131
132pub const SUSPENDED: Error = Error::new_const(codes::SUSPENDED, "Suspended invocation");
133
134// Other errors
135
136#[derive(Debug, Clone, thiserror::Error)]
137#[error("The execution replay ended unexpectedly. Expecting to read '{expected}' from the recorded journal, but the buffered messages were already drained.")]
138pub struct UnavailableEntryError {
139    expected: MessageType,
140}
141
142impl UnavailableEntryError {
143    pub fn new(expected: MessageType) -> Self {
144        Self { expected }
145    }
146}
147
148#[derive(Debug, thiserror::Error)]
149#[error("Unexpected state '{state:?}' when invoking '{event:?}'")]
150pub struct UnexpectedStateError {
151    state: &'static str,
152    event: String,
153}
154
155impl UnexpectedStateError {
156    pub fn new(state: &'static str, event: String) -> Self {
157        Self { state, event }
158    }
159}
160
161#[derive(Debug)]
162pub struct ClosedError {
163    event: String,
164}
165
166impl ClosedError {
167    pub fn new(event: String) -> Self {
168        Self { event }
169    }
170}
171
172impl fmt::Display for ClosedError {
173    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
174        display_closed_error(f, &self.event)
175    }
176}
177
178impl std::error::Error for ClosedError {}
179
180#[derive(Debug)]
181pub struct CommandTypeMismatchError {
182    actual: MessageType,
183    command_index: i64,
184    expected: MessageType,
185}
186
187impl CommandTypeMismatchError {
188    pub fn new(
189        command_index: i64,
190        actual: MessageType,
191        expected: MessageType,
192    ) -> CommandTypeMismatchError {
193        Self {
194            command_index,
195            actual,
196            expected,
197        }
198    }
199}
200
201impl fmt::Display for CommandTypeMismatchError {
202    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
203        write!(f,
204               "Found a mismatch between the code paths taken during the previous execution and the paths taken during this execution.
205This typically happens when some parts of the code are non-deterministic.
206 - The previous execution ran and recorded the following: '{}' (index '{}')
207 - The current execution attempts to perform the following: '{}'",
208               self.expected,
209            self.command_index,
210               self.actual,
211        )
212    }
213}
214
215impl std::error::Error for CommandTypeMismatchError {}
216
217#[derive(Debug)]
218pub struct CommandMismatchError<M> {
219    command_index: i64,
220    actual: M,
221    expected: M,
222}
223
224impl<M> CommandMismatchError<M> {
225    pub fn new(command_index: i64, actual: M, expected: M) -> CommandMismatchError<M> {
226        Self {
227            command_index,
228            actual,
229            expected,
230        }
231    }
232}
233
234impl<M: RestateMessage + CommandMessageHeaderDiff> fmt::Display for CommandMismatchError<M> {
235    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
236        write!(f,
237"Found a mismatch between the code paths taken during the previous execution and the paths taken during this execution.
238This typically happens when some parts of the code are non-deterministic.
239- The mismatch happened while executing '{}' (index '{}')
240- Difference:",
241            M::ty(), self.command_index,
242        )?;
243        self.actual
244            .write_diff(&self.expected, DiffFormatter::new(f, "   "))
245    }
246}
247
248impl<M: RestateMessage + CommandMessageHeaderDiff + std::fmt::Debug> std::error::Error
249    for CommandMismatchError<M>
250{
251}
252
253#[derive(Debug)]
254pub struct UncompletedDoProgressDuringReplay {
255    notification_ids: Vec<NotificationId>,
256    additional_known_metadata: HashMap<NotificationId, NotificationMetadata>,
257}
258
259impl UncompletedDoProgressDuringReplay {
260    pub(crate) fn new(
261        notification_ids: HashSet<NotificationId>,
262        additional_known_metadata: HashMap<NotificationId, NotificationMetadata>,
263    ) -> Self {
264        // Order notifications: completions first (by id), then named signals, then unnamed signals (awakeables by id), then built-in signals last
265        let mut ordered_notification_ids = Vec::from_iter(notification_ids);
266        ordered_notification_ids.sort_by(|a, b| match (a, b) {
267            (NotificationId::CompletionId(a_id), NotificationId::CompletionId(b_id)) => {
268                a_id.cmp(b_id)
269            }
270            (NotificationId::CompletionId(_), _) => std::cmp::Ordering::Less,
271            (_, NotificationId::CompletionId(_)) => std::cmp::Ordering::Greater,
272
273            (NotificationId::SignalName(a_name), NotificationId::SignalName(b_name)) => {
274                a_name.cmp(b_name)
275            }
276            (NotificationId::SignalName(_), NotificationId::SignalId(_)) => {
277                std::cmp::Ordering::Less
278            }
279            (NotificationId::SignalId(_), NotificationId::SignalName(_)) => {
280                std::cmp::Ordering::Greater
281            }
282
283            (NotificationId::SignalId(a_id), NotificationId::SignalId(b_id)) => {
284                let a_is_cancel = *a_id == crate::service_protocol::CANCEL_SIGNAL_ID;
285                let b_is_cancel = *b_id == crate::service_protocol::CANCEL_SIGNAL_ID;
286                match (a_is_cancel, b_is_cancel) {
287                    (true, false) => std::cmp::Ordering::Greater,
288                    (false, true) => std::cmp::Ordering::Less,
289                    _ => a_id.cmp(b_id),
290                }
291            }
292        });
293        Self {
294            notification_ids: ordered_notification_ids,
295            additional_known_metadata,
296        }
297    }
298}
299
300impl fmt::Display for UncompletedDoProgressDuringReplay {
301    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
302        write!(f,
303"Found a mismatch between the code paths taken during the previous execution and the paths taken during this execution.
304'await' could not be replayed. This usually means the code was mutated adding an 'await' without registering a new service revision.
305Notifications awaited on this await point:",
306        )?;
307
308        for notification_id in &self.notification_ids {
309            write!(f, "\n - ")?;
310            if let Some(metadata) = self.additional_known_metadata.get(notification_id) {
311                write!(f, "{}", metadata)?;
312            } else {
313                match notification_id {
314                    NotificationId::CompletionId(completion_id) => {
315                        write!(f, "completion id {}", completion_id)?;
316                    }
317                    NotificationId::SignalId(signal_id) => {
318                        write!(f, "signal [{}]", signal_id)?;
319                    }
320                    NotificationId::SignalName(signal_name) => {
321                        write!(f, "Named signal: {}", signal_name)?;
322                    }
323                }
324            }
325        }
326
327        Ok(())
328    }
329}
330
331impl std::error::Error for UncompletedDoProgressDuringReplay {}
332
333#[derive(Debug, Clone, thiserror::Error)]
334#[error("Cannot convert a eager state key into UTF-8 String: {0:?}")]
335pub struct BadEagerStateKeyError(#[from] pub(crate) std::string::FromUtf8Error);
336
337pub const EMPTY_GET_EAGER_STATE: Error = Error::new_const(
338    codes::PROTOCOL_VIOLATION,
339    "Unexpected empty value variant for get eager state.",
340);
341
342pub const EMPTY_GET_EAGER_STATE_KEYS: Error = Error::new_const(
343    codes::PROTOCOL_VIOLATION,
344    "Unexpected empty value variant for state keys.",
345);
346
347#[derive(Debug, thiserror::Error)]
348#[error("Feature '{feature}' is not supported by the negotiated protocol version '{current_version}', the minimum required version is '{minimum_required_version}'")]
349pub struct UnsupportedFeatureForNegotiatedVersion {
350    feature: &'static str,
351    current_version: Version,
352    minimum_required_version: Version,
353}
354
355impl UnsupportedFeatureForNegotiatedVersion {
356    pub fn new(
357        feature: &'static str,
358        current_version: Version,
359        minimum_required_version: Version,
360    ) -> Self {
361        Self {
362            feature,
363            current_version,
364            minimum_required_version,
365        }
366    }
367}
368
369#[derive(Debug, thiserror::Error)]
370#[error("Received a run completion ack for completion id {completion_id}, but the related run was not proposed during this attempt.")]
371pub struct BadProposeRunCompletionAck {
372    completion_id: CompletionId,
373}
374
375impl BadProposeRunCompletionAck {
376    pub fn new(completion_id: CompletionId) -> Self {
377        Self { completion_id }
378    }
379}
380
381#[derive(Debug, Clone, thiserror::Error)]
382#[error("The provided duration for '{0}' is out of bounds: {1:?}")]
383pub struct OutOfBoundsDuration(
384    pub(crate) &'static str,
385    pub(crate) std::num::TryFromIntError,
386);
387
388// Conversions to VMError
389
390trait WithInvocationErrorCode {
391    fn code(&self) -> InvocationErrorCode;
392}
393
394impl<T: WithInvocationErrorCode + fmt::Display> From<T> for Error {
395    fn from(value: T) -> Self {
396        Error::new(value.code().0, value.to_string())
397    }
398}
399
400macro_rules! impl_error_code {
401    ($error_type:ident, $code:ident) => {
402        impl WithInvocationErrorCode for $error_type {
403            fn code(&self) -> InvocationErrorCode {
404                codes::$code
405            }
406        }
407    };
408}
409
410impl_error_code!(ContentTypeError, UNSUPPORTED_MEDIA_TYPE);
411impl WithInvocationErrorCode for DecodingError {
412    fn code(&self) -> InvocationErrorCode {
413        match self {
414            DecodingError::UnexpectedMessageType { .. } => codes::JOURNAL_MISMATCH,
415            _ => codes::INTERNAL,
416        }
417    }
418}
419impl_error_code!(UnavailableEntryError, PROTOCOL_VIOLATION);
420impl_error_code!(UnexpectedStateError, PROTOCOL_VIOLATION);
421impl_error_code!(ClosedError, CLOSED);
422impl_error_code!(CommandTypeMismatchError, JOURNAL_MISMATCH);
423impl_error_code!(UncompletedDoProgressDuringReplay, JOURNAL_MISMATCH);
424impl<M: RestateMessage + CommandMessageHeaderDiff> WithInvocationErrorCode
425    for CommandMismatchError<M>
426{
427    fn code(&self) -> InvocationErrorCode {
428        codes::JOURNAL_MISMATCH
429    }
430}
431impl_error_code!(BadEagerStateKeyError, INTERNAL);
432impl_error_code!(UnsupportedFeatureForNegotiatedVersion, UNSUPPORTED_FEATURE);
433impl_error_code!(BadProposeRunCompletionAck, PROTOCOL_VIOLATION);
434impl_error_code!(OutOfBoundsDuration, INTERNAL);
restate_sdk_shared_core/vm/errors.rs

restate_sdk_shared_core/vm/
errors.rs
