Reset stale actor objects using the daprd activation id

dapr/actor/runtime/_grpc_callbacks.py

@@ -26,7 +26,7 @@
     ActorRuntimeConfig,
     ActorTypeConfig,
 )
-from dapr.clients.base import DAPR_REENTRANCY_ID_HEADER
+from dapr.clients.base import DAPR_ACTOR_ACTIVATION_ID_HEADER, DAPR_REENTRANCY_ID_HEADER
 from dapr.clients.exceptions import ERROR_CODE_UNKNOWN, DaprInternalError
 from dapr.proto import api_v1
 
@@ -90,8 +90,17 @@ def _entity_config_to_proto(type_config: ActorTypeConfig) -> api_v1.ActorEntityC
 
 def extract_reentrancy_id(metadata: Mapping[str, str]) -> Optional[str]:
     """Looks up the reentrancy id header case-insensitively in callback metadata."""
+    return _find_metadata_value(metadata, DAPR_REENTRANCY_ID_HEADER)
+
+
+def extract_activation_id(metadata: Mapping[str, str]) -> Optional[str]:
+    return _find_metadata_value(metadata, DAPR_ACTOR_ACTIVATION_ID_HEADER)
+
+
+def _find_metadata_value(metadata: Mapping[str, str], header: str) -> Optional[str]:
+    header_lower = header.lower()
     for key, value in metadata.items():
-        if key.lower() == DAPR_REENTRANCY_ID_HEADER.lower():
+        if value and key.lower() == header_lower:
             return value
     return None
 

dapr/actor/runtime/actor.py

@@ -15,7 +15,7 @@
 
 import uuid
 from datetime import timedelta
-from typing import Any, Optional
+from typing import Any, Callable, Optional
 
 from dapr.actor.id import ActorId
 from dapr.actor.runtime._method_context import ActorMethodContext
@@ -156,15 +156,16 @@ async def unregister_reminder(self, name: str) -> None:
             self._runtime_ctx.actor_type_info.type_name, self.id.id, name
         )
 
-    async def _on_activate_internal(self) -> None:
+    async def _on_activate_internal(self, can_save: Optional[Callable[[], bool]] = None) -> None:
         """Clears all state cache, calls the overridden :meth:`_on_activate`,
         and then save the states.
 
         This internal callback is called when actor is activated.
         """
         await self._reset_state_internal()
         await self._on_activate()
-        await self._save_state_internal()
+        if can_save is None or can_save():
+            await self._save_state_internal()
 
     async def _on_deactivate_internal(self) -> None:
         """Clears all state cache, calls the overridden :meth:`_on_deactivate`.
@@ -181,13 +182,19 @@ async def _on_pre_actor_method_internal(self, method_context: ActorMethodContext
         """
         await self._on_pre_actor_method(method_context)
 
-    async def _on_post_actor_method_internal(self, method_context: ActorMethodContext) -> None:
+    async def _on_post_actor_method_internal(
+        self,
+        method_context: ActorMethodContext,
+        before_save: Optional[Callable[[], None]] = None,
+    ) -> None:
         """Calls the overridden :meth:`_on_post_actor_method` and
         saves the states.
 
         This internal callback is called after actor method is invoked.
         """
         await self._on_post_actor_method(method_context)
+        if before_save is not None:
+            before_save()
         await self._save_state_internal()
 
     async def _on_invoke_failed_internal(self, exception=None):

dapr/actor/runtime/grpc_host.py

@@ -35,6 +35,7 @@
     build_invoke_error_payload,
     build_reminder_fire_body,
     build_timer_fire_body,
+    extract_activation_id,
     extract_reentrancy_id,
     status_code_for_exception,
 )
@@ -522,8 +523,14 @@ async def _on_invoke(
     ) -> None:
         try:
             reentrancy_id = extract_reentrancy_id(request.metadata)
+            activation_id = extract_activation_id(request.metadata)
             result = await ActorRuntime.dispatch(
-                request.actor_type, request.actor_id, request.method, request.data, reentrancy_id
+                request.actor_type,
+                request.actor_id,
+                request.method,
+                request.data,
+                reentrancy_id,
+                activation_id=activation_id,
             )
             response = api_v1.SubscribeActorEventsRequestInvokeResponseAlpha1(
                 id=request.id,
@@ -555,8 +562,13 @@ async def _on_reminder(
     ) -> None:
         try:
             body = build_reminder_fire_body(request)
+            activation_id = extract_activation_id(request.metadata)
             await ActorRuntime.fire_reminder(
-                request.actor_type, request.actor_id, request.name, body
+                request.actor_type,
+                request.actor_id,
+                request.name,
+                body,
+                activation_id=activation_id,
             )
         except Exception as exception:  # noqa: BLE001
             await self._send_request_failed(session, request.id, exception)
@@ -573,7 +585,14 @@ async def _on_timer(
     ) -> None:
         try:
             body = build_timer_fire_body(request)
-            await ActorRuntime.fire_timer(request.actor_type, request.actor_id, request.name, body)
+            activation_id = extract_activation_id(request.metadata)
+            await ActorRuntime.fire_timer(
+                request.actor_type,
+                request.actor_id,
+                request.name,
+                body,
+                activation_id=activation_id,
+            )
         except Exception as exception:  # noqa: BLE001
             await self._send_request_failed(session, request.id, exception)
             return

dapr/actor/runtime/manager.py

@@ -14,8 +14,10 @@
 """
 
 import asyncio
+import logging
 import uuid
-from typing import Any, Callable, Coroutine, Dict, Optional
+from functools import partial
+from typing import Any, Callable, Coroutine, Dict, NamedTuple, Optional
 
 from dapr.actor.id import ActorId
 from dapr.actor.runtime._method_context import ActorMethodContext
@@ -26,41 +28,67 @@
 from dapr.clients.exceptions import DaprInternalError
 from dapr.common.reentrancy_context import reentrancy_ctx
 
+logger = logging.getLogger(__name__)
+
 TIMER_METHOD_NAME = 'fire_timer'
 REMINDER_METHOD_NAME = 'receive_reminder'
 
 
+def _describe_activation(activation_id: Optional[str]) -> str:
+    return f'activation {activation_id}' if activation_id else 'an activation with no id'
+
+
+class _ActiveActor(NamedTuple):
+    actor: Actor
+    activation_id: Optional[str]
+
+
 class ActorManager:
     """A Actor Manager manages actors of a specific actor type."""
 
     def __init__(self, ctx: ActorRuntimeContext):
         self._runtime_ctx = ctx
         self._message_serializer = self._runtime_ctx.message_serializer
 
-        self._active_actors: Dict[str, Actor] = {}
+        self._active_actors: Dict[str, _ActiveActor] = {}
         self._active_actors_lock = asyncio.Lock()
 
         self._dispatcher = ActorMethodDispatcher(ctx.actor_type_info)
         self._timer_method_context = ActorMethodContext.create_for_timer(TIMER_METHOD_NAME)
         self._reminder_method_context = ActorMethodContext.create_for_reminder(REMINDER_METHOD_NAME)
 
-    async def activate_actor(self, actor_id: ActorId):
+    async def activate_actor(self, actor_id: ActorId, activation_id: Optional[str] = None) -> Actor:
         """Activates actor."""
         actor = self._runtime_ctx.create_actor(actor_id)
-        await actor._on_activate_internal()
+        await actor._on_activate_internal(lambda: actor_id.id not in self._active_actors)
 
         async with self._active_actors_lock:
-            self._active_actors[actor_id.id] = actor
+            registered = self._active_actors.get(actor_id.id)
+            if registered is None:
+                self._active_actors[actor_id.id] = _ActiveActor(actor, activation_id or None)
+                return actor
+
+        await self._deactivate_discarded(actor_id, actor)
+        if activation_id and activation_id != registered.activation_id:
+            registered_by = _describe_activation(registered.activation_id)
+            raise DaprInternalError(
+                f'{actor_id} was activated by {registered_by} during this activation'
+            )
+        return registered.actor
 
     async def deactivate_actor(self, actor_id: ActorId):
         async with self._active_actors_lock:
-            deactivated_actor = self._active_actors.pop(actor_id.id, None)
-            if not deactivated_actor:
+            deactivated = self._active_actors.pop(actor_id.id, None)
+            if not deactivated:
                 raise ValueError(f'{actor_id} is not activated')
-        await deactivated_actor._on_deactivate_internal()
+        await deactivated.actor._on_deactivate_internal()
 
     async def fire_reminder(
-        self, actor_id: ActorId, reminder_name: str, request_body: bytes
+        self,
+        actor_id: ActorId,
+        reminder_name: str,
+        request_body: bytes,
+        activation_id: Optional[str] = None,
     ) -> None:
         if not self._runtime_ctx.actor_type_info.is_remindable():
             raise ValueError(
@@ -84,19 +112,33 @@ async def invoke_reminder(actor: Actor) -> Optional[bytes]:
                 )
             return None
 
-        await self._dispatch_internal(actor_id, self._reminder_method_context, invoke_reminder)
+        await self._dispatch_internal(
+            actor_id, self._reminder_method_context, invoke_reminder, activation_id
+        )
 
-    async def fire_timer(self, actor_id: ActorId, timer_name: str, request_body: bytes) -> None:
+    async def fire_timer(
+        self,
+        actor_id: ActorId,
+        timer_name: str,
+        request_body: bytes,
+        activation_id: Optional[str] = None,
+    ) -> None:
         timer = self._message_serializer.deserialize(request_body, object)
 
         async def invoke_timer(actor: Actor) -> Optional[bytes]:
             await actor._fire_timer_internal(timer['callback'], timer['data'])
             return None
 
-        await self._dispatch_internal(actor_id, self._timer_method_context, invoke_timer)
+        await self._dispatch_internal(
+            actor_id, self._timer_method_context, invoke_timer, activation_id
+        )
 
     async def dispatch(
-        self, actor_id: ActorId, actor_method_name: str, request_body: bytes
+        self,
+        actor_id: ActorId,
+        actor_method_name: str,
+        request_body: bytes,
+        activation_id: Optional[str] = None,
     ) -> bytes:
         method_context = ActorMethodContext.create_for_actor(actor_method_name)
         arg_types = self._dispatcher.get_arg_types(actor_method_name)
@@ -113,30 +155,37 @@ async def invoke_method(actor):
                 rtnval = await self._dispatcher.dispatch(actor, actor_method_name)
             return rtnval
 
-        rtn_obj = await self._dispatch_internal(actor_id, method_context, invoke_method)
+        rtn_obj = await self._dispatch_internal(
+            actor_id, method_context, invoke_method, activation_id
+        )
         return self._message_serializer.serialize(rtn_obj)
 
     async def _dispatch_internal(
         self,
         actor_id: ActorId,
         method_context: ActorMethodContext,
         dispatch_action: Callable[[Actor], Coroutine[Any, Any, Optional[bytes]]],
+        activation_id: Optional[str] = None,
     ) -> object:
-        # Activate actor when actor is invoked.
-        if actor_id.id not in self._active_actors:
-            await self.activate_actor(actor_id)
-        actor = None
+        await self._deactivate_if_stale(actor_id, activation_id)
         async with self._active_actors_lock:
-            actor = self._active_actors.get(actor_id.id, None)
-        if not actor:
-            raise ValueError(f'{actor_id} is not activated')
+            active = self._active_actors.get(actor_id.id)
+        if active is not None and activation_id and activation_id != active.activation_id:
+            registered_by = _describe_activation(active.activation_id)
+            raise DaprInternalError(f'{actor_id} was activated by {registered_by} during this call')
+        if active is None:
+            actor = await self.activate_actor(actor_id, activation_id)
+        else:
+            actor = active.actor
 
         try:
             if reentrancy_ctx.get(None) is not None:
                 actor._state_manager.set_state_context(str(uuid.uuid4()))
             await actor._on_pre_actor_method_internal(method_context)
             retval = await dispatch_action(actor)
-            await actor._on_post_actor_method_internal(method_context)
+            await actor._on_post_actor_method_internal(
+                method_context, partial(self._ensure_still_registered, actor_id, actor)
+            )
         except DaprInternalError as ex:
             await actor._on_invoke_failed_internal(ex)
             raise ex
@@ -145,3 +194,29 @@ async def _dispatch_internal(
                 actor._state_manager.set_state_context(None)
 
         return retval
+
+    def _ensure_still_registered(self, actor_id: ActorId, actor: Actor) -> None:
+        active = self._active_actors.get(actor_id.id)
+        if active is None or active.actor is not actor:
+            raise DaprInternalError(
+                f'{actor_id} was deactivated or replaced during the call, state not saved'
+            )
+
+    async def _deactivate_if_stale(self, actor_id: ActorId, activation_id: Optional[str]) -> None:
+        """Daprd can re-activate without a DELETE after a SIGKILL or OOM restart."""
+        if not activation_id:
+            return
+        async with self._active_actors_lock:
+            stale = self._active_actors.get(actor_id.id)
+            if stale is None or stale.activation_id == activation_id:
+                return
+            del self._active_actors[actor_id.id]
+        await self._deactivate_discarded(actor_id, stale.actor)
+
+    async def _deactivate_discarded(self, actor_id: ActorId, actor: Actor) -> None:
+        """Logs instead of raising so the unrelated call does not fail."""
+        actor._state_manager._discard()
+        try:
+            await actor._on_deactivate_internal()
+        except Exception:  # noqa: BLE001
+            logger.exception('Deactivating discarded object for %s failed', actor_id)