Log each unexpected history event once per execution

dapr/ext/workflow/_durabletask/worker.py

@@ -1418,6 +1418,7 @@ def __init__(self, instance_id: str):
         self._encoded_custom_status: Optional[str] = None
         self._orchestrator_version_name: Optional[str] = None
         self._version_name: Optional[str] = None
+        self._reported_unexpected_events: set[tuple[str, int]] = set()
         self._history_patches: dict[str, bool] = {}
         self._applied_patches: dict[str, bool] = {}
         self._encountered_patches: list[str] = []
@@ -1983,6 +1984,28 @@ def execute(
             patches=ctx._encountered_patches,
         )
 
+    def _log_unexpected_event(
+        self, ctx: _RuntimeOrchestrationContext, event_name: str, event_id: int
+    ) -> None:
+        """Logs an unexpected history event, once per execution.
+
+        A single work item can carry the same unexpected event many times, so
+        logging every occurrence floods the caller's logs with identical lines.
+
+        Args:
+            ctx: The orchestration context the event belongs to.
+            event_name: Name of the history event field, as sent by the sidecar.
+            event_id: Id of the task or timer the event refers to.
+        """
+        if ctx.is_replaying:
+            return
+        if (event_name, event_id) in ctx._reported_unexpected_events:
+            return
+        ctx._reported_unexpected_events.add((event_name, event_id))
+        self._logger.warning(
+            f'{ctx.instance_id}: Ignoring unexpected {event_name} event with ID = {event_id}.'
+        )
+
     def process_event(self, ctx: _RuntimeOrchestrationContext, event: pb.HistoryEvent) -> None:
         if self._is_suspended and _is_suspendable(event):
             # We are suspended, so we need to buffer this event until we are resumed
@@ -2074,10 +2097,7 @@ def process_event(self, ctx: _RuntimeOrchestrationContext, event: pb.HistoryEven
                 timer_task = ctx._pending_tasks.pop(timer_id, None)
                 if not timer_task:
                     # TODO: Should this be an error? When would it ever happen?
-                    if not ctx._is_replaying:
-                        self._logger.warning(
-                            f'{ctx.instance_id}: Ignoring unexpected timerFired event with ID = {timer_id}.'
-                        )
+                    self._log_unexpected_event(ctx, 'timerFired', timer_id)
                     return
                 timer_task.complete(None)
                 if timer_task._retryable_parent is not None:
@@ -2128,10 +2148,7 @@ def process_event(self, ctx: _RuntimeOrchestrationContext, event: pb.HistoryEven
                 activity_task = ctx._pending_tasks.pop(task_id, None)
                 if not activity_task:
                     # TODO: Should this be an error? When would it ever happen?
-                    if not ctx.is_replaying:
-                        self._logger.warning(
-                            f'{ctx.instance_id}: Ignoring unexpected taskCompleted event with ID = {task_id}.'
-                        )
+                    self._log_unexpected_event(ctx, 'taskCompleted', task_id)
                     return
                 result = None
                 if not ph.is_empty(event.taskCompleted.result):
@@ -2143,10 +2160,7 @@ def process_event(self, ctx: _RuntimeOrchestrationContext, event: pb.HistoryEven
                 activity_task = ctx._pending_tasks.pop(task_id, None)
                 if not activity_task:
                     # TODO: Should this be an error? When would it ever happen?
-                    if not ctx.is_replaying:
-                        self._logger.warning(
-                            f'{ctx.instance_id}: Ignoring unexpected taskFailed event with ID = {task_id}.'
-                        )
+                    self._log_unexpected_event(ctx, 'taskFailed', task_id)
                     return
 
                 if isinstance(activity_task, task.RetryableTask):
@@ -2214,10 +2228,7 @@ def process_event(self, ctx: _RuntimeOrchestrationContext, event: pb.HistoryEven
                 sub_orch_task = ctx._pending_tasks.pop(task_id, None)
                 if not sub_orch_task:
                     # TODO: Should this be an error? When would it ever happen?
-                    if not ctx.is_replaying:
-                        self._logger.warning(
-                            f'{ctx.instance_id}: Ignoring unexpected childWorkflowInstanceCompleted event with ID = {task_id}.'
-                        )
+                    self._log_unexpected_event(ctx, 'childWorkflowInstanceCompleted', task_id)
                     return
                 result = None
                 if not ph.is_empty(event.childWorkflowInstanceCompleted.result):
@@ -2230,10 +2241,7 @@ def process_event(self, ctx: _RuntimeOrchestrationContext, event: pb.HistoryEven
                 sub_orch_task = ctx._pending_tasks.pop(task_id, None)
                 if not sub_orch_task:
                     # TODO: Should this be an error? When would it ever happen?
-                    if not ctx.is_replaying:
-                        self._logger.warning(
-                            f'{ctx.instance_id}: Ignoring unexpected childWorkflowInstanceFailed event with ID = {task_id}.'
-                        )
+                    self._log_unexpected_event(ctx, 'childWorkflowInstanceFailed', task_id)
                     return
                 if isinstance(sub_orch_task, task.RetryableTask):
                     if sub_orch_task._retry_policy is not None:

tests/ext/workflow/durabletask/test_orchestration_executor.py

@@ -2730,3 +2730,34 @@ def get_and_validate_single_complete_workflow_action(
     assert type(actions[0]) is pb.WorkflowAction
     assert actions[0].HasField('completeWorkflow')
     return actions[0].completeWorkflow
+
+
+def test_unexpected_timer_fired_is_logged_once_per_id(caplog):
+    """Tests that a repeated unexpected timerFired event is only reported once"""
+
+    def orchestrator(ctx: task.OrchestrationContext, _):
+        yield ctx.wait_for_external_event('never_arrives')
+        return 'done'
+
+    registry = worker._Registry()
+    name = registry.add_orchestrator(orchestrator)
+
+    old_events = [
+        helpers.new_workflow_started_event(),
+        helpers.new_execution_started_event(name, TEST_INSTANCE_ID, encoded_input=None),
+    ]
+    fire_at = datetime.now()
+    new_events = [
+        helpers.new_timer_fired_event(timer_id=5, fire_at=fire_at),
+        helpers.new_timer_fired_event(timer_id=5, fire_at=fire_at),
+        helpers.new_timer_fired_event(timer_id=7, fire_at=fire_at),
+    ]
+
+    executor = worker._OrchestrationExecutor(registry, TEST_LOGGER)
+    with caplog.at_level(logging.WARNING, logger=TEST_LOGGER.name):
+        executor.execute(TEST_INSTANCE_ID, old_events, new_events)
+
+    assert [r.getMessage() for r in caplog.records] == [
+        f'{TEST_INSTANCE_ID}: Ignoring unexpected timerFired event with ID = 5.',
+        f'{TEST_INSTANCE_ID}: Ignoring unexpected timerFired event with ID = 7.',
+    ]