From 64448c23ee399cd53c22eb857bb14a7e6ef1ae3a Mon Sep 17 00:00:00 2001 From: Casper Nielsen Date: Mon, 12 Oct 2026 21:37:40 +0000 Subject: [PATCH 1/3] Refresh the default actor state tracker in place after a reentrant save (#1237) * Actors: invalidate the default state tracker after a reentrant save With reentrancy enabled, each dispatched method call gets its own state change tracker, but activation, reminders and timers run on the default tracker because no reentrancy id reaches them. A key read during activation stays cached there with change kind none forever, while method calls write the same key through their own trackers. A reminder callback that later reads that key is served the stale activation value. An app that skips its write because the value looks unchanged loses that write silently: nothing is logged anywhere, because no write is ever issued. Drop the default tracker's clean copies of keys written through a reentrancy-scoped tracker, so the next read reloads them from the runtime. Reported in dapr/dapr#10532, where a reminder callback's read-modify-write of an actor state key never persisted while the identical write from an ordinary method call did, and only with reentrancy enabled. Should be backported. Signed-off-by: joshvanl * fix(actor): refresh the default state tracker in place after a reentrant save Instead of dropping the default tracker's clean copy of a key that a reentrant call saved, replace it with the saved value and ttl so the next read from activation, a reminder or a timer is served from cache rather than costing an extra state store read. Removed keys are still dropped and entries with pending changes are left alone. The cached value is passed through the state serializer first, so it has the same shape a fresh read would return (for example a tuple comes back as a list). This mirrors dapr/dotnet-sdk#1912. Signed-off-by: Casper Nielsen Co-Authored-By: Claude Opus 5.5 * fix(actor): drop the default entry when a reentrant refresh cannot match a fresh read Two cases fall back to #1227's eviction instead of an in-place refresh: a saved None value, which the state provider leaves out of the write, and a state serializer that fails to decode the value after the save has already committed. The save no longer raises after a successful write. Signed-off-by: Casper Nielsen Co-Authored-By: Claude Opus 5.5 * fix(actor): evict TTL'd keys from the default tracker and log refresh failures Co-Authored-By: Claude Opus 5.5 Signed-off-by: Casper Nielsen --------- Signed-off-by: joshvanl Signed-off-by: Casper Nielsen Co-authored-by: joshvanl Co-authored-by: Claude Opus 5.5 (cherry picked from commit fb2cd884cdda22e2e3a623d0225bdb352e84a433) Signed-off-by: Samantha Coyle --- dapr/actor/runtime/_state_provider.py | 4 + dapr/actor/runtime/state_manager.py | 38 +++++++ tests/actor/test_state_manager.py | 151 ++++++++++++++++++++++++++ 3 files changed, 193 insertions(+) diff --git a/dapr/actor/runtime/_state_provider.py b/dapr/actor/runtime/_state_provider.py index eeb1e4995..86752d150 100644 --- a/dapr/actor/runtime/_state_provider.py +++ b/dapr/actor/runtime/_state_provider.py @@ -51,6 +51,10 @@ async def try_load_state( result = self._state_serializer.deserialize(raw_state_value, state_type) return (True, result) + def round_trip_state_value(self, value: Any) -> Any: + """Returns value as try_load_state would decode it after a save, without a store read.""" + return self._state_serializer.deserialize(self._state_serializer.serialize(value), object) + async def contains_state(self, actor_type: str, actor_id: str, state_name: str) -> bool: raw_state_value = await self._state_client.get_state(actor_type, actor_id, state_name) return (raw_state_value is not None) and len(raw_state_value) > 0 diff --git a/dapr/actor/runtime/state_manager.py b/dapr/actor/runtime/state_manager.py index c2882debb..b04e50892 100644 --- a/dapr/actor/runtime/state_manager.py +++ b/dapr/actor/runtime/state_manager.py @@ -14,6 +14,7 @@ """ import asyncio +import logging from contextvars import ContextVar from typing import TYPE_CHECKING, Any, Callable, Dict, Generic, List, Optional, Tuple, TypeVar @@ -23,6 +24,8 @@ if TYPE_CHECKING: from dapr.actor.runtime.actor import Actor +logger = logging.getLogger(__name__) + T = TypeVar('T') CONTEXT: ContextVar[Optional[Dict[str, Any]]] = ContextVar('state_tracker_context') @@ -267,6 +270,41 @@ async def save_state(self) -> None: ) for state_name in states_to_remove: state_change_tracker.pop(state_name, None) + if state_change_tracker is not self._default_state_change_tracker: + self._refresh_default_tracker(state_changes) + + def _refresh_default_tracker(self, state_changes: List[ActorStateChange]) -> None: + # Writes made through a reentrancy-scoped tracker are invisible to the default + # tracker, which activation, reminders and timers read from. Refresh its clean + # copies of the written keys in place, in the shape a fresh read would return, + # and drop removed keys. Entries with pending changes are left alone. + state_provider = self._actor.runtime_ctx.state_provider + for change in state_changes: + metadata = self._default_state_change_tracker.get(change.state_name) + if metadata is None or metadata.change_kind != StateChangeKind.none: + continue + # A None value is not written to the store, and a TTL'd key can expire there + # while a cached copy would not, so let the next read reload both. + is_evicted = ( + change.change_kind == StateChangeKind.remove + or change.value is None + or change.ttl_in_seconds is not None + ) + if is_evicted: + self._default_state_change_tracker.pop(change.state_name) + continue + try: + value = state_provider.round_trip_state_value(change.value) + except Exception as exc: + # The save has already committed; fall back to reloading on the next read. + logger.warning( + 'could not refresh cached actor state %s: %s', change.state_name, exc + ) + self._default_state_change_tracker.pop(change.state_name) + continue + self._default_state_change_tracker[change.state_name] = StateMetadata( + value, StateChangeKind.none + ) def is_state_marked_for_remove(self, state_name: str) -> bool: state_change_tracker = self._get_contextual_state_tracker() diff --git a/tests/actor/test_state_manager.py b/tests/actor/test_state_manager.py index 5ed4e24e6..34ff332d2 100644 --- a/tests/actor/test_state_manager.py +++ b/tests/actor/test_state_manager.py @@ -22,6 +22,7 @@ from dapr.actor.runtime.context import ActorRuntimeContext from dapr.actor.runtime.state_change import StateChangeKind from dapr.actor.runtime.state_manager import ActorStateManager, StateMetadata +from dapr.common.reentrancy_context import reentrancy_ctx from dapr.serializers import DefaultJSONSerializer from tests.actor.fake_actor_classes import FakeSimpleActor from tests.actor.fake_client import FakeDaprActorClient @@ -118,6 +119,156 @@ def test_get_state_for_removed_value(self): self.assertFalse(has_value) self.assertIsNone(val) + def _run_reentrant(self, state_manager, coro_fn): + # Runs coro_fn inside a reentrancy-scoped call, then saves its tracker. + token = reentrancy_ctx.set('reentrancy-id') + try: + state_manager.set_state_context('ctx1') + try: + _run(coro_fn()) + _run(state_manager.save_state()) + finally: + state_manager.set_state_context(None) + finally: + reentrancy_ctx.reset(token) + + @mock.patch( + 'tests.actor.fake_client.FakeDaprActorClient.get_state', + new=_async_mock(return_value=b'"value1"'), + ) + @mock.patch( + 'tests.actor.fake_client.FakeDaprActorClient.save_state_transactionally', new=_async_mock() + ) + def test_reentrant_update_refreshes_default_tracker(self): + state_manager = ActorStateManager(self._fake_actor) + _run(state_manager.try_get_state('state1')) + + self._run_reentrant(state_manager, lambda: state_manager.set_state('state1', 'v2')) + + # The default read is served from the refreshed entry, not the state store. + calls = self._fake_client.get_state.mock.call_count + has_value, val = _run(state_manager.try_get_state('state1')) + self.assertTrue(has_value) + self.assertEqual('v2', val) + self.assertEqual(calls, self._fake_client.get_state.mock.call_count) + state = state_manager._default_state_change_tracker['state1'] + self.assertEqual(StateChangeKind.none, state.change_kind) + + @mock.patch( + 'tests.actor.fake_client.FakeDaprActorClient.get_state', + new=_async_mock(return_value=b'"value1"'), + ) + @mock.patch( + 'tests.actor.fake_client.FakeDaprActorClient.save_state_transactionally', new=_async_mock() + ) + def test_reentrant_ttl_save_evicts_default_tracker(self): + state_manager = ActorStateManager(self._fake_actor) + _run(state_manager.try_get_state('state1')) + + self._run_reentrant(state_manager, lambda: state_manager.set_state_ttl('state1', 'v2', 60)) + + # A cached copy would outlive the key's expiry in the store. + self.assertNotIn('state1', state_manager._default_state_change_tracker) + + @mock.patch( + 'tests.actor.fake_client.FakeDaprActorClient.get_state', + new=_async_mock(return_value=b'"value1"'), + ) + @mock.patch( + 'tests.actor.fake_client.FakeDaprActorClient.save_state_transactionally', new=_async_mock() + ) + def test_reentrant_remove_evicts_default_tracker(self): + state_manager = ActorStateManager(self._fake_actor) + _run(state_manager.try_get_state('state1')) + + self._run_reentrant(state_manager, lambda: state_manager.remove_state('state1')) + + self.assertNotIn('state1', state_manager._default_state_change_tracker) + self._fake_client.get_state.mock.return_value = None + has_value, val = _run(state_manager.try_get_state('state1')) + self.assertFalse(has_value) + self.assertIsNone(val) + + @mock.patch( + 'tests.actor.fake_client.FakeDaprActorClient.get_state', + new=_async_mock(return_value=b'"value1"'), + ) + @mock.patch( + 'tests.actor.fake_client.FakeDaprActorClient.save_state_transactionally', new=_async_mock() + ) + def test_reentrant_save_keeps_dirty_default_entry(self): + state_manager = ActorStateManager(self._fake_actor) + _run(state_manager.set_state('state1', 'pending')) + + self._run_reentrant(state_manager, lambda: state_manager.set_state('state1', 'v2')) + + state = state_manager._default_state_change_tracker['state1'] + self.assertEqual('pending', state.value) + self.assertEqual(StateChangeKind.update, state.change_kind) + + @mock.patch( + 'tests.actor.fake_client.FakeDaprActorClient.get_state', + new=_async_mock(return_value=b'[1, 2]'), + ) + @mock.patch( + 'tests.actor.fake_client.FakeDaprActorClient.save_state_transactionally', new=_async_mock() + ) + def test_reentrant_refresh_matches_fresh_read(self): + state_manager = ActorStateManager(self._fake_actor) + _run(state_manager.try_get_state('state1')) + + self._run_reentrant(state_manager, lambda: state_manager.set_state('state1', (3, 4))) + + _, refreshed = _run(state_manager.try_get_state('state1')) + self._fake_client.get_state.mock.return_value = self._serializer.serialize((3, 4)) + _, fresh = _run(ActorStateManager(self._fake_actor).try_get_state('state1')) + self.assertEqual([3, 4], fresh) + self.assertEqual(fresh, refreshed) + + @mock.patch( + 'tests.actor.fake_client.FakeDaprActorClient.get_state', + new=_async_mock(return_value=b'"value1"'), + ) + @mock.patch( + 'tests.actor.fake_client.FakeDaprActorClient.save_state_transactionally', new=_async_mock() + ) + def test_reentrant_save_of_none_evicts_default_tracker(self): + state_manager = ActorStateManager(self._fake_actor) + _run(state_manager.try_get_state('state1')) + + self._run_reentrant(state_manager, lambda: state_manager.set_state('state1', None)) + + self.assertNotIn('state1', state_manager._default_state_change_tracker) + + @mock.patch( + 'tests.actor.fake_client.FakeDaprActorClient.get_state', + new=_async_mock(return_value=b'"value1"'), + ) + @mock.patch( + 'tests.actor.fake_client.FakeDaprActorClient.save_state_transactionally', new=_async_mock() + ) + def test_reentrant_refresh_failure_evicts_default_tracker(self): + state_manager = ActorStateManager(self._fake_actor) + _run(state_manager.try_get_state('state1')) + _run(state_manager.try_get_state('state2')) + + async def set_both(): + await state_manager.set_state('state1', 'v2') + await state_manager.set_state('state2', 'v2') + + # The save must not raise after the write has committed, and every key is handled. + with mock.patch.object( + self._runtime_ctx.state_provider, + 'round_trip_state_value', + side_effect=ValueError('cannot decode'), + ): + with self.assertLogs('dapr.actor.runtime.state_manager', level='WARNING') as logs: + self._run_reentrant(state_manager, set_both) + + self.assertEqual(2, len(logs.records)) + self.assertNotIn('state1', state_manager._default_state_change_tracker) + self.assertNotIn('state2', state_manager._default_state_change_tracker) + @mock.patch('tests.actor.fake_client.FakeDaprActorClient.get_state', new=_async_mock()) def test_set_state_for_new_state(self): state_manager = ActorStateManager(self._fake_actor) From f9aefe55ae9d59964b7618f5aa888a5ad17c1563 Mon Sep 17 00:00:00 2001 From: Casper Nielsen Date: Mon, 12 Oct 2026 22:23:59 +0000 Subject: [PATCH 2/3] Remember missing actor state keys instead of re-reading them (#1238) * Actors: invalidate the default state tracker after a reentrant save With reentrancy enabled, each dispatched method call gets its own state change tracker, but activation, reminders and timers run on the default tracker because no reentrancy id reaches them. A key read during activation stays cached there with change kind none forever, while method calls write the same key through their own trackers. A reminder callback that later reads that key is served the stale activation value. An app that skips its write because the value looks unchanged loses that write silently: nothing is logged anywhere, because no write is ever issued. Drop the default tracker's clean copies of keys written through a reentrancy-scoped tracker, so the next read reloads them from the runtime. Reported in dapr/dapr#10532, where a reminder callback's read-modify-write of an actor state key never persisted while the identical write from an ordinary method call did, and only with reentrancy enabled. Should be backported. Signed-off-by: joshvanl * fix(actor): refresh the default state tracker in place after a reentrant save Instead of dropping the default tracker's clean copy of a key that a reentrant call saved, replace it with the saved value and ttl so the next read from activation, a reminder or a timer is served from cache rather than costing an extra state store read. Removed keys are still dropped and entries with pending changes are left alone. The cached value is passed through the state serializer first, so it has the same shape a fresh read would return (for example a tuple comes back as a list). This mirrors dapr/dotnet-sdk#1912. Signed-off-by: Casper Nielsen Co-Authored-By: Claude Opus 5.5 * fix(actor): drop the default entry when a reentrant refresh cannot match a fresh read Two cases fall back to #1227's eviction instead of an in-place refresh: a saved None value, which the state provider leaves out of the write, and a state serializer that fails to decode the value after the save has already committed. The save no longer raises after a successful write. Signed-off-by: Casper Nielsen Co-Authored-By: Claude Opus 5.5 * feat(actor): cache state that is not found in the default state tracker Reading a key that does not exist went back to the state store every time, because a miss was not cached. try_get_state now records a miss as a clean "not found" entry in the default tracker, so later reads, contains_state, try_remove_state and remove_state answer without I/O, and try_add_state, set_state, set_state_ttl, get_or_add_state and add_or_update_state turn it into an add without asking the store again. get_state_names skips it and save_state never sends it. Misses are not cached in reentrant trackers. Nothing refreshes an outer reentrant call's tracker when a nested reentrant call (A -> B -> A) saves, so a cached miss there would hide the nested write and then overwrite it through get_or_add_state or try_add_state. The entry is a private StateMetadata subclass that reports the existing "none" change kind, so the public StateChangeKind enum, ActorStateChange and subclasses of ActorStateManager are unchanged. A reentrant save refreshes a "not found" default entry the same way as a clean one: a write replaces it with the saved value and a remove evicts it. Without that, a reminder that cached a miss would keep seeing the key as absent after a reentrant call created it. Ports dapr/dotnet-sdk#1913 and dapr/dotnet-sdk#1916. Signed-off-by: Casper Nielsen Co-Authored-By: Claude Opus 5.5 * test(actor): run the nested reentrant test on Python 3.10 create_task(context=...) only exists from Python 3.11. A task already runs in a copy of the caller's context, so the argument isn't needed. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Casper Nielsen * test(actor): read the empty payload through a fresh state manager The cached miss served the second read from the tracker, so the empty payload never reached the state provider. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Casper Nielsen * fix(actor): log when a reentrant save cannot refresh cached state Co-Authored-By: Claude Opus 5.5 Signed-off-by: Casper Nielsen --------- Signed-off-by: joshvanl Signed-off-by: Casper Nielsen Co-authored-by: joshvanl Co-authored-by: Claude Opus 5.5 (cherry picked from commit dbf98673b525ea820a677000a457d605c2f023ef) Signed-off-by: Samantha Coyle --- dapr/actor/runtime/state_manager.py | 54 ++++++-- tests/actor/test_state_manager.py | 194 +++++++++++++++++++++++++++- 2 files changed, 232 insertions(+), 16 deletions(-) diff --git a/dapr/actor/runtime/state_manager.py b/dapr/actor/runtime/state_manager.py index b04e50892..293a8b5d3 100644 --- a/dapr/actor/runtime/state_manager.py +++ b/dapr/actor/runtime/state_manager.py @@ -63,6 +63,21 @@ def ttl_in_seconds(self, new_ttl_in_seconds: Optional[int]) -> None: self._ttl_in_seconds = new_ttl_in_seconds +class _NotFoundStateMetadata(StateMetadata[None]): + # A cached miss: the key is known to be absent from the state store. It reports the + # public 'none' change kind, so it is never saved, and is told apart by its type. + def __init__(self) -> None: + super().__init__(None, StateChangeKind.none) + + +def _is_not_found(state_metadata: StateMetadata) -> bool: + return isinstance(state_metadata, _NotFoundStateMetadata) + + +def _is_absent(state_metadata: StateMetadata) -> bool: + return _is_not_found(state_metadata) or state_metadata.change_kind == StateChangeKind.remove + + class ActorStateManager(Generic[T]): def __init__(self, actor: 'Actor'): self._actor = actor @@ -80,6 +95,9 @@ async def try_add_state(self, state_name: str, value: T) -> bool: state_change_tracker = self._get_contextual_state_tracker() if state_name in state_change_tracker: state_metadata = state_change_tracker[state_name] + if _is_not_found(state_metadata): + state_change_tracker[state_name] = StateMetadata(value, StateChangeKind.add) + return True if state_metadata.change_kind == StateChangeKind.remove: state_change_tracker[state_name] = StateMetadata(value, StateChangeKind.update) return True @@ -105,7 +123,7 @@ async def try_get_state(self, state_name: str) -> Tuple[bool, Optional[T]]: state_change_tracker = self._get_contextual_state_tracker() if state_name in state_change_tracker: state_metadata = state_change_tracker[state_name] - if state_metadata.change_kind == StateChangeKind.remove: + if _is_absent(state_metadata): return False, None return True, state_metadata.value has_value, val = await self._actor.runtime_ctx.state_provider.try_load_state( @@ -113,6 +131,10 @@ async def try_get_state(self, state_name: str) -> Tuple[bool, Optional[T]]: ) if has_value: state_change_tracker[state_name] = StateMetadata(val, StateChangeKind.none) + elif state_change_tracker is self._default_state_change_tracker: + # Only the default tracker is refreshed by reentrant saves, so a miss cached + # in a reentrant tracker could hide a nested call's write and overwrite it. + state_change_tracker[state_name] = _NotFoundStateMetadata() return has_value, val async def set_state(self, state_name: str, value: T) -> None: @@ -125,6 +147,11 @@ async def set_state_ttl(self, state_name: str, value: T, ttl_in_seconds: Optiona state_change_tracker = self._get_contextual_state_tracker() if state_name in state_change_tracker: state_metadata = state_change_tracker[state_name] + if _is_not_found(state_metadata): + state_change_tracker[state_name] = StateMetadata( + value, StateChangeKind.add, ttl_in_seconds + ) + return state_metadata.value = value state_metadata.ttl_in_seconds = ttl_in_seconds @@ -156,7 +183,7 @@ async def try_remove_state(self, state_name: str) -> bool: state_change_tracker = self._get_contextual_state_tracker() if state_name in state_change_tracker: state_metadata = state_change_tracker[state_name] - if state_metadata.change_kind == StateChangeKind.remove: + if _is_absent(state_metadata): return False elif state_metadata.change_kind == StateChangeKind.add: state_change_tracker.pop(state_name, None) @@ -175,8 +202,7 @@ async def try_remove_state(self, state_name: str) -> bool: async def contains_state(self, state_name: str) -> bool: state_change_tracker = self._get_contextual_state_tracker() if state_name in state_change_tracker: - state_metadata = state_change_tracker[state_name] - return state_metadata.change_kind != StateChangeKind.remove + return not _is_absent(state_change_tracker[state_name]) return await self._actor.runtime_ctx.state_provider.contains_state( self._type_name, self._actor.id.id, state_name ) @@ -203,6 +229,9 @@ async def add_or_update_state( state_change_tracker = self._get_contextual_state_tracker() if state_name in state_change_tracker: state_metadata = state_change_tracker[state_name] + if _is_not_found(state_metadata): + state_change_tracker[state_name] = StateMetadata(value, StateChangeKind.add) + return value if state_metadata.change_kind == StateChangeKind.remove: state_change_tracker[state_name] = StateMetadata(value, StateChangeKind.update) return value @@ -228,11 +257,7 @@ async def get_state_names(self) -> List[str]: # TODO: Get all state names from Dapr once implemented. def append_names_sync(): state_change_tracker = self._get_contextual_state_tracker() - return [ - key - for key, value in state_change_tracker.items() - if value.change_kind != StateChangeKind.remove - ] + return [key for key, value in state_change_tracker.items() if not _is_absent(value)] default_loop = asyncio.get_running_loop() return await default_loop.run_in_executor(None, append_names_sync) @@ -276,8 +301,8 @@ async def save_state(self) -> None: def _refresh_default_tracker(self, state_changes: List[ActorStateChange]) -> None: # Writes made through a reentrancy-scoped tracker are invisible to the default # tracker, which activation, reminders and timers read from. Refresh its clean - # copies of the written keys in place, in the shape a fresh read would return, - # and drop removed keys. Entries with pending changes are left alone. + # copies of the written keys in place, cached misses included, in the shape a fresh + # read would return, and drop removed keys. Entries with pending changes are left alone. state_provider = self._actor.runtime_ctx.state_provider for change in state_changes: metadata = self._default_state_change_tracker.get(change.state_name) @@ -295,10 +320,13 @@ def _refresh_default_tracker(self, state_changes: List[ActorStateChange]) -> Non continue try: value = state_provider.round_trip_state_value(change.value) - except Exception as exc: + except Exception: # The save has already committed; fall back to reloading on the next read. logger.warning( - 'could not refresh cached actor state %s: %s', change.state_name, exc + 'could not refresh cached actor state %s after reentrant save; ' + 'it will be reloaded', + change.state_name, + exc_info=True, ) self._default_state_change_tracker.pop(change.state_name) continue diff --git a/tests/actor/test_state_manager.py b/tests/actor/test_state_manager.py index 34ff332d2..b201b3fe9 100644 --- a/tests/actor/test_state_manager.py +++ b/tests/actor/test_state_manager.py @@ -13,6 +13,7 @@ limitations under the License. """ +import asyncio import base64 import unittest from unittest import mock @@ -86,11 +87,15 @@ def test_get_state_for_no_state(self): self.assertFalse(has_value) self.assertIsNone(val) - # Test if the test value is empty string - self._fake_client.get_state.return_value = '' - has_value, val = _run(state_manager.try_get_state('state1')) + # A fresh manager, because the miss above is now cached in the default tracker. + get_state_mock = self._fake_client.get_state.mock + get_state_mock.return_value = '' + get_state_mock.reset_mock(return_value=False) + state_manager_empty = ActorStateManager(self._fake_actor) + has_value, val = _run(state_manager_empty.try_get_state('state1')) self.assertFalse(has_value) self.assertIsNone(val) + get_state_mock.assert_called_once() @mock.patch( 'tests.actor.fake_client.FakeDaprActorClient.get_state', @@ -269,6 +274,189 @@ async def set_both(): self.assertNotIn('state1', state_manager._default_state_change_tracker) self.assertNotIn('state2', state_manager._default_state_change_tracker) + def _cache_miss(self, state_manager, state_name='state1'): + # Reads an absent key so the tracker caches the miss; returns the store read count. + has_value, _ = _run(state_manager.try_get_state(state_name)) + self.assertFalse(has_value) + return self._fake_client.get_state.mock.call_count + + @mock.patch('tests.actor.fake_client.FakeDaprActorClient.get_state', new=_async_mock()) + def test_missing_state_is_cached_as_absent(self): + state_manager = ActorStateManager(self._fake_actor) + calls = self._cache_miss(state_manager) + + self.assertEqual((False, None), _run(state_manager.try_get_state('state1'))) + with self.assertRaises(KeyError): + _run(state_manager.get_state('state1')) + self.assertFalse(_run(state_manager.contains_state('state1'))) + self.assertFalse(state_manager.is_state_marked_for_remove('state1')) + self.assertEqual([], _run(state_manager.get_state_names())) + self.assertEqual(calls, self._fake_client.get_state.mock.call_count) + + @mock.patch('tests.actor.fake_client.FakeDaprActorClient.get_state', new=_async_mock()) + def test_try_add_state_after_cached_miss(self): + state_manager = ActorStateManager(self._fake_actor) + calls = self._cache_miss(state_manager) + + self.assertTrue(_run(state_manager.try_add_state('state1', 'value1'))) + self.assertEqual(calls, self._fake_client.get_state.mock.call_count) + state = state_manager._default_state_change_tracker['state1'] + self.assertEqual(StateChangeKind.add, state.change_kind) + self.assertEqual('value1', state.value) + self.assertEqual((True, 'value1'), _run(state_manager.try_get_state('state1'))) + + @mock.patch('tests.actor.fake_client.FakeDaprActorClient.get_state', new=_async_mock()) + def test_set_state_after_cached_miss(self): + state_manager = ActorStateManager(self._fake_actor) + self._cache_miss(state_manager, 'state1') + calls = self._cache_miss(state_manager, 'state2') + + _run(state_manager.set_state('state1', 'value1')) + _run(state_manager.set_state_ttl('state2', 'value2', 60)) + self.assertEqual(calls, self._fake_client.get_state.mock.call_count) + tracker = state_manager._default_state_change_tracker + self.assertEqual(StateChangeKind.add, tracker['state1'].change_kind) + self.assertEqual('value1', tracker['state1'].value) + self.assertIsNone(tracker['state1'].ttl_in_seconds) + self.assertEqual(StateChangeKind.add, tracker['state2'].change_kind) + self.assertEqual('value2', tracker['state2'].value) + self.assertEqual(60, tracker['state2'].ttl_in_seconds) + + @mock.patch('tests.actor.fake_client.FakeDaprActorClient.get_state', new=_async_mock()) + def test_remove_state_after_cached_miss(self): + state_manager = ActorStateManager(self._fake_actor) + calls = self._cache_miss(state_manager) + + self.assertFalse(_run(state_manager.try_remove_state('state1'))) + with self.assertRaises(KeyError): + _run(state_manager.remove_state('state1')) + self.assertEqual(calls, self._fake_client.get_state.mock.call_count) + self.assertFalse(state_manager.is_state_marked_for_remove('state1')) + + @mock.patch('tests.actor.fake_client.FakeDaprActorClient.get_state', new=_async_mock()) + def test_get_or_add_state_after_cached_miss(self): + state_manager = ActorStateManager(self._fake_actor) + calls = self._cache_miss(state_manager) + + self.assertEqual('value1', _run(state_manager.get_or_add_state('state1', 'value1'))) + self.assertEqual(calls, self._fake_client.get_state.mock.call_count) + state = state_manager._default_state_change_tracker['state1'] + self.assertEqual(StateChangeKind.add, state.change_kind) + self.assertEqual('value1', state.value) + + @mock.patch('tests.actor.fake_client.FakeDaprActorClient.get_state', new=_async_mock()) + def test_add_or_update_state_after_cached_miss(self): + state_manager = ActorStateManager(self._fake_actor) + calls = self._cache_miss(state_manager) + update_value = mock.MagicMock(return_value='updated') + + val = _run(state_manager.add_or_update_state('state1', 'value1', update_value)) + self.assertEqual('value1', val) + update_value.assert_not_called() + self.assertEqual(calls, self._fake_client.get_state.mock.call_count) + state = state_manager._default_state_change_tracker['state1'] + self.assertEqual(StateChangeKind.add, state.change_kind) + self.assertEqual('value1', state.value) + + @mock.patch('tests.actor.fake_client.FakeDaprActorClient.get_state', new=_async_mock()) + @mock.patch( + 'tests.actor.fake_client.FakeDaprActorClient.save_state_transactionally', new=_async_mock() + ) + def test_save_state_skips_cached_miss(self): + state_manager = ActorStateManager(self._fake_actor) + self._cache_miss(state_manager, 'state1') + save = self._fake_client.save_state_transactionally.mock + save.reset_mock() + + _run(state_manager.save_state()) + save.assert_not_called() + + self._cache_miss(state_manager, 'state2') + _run(state_manager.set_state('state2', 'value2')) + _run(state_manager.save_state()) + save.assert_called_once() + self.assertEqual( + b'[{"operation":"upsert","request":{"key":"state2","value":"value2"}}]', + save.call_args.args[2], + ) + self.assertEqual((False, None), _run(state_manager.try_get_state('state1'))) + + @mock.patch('tests.actor.fake_client.FakeDaprActorClient.get_state', new=_async_mock()) + @mock.patch( + 'tests.actor.fake_client.FakeDaprActorClient.save_state_transactionally', new=_async_mock() + ) + def test_reentrant_write_replaces_cached_miss_in_default_tracker(self): + state_manager = ActorStateManager(self._fake_actor) + self._cache_miss(state_manager) + + # A reentrant call reads the absent key in its own tracker, then creates it. + async def create(): + self.assertFalse(await state_manager.contains_state('state1')) + await state_manager.set_state('state1', 'value2') + + self._run_reentrant(state_manager, create) + + # The reminder-facing default tracker must not keep reporting the key as absent. + calls = self._fake_client.get_state.mock.call_count + self.assertEqual((True, 'value2'), _run(state_manager.try_get_state('state1'))) + self.assertTrue(_run(state_manager.contains_state('state1'))) + self.assertEqual(calls, self._fake_client.get_state.mock.call_count) + + @mock.patch('tests.actor.fake_client.FakeDaprActorClient.get_state', new=_async_mock()) + @mock.patch( + 'tests.actor.fake_client.FakeDaprActorClient.save_state_transactionally', new=_async_mock() + ) + def test_reentrant_remove_evicts_cached_miss_from_default_tracker(self): + state_manager = ActorStateManager(self._fake_actor) + self._cache_miss(state_manager) + + # The key is created elsewhere, then removed by a reentrant call. + self._fake_client.get_state.mock.return_value = b'"value1"' + self._run_reentrant(state_manager, lambda: state_manager.remove_state('state1')) + + self.assertNotIn('state1', state_manager._default_state_change_tracker) + + @mock.patch('tests.actor.fake_client.FakeDaprActorClient.get_state', new=_async_mock()) + @mock.patch( + 'tests.actor.fake_client.FakeDaprActorClient.save_state_transactionally', new=_async_mock() + ) + def test_nested_reentrant_write_is_seen_by_outer_call(self): + state_manager = ActorStateManager(self._fake_actor) + get_state = self._fake_client.get_state.mock + save = self._fake_client.save_state_transactionally.mock + save.reset_mock() + + async def inner(): + state_manager.set_state_context('inner') + await state_manager.set_state('state1', 'from-inner') + await state_manager.save_state() + get_state.return_value = b'"from-inner"' + + async def outer(): + state_manager.set_state_context('outer') + self.assertFalse(await state_manager.contains_state('state1')) + self.assertEqual((False, None), await state_manager.try_get_state('state1')) + # A nested reentrant call (A -> B -> A) creates the key in its own tracker. + await asyncio.create_task(inner()) + self.assertTrue(await state_manager.contains_state('state1')) + self.assertEqual( + 'from-inner', await state_manager.get_or_add_state('state1', 'default') + ) + await state_manager.save_state() + + token = reentrancy_ctx.set('reentrancy-id') + try: + _run(outer()) + finally: + state_manager.set_state_context(None) + reentrancy_ctx.reset(token) + + save.assert_called_once() + self.assertEqual( + b'[{"operation":"upsert","request":{"key":"state1","value":"from-inner"}}]', + save.call_args.args[2], + ) + @mock.patch('tests.actor.fake_client.FakeDaprActorClient.get_state', new=_async_mock()) def test_set_state_for_new_state(self): state_manager = ActorStateManager(self._fake_actor) From 8ae8387ef18c8c689505e4064ea45ba719aa6e93 Mon Sep 17 00:00:00 2001 From: Casper Nielsen Date: Tue, 13 Oct 2026 13:49:00 +0000 Subject: [PATCH 3/3] Skip the state store existence check when setting actor state (#1239) * Actors: invalidate the default state tracker after a reentrant save With reentrancy enabled, each dispatched method call gets its own state change tracker, but activation, reminders and timers run on the default tracker because no reentrancy id reaches them. A key read during activation stays cached there with change kind none forever, while method calls write the same key through their own trackers. A reminder callback that later reads that key is served the stale activation value. An app that skips its write because the value looks unchanged loses that write silently: nothing is logged anywhere, because no write is ever issued. Drop the default tracker's clean copies of keys written through a reentrancy-scoped tracker, so the next read reloads them from the runtime. Reported in dapr/dapr#10532, where a reminder callback's read-modify-write of an actor state key never persisted while the identical write from an ordinary method call did, and only with reentrancy enabled. Should be backported. Signed-off-by: joshvanl * fix(actor): refresh the default state tracker in place after a reentrant save Instead of dropping the default tracker's clean copy of a key that a reentrant call saved, replace it with the saved value and ttl so the next read from activation, a reminder or a timer is served from cache rather than costing an extra state store read. Removed keys are still dropped and entries with pending changes are left alone. The cached value is passed through the state serializer first, so it has the same shape a fresh read would return (for example a tuple comes back as a list). This mirrors dapr/dotnet-sdk#1912. Signed-off-by: Casper Nielsen Co-Authored-By: Claude Opus 5.5 * fix(actor): drop the default entry when a reentrant refresh cannot match a fresh read Two cases fall back to #1227's eviction instead of an in-place refresh: a saved None value, which the state provider leaves out of the write, and a state serializer that fails to decode the value after the save has already committed. The save no longer raises after a successful write. Signed-off-by: Casper Nielsen Co-Authored-By: Claude Opus 5.5 * feat(actor): cache state that is not found in the default state tracker Reading a key that does not exist went back to the state store every time, because a miss was not cached. try_get_state now records a miss as a clean "not found" entry in the default tracker, so later reads, contains_state, try_remove_state and remove_state answer without I/O, and try_add_state, set_state, set_state_ttl, get_or_add_state and add_or_update_state turn it into an add without asking the store again. get_state_names skips it and save_state never sends it. Misses are not cached in reentrant trackers. Nothing refreshes an outer reentrant call's tracker when a nested reentrant call (A -> B -> A) saves, so a cached miss there would hide the nested write and then overwrite it through get_or_add_state or try_add_state. The entry is a private StateMetadata subclass that reports the existing "none" change kind, so the public StateChangeKind enum, ActorStateChange and subclasses of ActorStateManager are unchanged. A reentrant save refreshes a "not found" default entry the same way as a clean one: a write replaces it with the saved value and a remove evicts it. Without that, a reminder that cached a miss would keep seeing the key as absent after a reentrant call created it. Ports dapr/dotnet-sdk#1913 and dapr/dotnet-sdk#1916. Signed-off-by: Casper Nielsen Co-Authored-By: Claude Opus 5.5 * test(actor): run the nested reentrant test on Python 3.10 create_task(context=...) only exists from Python 3.11. A task already runs in a copy of the caller's context, so the argument isn't needed. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Casper Nielsen * perf(actor): skip the state store read when setting an untracked key set_state and set_state_ttl read the state store for a key that was not in the tracker only to choose between add and update, but the state provider saves both as the same upsert. The key is now recorded as an update without that read. Update rather than add is deliberate: the key may already exist in the store, so a remove later in the same turn has to send a delete instead of dropping the pending entry. Deleting a key that turns out not to exist is harmless; skipping the delete of one that does is not. Keys the manager knows are absent keep the add kind, so add-then-remove still sends nothing: try_add_state still reads the store and fails when the key exists, and a cached "not found" entry still turns into an add. try_remove_state still reads the store to report whether anything was removed. A reentrant save refreshes the default tracker the same way whatever kind the saved change had. Ports dapr/dotnet-sdk#1914. Signed-off-by: Casper Nielsen Co-Authored-By: Claude Opus 5.5 * fix(actor): log the cause when a reentrant refresh evicts a state entry Co-Authored-By: Claude Opus 5.5 Signed-off-by: Casper Nielsen --------- Signed-off-by: joshvanl Signed-off-by: Casper Nielsen Co-authored-by: joshvanl Co-authored-by: Claude Opus 5.5 (cherry picked from commit fceda096040adcf9d203fdb567b7dcd251ace16d) Signed-off-by: Samantha Coyle --- dapr/actor/runtime/state_manager.py | 18 ++-- tests/actor/test_state_manager.py | 141 ++++++++++++++++++++++++---- 2 files changed, 129 insertions(+), 30 deletions(-) diff --git a/dapr/actor/runtime/state_manager.py b/dapr/actor/runtime/state_manager.py index 293a8b5d3..59f3ade1b 100644 --- a/dapr/actor/runtime/state_manager.py +++ b/dapr/actor/runtime/state_manager.py @@ -163,17 +163,11 @@ async def set_state_ttl(self, state_name: str, value: T, ttl_in_seconds: Optiona state_change_tracker[state_name] = state_metadata return - existed = await self._actor.runtime_ctx.state_provider.contains_state( - self._type_name, self._actor.id.id, state_name + # Add and update are both saved as an upsert, so skip the store read. Update also + # makes a later remove send a delete, since the key may already be in the store. + state_change_tracker[state_name] = StateMetadata( + value, StateChangeKind.update, ttl_in_seconds ) - if existed: - state_change_tracker[state_name] = StateMetadata( - value, StateChangeKind.update, ttl_in_seconds - ) - else: - state_change_tracker[state_name] = StateMetadata( - value, StateChangeKind.add, ttl_in_seconds - ) async def remove_state(self, state_name: str) -> None: if not await self.try_remove_state(state_name): @@ -323,8 +317,8 @@ def _refresh_default_tracker(self, state_changes: List[ActorStateChange]) -> Non except Exception: # The save has already committed; fall back to reloading on the next read. logger.warning( - 'could not refresh cached actor state %s after reentrant save; ' - 'it will be reloaded', + 'Could not round-trip actor state %r through the state serializer; ' + 'evicting it from the default state tracker', change.state_name, exc_info=True, ) diff --git a/tests/actor/test_state_manager.py b/tests/actor/test_state_manager.py index b201b3fe9..a46a0198b 100644 --- a/tests/actor/test_state_manager.py +++ b/tests/actor/test_state_manager.py @@ -15,6 +15,7 @@ import asyncio import base64 +import json import unittest from unittest import mock @@ -30,6 +31,10 @@ from tests.actor.utils import _async_mock, _run +def _operations(data): + return [(op['operation'], op['request']['key']) for op in json.loads(data)] + + class ActorStateManagerTests(unittest.TestCase): def setUp(self): # Create mock client @@ -271,6 +276,8 @@ async def set_both(): self._run_reentrant(state_manager, set_both) self.assertEqual(2, len(logs.records)) + self.assertIn("'state1'", logs.records[0].getMessage()) + self.assertIsNotNone(logs.records[0].exc_info) self.assertNotIn('state1', state_manager._default_state_change_tracker) self.assertNotIn('state2', state_manager._default_state_change_tracker) @@ -464,7 +471,7 @@ def test_set_state_for_new_state(self): _run(state_manager.set_state('state1', 'value1')) state = state_change_tracker['state1'] - self.assertEqual(StateChangeKind.add, state.change_kind) + self.assertEqual(StateChangeKind.update, state.change_kind) self.assertEqual('value1', state.value) self.assertEqual(None, state.ttl_in_seconds) @@ -475,12 +482,12 @@ def test_set_state_for_existing_state_only_in_mem(self): _run(state_manager.set_state('state1', 'value1')) state = state_change_tracker['state1'] - self.assertEqual(StateChangeKind.add, state.change_kind) + self.assertEqual(StateChangeKind.update, state.change_kind) self.assertEqual('value1', state.value) _run(state_manager.set_state('state1', 'value2')) state = state_change_tracker['state1'] - self.assertEqual(StateChangeKind.add, state.change_kind) + self.assertEqual(StateChangeKind.update, state.change_kind) self.assertEqual('value2', state.value) self.assertEqual(None, state.ttl_in_seconds) @@ -504,7 +511,7 @@ def test_set_state_ttl_for_new_state(self): _run(state_manager.set_state_ttl('state1', 'value1', 3600)) state = state_change_tracker['state1'] - self.assertEqual(StateChangeKind.add, state.change_kind) + self.assertEqual(StateChangeKind.update, state.change_kind) self.assertEqual('value1', state.value) self.assertEqual(3600, state.ttl_in_seconds) @@ -515,13 +522,13 @@ def test_set_state_ttl_for_existing_state_only_in_mem(self): _run(state_manager.set_state_ttl('state1', 'value1', 3600)) state = state_change_tracker['state1'] - self.assertEqual(StateChangeKind.add, state.change_kind) + self.assertEqual(StateChangeKind.update, state.change_kind) self.assertEqual('value1', state.value) self.assertEqual(3600, state.ttl_in_seconds) _run(state_manager.set_state_ttl('state1', 'value2', 7200)) state = state_change_tracker['state1'] - self.assertEqual(StateChangeKind.add, state.change_kind) + self.assertEqual(StateChangeKind.update, state.change_kind) self.assertEqual('value2', state.value) self.assertEqual(7200, state.ttl_in_seconds) @@ -539,6 +546,113 @@ def test_set_state_ttl_for_existing_state(self): self.assertEqual('value2', state.value) self.assertEqual(3600, state.ttl_in_seconds) + @mock.patch('tests.actor.fake_client.FakeDaprActorClient.get_state', new=_async_mock()) + @mock.patch( + 'tests.actor.fake_client.FakeDaprActorClient.save_state_transactionally', new=_async_mock() + ) + def test_set_state_does_not_read_state_store(self): + state_manager = ActorStateManager(self._fake_actor) + get_state = self._fake_client.get_state.mock + save = self._fake_client.save_state_transactionally.mock + _run(state_manager.set_state('state1', 'value1')) + _run(state_manager.set_state_ttl('state2', 'value2', 60)) + + self.assertEqual('value1', _run(state_manager.get_state('state1'))) + self.assertTrue(_run(state_manager.contains_state('state2'))) + self.assertFalse(_run(state_manager.try_add_state('state1', 'other'))) + _run(state_manager.save_state()) + + get_state.assert_not_called() + self.assertEqual( + b'[{"operation":"upsert","request":{"key":"state1","value":"value1"}},' + b'{"operation":"upsert","request":{"key":"state2","value":"value2",' + b'"metadata":{"ttlInSeconds":"60"}}}]', + save.call_args.args[2], + ) + + @mock.patch( + 'tests.actor.fake_client.FakeDaprActorClient.get_state', + new=_async_mock(return_value=b'"stored"'), + ) + @mock.patch( + 'tests.actor.fake_client.FakeDaprActorClient.save_state_transactionally', new=_async_mock() + ) + def test_set_then_remove_in_one_turn_sends_delete(self): + # The key may already be in the store, so removing it must still send a delete. + state_manager = ActorStateManager(self._fake_actor) + save = self._fake_client.save_state_transactionally.mock + _run(state_manager.set_state('state1', 'value1')) + _run(state_manager.set_state_ttl('state2', 'value2', 60)) + + self.assertTrue(_run(state_manager.try_remove_state('state1'))) + _run(state_manager.remove_state('state2')) + self.assertFalse(_run(state_manager.contains_state('state1'))) + _run(state_manager.save_state()) + + self._fake_client.get_state.mock.assert_not_called() + self.assertEqual( + [('delete', 'state1'), ('delete', 'state2')], _operations(save.call_args.args[2]) + ) + self.assertEqual({}, state_manager._default_state_change_tracker) + + @mock.patch('tests.actor.fake_client.FakeDaprActorClient.get_state', new=_async_mock()) + @mock.patch( + 'tests.actor.fake_client.FakeDaprActorClient.save_state_transactionally', new=_async_mock() + ) + def test_add_then_remove_in_one_turn_sends_nothing(self): + # try_add_state and a cached miss both know the key is absent, so nothing is sent. + state_manager = ActorStateManager(self._fake_actor) + save = self._fake_client.save_state_transactionally.mock + self.assertTrue(_run(state_manager.try_add_state('state1', 'value1'))) + self._cache_miss(state_manager, 'state2') + _run(state_manager.set_state('state2', 'value2')) + + self.assertTrue(_run(state_manager.try_remove_state('state1'))) + self.assertTrue(_run(state_manager.try_remove_state('state2'))) + _run(state_manager.save_state()) + + save.assert_not_called() + + @mock.patch( + 'tests.actor.fake_client.FakeDaprActorClient.get_state', + new=_async_mock(return_value=b'"stored"'), + ) + @mock.patch( + 'tests.actor.fake_client.FakeDaprActorClient.save_state_transactionally', new=_async_mock() + ) + def test_reentrant_set_then_remove_sends_delete_and_evicts_default(self): + state_manager = ActorStateManager(self._fake_actor) + save = self._fake_client.save_state_transactionally.mock + _run(state_manager.try_get_state('state1')) + calls = self._fake_client.get_state.mock.call_count + + async def set_then_remove(): + await state_manager.set_state('state1', 'value1') + await state_manager.remove_state('state1') + + self._run_reentrant(state_manager, set_then_remove) + + self.assertEqual(calls, self._fake_client.get_state.mock.call_count) + self.assertEqual([('delete', 'state1')], _operations(save.call_args.args[2])) + self.assertNotIn('state1', state_manager._default_state_change_tracker) + + @mock.patch( + 'tests.actor.fake_client.FakeDaprActorClient.get_state', + new=_async_mock(return_value=b'"stored"'), + ) + def test_try_add_and_try_remove_still_check_state_store(self): + state_manager = ActorStateManager(self._fake_actor) + get_state = self._fake_client.get_state.mock + self.assertFalse(_run(state_manager.try_add_state('state1', 'value1'))) + self.assertTrue(_run(state_manager.try_remove_state('state2'))) + self.assertEqual(2, get_state.call_count) + + get_state.return_value = None + state_manager = ActorStateManager(self._fake_actor) + self.assertFalse(_run(state_manager.try_remove_state('state1'))) + self.assertTrue(_run(state_manager.try_add_state('state2', 'value2'))) + self.assertEqual(4, get_state.call_count) + @mock.patch('tests.actor.fake_client.FakeDaprActorClient.get_state', new=_async_mock()) def test_set_state_ttl_lt_0_for_new_state(self): state_manager = ActorStateManager(self._fake_actor) @@ -553,13 +667,13 @@ def test_set_state_ttl_lt_0_for_existing_state_only_in_mem(self): _run(state_manager.set_state_ttl('state1', 'value1', 3600)) state = state_change_tracker['state1'] - self.assertEqual(StateChangeKind.add, state.change_kind) + self.assertEqual(StateChangeKind.update, state.change_kind) self.assertEqual('value1', state.value) self.assertEqual(3600, state.ttl_in_seconds) _run(state_manager.set_state_ttl('state1', 'value2', -3600)) state = state_change_tracker['state1'] - self.assertEqual(StateChangeKind.add, state.change_kind) + self.assertEqual(StateChangeKind.update, state.change_kind) self.assertEqual('value1', state.value) self.assertEqual(3600, state.ttl_in_seconds) @@ -734,16 +848,7 @@ def test_get_state_names(self): _run(state_manager.set_state('state3', 'value3')) names = _run(state_manager.get_state_names()) self.assertEqual(['state1', 'state2', 'state3'], names) - - self._fake_client.get_state.mock.assert_any_call( - self._test_type_info._name, self._test_actor_id.id, 'state1' - ) - self._fake_client.get_state.mock.assert_any_call( - self._test_type_info._name, self._test_actor_id.id, 'state2' - ) - self._fake_client.get_state.mock.assert_any_call( - self._test_type_info._name, self._test_actor_id.id, 'state3' - ) + self._fake_client.get_state.mock.assert_not_called() def test_get_state_names_excludes_only_removed(self): """States that still exist (add/update/none) must be listed; only a