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/// Ephemeral completion id, identifying an ephemeral command and the ephemeral notification completing it, e.g. [VM::ephemeral_state_get].
176///
177/// Ephemeral commands are neither recorded in the journal, nor replayed.
178/// Their result is delivered as an ephemeral notification, see [AwaitResponse::EphemeralNotificationReady].
179///
180/// Ids are scoped to the current attempt and are unrelated to [NotificationHandle]s.
181#[derive(Debug, Hash, Clone, Copy, Eq, PartialEq, Ord, PartialOrd)]
182pub struct EphemeralCompletionId(u32);
183
184impl From<u32> for EphemeralCompletionId {
185    fn from(value: u32) -> Self {
186        EphemeralCompletionId(value)
187    }
188}
189
190impl From<EphemeralCompletionId> for u32 {
191    fn from(value: EphemeralCompletionId) -> Self {
192        value.0
193    }
194}
195
196#[derive(Debug, Clone, Copy, Eq, PartialEq)]
197pub struct CallHandle {
198    pub invocation_id_notification_handle: NotificationHandle,
199    pub call_notification_handle: NotificationHandle,
200}
201
202#[derive(Debug, Clone, Copy, Eq, PartialEq)]
203pub struct SendHandle {
204    pub invocation_id_notification_handle: NotificationHandle,
205}
206
207#[derive(Debug, Eq, PartialEq, Clone, strum::IntoStaticStr)]
208pub enum Value {
209    /// a void/None/undefined success
210    Void,
211    Success(Bytes),
212    Failure(TerminalFailure),
213    /// Only returned for get_state_keys
214    StateKeys(Vec<String>),
215    /// Only returned for get_call_invocation_id
216    InvocationId(String),
217}
218
219/// Terminal failure
220#[derive(Debug, Clone, Eq, PartialEq)]
221pub struct TerminalFailure {
222    pub code: u16,
223    pub message: String,
224    pub metadata: Vec<(String, String)>,
225}
226
227#[derive(Debug, Default)]
228pub struct EntryRetryInfo {
229    /// Number of retries that happened so far for this entry.
230    pub retry_count: u32,
231    /// Time spent in the current retry loop.
232    pub retry_loop_duration: Duration,
233}
234
235#[derive(Debug, Clone)]
236pub enum RunExitResult {
237    Success(Bytes),
238    TerminalFailure(TerminalFailure),
239    RetryableFailure {
240        attempt_duration: Duration,
241        error: Error,
242    },
243}
244
245#[derive(Debug, Clone)]
246pub enum NonEmptyValue {
247    Success(Bytes),
248    Failure(TerminalFailure),
249}
250
251impl From<NonEmptyValue> for Value {
252    fn from(value: NonEmptyValue) -> Self {
253        match value {
254            NonEmptyValue::Success(s) => Value::Success(s),
255            NonEmptyValue::Failure(f) => Value::Failure(f),
256        }
257    }
258}
259
260#[derive(Debug, Eq, PartialEq)]
261pub enum AttachInvocationTarget {
262    InvocationId(String),
263    WorkflowId {
264        name: String,
265        key: String,
266        /// Scope for the invocation target. `None` means unscoped. Since V7.
267        scope: Option<String>,
268    },
269    IdempotencyId {
270        service_name: String,
271        service_key: Option<String>,
272        handler_name: String,
273        idempotency_key: String,
274        /// Scope for the invocation target. `None` means unscoped. Since V7.
275        scope: Option<String>,
276    },
277}
278
279pub type VMResult<T> = Result<T, Error>;
280
281#[derive(Debug, Copy, Clone, Default)]
282pub enum AwaitingOnPolicy {
283    SendAlways,
284    #[default]
285    DontSendWhenExecutingRun,
286    DontSend,
287}
288
289#[derive(Debug)]
290pub enum ImplicitCancellationOption {
291    Disabled,
292    Enabled {
293        cancel_children_calls: bool,
294        cancel_children_one_way_calls: bool,
295    },
296}
297
298#[derive(Debug, Default)]
299pub enum NonDeterministicChecksOption {
300    /// This will disable checking payloads (state values, rpc request, complete awakeable value),
301    /// but will still check all the other commands parameters.
302    PayloadChecksDisabled,
303    #[default]
304    Enabled,
305}
306
307/// Defines how the state machine reacts to journal mismatch (non-determinism) errors,
308/// that is errors carrying the `JOURNAL_MISMATCH` status code.
309#[derive(Debug, Default)]
310pub enum JournalMismatchRetryBehavior {
311    /// On journal mismatch error, pause the invocation instead of retrying.
312    ///
313    /// Requires service protocol V7 or newer; on older versions the error follows
314    /// the retry policy, as with [`JournalMismatchRetryBehavior::FollowRetryPolicy`].
315    Pause,
316    /// On journal mismatch error, fail the invocation terminally instead of retrying.
317    ///
318    /// Requires service protocol V7 or newer; on older versions the error follows
319    /// the retry policy, as with [`JournalMismatchRetryBehavior::FollowRetryPolicy`].
320    FailTerminally,
321    /// Just follow the retry policy for non journal mismatch.
322    #[default]
323    FollowRetryPolicy,
324}
325
326/// What a retried or resumed invocation does with the journal of its earlier attempts.
327#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
328pub enum JournalReplay {
329    /// Replay it: commands return their recorded results, and must match them.
330    #[default]
331    Enabled,
332    /// Skip it: the handler runs again from the start and every command executes anew,
333    /// reading current state. Effects of earlier attempts may repeat (at least once), and
334    /// awakeables are unsupported.
335    Disabled,
336}
337
338/// How repeated deliveries of the same signal are handed out to handles.
339#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
340pub enum SignalDelivery {
341    /// Every delivery is kept and handed out in journal order, one per take.
342    #[default]
343    Queued,
344    /// A delivery not yet taken is replaced by the next delivery with the same signal id,
345    /// so earlier values can be lost.
346    Overwrite,
347}
348
349#[derive(Debug)]
350pub struct VMOptions {
351    pub implicit_cancellation: ImplicitCancellationOption,
352    pub non_determinism_checks: NonDeterministicChecksOption,
353    pub awaiting_on_policy: AwaitingOnPolicy,
354    pub journal_mismatch_retry_behavior: JournalMismatchRetryBehavior,
355    pub journal_replay: JournalReplay,
356    pub signal_delivery: SignalDelivery,
357}
358
359impl Default for VMOptions {
360    fn default() -> Self {
361        Self {
362            implicit_cancellation: ImplicitCancellationOption::Enabled {
363                cancel_children_calls: true,
364                cancel_children_one_way_calls: false,
365            },
366            non_determinism_checks: Default::default(),
367            awaiting_on_policy: Default::default(),
368            journal_mismatch_retry_behavior: Default::default(),
369            journal_replay: Default::default(),
370            signal_delivery: Default::default(),
371        }
372    }
373}
374
375#[derive(Clone)]
376#[cfg_attr(test, derive(Eq, PartialEq))]
377#[non_exhaustive]
378pub enum UnresolvedFuture {
379    /// Waiting only this handle.
380    Single(NotificationHandle),
381    /// Resolve as soon as any one child future completes with success, or with failure (same as JS Promise.race).
382    FirstCompleted(Vec<UnresolvedFuture>),
383    /// Wait for every child to complete, regardless of success or failure (same as JS Promise.allSettled).
384    AllCompleted(Vec<UnresolvedFuture>),
385    /// Resolve on the first success; fail only if all children fail (same as JS Promise.any).
386    FirstSucceededOrAllFailed(Vec<UnresolvedFuture>),
387    /// Resolve when all children succeed; short-circuit on the first failure (same as JS Promise.all).
388    AllSucceededOrFirstFailed(Vec<UnresolvedFuture>),
389    /// Unknown combinator. This should be used when the future type is not known by the SDK,
390    /// or not representable by the other future types.
391    Unknown(Vec<UnresolvedFuture>),
392}
393
394impl From<NotificationHandle> for UnresolvedFuture {
395    fn from(value: NotificationHandle) -> Self {
396        Self::Single(value)
397    }
398}
399
400impl std::fmt::Debug for UnresolvedFuture {
401    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
402        fn fmt_combinator(
403            f: &mut std::fmt::Formatter<'_>,
404            name: &str,
405            children: &[UnresolvedFuture],
406        ) -> std::fmt::Result {
407            write!(f, "{name}(")?;
408            for (i, child) in children.iter().enumerate() {
409                if i > 0 {
410                    write!(f, ", ")?;
411                }
412                write!(f, "{child:?}")?;
413            }
414            write!(f, ")")
415        }
416
417        match self {
418            UnresolvedFuture::Unknown(children) => fmt_combinator(f, "unknown", children),
419            UnresolvedFuture::Single(h) => write!(f, "{}", u32::from(*h)),
420            UnresolvedFuture::FirstCompleted(children) => {
421                fmt_combinator(f, "first_completed", children)
422            }
423            UnresolvedFuture::AllCompleted(children) => {
424                fmt_combinator(f, "all_completed", children)
425            }
426            UnresolvedFuture::FirstSucceededOrAllFailed(children) => {
427                fmt_combinator(f, "first_succeeded_or_all_failed", children)
428            }
429            UnresolvedFuture::AllSucceededOrFirstFailed(children) => {
430                fmt_combinator(f, "all_succeeded_or_first_failed", children)
431            }
432        }
433    }
434}
435
436#[derive(Debug, PartialEq, Eq)]
437pub enum AwaitResponse {
438    /// Any of the futures in the tree completed.
439    ///
440    /// 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.
441    AnyCompleted,
442    /// There isn't enough information to make progress.
443    /// The SDK should call do_progress again after any of notify_input/notify_input_closed/propose_run_completion are called.
444    WaitingExternalProgress {
445        // The VM expects any of notify_input/notify_input_closed to be called.
446        waiting_input: bool,
447        // The VM expects propose_run_completion to be called.
448        waiting_run_proposal: bool,
449    },
450    /// The SDK should execute a pending run
451    ExecuteRun(NotificationHandle),
452    /// Returned only when [ImplicitCancellationOption::Enabled].
453    CancelSignalReceived,
454    /// The notification of an ephemeral command is ready, see [VM::ephemeral_state_get].
455    ///
456    /// This takes precedence over any other response.
457    /// The SDK should take the notification using [VM::take_ephemeral_notification], complete the related SDK promise,
458    /// and then call [VM::do_await] again.
459    EphemeralNotificationReady(EphemeralCompletionId),
460}
461
462#[derive(Debug, Copy, Clone, Eq, PartialEq, strum::EnumIs)]
463#[repr(u8)]
464pub enum State {
465    WaitingPreFlight = 0,
466    Replaying = 1,
467    Processing = 2,
468    Closed = 3,
469}
470
471#[derive(Debug, Clone, Eq, PartialEq)]
472pub struct AwakeableHandle {
473    pub id: String,
474    pub handle: NotificationHandle,
475}
476
477#[derive(Debug, Clone, Eq, PartialEq)]
478pub struct RunHandle {
479    /// If true, the run result will be replayed, meaning the SDK don't need to schedule the closure for execution.
480    pub replayed: bool,
481    pub handle: NotificationHandle,
482}
483
484pub trait VM: Sized {
485    fn new(request_headers: impl HeaderMap, options: VMOptions) -> VMResult<Self>;
486
487    fn get_response_head(&self) -> ResponseHead;
488
489    // --- Input stream
490
491    fn notify_input(&mut self, buffer: Bytes);
492
493    fn notify_input_closed(&mut self);
494
495    // --- Errors
496
497    fn notify_error(&mut self, error: Error, related_command: Option<CommandRelationship>);
498
499    // --- Output stream
500
501    /// Returns all the bytes currently buffered in the output buffer.
502    fn take_output(&mut self) -> Bytes;
503
504    // --- Execution start waiting point
505
506    fn is_ready_to_execute(&self) -> VMResult<bool>;
507
508    // --- Async results
509
510    fn is_completed(&self, handle: NotificationHandle) -> bool;
511
512    fn do_await(&mut self, unresolved_future: UnresolvedFuture) -> VMResult<AwaitResponse>;
513
514    fn take_notification(&mut self, handle: NotificationHandle) -> VMResult<Option<Value>>;
515
516    // --- Syscall(s)
517
518    fn sys_input(&mut self) -> VMResult<Input>;
519
520    fn sys_state_get(
521        &mut self,
522        key: String,
523        name: Option<String>,
524        options: PayloadOptions,
525    ) -> VMResult<NotificationHandle>;
526
527    fn sys_state_get_keys(&mut self, name: Option<String>) -> VMResult<NotificationHandle>;
528
529    /// Ephemeral command to get the state value of the given key: neither the command nor its notification are recorded in the journal.
530    ///
531    /// The value is answered locally when known (eager state, or state modified by this invocation),
532    /// otherwise it's requested to the runtime. In both cases, the result is made available through [VM::do_await],
533    /// returning [AwaitResponse::EphemeralNotificationReady] with the returned id. Then the SDK should take the value with [VM::take_ephemeral_notification].
534    ///
535    /// Because the result is not recorded, the SDK should use it only in contexts where its usage is recorded otherwise, e.g. within a `ctx.run`.
536    ///
537    /// This operation is allowed only in the processing phase, and requires protocol V8.
538    fn ephemeral_state_get(&mut self, key: String) -> VMResult<EphemeralCompletionId>;
539
540    /// Take the notification of an ephemeral command, see [VM::ephemeral_state_get].
541    ///
542    /// Returns [Value::Void] when the state is empty, [Value::Success] otherwise.
543    /// Returns `None` if the notification is not ready yet, or it was already taken.
544    fn take_ephemeral_notification(
545        &mut self,
546        completion_id: EphemeralCompletionId,
547    ) -> VMResult<Option<Value>>;
548
549    fn sys_state_set(
550        &mut self,
551        key: String,
552        value: Bytes,
553        name: Option<String>,
554        options: PayloadOptions,
555    ) -> VMResult<()>;
556
557    fn sys_state_clear(&mut self, key: String, name: Option<String>) -> VMResult<()>;
558
559    fn sys_state_clear_all(&mut self, name: Option<String>) -> VMResult<()>;
560
561    /// Note: `now_since_unix_epoch` is only used for debugging purposes
562    fn sys_sleep(
563        &mut self,
564        name: String,
565        wake_up_time_since_unix_epoch: Duration,
566        now_since_unix_epoch: Option<Duration>,
567    ) -> VMResult<NotificationHandle>;
568
569    fn sys_call(
570        &mut self,
571        target: Target,
572        input: Bytes,
573        name: Option<String>,
574        options: PayloadOptions,
575    ) -> VMResult<CallHandle>;
576
577    fn sys_send(
578        &mut self,
579        target: Target,
580        input: Bytes,
581        execution_time_since_unix_epoch: Option<Duration>,
582        name: Option<String>,
583        options: PayloadOptions,
584    ) -> VMResult<SendHandle>;
585
586    fn sys_awakeable(&mut self) -> VMResult<AwakeableHandle>;
587
588    fn sys_complete_awakeable(
589        &mut self,
590        id: String,
591        value: NonEmptyValue,
592        options: PayloadOptions,
593    ) -> VMResult<()>;
594
595    fn create_signal_handle(&mut self, signal_name: String) -> VMResult<NotificationHandle>;
596
597    fn sys_complete_signal(
598        &mut self,
599        target_invocation_id: String,
600        signal_name: String,
601        value: NonEmptyValue,
602    ) -> VMResult<()>;
603
604    fn sys_get_promise(&mut self, key: String) -> VMResult<NotificationHandle>;
605
606    fn sys_peek_promise(&mut self, key: String) -> VMResult<NotificationHandle>;
607
608    fn sys_complete_promise(
609        &mut self,
610        key: String,
611        value: NonEmptyValue,
612        options: PayloadOptions,
613    ) -> VMResult<NotificationHandle>;
614
615    fn sys_run(&mut self, name: String) -> VMResult<RunHandle>;
616
617    fn propose_run_completion(
618        &mut self,
619        notification_handle: NotificationHandle,
620        value: RunExitResult,
621        retry_policy: RetryPolicy,
622    ) -> VMResult<()>;
623
624    fn sys_cancel_invocation(&mut self, target_invocation_id: String) -> VMResult<()>;
625
626    fn sys_attach_invocation(
627        &mut self,
628        target: AttachInvocationTarget,
629    ) -> VMResult<NotificationHandle>;
630
631    fn sys_get_invocation_output(
632        &mut self,
633        target: AttachInvocationTarget,
634    ) -> VMResult<NotificationHandle>;
635
636    fn sys_write_output(&mut self, value: NonEmptyValue, options: PayloadOptions) -> VMResult<()>;
637
638    fn sys_end(&mut self) -> VMResult<()>;
639
640    /// Returns the current state of the state machine.
641    fn state(&self) -> State;
642
643    /// Returns last command index. Returns `-1` if there was no progress in the journal.
644    fn last_command_index(&self) -> i64;
645}
646
647#[cfg(test)]
648mod tests;
restate_sdk_shared_core/lib.rs

restate_sdk_shared_core/
lib.rs
