From 611d2830e6af2992df322dc48c59b6d9233d4e5b Mon Sep 17 00:00:00 2001 From: "Jean Pierre Mandujano G." Date: Sun, 13 Sep 2026 00:47:12 -0500 Subject: [PATCH] Log each unexpected history event once per execution A work item can carry the same unexpected event many times, and every occurrence was logged, filling the caller's logs with identical warnings. Report each (event, id) pair once per execution instead. Signed-off-by: Jean Pierre Mandujano G. --- dapr/ext/workflow/_durabletask/worker.py | 48 +++++++++++-------- .../test_orchestration_executor.py | 31 ++++++++++++ 2 files changed, 59 insertions(+), 20 deletions(-) diff --git a/dapr/ext/workflow/_durabletask/worker.py b/dapr/ext/workflow/_durabletask/worker.py index 090eb747a..c8407b2f2 100644 --- a/dapr/ext/workflow/_durabletask/worker.py +++ b/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: diff --git a/tests/ext/workflow/durabletask/test_orchestration_executor.py b/tests/ext/workflow/durabletask/test_orchestration_executor.py index cdbbbdee0..7222abd2d 100644 --- a/tests/ext/workflow/durabletask/test_orchestration_executor.py +++ b/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.', + ]