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    /// Type of the command recorded in the journal at `command_index`.
183    actual: MessageType,
184    command_index: i64,
185    /// Type of the command the handler issues now, during replay.
186    expected: MessageType,
187}
188
189impl CommandTypeMismatchError {
190    pub fn new(
191        command_index: i64,
192        actual: MessageType,
193        expected: MessageType,
194    ) -> CommandTypeMismatchError {
195        Self {
196            command_index,
197            actual,
198            expected,
199        }
200    }
201}
202
203impl fmt::Display for CommandTypeMismatchError {
204    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
205        write!(f,
206               "Found a mismatch between the code paths taken during the previous execution and the paths taken during this execution.
207This typically happens when some parts of the code are non-deterministic.
208 - The previous execution ran and recorded the following: '{}' (index '{}')
209 - The current execution attempts to perform the following: '{}'",
210               self.actual,
211            self.command_index,
212               self.expected,
213        )
214    }
215}
216
217impl std::error::Error for CommandTypeMismatchError {}
218
219#[derive(Debug)]
220pub struct CommandMismatchError<M> {
221    command_index: i64,
222    actual: M,
223    expected: M,
224}
225
226impl<M> CommandMismatchError<M> {
227    pub fn new(command_index: i64, actual: M, expected: M) -> CommandMismatchError<M> {
228        Self {
229            command_index,
230            actual,
231            expected,
232        }
233    }
234}
235
236impl<M: RestateMessage + CommandMessageHeaderDiff> fmt::Display for CommandMismatchError<M> {
237    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
238        write!(f,
239"Found a mismatch between the code paths taken during the previous execution and the paths taken during this execution.
240This typically happens when some parts of the code are non-deterministic.
241- The mismatch happened while executing '{}' (index '{}')
242- Difference:",
243            M::ty(), self.command_index,
244        )?;
245        self.actual
246            .write_diff(&self.expected, DiffFormatter::new(f, "   "))
247    }
248}
249
250impl<M: RestateMessage + CommandMessageHeaderDiff + std::fmt::Debug> std::error::Error
251    for CommandMismatchError<M>
252{
253}
254
255#[derive(Debug)]
256pub struct UncompletedDoProgressDuringReplay {
257    notification_ids: Vec<NotificationId>,
258    additional_known_metadata: HashMap<NotificationId, NotificationMetadata>,
259}
260
261impl UncompletedDoProgressDuringReplay {
262    pub(crate) fn new(
263        notification_ids: HashSet<NotificationId>,
264        additional_known_metadata: HashMap<NotificationId, NotificationMetadata>,
265    ) -> Self {
266        // Order notifications: completions first (by id), then named signals, then unnamed signals (awakeables by id), then built-in signals last
267        let mut ordered_notification_ids = Vec::from_iter(notification_ids);
268        ordered_notification_ids.sort_by(|a, b| match (a, b) {
269            (NotificationId::CompletionId(a_id), NotificationId::CompletionId(b_id)) => {
270                a_id.cmp(b_id)
271            }
272            (NotificationId::CompletionId(_), _) => std::cmp::Ordering::Less,
273            (_, NotificationId::CompletionId(_)) => std::cmp::Ordering::Greater,
274
275            (NotificationId::SignalName(a_name), NotificationId::SignalName(b_name)) => {
276                a_name.cmp(b_name)
277            }
278            (NotificationId::SignalName(_), NotificationId::SignalId(_)) => {
279                std::cmp::Ordering::Less
280            }
281            (NotificationId::SignalId(_), NotificationId::SignalName(_)) => {
282                std::cmp::Ordering::Greater
283            }
284
285            (NotificationId::SignalId(a_id), NotificationId::SignalId(b_id)) => {
286                let a_is_cancel = *a_id == crate::service_protocol::CANCEL_SIGNAL_ID;
287                let b_is_cancel = *b_id == crate::service_protocol::CANCEL_SIGNAL_ID;
288                match (a_is_cancel, b_is_cancel) {
289                    (true, false) => std::cmp::Ordering::Greater,
290                    (false, true) => std::cmp::Ordering::Less,
291                    _ => a_id.cmp(b_id),
292                }
293            }
294        });
295        Self {
296            notification_ids: ordered_notification_ids,
297            additional_known_metadata,
298        }
299    }
300}
301
302impl fmt::Display for UncompletedDoProgressDuringReplay {
303    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
304        write!(f,
305"Found a mismatch between the code paths taken during the previous execution and the paths taken during this execution.
306'await' could not be replayed. This usually means the code was mutated adding an 'await' without registering a new service revision.
307Notifications awaited on this await point:",
308        )?;
309
310        for notification_id in &self.notification_ids {
311            write!(f, "\n - ")?;
312            if let Some(metadata) = self.additional_known_metadata.get(notification_id) {
313                write!(f, "{}", metadata)?;
314            } else {
315                match notification_id {
316                    NotificationId::CompletionId(completion_id) => {
317                        write!(f, "completion id {}", completion_id)?;
318                    }
319                    NotificationId::SignalId(signal_id) => {
320                        write!(f, "signal [{}]", signal_id)?;
321                    }
322                    NotificationId::SignalName(signal_name) => {
323                        write!(f, "Named signal: {}", signal_name)?;
324                    }
325                }
326            }
327        }
328
329        Ok(())
330    }
331}
332
333impl std::error::Error for UncompletedDoProgressDuringReplay {}
334
335#[derive(Debug, Clone, thiserror::Error)]
336#[error("Cannot convert a eager state key into UTF-8 String: {0:?}")]
337pub struct BadEagerStateKeyError(#[from] pub(crate) std::string::FromUtf8Error);
338
339pub const EMPTY_GET_EAGER_STATE: Error = Error::new_const(
340    codes::PROTOCOL_VIOLATION,
341    "Unexpected empty value variant for get eager state.",
342);
343
344pub const EMPTY_GET_EAGER_STATE_KEYS: Error = Error::new_const(
345    codes::PROTOCOL_VIOLATION,
346    "Unexpected empty value variant for state keys.",
347);
348
349#[derive(Debug, thiserror::Error)]
350#[error("Feature '{feature}' is not supported by the negotiated protocol version '{current_version}', the minimum required version is '{minimum_required_version}'")]
351pub struct UnsupportedFeatureForNegotiatedVersion {
352    feature: &'static str,
353    current_version: Version,
354    minimum_required_version: Version,
355}
356
357impl UnsupportedFeatureForNegotiatedVersion {
358    pub fn new(
359        feature: &'static str,
360        current_version: Version,
361        minimum_required_version: Version,
362    ) -> Self {
363        Self {
364            feature,
365            current_version,
366            minimum_required_version,
367        }
368    }
369}
370
371#[derive(Debug, thiserror::Error)]
372#[error("Received a run completion ack for completion id {completion_id}, but the related run was not proposed during this attempt.")]
373pub struct BadProposeRunCompletionAck {
374    completion_id: CompletionId,
375}
376
377impl BadProposeRunCompletionAck {
378    pub fn new(completion_id: CompletionId) -> Self {
379        Self { completion_id }
380    }
381}
382
383pub const EMPTY_GET_STATE_EPHEMERAL_NOTIFICATION: Error = Error::new_const(
384    codes::PROTOCOL_VIOLATION,
385    "Unexpected empty result variant for get state ephemeral notification.",
386);
387
388#[derive(Debug, thiserror::Error)]
389#[error("Cannot get state '{key}' without recording it in the journal: the input stream is closed, thus Restate cannot answer the request. Getting state without recording it in the journal requires a bidirectional stream between Restate and the service (not available in request/response mode, e.g. AWS Lambda), unless the state value is available in the eager state sent by Restate.")]
390pub struct EphemeralStateGetWithClosedInput {
391    key: String,
392}
393
394impl EphemeralStateGetWithClosedInput {
395    pub fn new(key: String) -> Self {
396        Self { key }
397    }
398}
399
400pub const INPUT_CLOSED_WHILE_WAITING_EPHEMERAL_NOTIFICATIONS: Error = Error::new_const(
401    codes::INTERNAL,
402    "The input stream was closed while waiting for the result of a get state not recorded in the journal. Getting state without recording it in the journal requires a bidirectional stream between Restate and the service (not available in request/response mode, e.g. AWS Lambda), unless the state value is available in the eager state sent by Restate.",
403);
404
405/// Ephemeral commands should really only be executed within ctx.run, so this is a "proxy safeguard" for that, because ctx.run
406/// are really only started when we're processing (just track AwaitResponse::ExecuteRun usages)
407pub const EPHEMERAL_COMMAND_DURING_REPLAY: Error = Error::new_const(
408    codes::JOURNAL_MISMATCH,
409    "Cannot execute ephemeral commands (e.g. get state without recording it in the journal) during replay, as their result might differ from the original execution. Ephemeral commands can be used only while processing new journal entries, e.g. within a ctx.run closure. This usually means the code uses them outside a ctx.run closure, or the code was changed.",
410);
411
412#[derive(Debug, Clone, thiserror::Error)]
413#[error("The provided duration for '{0}' is out of bounds: {1:?}")]
414pub struct OutOfBoundsDuration(
415    pub(crate) &'static str,
416    pub(crate) std::num::TryFromIntError,
417);
418
419// Conversions to VMError
420
421trait WithInvocationErrorCode {
422    fn code(&self) -> InvocationErrorCode;
423}
424
425impl<T: WithInvocationErrorCode + fmt::Display> From<T> for Error {
426    fn from(value: T) -> Self {
427        Error::new(value.code().0, value.to_string())
428    }
429}
430
431macro_rules! impl_error_code {
432    ($error_type:ident, $code:ident) => {
433        impl WithInvocationErrorCode for $error_type {
434            fn code(&self) -> InvocationErrorCode {
435                codes::$code
436            }
437        }
438    };
439}
440
441impl_error_code!(ContentTypeError, UNSUPPORTED_MEDIA_TYPE);
442impl WithInvocationErrorCode for DecodingError {
443    fn code(&self) -> InvocationErrorCode {
444        match self {
445            DecodingError::UnexpectedMessageType { .. } => codes::JOURNAL_MISMATCH,
446            _ => codes::INTERNAL,
447        }
448    }
449}
450impl_error_code!(UnavailableEntryError, PROTOCOL_VIOLATION);
451impl_error_code!(UnexpectedStateError, PROTOCOL_VIOLATION);
452impl_error_code!(ClosedError, CLOSED);
453impl_error_code!(CommandTypeMismatchError, JOURNAL_MISMATCH);
454impl_error_code!(UncompletedDoProgressDuringReplay, JOURNAL_MISMATCH);
455impl<M: RestateMessage + CommandMessageHeaderDiff> WithInvocationErrorCode
456    for CommandMismatchError<M>
457{
458    fn code(&self) -> InvocationErrorCode {
459        codes::JOURNAL_MISMATCH
460    }
461}
462impl_error_code!(BadEagerStateKeyError, INTERNAL);
463impl_error_code!(UnsupportedFeatureForNegotiatedVersion, UNSUPPORTED_FEATURE);
464impl_error_code!(BadProposeRunCompletionAck, PROTOCOL_VIOLATION);
465impl_error_code!(EphemeralStateGetWithClosedInput, INTERNAL);
466impl_error_code!(OutOfBoundsDuration, INTERNAL);
restate_sdk_shared_core/vm/errors.rs

restate_sdk_shared_core/vm/
errors.rs
