lib.rs

pub mod error;
pub mod fmt;
mod headers;
#[cfg(feature = "request_identity")]
mod request_identity;
mod retries;
mod service_protocol;
#[cfg(feature = "tracing_pretty")]
pub mod tracing_pretty;
mod vm;

use bytes::Bytes;
use std::borrow::Cow;
use std::time::Duration;

pub use crate::retries::{OnMaxAttempts, RetryPolicy};
pub use error::Error;
pub use headers::HeaderMap;
#[cfg(feature = "request_identity")]
pub use request_identity::*;
pub use service_protocol::Version;
pub use vm::CoreVM;

/// Options for syscalls that involve payload serialization.
#[derive(Debug, Clone, Copy, Default)]
#[non_exhaustive]
pub struct PayloadOptions {
    /// If true, skip payload byte equality checks during replay.
    /// Use this when the serialization format is non-deterministic.
    pub unstable_serialization: bool,
}

impl PayloadOptions {
    /// Create options indicating stable (deterministic) serialization (default).
    pub fn stable_serialization() -> Self {
        Self {
            unstable_serialization: false,
        }
    }

    /// Create options indicating unstable (non-deterministic) serialization.
    /// Payload byte equality will be skipped during replay.
    pub fn unstable_serialization() -> Self {
        Self {
            unstable_serialization: true,
        }
    }
}

#[derive(Debug, Eq, PartialEq)]
pub struct Header {
    pub key: Cow<'static, str>,
    pub value: Cow<'static, str>,
}

#[derive(Debug)]
#[non_exhaustive]
pub struct ResponseHead {
    pub status_code: u16,
    pub headers: Vec<Header>,
    pub version: Version,
}

#[derive(Debug, Eq, PartialEq)]
#[non_exhaustive]
pub struct Input {
    pub invocation_id: String,
    pub random_seed: u64,
    pub key: String,
    pub headers: Vec<Header>,
    pub input: Bytes,
    /// Scope for this invocation, if it was called within a scope. Since V7.
    pub scope: Option<String>,
    /// Limit key for this invocation, if one was provided. Since V7.
    pub limit_key: Option<String>,
    /// Idempotency key for this invocation, if one was provided. Since V7.
    pub idempotency_key: Option<String>,
}

#[derive(Debug, Copy, Clone, Eq, PartialEq)]
#[non_exhaustive]
pub enum CommandType {
    Input,
    Output,
    GetState,
    GetStateKeys,
    SetState,
    ClearState,
    ClearAllState,
    GetPromise,
    PeekPromise,
    CompletePromise,
    Sleep,
    Call,
    OneWayCall,
    SendSignal,
    Run,
    AttachInvocation,
    GetInvocationOutput,
    CompleteAwakeable,
    CancelInvocation,
}

impl std::fmt::Display for CommandType {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(match self {
            CommandType::Input => "handler input",
            CommandType::Output => "handler return",
            CommandType::GetState => "get state",
            CommandType::GetStateKeys => "get state keys",
            CommandType::SetState => "set state",
            CommandType::ClearState => "clear state",
            CommandType::ClearAllState => "clear all state",
            CommandType::GetPromise => "get promise",
            CommandType::PeekPromise => "peek promise",
            CommandType::CompletePromise => "complete promise",
            CommandType::Sleep => "sleep",
            CommandType::Call => "call",
            CommandType::OneWayCall => "one way call/send",
            CommandType::SendSignal => "send signal",
            CommandType::Run => "run",
            CommandType::AttachInvocation => "attach invocation",
            CommandType::GetInvocationOutput => "get invocation output",
            CommandType::CompleteAwakeable => "complete awakeable",
            CommandType::CancelInvocation => "cancel invocation",
        })
    }
}

/// Used in `notify_error` to specify which command this error relates to.
#[derive(Debug, Clone, Eq, PartialEq)]
pub enum CommandRelationship {
    Last,
    Next {
        ty: CommandType,
        name: Option<Cow<'static, str>>,
    },
    Specific {
        command_index: u32,
        ty: CommandType,
        name: Option<Cow<'static, str>>,
    },
}

#[derive(Debug, Eq, PartialEq)]
pub struct Target {
    pub service: String,
    pub handler: String,
    pub key: Option<String>,
    pub idempotency_key: Option<String>,
    /// Scope for the invocation target. `None` means unscoped. Since V7.
    pub scope: Option<String>,
    /// Limit key for the invocation. Only valid if `scope` is set. If present, must be non-empty. Since V7.
    pub limit_key: Option<String>,
    pub headers: Vec<Header>,
}

pub const CANCEL_NOTIFICATION_HANDLE: NotificationHandle = NotificationHandle(1);

#[derive(Debug, Hash, Clone, Copy, Eq, PartialEq, Ord, PartialOrd)]
pub struct NotificationHandle(u32);

impl From<u32> for NotificationHandle {
    fn from(value: u32) -> Self {
        NotificationHandle(value)
    }
}

impl From<NotificationHandle> for u32 {
    fn from(value: NotificationHandle) -> Self {
        value.0
    }
}

#[derive(Debug, Clone, Copy, Eq, PartialEq)]
pub struct CallHandle {
    pub invocation_id_notification_handle: NotificationHandle,
    pub call_notification_handle: NotificationHandle,
}

#[derive(Debug, Clone, Copy, Eq, PartialEq)]
pub struct SendHandle {
    pub invocation_id_notification_handle: NotificationHandle,
}

#[derive(Debug, Eq, PartialEq, Clone, strum::IntoStaticStr)]
pub enum Value {
    /// a void/None/undefined success
    Void,
    Success(Bytes),
    Failure(TerminalFailure),
    /// Only returned for get_state_keys
    StateKeys(Vec<String>),
    /// Only returned for get_call_invocation_id
    InvocationId(String),
}

/// Terminal failure
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct TerminalFailure {
    pub code: u16,
    pub message: String,
    pub metadata: Vec<(String, String)>,
}

#[derive(Debug, Default)]
pub struct EntryRetryInfo {
    /// Number of retries that happened so far for this entry.
    pub retry_count: u32,
    /// Time spent in the current retry loop.
    pub retry_loop_duration: Duration,
}

#[derive(Debug, Clone)]
pub enum RunExitResult {
    Success(Bytes),
    TerminalFailure(TerminalFailure),
    RetryableFailure {
        attempt_duration: Duration,
        error: Error,
    },
}

#[derive(Debug, Clone)]
pub enum NonEmptyValue {
    Success(Bytes),
    Failure(TerminalFailure),
}

impl From<NonEmptyValue> for Value {
    fn from(value: NonEmptyValue) -> Self {
        match value {
            NonEmptyValue::Success(s) => Value::Success(s),
            NonEmptyValue::Failure(f) => Value::Failure(f),
        }
    }
}

#[derive(Debug, Eq, PartialEq)]
pub enum AttachInvocationTarget {
    InvocationId(String),
    WorkflowId {
        name: String,
        key: String,
        /// Scope for the invocation target. `None` means unscoped. Since V7.
        scope: Option<String>,
    },
    IdempotencyId {
        service_name: String,
        service_key: Option<String>,
        handler_name: String,
        idempotency_key: String,
        /// Scope for the invocation target. `None` means unscoped. Since V7.
        scope: Option<String>,
    },
}

pub type VMResult<T> = Result<T, Error>;

#[derive(Debug, Copy, Clone, Default)]
pub enum AwaitingOnPolicy {
    SendAlways,
    #[default]
    DontSendWhenExecutingRun,
    DontSend,
}

#[derive(Debug)]
pub enum ImplicitCancellationOption {
    Disabled,
    Enabled {
        cancel_children_calls: bool,
        cancel_children_one_way_calls: bool,
    },
}

#[derive(Debug, Default)]
pub enum NonDeterministicChecksOption {
    /// This will disable checking payloads (state values, rpc request, complete awakeable value),
    /// but will still check all the other commands parameters.
    PayloadChecksDisabled,
    #[default]
    Enabled,
}

/// Defines how the state machine reacts to journal mismatch (non-determinism) errors,
/// that is errors carrying the `JOURNAL_MISMATCH` status code.
#[derive(Debug, Default)]
pub enum JournalMismatchRetryBehavior {
    /// On journal mismatch error, pause the invocation instead of retrying.
    ///
    /// Requires service protocol V7 or newer; on older versions the error follows
    /// the retry policy, as with [`JournalMismatchRetryBehavior::FollowRetryPolicy`].
    Pause,
    /// On journal mismatch error, fail the invocation terminally instead of retrying.
    ///
    /// Requires service protocol V7 or newer; on older versions the error follows
    /// the retry policy, as with [`JournalMismatchRetryBehavior::FollowRetryPolicy`].
    FailTerminally,
    /// Just follow the retry policy for non journal mismatch.
    #[default]
    FollowRetryPolicy,
}

#[derive(Debug)]
pub struct VMOptions {
    pub implicit_cancellation: ImplicitCancellationOption,
    pub non_determinism_checks: NonDeterministicChecksOption,
    pub awaiting_on_policy: AwaitingOnPolicy,
    pub journal_mismatch_retry_behavior: JournalMismatchRetryBehavior,
}

impl Default for VMOptions {
    fn default() -> Self {
        Self {
            implicit_cancellation: ImplicitCancellationOption::Enabled {
                cancel_children_calls: true,
                cancel_children_one_way_calls: false,
            },
            non_determinism_checks: Default::default(),
            awaiting_on_policy: Default::default(),
            journal_mismatch_retry_behavior: Default::default(),
        }
    }
}

#[derive(Clone)]
#[cfg_attr(test, derive(Eq, PartialEq))]
#[non_exhaustive]
pub enum UnresolvedFuture {
    /// Waiting only this handle.
    Single(NotificationHandle),
    /// Resolve as soon as any one child future completes with success, or with failure (same as JS Promise.race).
    FirstCompleted(Vec<UnresolvedFuture>),
    /// Wait for every child to complete, regardless of success or failure (same as JS Promise.allSettled).
    AllCompleted(Vec<UnresolvedFuture>),
    /// Resolve on the first success; fail only if all children fail (same as JS Promise.any).
    FirstSucceededOrAllFailed(Vec<UnresolvedFuture>),
    /// Resolve when all children succeed; short-circuit on the first failure (same as JS Promise.all).
    AllSucceededOrFirstFailed(Vec<UnresolvedFuture>),
    /// Unknown combinator. This should be used when the future type is not known by the SDK,
    /// or not representable by the other future types.
    Unknown(Vec<UnresolvedFuture>),
}

impl From<NotificationHandle> for UnresolvedFuture {
    fn from(value: NotificationHandle) -> Self {
        Self::Single(value)
    }
}

impl std::fmt::Debug for UnresolvedFuture {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        fn fmt_combinator(
            f: &mut std::fmt::Formatter<'_>,
            name: &str,
            children: &[UnresolvedFuture],
        ) -> std::fmt::Result {
            write!(f, "{name}(")?;
            for (i, child) in children.iter().enumerate() {
                if i > 0 {
                    write!(f, ", ")?;
                }
                write!(f, "{child:?}")?;
            }
            write!(f, ")")
        }

        match self {
            UnresolvedFuture::Unknown(children) => fmt_combinator(f, "unknown", children),
            UnresolvedFuture::Single(h) => write!(f, "{}", u32::from(*h)),
            UnresolvedFuture::FirstCompleted(children) => {
                fmt_combinator(f, "first_completed", children)
            }
            UnresolvedFuture::AllCompleted(children) => {
                fmt_combinator(f, "all_completed", children)
            }
            UnresolvedFuture::FirstSucceededOrAllFailed(children) => {
                fmt_combinator(f, "first_succeeded_or_all_failed", children)
            }
            UnresolvedFuture::AllSucceededOrFirstFailed(children) => {
                fmt_combinator(f, "all_succeeded_or_first_failed", children)
            }
        }
    }
}

#[derive(Debug, PartialEq, Eq)]
pub enum AwaitResponse {
    /// Any of the futures in the tree completed.
    ///
    /// 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.
    AnyCompleted,
    /// There isn't enough information to make progress.
    /// The SDK should call do_progress again after any of notify_input/notify_input_closed/propose_run_completion are called.
    WaitingExternalProgress {
        // The VM expects any of notify_input/notify_input_closed to be called.
        waiting_input: bool,
        // The VM expects propose_run_completion to be called.
        waiting_run_proposal: bool,
    },
    /// The SDK should execute a pending run
    ExecuteRun(NotificationHandle),
    /// Returned only when [ImplicitCancellationOption::Enabled].
    CancelSignalReceived,
}

#[derive(Debug, Copy, Clone, Eq, PartialEq, strum::EnumIs)]
#[repr(u8)]
pub enum State {
    WaitingPreFlight = 0,
    Replaying = 1,
    Processing = 2,
    Closed = 3,
}

#[derive(Debug, Clone, Eq, PartialEq)]
pub struct AwakeableHandle {
    pub id: String,
    pub handle: NotificationHandle,
}

#[derive(Debug, Clone, Eq, PartialEq)]
pub struct RunHandle {
    /// If true, the run result will be replayed, meaning the SDK don't need to schedule the closure for execution.
    pub replayed: bool,
    pub handle: NotificationHandle,
}

pub trait VM: Sized {
    fn new(request_headers: impl HeaderMap, options: VMOptions) -> VMResult<Self>;

    fn get_response_head(&self) -> ResponseHead;

    // --- Input stream

    fn notify_input(&mut self, buffer: Bytes);

    fn notify_input_closed(&mut self);

    // --- Errors

    fn notify_error(&mut self, error: Error, related_command: Option<CommandRelationship>);

    // --- Output stream

    /// Returns all the bytes currently buffered in the output buffer.
    fn take_output(&mut self) -> Bytes;

    // --- Execution start waiting point

    fn is_ready_to_execute(&self) -> VMResult<bool>;

    // --- Async results

    fn is_completed(&self, handle: NotificationHandle) -> bool;

    fn do_await(&mut self, unresolved_future: UnresolvedFuture) -> VMResult<AwaitResponse>;

    fn take_notification(&mut self, handle: NotificationHandle) -> VMResult<Option<Value>>;

    // --- Syscall(s)

    fn sys_input(&mut self) -> VMResult<Input>;

    fn sys_state_get(
        &mut self,
        key: String,
        options: PayloadOptions,
    ) -> VMResult<NotificationHandle>;

    fn sys_state_get_keys(&mut self) -> VMResult<NotificationHandle>;

    fn sys_state_set(&mut self, key: String, value: Bytes, options: PayloadOptions)
        -> VMResult<()>;

    fn sys_state_clear(&mut self, key: String) -> VMResult<()>;

    fn sys_state_clear_all(&mut self) -> VMResult<()>;

    /// Note: `now_since_unix_epoch` is only used for debugging purposes
    fn sys_sleep(
        &mut self,
        name: String,
        wake_up_time_since_unix_epoch: Duration,
        now_since_unix_epoch: Option<Duration>,
    ) -> VMResult<NotificationHandle>;

    fn sys_call(
        &mut self,
        target: Target,
        input: Bytes,
        name: Option<String>,
        options: PayloadOptions,
    ) -> VMResult<CallHandle>;

    fn sys_send(
        &mut self,
        target: Target,
        input: Bytes,
        execution_time_since_unix_epoch: Option<Duration>,
        name: Option<String>,
        options: PayloadOptions,
    ) -> VMResult<SendHandle>;

    fn sys_awakeable(&mut self) -> VMResult<AwakeableHandle>;

    fn sys_complete_awakeable(
        &mut self,
        id: String,
        value: NonEmptyValue,
        options: PayloadOptions,
    ) -> VMResult<()>;

    fn create_signal_handle(&mut self, signal_name: String) -> VMResult<NotificationHandle>;

    fn sys_complete_signal(
        &mut self,
        target_invocation_id: String,
        signal_name: String,
        value: NonEmptyValue,
    ) -> VMResult<()>;

    fn sys_get_promise(&mut self, key: String) -> VMResult<NotificationHandle>;

    fn sys_peek_promise(&mut self, key: String) -> VMResult<NotificationHandle>;

    fn sys_complete_promise(
        &mut self,
        key: String,
        value: NonEmptyValue,
        options: PayloadOptions,
    ) -> VMResult<NotificationHandle>;

    fn sys_run(&mut self, name: String) -> VMResult<RunHandle>;

    fn propose_run_completion(
        &mut self,
        notification_handle: NotificationHandle,
        value: RunExitResult,
        retry_policy: RetryPolicy,
    ) -> VMResult<()>;

    fn sys_cancel_invocation(&mut self, target_invocation_id: String) -> VMResult<()>;

    fn sys_attach_invocation(
        &mut self,
        target: AttachInvocationTarget,
    ) -> VMResult<NotificationHandle>;

    fn sys_get_invocation_output(
        &mut self,
        target: AttachInvocationTarget,
    ) -> VMResult<NotificationHandle>;

    fn sys_write_output(&mut self, value: NonEmptyValue, options: PayloadOptions) -> VMResult<()>;

    fn sys_end(&mut self) -> VMResult<()>;

    /// Returns the current state of the state machine.
    fn state(&self) -> State;

    /// Returns last command index. Returns `-1` if there was no progress in the journal.
    fn last_command_index(&self) -> i64;
}

#[cfg(test)]
mod tests;