diff --git a/python/restate/_preview.py b/python/restate/_preview.py index 78bbb6c1..d3ce425c 100644 --- a/python/restate/_preview.py +++ b/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]: diff --git a/python/restate/handler.py b/python/restate/handler.py index b274ba51..a2fe6772 100644 --- a/python/restate/handler.py +++ b/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, ) diff --git a/python/restate/object.py b/python/restate/object.py index 06b0b6ed..a8d726aa 100644 --- a/python/restate/object.py +++ b/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: diff --git a/python/restate/service.py b/python/restate/service.py index f9eb7827..41ad5764 100644 --- a/python/restate/service.py +++ b/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: diff --git a/python/restate/workflow.py b/python/restate/workflow.py index 4a20173d..7a6a1d93 100644 --- a/python/restate/workflow.py +++ b/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: diff --git a/tests/explicit_recovery.py b/tests/explicit_recovery.py index 6ea8cdc4..9ed02c5d 100644 --- a/tests/explicit_recovery.py +++ b/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