Expose the Run retry configuration

python/restate/context.py

@@ -79,6 +79,19 @@ def clear(self, name: str) -> None:
     def clear_all(self) -> None:
         """clear all the values in the store."""
 
+@dataclass
+class RunExponentialBackoffRetryConfig:
+    """
+    Exponential backoff Retry Configuration.
+
+    Attributes:
+        initial_interval: The initial interval in milliseconds to wait before the first retry.
+        max_attempts: The maximum number of attempts to retry.
+        max_duration: The maximum duration in milliseconds to retry.
+    """
+    initial_interval: typing.Optional[int]
+    max_attempts: typing.Optional[int]
+    max_duration: typing.Optional[timedelta]
 
 class Context(abc.ABC):
     """
@@ -95,7 +108,8 @@ def request(self) -> Request:
     def run(self,
             name: str,
             action: RunAction[T],
-            serde: Serde[T] = JsonSerde()) -> Awaitable[T]:
+            serde: Serde[T] = JsonSerde(),
+            retry: typing.Optional[RunExponentialBackoffRetryConfig] = None) -> Awaitable[T]:
         """
         Runs the given action with the given name.
         """

python/restate/server_context.py

@@ -16,12 +16,12 @@
 import typing
 import traceback
 
-from restate.context import DurablePromise, ObjectContext, Request
+from restate.context import DurablePromise, ObjectContext, Request, RunExponentialBackoffRetryConfig
 from restate.exceptions import TerminalError
 from restate.handler import Handler, handler_from_callable, invoke_handler
 from restate.serde import BytesSerde, JsonSerde, Serde
 from restate.server_types import Receive, Send
-from restate.vm import Failure, Invocation, NotReady, SuspendedException, VMWrapper
+from restate.vm import Failure, Invocation, NotReady, SuspendedException, VMWrapper, RunRetryConfig
 
 
 T = TypeVar('T')
@@ -230,7 +230,8 @@ def request(self) -> Request:
     async def run(self,
                   name: str,
                   action: Callable[[], T] | Callable[[], Awaitable[T]],
-                  serde: Optional[Serde[T]] = JsonSerde()) -> T:
+                  serde: Optional[Serde[T]] = JsonSerde(),
+                  retry: typing.Optional[RunExponentialBackoffRetryConfig] = None) -> T:
         assert serde is not None
         res = self.vm.sys_run_enter(name)
         if isinstance(res, Failure):
@@ -254,6 +255,18 @@ async def run(self,
             await self.create_poll_coroutine(handle)
             # unreachable
             assert False
+        except Exception as e:
+            if retry is None:
+                raise e
+            failure = Failure(code=500, message=str(e))
+            max_duration_ms = None if retry.max_duration is None else int(retry.max_duration.total_seconds() * 1000)
+            config = RunRetryConfig(initial_interval=retry.initial_interval,
+                                    max_attempts=retry.max_attempts,
+                                    max_duration=max_duration_ms)
+            self.vm.sys_run_exit_transient(failure=failure,
+                                                    config=config,
+                                                    attempt_duration_ms=1)
+            raise e
 
     def sleep(self, delta: timedelta) -> Awaitable[None]:
         # convert timedelta to milliseconds