Add entry names and explicit recovery preview

Cargo.toml

@@ -14,4 +14,4 @@ doc = false
 [dependencies]
 pyo3 = { version = "0.25.1", features = ["extension-module"] }
 tracing-subscriber = { version = "0.3", features = ["fmt", "env-filter"] }
-restate-sdk-shared-core = { version = "7.0.3", features = ["request_identity", "sha2_random_seed", "rust_crypto"] }
+restate-sdk-shared-core = { version = "8.1.0", features = ["request_identity", "sha2_random_seed", "rust_crypto"] }

python/restate/_preview.py

@@ -0,0 +1,55 @@
+#
+#  Copyright (c) 2023-2024 - Restate Software, Inc., Restate GmbH
+#
+#  This file is part of the Restate SDK for Python,
+#  which is released under the MIT license.
+#
+#  You can find a copy of the license in file LICENSE in the root
+#  directory of this repository or package, or at
+#  https://github.com/restatedev/sdk-typescript/blob/main/LICENSE
+#
+"""Preview apis: they might change or be removed without notice."""
+
+import typing
+
+from .handler import handler_from_callable
+from .service import Service
+from .object import VirtualObject
+from .workflow import Workflow
+
+T = typing.TypeVar("T")
+
+
+def explicit_recovery(target: T) -> T:
+    """
+    Run the handlers with explicit recovery: a retried or resumed invocation doesn't replay the journal of its
+    earlier attempts, but runs the handler again from the start, executing every context operation anew,
+    e.g. reading the current state.
+
+    The handler takes care of recovering from the effects of the earlier attempts, which might repeat (at least once):
+
+    * Each attempt's calls and sends are new requests. To reach the invocation an earlier attempt started,
+      give it an idempotency key that is the same on every attempt, such as one made from `ctx.request().id`.
+    * `ctx.run` executes its action again on every attempt.
+    * Awakeables are not supported.
+
+    Apply it to a service, virtual object or workflow to enable it for all its handlers,
+    or as a decorator on top of a handler to enable it for that handler only.
+
+    Example:
+        counter = explicit_recovery(VirtualObject("counter"))
+
+        greeter = Service("greeter")
+
+        @explicit_recovery
+        @greeter.handler()
+        async def greet(ctx: Context, name: str) -> str:
+            ...
+
+    NOTE: This API is in preview and might change without notice.
+    """
+    if isinstance(target, (Service, VirtualObject, Workflow)):
+        target.explicit_recovery = True
+    else:
+        handler_from_callable(target).explicit_recovery = True  # type: ignore
+    return target

python/restate/context.py

@@ -178,7 +178,11 @@ class KeyValueStore(abc.ABC):
 
     @abc.abstractmethod
     def get(
-        self, name: str, serde: Serde[T] = DefaultSerde(), type_hint: Optional[typing.Type[T]] = None
+        self,
+        name: str,
+        serde: Serde[T] = DefaultSerde(),
+        type_hint: Optional[typing.Type[T]] = None,
+        entry_name: Optional[str] = None,
     ) -> Awaitable[Optional[T]]:
         """
         Retrieves the value associated with the given name.
@@ -188,23 +192,48 @@ def get(
             serde: The serialization/deserialization mechanism. - if the default serde is used, a default serializer will be used based on the type.
                     See also 'type_hint'.
             type_hint: The type hint of the return value. This is used to pick the serializer. If None, the type hint will be inferred from the action's return type, or the provided serializer.
+            entry_name: The name of the journal entry, shown in the Restate journal for observability.
         """
 
     @abc.abstractmethod
-    def state_keys(self) -> Awaitable[List[str]]:
-        """Returns the list of keys in the store."""
+    def state_keys(self, entry_name: Optional[str] = None) -> Awaitable[List[str]]:
+        """
+        Returns the list of keys in the store.
+
+        Args:
+            entry_name: The name of the journal entry, shown in the Restate journal for observability.
+        """
 
     @abc.abstractmethod
-    def set(self, name: str, value: T, serde: Serde[T] = DefaultSerde()) -> None:
-        """set the value associated with the given name."""
+    def set(self, name: str, value: T, serde: Serde[T] = DefaultSerde(), entry_name: Optional[str] = None) -> None:
+        """
+        set the value associated with the given name.
+
+        Args:
+            name: The state name
+            value: The value to set
+            serde: The serialization/deserialization mechanism.
+            entry_name: The name of the journal entry, shown in the Restate journal for observability.
+        """
 
     @abc.abstractmethod
-    def clear(self, name: str) -> None:
-        """clear the value associated with the given name."""
+    def clear(self, name: str, entry_name: Optional[str] = None) -> None:
+        """
+        clear the value associated with the given name.
+
+        Args:
+            name: The state name
+            entry_name: The name of the journal entry, shown in the Restate journal for observability.
+        """
 
     @abc.abstractmethod
-    def clear_all(self) -> None:
-        """clear all the values in the store."""
+    def clear_all(self, entry_name: Optional[str] = None) -> None:
+        """
+        clear all the values in the store.
+
+        Args:
+            entry_name: The name of the journal entry, shown in the Restate journal for observability.
+        """
 
 
 # pylint: disable=R0903
@@ -797,7 +826,11 @@ def key(self) -> str:
 
     @abc.abstractmethod
     def get(
-        self, name: str, serde: Serde[T] = DefaultSerde(), type_hint: Optional[typing.Type[T]] = None
+        self,
+        name: str,
+        serde: Serde[T] = DefaultSerde(),
+        type_hint: Optional[typing.Type[T]] = None,
+        entry_name: Optional[str] = None,
     ) -> RestateDurableFuture[Optional[T]]:
         """
         Retrieves the value associated with the given name.
@@ -807,12 +840,16 @@ def get(
             serde: The serialization/deserialization mechanism. - if the default serde is used, a default serializer will be used based on the type.
                     See also 'type_hint'.
             type_hint: The type hint of the return value. This is used to pick the serializer. If None, the type hint will be inferred from the action's return type, or the provided serializer.
+            entry_name: The name of the journal entry, shown in the Restate journal for observability.
         """
 
     @abc.abstractmethod
-    def state_keys(self) -> Awaitable[List[str]]:
+    def state_keys(self, entry_name: Optional[str] = None) -> Awaitable[List[str]]:
         """
         Returns the list of keys in the store.
+
+        Args:
+            entry_name: The name of the journal entry, shown in the Restate journal for observability.
         """
 
 

python/restate/handler.py

@@ -239,6 +239,7 @@ class Handler(Generic[I, O]):
         enable_lazy_state: If true, lazy state is enabled.
         ingress_private: If true, the handler cannot be invoked from the HTTP nor Kafka ingress.
         invocation_retry_policy: Optional retry policy configuration applied to this handler.
+        explicit_recovery: If true, the handler runs with explicit recovery, see restate._preview.explicit_recovery.
     """
 
     service_tag: ServiceTag
@@ -258,6 +259,7 @@ class Handler(Generic[I, O]):
     ingress_private: Optional[bool] = None
     invocation_retry_policy: Optional[InvocationRetryPolicy] = None
     context_managers: Optional[List[Callable[[], AsyncContextManager[None]]]] = None
+    explicit_recovery: bool = False
 
 
 # disable too many arguments warning

python/restate/object.py

@@ -101,6 +101,7 @@ def __init__(
         self.ingress_private = ingress_private
         self.invocation_retry_policy = invocation_retry_policy
         self.context_managers = invocation_context_managers
+        self.explicit_recovery = False
 
     @property
     def name(self):

python/restate/server.py

@@ -320,7 +320,11 @@ async def app(scope: Scope, receive: Receive, send: Send):
             try:
                 journal_codec = await get_codec()
                 await process_invocation_to_completion(
-                    VMWrapper(request_headers, disable_payload_checks=disable_payload_checks),
+                    VMWrapper(
+                        request_headers,
+                        disable_payload_checks=disable_payload_checks,
+                        explicit_recovery=handler.explicit_recovery or service.explicit_recovery,
+                    ),
                     handler,
                     dict(request_headers),
                     receive_channel,

python/restate/server_context.py

@@ -766,33 +766,37 @@ async def inv_id_factory():
         return ServerCallDurableFuture(self, handle, self._create_fetch_result_coroutine(handle, serde), inv_id_factory)
 
     def get(
-        self, name: str, serde: Serde[T] = DefaultSerde(), type_hint: Optional[typing.Type[T]] = None
+        self,
+        name: str,
+        serde: Serde[T] = DefaultSerde(),
+        type_hint: Optional[typing.Type[T]] = None,
+        entry_name: Optional[str] = None,
     ) -> Awaitable[Optional[T]]:
-        handle = self.vm.sys_get_state(name)
+        handle = self.vm.sys_get_state(name, entry_name)
         update_restate_context_is_replaying(self.vm)
         if isinstance(serde, DefaultSerde):
             serde = serde.with_maybe_type(type_hint)
         return self.create_future(handle, serde)  # type: ignore
 
-    def state_keys(self) -> Awaitable[List[str]]:
-        handle = self.vm.sys_get_state_keys()
+    def state_keys(self, entry_name: Optional[str] = None) -> Awaitable[List[str]]:
+        handle = self.vm.sys_get_state_keys(entry_name)
         update_restate_context_is_replaying(self.vm)
         return self.create_future(handle)
 
-    def set(self, name: str, value: T, serde: Serde[T] = DefaultSerde()) -> None:
+    def set(self, name: str, value: T, serde: Serde[T] = DefaultSerde(), entry_name: Optional[str] = None) -> None:
         """Set the value associated with the given name."""
         if isinstance(serde, DefaultSerde):
             serde = serde.with_maybe_type(type(value))
         buffer = self._encode(serde.serialize(value))
-        self.vm.sys_set_state(name, bytes(buffer))
+        self.vm.sys_set_state(name, bytes(buffer), entry_name)
         update_restate_context_is_replaying(self.vm)
 
-    def clear(self, name: str) -> None:
-        self.vm.sys_clear_state(name)
+    def clear(self, name: str, entry_name: Optional[str] = None) -> None:
+        self.vm.sys_clear_state(name, entry_name)
         update_restate_context_is_replaying(self.vm)
 
-    def clear_all(self) -> None:
-        self.vm.sys_clear_all_state()
+    def clear_all(self, entry_name: Optional[str] = None) -> None:
+        self.vm.sys_clear_all_state(entry_name)
         update_restate_context_is_replaying(self.vm)
 
     def request(self) -> Request:

python/restate/service.py

@@ -95,6 +95,7 @@ def __init__(
         self.ingress_private = ingress_private
         self.invocation_retry_policy = invocation_retry_policy
         self.context_managers = invocation_context_managers
+        self.explicit_recovery = False
 
     @property
     def name(self):

python/restate/vm.py

@@ -205,8 +205,13 @@ class VMWrapper:
     It provides a type-friendly interface to our shared vm.
     """
 
-    def __init__(self, headers: typing.List[typing.Tuple[str, str]], disable_payload_checks: bool = False):
-        self.vm = PyVM(headers, disable_payload_checks)
+    def __init__(
+        self,
+        headers: typing.List[typing.Tuple[str, str]],
+        disable_payload_checks: bool = False,
+        explicit_recovery: bool = False,
+    ):
+        self.vm = PyVM(headers, disable_payload_checks, explicit_recovery)
 
     def get_response_head(self) -> typing.Tuple[int, typing.Iterable[typing.Tuple[str, str]]]:
         """
@@ -345,47 +350,52 @@ def sys_write_output_failure(self, output: Failure):
         res = PyFailure(output.code, output.message, metadata=metadata)
         self.vm.sys_write_output_failure(res)
 
-    def sys_get_state(self, name) -> int:
+    def sys_get_state(self, name, entry_name: typing.Optional[str] = None) -> int:
         """
         Retrieves a key-value binding.
 
         Args:
             name: The name of the value to be retrieved.
+            entry_name: The name of the journal entry.
 
         Returns:
             The value associated with the given name.
         """
-        return self.vm.sys_get_state(name)
+        return self.vm.sys_get_state(name, entry_name)
 
-    def sys_get_state_keys(self) -> int:
+    def sys_get_state_keys(self, entry_name: typing.Optional[str] = None) -> int:
         """
         Retrieves all keys.
 
+        Args:
+            entry_name: The name of the journal entry.
+
         Returns:
             The state keys
         """
-        return self.vm.sys_get_state_keys()
+        return self.vm.sys_get_state_keys(entry_name)
 
-    def sys_set_state(self, name: str, value: bytes):
+    def sys_set_state(self, name: str, value: bytes, entry_name: typing.Optional[str] = None):
         """
         Sets a key-value binding.
 
         Args:
             name: The name of the value to be set.
             value: The value to be set.
+            entry_name: The name of the journal entry.
 
         Returns:
             None
         """
-        self.vm.sys_set_state(name, value)
+        self.vm.sys_set_state(name, value, entry_name)
 
-    def sys_clear_state(self, name: str):
+    def sys_clear_state(self, name: str, entry_name: typing.Optional[str] = None):
         """Clear the state associated with the given name."""
-        self.vm.sys_clear_state(name)
+        self.vm.sys_clear_state(name, entry_name)
 
-    def sys_clear_all_state(self):
+    def sys_clear_all_state(self, entry_name: typing.Optional[str] = None):
         """Clear the state associated with the given name."""
-        self.vm.sys_clear_all_state()
+        self.vm.sys_clear_all_state(entry_name)
 
     def sys_sleep(self, millis: int, name: typing.Optional[str] = None):
         """Ask to sleep for a given duration"""

python/restate/workflow.py

@@ -107,6 +107,7 @@ def __init__(
         self.ingress_private = ingress_private
         self.invocation_retry_policy = invocation_retry_policy
         self.context_managers = invocation_context_managers
+        self.explicit_recovery = False
 
     @property
     def name(self):