Allow infinite retries for workflows

Signed-off-by: Matheus André <92062874+matheusandre1@users.noreply.github.com>

Author: matheusandre1

examples/workflow/simple.py

@@ -31,6 +31,7 @@
 child_orchestrator_count = 0
 child_orchestrator_string = ''
 child_act_retry_count = 0
+infinite_retry_count = 0
 instance_id = 'exampleInstanceID'
 child_instance_id = 'childInstanceID'
 workflow_name = 'hello_world_wf'
@@ -48,6 +49,11 @@
     retry_timeout=timedelta(seconds=100),
 )
 
+infinite_retry_policy = RetryPolicy(
+    first_retry_interval=timedelta(seconds=1),
+    max_number_of_attempts=-1,
+)
+
 wfr = WorkflowRuntime()
 
 
@@ -57,6 +63,7 @@ def hello_world_wf(ctx: DaprWorkflowContext, wf_input):
     yield ctx.call_activity(hello_act, input=1)
     yield ctx.call_activity(hello_act, input=10)
     yield ctx.call_activity(hello_retryable_act, retry_policy=retry_policy)
+    yield ctx.call_activity(hello_infinite_retryable_act, retry_policy=infinite_retry_policy)
     yield ctx.call_child_workflow(child_retryable_wf, retry_policy=retry_policy)
 
     # Change in event handling: Use when_any to handle both event and timeout
@@ -91,6 +98,15 @@ def hello_retryable_act(ctx: WorkflowActivityContext):
     retry_count += 1
 
 
+@wfr.activity(name='hello_infinite_retryable_act')
+def hello_infinite_retryable_act(ctx: WorkflowActivityContext):
+    global infinite_retry_count
+    infinite_retry_count += 1
+    print(f'Infinite retry attempt: {infinite_retry_count}', flush=True)
+    if infinite_retry_count <= 10:
+        raise ValueError('Retryable Error')
+
+
 @wfr.workflow(name='child_retryable_wf')
 def child_retryable_wf(ctx: DaprWorkflowContext):
     global child_orchestrator_string, child_orchestrator_count
@@ -128,11 +144,12 @@ def main():
 
     wf_client.wait_for_workflow_start(instance_id)
 
-    # Sleep to let the workflow run initial activities
-    sleep(12)
+    # Sleep to let the workflow run initial activities and infinite retries
+    sleep(24)
 
     assert counter == 11
     assert retry_count == 2
+    assert infinite_retry_count == 11
     assert child_orchestrator_string == '1aa2bb3cc'
 
     # Pause Test
@@ -153,10 +170,13 @@ def main():
         state = wf_client.wait_for_workflow_completion(instance_id, timeout_in_seconds=30)
         if state.runtime_status.name == 'COMPLETED':
             print('Workflow completed! Result: {}'.format(state.serialized_output.strip('"')))
+            assert state.serialized_output.strip('"') == 'Completed'
         else:
             print(f'Workflow failed! Status: {state.runtime_status.name}')
+            raise AssertionError(f'Unexpected workflow status: {state.runtime_status.name}')
     except TimeoutError:
         print('*** Workflow timed out!')
+        raise
 
     wf_client.purge_workflow(instance_id=instance_id)
     try:

ext/dapr-ext-workflow/dapr/ext/workflow/retry_policy.py

@@ -40,6 +40,7 @@ def __init__(
         Args:
             first_retry_interval(timedelta): The retry interval to use for the first retry attempt.
             max_number_of_attempts(int):  The maximum number of retry attempts.
+                Use ``-1`` for infinite retries.
             backoff_coefficient(Optional[float]): The backoff coefficient to use for calculating
                 the next retry interval.
             max_retry_interval(Optional[timedelta]): The maximum retry interval to use for any
@@ -50,8 +51,8 @@ def __init__(
         # validate inputs
         if first_retry_interval < timedelta(seconds=0):
             raise ValueError('first_retry_interval must be >= 0')
-        if max_number_of_attempts < 1:
-            raise ValueError('max_number_of_attempts must be >= 1')
+        if max_number_of_attempts == 0 or max_number_of_attempts < -1:
+            raise ValueError('max_number_of_attempts must be >= 1 or -1 for infinite retries')
         if backoff_coefficient is not None and backoff_coefficient < 1:
             raise ValueError('backoff_coefficient must be >= 1')
         if max_retry_interval is not None and max_retry_interval < timedelta(seconds=0):

ext/dapr-ext-workflow/dapr/ext/workflow/workflow_runtime.py

@@ -246,7 +246,10 @@ def start(self):
                 try:
                     is_ready = self.wait_for_worker_ready(timeout=self._worker_ready_timeout)
                     if not is_ready:
-                        raise RuntimeError('WorkflowRuntime worker and its stream are not ready')
+                        self._logger.warning(
+                            'WorkflowRuntime worker and its stream are not ready. '
+                            'Continuing; workflows scheduled immediately may not be received.'
+                        )
                     else:
                         self._logger.debug(
                             'WorkflowRuntime worker is ready and its stream can receive work items'

ext/dapr-ext-workflow/tests/test_retry_policy.py

@@ -0,0 +1,63 @@
+# -*- 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 datetime import timedelta
+
+from dapr.ext.workflow.retry_policy import RetryPolicy
+
+
+class RetryPolicyTests(unittest.TestCase):
+    @staticmethod
+    def _execute_with_retry(operation, retry_policy: RetryPolicy):
+        attempts = 0
+        while True:
+            attempts += 1
+            try:
+                operation(attempts)
+                return attempts
+            except ValueError:
+                if (
+                    retry_policy.max_number_of_attempts != -1
+                    and attempts >= retry_policy.max_number_of_attempts
+                ):
+                    raise
+
+    def test_allow_infinite_max_number_of_attempts(self):
+        retry_policy = RetryPolicy(
+            first_retry_interval=timedelta(seconds=1), max_number_of_attempts=-1
+        )
+
+        self.assertEqual(-1, retry_policy.max_number_of_attempts)
+
+    def test_infinite_retries_succeeds_after_ten_failures(self):
+        retry_policy = RetryPolicy(
+            first_retry_interval=timedelta(seconds=1), max_number_of_attempts=-1
+        )
+
+        def flaky_operation(attempt: int):
+            if attempt <= 10:
+                raise ValueError('Retryable Error')
+
+        attempts = self._execute_with_retry(flaky_operation, retry_policy=retry_policy)
+
+        self.assertEqual(11, attempts)
+
+    def test_reject_invalid_max_number_of_attempts(self):
+        with self.assertRaises(ValueError):
+            RetryPolicy(first_retry_interval=timedelta(seconds=1), max_number_of_attempts=0)
+
+        with self.assertRaises(ValueError):
+            RetryPolicy(first_retry_interval=timedelta(seconds=1), max_number_of_attempts=-2)

ext/dapr-ext-workflow/tests/test_workflow_runtime.py

@@ -246,17 +246,18 @@ def test_start_logs_exception_when_worker_start_fails(self):
         mock_exception.assert_called_once()
         self.assertIn('did not start', mock_exception.call_args[0][0])
 
-    def test_start_raises_when_worker_not_ready(self):
+    def test_start_logs_warning_when_worker_not_ready(self):
         listActivities.clear()
         listOrchestrators.clear()
         mock.patch('durabletask.worker._Registry', return_value=FakeTaskHubGrpcWorker()).start()
         runtime = WorkflowRuntime(worker_ready_timeout=0.2)
         mock_worker = mock.MagicMock()
         mock_worker.is_worker_ready.return_value = False
         runtime._WorkflowRuntime__worker = mock_worker
-        with self.assertRaises(RuntimeError) as ctx:
+        with mock.patch.object(runtime._logger, 'warning') as mock_warning:
             runtime.start()
-        self.assertIn('not ready', str(ctx.exception))
+        mock_worker.start.assert_called_once()
+        self.assertGreaterEqual(mock_warning.call_count, 1)
 
     def test_start_logs_warning_when_no_is_worker_ready(self):
         mock_worker = mock.MagicMock(spec=['start', 'stop', '_registry'])