First pass at the implementation of side effect retry

Author: slinkydeveloper

Cargo.toml

@@ -22,7 +22,7 @@ pin-project-lite = "0.2"
 rand = { version = "0.8.5", optional = true }
 regress = "0.10"
 restate-sdk-macros = { version = "0.2.1", path = "macros" }
-restate-sdk-shared-core = { version = "0.0.5" }
+restate-sdk-shared-core = { path = "../sdk-shared-core" }
 serde = "1.0"
 serde_json = "1.0"
 thiserror = "1.0.63"

src/context/mod.rs

@@ -10,7 +10,7 @@ mod request;
 mod run;
 
 pub use request::{Request, RequestTarget};
-pub use run::RunClosure;
+pub use run::{RunClosure, RunRetryPolicy};
 pub type HeaderMap = http::HeaderMap<String>;
 
 /// Service handler context.
@@ -385,6 +385,23 @@ pub trait ContextSideEffects<'ctx>: private::SealedContext<'ctx> {
         self.inner_context().run(name, run_closure)
     }
 
+    /// Run a non-deterministic operation and record its result.
+    ///
+    fn run_with_retry<R, F, T>(
+        &self,
+        name: &'ctx str,
+        retry_policy: RunRetryPolicy,
+        run_closure: R,
+    ) -> impl Future<Output = Result<T, TerminalError>> + 'ctx
+    where
+        R: RunClosure<Fut = F, Output = T> + Send + Sync + 'ctx,
+        T: Serialize + Deserialize,
+        F: Future<Output = HandlerResult<T>> + Send + Sync + 'ctx,
+    {
+        self.inner_context()
+            .run_with_retry(name, retry_policy, run_closure)
+    }
+
     /// Return a random seed inherently predictable, based on the invocation id, which is not secret.
     ///
     /// This value is stable during the invocation lifecycle, thus across retries.

src/context/run.rs

@@ -1,6 +1,7 @@
 use crate::errors::HandlerResult;
 use crate::serde::{Deserialize, Serialize};
 use std::future::Future;
+use std::time::Duration;
 
 /// Run closure trait
 pub trait RunClosure {
@@ -23,3 +24,72 @@ where
         self()
     }
 }
+
+/// This struct represents the policy to execute retries for run closures.
+#[derive(Debug, Clone)]
+pub struct RunRetryPolicy {
+    pub(crate) initial_interval: Duration,
+    pub(crate) factor: f32,
+    pub(crate) max_interval: Option<Duration>,
+    pub(crate) max_attempts: Option<u32>,
+    pub(crate) max_duration: Option<Duration>,
+}
+
+impl Default for RunRetryPolicy {
+    fn default() -> Self {
+        Self {
+            initial_interval: Duration::from_millis(100),
+            factor: 2.0,
+            max_interval: Some(Duration::from_secs(2)),
+            max_attempts: None,
+            max_duration: Some(Duration::from_secs(50)),
+        }
+    }
+}
+
+impl RunRetryPolicy {
+    /// Create a new retry policy.
+    pub fn new() -> Self {
+        Self {
+            initial_interval: Duration::from_millis(100),
+            factor: 1.0,
+            max_interval: None,
+            max_attempts: None,
+            max_duration: None,
+        }
+    }
+
+    /// Initial interval for the first retry attempt.
+    pub fn with_initial_interval(mut self, initial_interval: Duration) -> Self {
+        self.initial_interval = initial_interval;
+        self
+    }
+
+    /// Maximum interval between retries.
+    pub fn with_factor(mut self, factor: f32) -> Self {
+        self.factor = factor;
+        self
+    }
+
+    /// Maximum interval between retries.
+    pub fn with_max_interval(mut self, max_interval: Duration) -> Self {
+        self.max_interval = Some(max_interval);
+        self
+    }
+
+    /// Gives up retrying when either this number of attempts is reached,
+    /// or `max_duration` (if set) is reached first.
+    /// Infinite retries if this field and `max_duration` are unset.
+    pub fn with_max_attempts(mut self, max_attempts: u32) -> Self {
+        self.max_attempts = Some(max_attempts);
+        self
+    }
+
+    /// Gives up retrying when either the retry loop lasted for this given max duration,
+    /// or `max_attempts` (if set) is reached first.
+    /// Infinite retries if this field and `max_attempts` are unset.
+    pub fn with_max_duration(mut self, max_duration: Duration) -> Self {
+        self.max_duration = Some(max_duration);
+        self
+    }
+}