From 50d6c19ef43edb013464d9d1896a9c7a985e989f Mon Sep 17 00:00:00 2001 From: mesrop2008 <104313591+mesrop2008@users.noreply.github.com> Date: Tue, 18 Aug 2026 02:27:43 +0000 Subject: [PATCH] fix(api): clear stale cancellation signals when resuming a paused workflow (#40905) --- api/core/app/apps/execution_coordinator.py | 40 +++++++ api/core/app/apps/workflow/app_runner.py | 3 +- .../app_generate/workflow_execute_task.py | 7 ++ .../app/apps/test_execution_coordinator.py | 55 +++++++++- .../tasks/test_workflow_execute_task.py | 102 ++++++++++++++++++ 5 files changed, 205 insertions(+), 2 deletions(-) diff --git a/api/core/app/apps/execution_coordinator.py b/api/core/app/apps/execution_coordinator.py index ce8e8986307..1372d55155e 100644 --- a/api/core/app/apps/execution_coordinator.py +++ b/api/core/app/apps/execution_coordinator.py @@ -8,6 +8,7 @@ from enum import Enum, auto from configs import dify_config from extensions.ext_redis import redis_client +from graphon.graph_engine.command_channels import RedisChannel from graphon.graph_engine.manager import GraphEngineManager logger = logging.getLogger(__name__) @@ -20,6 +21,11 @@ class AppExecutionState(Enum): TERMINAL = auto() +def app_task_command_channel_key(task_id: str) -> str: + """Redis key of the GraphEngine command channel for one app task.""" + return f"workflow:{task_id}:commands" + + def set_app_task_stop_flag(task_id: str) -> None: if not task_id: return @@ -27,6 +33,40 @@ def set_app_task_stop_flag(task_id: str) -> None: redis_client.setex(f"generate_task_stopped:{task_id}", 600, 1) +def clear_app_task_cancellation_signals(task_id: str) -> None: + """Discard cancellation signals left over from earlier attempts of one task. + + Both cancellation channels are keyed by task ID and outlive the attempt that + armed them: the stop flag lives for 600 seconds and a queued ``AbortCommand`` + for an hour, and neither is consumed while no engine is running. A resumed + workflow deliberately reuses the paused run's task ID, so without this reset + it inherits those signals and aborts itself as soon as it starts. Call this + only when starting a new attempt that is meant to run, never mid-execution. + """ + if not task_id: + return + + try: + redis_client.delete(f"generate_task_stopped:{task_id}") + except Exception: + logger.exception("Failed to clear stop flag for app task %s", task_id) + + channel_key = app_task_command_channel_key(task_id) + try: + # fetch_commands() drains the queue and its pending marker together; the + # explicit delete covers a queue whose marker was already consumed. + discarded = RedisChannel(redis_client, channel_key).fetch_commands() + redis_client.delete(channel_key) + if discarded: + logger.info( + "Discarded %s stale GraphEngine command(s) for app task %s", + len(discarded), + task_id, + ) + except Exception: + logger.exception("Failed to clear pending GraphEngine commands for app task %s", task_id) + + class AppExecutionCoordinator: """Own cancellation policy for one app execution attempt. diff --git a/api/core/app/apps/workflow/app_runner.py b/api/core/app/apps/workflow/app_runner.py index d7427408792..1ea7b26da29 100644 --- a/api/core/app/apps/workflow/app_runner.py +++ b/api/core/app/apps/workflow/app_runner.py @@ -4,6 +4,7 @@ from collections.abc import Sequence from typing import cast from core.app.apps.base_app_queue_manager import AppQueueManager +from core.app.apps.execution_coordinator import app_task_command_channel_key from core.app.apps.workflow.app_config_manager import WorkflowAppConfig from core.app.apps.workflow.command_channels import ( CelerySignalCommandChannel, @@ -153,7 +154,7 @@ class WorkflowAppRunner(WorkflowBasedAppRunner): # RUN WORKFLOW # Create Redis command channel for this workflow execution task_id = self.application_generate_entity.task_id - channel_key = f"workflow:{task_id}:commands" + channel_key = app_task_command_channel_key(task_id) celery_signal_channel = CelerySignalCommandChannel( shutdown_state_getter=celery_warm_shutdown_started, abort_reason=WORKFLOW_WARM_SHUTDOWN_ABORT_REASON, diff --git a/api/tasks/app_generate/workflow_execute_task.py b/api/tasks/app_generate/workflow_execute_task.py index 8a88ff4dfa6..3cd30f3b5b7 100644 --- a/api/tasks/app_generate/workflow_execute_task.py +++ b/api/tasks/app_generate/workflow_execute_task.py @@ -12,6 +12,7 @@ from sqlalchemy import Engine, select from sqlalchemy.orm import Session, sessionmaker from core.app.apps.advanced_chat.app_generator import AdvancedChatAppGenerator +from core.app.apps.execution_coordinator import clear_app_task_cancellation_signals from core.app.apps.message_based_app_generator import MessageBasedAppGenerator from core.app.apps.workflow.app_generator import WorkflowAppGenerator from core.app.entities.app_invoke_entities import ( @@ -558,6 +559,12 @@ def _resume_app_execution(payload: dict[str, Any]) -> None: ) return + # The resumed attempt reuses the paused run's task ID, so cancellation + # signals armed against that ID before or during the pause would abort it + # immediately and report it as stopped by the user. This attempt is starting + # deliberately, so drop them before any engine can observe them. + clear_app_task_cancellation_signals(generate_entity.task_id) + workflow_run_repo.resume_workflow_pause(workflow_run_id, pause_entity) pause_config = PauseStateLayerConfig( diff --git a/api/tests/unit_tests/core/app/apps/test_execution_coordinator.py b/api/tests/unit_tests/core/app/apps/test_execution_coordinator.py index 4e25532650d..b926e694202 100644 --- a/api/tests/unit_tests/core/app/apps/test_execution_coordinator.py +++ b/api/tests/unit_tests/core/app/apps/test_execution_coordinator.py @@ -2,7 +2,12 @@ from unittest.mock import Mock, patch import pytest -from core.app.apps.execution_coordinator import AppExecutionCoordinator, AppExecutionState +from core.app.apps.execution_coordinator import ( + AppExecutionCoordinator, + AppExecutionState, + app_task_command_channel_key, + clear_app_task_cancellation_signals, +) def test_listener_close_does_not_abort_running_attempt() -> None: @@ -74,6 +79,54 @@ def test_pausing_started_attempt_cancels_watchdog() -> None: assert coordinator.state is AppExecutionState.PAUSED +def test_clearing_cancellation_signals_drops_stop_flag_and_queued_commands() -> None: + channel = Mock() + channel.fetch_commands.return_value = [Mock()] + with ( + patch("core.app.apps.execution_coordinator.redis_client") as redis_client, + patch("core.app.apps.execution_coordinator.RedisChannel", return_value=channel) as redis_channel, + ): + clear_app_task_cancellation_signals("task") + + redis_channel.assert_called_once_with(redis_client, "workflow:task:commands") + channel.fetch_commands.assert_called_once_with() + assert redis_client.delete.call_args_list == [ + (("generate_task_stopped:task",), {}), + (("workflow:task:commands",), {}), + ] + + +def test_clearing_cancellation_signals_ignores_empty_task_id() -> None: + with ( + patch("core.app.apps.execution_coordinator.redis_client") as redis_client, + patch("core.app.apps.execution_coordinator.RedisChannel") as redis_channel, + ): + clear_app_task_cancellation_signals("") + + redis_client.delete.assert_not_called() + redis_channel.assert_not_called() + + +def test_clearing_cancellation_signals_survives_command_channel_failure( + caplog: pytest.LogCaptureFixture, +) -> None: + with ( + patch("core.app.apps.execution_coordinator.redis_client") as redis_client, + patch("core.app.apps.execution_coordinator.RedisChannel") as redis_channel, + ): + redis_channel.return_value.fetch_commands.side_effect = RuntimeError("redis read failed") + + clear_app_task_cancellation_signals("task") + + # The stop flag is cleared first, so a command-channel failure cannot leave it armed. + redis_client.delete.assert_called_once_with("generate_task_stopped:task") + assert "Failed to clear pending GraphEngine commands for app task task" in caplog.text + + +def test_command_channel_key_matches_the_channel_the_stop_command_targets() -> None: + assert app_task_command_channel_key("task") == "workflow:task:commands" + + def test_stop_flag_failure_does_not_block_graph_stop(caplog: pytest.LogCaptureFixture) -> None: on_timeout = Mock() with ( diff --git a/api/tests/unit_tests/tasks/test_workflow_execute_task.py b/api/tests/unit_tests/tasks/test_workflow_execute_task.py index 7c7f8f34b08..c319d307c56 100644 --- a/api/tests/unit_tests/tasks/test_workflow_execute_task.py +++ b/api/tests/unit_tests/tasks/test_workflow_execute_task.py @@ -852,6 +852,108 @@ def test_resume_app_execution_returns_early_when_advanced_chat_missing_conversat resume_advanced_chat.assert_not_called() +def test_resume_app_execution_clears_stale_cancellation_signals_before_resuming( + monkeypatch: pytest.MonkeyPatch, + sqlite_engine: Engine, + sqlite_session_factory: sessionmaker[Session], +): + """A resumed run reuses the paused task ID, so it must not inherit its cancellation signals. + + Regression test for #40878: a stop flag or queued AbortCommand left over from + an earlier attempt of the same task aborted the resumed run, which then + finished as "Stopped by user". + """ + workflow_run_id = "run-id" + _persist_resumption_models(sqlite_session_factory, workflow_run_id=workflow_run_id) + + monkeypatch.setattr("tasks.app_generate.workflow_execute_task.db", SimpleNamespace(engine=sqlite_engine)) + + pause_entity = MagicMock() + pause_entity.get_state.return_value = b"state" + + workflow_run_repo = MagicMock() + workflow_run_repo.get_workflow_pause.return_value = pause_entity + monkeypatch.setattr( + "tasks.app_generate.workflow_execute_task.DifyAPIRepositoryFactory.create_api_workflow_run_repository", + lambda *_args, **_kwargs: workflow_run_repo, + ) + + generate_entity = _build_workflow_generate_entity(stream=False) + resumption_context = MagicMock() + resumption_context.serialized_graph_runtime_state = "{}" + resumption_context.get_generate_entity.return_value = generate_entity + monkeypatch.setattr( + "tasks.app_generate.workflow_execute_task.WorkflowResumptionContext.loads", + lambda *_args, **_kwargs: resumption_context, + ) + monkeypatch.setattr( + "tasks.app_generate.workflow_execute_task.GraphRuntimeState.from_snapshot", + lambda *_args, **_kwargs: MagicMock(), + ) + monkeypatch.setattr( + "tasks.app_generate.workflow_execute_task._resolve_user_for_run", lambda *_args, **_kwargs: MagicMock() + ) + + calls: list[str] = [] + clear_signals = MagicMock(side_effect=lambda task_id: calls.append(f"clear:{task_id}")) + resume_workflow = MagicMock(side_effect=lambda **_kwargs: calls.append("resume")) + monkeypatch.setattr("tasks.app_generate.workflow_execute_task.clear_app_task_cancellation_signals", clear_signals) + monkeypatch.setattr("tasks.app_generate.workflow_execute_task._resume_workflow", resume_workflow) + + _resume_app_execution({"workflow_run_id": workflow_run_id}) + + clear_signals.assert_called_once_with(generate_entity.task_id) + # Clearing after the engine started would let it observe the stale abort first. + assert calls == [f"clear:{generate_entity.task_id}", "resume"] + + +def test_resume_app_execution_keeps_cancellation_signals_when_resume_is_abandoned( + monkeypatch: pytest.MonkeyPatch, + sqlite_engine: Engine, + sqlite_session_factory: sessionmaker[Session], +): + """No attempt is starting, so nothing may clear the signals guarding this task.""" + workflow_run_id = "run-id" + _persist_resumption_models(sqlite_session_factory, workflow_run_id=workflow_run_id) + + monkeypatch.setattr("tasks.app_generate.workflow_execute_task.db", SimpleNamespace(engine=sqlite_engine)) + + pause_entity = MagicMock() + pause_entity.get_state.return_value = b"state" + + workflow_run_repo = MagicMock() + workflow_run_repo.get_workflow_pause.return_value = pause_entity + monkeypatch.setattr( + "tasks.app_generate.workflow_execute_task.DifyAPIRepositoryFactory.create_api_workflow_run_repository", + lambda *_args, **_kwargs: workflow_run_repo, + ) + + # Missing conversation id makes the advanced-chat resume bail out before running. + generate_entity = _build_advanced_chat_generate_entity(conversation_id=None) + resumption_context = MagicMock() + resumption_context.serialized_graph_runtime_state = "{}" + resumption_context.get_generate_entity.return_value = generate_entity + monkeypatch.setattr( + "tasks.app_generate.workflow_execute_task.WorkflowResumptionContext.loads", + lambda *_args, **_kwargs: resumption_context, + ) + monkeypatch.setattr( + "tasks.app_generate.workflow_execute_task.GraphRuntimeState.from_snapshot", + lambda *_args, **_kwargs: MagicMock(), + ) + monkeypatch.setattr( + "tasks.app_generate.workflow_execute_task._resolve_user_for_run", lambda *_args, **_kwargs: MagicMock() + ) + + clear_signals = MagicMock() + monkeypatch.setattr("tasks.app_generate.workflow_execute_task.clear_app_task_cancellation_signals", clear_signals) + monkeypatch.setattr("tasks.app_generate.workflow_execute_task._resume_advanced_chat", MagicMock()) + + _resume_app_execution({"workflow_run_id": workflow_run_id}) + + clear_signals.assert_not_called() + + def test_resume_advanced_chat_publishes_events_for_originally_blocking_runs( monkeypatch: pytest.MonkeyPatch, sqlite_session: Session,