explicit_recovery as a handler and service key

python/restate/_preview.py

@@ -14,48 +14,7 @@
 
 from .context import Context, RestateDurableFuture
 from .exceptions import TerminalError
-from .handler import handler_from_callable
 from .server_context import ServerInvocationContext
-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
 
 
 def cancellation(ctx: Context) -> RestateDurableFuture[None]:

python/restate/handler.py

@@ -255,7 +255,10 @@ class Handler(Generic[I, O]):
             True preloads all state, False preloads nothing, a list preloads only the listed keys.
         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.
+        explicit_recovery: If true, the handler runs with explicit recovery (preview): a retried or resumed
+            invocation runs the handler again from the start, instead of replaying its journal. See the
+            ``explicit_recovery`` option of the handler decorators for what that means for calls, sends,
+            ``ctx.run`` and awakeables.
         explicit_cancellation: If true, the handler runs with explicit cancellation (preview):
             see restate._preview.cancellation.
     """
@@ -304,6 +307,7 @@ def make_handler(
     ingress_private: Optional[bool] = None,
     invocation_retry_policy: Optional[InvocationRetryPolicy] = None,
     context_managers: Optional[List[Callable[[], AsyncContextManager[None]]]] = None,
+    explicit_recovery: bool = False,
     explicit_cancellation: bool = False,
 ) -> Handler[I, O]:
     """
@@ -329,6 +333,7 @@ def make_handler(
         ingress_private: If true, the handler cannot be invoked from the HTTP nor Kafka ingress.
         invocation_retry_policy: Retry policy used by Restate when invoking this handler.
             NOTE: only supported with restate-server >= 1.5.
+        explicit_recovery: If true, the handler runs with explicit recovery (preview).
         explicit_cancellation: If true, the handler runs with explicit cancellation (preview).
     """
     # try to deduce the handler name
@@ -365,6 +370,7 @@ def make_handler(
         ingress_private=ingress_private,
         invocation_retry_policy=invocation_retry_policy,
         context_managers=context_managers,
+        explicit_recovery=explicit_recovery,
         explicit_cancellation=explicit_cancellation,
     )
 

python/restate/object.py

@@ -75,6 +75,8 @@ class VirtualObject:
             NOTE: You can set this field only if you register this service against restate-server >= 1.4,
             otherwise the service discovery will fail.
         invocation_retry_policy (InvocationRetryPolicy, optional): Retry policy applied for all invocations to this virtual object.
+        explicit_recovery (bool, optional): (Preview) Explicit recovery for all handlers of this virtual object:
+            see the handler option of the same name. NOTE: in preview, it might change without notice.
         explicit_cancellation (bool, optional): (Preview) Explicit cancellation for all handlers of this virtual object:
             see the handler option of the same name. NOTE: in preview, it might change without notice.
     """
@@ -97,6 +99,7 @@ def __init__(
         invocation_context_managers: typing.Optional[
             typing.List[typing.Callable[[], typing.AsyncContextManager[None]]]
         ] = None,
+        explicit_recovery: bool = False,
         explicit_cancellation: bool = False,
     ):
         self.service_tag = ServiceTag("object", name, description, metadata)
@@ -111,7 +114,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
+        self.explicit_recovery = explicit_recovery
         self.explicit_cancellation = explicit_cancellation
 
     @property
@@ -142,6 +145,7 @@ def handler(
         invocation_context_managers: typing.Optional[
             typing.List[typing.Callable[[], typing.AsyncContextManager[None]]]
         ] = None,
+        explicit_recovery: bool = False,
         explicit_cancellation: bool = False,
     ) -> typing.Callable[[T], T]:
         """
@@ -188,6 +192,12 @@ def handler(
                 NOTE: You can set this field only if you register this service against restate-server >= 1.4,
                 otherwise the service discovery will fail.
             invocation_retry_policy (InvocationRetryPolicy, optional): Retry policy applied for all invocations to this handler.
+            explicit_recovery: (Preview) If true, a retried or resumed invocation does not replay its journal:
+                the handler runs again from the start, executing every context operation anew. So each
+                attempt's calls and sends are new requests (give them an idempotency key that is the same on
+                every attempt, such as one made from ``ctx.request().id``), ``ctx.run`` runs its action again
+                on every attempt, and awakeables are not supported.
+                NOTE: in preview, it might change without notice.
             explicit_cancellation: (Preview) If true, Restate cancelling an invocation neither fails the handler's
                 awaits nor cancels its calls: the handler awaits `restate._preview.cancellation(ctx)` and decides
                 what to stop. NOTE: in preview, it might change without notice.
@@ -236,6 +246,7 @@ def wrapped(*args, **kwargs):
                 ingress_private,
                 invocation_retry_policy,
                 combined_context_managers,
+                explicit_recovery=explicit_recovery,
                 explicit_cancellation=explicit_cancellation,
             )
             if handler.name in self.handlers:

python/restate/service.py

@@ -69,6 +69,8 @@ class Service:
             otherwise the service discovery will fail.
         invocation_retry_policy (InvocationRetryPolicy, optional): Retry policy applied for all invocations to this service.
         invocation_context_managers (List[Callable[[], AsyncContextManager[None]]], optional): List of context manager factories that will be applied to each invocation of this service.
+        explicit_recovery (bool, optional): (Preview) Explicit recovery for all handlers of this service:
+            see the handler option of the same name. NOTE: in preview, it might change without notice.
         explicit_cancellation (bool, optional): (Preview) Explicit cancellation for all handlers of this service:
             see the handler option of the same name. NOTE: in preview, it might change without notice.
     """
@@ -87,6 +89,7 @@ def __init__(
         invocation_context_managers: typing.Optional[
             typing.List[typing.Callable[[], typing.AsyncContextManager[None]]]
         ] = None,
+        explicit_recovery: bool = False,
         explicit_cancellation: bool = False,
     ) -> None:
         self.service_tag = ServiceTag("service", name, description, metadata)
@@ -98,7 +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
+        self.explicit_recovery = explicit_recovery
         self.explicit_cancellation = explicit_cancellation
 
     @property
@@ -125,6 +128,7 @@ def handler(
         invocation_context_managers: typing.Optional[
             typing.List[typing.Callable[[], typing.AsyncContextManager[None]]]
         ] = None,
+        explicit_recovery: bool = False,
         explicit_cancellation: bool = False,
     ) -> typing.Callable[[T], T]:
         """
@@ -164,6 +168,12 @@ def handler(
                 otherwise the service discovery will fail.
             invocation_retry_policy (InvocationRetryPolicy, optional): Retry policy applied for all invocations to this handler.
             invocation_context_managers: List of context manager factories that will be applied to each invocation of this handler.
+            explicit_recovery: (Preview) If true, a retried or resumed invocation does not replay its journal:
+                the handler runs again from the start, executing every context operation anew. So each
+                attempt's calls and sends are new requests (give them an idempotency key that is the same on
+                every attempt, such as one made from ``ctx.request().id``), ``ctx.run`` runs its action again
+                on every attempt, and awakeables are not supported.
+                NOTE: in preview, it might change without notice.
             explicit_cancellation: (Preview) If true, Restate cancelling an invocation neither fails the handler's
                 awaits nor cancels its calls: the handler awaits `restate._preview.cancellation(ctx)` and decides
                 what to stop. NOTE: in preview, it might change without notice.
@@ -215,6 +225,7 @@ def wrapped(*args, **kwargs):
                 ingress_private,
                 invocation_retry_policy,
                 combined_context_managers,
+                explicit_recovery=explicit_recovery,
                 explicit_cancellation=explicit_cancellation,
             )
             if handler.name in self.handlers:

python/restate/workflow.py

@@ -81,6 +81,8 @@ class Workflow:
             otherwise the service discovery will fail.
         invocation_retry_policy (InvocationRetryPolicy, optional): Retry policy applied for all invocations to this workflow.
         invocation_context_managers (List[Callable[[], AsyncContextManager[None]]], optional): List of context manager factories that will be applied to each invocation of this workflow.
+        explicit_recovery (bool, optional): (Preview) Explicit recovery for all handlers of this workflow:
+            see the handler option of the same name. NOTE: in preview, it might change without notice.
         explicit_cancellation (bool, optional): (Preview) Explicit cancellation for all handlers of this workflow:
             see the handler option of the same name. NOTE: in preview, it might change without notice.
     """
@@ -103,6 +105,7 @@ def __init__(
         invocation_context_managers: typing.Optional[
             typing.List[typing.Callable[[], typing.AsyncContextManager[None]]]
         ] = None,
+        explicit_recovery: bool = False,
         explicit_cancellation: bool = False,
     ):
         self.service_tag = ServiceTag("workflow", name, description, metadata)
@@ -117,7 +120,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
+        self.explicit_recovery = explicit_recovery
         self.explicit_cancellation = explicit_cancellation
 
     @property
@@ -146,6 +149,7 @@ def main(
         invocation_context_managers: typing.Optional[
             typing.List[typing.Callable[[], typing.AsyncContextManager[None]]]
         ] = None,
+        explicit_recovery: bool = False,
         explicit_cancellation: bool = False,
     ) -> typing.Callable[[T], T]:
         """
@@ -191,6 +195,12 @@ def main(
                 NOTE: You can set this field only if you register this service against restate-server >= 1.4,
                 otherwise the service discovery will fail.
             invocation_retry_policy (InvocationRetryPolicy, optional): Retry policy applied for all invocations to this workflow.
+            explicit_recovery: (Preview) If true, a retried or resumed invocation does not replay its journal:
+                the handler runs again from the start, executing every context operation anew. So each
+                attempt's calls and sends are new requests (give them an idempotency key that is the same on
+                every attempt, such as one made from ``ctx.request().id``), ``ctx.run`` runs its action again
+                on every attempt, and awakeables are not supported.
+                NOTE: in preview, it might change without notice.
             explicit_cancellation: (Preview) If true, Restate cancelling an invocation neither fails the handler's
                 awaits nor cancels its calls: the handler awaits `restate._preview.cancellation(ctx)` and decides
                 what to stop. NOTE: in preview, it might change without notice.
@@ -213,6 +223,7 @@ def main(
             ingress_private=ingress_private,
             invocation_retry_policy=invocation_retry_policy,
             invocation_context_managers=invocation_context_managers,
+            explicit_recovery=explicit_recovery,
             explicit_cancellation=explicit_cancellation,
         )
 
@@ -235,6 +246,7 @@ def handler(
         invocation_context_managers: typing.Optional[
             typing.List[typing.Callable[[], typing.AsyncContextManager[None]]]
         ] = None,
+        explicit_recovery: bool = False,
         explicit_cancellation: bool = False,
     ) -> typing.Callable[[T], T]:
         """
@@ -281,6 +293,12 @@ def handler(
                 otherwise the service discovery will fail.
             invocation_retry_policy (InvocationRetryPolicy, optional): Retry policy applied for all invocations to this handler.
             invocation_context_managers (List[Callable[[], AsyncContextManager[None]]], optional): List of context manager factories that will be applied to each invocation of this handler.
+            explicit_recovery: (Preview) If true, a retried or resumed invocation does not replay its journal:
+                the handler runs again from the start, executing every context operation anew. So each
+                attempt's calls and sends are new requests (give them an idempotency key that is the same on
+                every attempt, such as one made from ``ctx.request().id``), ``ctx.run`` runs its action again
+                on every attempt, and awakeables are not supported.
+                NOTE: in preview, it might change without notice.
             explicit_cancellation: (Preview) If true, Restate cancelling an invocation neither fails the handler's
                 awaits nor cancels its calls: the handler awaits `restate._preview.cancellation(ctx)` and decides
                 what to stop. NOTE: in preview, it might change without notice.
@@ -303,6 +321,7 @@ def handler(
             ingress_private,
             invocation_retry_policy,
             invocation_context_managers,
+            explicit_recovery=explicit_recovery,
             explicit_cancellation=explicit_cancellation,
         )
 
@@ -328,6 +347,7 @@ def _add_handler(
         invocation_context_managers: typing.Optional[
             typing.List[typing.Callable[[], typing.AsyncContextManager[None]]]
         ] = None,
+        explicit_recovery: bool = False,
         explicit_cancellation: bool = False,
     ) -> typing.Callable[[T], T]:
         """
@@ -423,6 +443,7 @@ def wrapped(*args, **kwargs):
                 ingress_private=ingress_private,
                 invocation_retry_policy=invocation_retry_policy,
                 context_managers=combined_context_managers,
+                explicit_recovery=explicit_recovery,
                 explicit_cancellation=explicit_cancellation,
             )
             if handler.name in self.handlers:

tests/explicit_recovery.py

@@ -16,7 +16,6 @@
 
 import restate
 from restate import Context, ObjectContext, Service, VirtualObject, Workflow, WorkflowContext
-from restate._preview import explicit_recovery
 from restate.vm import VMWrapper
 
 
@@ -36,21 +35,22 @@ async def greet(ctx: Context, name: str) -> str:
     assert not greeter.handlers["greet"].explicit_recovery
 
 
-def test_applies_to_every_handler_from_the_service():
-    counter = explicit_recovery(VirtualObject("counter"))
+def test_applies_to_every_handler_of_the_service():
+    counter = VirtualObject("counter", explicit_recovery=True)
 
     @counter.handler()
     async def add(ctx: ObjectContext, n: int) -> None:
         pass
 
     assert counter.explicit_recovery
+    assert Service("greeter", explicit_recovery=True).explicit_recovery
+    assert Workflow("flow", explicit_recovery=True).explicit_recovery
 
 
 def test_applies_to_a_single_handler():
     flow = Workflow("flow")
 
-    @explicit_recovery
-    @flow.main()
+    @flow.main(explicit_recovery=True)
     async def run(ctx: WorkflowContext) -> None:
         pass
 
@@ -63,14 +63,6 @@ async def query(ctx: WorkflowContext) -> None:
     assert not flow.handlers["query"].explicit_recovery
 
 
-def test_must_be_applied_on_top_of_the_handler_decorator():
-    async def greet(ctx: Context, name: str) -> str:
-        return name
-
-    with pytest.raises(ValueError):
-        explicit_recovery(greet)
-
-
 def test_the_vm_takes_it():
     headers = [("content-type", "application/vnd.restate.invocation.v6")]
     VMWrapper(headers, explicit_recovery=True)
@@ -80,7 +72,7 @@ def test_the_vm_takes_it():
 @pytest.mark.anyio
 async def test_a_retry_runs_the_handler_again_and_entries_are_named():
     calls = 0
-    counter = explicit_recovery(VirtualObject("counter"))
+    counter = VirtualObject("counter", explicit_recovery=True)
 
     def step() -> int:
         nonlocal calls