extended tests to ensure working as expected

Author: Andrei Neagu

packages/service-library/src/servicelib/resilent_long_running/_client.py

@@ -9,7 +9,6 @@
 from settings_library.redis import RedisSettings
 
 from ._errors import (
-    AlreadyStartedError,
     FinishedWithError,
     NoMoreRetryAttemptsError,
     TimedOutError,
@@ -161,17 +160,8 @@ async def _start_job_if_missing(
     ) -> None:
         # if job is missing on server side start it
         if await self._rpc_interface.get_status(unique_id) == JobStatus.NOT_FOUND:
-            try:
-                await self._rpc_interface.start(
-                    name, unique_id, timeout=timeout, **params
-                )
-                await self._store_interface.update_entry_expiry(
-                    unique_id, expire=timeout
-                )
-            except AlreadyStartedError:
-                _logger.info(
-                    "unique_id='%s', was already running, did not start", unique_id
-                )
+            await self._rpc_interface.start(name, unique_id, timeout=timeout, **params)
+            await self._store_interface.update_entry_expiry(unique_id, expire=timeout)
 
     async def _format_remaining_attempts(
         self, unique_id: JobUniqueId, retry_count: NonNegativeInt
@@ -263,14 +253,15 @@ async def ensure_result(
             timeout -- maximum time to wait for the result
 
         Keyword Arguments:
-            is_unique -- only one instance of this task can exsist at any given time (default: {False})
+            is_unique -- only one instance of the remote handler can exsist at any given time (default: {False})
             retry_count -- how many times to retry execution before giving up (default: {3})
 
         Raises:
             UnexpectedResultTypeError: wrong result type
             TimedOutError: did not finish in time
+            AlreadyStartedError: a handler marked as unique with the same params is already running
             NoMoreRetryAttemptsError: no more retry attempts left and still raised an error
-            FinishedWithError: task on remove finised with an error
+            FinishedWithError: handler on remove finised with an error
 
         Returns:
             the resturn value of the handler invoked remotley

packages/service-library/src/servicelib/resilent_long_running/_rpc/server.py

@@ -48,7 +48,11 @@ async def setup(self) -> None:
             self.result,
         ):
             router.expose(
-                reraise_if_error_type=(JobNotFoundError, NoResultIsAvailableError)
+                reraise_if_error_type=(
+                    JobNotFoundError,
+                    AlreadyStartedError,
+                    NoResultIsAvailableError,
+                )
             )(handler)
 
         await self._rabbitmq_rpc_server.register_router(router, self._rpc_namespace)

packages/service-library/tests/resilient_long_running/test_workflow.py

@@ -18,11 +18,13 @@
     Server,
     TimedOutError,
 )
+from servicelib.resilent_long_running._client import AlreadyStartedError
 from servicelib.resilent_long_running._models import JobUniqueId
 from servicelib.resilent_long_running.runners.asyncio_tasks import (
     AsyncioTasksJobInterface,
     AsyncTaskRegistry,
 )
+from servicelib.utils import limited_gather
 from settings_library.rabbit import RabbitSettings
 from settings_library.redis import RedisSettings
 from tenacity import (
@@ -66,6 +68,10 @@ async def some_f() -> None:
     async def echo_f(data: Any) -> Any:
         return data
 
+    @registry.expose()
+    async def inc_number_f(number: int) -> int:
+        return number + 1
+
     @registry.expose()
     async def raising_f() -> None:
         msg = "I always raise an error"
@@ -134,6 +140,25 @@ async def test_workflow(
     assert type(result) is expected_type
 
 
+@pytest.mark.parametrize("is_unique", [True, False])
+async def test_workflow_paralle_calls_to_same_handler(
+    server: Server, client: Client, is_unique: bool
+):
+    async def _to_run(number: int) -> None:
+        result = await client.ensure_result(
+            "inc_number_f",
+            expected_type=int,
+            timeout=timedelta(seconds=30),
+            number=number,
+            is_unique=is_unique,
+        )
+        assert result == number + 1
+
+    count = 100
+    await limited_gather(*(_to_run(number) for number in range(count)), limit=count)
+    await _assert_tasks_count(server, count=0)
+
+
 @pytest.mark.parametrize("is_unique", [True, False])
 async def test_timeout_error(server: Server, client: Client, is_unique: bool):
     with pytest.raises(TimedOutError):
@@ -319,18 +344,36 @@ async def main():
         process.kill()
 
 
-@pytest.mark.parametrize("is_unique", [False])
+async def test_start_unique_task_twice_is_not_allowed(server: Server, client: Client):
+    async def _runner() -> None:
+        await client.ensure_result(
+            "sleep_for_f",
+            expected_type=type(None),
+            timeout=timedelta(seconds=30),
+            duration=2,
+            is_unique=True,
+        )
+
+    with pytest.raises(AlreadyStartedError):
+        await limited_gather(*(_runner() for _ in range(2)), limit=2)
+
+
 async def test_cancellation_of_client_can_resume_process(
     server: Server,
     client_process: Callable[[Callable[[Client], Awaitable[None]]], Process],
     client: Client,
-    is_unique: bool,
 ):
     await _assert_tasks_count(server, count=0)
 
+    # NOTE: only when the task is maked as unique can the client pickup
+    # the running handler running on remote once it's restarted
+
+    # NOTE: you cannot start the same unique task twice
+    # a task is unique when all of it's parameters are the same and is maked as unique
+
     async def _runner(client_: Client) -> None:
         await _sleep_for_ensure_result(
-            client_, retry_count=3, timeout=timedelta(minutes=1), is_unique=is_unique
+            client_, retry_count=3, timeout=timedelta(minutes=1), is_unique=True
         )
 
     # start task in process