diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5e4c5687..5ee0902b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -20,7 +20,7 @@ jobs: run: | python -m pip install --upgrade pip pip install -r requirements.txt # test with requirements file so can easily bump with dependabot - pip install pytest + pip install pytest fakeredis - name: Test run: | python -m pytest test_gui.py tests/ diff --git a/src/workflow/QueueManager.py b/src/workflow/QueueManager.py index 63bbdae9..21db9a8b 100644 --- a/src/workflow/QueueManager.py +++ b/src/workflow/QueueManager.py @@ -178,7 +178,8 @@ def get_job_info(self, job_id: str) -> Optional[JobInfo]: job = Job.fetch(job_id, connection=self._redis) - # Map RQ status to our enum + # 'stopped' is what RQ records after send_stop_job_command runs; + # surface it as CANCELED so the UI doesn't show stopped jobs as queued. status_map = { "queued": JobStatus.QUEUED, "started": JobStatus.STARTED, @@ -186,6 +187,7 @@ def get_job_info(self, job_id: str) -> Optional[JobInfo]: "failed": JobStatus.FAILED, "deferred": JobStatus.DEFERRED, "canceled": JobStatus.CANCELED, + "stopped": JobStatus.CANCELED, } status = status_map.get(job.get_status(), JobStatus.QUEUED) @@ -232,24 +234,61 @@ def cancel_job(self, job_id: str) -> bool: """ Cancel a queued or running job. + For queued jobs, this removes them from the queue. For jobs that are + already executing in a worker, Job.cancel() alone is not enough — it + only updates Redis registries while the worker keeps running the + workflow. We send a stop-job command to the worker so the work-horse + is actually interrupted. + Args: job_id: The job ID to cancel Returns: - True if successfully canceled + True if the job is canceled (or already was), False otherwise. """ if not self.is_available: return False try: + from rq.command import send_stop_job_command + from rq.exceptions import InvalidJobOperation, NoSuchJobError from rq.job import Job + except ImportError: + return False + try: job = Job.fetch(job_id, connection=self._redis) - job.cancel() + except NoSuchJobError: + return False + except Exception: + return False + + # Idempotent: a second Stop click (or rerun) should not surface an error. + if job.is_canceled or job.is_stopped: return True + + # Tell the worker to interrupt the work-horse before marking canceled. + if job.is_started and job.worker_name: + try: + send_stop_job_command(self._redis, job_id) + except InvalidJobOperation: + # The worker just finished or the job moved out of 'started'; + # fall through to cancel() to settle registry state. + pass + except Exception: + pass + + try: + job.cancel() + except InvalidJobOperation: + # Worker already transitioned the job (e.g. to 'stopped'); that + # satisfies the user's intent to stop. + pass except Exception: return False + return True + def get_queue_stats(self) -> dict: """ Get queue statistics. diff --git a/tests/test_queue_manager_cancel.py b/tests/test_queue_manager_cancel.py new file mode 100644 index 00000000..0f877082 --- /dev/null +++ b/tests/test_queue_manager_cancel.py @@ -0,0 +1,161 @@ +""" +Tests for QueueManager.cancel_job - the "stop workflow" path used in online +mode where workflows are executed by RQ workers (the vendor's queue). + +Bug being fixed: when a workflow is mid-execution in an RQ worker and the +user clicks "Stop Workflow", QueueManager.cancel_job calls Job.cancel() on +the RQ Job. For a job in the "started" state this only marks the job as +canceled in the Redis registries; the worker keeps executing the workflow +and the user sees inconsistent / "weird" state (worker still appending to +logs, status flipping around, etc.). + +To actually stop a running RQ job, RQ exposes +rq.command.send_stop_job_command(connection, job_id) which messages the +worker over Redis pubsub to interrupt the work-horse. +""" + +import os +import sys + +import pytest + +fakeredis = pytest.importorskip("fakeredis") +rq = pytest.importorskip("rq") +from rq import Queue +from rq.job import Job, JobStatus + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from src.workflow.QueueManager import QueueManager + + +def _make_queue_manager() -> QueueManager: + """Build a QueueManager wired to fake Redis, bypassing __init__.""" + qm = QueueManager.__new__(QueueManager) + qm._redis = fakeredis.FakeStrictRedis() + qm._queue = Queue(QueueManager.QUEUE_NAME, connection=qm._redis) + qm._is_online = True + qm._init_attempted = True + qm._default_timeout = 7200 + qm._default_result_ttl = 86400 + return qm + + +def _force_started(job: Job, worker_name: str = "rq:worker:test-worker") -> None: + """Move a queued job into the 'started' state with a worker assigned.""" + job.set_status(JobStatus.STARTED) + job.worker_name = worker_name + job.save() + + +def test_cancel_queued_job_marks_it_canceled(): + qm = _make_queue_manager() + qm._queue.enqueue(os.getcwd, job_id="queued-job") + + assert qm.cancel_job("queued-job") is True + + refreshed = Job.fetch("queued-job", connection=qm._redis) + assert refreshed.get_status() == JobStatus.CANCELED + + +def test_cancel_started_job_sends_stop_command_to_worker(monkeypatch): + """ + Reproduces the vendor-queue stop bug. + + A workflow that is actively running in a worker must be stopped by + sending a stop-job command to the worker. The previous implementation + only called Job.cancel(), which left the worker running. + """ + qm = _make_queue_manager() + job = qm._queue.enqueue(os.getcwd, job_id="started-job") + _force_started(job) + + stop_calls: list[str] = [] + + def fake_send_stop_job_command(connection, job_id, *args, **kwargs): + stop_calls.append(job_id) + + import rq.command as rq_command + monkeypatch.setattr( + rq_command, "send_stop_job_command", fake_send_stop_job_command + ) + + result = qm.cancel_job("started-job") + + assert result is True, "cancel_job should report success for started jobs" + assert stop_calls == ["started-job"], ( + "cancel_job must call send_stop_job_command for started jobs - " + "otherwise the RQ worker keeps running the workflow." + ) + + +def test_cancel_already_canceled_job_does_not_raise(): + """ + User double-clicks 'Stop Workflow' or stop is invoked twice on rerun. + The second call must not surface InvalidJobOperation as a 'weird error'. + """ + qm = _make_queue_manager() + job = qm._queue.enqueue(os.getcwd, job_id="dup-cancel") + job.cancel() + assert Job.fetch("dup-cancel", connection=qm._redis).get_status() == JobStatus.CANCELED + + # Must not raise; intent (job is canceled) is already satisfied. + assert qm.cancel_job("dup-cancel") is True + + +def test_cancel_missing_job_returns_false(): + qm = _make_queue_manager() + assert qm.cancel_job("does-not-exist") is False + + +def test_started_status_without_worker_is_handled_gracefully(monkeypatch): + """ + Edge case: job is marked started but has no worker_name yet (race between + worker pickup and stop click). cancel_job must not raise; it should fall + back to canceling the job in the registry. + """ + qm = _make_queue_manager() + job = qm._queue.enqueue(os.getcwd, job_id="started-no-worker") + job.set_status(JobStatus.STARTED) + job.save() + + stop_calls: list[str] = [] + + def fake_send_stop_job_command(connection, job_id, *args, **kwargs): + stop_calls.append(job_id) + + import rq.command as rq_command + monkeypatch.setattr( + rq_command, "send_stop_job_command", fake_send_stop_job_command + ) + + result = qm.cancel_job("started-no-worker") + + assert result is True + # Without a worker_name there is nothing to send the stop command to. + assert stop_calls == [] + assert ( + Job.fetch("started-no-worker", connection=qm._redis).get_status() + == JobStatus.CANCELED + ) + + +def test_stopped_status_is_mapped_in_get_job_info(monkeypatch): + """ + After send_stop_job_command runs, RQ marks the job 'stopped'. The status + map in get_job_info must recognise it; otherwise the UI would show the + job as still 'queued', which is the user-visible 'weird error'. + """ + qm = _make_queue_manager() + job = qm._queue.enqueue(os.getcwd, job_id="stopped-job") + job.set_status(JobStatus.STOPPED) + job.save() + + info = qm.get_job_info("stopped-job") + assert info is not None + assert info.status == __import__( + "src.workflow.QueueManager", fromlist=["JobStatus"] + ).JobStatus.CANCELED, ( + "RQ 'stopped' status should be reported as CANCELED to the UI; " + "otherwise stopped jobs appear stuck in 'queued'." + )