Remove parameter caching inside task class (#370)

Closes #369

The validated and processed parameters from a plan request were being
cached inside the `Task` class because they were needed in two separate
threads. Unfortunately, with the introduction of the subprocess (#343),
pickling was causing issues with the cached object (see #369 for more
details).

This PR is the simplest solution: remove the cache and generate the
parameters twice. It also adds regression tests for the bug case in #369

src/blueapi/worker/reworker.py

@@ -129,7 +129,7 @@ def begin_task(self, task_id: str) -> None:
             raise KeyError(f"No pending task with ID {task_id}")
 
     def submit_task(self, task: Task) -> str:
-        task.prepare_params(self._ctx)
+        task.prepare_params(self._ctx)  # Will raise if parameters are invalid
         task_id: str = str(uuid.uuid4())
         trackable_task = TrackableTask(task_id=task_id, task=task)
         self._pending_tasks[task_id] = trackable_task

src/blueapi/worker/task.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Any, Mapping, Optional
+from typing import Any, Mapping
 
 from pydantic import BaseModel, Field
 
@@ -18,25 +18,19 @@ class Task(BlueapiBaseModel):
     params: Mapping[str, Any] = Field(
         description="Values for parameters to plan, if any", default_factory=dict
     )
-    _prepared_params: Optional[BaseModel] = None
 
-    def prepare_params(self, ctx: BlueskyContext) -> None:
-        self._ensure_params(ctx)
+    def prepare_params(self, ctx: BlueskyContext) -> BaseModel:
+        return _lookup_params(ctx, self)
 
     def do_task(self, ctx: BlueskyContext) -> None:
         LOGGER.info(f"Asked to run plan {self.name} with {self.params}")
 
         func = ctx.plan_functions[self.name]
-        prepared_params = self._ensure_params(ctx)
+        prepared_params = self.prepare_params(ctx)
         plan_generator = func(**prepared_params.dict())
         wrapped_plan_generator = ctx.wrap(plan_generator)
         ctx.run_engine(wrapped_plan_generator)
 
-    def _ensure_params(self, ctx: BlueskyContext) -> BaseModel:
-        if self._prepared_params is None:
-            self._prepared_params = _lookup_params(ctx, self)
-        return self._prepared_params
-
 
 # Here for backward compatibility pending
 # https://github.com/DiamondLightSource/blueapi/issues/253

tests/service/test_rest_api.py

@@ -12,6 +12,7 @@
 from blueapi.core.bluesky_types import Plan
 from blueapi.service.handler import Handler
 from blueapi.service.main import get_handler, setup_handler, teardown_handler
+from blueapi.service.model import WorkerTask
 from blueapi.worker.task import RunPlan
 from src.blueapi.worker import WorkerState
 
@@ -215,6 +216,31 @@ def test_put_plan_begins_task(handler: Handler, client: TestClient) -> None:
     task_json = {"task_id": task_id}
     client.put("/worker/task", json=task_json)
 
+    resp = client.get("/worker/task")
+    assert resp.status_code == 200
+    active_task = WorkerTask(**resp.json())
+    assert active_task is not None
+    assert active_task.task_id == task_id
+    handler.stop()
+
+
+def test_worker_task_is_none_on_startup(handler: Handler, client: TestClient) -> None:
+    handler.start()
+    resp = client.get("/worker/task")
+    assert resp.status_code == 200
+    active_task = WorkerTask(**resp.json())
+    assert active_task.task_id is None
+    handler.stop()
+
+
+def test_get_worker_task(handler: Handler, client: TestClient) -> None:
+    handler.start()
+    response = client.post("/tasks", json=_TASK.dict())
+    task_id = response.json()["task_id"]
+
+    task_json = {"task_id": task_id}
+    client.put("/worker/task", json=task_json)
+
     active_task = handler.active_task
     assert active_task is not None
     assert active_task.task_id == task_id