restate_sdk_shared_core/
lib.rs

1pub mod error;
2pub mod fmt;
3mod headers;
4#[cfg(feature = "request_identity")]
5mod request_identity;
6mod retries;
7mod service_protocol;
8#[cfg(feature = "tracing_pretty")]
9pub mod tracing_pretty;
10mod vm;
11
12use bytes::Bytes;
13use std::borrow::Cow;
14use std::time::Duration;
15
16pub use crate::retries::{OnMaxAttempts, RetryPolicy};
17pub use error::Error;
18pub use headers::HeaderMap;
19#[cfg(feature = "request_identity")]
20pub use request_identity::*;
21pub use service_protocol::Version;
22pub use vm::CoreVM;
23
24/// Options for syscalls that involve payload serialization.
25#[derive(Debug, Clone, Copy, Default)]
26#[non_exhaustive]
27pub struct PayloadOptions {
28    /// If true, skip payload byte equality checks during replay.
29    /// Use this when the serialization format is non-deterministic.
30    pub unstable_serialization: bool,
31}
32
33impl PayloadOptions {
34    /// Create options indicating stable (deterministic) serialization (default).
35    pub fn stable_serialization() -> Self {
36        Self {
37            unstable_serialization: false,
38        }
39    }
40
41    /// Create options indicating unstable (non-deterministic) serialization.
42    /// Payload byte equality will be skipped during replay.
43    pub fn unstable_serialization() -> Self {
44        Self {
45            unstable_serialization: true,
46        }
47    }
48}
49
50#[derive(Debug, Eq, PartialEq)]
51pub struct Header {
52    pub key: Cow<'static, str>,
53    pub value: Cow<'static, str>,
54}
55
56#[derive(Debug)]
57#[non_exhaustive]
58pub struct ResponseHead {
59    pub status_code: u16,
60    pub headers: Vec<Header>,
61    pub version: Version,
62}
63
64#[derive(Debug, Eq, PartialEq)]
65#[non_exhaustive]
66pub struct Input {
67    pub invocation_id: String,
68    pub random_seed: u64,
69    pub key: String,
70    pub headers: Vec<Header>,
71    pub input: Bytes,
72    /// Scope for this invocation, if it was called within a scope. Since V7.
73    pub scope: Option<String>,
74    /// Limit key for this invocation, if one was provided. Since V7.
75    pub limit_key: Option<String>,
76    /// Idempotency key for this invocation, if one was provided. Since V7.
77    pub idempotency_key: Option<String>,
78}
79
80#[derive(Debug, Copy, Clone, Eq, PartialEq)]
81#[non_exhaustive]
82pub enum CommandType {
83    Input,
84    Output,
85    GetState,
86    GetStateKeys,
87    SetState,
88    ClearState,
89    ClearAllState,
90    GetPromise,
91    PeekPromise,
92    CompletePromise,
93    Sleep,
94    Call,
95    OneWayCall,
96    SendSignal,
97    Run,
98    AttachInvocation,
99    GetInvocationOutput,
100    CompleteAwakeable,
101    CancelInvocation,
102}
103
104impl std::fmt::Display for CommandType {
105    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
106        f.write_str(match self {
107            CommandType::Input => "handler input",
108            CommandType::Output => "handler return",
109            CommandType::GetState => "get state",
110            CommandType::GetStateKeys => "get state keys",
111            CommandType::SetState => "set state",
112            CommandType::ClearState => "clear state",
113            CommandType::ClearAllState => "clear all state",
114            CommandType::GetPromise => "get promise",
115            CommandType::PeekPromise => "peek promise",
116            CommandType::CompletePromise => "complete promise",
117            CommandType::Sleep => "sleep",
118            CommandType::Call => "call",
119            CommandType::OneWayCall => "one way call/send",
120            CommandType::SendSignal => "send signal",
121            CommandType::Run => "run",
122            CommandType::AttachInvocation => "attach invocation",
123            CommandType::GetInvocationOutput => "get invocation output",
124            CommandType::CompleteAwakeable => "complete awakeable",
125            CommandType::CancelInvocation => "cancel invocation",
126        })
127    }
128}
129
130/// Used in `notify_error` to specify which command this error relates to.
131#[derive(Debug, Clone, Eq, PartialEq)]
132pub enum CommandRelationship {
133    Last,
134    Next {
135        ty: CommandType,
136        name: Option<Cow<'static, str>>,
137    },
138    Specific {
139        command_index: u32,
140        ty: CommandType,
141        name: Option<Cow<'static, str>>,
142    },
143}
144
145#[derive(Debug, Eq, PartialEq)]
146pub struct Target {
147    pub service: String,
148    pub handler: String,
149    pub key: Option<String>,
150    pub idempotency_key: Option<String>,
151    /// Scope for the invocation target. `None` means unscoped. Since V7.
152    pub scope: Option<String>,
153    /// Limit key for the invocation. Only valid if `scope` is set. If present, must be non-empty. Since V7.
154    pub limit_key: Option<String>,
155    pub headers: Vec<Header>,
156}
157
158pub const CANCEL_NOTIFICATION_HANDLE: NotificationHandle = NotificationHandle(1);
159
160#[derive(Debug, Hash, Clone, Copy, Eq, PartialEq, Ord, PartialOrd)]
161pub struct NotificationHandle(u32);
162
163impl From<u32> for NotificationHandle {
164    fn from(value: u32) -> Self {
165        NotificationHandle(value)
166    }
167}
168
169impl From<NotificationHandle> for u32 {
170    fn from(value: NotificationHandle) -> Self {
171        value.0
172    }
173}
174
175#[derive(Debug, Clone, Copy, Eq, PartialEq)]
176pub struct CallHandle {
177    pub invocation_id_notification_handle: NotificationHandle,
178    pub call_notification_handle: NotificationHandle,
179}
180
181#[derive(Debug, Clone, Copy, Eq, PartialEq)]
182pub struct SendHandle {
183    pub invocation_id_notification_handle: NotificationHandle,
184}
185
186#[derive(Debug, Eq, PartialEq, Clone, strum::IntoStaticStr)]
187pub enum Value {
188    /// a void/None/undefined success
189    Void,
190    Success(Bytes),
191    Failure(TerminalFailure),
192    /// Only returned for get_state_keys
193    StateKeys(Vec<String>),
194    /// Only returned for get_call_invocation_id
195    InvocationId(String),
196}
197
198/// Terminal failure
199#[derive(Debug, Clone, Eq, PartialEq)]
200pub struct TerminalFailure {
201    pub code: u16,
202    pub message: String,
203    pub metadata: Vec<(String, String)>,
204}
205
206#[derive(Debug, Default)]
207pub struct EntryRetryInfo {
208    /// Number of retries that happened so far for this entry.
209    pub retry_count: u32,
210    /// Time spent in the current retry loop.
211    pub retry_loop_duration: Duration,
212}
213
214#[derive(Debug, Clone)]
215pub enum RunExitResult {
216    Success(Bytes),
217    TerminalFailure(TerminalFailure),
218    RetryableFailure {
219        attempt_duration: Duration,
220        error: Error,
221    },
222}
223
224#[derive(Debug, Clone)]
225pub enum NonEmptyValue {
226    Success(Bytes),
227    Failure(TerminalFailure),
228}
229
230impl From<NonEmptyValue> for Value {
231    fn from(value: NonEmptyValue) -> Self {
232        match value {
233            NonEmptyValue::Success(s) => Value::Success(s),
234            NonEmptyValue::Failure(f) => Value::Failure(f),
235        }
236    }
237}
238
239#[derive(Debug, Eq, PartialEq)]
240pub enum AttachInvocationTarget {
241    InvocationId(String),
242    WorkflowId {
243        name: String,
244        key: String,
245        /// Scope for the invocation target. `None` means unscoped. Since V7.
246        scope: Option<String>,
247    },
248    IdempotencyId {
249        service_name: String,
250        service_key: Option<String>,
251        handler_name: String,
252        idempotency_key: String,
253        /// Scope for the invocation target. `None` means unscoped. Since V7.
254        scope: Option<String>,
255    },
256}
257
258pub type VMResult<T> = Result<T, Error>;
259
260#[derive(Debug, Copy, Clone, Default)]
261pub enum AwaitingOnPolicy {
262    SendAlways,
263    #[default]
264    DontSendWhenExecutingRun,
265    DontSend,
266}
267
268#[derive(Debug)]
269pub enum ImplicitCancellationOption {
270    Disabled,
271    Enabled {
272        cancel_children_calls: bool,
273        cancel_children_one_way_calls: bool,
274    },
275}
276
277#[derive(Debug, Default)]
278pub enum NonDeterministicChecksOption {
279    /// This will disable checking payloads (state values, rpc request, complete awakeable value),
280    /// but will still check all the other commands parameters.
281    PayloadChecksDisabled,
282    #[default]
283    Enabled,
284}
285
286/// Defines how the state machine reacts to journal mismatch (non-determinism) errors,
287/// that is errors carrying the `JOURNAL_MISMATCH` status code.
288#[derive(Debug, Default)]
289pub enum JournalMismatchRetryBehavior {
290    /// On journal mismatch error, pause the invocation instead of retrying.
291    ///
292    /// Requires service protocol V7 or newer; on older versions the error follows
293    /// the retry policy, as with [`JournalMismatchRetryBehavior::FollowRetryPolicy`].
294    Pause,
295    /// On journal mismatch error, fail the invocation terminally instead of retrying.
296    ///
297    /// Requires service protocol V7 or newer; on older versions the error follows
298    /// the retry policy, as with [`JournalMismatchRetryBehavior::FollowRetryPolicy`].
299    FailTerminally,
300    /// Just follow the retry policy for non journal mismatch.
301    #[default]
302    FollowRetryPolicy,
303}
304
305#[derive(Debug)]
306pub struct VMOptions {
307    pub implicit_cancellation: ImplicitCancellationOption,
308    pub non_determinism_checks: NonDeterministicChecksOption,
309    pub awaiting_on_policy: AwaitingOnPolicy,
310    pub journal_mismatch_retry_behavior: JournalMismatchRetryBehavior,
311}
312
313impl Default for VMOptions {
314    fn default() -> Self {
315        Self {
316            implicit_cancellation: ImplicitCancellationOption::Enabled {
317                cancel_children_calls: true,
318                cancel_children_one_way_calls: false,
319            },
320            non_determinism_checks: Default::default(),
321            awaiting_on_policy: Default::default(),
322            journal_mismatch_retry_behavior: Default::default(),
323        }
324    }
325}
326
327#[derive(Clone)]
328#[cfg_attr(test, derive(Eq, PartialEq))]
329#[non_exhaustive]
330pub enum UnresolvedFuture {
331    /// Waiting only this handle.
332    Single(NotificationHandle),
333    /// Resolve as soon as any one child future completes with success, or with failure (same as JS Promise.race).
334    FirstCompleted(Vec<UnresolvedFuture>),
335    /// Wait for every child to complete, regardless of success or failure (same as JS Promise.allSettled).
336    AllCompleted(Vec<UnresolvedFuture>),
337    /// Resolve on the first success; fail only if all children fail (same as JS Promise.any).
338    FirstSucceededOrAllFailed(Vec<UnresolvedFuture>),
339    /// Resolve when all children succeed; short-circuit on the first failure (same as JS Promise.all).
340    AllSucceededOrFirstFailed(Vec<UnresolvedFuture>),
341    /// Unknown combinator. This should be used when the future type is not known by the SDK,
342    /// or not representable by the other future types.
343    Unknown(Vec<UnresolvedFuture>),
344}
345
346impl From<NotificationHandle> for UnresolvedFuture {
347    fn from(value: NotificationHandle) -> Self {
348        Self::Single(value)
349    }
350}
351
352impl std::fmt::Debug for UnresolvedFuture {
353    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
354        fn fmt_combinator(
355            f: &mut std::fmt::Formatter<'_>,
356            name: &str,
357            children: &[UnresolvedFuture],
358        ) -> std::fmt::Result {
359            write!(f, "{name}(")?;
360            for (i, child) in children.iter().enumerate() {
361                if i > 0 {
362                    write!(f, ", ")?;
363                }
364                write!(f, "{child:?}")?;
365            }
366            write!(f, ")")
367        }
368
369        match self {
370            UnresolvedFuture::Unknown(children) => fmt_combinator(f, "unknown", children),
371            UnresolvedFuture::Single(h) => write!(f, "{}", u32::from(*h)),
372            UnresolvedFuture::FirstCompleted(children) => {
373                fmt_combinator(f, "first_completed", children)
374            }
375            UnresolvedFuture::AllCompleted(children) => {
376                fmt_combinator(f, "all_completed", children)
377            }
378            UnresolvedFuture::FirstSucceededOrAllFailed(children) => {
379                fmt_combinator(f, "first_succeeded_or_all_failed", children)
380            }
381            UnresolvedFuture::AllSucceededOrFirstFailed(children) => {
382                fmt_combinator(f, "all_succeeded_or_first_failed", children)
383            }
384        }
385    }
386}
387
388#[derive(Debug, PartialEq, Eq)]
389pub enum AwaitResponse {
390    /// Any of the futures in the tree completed.
391    ///
392    /// This doesn't mean the future is fully completed! Instead, it means a subtree of the future tree is ready to be completed using take_notification.
393    AnyCompleted,
394    /// There isn't enough information to make progress.
395    /// The SDK should call do_progress again after any of notify_input/notify_input_closed/propose_run_completion are called.
396    WaitingExternalProgress {
397        // The VM expects any of notify_input/notify_input_closed to be called.
398        waiting_input: bool,
399        // The VM expects propose_run_completion to be called.
400        waiting_run_proposal: bool,
401    },
402    /// The SDK should execute a pending run
403    ExecuteRun(NotificationHandle),
404    /// Returned only when [ImplicitCancellationOption::Enabled].
405    CancelSignalReceived,
406}
407
408#[derive(Debug, Copy, Clone, Eq, PartialEq, strum::EnumIs)]
409#[repr(u8)]
410pub enum State {
411    WaitingPreFlight = 0,
412    Replaying = 1,
413    Processing = 2,
414    Closed = 3,
415}
416
417#[derive(Debug, Clone, Eq, PartialEq)]
418pub struct AwakeableHandle {
419    pub id: String,
420    pub handle: NotificationHandle,
421}
422
423#[derive(Debug, Clone, Eq, PartialEq)]
424pub struct RunHandle {
425    /// If true, the run result will be replayed, meaning the SDK don't need to schedule the closure for execution.
426    pub replayed: bool,
427    pub handle: NotificationHandle,
428}
429
430pub trait VM: Sized {
431    fn new(request_headers: impl HeaderMap, options: VMOptions) -> VMResult<Self>;
432
433    fn get_response_head(&self) -> ResponseHead;
434
435    // --- Input stream
436
437    fn notify_input(&mut self, buffer: Bytes);
438
439    fn notify_input_closed(&mut self);
440
441    // --- Errors
442
443    fn notify_error(&mut self, error: Error, related_command: Option<CommandRelationship>);
444
445    // --- Output stream
446
447    /// Returns all the bytes currently buffered in the output buffer.
448    fn take_output(&mut self) -> Bytes;
449
450    // --- Execution start waiting point
451
452    fn is_ready_to_execute(&self) -> VMResult<bool>;
453
454    // --- Async results
455
456    fn is_completed(&self, handle: NotificationHandle) -> bool;
457
458    fn do_await(&mut self, unresolved_future: UnresolvedFuture) -> VMResult<AwaitResponse>;
459
460    fn take_notification(&mut self, handle: NotificationHandle) -> VMResult<Option<Value>>;
461
462    // --- Syscall(s)
463
464    fn sys_input(&mut self) -> VMResult<Input>;
465
466    fn sys_state_get(
467        &mut self,
468        key: String,
469        options: PayloadOptions,
470    ) -> VMResult<NotificationHandle>;
471
472    fn sys_state_get_keys(&mut self) -> VMResult<NotificationHandle>;
473
474    fn sys_state_set(&mut self, key: String, value: Bytes, options: PayloadOptions)
475        -> VMResult<()>;
476
477    fn sys_state_clear(&mut self, key: String) -> VMResult<()>;
478
479    fn sys_state_clear_all(&mut self) -> VMResult<()>;
480
481    /// Note: `now_since_unix_epoch` is only used for debugging purposes
482    fn sys_sleep(
483        &mut self,
484        name: String,
485        wake_up_time_since_unix_epoch: Duration,
486        now_since_unix_epoch: Option<Duration>,
487    ) -> VMResult<NotificationHandle>;
488
489    fn sys_call(
490        &mut self,
491        target: Target,
492        input: Bytes,
493        name: Option<String>,
494        options: PayloadOptions,
495    ) -> VMResult<CallHandle>;
496
497    fn sys_send(
498        &mut self,
499        target: Target,
500        input: Bytes,
501        execution_time_since_unix_epoch: Option<Duration>,
502        name: Option<String>,
503        options: PayloadOptions,
504    ) -> VMResult<SendHandle>;
505
506    fn sys_awakeable(&mut self) -> VMResult<AwakeableHandle>;
507
508    fn sys_complete_awakeable(
509        &mut self,
510        id: String,
511        value: NonEmptyValue,
512        options: PayloadOptions,
513    ) -> VMResult<()>;
514
515    fn create_signal_handle(&mut self, signal_name: String) -> VMResult<NotificationHandle>;
516
517    fn sys_complete_signal(
518        &mut self,
519        target_invocation_id: String,
520        signal_name: String,
521        value: NonEmptyValue,
522    ) -> VMResult<()>;
523
524    fn sys_get_promise(&mut self, key: String) -> VMResult<NotificationHandle>;
525
526    fn sys_peek_promise(&mut self, key: String) -> VMResult<NotificationHandle>;
527
528    fn sys_complete_promise(
529        &mut self,
530        key: String,
531        value: NonEmptyValue,
532        options: PayloadOptions,
533    ) -> VMResult<NotificationHandle>;
534
535    fn sys_run(&mut self, name: String) -> VMResult<RunHandle>;
536
537    fn propose_run_completion(
538        &mut self,
539        notification_handle: NotificationHandle,
540        value: RunExitResult,
541        retry_policy: RetryPolicy,
542    ) -> VMResult<()>;
543
544    fn sys_cancel_invocation(&mut self, target_invocation_id: String) -> VMResult<()>;
545
546    fn sys_attach_invocation(
547        &mut self,
548        target: AttachInvocationTarget,
549    ) -> VMResult<NotificationHandle>;
550
551    fn sys_get_invocation_output(
552        &mut self,
553        target: AttachInvocationTarget,
554    ) -> VMResult<NotificationHandle>;
555
556    fn sys_write_output(&mut self, value: NonEmptyValue, options: PayloadOptions) -> VMResult<()>;
557
558    fn sys_end(&mut self) -> VMResult<()>;
559
560    /// Returns the current state of the state machine.
561    fn state(&self) -> State;
562
563    /// Returns last command index. Returns `-1` if there was no progress in the journal.
564    fn last_command_index(&self) -> i64;
565}
566
567#[cfg(test)]
568mod tests;
restate_sdk_shared_core/lib.rs

restate_sdk_shared_core/
lib.rs
