diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7baf813..6d79f8e 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/k8s/components/memory-tier-high/streamlit-resources.yaml b/k8s/components/memory-tier-high/streamlit-resources.yaml index e84d9fd..ed7b6e9 100644 --- a/k8s/components/memory-tier-high/streamlit-resources.yaml +++ b/k8s/components/memory-tier-high/streamlit-resources.yaml @@ -12,5 +12,5 @@ spec: memory: "512Mi" cpu: "500m" limits: - memory: "4Gi" + memory: "8Gi" cpu: "4" diff --git a/src/workflow/QueueManager.py b/src/workflow/QueueManager.py index 68878bc..21db9a8 100644 --- a/src/workflow/QueueManager.py +++ b/src/workflow/QueueManager.py @@ -55,25 +55,34 @@ class QueueManager: def __init__(self): self._redis = None self._queue = None - self._is_online = self._check_online_mode() self._init_attempted = False + settings = self._load_settings() + self._is_online = self._check_online_mode(settings) + + queue_settings = settings.get("queue_settings", {}) + self._default_timeout = queue_settings.get("default_timeout", 7200) + self._default_result_ttl = queue_settings.get("result_ttl", 86400) + if self._is_online: self._init_redis() - def _check_online_mode(self) -> bool: + @staticmethod + def _load_settings() -> dict: + """Load settings.json once; return empty dict on failure.""" + try: + with open("settings.json", "r") as f: + return json.load(f) + except Exception: + return {} + + def _check_online_mode(self, settings: dict) -> bool: """Check if running in online mode""" # Check environment variable first (set in Docker) if os.environ.get("REDIS_URL"): return True - # Fallback: check settings file - try: - with open("settings.json", "r") as f: - settings = json.load(f) - return settings.get("online_deployment", False) - except Exception: - return False + return settings.get("online_deployment", False) def _init_redis(self) -> None: """Initialize Redis connection and queue""" @@ -108,8 +117,8 @@ def submit_job( args: tuple = (), kwargs: dict = None, job_id: Optional[str] = None, - timeout: int = 7200, # 2 hour default - result_ttl: int = 86400, # 24 hours + timeout: Optional[int] = None, + result_ttl: Optional[int] = None, description: str = "" ) -> Optional[str]: """ @@ -120,8 +129,8 @@ def submit_job( args: Positional arguments for the function kwargs: Keyword arguments for the function job_id: Optional custom job ID (defaults to UUID) - timeout: Job timeout in seconds - result_ttl: How long to keep results + timeout: Job timeout in seconds (defaults to settings.json queue_settings.default_timeout) + result_ttl: How long to keep results (defaults to settings.json queue_settings.result_ttl) description: Human-readable job description Returns: @@ -131,6 +140,10 @@ def submit_job( return None kwargs = kwargs or {} + if timeout is None: + timeout = self._default_timeout + if result_ttl is None: + result_ttl = self._default_result_ttl try: job = self._queue.enqueue( @@ -165,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, @@ -173,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) @@ -219,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/src/workflow/StreamlitUI.py b/src/workflow/StreamlitUI.py index 3cca528..0af89aa 100644 --- a/src/workflow/StreamlitUI.py +++ b/src/workflow/StreamlitUI.py @@ -20,6 +20,7 @@ tk_directory_dialog, tk_file_dialog, ) +from src.workflow._log_status import classify_log_outcome class StreamlitUI: @@ -1268,9 +1269,11 @@ def execution_section( with open(log_path, "r", encoding="utf-8") as f: lines = f.readlines() content = "".join(lines) - # Check if workflow finished successfully - if "WORKFLOW FINISHED" in content: + outcome = classify_log_outcome(content) + if outcome == "finished": st.success("**Workflow completed successfully.**") + elif outcome == "cancelled": + st.warning("**Workflow was cancelled.**") else: st.error("**Errors occurred, check log file.**") # Apply line limit to static display @@ -1324,6 +1327,9 @@ def _show_queue_status(self, status: dict) -> None: with st.expander("Error Details", expanded=True): st.code(job_error) + elif job_status == "canceled": + st.warning(f"**Status: {label}** - Workflow was cancelled.") + # Expandable job details with st.expander("Job Details", expanded=False): st.code(f"""Job ID: {status.get('job_id', 'N/A')} diff --git a/src/workflow/WorkflowManager.py b/src/workflow/WorkflowManager.py index 08a4b9e..302b079 100644 --- a/src/workflow/WorkflowManager.py +++ b/src/workflow/WorkflowManager.py @@ -72,7 +72,6 @@ def _start_workflow_queued(self) -> None: "workflow_module": self.__class__.__module__, }, job_id=job_id, - timeout=7200, # 2 hour timeout description=f"Workflow: {self.name}" ) @@ -179,16 +178,28 @@ def stop_workflow(self) -> bool: """ Stop a running workflow. + Writes a "WORKFLOW CANCELLED" marker to the log so the static + run-page display can render a "Workflow was cancelled" message + instead of "Errors occurred". Cleans up the worker-side pid_dir + left behind when the RQ worker is force-stopped, so a subsequent + get_workflow_status does not flip running back to True via the + local-mode fallback. + + .job_id is intentionally left in place: get_job_info will report + the canceled status to the UI so _show_queue_status can render the + Cancelled pill. Resubmission overwrites it; RQ's result_ttl + eventually evicts the job and get_workflow_status self-heals. + Returns: - True if successfully stopped + True if a stop action was taken (queue cancel or local kill). """ # Try to cancel queue job first (online mode) if self._queue_manager and self._queue_manager.is_available: job_id = self._queue_manager.load_job_id(self.workflow_dir) if job_id: - success = self._queue_manager.cancel_job(job_id) - if success: - self._queue_manager.clear_job_id(self.workflow_dir) + if self._queue_manager.cancel_job(job_id): + self.logger.log("WORKFLOW CANCELLED") + shutil.rmtree(self.executor.pid_dir, ignore_errors=True) return True # Fallback: stop local process @@ -223,6 +234,8 @@ def _stop_local_workflow(self) -> bool: # Clean up the pid directory shutil.rmtree(pid_dir, ignore_errors=True) + if stopped: + self.logger.log("WORKFLOW CANCELLED") return stopped def show_file_upload_section(self) -> None: diff --git a/src/workflow/_log_status.py b/src/workflow/_log_status.py new file mode 100644 index 0000000..68adecf --- /dev/null +++ b/src/workflow/_log_status.py @@ -0,0 +1,30 @@ +""" +Pure helper for classifying a workflow log file's terminal state. + +Kept streamlit-free so the static-display dispatch in StreamlitUI can be +unit-tested without a Streamlit runtime. +""" + +from typing import Literal + +LogOutcome = Literal["finished", "cancelled", "error"] + +CANCELLED_MARKER = "WORKFLOW CANCELLED" +FINISHED_MARKER = "WORKFLOW FINISHED" + + +def classify_log_outcome(content: str) -> LogOutcome: + """ + Classify a workflow log's terminal state from its full text. + + Order matters: a TOPP subprocess often dies as the worker is being torn + down, so a partial 'ERROR:' line followed by the cancellation marker is + still a cancellation, not a crash. Cancellation therefore wins over + finished (defensive — both shouldn't appear) and over the implicit error + fallback. + """ + if CANCELLED_MARKER in content: + return "cancelled" + if FINISHED_MARKER in content: + return "finished" + return "error" diff --git a/tests/test_log_status.py b/tests/test_log_status.py new file mode 100644 index 0000000..951f290 --- /dev/null +++ b/tests/test_log_status.py @@ -0,0 +1,50 @@ +""" +Tests for the classify_log_outcome helper used by the run-page static display. + +The UI must render three different messages depending on what's in the +workflow log: + finished -> "Workflow completed successfully" (success) + cancelled -> "Workflow was cancelled" (warning) + error -> "Errors occurred, check log file" (error) + +This helper is split out so the dispatch is unit-testable without booting +Streamlit and without pulling in pyopenms. +""" + +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from src.workflow._log_status import classify_log_outcome + + +def test_finished_marker_returns_finished(): + assert classify_log_outcome( + "STARTING WORKFLOW\n\nstep 1\n\nWORKFLOW FINISHED\n\n" + ) == "finished" + + +def test_cancelled_marker_returns_cancelled(): + assert classify_log_outcome( + "STARTING WORKFLOW\n\nstep 1\n\nWORKFLOW CANCELLED\n\n" + ) == "cancelled" + + +def test_cancelled_takes_precedence_over_partial_error(): + """ + A TOPP subprocess often dies as the worker is being torn down, leaving + an ERROR line followed by the cancellation marker. The user-meaningful + state is 'cancelled', not 'error'. + """ + assert classify_log_outcome( + "STARTING WORKFLOW\n\nERROR: subprocess died\n\nWORKFLOW CANCELLED\n\n" + ) == "cancelled" + + +def test_truncated_log_returns_error(): + assert classify_log_outcome("STARTING WORKFLOW\n\nstep 1\n\n") == "error" + + +def test_empty_log_returns_error(): + assert classify_log_outcome("") == "error" diff --git a/tests/test_queue_manager_cancel.py b/tests/test_queue_manager_cancel.py new file mode 100644 index 0000000..0f87708 --- /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'." + ) diff --git a/tests/test_workflow_manager_stop.py b/tests/test_workflow_manager_stop.py new file mode 100644 index 0000000..8ebcbd4 --- /dev/null +++ b/tests/test_workflow_manager_stop.py @@ -0,0 +1,162 @@ +""" +Tests for WorkflowManager.stop_workflow / get_workflow_status interaction. + +Bug being fixed (follow-up to PR #383): the RQ worker actually terminates when +the user clicks "Stop Workflow", but the Streamlit UI still shows +"workflow is running" and pressing Stop a second time produces an +"error has occurred" message instead of "workflow has been cancelled". + +Root causes: + 1. stop_workflow clears .job_id on success, so the next get_workflow_status + poll falls through to the local-mode pid_dir fallback. The killed worker + left stale child PID files in pid_dir, so the fallback wrongly returns + running=True. + 2. The worker never wrote 'WORKFLOW FINISHED' to the log because it was + killed mid-execution. The UI's static-display branch only knows two + outcomes (FINISHED -> success, else -> error), so a cancelled run is + misreported as an error. + +These tests pin both behaviours. +""" + +import os +import sys +import types + +import pytest + +fakeredis = pytest.importorskip("fakeredis") +rq = pytest.importorskip("rq") +streamlit = pytest.importorskip("streamlit") +pyopenms = pytest.importorskip("pyopenms") + +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.Logger import Logger +from src.workflow.QueueManager import QueueManager +from src.workflow.WorkflowManager import WorkflowManager + + +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: + job.set_status(JobStatus.STARTED) + job.worker_name = worker_name + job.save() + + +def _make_workflow_manager(tmp_path, monkeypatch) -> WorkflowManager: + """ + Build a minimal WorkflowManager wired to a fakeredis-backed QueueManager. + + Bypasses __init__ (which constructs a StreamlitUI and reads session + state). Only the attributes used by stop_workflow / get_workflow_status + are populated: + - workflow_dir (real tmp dir) + - logger (real Logger; streamlit-free) + - executor (SimpleNamespace exposing pid_dir; CommandExecutor + itself imports streamlit so we cannot instantiate it) + - _queue_manager (fakeredis-backed QueueManager) + + A stale child PID file is dropped in pid_dir to simulate the state the + worker leaves behind when it is force-killed mid-execution. + """ + workflow_dir = tmp_path / "wf" + workflow_dir.mkdir() + + pid_dir = workflow_dir / "pids" + pid_dir.mkdir() + (pid_dir / "12345").touch() + + qm = _make_queue_manager() + job = qm._queue.enqueue(os.getcwd, job_id="wf-job") + _force_started(job) + qm.store_job_id(workflow_dir, "wf-job") + + monkeypatch.setattr( + "rq.command.send_stop_job_command", + lambda *a, **kw: None, + ) + + wm = WorkflowManager.__new__(WorkflowManager) + wm.workflow_dir = workflow_dir + wm.logger = Logger(workflow_dir) + wm.executor = types.SimpleNamespace(pid_dir=pid_dir) + wm._queue_manager = qm + return wm + + +def test_stop_workflow_clears_running_state_in_queue_mode(tmp_path, monkeypatch): + """ + Bug #1: after a successful queue cancel, get_workflow_status must report + running=False. Currently the stale pid_dir keeps the local-mode fallback + returning running=True. + """ + wm = _make_workflow_manager(tmp_path, monkeypatch) + + assert wm.stop_workflow() is True + + status = wm.get_workflow_status() + assert status["running"] is False, ( + "After cancel, get_workflow_status must not report the workflow as " + "still running." + ) + + pid_dir = wm.executor.pid_dir + assert not (pid_dir.exists() and any(pid_dir.iterdir())), ( + "stop_workflow must clean up the stale pid_dir left behind by the " + "killed worker; otherwise the local-mode fallback in " + "get_workflow_status flips running back to True." + ) + + +def test_stop_workflow_writes_cancellation_marker_to_log(tmp_path, monkeypatch): + """ + Bug #2: the static log-display branch needs a way to tell 'cancelled' + apart from 'crashed'. stop_workflow must drop a 'WORKFLOW CANCELLED' + marker into the log so the UI can render the right message. + """ + wm = _make_workflow_manager(tmp_path, monkeypatch) + wm.logger.log("STARTING WORKFLOW") # mimic a partial run + + assert wm.stop_workflow() is True + + logs_dir = wm.workflow_dir / "logs" + for log_name in ("minimal.log", "commands-and-run-times.log", "all.log"): + content = (logs_dir / log_name).read_text(encoding="utf-8") + assert "WORKFLOW CANCELLED" in content, ( + f"{log_name} should contain the WORKFLOW CANCELLED marker." + ) + assert "WORKFLOW FINISHED" not in content, ( + f"{log_name} must not claim the workflow finished." + ) + + +def test_stop_workflow_is_idempotent(tmp_path, monkeypatch): + """ + Pressing Stop a second time (or stop firing twice on Streamlit rerun) + must not raise and must keep running=False. The first call's user intent + has already been satisfied; subsequent calls should be safe no-ops. + """ + wm = _make_workflow_manager(tmp_path, monkeypatch) + + assert wm.stop_workflow() is True + + # Second call: must not raise, get_workflow_status must remain not-running. + wm.stop_workflow() + assert wm.get_workflow_status()["running"] is False + +