From 1cf8128f6c787b192113bfbea5756d34733567fc Mon Sep 17 00:00:00 2001 From: Cassandra Coyle Date: Mon, 26 Oct 2026 15:19:42 -0500 Subject: [PATCH 1/2] actor: reset actor object when daprd reports a new activation id Signed-off-by: Cassandra Coyle --- dapr/actor/runtime/_grpc_callbacks.py | 13 +- dapr/actor/runtime/actor.py | 15 +- dapr/actor/runtime/grpc_host.py | 25 +- dapr/actor/runtime/manager.py | 110 ++++- dapr/actor/runtime/runtime.py | 28 +- dapr/clients/base.py | 1 + dapr/ext/fastapi/actor.py | 19 +- dapr/ext/flask/actor.py | 25 +- tests/actor/test_actor_manager.py | 615 ++++++++++++++++++++++++++ tests/actor/test_actor_reentrancy.py | 14 +- tests/actor/test_actor_runtime.py | 87 ++++ tests/actor/test_grpc_callbacks.py | 26 ++ tests/actor/test_grpc_host.py | 133 +++++- tests/clients/fake_dapr_server.py | 7 +- tests/ext/fastapi/test_dapractor.py | 113 +++++ tests/ext/flask/test_dapractor.py | 137 ++++++ 16 files changed, 1326 insertions(+), 42 deletions(-) create mode 100644 tests/ext/flask/test_dapractor.py diff --git a/dapr/actor/runtime/_grpc_callbacks.py b/dapr/actor/runtime/_grpc_callbacks.py index 92544882f..64046e46e 100644 --- a/dapr/actor/runtime/_grpc_callbacks.py +++ b/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 diff --git a/dapr/actor/runtime/actor.py b/dapr/actor/runtime/actor.py index a666b06e1..81ff0909c 100644 --- a/dapr/actor/runtime/actor.py +++ b/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,7 +156,7 @@ 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. @@ -164,7 +164,8 @@ async def _on_activate_internal(self) -> None: """ 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): diff --git a/dapr/actor/runtime/grpc_host.py b/dapr/actor/runtime/grpc_host.py index 93cbcdca4..8f76ebfb1 100644 --- a/dapr/actor/runtime/grpc_host.py +++ b/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, ) @@ -518,8 +519,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, @@ -551,8 +558,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) @@ -569,7 +581,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 diff --git a/dapr/actor/runtime/manager.py b/dapr/actor/runtime/manager.py index 7199689ab..cda3b1a6b 100644 --- a/dapr/actor/runtime/manager.py +++ b/dapr/actor/runtime/manager.py @@ -14,7 +14,9 @@ """ import asyncio +import logging import uuid +from functools import partial from typing import Any, Callable, Coroutine, Dict, Optional from dapr.actor.id import ActorId @@ -26,10 +28,16 @@ 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 ActorManager: """A Actor Manager manages actors of a specific actor type.""" @@ -38,29 +46,51 @@ def __init__(self, ctx: ActorRuntimeContext): self._message_serializer = self._runtime_ctx.message_serializer self._active_actors: Dict[str, Actor] = {} + self._activation_ids: Dict[str, str] = {} 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] = actor + if activation_id: + self._activation_ids[actor_id.id] = activation_id + else: + self._activation_ids.pop(actor_id.id, None) + return actor + registered_id = self._activation_ids.get(actor_id.id) + + await self._deactivate_discarded(actor_id, actor) + if activation_id and activation_id != registered_id: + registered_by = _describe_activation(registered_id) + raise DaprInternalError( + f'{actor_id} was activated by {registered_by} during this activation' + ) + return registered async def deactivate_actor(self, actor_id: ActorId): async with self._active_actors_lock: deactivated_actor = self._active_actors.pop(actor_id.id, None) + self._activation_ids.pop(actor_id.id, None) if not deactivated_actor: raise ValueError(f'{actor_id} is not activated') 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 +114,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,7 +157,9 @@ 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( @@ -121,22 +167,26 @@ async def _dispatch_internal( 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') + actor = self._active_actors.get(actor_id.id) + registered_id = self._activation_ids.get(actor_id.id) + if actor is not None and activation_id and activation_id != registered_id: + registered_by = _describe_activation(registered_id) + raise DaprInternalError(f'{actor_id} was activated by {registered_by} during this call') + if actor is None: + actor = await self.activate_actor(actor_id, activation_id) 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 +195,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: + if self._active_actors.get(actor_id.id) 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: + if actor_id.id not in self._active_actors: + return + if self._activation_ids.get(actor_id.id) == activation_id: + return + stale_actor = self._active_actors.pop(actor_id.id) + self._activation_ids.pop(actor_id.id, None) + 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.""" + try: + await actor._on_deactivate_internal() + except Exception: # noqa: BLE001 + logger.exception('Deactivating discarded object for %s failed', actor_id) diff --git a/dapr/actor/runtime/runtime.py b/dapr/actor/runtime/runtime.py index 6f5462748..4705491ff 100644 --- a/dapr/actor/runtime/runtime.py +++ b/dapr/actor/runtime/runtime.py @@ -103,6 +103,7 @@ async def dispatch( actor_method_name: str, request_body: bytes, reentrancy_id: Optional[str] = None, + activation_id: Optional[str] = None, ) -> bytes: """Dispatches actor method defined in actor_type. @@ -113,6 +114,7 @@ async def dispatch( request_body (bytes): the body of request that is passed to actor method arguments. reentrancy_id (str): reentrancy ID obtained from the dapr_reentrancy_id header if present. + activation_id (str, optional): the Dapr-Actor-Activation-Id header value. Returns: bytes: serialized response data. @@ -124,11 +126,18 @@ async def dispatch( manager = await cls._get_actor_manager(actor_type_name) if not manager: raise ValueError(f'{actor_type_name} is not registered.') - return await manager.dispatch(ActorId(actor_id), actor_method_name, request_body) + return await manager.dispatch( + ActorId(actor_id), actor_method_name, request_body, activation_id + ) @classmethod async def fire_reminder( - cls, actor_type_name: str, actor_id: str, name: str, state: bytes + cls, + actor_type_name: str, + actor_id: str, + name: str, + state: bytes, + activation_id: Optional[str] = None, ) -> None: """Fires a reminder for the Actor. @@ -137,6 +146,7 @@ async def fire_reminder( actor_id (str): Actor ID. name (str): the name of reminder. state (bytes): the body of request that is passed to reminder callback. + activation_id (str, optional): the Dapr-Actor-Activation-Id header value. Raises: ValueError: `actor_type_name` actor type is not registered. @@ -145,10 +155,17 @@ async def fire_reminder( manager = await cls._get_actor_manager(actor_type_name) if not manager: raise ValueError(f'{actor_type_name} is not registered.') - await manager.fire_reminder(ActorId(actor_id), name, state) + await manager.fire_reminder(ActorId(actor_id), name, state, activation_id) @classmethod - async def fire_timer(cls, actor_type_name: str, actor_id: str, name: str, state: bytes) -> None: + async def fire_timer( + cls, + actor_type_name: str, + actor_id: str, + name: str, + state: bytes, + activation_id: Optional[str] = None, + ) -> None: """Fires a timer for the Actor. Args: @@ -156,6 +173,7 @@ async def fire_timer(cls, actor_type_name: str, actor_id: str, name: str, state: actor_id (str): Actor ID. name (str): the timer's name. state (bytes): the timer's trigger body. + activation_id (str, optional): the Dapr-Actor-Activation-Id header value. Raises: ValueError: `actor_type_name` actor type is not registered. @@ -163,7 +181,7 @@ async def fire_timer(cls, actor_type_name: str, actor_id: str, name: str, state: manager = await cls._get_actor_manager(actor_type_name) if not manager: raise ValueError(f'{actor_type_name} is not registered.') - await manager.fire_timer(ActorId(actor_id), name, state) + await manager.fire_timer(ActorId(actor_id), name, state, activation_id) @classmethod def set_actor_config(cls, config: ActorRuntimeConfig) -> None: diff --git a/dapr/clients/base.py b/dapr/clients/base.py index 6d82059c0..304d4ed2c 100644 --- a/dapr/clients/base.py +++ b/dapr/clients/base.py @@ -17,6 +17,7 @@ from typing import Optional DAPR_REENTRANCY_ID_HEADER = 'Dapr-Reentrancy-Id' +DAPR_ACTOR_ACTIVATION_ID_HEADER = 'Dapr-Actor-Activation-Id' class DaprActorClientBase(ABC): diff --git a/dapr/ext/fastapi/actor.py b/dapr/ext/fastapi/actor.py index a153581dd..28c09346c 100644 --- a/dapr/ext/fastapi/actor.py +++ b/dapr/ext/fastapi/actor.py @@ -25,6 +25,7 @@ DEFAULT_CONTENT_TYPE = 'application/json; utf-8' DAPR_REENTRANCY_ID_HEADER = 'Dapr-Reentrancy-Id' +DAPR_ACTOR_ACTIVATION_ID_HEADER = 'Dapr-Actor-Activation-Id' def _wrap_response( @@ -92,8 +93,14 @@ async def actor_method( # Read raw bytes from request stream req_body = await request.body() reentrancy_id = request.headers.get(DAPR_REENTRANCY_ID_HEADER) + activation_id = request.headers.get(DAPR_ACTOR_ACTIVATION_ID_HEADER) result = await ActorRuntime.dispatch( - actor_type_name, actor_id, method_name, req_body, reentrancy_id + actor_type_name, + actor_id, + method_name, + req_body, + reentrancy_id, + activation_id=activation_id, ) except DaprInternalError as ex: return _wrap_response(status.HTTP_500_INTERNAL_SERVER_ERROR, ex.as_json_safe_dict()) @@ -115,7 +122,10 @@ async def actor_timer( try: # Read raw bytes from request stream req_body = await request.body() - await ActorRuntime.fire_timer(actor_type_name, actor_id, timer_name, req_body) + activation_id = request.headers.get(DAPR_ACTOR_ACTIVATION_ID_HEADER) + await ActorRuntime.fire_timer( + actor_type_name, actor_id, timer_name, req_body, activation_id=activation_id + ) except DaprInternalError as ex: return _wrap_response(status.HTTP_500_INTERNAL_SERVER_ERROR, ex.as_json_safe_dict()) except Exception as ex: @@ -137,7 +147,10 @@ async def actor_reminder( try: # Read raw bytes from request stream req_body = await request.body() - await ActorRuntime.fire_reminder(actor_type_name, actor_id, reminder_name, req_body) + activation_id = request.headers.get(DAPR_ACTOR_ACTIVATION_ID_HEADER) + await ActorRuntime.fire_reminder( + actor_type_name, actor_id, reminder_name, req_body, activation_id=activation_id + ) except DaprInternalError as ex: return _wrap_response(status.HTTP_500_INTERNAL_SERVER_ERROR, ex.as_json_safe_dict()) except Exception as ex: diff --git a/dapr/ext/flask/actor.py b/dapr/ext/flask/actor.py index c8bdf1a50..eddc5ce01 100644 --- a/dapr/ext/flask/actor.py +++ b/dapr/ext/flask/actor.py @@ -24,6 +24,7 @@ DEFAULT_CONTENT_TYPE = 'application/json; utf-8' DAPR_REENTRANCY_ID_HEADER = 'Dapr-Reentrancy-Id' +DAPR_ACTOR_ACTIVATION_ID_HEADER = 'Dapr-Actor-Activation-Id' class DaprActor(object): @@ -93,9 +94,15 @@ def _method_handler(self, actor_type_name, actor_id, method_name): # Read raw bytes from request stream req_body = request.stream.read() reentrancy_id = request.headers.get(DAPR_REENTRANCY_ID_HEADER) + activation_id = request.headers.get(DAPR_ACTOR_ACTIVATION_ID_HEADER) result = asyncio.run( ActorRuntime.dispatch( - actor_type_name, actor_id, method_name, req_body, reentrancy_id + actor_type_name, + actor_id, + method_name, + req_body, + reentrancy_id, + activation_id=activation_id, ) ) except DaprInternalError as ex: @@ -111,7 +118,12 @@ def _timer_handler(self, actor_type_name, actor_id, timer_name): try: # Read raw bytes from request stream req_body = request.stream.read() - asyncio.run(ActorRuntime.fire_timer(actor_type_name, actor_id, timer_name, req_body)) + activation_id = request.headers.get(DAPR_ACTOR_ACTIVATION_ID_HEADER) + asyncio.run( + ActorRuntime.fire_timer( + actor_type_name, actor_id, timer_name, req_body, activation_id=activation_id + ) + ) except DaprInternalError as ex: return wrap_response(500, ex.as_json_safe_dict()) except Exception as ex: @@ -125,8 +137,15 @@ def _reminder_handler(self, actor_type_name, actor_id, reminder_name): try: # Read raw bytes from request stream req_body = request.stream.read() + activation_id = request.headers.get(DAPR_ACTOR_ACTIVATION_ID_HEADER) asyncio.run( - ActorRuntime.fire_reminder(actor_type_name, actor_id, reminder_name, req_body) + ActorRuntime.fire_reminder( + actor_type_name, + actor_id, + reminder_name, + req_body, + activation_id=activation_id, + ) ) except DaprInternalError as ex: return wrap_response(500, ex.as_json_safe_dict()) diff --git a/tests/actor/test_actor_manager.py b/tests/actor/test_actor_manager.py index 47ba2f2bf..2eaf5977f 100644 --- a/tests/actor/test_actor_manager.py +++ b/tests/actor/test_actor_manager.py @@ -13,14 +13,22 @@ limitations under the License. """ +import asyncio +import json import unittest from datetime import timedelta +from functools import partial +from typing import Awaitable, Callable, Dict, List, Optional from unittest import mock +from dapr.actor.actor_interface import ActorInterface, actormethod from dapr.actor.id import ActorId from dapr.actor.runtime._type_information import ActorTypeInformation +from dapr.actor.runtime.actor import Actor from dapr.actor.runtime.context import ActorRuntimeContext from dapr.actor.runtime.manager import ActorManager +from dapr.actor.runtime.remindable import Remindable +from dapr.clients.exceptions import DaprInternalError from dapr.serializers import DefaultJSONSerializer from tests.actor.fake_actor_classes import ( FakeMultiInterfacesActor, @@ -173,3 +181,610 @@ def test_fire_timer_success(self): ) self.assertTrue(actor.timer_called) + + +class ActivationTrackingActorInterface(ActorInterface): + @actormethod(name='Remember') + async def remember(self, value: object) -> None: ... + + @actormethod(name='Recall') + async def recall(self) -> object: ... + + +class ActivationTrackingActor(Actor, ActivationTrackingActorInterface, Remindable): + def __init__(self, ctx, actor_id): + super().__init__(ctx, actor_id) + self.activate_calls = 0 + self.deactivate_calls = 0 + self.reminders_received: List[str] = [] + self.timer_data_received: List[object] = [] + + async def remember(self, value: object) -> None: + await self._state_manager.set_state('value', value) + + async def recall(self) -> object: + return await self._state_manager.get_state('value') + + async def timer_callback(self, data: object) -> None: + self.timer_data_received.append(data) + + async def receive_reminder( + self, + name: str, + state: bytes, + due_time: timedelta, + period: timedelta, + ttl: Optional[timedelta], + ) -> None: + self.reminders_received.append(name) + + async def _on_activate(self) -> None: + self.activate_calls += 1 + + async def _on_deactivate(self) -> None: + self.deactivate_calls += 1 + + +STORED_VALUE = 'expected_response' +REMINDER_BODY = DefaultJSONSerializer().serialize( + {'dueTime': '0h0m1s', 'period': '0h0m1s', 'data': 'cmVtaW5kZXJfc3RhdGU='} +) +TIMER_BODY = b'{ "callback": "timer_callback", "data": "timer call" }' + + +class ActorManagerActivationIdTests(unittest.TestCase): + def setUp(self): + self._serializer = DefaultJSONSerializer() + type_info = ActorTypeInformation.create(ActivationTrackingActor) + ctx = ActorRuntimeContext( + type_info, self._serializer, self._serializer, FakeDaprActorClient() + ) + self._manager = ActorManager(ctx) + self._actor_id = ActorId('activation') + + def _remember(self, value: str, activation_id: Optional[str]) -> None: + body = self._serializer.serialize(value) + _run(self._manager.dispatch(self._actor_id, 'Remember', body, activation_id)) + + def _recall(self, activation_id: Optional[str]) -> object: + response = _run(self._manager.dispatch(self._actor_id, 'Recall', b'', activation_id)) + return self._serializer.deserialize(response) + + def _active_actor(self) -> ActivationTrackingActor: + return self._manager._active_actors[self._actor_id.id] + + def _stored_activation_id(self) -> Optional[str]: + return self._manager._activation_ids.get(self._actor_id.id) + + def test_first_call_stores_activation_id(self): + self._remember('cached', 'act-1') + + self.assertEqual('act-1', self._stored_activation_id()) + self.assertEqual(1, self._active_actor().activate_calls) + + def test_same_activation_id_reuses_actor_and_cache(self): + self._remember('cached', 'act-1') + actor_first = self._active_actor() + + recalled = self._recall('act-1') + + self.assertIs(actor_first, self._active_actor()) + self.assertEqual('cached', recalled) + self.assertEqual(0, actor_first.deactivate_calls) + + def test_new_activation_id_replaces_actor_and_drops_cache(self): + self._remember('cached', 'act-1') + actor_stale = self._active_actor() + + recalled = self._recall('act-2') + + actor_fresh = self._active_actor() + self.assertIsNot(actor_stale, actor_fresh) + self.assertEqual(1, actor_stale.deactivate_calls) + self.assertEqual({}, actor_stale._state_manager._default_state_change_tracker) + self.assertEqual(1, actor_fresh.activate_calls) + self.assertEqual(0, actor_fresh.deactivate_calls) + self.assertEqual(STORED_VALUE, recalled) + self.assertEqual('act-2', self._stored_activation_id()) + + def test_missing_activation_id_never_resets(self): + self._remember('cached', 'act-1') + actor_first = self._active_actor() + + recalled_none = self._recall(None) + recalled_empty = self._recall('') + + self.assertIs(actor_first, self._active_actor()) + self.assertEqual('cached', recalled_none) + self.assertEqual('cached', recalled_empty) + self.assertEqual(0, actor_first.deactivate_calls) + self.assertEqual('act-1', self._stored_activation_id()) + + def test_without_activation_ids_actor_is_reused(self): + self._remember('cached', None) + actor_first = self._active_actor() + + recalled = self._recall(None) + + self.assertIs(actor_first, self._active_actor()) + self.assertEqual('cached', recalled) + self.assertIsNone(self._stored_activation_id()) + + def test_stored_empty_then_activation_id_resets(self): + self._remember('cached', None) + actor_stale = self._active_actor() + + recalled = self._recall('act-1') + + self.assertIsNot(actor_stale, self._active_actor()) + self.assertEqual(1, actor_stale.deactivate_calls) + self.assertEqual(STORED_VALUE, recalled) + self.assertEqual('act-1', self._stored_activation_id()) + + def test_activate_actor_without_id_then_activation_id_resets(self): + _run(self._manager.activate_actor(self._actor_id)) + actor_stale = self._active_actor() + + self._recall('act-1') + + self.assertIsNot(actor_stale, self._active_actor()) + self.assertEqual(1, actor_stale.deactivate_calls) + + def test_deactivate_actor_clears_activation_id(self): + self._remember('cached', 'act-1') + actor_first = self._active_actor() + + _run(self._manager.deactivate_actor(self._actor_id)) + + self.assertIsNone(self._stored_activation_id()) + self.assertEqual(1, actor_first.deactivate_calls) + + self._recall('act-1') + + self.assertIsNot(actor_first, self._active_actor()) + self.assertEqual(1, actor_first.deactivate_calls) + self.assertEqual('act-1', self._stored_activation_id()) + + def test_reminder_same_activation_id_reuses_actor(self): + self._remember('cached', 'act-1') + actor_first = self._active_actor() + + _run(self._manager.fire_reminder(self._actor_id, 'r1', REMINDER_BODY, 'act-1')) + + self.assertIs(actor_first, self._active_actor()) + self.assertEqual(['r1'], actor_first.reminders_received) + self.assertEqual(0, actor_first.deactivate_calls) + + def test_reminder_new_activation_id_replaces_actor(self): + self._remember('cached', 'act-1') + actor_stale = self._active_actor() + + _run(self._manager.fire_reminder(self._actor_id, 'r1', REMINDER_BODY, 'act-2')) + + actor_fresh = self._active_actor() + self.assertIsNot(actor_stale, actor_fresh) + self.assertEqual(1, actor_stale.deactivate_calls) + self.assertEqual([], actor_stale.reminders_received) + self.assertEqual(['r1'], actor_fresh.reminders_received) + self.assertEqual('act-2', self._stored_activation_id()) + self.assertEqual(STORED_VALUE, self._recall('act-2')) + + def test_reminder_missing_activation_id_never_resets(self): + self._remember('cached', 'act-1') + actor_first = self._active_actor() + + _run(self._manager.fire_reminder(self._actor_id, 'r1', REMINDER_BODY)) + + self.assertIs(actor_first, self._active_actor()) + self.assertEqual(['r1'], actor_first.reminders_received) + + def test_timer_same_activation_id_reuses_actor(self): + self._remember('cached', 'act-1') + actor_first = self._active_actor() + + _run(self._manager.fire_timer(self._actor_id, 't1', TIMER_BODY, 'act-1')) + + self.assertIs(actor_first, self._active_actor()) + self.assertEqual(['timer call'], actor_first.timer_data_received) + self.assertEqual(0, actor_first.deactivate_calls) + + def test_timer_new_activation_id_replaces_actor(self): + self._remember('cached', 'act-1') + actor_stale = self._active_actor() + + _run(self._manager.fire_timer(self._actor_id, 't1', TIMER_BODY, 'act-2')) + + actor_fresh = self._active_actor() + self.assertIsNot(actor_stale, actor_fresh) + self.assertEqual(1, actor_stale.deactivate_calls) + self.assertEqual([], actor_stale.timer_data_received) + self.assertEqual(['timer call'], actor_fresh.timer_data_received) + self.assertEqual('act-2', self._stored_activation_id()) + self.assertEqual(STORED_VALUE, self._recall('act-2')) + + def test_timer_missing_activation_id_never_resets(self): + self._remember('cached', 'act-1') + actor_first = self._active_actor() + + _run(self._manager.fire_timer(self._actor_id, 't1', TIMER_BODY, '')) + + self.assertIs(actor_first, self._active_actor()) + self.assertEqual(['timer call'], actor_first.timer_data_received) + + def test_activation_ids_are_tracked_per_actor(self): + other_actor_id = ActorId('other') + self._remember('cached', 'act-1') + actor_first = self._active_actor() + + _run(self._manager.dispatch(other_actor_id, 'Recall', b'', 'act-other')) + + self.assertIs(actor_first, self._active_actor()) + self.assertEqual('act-1', self._stored_activation_id()) + self.assertEqual('act-other', self._manager._activation_ids[other_actor_id.id]) + + +class StoreBackedActorClient(FakeDaprActorClient): + def __init__(self): + self.store: Dict[str, bytes] = {} + + async def get_state(self, actor_type: str, actor_id: str, name: str) -> bytes: + return self.store.get(name, b'') + + async def save_state_transactionally(self, actor_type: str, actor_id: str, data: bytes) -> None: + for operation in json.loads(data): + key = operation['request']['key'] + if operation['operation'] == 'delete': + self.store.pop(key, None) + continue + self.store[key] = json.dumps(operation['request']['value']).encode('utf-8') + + +class AppendActorInterface(ActorInterface): + @actormethod(name='Append') + async def append(self, item: str) -> None: ... + + +class AppendActor(Actor, AppendActorInterface, Remindable): + def __init__(self, ctx, actor_id): + super().__init__(ctx, actor_id) + self.held: Optional[asyncio.Event] = None + self.release: Optional[asyncio.Event] = None + + async def append(self, item: str) -> None: + items_found, items = await self._state_manager.try_get_state('items') + items_current = items if items_found else [] + if self.held is not None and self.release is not None: + self.held.set() + await self.release.wait() + await self._state_manager.set_state('items', items_current + [item]) + + async def timer_callback(self, item: str) -> None: + await self.append(item) + + async def failing_timer_callback(self, item: str) -> None: + await self.append(item) + raise RuntimeError('timer failed') + + async def receive_reminder( + self, + name: str, + state: bytes, + due_time: timedelta, + period: timedelta, + ttl: Optional[timedelta], + ) -> None: + await self.append(name) + + +class ActorManagerReplacedSaveTests(unittest.TestCase): + def setUp(self): + self._serializer = DefaultJSONSerializer() + self._client = StoreBackedActorClient() + type_info = ActorTypeInformation.create(AppendActor) + ctx = ActorRuntimeContext(type_info, self._serializer, self._serializer, self._client) + self._manager = ActorManager(ctx) + self._actor_id = ActorId('replaced') + + def _stored_items(self) -> object: + return json.loads(self._client.store['items']) + + async def _append(self, item: str, activation_id: Optional[str]) -> None: + body = self._serializer.serialize(item) + await self._manager.dispatch(self._actor_id, 'Append', body, activation_id) + + async def _run_replaced_during(self, stale_call: Callable[[], Awaitable[None]]) -> None: + await self._append('x', 'act-1') + actor_stale: AppendActor = self._manager._active_actors[self._actor_id.id] + actor_stale.held = asyncio.Event() + actor_stale.release = asyncio.Event() + stale_task = asyncio.ensure_future(stale_call()) + await asyncio.wait_for(actor_stale.held.wait(), 5) + + await self._append('b', 'act-2') + actor_stale.release.set() + + with self.assertRaises(DaprInternalError): + await asyncio.wait_for(stale_task, 5) + self.assertEqual(['x', 'b'], self._stored_items()) + + def test_replaced_method_call_skips_save(self): + _run(self._run_replaced_during(lambda: self._append('a', 'act-1'))) + + def test_replaced_timer_skips_save(self): + timer_body = self._serializer.serialize({'callback': 'timer_callback', 'data': 'a'}) + stale_call = partial(self._manager.fire_timer, self._actor_id, 't1', timer_body, 'act-1') + + _run(self._run_replaced_during(stale_call)) + + def test_replaced_reminder_skips_save(self): + stale_call = partial( + self._manager.fire_reminder, self._actor_id, 'a', REMINDER_BODY, 'act-1' + ) + + _run(self._run_replaced_during(stale_call)) + + def test_raising_timer_skips_save(self): + _run(self._append('x', 'act-1')) + timer_body = self._serializer.serialize({'callback': 'failing_timer_callback', 'data': 'a'}) + + with self.assertRaises(RuntimeError): + _run(self._manager.fire_timer(self._actor_id, 't1', timer_body, 'act-1')) + + self.assertEqual(['x'], self._stored_items()) + + def test_replaced_by_missing_id_call_still_saves(self): + async def scenario() -> None: + await self._append('x', 'act-1') + actor_first: AppendActor = self._manager._active_actors[self._actor_id.id] + actor_first.held = asyncio.Event() + actor_first.release = asyncio.Event() + first_task = asyncio.ensure_future(self._append('a', 'act-1')) + await asyncio.wait_for(actor_first.held.wait(), 5) + + actor_first.held = None + await self._append('b', None) + actor_first.release.set() + await asyncio.wait_for(first_task, 5) + + _run(scenario()) + + self.assertIn('a', self._stored_items()) + + def test_deactivated_during_call_skips_save(self): + async def scenario() -> None: + await self._append('x', 'act-1') + actor_first: AppendActor = self._manager._active_actors[self._actor_id.id] + actor_first.held = asyncio.Event() + actor_first.release = asyncio.Event() + first_task = asyncio.ensure_future(self._append('a', 'act-1')) + await asyncio.wait_for(actor_first.held.wait(), 5) + + await self._manager.deactivate_actor(self._actor_id) + actor_first.release.set() + + with self.assertRaises(DaprInternalError): + await asyncio.wait_for(first_task, 5) + + _run(scenario()) + + self.assertEqual(['x'], self._stored_items()) + + +class PostHookOverrideActor(AppendActor): + def __init__(self, ctx, actor_id): + super().__init__(ctx, actor_id) + self.post_internal_calls: List[str] = [] + + async def _on_post_actor_method_internal(self, method_context, before_save=None) -> None: + self.post_internal_calls.append(method_context.method_name) + await super()._on_post_actor_method_internal(method_context, before_save) + + +class ActorInternalHookCompatTests(unittest.TestCase): + def setUp(self): + self._serializer = DefaultJSONSerializer() + self._client = StoreBackedActorClient() + + def _ctx(self, actor_class: type) -> ActorRuntimeContext: + type_info = ActorTypeInformation.create(actor_class) + return ActorRuntimeContext(type_info, self._serializer, self._serializer, self._client) + + def test_on_activate_internal_without_can_save_saves(self): + actor = GatedActivateWriteActor(self._ctx(GatedActivateWriteActor), ActorId('compat')) + + _run(actor._on_activate_internal()) + + self.assertEqual(['init'], json.loads(self._client.store['items'])) + + def test_manager_calls_overridden_post_actor_method_internal(self): + manager = ActorManager(self._ctx(PostHookOverrideActor)) + body = self._serializer.serialize('a') + + _run(manager.dispatch(ActorId('compat'), 'Append', body, 'act-1')) + + actor: PostHookOverrideActor = manager._active_actors['compat'] + self.assertEqual(['Append'], actor.post_internal_calls) + self.assertEqual(['a'], json.loads(self._client.store['items'])) + + +class GatedActivateActor(AppendActor): + activation_started: Optional[asyncio.Event] = None + activation_release: Optional[asyncio.Event] = None + + async def _on_activate(self) -> None: + await self._state_manager.try_get_state('items') + release = GatedActivateActor.activation_release + if release is None: + return + GatedActivateActor.activation_release = None + if GatedActivateActor.activation_started is not None: + GatedActivateActor.activation_started.set() + await release.wait() + + +class GatedActivateWriteActor(AppendActor): + async def _on_activate(self) -> None: + items_found, items = await self._state_manager.try_get_state('items') + release = GatedActivateActor.activation_release + if release is not None: + GatedActivateActor.activation_release = None + if GatedActivateActor.activation_started is not None: + GatedActivateActor.activation_started.set() + await release.wait() + await self._state_manager.set_state('items', (items if items_found else []) + ['init']) + + +class GatedDeactivateActor(AppendActor): + def __init__(self, ctx, actor_id): + super().__init__(ctx, actor_id) + self.deactivate_started: Optional[asyncio.Event] = None + self.deactivate_release: Optional[asyncio.Event] = None + self.items_appended: List[str] = [] + + async def append(self, item: str) -> None: + self.items_appended.append(item) + await super().append(item) + + async def _on_deactivate(self) -> None: + if self.deactivate_started is not None and self.deactivate_release is not None: + self.deactivate_started.set() + await self.deactivate_release.wait() + + +class RaisingDeactivateActor(AppendActor): + async def _on_deactivate(self) -> None: + raise RuntimeError('deactivate failed') + + +class ActorManagerConcurrentActivationTests(unittest.TestCase): + def _manager_for(self, actor_class: type) -> ActorManager: + serializer = DefaultJSONSerializer() + self._serializer = serializer + self._client = StoreBackedActorClient() + type_info = ActorTypeInformation.create(actor_class) + ctx = ActorRuntimeContext(type_info, serializer, serializer, self._client) + return ActorManager(ctx) + + async def _append(self, manager: ActorManager, item: str, activation_id: Optional[str]) -> None: + body = self._serializer.serialize(item) + await manager.dispatch(ActorId('race'), 'Append', body, activation_id) + + def tearDown(self): + GatedActivateActor.activation_started = None + GatedActivateActor.activation_release = None + + def test_late_activation_does_not_overwrite_newer_activation(self): + manager = self._manager_for(GatedActivateActor) + self._client.store['items'] = json.dumps(['x']).encode('utf-8') + + async def scenario() -> None: + GatedActivateActor.activation_started = asyncio.Event() + release = asyncio.Event() + GatedActivateActor.activation_release = release + late_task = asyncio.ensure_future(self._append(manager, 'a', 'act-A')) + await asyncio.wait_for(GatedActivateActor.activation_started.wait(), 5) + + await self._append(manager, 'b', 'act-B') + actor_newer = manager._active_actors['race'] + release.set() + + with self.assertRaises(DaprInternalError): + await asyncio.wait_for(late_task, 5) + self.assertIs(actor_newer, manager._active_actors['race']) + self.assertEqual('act-B', manager._activation_ids['race']) + + _run(scenario()) + + self.assertEqual(['x', 'b'], json.loads(self._client.store['items'])) + + def test_late_activation_save_does_not_overwrite_newer_activation(self): + manager = self._manager_for(GatedActivateWriteActor) + self._client.store['items'] = json.dumps(['x']).encode('utf-8') + + async def scenario() -> None: + GatedActivateActor.activation_started = asyncio.Event() + release = asyncio.Event() + GatedActivateActor.activation_release = release + late_task = asyncio.ensure_future(self._append(manager, 'a', 'act-A')) + await asyncio.wait_for(GatedActivateActor.activation_started.wait(), 5) + + await self._append(manager, 'b', 'act-B') + release.set() + + with self.assertRaises(DaprInternalError): + await asyncio.wait_for(late_task, 5) + + _run(scenario()) + + self.assertEqual(['x', 'init', 'b'], json.loads(self._client.store['items'])) + + def test_late_activation_with_same_id_reuses_winner(self): + manager = self._manager_for(GatedActivateActor) + self._client.store['items'] = json.dumps(['x']).encode('utf-8') + + async def scenario() -> None: + GatedActivateActor.activation_started = asyncio.Event() + release = asyncio.Event() + GatedActivateActor.activation_release = release + late_task = asyncio.ensure_future(self._append(manager, 'a', 'act-A')) + await asyncio.wait_for(GatedActivateActor.activation_started.wait(), 5) + + await self._append(manager, 'b', 'act-A') + actor_winner = manager._active_actors['race'] + release.set() + + await asyncio.wait_for(late_task, 5) + self.assertIs(actor_winner, manager._active_actors['race']) + self.assertEqual('act-A', manager._activation_ids['race']) + + _run(scenario()) + + self.assertEqual(['x', 'b', 'a'], json.loads(self._client.store['items'])) + + def test_raising_deactivate_on_replace_does_not_fail_new_call(self): + manager = self._manager_for(RaisingDeactivateActor) + + async def scenario() -> None: + await self._append(manager, 'a', 'act-A') + actor_stale = manager._active_actors['race'] + + await self._append(manager, 'b', 'act-B') + + self.assertIsNot(actor_stale, manager._active_actors['race']) + self.assertEqual('act-B', manager._activation_ids['race']) + + with self.assertLogs('dapr.actor.runtime.manager', level='ERROR'): + _run(scenario()) + + self.assertEqual(['a', 'b'], json.loads(self._client.store['items'])) + + def test_old_id_registered_during_reset_fails_new_id_call(self): + manager = self._manager_for(GatedDeactivateActor) + + async def scenario() -> None: + await self._append(manager, 'x', 'act-A') + actor_stale: GatedDeactivateActor = manager._active_actors['race'] + actor_stale.deactivate_started = asyncio.Event() + actor_stale.deactivate_release = asyncio.Event() + new_task = asyncio.ensure_future(self._append(manager, 'b', 'act-B')) + await asyncio.wait_for(actor_stale.deactivate_started.wait(), 5) + + await self._append(manager, 'a', 'act-A') + actor_late: GatedDeactivateActor = manager._active_actors['race'] + actor_stale.deactivate_release.set() + + with self.assertRaises(DaprInternalError): + await asyncio.wait_for(new_task, 5) + self.assertEqual(['a'], actor_late.items_appended) + self.assertEqual('act-A', manager._activation_ids['race']) + + await self._append(manager, 'c', 'act-B') + + actor_fresh: GatedDeactivateActor = manager._active_actors['race'] + self.assertIsNot(actor_late, actor_fresh) + self.assertEqual(['c'], actor_fresh.items_appended) + self.assertEqual('act-B', manager._activation_ids['race']) + + _run(scenario()) + + self.assertEqual(['x', 'a', 'c'], json.loads(self._client.store['items'])) diff --git a/tests/actor/test_actor_reentrancy.py b/tests/actor/test_actor_reentrancy.py index 69f11160e..01af97312 100644 --- a/tests/actor/test_actor_reentrancy.py +++ b/tests/actor/test_actor_reentrancy.py @@ -238,7 +238,12 @@ def test_parse_incoming_reentrancy_header_flask(self): data=request_body, ) mocked.assert_called_with( - actor_type_name, actor_id, method_name, request_body, reentrancy_id + actor_type_name, + actor_id, + method_name, + request_body, + reentrancy_id, + activation_id=None, ) def test_parse_incoming_reentrancy_header_fastapi(self): @@ -272,5 +277,10 @@ def test_parse_incoming_reentrancy_header_fastapi(self): data=request_body, ) mocked.assert_called_with( - actor_type_name, actor_id, method_name, request_body, reentrancy_id + actor_type_name, + actor_id, + method_name, + request_body, + reentrancy_id, + activation_id=None, ) diff --git a/tests/actor/test_actor_runtime.py b/tests/actor/test_actor_runtime.py index c07085ef1..725c7b5ab 100644 --- a/tests/actor/test_actor_runtime.py +++ b/tests/actor/test_actor_runtime.py @@ -23,6 +23,7 @@ from tests.actor.fake_actor_classes import ( FakeMultiInterfacesActor, FakeSimpleActor, + FakeSimpleReminderActor, FakeSimpleTimerActor, ) from tests.actor.utils import _run @@ -136,3 +137,89 @@ def test_fire_timer_unregistered(self): '{ "callback": "timer_callback", "data": "timer call" }'.encode('UTF8'), ) ) + + def test_dispatch_new_activation_id_replaces_actor(self): + actor_type = FakeMultiInterfacesActor.__name__ + request_body = self._serializer.serialize({'message': 'hello dapr'}) + _run( + ActorRuntime.dispatch( + actor_type, 'test-id', 'ActionMethod', request_body, activation_id='act-1' + ) + ) + manager = ActorRuntime._actor_managers[actor_type] + actor_stale = manager._active_actors['test-id'] + + _run( + ActorRuntime.dispatch( + actor_type, 'test-id', 'ActionMethod', request_body, activation_id='act-2' + ) + ) + + self.assertIsNot(actor_stale, manager._active_actors['test-id']) + self.assertTrue(actor_stale.deactivated) + self.assertEqual('act-2', manager._activation_ids['test-id']) + + def test_dispatch_same_activation_id_reuses_actor(self): + actor_type = FakeMultiInterfacesActor.__name__ + request_body = self._serializer.serialize({'message': 'hello dapr'}) + _run( + ActorRuntime.dispatch( + actor_type, 'test-id', 'ActionMethod', request_body, activation_id='act-1' + ) + ) + manager = ActorRuntime._actor_managers[actor_type] + actor_first = manager._active_actors['test-id'] + + _run( + ActorRuntime.dispatch( + actor_type, 'test-id', 'ActionMethod', request_body, activation_id='act-1' + ) + ) + + self.assertIs(actor_first, manager._active_actors['test-id']) + self.assertFalse(actor_first.deactivated) + + def test_fire_timer_new_activation_id_replaces_actor(self): + actor_type = FakeSimpleTimerActor.__name__ + timer_body = b'{ "callback": "timer_callback", "data": "timer call" }' + _run( + ActorRuntime.fire_timer( + actor_type, 'test-id', 'test_timer', timer_body, activation_id='act-1' + ) + ) + manager = ActorRuntime._actor_managers[actor_type] + actor_stale = manager._active_actors['test-id'] + + _run( + ActorRuntime.fire_timer( + actor_type, 'test-id', 'test_timer', timer_body, activation_id='act-2' + ) + ) + + actor_fresh = manager._active_actors['test-id'] + self.assertIsNot(actor_stale, actor_fresh) + self.assertTrue(actor_fresh.timer_called) + self.assertEqual('act-2', manager._activation_ids['test-id']) + + def test_fire_reminder_new_activation_id_replaces_actor(self): + _run(ActorRuntime.register_actor(FakeSimpleReminderActor)) + actor_type = FakeSimpleReminderActor.__name__ + reminder_body = self._serializer.serialize( + {'dueTime': '0h0m1s', 'period': '0h0m1s', 'data': 'cmVtaW5kZXJfc3RhdGU='} + ) + _run( + ActorRuntime.fire_reminder( + actor_type, 'test-id', 'test_reminder', reminder_body, activation_id='act-1' + ) + ) + manager = ActorRuntime._actor_managers[actor_type] + actor_stale = manager._active_actors['test-id'] + + _run( + ActorRuntime.fire_reminder( + actor_type, 'test-id', 'test_reminder', reminder_body, activation_id='act-2' + ) + ) + + self.assertIsNot(actor_stale, manager._active_actors['test-id']) + self.assertEqual('act-2', manager._activation_ids['test-id']) diff --git a/tests/actor/test_grpc_callbacks.py b/tests/actor/test_grpc_callbacks.py index a3c5f409c..502731adb 100644 --- a/tests/actor/test_grpc_callbacks.py +++ b/tests/actor/test_grpc_callbacks.py @@ -31,6 +31,7 @@ build_invoke_error_payload, build_reminder_fire_body, build_timer_fire_body, + extract_activation_id, extract_reentrancy_id, status_code_for_exception, ) @@ -126,6 +127,31 @@ def test_missing(self): metadata = {'content-type': 'application/json'} self.assertIsNone(extract_reentrancy_id(metadata)) + def test_empty_duplicate_is_skipped(self): + metadata = {'dapr-reentrancy-id': '', 'Dapr-Reentrancy-Id': 'rid-1'} + self.assertEqual('rid-1', extract_reentrancy_id(metadata)) + + +class ActivationIdExtractionTests(unittest.TestCase): + def test_exact_header(self): + metadata = {'Dapr-Actor-Activation-Id': 'act-1'} + self.assertEqual('act-1', extract_activation_id(metadata)) + + def test_case_insensitive(self): + self.assertEqual('act-2', extract_activation_id({'dapr-actor-activation-id': 'act-2'})) + self.assertEqual('act-3', extract_activation_id({'DAPR-ACTOR-ACTIVATION-ID': 'act-3'})) + + def test_missing(self): + metadata = {'Dapr-Reentrancy-Id': 'rid-1'} + self.assertIsNone(extract_activation_id(metadata)) + + def test_empty_duplicate_is_skipped(self): + metadata = {'dapr-actor-activation-id': '', 'Dapr-Actor-Activation-Id': 'act-1'} + self.assertEqual('act-1', extract_activation_id(metadata)) + + def test_only_empty_is_none(self): + self.assertIsNone(extract_activation_id({'Dapr-Actor-Activation-Id': ''})) + class ReminderFireBodyTests(unittest.TestCase): def test_with_base64_string_data(self): diff --git a/tests/actor/test_grpc_host.py b/tests/actor/test_grpc_host.py index fe9e7e7e6..a65bffa39 100644 --- a/tests/actor/test_grpc_host.py +++ b/tests/actor/test_grpc_host.py @@ -27,6 +27,7 @@ import asyncio import base64 import json +import threading import time import unittest from datetime import timedelta @@ -65,6 +66,9 @@ async def save_data(self, data: object) -> None: ... @actormethod(name='WhoAmI') async def who_am_i(self) -> Optional[str]: ... + @actormethod(name='Append') + async def append(self, item: str) -> None: ... + @actormethod(name='Fail') async def fail(self) -> None: ... @@ -81,6 +85,8 @@ class HostTestActor(Actor, HostTestActorInterface, Remindable): reminders_fired: list = [] timers_fired: list = [] deactivated_ids: list = [] + append_held: Optional[asyncio.Event] = None + append_release: Optional[asyncio.Event] = None async def get_name(self) -> dict: return {'name': 'HostTestActor'} @@ -95,6 +101,15 @@ async def save_data(self, data: object) -> None: async def who_am_i(self) -> Optional[str]: return reentrancy_ctx.get() + async def append(self, item: str) -> None: + items_found, items = await self._state_manager.try_get_state('items') + items_current = items if items_found else [] + held, release = HostTestActor.append_held, HostTestActor.append_release + if item == 'stale' and held is not None and release is not None: + held.set() + await release.wait() + await self._state_manager.set_state('items', items_current + [item]) + async def fail(self) -> None: raise RuntimeError('handler exploded') @@ -174,7 +189,7 @@ def _packed_bytes_value(value: bytes) -> any_pb2.Any: return packed -def _reminder_message(request_id, name, state: bytes): +def _reminder_message(request_id, name, state: bytes, metadata=None): """Builds the reminder-fire callback for a reminder registered with `state`.""" json_value = json.dumps(base64.b64encode(state).decode('utf-8')).encode('utf-8') reminder_request = api_v1.SubscribeActorEventsResponseReminderRequestAlpha1( @@ -185,11 +200,12 @@ def _reminder_message(request_id, name, state: bytes): due_time='0h0m5s', period='0h0m10s', data=_packed_bytes_value(json_value), + metadata=metadata or {}, ) return api_v1.SubscribeActorEventsResponseAlpha1(reminder_request=reminder_request) -def _timer_message(request_id, name, callback, data_json: bytes): +def _timer_message(request_id, name, callback, data_json: bytes, metadata=None): """Builds the timer-fire callback for a timer with a JSON data payload.""" timer_request = api_v1.SubscribeActorEventsResponseTimerRequestAlpha1( id=request_id, @@ -200,6 +216,7 @@ def _timer_message(request_id, name, callback, data_json: bytes): period='0h0m10s', callback=callback, data=_packed_bytes_value(data_json), + metadata=metadata or {}, ) return api_v1.SubscribeActorEventsResponseAlpha1(timer_request=timer_request) @@ -235,6 +252,8 @@ def setUp(self): HostTestActor.reminders_fired.clear() HostTestActor.timers_fired.clear() HostTestActor.deactivated_ids.clear() + HostTestActor.append_held = None + HostTestActor.append_release = None self._saved_managers = dict(ActorRuntime._actor_managers) self._saved_config = ActorRuntime.get_actor_config() ActorRuntime._actor_managers.clear() @@ -375,6 +394,116 @@ async def test_reentrancy_id_reaches_actor_context(self): self.assertEqual('rid-42', json.loads(replies[0].invoke_response.data)) + def _held_activation_id(self) -> Optional[str]: + manager = ActorRuntime._actor_managers['HostTestActor'] + return manager._activation_ids.get('a1') + + async def test_invoke_activation_id_is_stored(self): + invoke = _invoke_message('req-1', 'GetName', metadata={'Dapr-Actor-Activation-Id': 'act-1'}) + await self._serve_callbacks([invoke]) + + await self._await_replies(1) + + self.assertEqual('act-1', self._held_activation_id()) + + async def test_invoke_new_activation_id_replaces_actor(self): + await self._serve_callbacks( + [_invoke_message('req-1', 'GetName', metadata={'Dapr-Actor-Activation-Id': 'act-1'})], + [_invoke_message('req-2', 'GetName', metadata={'Dapr-Actor-Activation-Id': 'act-1'})], + [_invoke_message('req-3', 'GetName', metadata={'dapr-actor-activation-id': 'act-2'})], + ) + + replies = await self._await_replies(3) + + self.assertFalse(any(reply.invoke_response.error for reply in replies)) + self.assertEqual(['a1'], HostTestActor.deactivated_ids) + self.assertEqual('act-2', self._held_activation_id()) + + async def test_invoke_without_activation_id_keeps_actor(self): + await self._serve_callbacks( + [_invoke_message('req-1', 'GetName', metadata={'Dapr-Actor-Activation-Id': 'act-1'})], + [_invoke_message('req-2', 'GetName')], + ) + + await self._await_replies(2) + + self.assertEqual([], HostTestActor.deactivated_ids) + self.assertEqual('act-1', self._held_activation_id()) + + async def test_reminder_new_activation_id_replaces_actor(self): + await self._serve_callbacks( + [_invoke_message('req-1', 'GetName', metadata={'Dapr-Actor-Activation-Id': 'act-1'})], + [ + _reminder_message( + 'req-2', + 'demo_reminder', + b'reminder_state', + metadata={'DAPR-ACTOR-ACTIVATION-ID': 'act-2'}, + ) + ], + ) + + replies = await self._await_replies(2) + + self.assertEqual('reminder_response', replies[1].WhichOneof('request_type')) + self.assertEqual(['a1'], HostTestActor.deactivated_ids) + self.assertEqual('act-2', self._held_activation_id()) + + async def test_timer_new_activation_id_replaces_actor(self): + await self._serve_callbacks( + [_invoke_message('req-1', 'GetName', metadata={'Dapr-Actor-Activation-Id': 'act-1'})], + [ + _timer_message( + 'req-2', + 'demo_timer', + 'timer_callback', + b'{"setting":1}', + metadata={'Dapr-Actor-Activation-Id': 'act-2'}, + ) + ], + ) + + replies = await self._await_replies(2) + + self.assertEqual('timer_response', replies[1].WhichOneof('request_type')) + self.assertEqual(['a1'], HostTestActor.deactivated_ids) + self.assertEqual('act-2', self._held_activation_id()) + + async def test_call_from_dropped_stream_does_not_save_over_new_activation(self): + HostTestActor.append_held = asyncio.Event() + HostTestActor.append_release = asyncio.Event() + new_stream_gate = threading.Event() + self.addCleanup(new_stream_gate.set) + stale_append = _invoke_message( + 'req-stale', + 'Append', + json.dumps('stale').encode('utf-8'), + metadata={'Dapr-Actor-Activation-Id': 'act-1'}, + ) + fresh_append = _invoke_message( + 'req-fresh', + 'Append', + json.dumps('fresh').encode('utf-8'), + metadata={'Dapr-Actor-Activation-Id': 'act-2'}, + ) + + with mock.patch('dapr.actor.runtime.grpc_host._RECONNECT_DELAY_SECONDS', 0.05): + await self._serve_plans( + {'rounds': [], 'unanswered': [stale_append], 'end': 'abort'}, + {'rounds': [[fresh_append]], 'end': 'hold', 'gate': new_stream_gate}, + ) + await asyncio.wait_for(HostTestActor.append_held.wait(), 5) + stale_tasks = list(self.host._dispatch_tasks) + new_stream_gate.set() + replies = await self._await_replies(1) + HostTestActor.append_release.set() + await asyncio.wait_for(asyncio.gather(*stale_tasks, return_exceptions=True), 5) + + self.assertEqual('req-fresh', replies[0].invoke_response.id) + self.assertFalse(replies[0].invoke_response.error) + stored = self._fake_dapr_server.actor_state[('HostTestActor', 'a1', 'items')] + self.assertEqual(['fresh'], json.loads(stored)) + async def test_reminder_round_trip(self): await self._serve_callbacks( [_reminder_message('req-1', 'demo_reminder', b'reminder_state')] diff --git a/tests/clients/fake_dapr_server.py b/tests/clients/fake_dapr_server.py index dd8b6a2e6..9f79a222e 100644 --- a/tests/clients/fake_dapr_server.py +++ b/tests/clients/fake_dapr_server.py @@ -63,7 +63,7 @@ def __init__(self, grpc_port: int = 50001, http_port: int = 8080): # dict with 'rounds' (lists of SubscribeActorEventsResponseAlpha1 pushed per round, the # server reads one correlated reply per pushed message before the next round) and 'end' # ('abort' simulates a sidecar drop, 'eof' closes cleanly, 'hold' keeps the stream open - # until the client goes away). + # until the client goes away). 'gate' delays the first round, 'unanswered' get no replies. self.actor_stream_plans: List[Dict[str, Any]] = [] self.actor_stream_initials: List[api_v1.SubscribeActorEventsRequestInitialAlpha1] = [] self.actor_stream_replies: List[api_v1.SubscribeActorEventsRequestAlpha1] = [] @@ -323,11 +323,16 @@ def SubscribeActorEventsAlpha1(self, request_iterator, context): yield api_v1.SubscribeActorEventsResponseAlpha1( initial_response=api_v1.SubscribeActorEventsResponseInitialAlpha1() ) + gate = plan.get('gate') + if gate is not None: + gate.wait(timeout=10) for round_messages in plan.get('rounds', []): for message in round_messages: yield message for _ in round_messages: self.actor_stream_replies.append(next(request_iterator)) + for message in plan.get('unanswered', []): + yield message end_behavior = plan.get('end', 'hold') if end_behavior == 'abort': diff --git a/tests/ext/fastapi/test_dapractor.py b/tests/ext/fastapi/test_dapractor.py index f3e2f82ca..1fa3e2dcf 100644 --- a/tests/ext/fastapi/test_dapractor.py +++ b/tests/ext/fastapi/test_dapractor.py @@ -15,9 +15,12 @@ import json import unittest +from unittest import mock from fastapi import FastAPI +from fastapi.testclient import TestClient +from dapr.actor import ActorRuntime from dapr.ext.fastapi.actor import DaprActor, _wrap_response EXPECTED_ACTOR_PATHS = { @@ -93,5 +96,115 @@ def test_router_tag(self): self.assertFalse(tags) +class DaprActorActivationIdTest(unittest.TestCase): + def setUp(self): + app = FastAPI() + DaprActor(app) + self.client = TestClient(app) + + def test_method_passes_activation_id(self): + with mock.patch.object( + ActorRuntime, 'dispatch', new=mock.AsyncMock(return_value=b'"ok"') + ) as dispatch: + response = self.client.put( + '/actors/FakeActor/a1/method/Echo', + content=b'"hi"', + headers={'Dapr-Actor-Activation-Id': 'act-1', 'Dapr-Reentrancy-Id': 'rid-1'}, + ) + + self.assertEqual(200, response.status_code) + dispatch.assert_awaited_once_with( + 'FakeActor', 'a1', 'Echo', b'"hi"', 'rid-1', activation_id='act-1' + ) + + def test_method_header_is_case_insensitive(self): + for header_name in ('dapr-actor-activation-id', 'DAPR-ACTOR-ACTIVATION-ID'): + with mock.patch.object( + ActorRuntime, 'dispatch', new=mock.AsyncMock(return_value=b'"ok"') + ) as dispatch: + self.client.put( + '/actors/FakeActor/a1/method/Echo', + content=b'"hi"', + headers={header_name: 'act-2'}, + ) + + dispatch.assert_awaited_once_with( + 'FakeActor', 'a1', 'Echo', b'"hi"', None, activation_id='act-2' + ) + + def test_method_without_header_passes_none(self): + with mock.patch.object( + ActorRuntime, 'dispatch', new=mock.AsyncMock(return_value=b'"ok"') + ) as dispatch: + self.client.put('/actors/FakeActor/a1/method/Echo', content=b'"hi"') + + dispatch.assert_awaited_once_with( + 'FakeActor', 'a1', 'Echo', b'"hi"', None, activation_id=None + ) + + def test_timer_passes_activation_id(self): + with mock.patch.object(ActorRuntime, 'fire_timer', new=mock.AsyncMock()) as fire_timer: + response = self.client.put( + '/actors/FakeActor/a1/method/timer/t1', + content=b'{}', + headers={'dapr-actor-activation-id': 'act-3'}, + ) + + self.assertEqual(200, response.status_code) + fire_timer.assert_awaited_once_with('FakeActor', 'a1', 't1', b'{}', activation_id='act-3') + + def test_reminder_passes_activation_id(self): + with mock.patch.object( + ActorRuntime, 'fire_reminder', new=mock.AsyncMock() + ) as fire_reminder: + response = self.client.put( + '/actors/FakeActor/a1/method/remind/r1', + content=b'{}', + headers={'Dapr-Actor-Activation-Id': 'act-4'}, + ) + + self.assertEqual(200, response.status_code) + fire_reminder.assert_awaited_once_with( + 'FakeActor', 'a1', 'r1', b'{}', activation_id='act-4' + ) + + def test_timer_header_is_case_insensitive(self): + for header_name in ('dapr-actor-activation-id', 'DAPR-ACTOR-ACTIVATION-ID'): + with mock.patch.object(ActorRuntime, 'fire_timer', new=mock.AsyncMock()) as fire_timer: + self.client.put( + '/actors/FakeActor/a1/method/timer/t1', + content=b'{}', + headers={header_name: 'act-6'}, + ) + + fire_timer.assert_awaited_once_with( + 'FakeActor', 'a1', 't1', b'{}', activation_id='act-6' + ) + + def test_reminder_header_is_case_insensitive(self): + for header_name in ('dapr-actor-activation-id', 'DAPR-ACTOR-ACTIVATION-ID'): + with mock.patch.object( + ActorRuntime, 'fire_reminder', new=mock.AsyncMock() + ) as fire_reminder: + self.client.put( + '/actors/FakeActor/a1/method/remind/r1', + content=b'{}', + headers={header_name: 'act-7'}, + ) + + fire_reminder.assert_awaited_once_with( + 'FakeActor', 'a1', 'r1', b'{}', activation_id='act-7' + ) + + def test_deactivate_is_unchanged(self): + with mock.patch.object(ActorRuntime, 'deactivate', new=mock.AsyncMock()) as deactivate: + response = self.client.delete( + '/actors/FakeActor/a1', headers={'Dapr-Actor-Activation-Id': 'act-5'} + ) + + self.assertEqual(200, response.status_code) + deactivate.assert_awaited_once_with('FakeActor', 'a1') + + if __name__ == '__main__': unittest.main() diff --git a/tests/ext/flask/test_dapractor.py b/tests/ext/flask/test_dapractor.py new file mode 100644 index 000000000..762b04a98 --- /dev/null +++ b/tests/ext/flask/test_dapractor.py @@ -0,0 +1,137 @@ +# -*- coding: utf-8 -*- + +""" +Copyright 2026 The Dapr Authors +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" + +import unittest +from unittest import mock + +from flask import Flask + +from dapr.actor import ActorRuntime +from dapr.ext.flask import DaprActor + + +class DaprActorActivationIdTest(unittest.TestCase): + def setUp(self): + app = Flask('test_actor_app') + app.testing = True + DaprActor(app) + self.client = app.test_client() + + def test_method_passes_activation_id(self): + with mock.patch.object( + ActorRuntime, 'dispatch', new=mock.AsyncMock(return_value=b'"ok"') + ) as dispatch: + response = self.client.put( + '/actors/FakeActor/a1/method/Echo', + data=b'"hi"', + headers={'Dapr-Actor-Activation-Id': 'act-1', 'Dapr-Reentrancy-Id': 'rid-1'}, + ) + + self.assertEqual(200, response.status_code) + dispatch.assert_awaited_once_with( + 'FakeActor', 'a1', 'Echo', b'"hi"', 'rid-1', activation_id='act-1' + ) + + def test_method_header_is_case_insensitive(self): + for header_name in ('dapr-actor-activation-id', 'DAPR-ACTOR-ACTIVATION-ID'): + with mock.patch.object( + ActorRuntime, 'dispatch', new=mock.AsyncMock(return_value=b'"ok"') + ) as dispatch: + self.client.put( + '/actors/FakeActor/a1/method/Echo', + data=b'"hi"', + headers={header_name: 'act-2'}, + ) + + dispatch.assert_awaited_once_with( + 'FakeActor', 'a1', 'Echo', b'"hi"', None, activation_id='act-2' + ) + + def test_method_without_header_passes_none(self): + with mock.patch.object( + ActorRuntime, 'dispatch', new=mock.AsyncMock(return_value=b'"ok"') + ) as dispatch: + self.client.put('/actors/FakeActor/a1/method/Echo', data=b'"hi"') + + dispatch.assert_awaited_once_with( + 'FakeActor', 'a1', 'Echo', b'"hi"', None, activation_id=None + ) + + def test_timer_passes_activation_id(self): + with mock.patch.object(ActorRuntime, 'fire_timer', new=mock.AsyncMock()) as fire_timer: + response = self.client.put( + '/actors/FakeActor/a1/method/timer/t1', + data=b'{}', + headers={'dapr-actor-activation-id': 'act-3'}, + ) + + self.assertEqual(200, response.status_code) + fire_timer.assert_awaited_once_with('FakeActor', 'a1', 't1', b'{}', activation_id='act-3') + + def test_reminder_passes_activation_id(self): + with mock.patch.object( + ActorRuntime, 'fire_reminder', new=mock.AsyncMock() + ) as fire_reminder: + response = self.client.put( + '/actors/FakeActor/a1/method/remind/r1', + data=b'{}', + headers={'Dapr-Actor-Activation-Id': 'act-4'}, + ) + + self.assertEqual(200, response.status_code) + fire_reminder.assert_awaited_once_with( + 'FakeActor', 'a1', 'r1', b'{}', activation_id='act-4' + ) + + def test_timer_header_is_case_insensitive(self): + for header_name in ('dapr-actor-activation-id', 'DAPR-ACTOR-ACTIVATION-ID'): + with mock.patch.object(ActorRuntime, 'fire_timer', new=mock.AsyncMock()) as fire_timer: + self.client.put( + '/actors/FakeActor/a1/method/timer/t1', + data=b'{}', + headers={header_name: 'act-6'}, + ) + + fire_timer.assert_awaited_once_with( + 'FakeActor', 'a1', 't1', b'{}', activation_id='act-6' + ) + + def test_reminder_header_is_case_insensitive(self): + for header_name in ('dapr-actor-activation-id', 'DAPR-ACTOR-ACTIVATION-ID'): + with mock.patch.object( + ActorRuntime, 'fire_reminder', new=mock.AsyncMock() + ) as fire_reminder: + self.client.put( + '/actors/FakeActor/a1/method/remind/r1', + data=b'{}', + headers={header_name: 'act-7'}, + ) + + fire_reminder.assert_awaited_once_with( + 'FakeActor', 'a1', 'r1', b'{}', activation_id='act-7' + ) + + def test_deactivate_is_unchanged(self): + with mock.patch.object(ActorRuntime, 'deactivate', new=mock.AsyncMock()) as deactivate: + response = self.client.delete( + '/actors/FakeActor/a1', headers={'Dapr-Actor-Activation-Id': 'act-5'} + ) + + self.assertEqual(200, response.status_code) + deactivate.assert_awaited_once_with('FakeActor', 'a1') + + +if __name__ == '__main__': + unittest.main() From e0968b7fb32d91bfc6b2e1ba3ae00f4db63814f7 Mon Sep 17 00:00:00 2001 From: Cassandra Coyle Date: Tue, 27 Oct 2026 11:57:06 -0500 Subject: [PATCH 2/2] pr feedback Signed-off-by: Cassandra Coyle --- dapr/actor/runtime/manager.py | 55 +++++---- dapr/actor/runtime/runtime.py | 50 ++++++++ dapr/actor/runtime/state_manager.py | 11 ++ dapr/clients/base.py | 4 + dapr/clients/grpc/dapr_actor_grpc_client.py | 5 + dapr/clients/http/dapr_actor_http_client.py | 8 ++ tests/actor/test_actor_factory.py | 6 +- tests/actor/test_actor_manager.py | 100 +++++++++++----- tests/actor/test_actor_runtime.py | 126 ++++++++++++++++++-- tests/actor/test_grpc_host.py | 20 +++- tests/clients/fake_dapr_server.py | 2 + 11 files changed, 315 insertions(+), 72 deletions(-) diff --git a/dapr/actor/runtime/manager.py b/dapr/actor/runtime/manager.py index cda3b1a6b..e9ac638e9 100644 --- a/dapr/actor/runtime/manager.py +++ b/dapr/actor/runtime/manager.py @@ -17,7 +17,7 @@ import logging import uuid from functools import partial -from typing import Any, Callable, Coroutine, Dict, Optional +from typing import Any, Callable, Coroutine, Dict, NamedTuple, Optional from dapr.actor.id import ActorId from dapr.actor.runtime._method_context import ActorMethodContext @@ -38,6 +38,11 @@ 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.""" @@ -45,8 +50,7 @@ def __init__(self, ctx: ActorRuntimeContext): self._runtime_ctx = ctx self._message_serializer = self._runtime_ctx.message_serializer - self._active_actors: Dict[str, Actor] = {} - self._activation_ids: Dict[str, str] = {} + self._active_actors: Dict[str, _ActiveActor] = {} self._active_actors_lock = asyncio.Lock() self._dispatcher = ActorMethodDispatcher(ctx.actor_type_info) @@ -61,29 +65,23 @@ async def activate_actor(self, actor_id: ActorId, activation_id: Optional[str] = async with self._active_actors_lock: registered = self._active_actors.get(actor_id.id) if registered is None: - self._active_actors[actor_id.id] = actor - if activation_id: - self._activation_ids[actor_id.id] = activation_id - else: - self._activation_ids.pop(actor_id.id, None) + self._active_actors[actor_id.id] = _ActiveActor(actor, activation_id or None) return actor - registered_id = self._activation_ids.get(actor_id.id) await self._deactivate_discarded(actor_id, actor) - if activation_id and activation_id != registered_id: - registered_by = _describe_activation(registered_id) + 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 + 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) - self._activation_ids.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, @@ -171,13 +169,14 @@ async def _dispatch_internal( ) -> object: await self._deactivate_if_stale(actor_id, activation_id) async with self._active_actors_lock: - actor = self._active_actors.get(actor_id.id) - registered_id = self._activation_ids.get(actor_id.id) - if actor is not None and activation_id and activation_id != registered_id: - registered_by = _describe_activation(registered_id) + 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 actor is None: + 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: @@ -197,7 +196,8 @@ async def _dispatch_internal( return retval def _ensure_still_registered(self, actor_id: ActorId, actor: Actor) -> None: - if self._active_actors.get(actor_id.id) is not actor: + 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' ) @@ -207,16 +207,15 @@ async def _deactivate_if_stale(self, actor_id: ActorId, activation_id: Optional[ if not activation_id: return async with self._active_actors_lock: - if actor_id.id not in self._active_actors: - return - if self._activation_ids.get(actor_id.id) == activation_id: + stale = self._active_actors.get(actor_id.id) + if stale is None or stale.activation_id == activation_id: return - stale_actor = self._active_actors.pop(actor_id.id) - self._activation_ids.pop(actor_id.id, None) - await self._deactivate_discarded(actor_id, stale_actor) + 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 diff --git a/dapr/actor/runtime/runtime.py b/dapr/actor/runtime/runtime.py index c3d61bf88..a3e87e57a 100644 --- a/dapr/actor/runtime/runtime.py +++ b/dapr/actor/runtime/runtime.py @@ -14,6 +14,8 @@ """ import asyncio +import logging +import re from typing import Callable, Dict, List, Optional, Type, Union from dapr.actor.id import ActorId @@ -30,6 +32,19 @@ _HTTP_TIMEOUT_FROM_SETTINGS = _FromSettings('DAPR_HTTP_TIMEOUT_SECONDS') +# Older sidecars forward caller headers, so a caller could spoof the activation id. +_MIN_ACTIVATION_ID_VERSION = (1, 19, 1) +_VERSION_LOOKUP_TIMEOUT_SECONDS = 5 + +logger = logging.getLogger(__name__) + + +def _supports_activation_id(version: str) -> bool: + match = re.match(r'(\d+)\.(\d+)\.(\d+)(?:[-+].*)?$', version) + if match is None: + return False + return tuple(int(part) for part in match.groups()) >= _MIN_ACTIVATION_ID_VERSION + class ActorRuntime: """The class that creates instances of :class:`Actor` and @@ -41,6 +56,9 @@ class ActorRuntime: _actor_managers: Dict[str, ActorManager] = {} _actor_managers_lock = asyncio.Lock() + _activation_id_trusted: Optional[bool] = None + _version_lookup: Optional['asyncio.Future[str]'] = None + @classmethod async def register_actor( cls, @@ -131,6 +149,7 @@ async def dispatch( manager = await cls._get_actor_manager(actor_type_name) if not manager: raise ValueError(f'{actor_type_name} is not registered.') + activation_id = await cls._trusted_activation_id(manager, activation_id) return await manager.dispatch( ActorId(actor_id), actor_method_name, request_body, activation_id ) @@ -160,6 +179,7 @@ async def fire_reminder( manager = await cls._get_actor_manager(actor_type_name) if not manager: raise ValueError(f'{actor_type_name} is not registered.') + activation_id = await cls._trusted_activation_id(manager, activation_id) await manager.fire_reminder(ActorId(actor_id), name, state, activation_id) @classmethod @@ -186,6 +206,7 @@ async def fire_timer( manager = await cls._get_actor_manager(actor_type_name) if not manager: raise ValueError(f'{actor_type_name} is not registered.') + activation_id = await cls._trusted_activation_id(manager, activation_id) await manager.fire_timer(ActorId(actor_id), name, state, activation_id) @classmethod @@ -215,3 +236,32 @@ async def _get_actor_manager(cls, actor_type_name: str) -> Optional[ActorManager """ async with cls._actor_managers_lock: return cls._actor_managers.get(actor_type_name) + + @classmethod + async def _trusted_activation_id( + cls, manager: ActorManager, activation_id: Optional[str] + ) -> Optional[str]: + """Returns activation_id only when the sidecar is new enough to set it itself.""" + if not activation_id: + return activation_id + if cls._activation_id_trusted is None: + lookup = cls._version_lookup + if lookup is None or lookup.get_loop() is not asyncio.get_running_loop(): + lookup = asyncio.ensure_future( + asyncio.wait_for( + manager._runtime_ctx.dapr_client.get_runtime_version(), + _VERSION_LOOKUP_TIMEOUT_SECONDS, + ) + ) + cls._version_lookup = lookup + try: + version = await asyncio.shield(lookup) + except Exception as exception: # noqa: BLE001 + if cls._version_lookup is lookup: + cls._version_lookup = None + logger.warning( + 'Ignoring actor activation id, sidecar version lookup failed: %r', exception + ) + return None + cls._activation_id_trusted = _supports_activation_id(version) + return activation_id if cls._activation_id_trusted else None diff --git a/dapr/actor/runtime/state_manager.py b/dapr/actor/runtime/state_manager.py index 59f3ade1b..f004ba92c 100644 --- a/dapr/actor/runtime/state_manager.py +++ b/dapr/actor/runtime/state_manager.py @@ -19,6 +19,7 @@ from typing import TYPE_CHECKING, Any, Callable, Dict, Generic, List, Optional, Tuple, TypeVar from dapr.actor.runtime.state_change import ActorStateChange, StateChangeKind +from dapr.clients.exceptions import DaprInternalError from dapr.common.reentrancy_context import reentrancy_ctx if TYPE_CHECKING: @@ -86,6 +87,7 @@ def __init__(self, actor: 'Actor'): self._type_name = actor.runtime_ctx.actor_type_info.type_name self._default_state_change_tracker: Dict[str, StateMetadata] = {} + self._discarded = False async def add_state(self, state_name: str, value: T) -> None: if not await self.try_add_state(state_name, value): @@ -261,7 +263,16 @@ async def clear_cache(self) -> None: default_loop = asyncio.get_running_loop() await default_loop.run_in_executor(None, state_change_tracker.clear) + def _discard(self) -> None: + """Refuses every later save once a newer activation replaced the actor.""" + self._discarded = True + async def save_state(self) -> None: + if self._discarded: + raise DaprInternalError( + f'{self._type_name} actor {self._actor.id} was replaced by a newer activation, ' + 'state not saved' + ) state_change_tracker = self._get_contextual_state_tracker() if len(state_change_tracker) == 0: return diff --git a/dapr/clients/base.py b/dapr/clients/base.py index 304d4ed2c..76c34765d 100644 --- a/dapr/clients/base.py +++ b/dapr/clients/base.py @@ -51,3 +51,7 @@ async def register_timer( @abstractmethod async def unregister_timer(self, actor_type: str, actor_id: str, name: str) -> None: ... + + async def get_runtime_version(self) -> str: + """Returns the sidecar's runtime version from the metadata API.""" + raise NotImplementedError diff --git a/dapr/clients/grpc/dapr_actor_grpc_client.py b/dapr/clients/grpc/dapr_actor_grpc_client.py index 85e310fb2..7ab806f77 100644 --- a/dapr/clients/grpc/dapr_actor_grpc_client.py +++ b/dapr/clients/grpc/dapr_actor_grpc_client.py @@ -84,6 +84,11 @@ async def close(self) -> None: if self._owns_channel: await self._channel.close() + async def get_runtime_version(self) -> str: + """Returns the sidecar's runtime version through the GetMetadata RPC.""" + response = await self._stub.GetMetadata(api_v1.GetMetadataRequest(), timeout=self._timeout) + return response.runtime_version + async def invoke_method( self, actor_type: str, actor_id: str, method: str, data: Optional[bytes] = None ) -> bytes: diff --git a/dapr/clients/http/dapr_actor_http_client.py b/dapr/clients/http/dapr_actor_http_client.py index d35f85441..f7429554b 100644 --- a/dapr/clients/http/dapr_actor_http_client.py +++ b/dapr/clients/http/dapr_actor_http_client.py @@ -13,6 +13,7 @@ limitations under the License. """ +import json from typing import TYPE_CHECKING, Callable, Dict, Optional, Union from dapr.clients.http.helpers import get_api_url @@ -145,5 +146,12 @@ async def unregister_timer(self, actor_type: str, actor_id: str, name: str) -> N url = f'{self._get_base_url(actor_type, actor_id)}/timers/{name}' await self._client.send_bytes(method='DELETE', url=url, data=None) + async def get_runtime_version(self) -> str: + """Returns the sidecar's runtime version from the metadata API.""" + body, _ = await self._client.send_bytes( + method='GET', url=f'{get_api_url()}/metadata', data=None + ) + return json.loads(body).get('runtimeVersion', '') + def _get_base_url(self, actor_type: str, actor_id: str) -> str: return '{}/actors/{}/{}'.format(get_api_url(), actor_type, actor_id) diff --git a/tests/actor/test_actor_factory.py b/tests/actor/test_actor_factory.py index 4f629bb25..4e2bede82 100644 --- a/tests/actor/test_actor_factory.py +++ b/tests/actor/test_actor_factory.py @@ -79,9 +79,9 @@ def test_activate_actor(self): _run(self._manager.activate_actor(test_actor_id)) # assert - self.assertEqual(test_actor_id, self._manager._active_actors[test_actor_id.id].id) - self.assertTrue(self._manager._active_actors[test_actor_id.id].activated) - self.assertFalse(self._manager._active_actors[test_actor_id.id].deactivated) + self.assertEqual(test_actor_id, self._manager._active_actors[test_actor_id.id].actor.id) + self.assertTrue(self._manager._active_actors[test_actor_id.id].actor.activated) + self.assertFalse(self._manager._active_actors[test_actor_id.id].actor.deactivated) def test_dispatch_success(self): """dispatch ActionMethod""" diff --git a/tests/actor/test_actor_manager.py b/tests/actor/test_actor_manager.py index 2eaf5977f..138452bda 100644 --- a/tests/actor/test_actor_manager.py +++ b/tests/actor/test_actor_manager.py @@ -60,9 +60,9 @@ def test_activate_actor(self): _run(self._manager.activate_actor(test_actor_id)) # assert - self.assertEqual(test_actor_id, self._manager._active_actors[test_actor_id.id].id) - self.assertTrue(self._manager._active_actors[test_actor_id.id].activated) - self.assertFalse(self._manager._active_actors[test_actor_id.id].deactivated) + self.assertEqual(test_actor_id, self._manager._active_actors[test_actor_id.id].actor.id) + self.assertTrue(self._manager._active_actors[test_actor_id.id].actor.activated) + self.assertFalse(self._manager._active_actors[test_actor_id.id].actor.deactivated) def test_deactivate_actor(self): """Activate ActorId('2') and deactivate it""" @@ -70,9 +70,9 @@ def test_deactivate_actor(self): _run(self._manager.activate_actor(test_actor_id)) # assert - self.assertEqual(test_actor_id, self._manager._active_actors[test_actor_id.id].id) - self.assertTrue(self._manager._active_actors[test_actor_id.id].activated) - self.assertFalse(self._manager._active_actors[test_actor_id.id].deactivated) + self.assertEqual(test_actor_id, self._manager._active_actors[test_actor_id.id].actor.id) + self.assertTrue(self._manager._active_actors[test_actor_id.id].actor.activated) + self.assertFalse(self._manager._active_actors[test_actor_id.id].actor.deactivated) _run(self._manager.deactivate_actor(test_actor_id)) self.assertIsNone(self._manager._active_actors.get(test_actor_id.id)) @@ -157,7 +157,7 @@ def test_fire_timer_success(self): manager = ActorManager(ctx) _run(manager.activate_actor(test_actor_id)) - actor = manager._active_actors.get(test_actor_id.id, None) + actor = manager._active_actors[test_actor_id.id].actor # Setup timer _run( @@ -251,10 +251,11 @@ def _recall(self, activation_id: Optional[str]) -> object: return self._serializer.deserialize(response) def _active_actor(self) -> ActivationTrackingActor: - return self._manager._active_actors[self._actor_id.id] + return self._manager._active_actors[self._actor_id.id].actor def _stored_activation_id(self) -> Optional[str]: - return self._manager._activation_ids.get(self._actor_id.id) + active = self._manager._active_actors.get(self._actor_id.id) + return active.activation_id if active else None def test_first_call_stores_activation_id(self): self._remember('cached', 'act-1') @@ -420,7 +421,7 @@ def test_activation_ids_are_tracked_per_actor(self): self.assertIs(actor_first, self._active_actor()) self.assertEqual('act-1', self._stored_activation_id()) - self.assertEqual('act-other', self._manager._activation_ids[other_actor_id.id]) + self.assertEqual('act-other', self._manager._active_actors[other_actor_id.id].activation_id) class StoreBackedActorClient(FakeDaprActorClient): @@ -494,7 +495,7 @@ async def _append(self, item: str, activation_id: Optional[str]) -> None: async def _run_replaced_during(self, stale_call: Callable[[], Awaitable[None]]) -> None: await self._append('x', 'act-1') - actor_stale: AppendActor = self._manager._active_actors[self._actor_id.id] + actor_stale: AppendActor = self._manager._active_actors[self._actor_id.id].actor actor_stale.held = asyncio.Event() actor_stale.release = asyncio.Event() stale_task = asyncio.ensure_future(stale_call()) @@ -535,7 +536,7 @@ def test_raising_timer_skips_save(self): def test_replaced_by_missing_id_call_still_saves(self): async def scenario() -> None: await self._append('x', 'act-1') - actor_first: AppendActor = self._manager._active_actors[self._actor_id.id] + actor_first: AppendActor = self._manager._active_actors[self._actor_id.id].actor actor_first.held = asyncio.Event() actor_first.release = asyncio.Event() first_task = asyncio.ensure_future(self._append('a', 'act-1')) @@ -553,7 +554,7 @@ async def scenario() -> None: def test_deactivated_during_call_skips_save(self): async def scenario() -> None: await self._append('x', 'act-1') - actor_first: AppendActor = self._manager._active_actors[self._actor_id.id] + actor_first: AppendActor = self._manager._active_actors[self._actor_id.id].actor actor_first.held = asyncio.Event() actor_first.release = asyncio.Event() first_task = asyncio.ensure_future(self._append('a', 'act-1')) @@ -602,7 +603,7 @@ def test_manager_calls_overridden_post_actor_method_internal(self): _run(manager.dispatch(ActorId('compat'), 'Append', body, 'act-1')) - actor: PostHookOverrideActor = manager._active_actors['compat'] + actor: PostHookOverrideActor = manager._active_actors['compat'].actor self.assertEqual(['Append'], actor.post_internal_calls) self.assertEqual(['a'], json.loads(self._client.store['items'])) @@ -651,6 +652,49 @@ async def _on_deactivate(self) -> None: await self.deactivate_release.wait() +class SaveOnDeactivateActor(AppendActor): + async def _on_deactivate(self) -> None: + await self._state_manager.set_state('items', ['deactivated']) + await self._state_manager.save_state() + + +class ActorManagerDiscardedSaveTests(unittest.TestCase): + def setUp(self): + self._serializer = DefaultJSONSerializer() + self._client = StoreBackedActorClient() + type_info = ActorTypeInformation.create(SaveOnDeactivateActor) + ctx = ActorRuntimeContext(type_info, self._serializer, self._serializer, self._client) + self._manager = ActorManager(ctx) + self._actor_id = ActorId('discarded') + + def _stored_items(self) -> object: + return json.loads(self._client.store['items']) + + async def _append(self, item: str, activation_id: Optional[str]) -> None: + body = self._serializer.serialize(item) + await self._manager.dispatch(self._actor_id, 'Append', body, activation_id) + + def test_replaced_actor_cannot_save_from_on_deactivate(self): + _run(self._append('x', 'act-1')) + _run(self._append('y', 'act-2')) + self.assertEqual(['x', 'y'], self._stored_items()) + + _run(self._manager.deactivate_actor(self._actor_id)) + self.assertEqual(['deactivated'], self._stored_items()) + + def test_replaced_actor_refuses_explicit_save(self): + async def scenario() -> None: + await self._append('x', 'act-1') + actor_stale = self._manager._active_actors[self._actor_id.id].actor + await self._append('y', 'act-2') + await actor_stale._state_manager.set_state('items', ['stale']) + with self.assertRaises(DaprInternalError): + await actor_stale._state_manager.save_state() + + _run(scenario()) + self.assertEqual(['x', 'y'], self._stored_items()) + + class RaisingDeactivateActor(AppendActor): async def _on_deactivate(self) -> None: raise RuntimeError('deactivate failed') @@ -685,13 +729,13 @@ async def scenario() -> None: await asyncio.wait_for(GatedActivateActor.activation_started.wait(), 5) await self._append(manager, 'b', 'act-B') - actor_newer = manager._active_actors['race'] + actor_newer = manager._active_actors['race'].actor release.set() with self.assertRaises(DaprInternalError): await asyncio.wait_for(late_task, 5) - self.assertIs(actor_newer, manager._active_actors['race']) - self.assertEqual('act-B', manager._activation_ids['race']) + self.assertIs(actor_newer, manager._active_actors['race'].actor) + self.assertEqual('act-B', manager._active_actors['race'].activation_id) _run(scenario()) @@ -730,12 +774,12 @@ async def scenario() -> None: await asyncio.wait_for(GatedActivateActor.activation_started.wait(), 5) await self._append(manager, 'b', 'act-A') - actor_winner = manager._active_actors['race'] + actor_winner = manager._active_actors['race'].actor release.set() await asyncio.wait_for(late_task, 5) - self.assertIs(actor_winner, manager._active_actors['race']) - self.assertEqual('act-A', manager._activation_ids['race']) + self.assertIs(actor_winner, manager._active_actors['race'].actor) + self.assertEqual('act-A', manager._active_actors['race'].activation_id) _run(scenario()) @@ -746,12 +790,12 @@ def test_raising_deactivate_on_replace_does_not_fail_new_call(self): async def scenario() -> None: await self._append(manager, 'a', 'act-A') - actor_stale = manager._active_actors['race'] + actor_stale = manager._active_actors['race'].actor await self._append(manager, 'b', 'act-B') - self.assertIsNot(actor_stale, manager._active_actors['race']) - self.assertEqual('act-B', manager._activation_ids['race']) + self.assertIsNot(actor_stale, manager._active_actors['race'].actor) + self.assertEqual('act-B', manager._active_actors['race'].activation_id) with self.assertLogs('dapr.actor.runtime.manager', level='ERROR'): _run(scenario()) @@ -763,27 +807,27 @@ def test_old_id_registered_during_reset_fails_new_id_call(self): async def scenario() -> None: await self._append(manager, 'x', 'act-A') - actor_stale: GatedDeactivateActor = manager._active_actors['race'] + actor_stale: GatedDeactivateActor = manager._active_actors['race'].actor actor_stale.deactivate_started = asyncio.Event() actor_stale.deactivate_release = asyncio.Event() new_task = asyncio.ensure_future(self._append(manager, 'b', 'act-B')) await asyncio.wait_for(actor_stale.deactivate_started.wait(), 5) await self._append(manager, 'a', 'act-A') - actor_late: GatedDeactivateActor = manager._active_actors['race'] + actor_late: GatedDeactivateActor = manager._active_actors['race'].actor actor_stale.deactivate_release.set() with self.assertRaises(DaprInternalError): await asyncio.wait_for(new_task, 5) self.assertEqual(['a'], actor_late.items_appended) - self.assertEqual('act-A', manager._activation_ids['race']) + self.assertEqual('act-A', manager._active_actors['race'].activation_id) await self._append(manager, 'c', 'act-B') - actor_fresh: GatedDeactivateActor = manager._active_actors['race'] + actor_fresh: GatedDeactivateActor = manager._active_actors['race'].actor self.assertIsNot(actor_late, actor_fresh) self.assertEqual(['c'], actor_fresh.items_appended) - self.assertEqual('act-B', manager._activation_ids['race']) + self.assertEqual('act-B', manager._active_actors['race'].activation_id) _run(scenario()) diff --git a/tests/actor/test_actor_runtime.py b/tests/actor/test_actor_runtime.py index f6bfa1379..b80f648ae 100644 --- a/tests/actor/test_actor_runtime.py +++ b/tests/actor/test_actor_runtime.py @@ -13,8 +13,11 @@ limitations under the License. """ +import asyncio import unittest from datetime import timedelta +from types import SimpleNamespace +from unittest import mock from dapr.actor.runtime.config import ActorRuntimeConfig from dapr.actor.runtime.runtime import ActorRuntime @@ -40,6 +43,11 @@ def setUpClass(cls): def setUp(self): ActorRuntime._actor_managers = {} ActorRuntime.set_actor_config(ActorRuntimeConfig()) + ActorRuntime._activation_id_trusted = None + ActorRuntime._version_lookup = None + self.addCleanup(setattr, ActorRuntime, '_activation_id_trusted', None) + self.server.set_response(b'{"runtimeVersion": "1.19.1"}') + self.addCleanup(self.server.set_response, b'') self._serializer = DefaultJSONSerializer() _run(ActorRuntime.register_actor(FakeSimpleActor)) _run(ActorRuntime.register_actor(FakeMultiInterfacesActor)) @@ -120,7 +128,7 @@ def test_fire_timer_success(self): ) manager = ActorRuntime._actor_managers[FakeSimpleTimerActor.__name__] - actor = manager._active_actors['test-id'] + actor = manager._active_actors['test-id'].actor self.assertTrue(actor.timer_called) def test_fire_timer_unregistered(self): @@ -143,7 +151,7 @@ def test_dispatch_new_activation_id_replaces_actor(self): ) ) manager = ActorRuntime._actor_managers[actor_type] - actor_stale = manager._active_actors['test-id'] + actor_stale = manager._active_actors['test-id'].actor _run( ActorRuntime.dispatch( @@ -151,9 +159,9 @@ def test_dispatch_new_activation_id_replaces_actor(self): ) ) - self.assertIsNot(actor_stale, manager._active_actors['test-id']) + self.assertIsNot(actor_stale, manager._active_actors['test-id'].actor) self.assertTrue(actor_stale.deactivated) - self.assertEqual('act-2', manager._activation_ids['test-id']) + self.assertEqual('act-2', manager._active_actors['test-id'].activation_id) def test_dispatch_same_activation_id_reuses_actor(self): actor_type = FakeMultiInterfacesActor.__name__ @@ -164,7 +172,7 @@ def test_dispatch_same_activation_id_reuses_actor(self): ) ) manager = ActorRuntime._actor_managers[actor_type] - actor_first = manager._active_actors['test-id'] + actor_first = manager._active_actors['test-id'].actor _run( ActorRuntime.dispatch( @@ -172,7 +180,7 @@ def test_dispatch_same_activation_id_reuses_actor(self): ) ) - self.assertIs(actor_first, manager._active_actors['test-id']) + self.assertIs(actor_first, manager._active_actors['test-id'].actor) self.assertFalse(actor_first.deactivated) def test_fire_timer_new_activation_id_replaces_actor(self): @@ -184,7 +192,7 @@ def test_fire_timer_new_activation_id_replaces_actor(self): ) ) manager = ActorRuntime._actor_managers[actor_type] - actor_stale = manager._active_actors['test-id'] + actor_stale = manager._active_actors['test-id'].actor _run( ActorRuntime.fire_timer( @@ -192,10 +200,10 @@ def test_fire_timer_new_activation_id_replaces_actor(self): ) ) - actor_fresh = manager._active_actors['test-id'] + actor_fresh = manager._active_actors['test-id'].actor self.assertIsNot(actor_stale, actor_fresh) self.assertTrue(actor_fresh.timer_called) - self.assertEqual('act-2', manager._activation_ids['test-id']) + self.assertEqual('act-2', manager._active_actors['test-id'].activation_id) def test_fire_reminder_new_activation_id_replaces_actor(self): _run(ActorRuntime.register_actor(FakeSimpleReminderActor)) @@ -209,7 +217,7 @@ def test_fire_reminder_new_activation_id_replaces_actor(self): ) ) manager = ActorRuntime._actor_managers[actor_type] - actor_stale = manager._active_actors['test-id'] + actor_stale = manager._active_actors['test-id'].actor _run( ActorRuntime.fire_reminder( @@ -217,5 +225,99 @@ def test_fire_reminder_new_activation_id_replaces_actor(self): ) ) - self.assertIsNot(actor_stale, manager._active_actors['test-id']) - self.assertEqual('act-2', manager._activation_ids['test-id']) + self.assertIsNot(actor_stale, manager._active_actors['test-id'].actor) + self.assertEqual('act-2', manager._active_actors['test-id'].activation_id) + + def test_dispatch_ignores_activation_id_from_old_sidecar(self): + self.server.set_response(b'{"runtimeVersion": "1.19.0"}') + actor_type = FakeMultiInterfacesActor.__name__ + request_body = self._serializer.serialize({'message': 'hello dapr'}) + _run( + ActorRuntime.dispatch( + actor_type, 'test-id', 'ActionMethod', request_body, activation_id='act-1' + ) + ) + manager = ActorRuntime._actor_managers[actor_type] + actor = manager._active_actors['test-id'].actor + + _run( + ActorRuntime.dispatch( + actor_type, 'test-id', 'ActionMethod', request_body, activation_id='act-2' + ) + ) + + self.assertIs(actor, manager._active_actors['test-id'].actor) + self.assertIsNone(manager._active_actors['test-id'].activation_id) + self.assertEqual('/v1.0/metadata', self.server.request_path()) + + +class ActivationIdGateTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + ActorRuntime._activation_id_trusted = None + ActorRuntime._version_lookup = None + self.addCleanup(setattr, ActorRuntime, '_activation_id_trusted', None) + self.addCleanup(setattr, ActorRuntime, '_version_lookup', None) + self.client = mock.Mock() + self.client.get_runtime_version = mock.AsyncMock() + self.manager = SimpleNamespace(_runtime_ctx=SimpleNamespace(dapr_client=self.client)) + + async def _gate(self, activation_id='act-1'): + return await ActorRuntime._trusted_activation_id(self.manager, activation_id) + + async def test_versions(self): + cases = { + '1.19.1': True, + '1.19.1-rc.1': True, + '1.19.2': True, + '1.20.0': True, + '2.0.0': True, + '1.19.0': False, + '1.19.0-rc.2': False, + '1.18.5': False, + 'edge': False, + 'dev': False, + '': False, + 'v1.19.1': False, + } + for version, trusted in cases.items(): + with self.subTest(version=version): + ActorRuntime._activation_id_trusted = None + ActorRuntime._version_lookup = None + self.client.get_runtime_version.return_value = version + self.assertEqual('act-1' if trusted else None, await self._gate()) + + async def test_result_is_cached(self): + self.client.get_runtime_version.return_value = '1.19.1' + self.assertEqual('act-1', await self._gate()) + self.assertEqual('act-2', await self._gate('act-2')) + self.client.get_runtime_version.assert_awaited_once() + + async def test_no_lookup_without_id(self): + self.assertIsNone(await self._gate(None)) + self.assertEqual('', await self._gate('')) + self.client.get_runtime_version.assert_not_awaited() + + async def test_failed_lookup_ignores_id_then_retries(self): + self.client.get_runtime_version.side_effect = [RuntimeError('down'), '1.19.1'] + self.assertIsNone(await self._gate()) + self.assertIsNone(ActorRuntime._activation_id_trusted) + self.assertEqual('act-1', await self._gate()) + self.assertEqual(2, self.client.get_runtime_version.await_count) + + async def test_lookup_without_override_ignores_id(self): + self.client.get_runtime_version.side_effect = NotImplementedError + self.assertIsNone(await self._gate()) + + async def test_concurrent_callers_share_one_lookup(self): + release = asyncio.Event() + + async def version(): + await release.wait() + return '1.19.1' + + self.client.get_runtime_version.side_effect = version + gates = [asyncio.ensure_future(self._gate(f'act-{i}')) for i in range(5)] + await asyncio.sleep(0) + release.set() + self.assertEqual([f'act-{i}' for i in range(5)], await asyncio.gather(*gates)) + self.client.get_runtime_version.assert_awaited_once() diff --git a/tests/actor/test_grpc_host.py b/tests/actor/test_grpc_host.py index 07efee668..754de4c89 100644 --- a/tests/actor/test_grpc_host.py +++ b/tests/actor/test_grpc_host.py @@ -256,6 +256,9 @@ def setUp(self): HostTestActor.deactivated_ids.clear() HostTestActor.append_held = None HostTestActor.append_release = None + server.runtime_version = '1.19.1' + ActorRuntime._activation_id_trusted = None + ActorRuntime._version_lookup = None self._saved_managers = dict(ActorRuntime._actor_managers) self._saved_config = ActorRuntime.get_actor_config() ActorRuntime._actor_managers.clear() @@ -270,6 +273,8 @@ async def asyncTearDown(self): ActorRuntime._actor_managers.clear() ActorRuntime._actor_managers.update(self._saved_managers) ActorRuntime.set_actor_config(self._saved_config) + ActorRuntime._activation_id_trusted = None + ActorRuntime._version_lookup = None async def _serve_callbacks(self, *rounds, end='hold'): """Registers HostTestActor and starts the host against one scripted @@ -398,7 +403,8 @@ async def test_reentrancy_id_reaches_actor_context(self): def _held_activation_id(self) -> Optional[str]: manager = ActorRuntime._actor_managers['HostTestActor'] - return manager._activation_ids.get('a1') + active = manager._active_actors.get('a1') + return active.activation_id if active else None async def test_invoke_activation_id_is_stored(self): invoke = _invoke_message('req-1', 'GetName', metadata={'Dapr-Actor-Activation-Id': 'act-1'}) @@ -421,6 +427,18 @@ async def test_invoke_new_activation_id_replaces_actor(self): self.assertEqual(['a1'], HostTestActor.deactivated_ids) self.assertEqual('act-2', self._held_activation_id()) + async def test_invoke_activation_id_ignored_on_old_sidecar(self): + self._fake_dapr_server.runtime_version = '1.19.0' + await self._serve_callbacks( + [_invoke_message('req-1', 'GetName', metadata={'Dapr-Actor-Activation-Id': 'act-1'})], + [_invoke_message('req-2', 'GetName', metadata={'Dapr-Actor-Activation-Id': 'act-2'})], + ) + + await self._await_replies(2) + + self.assertEqual([], HostTestActor.deactivated_ids) + self.assertIsNone(self._held_activation_id()) + async def test_invoke_without_activation_id_keeps_actor(self): await self._serve_callbacks( [_invoke_message('req-1', 'GetName', metadata={'Dapr-Actor-Activation-Id': 'act-1'})], diff --git a/tests/clients/fake_dapr_server.py b/tests/clients/fake_dapr_server.py index e212834b8..fbf514c87 100644 --- a/tests/clients/fake_dapr_server.py +++ b/tests/clients/fake_dapr_server.py @@ -49,6 +49,7 @@ def __init__(self, grpc_port: int = 0, http_port: int = 0): self.shutdown_received = False self.locks_to_owner = {} # (store_name, resource_id) -> lock_owner self.metadata: Dict[str, str] = {} + self.runtime_version = '1.19.1' self.jobs: Dict[str, api_v1.Job] = {} self.job_overwrites: Dict[str, bool] = {} # When True the stable Job RPCs return UNIMPLEMENTED so the client must @@ -642,6 +643,7 @@ def GetMetadata(self, request, context): ), ], extended_metadata=self.metadata, + runtime_version=self.runtime_version, ) def ConverseAlpha1(self, request, context):