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.', + ]