Don't sync after read-only operations

src/daq_queuing_service/task_queue/queue.py

@@ -70,20 +70,14 @@ class QueueContents(TypedDict):
 class Modifying(asyncio.Condition):
     def __init__(
         self,
-        on_enter: Callable[[], None],
+        lock: asyncio.Lock,
         on_exit: Callable[[], None],
         on_error: Callable[[], None],
     ):
-        super().__init__()
-        self._on_enter = on_enter
+        super().__init__(lock=lock)
         self._on_exit = on_exit
         self._on_error = on_error
 
-    async def __aenter__(self):
-        result = await super().__aenter__()
-        self._on_enter()
-        return result
-
     async def __aexit__(
         self,
         exc_type: type[BaseException] | None,
@@ -129,8 +123,9 @@ def __init__(self, converter: Converter, broadcaster: Broadcaster[QUEUE_EVENTS])
         )
         self._converter = converter
         self._broadcaster = broadcaster
+        self._lock = asyncio.Lock()
         self._modifying = Modifying(
-            on_enter=self._save_contents,
+            lock=self._lock,
             on_exit=self._sync,
             on_error=self._restore_latest_good_contents,
         )
@@ -217,6 +212,7 @@ def _sync(self):
         if not self._call_queue:
             self._pause_queue(PauseReason.EMPTY_QUEUE)
 
+        self._save_contents()
         self._broadcast_changes()
         self._modifying.notify_all()
 
@@ -231,17 +227,17 @@ def _copy_contents(self) -> QueueContents:
             }
         )
 
-    def _save_contents(self) -> None:
+    def _save_contents(self):
         self._last_good_contents = self._copy_contents()
 
-    def _restore_from_contents(self, contents: QueueContents) -> None:
+    def _restore_from_contents(self, contents: QueueContents):
         self._tasks = TaskRegistry(contents["tasks"])
         self._queue = contents["queue"]
         self._history = contents["history"]
         self._call_queue = contents["call_queue"]
         self._call_history = contents["call_history"]
 
-    def _restore_latest_good_contents(self) -> None:
+    def _restore_latest_good_contents(self):
         self._restore_from_contents(self._last_good_contents)
 
     def _broadcast_changes(self):
@@ -350,7 +346,7 @@ async def get_task_by_id(self, task_id: str) -> TaskWithPosition:
             TaskNotFoundError: Raised if the no task exists with the requested task ID.
         """
         # Returns copy so don't have to be worried about caller modifying task.
-        async with self._modifying:
+        async with self._lock:
             return self._get_task_by_id(task_id)
 
     def _get_task_by_id(self, task_id: str) -> TaskWithPosition:
@@ -369,7 +365,7 @@ async def get_task_by_position(self, position: int) -> TaskWithPosition | None:
             if no task exists at the requested position.
         """
         # Returns copy so don't have to be worried about caller modifying task.
-        async with self._modifying:
+        async with self._lock:
             if position < -self.length or position >= self.length:
                 return None
             return self._get_task_by_id(self._queue[position])
@@ -382,7 +378,7 @@ async def get_queue(self) -> list[TaskWithPosition]:
             will be run in.
         """
         # Returns copies so don't have to be worried about caller modifying tasks.
-        async with self._modifying:
+        async with self._lock:
             return self._get_queue()
 
     async def get_history(self) -> list[TaskWithPosition]:
@@ -393,7 +389,7 @@ async def get_history(self) -> list[TaskWithPosition]:
             chronological order.
         """
         # Returns copies so don't have to be worried about caller modifying tasks.
-        async with self._modifying:
+        async with self._lock:
             return self._get_history()
 
     async def get_tasks(self) -> list[TaskWithPosition]:
@@ -404,7 +400,7 @@ async def get_tasks(self) -> list[TaskWithPosition]:
             with the history.
         """
         # Returns copies so don't have to be worried about caller modifying tasks.
-        async with self._modifying:
+        async with self._lock:
             return self._get_history() + self._get_queue()
 
     async def add_tasks(self, tasks: list[Task], position: int | None = None) -> None:
@@ -640,14 +636,14 @@ def _get_history(self) -> list[TaskWithPosition]:
         ]
 
     async def get_call_queue(self) -> list[BlueapiCallResponse]:
-        async with self._modifying:
+        async with self._lock:
             return self._get_call_queue()
 
     def _get_call_queue(self) -> list[BlueapiCallResponse]:
         return [call.to_response() for call in self._call_queue]
 
     async def get_call_history(self) -> list[BlueapiCallResponse]:
-        async with self._modifying:
+        async with self._lock:
             return self._get_call_history()
 
     def _get_call_history(self) -> list[BlueapiCallResponse]:

tests/unit_tests/test_queue.py

@@ -1,5 +1,6 @@
 import asyncio
 import copy
+from typing import Any
 from unittest.mock import MagicMock
 
 import pytest
@@ -1099,14 +1100,23 @@ def test__restore_from_contents_replaces_queue_contents(task_queue: TaskQueue):
     assert task_queue._call_history == new_call_history
 
 
-async def test__last_good_contents_updated_when_modifying_lock_entered(
+async def test__last_good_contents_updated_when_modifying_lock_exited(
     task_queue: TaskQueue,
 ):
-    task_queue._queue = ["should be copied"]
+    task_queue._queue = []
 
     async with task_queue._modifying:
-        task_queue._queue = []
+        task_queue._add_tasks(
+            [
+                Task(
+                    id="should be copied",
+                    experiment=TaskRequest(name="", instrument_session=""),
+                )
+            ],
+            position=0,
+        )
 
+    task_queue._queue = []
     assert task_queue._last_good_contents["queue"] == ["should be copied"]
     assert task_queue._queue == []
 
@@ -1128,7 +1138,7 @@ def convert(
     task_queue._converter.construct_blueapi_calls = convert
 
     with pytest.raises(ConverterError):
-        await task_queue.get_queue()
+        await task_queue.move_task("0", 0)
 
     assert task_queue._queue == ["0", "1", "2", "3", "4"]
     assert list(task_queue._tasks.keys()) == ["0", "1", "2", "3", "4"]
@@ -1149,6 +1159,33 @@ def convert(
     task_queue.__init__(task_queue._converter, task_queue._broadcaster)
 
     with pytest.raises(ConverterError):
-        await task_queue.get_queue()
+        await task_queue.add_tasks(MagicMock())
 
     task_queue._restore_latest_good_contents.assert_called_once()
+
+
+@pytest.mark.parametrize(
+    "method_name, args",
+    [
+        ("get_queue", []),
+        ("get_tasks", []),
+        ("get_history", []),
+        ("get_task_by_id", ["0"]),
+        ("get_task_by_position", [0]),
+        ("get_call_queue", []),
+        ("get_call_history", []),
+    ],
+)
+async def test__sync_not_called_for_read_only_methods(
+    task_queue: TaskQueue, method_name: str, args: list[Any]
+):
+    task_queue._sync = MagicMock()
+
+    # Need to reinitialise so that mocked _sync is injected into Modifying object
+    contents = copy.copy(task_queue._last_good_contents)
+    task_queue.__init__(task_queue._converter, task_queue._broadcaster)
+    task_queue._restore_from_contents(contents)
+
+    await getattr(task_queue, method_name)(*args)
+
+    task_queue._sync.assert_not_called()