From fe92541167e9142df0e6b3e0a71bd31443f45f54 Mon Sep 17 00:00:00 2001 From: Zoheb Shaikh <26975142+ZohebShaikh@users.noreply.github.com> Date: Fri, 13 Nov 2026 23:43:38 +0000 Subject: [PATCH 1/2] check if I get 409 --- tests/system_tests/test_blueapi_system.py | 34 +++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/tests/system_tests/test_blueapi_system.py b/tests/system_tests/test_blueapi_system.py index f393c7ab8..2d7f8a548 100644 --- a/tests/system_tests/test_blueapi_system.py +++ b/tests/system_tests/test_blueapi_system.py @@ -3,9 +3,11 @@ import time from asyncio import Queue from collections.abc import Generator +from concurrent.futures import ThreadPoolExecutor from enum import StrEnum, auto from itertools import count from pathlib import Path +from threading import Barrier from unittest.mock import MagicMock, patch import pytest @@ -442,6 +444,38 @@ def test_put_worker_task_fails_if_not_idle( rest_client.clear_task(long_task_id) +def _wait_for_idle(rest_client: BlueapiRestClient, timeout: float = 10.0) -> None: + deadline = time.monotonic() + timeout + while rest_client.get_state() is not WorkerState.IDLE: + assert time.monotonic() < deadline, "Worker did not return to IDLE" + time.sleep(0.1) + + +@pytest.mark.parametrize("attempt", range(5)) +def test_concurrent_put_worker_task_returns_409_not_500( + rest_client: BlueapiRestClient, long_task: TaskRequest, attempt: int +): + """Two clients starting tasks at the same moment must get one success and one + 409 Conflict, never a 500 from the worker rejecting the second begin_task.""" + task_ids = [rest_client.create_task(long_task).task_id for _ in range(2)] + barrier = Barrier(len(task_ids)) + + def start(task_id: str) -> int | None: + barrier.wait() + try: + rest_client.update_worker_task(WorkerTask(task_id=task_id)) + return None + except BlueskyRemoteControlError as e: + return e.args[0] + + with ThreadPoolExecutor(len(task_ids)) as pool: + results = list(pool.map(start, task_ids)) + + rest_client.cancel_current_task(WorkerState.ABORTING) + _wait_for_idle(rest_client) + assert sorted(results, key=str) == [409, None] + + def test_get_worker_state(client: BlueapiClient): assert client.state == WorkerState.IDLE From c341e79fbf42168f2037731c1cf26b90689d675f Mon Sep 17 00:00:00 2001 From: Zoheb Shaikh <26975142+ZohebShaikh@users.noreply.github.com> Date: Fri, 13 Nov 2026 23:44:27 +0000 Subject: [PATCH 2/2] fix: blueapi conflicts on submit_task --- src/blueapi/service/main.py | 16 +++++++++++----- tests/unit_tests/service/test_rest_api.py | 17 +++++++++++++++++ 2 files changed, 28 insertions(+), 5 deletions(-) diff --git a/src/blueapi/service/main.py b/src/blueapi/service/main.py index bbdadfa97..fcaf4a512 100644 --- a/src/blueapi/service/main.py +++ b/src/blueapi/service/main.py @@ -434,11 +434,17 @@ def set_active_task( raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="Worker already active" ) - runner.run( - interface.begin_task, - task=task, - pass_through_headers=get_passthrough_headers(request), - ) + try: + runner.run( + interface.begin_task, + task=task, + pass_through_headers=get_passthrough_headers(request), + ) + except WorkerBusyError as e: + # Another request started a task between the check above and this call + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, detail="Worker already active" + ) from e return task diff --git a/tests/unit_tests/service/test_rest_api.py b/tests/unit_tests/service/test_rest_api.py index 4315eecf6..358017ca3 100644 --- a/tests/unit_tests/service/test_rest_api.py +++ b/tests/unit_tests/service/test_rest_api.py @@ -47,6 +47,7 @@ from blueapi.worker.event import TaskStatus, WorkerEvent, WorkerState from blueapi.worker.task import InvalidTaskParameter, Task, TaskParameterError from blueapi.worker.task_worker import TrackableTask +from blueapi.worker.worker_errors import WorkerBusyError class MockCountModel(BaseModel): ... @@ -553,6 +554,22 @@ def test_cant_delete_other_users_task( mock_runner.run.assert_called_once() +def test_put_plan_returns_409_if_worker_becomes_busy( + mock_runner: Mock, client: TestClient +) -> None: + def run(function, *args, **kwargs): + if function is interface.begin_task: + raise WorkerBusyError("Worker is in state RUNNING") + return None + + mock_runner.run.side_effect = run + + resp = client.put("/worker/task", json={"task_id": str(uuid.uuid4())}) + + assert resp.status_code == status.HTTP_409_CONFLICT + assert resp.json() == {"detail": "Worker already active"} + + def test_set_active_task(client: TestClient) -> None: task_id = str(uuid.uuid4()) task = WorkerTask(task_id=task_id)