fix: begin task conflict should send 409

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
 
 

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
 

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)