feat(actor): Add reminder failure policy

dapr/actor/__init__.py

@@ -16,6 +16,7 @@
 from dapr.actor.actor_interface import ActorInterface, actormethod
 from dapr.actor.client.proxy import ActorProxy, ActorProxyFactory
 from dapr.actor.id import ActorId
+from dapr.actor.runtime._failure_policy import ActorReminderFailurePolicy
 from dapr.actor.runtime.actor import Actor
 from dapr.actor.runtime.remindable import Remindable
 from dapr.actor.runtime.runtime import ActorRuntime
@@ -26,6 +27,7 @@
     'ActorProxyFactory',
     'ActorId',
     'Actor',
+    'ActorReminderFailurePolicy',
     'ActorRuntime',
     'Remindable',
     'actormethod',

dapr/actor/runtime/_failure_policy.py

@@ -0,0 +1,107 @@
+# -*- coding: utf-8 -*-
+
+"""
+Copyright 2024 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.
+"""
+
+from datetime import timedelta
+from typing import Any, Dict, Optional
+
+
+class ActorReminderFailurePolicy:
+    """Defines what happens when an actor reminder fails to trigger.
+
+    Use :meth:`drop_policy` to discard failed ticks without retrying, or
+    :meth:`constant_policy` to retry at a fixed interval.
+
+    Attributes:
+        drop: whether this is a drop (no-retry) policy.
+        interval: the retry interval for a constant policy.
+        max_retries: the maximum number of retries for a constant policy.
+    """
+
+    def __init__(
+        self,
+        *,
+        drop: bool = False,
+        interval: Optional[timedelta] = None,
+        max_retries: Optional[int] = None,
+    ):
+        """Creates a new :class:`ActorReminderFailurePolicy` instance.
+
+        Args:
+            drop (bool): if True, creates a drop policy that discards the reminder
+                tick on failure without retrying. Cannot be combined with interval
+                or max_retries.
+            interval (datetime.timedelta): the retry interval for a constant policy.
+            max_retries (int): the maximum number of retries for a constant policy.
+                If not set, retries indefinitely.
+
+        Raises:
+            ValueError: if drop is combined with interval or max_retries, or if
+                neither drop=True nor at least one of interval/max_retries is provided.
+        """
+        if drop and (interval is not None or max_retries is not None):
+            raise ValueError('drop policy cannot be combined with interval or max_retries')
+        if not drop and interval is None and max_retries is None:
+            raise ValueError(
+                'specify either drop=True or at least one of interval or max_retries'
+            )
+        self._drop = drop
+        self._interval = interval
+        self._max_retries = max_retries
+
+    @classmethod
+    def drop_policy(cls) -> 'ActorReminderFailurePolicy':
+        """Returns a policy that drops the reminder tick on failure (no retry)."""
+        return cls(drop=True)
+
+    @classmethod
+    def constant_policy(
+        cls,
+        interval: Optional[timedelta] = None,
+        max_retries: Optional[int] = None,
+    ) -> 'ActorReminderFailurePolicy':
+        """Returns a policy that retries at a constant interval on failure.
+
+        Args:
+            interval (datetime.timedelta): the time between retry attempts.
+            max_retries (int): the maximum number of retry attempts. If not set,
+                retries indefinitely.
+        """
+        return cls(interval=interval, max_retries=max_retries)
+
+    @property
+    def drop(self) -> bool:
+        """Returns True if this is a drop policy."""
+        return self._drop
+
+    @property
+    def interval(self) -> Optional[timedelta]:
+        """Returns the retry interval for a constant policy."""
+        return self._interval
+
+    @property
+    def max_retries(self) -> Optional[int]:
+        """Returns the maximum retries for a constant policy."""
+        return self._max_retries
+
+    def as_dict(self) -> Dict[str, Any]:
+        """Gets :class:`ActorReminderFailurePolicy` as a dict object."""
+        if self._drop:
+            return {'drop': {}}
+        d: Dict[str, Any] = {}
+        if self._interval is not None:
+            d['interval'] = self._interval
+        if self._max_retries is not None:
+            d['maxRetries'] = self._max_retries
+        return {'constant': d}

dapr/actor/runtime/_reminder_data.py

@@ -17,6 +17,8 @@
 from datetime import timedelta
 from typing import Any, Dict, Optional
 
+from dapr.actor.runtime._failure_policy import ActorReminderFailurePolicy
+
 
 class ActorReminderData:
     """The class that holds actor reminder data.
@@ -28,6 +30,7 @@ class ActorReminderData:
             for the first time.
         period: the time interval between reminder invocations after
             the first invocation.
+        failure_policy: the optional policy for handling reminder failures.
     """
 
     def __init__(
@@ -37,6 +40,7 @@ def __init__(
         due_time: timedelta,
         period: Optional[timedelta] = None,
         ttl: Optional[timedelta] = None,
+        failure_policy: Optional[ActorReminderFailurePolicy] = None,
     ):
         """Creates new :class:`ActorReminderData` instance.
 
@@ -49,11 +53,14 @@ def __init__(
             period (datetime.timedelta): the time interval between reminder
                 invocations after the first invocation.
             ttl (Optional[datetime.timedelta]): the time interval before the reminder stops firing.
+            failure_policy (Optional[ActorReminderFailurePolicy]): the policy for handling
+                reminder failures. If not set, the Dapr runtime default applies (3 retries).
         """
         self._reminder_name = reminder_name
         self._due_time = due_time
         self._period = period
         self._ttl = ttl
+        self._failure_policy = failure_policy
 
         if not isinstance(state, bytes):
             raise ValueError(f'only bytes are allowed for state: {type(state)}')
@@ -85,6 +92,11 @@ def ttl(self) -> Optional[timedelta]:
         """Gets ttl of Actor Reminder."""
         return self._ttl
 
+    @property
+    def failure_policy(self) -> Optional[ActorReminderFailurePolicy]:
+        """Gets the failure policy of Actor Reminder."""
+        return self._failure_policy
+
     def as_dict(self) -> Dict[str, Any]:
         """Gets :class:`ActorReminderData` as a dict object."""
         encoded_state = None
@@ -100,6 +112,9 @@ def as_dict(self) -> Dict[str, Any]:
         if self._ttl is not None:
             reminderDict.update({'ttl': self._ttl})
 
+        if self._failure_policy is not None:
+            reminderDict.update({'failurePolicy': self._failure_policy.as_dict()})
+
         return reminderDict
 
     @classmethod

dapr/actor/runtime/actor.py

@@ -18,6 +18,7 @@
 from typing import Any, Optional
 
 from dapr.actor.id import ActorId
+from dapr.actor.runtime._failure_policy import ActorReminderFailurePolicy
 from dapr.actor.runtime._method_context import ActorMethodContext
 from dapr.actor.runtime._reminder_data import ActorReminderData
 from dapr.actor.runtime._timer_data import TIMER_CALLBACK, ActorTimerData
@@ -113,6 +114,7 @@ async def register_reminder(
         due_time: timedelta,
         period: Optional[timedelta] = None,
         ttl: Optional[timedelta] = None,
+        failure_policy: Optional[ActorReminderFailurePolicy] = None,
     ) -> None:
         """Registers actor reminder.
 
@@ -131,9 +133,11 @@ async def register_reminder(
                 for the first time.
             period (datetime.timedelta): the time interval between reminder invocations after
                 the first invocation.
-            ttl (datetime.timedelta): the time interval before the reminder stops firing
+            ttl (datetime.timedelta): the time interval before the reminder stops firing.
+            failure_policy (ActorReminderFailurePolicy): the optional policy for handling reminder
+                failures. If not set, the Dapr runtime default applies (3 retries per tick).
         """
-        reminder = ActorReminderData(name, state, due_time, period, ttl)
+        reminder = ActorReminderData(name, state, due_time, period, ttl, failure_policy)
         req_body = self._runtime_ctx.message_serializer.serialize(reminder.as_dict())
         await self._runtime_ctx.dapr_client.register_reminder(
             self._runtime_ctx.actor_type_info.type_name, self.id.id, name, req_body

dapr/actor/runtime/mock_actor.py

@@ -17,6 +17,7 @@
 from typing import Any, Optional, TypeVar
 
 from dapr.actor.id import ActorId
+from dapr.actor.runtime._failure_policy import ActorReminderFailurePolicy
 from dapr.actor.runtime._reminder_data import ActorReminderData
 from dapr.actor.runtime._timer_data import TIMER_CALLBACK, ActorTimerData
 from dapr.actor.runtime.actor import Actor
@@ -88,6 +89,7 @@ async def register_reminder(
         due_time: timedelta,
         period: Optional[timedelta] = None,
         ttl: Optional[timedelta] = None,
+        failure_policy: Optional[ActorReminderFailurePolicy] = None,
     ) -> None:
         """Adds actor reminder to self._state_manager._mock_reminders.
 
@@ -98,9 +100,11 @@ async def register_reminder(
                 for the first time.
             period (datetime.timedelta): the time interval between reminder invocations after
                 the first invocation.
-            ttl (datetime.timedelta): the time interval before the reminder stops firing
+            ttl (datetime.timedelta): the time interval before the reminder stops firing.
+            failure_policy (ActorReminderFailurePolicy): the optional policy for handling reminder
+                failures. If not set, the Dapr runtime default applies (3 retries per tick).
         """
-        reminder = ActorReminderData(name, state, due_time, period, ttl)
+        reminder = ActorReminderData(name, state, due_time, period, ttl, failure_policy)
         self._state_manager._mock_reminders[name] = reminder  # type: ignore
 
     async def unregister_reminder(self, name: str) -> None:

tests/actor/test_failure_policy.py

@@ -0,0 +1,87 @@
+# -*- coding: utf-8 -*-
+
+"""
+Copyright 2024 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 datetime import timedelta
+
+from dapr.actor.runtime._failure_policy import ActorReminderFailurePolicy
+
+
+class ActorReminderFailurePolicyTests(unittest.TestCase):
+    # --- drop_policy ---
+
+    def test_drop_policy_factory(self):
+        p = ActorReminderFailurePolicy.drop_policy()
+        self.assertTrue(p.drop)
+        self.assertIsNone(p.interval)
+        self.assertIsNone(p.max_retries)
+
+    def test_drop_policy_as_dict(self):
+        p = ActorReminderFailurePolicy.drop_policy()
+        self.assertEqual({'drop': {}}, p.as_dict())
+
+    # --- constant_policy ---
+
+    def test_constant_policy_interval_and_max_retries(self):
+        p = ActorReminderFailurePolicy.constant_policy(
+            interval=timedelta(seconds=5), max_retries=3
+        )
+        self.assertFalse(p.drop)
+        self.assertEqual(timedelta(seconds=5), p.interval)
+        self.assertEqual(3, p.max_retries)
+
+    def test_constant_policy_as_dict_full(self):
+        p = ActorReminderFailurePolicy.constant_policy(
+            interval=timedelta(seconds=5), max_retries=3
+        )
+        self.assertEqual({'constant': {'interval': timedelta(seconds=5), 'maxRetries': 3}}, p.as_dict())
+
+    def test_constant_policy_interval_only(self):
+        p = ActorReminderFailurePolicy.constant_policy(interval=timedelta(seconds=10))
+        self.assertEqual({'constant': {'interval': timedelta(seconds=10)}}, p.as_dict())
+
+    def test_constant_policy_max_retries_only(self):
+        p = ActorReminderFailurePolicy.constant_policy(max_retries=5)
+        self.assertEqual({'constant': {'maxRetries': 5}}, p.as_dict())
+
+    # --- validation errors ---
+
+    def test_drop_with_interval_raises(self):
+        with self.assertRaises(ValueError):
+            ActorReminderFailurePolicy(drop=True, interval=timedelta(seconds=1))
+
+    def test_drop_with_max_retries_raises(self):
+        with self.assertRaises(ValueError):
+            ActorReminderFailurePolicy(drop=True, max_retries=3)
+
+    def test_drop_with_both_raises(self):
+        with self.assertRaises(ValueError):
+            ActorReminderFailurePolicy(drop=True, interval=timedelta(seconds=1), max_retries=3)
+
+    def test_no_policy_specified_raises(self):
+        with self.assertRaises(ValueError):
+            ActorReminderFailurePolicy()
+
+    # --- direct constructor ---
+
+    def test_direct_drop_constructor(self):
+        p = ActorReminderFailurePolicy(drop=True)
+        self.assertEqual({'drop': {}}, p.as_dict())
+
+    def test_direct_constant_constructor(self):
+        p = ActorReminderFailurePolicy(interval=timedelta(seconds=2), max_retries=1)
+        self.assertEqual(
+            {'constant': {'interval': timedelta(seconds=2), 'maxRetries': 1}}, p.as_dict()
+        )