moved exclusive start to its own module

Author: sanderegg

packages/service-library/src/servicelib/background_task.py

@@ -3,17 +3,14 @@
 import datetime
 import functools
 import logging
-import socket
 from collections.abc import AsyncIterator, Awaitable, Callable, Coroutine
 from typing import Any, Final, ParamSpec, TypeVar
 
-import arrow
 from tenacity import TryAgain, before_sleep_log, retry, retry_if_exception_type
 from tenacity.wait import wait_fixed
 
 from .async_utils import cancel_wait_task, with_delay
 from .logging_utils import log_context
-from .redis import RedisClientSDK, exclusive
 
 _logger = logging.getLogger(__name__)
 
@@ -140,41 +137,3 @@ async def periodic_task(
             # NOTE: this stopping is shielded to prevent the cancellation to propagate
             # into the stopping procedure
             await asyncio.shield(cancel_wait_task(asyncio_task, max_delay=stop_timeout))
-
-
-def exclusive_periodic(
-    client: RedisClientSDK,
-    *,
-    task_interval: datetime.timedelta,
-    retry_after: datetime.timedelta,
-) -> Callable[
-    [Callable[P, Coroutine[Any, Any, None]]], Callable[P, Coroutine[Any, Any, None]]
-]:
-    """decorates a function to become exclusive and periodic.
-
-    Arguments:
-        client -- The Redis client
-        task_interval -- the task periodicity
-        retry_after -- in case the exclusive lock cannot be acquired or is lost, this is the retry interval
-
-    Returns:
-        Nothing, a periodic method does not return anything as it runs forever.
-    """
-
-    def _decorator(
-        func: Callable[P, Coroutine[Any, Any, None]],
-    ) -> Callable[P, Coroutine[Any, Any, None]]:
-        @periodic(interval=retry_after)
-        @exclusive(
-            client,
-            lock_key=f"lock:exclusive_periodic_task:{func.__name__}",
-            lock_value=f"locked since {arrow.utcnow().format()} by {client.client_name} on {socket.gethostname()}",
-        )
-        @periodic(interval=task_interval)
-        @functools.wraps(func)
-        async def _wrapper(*args: P.args, **kwargs: P.kwargs) -> None:
-            return await func(*args, **kwargs)
-
-        return _wrapper
-
-    return _decorator

packages/service-library/src/servicelib/background_task_utils.py

@@ -0,0 +1,47 @@
+import datetime
+import functools
+from collections.abc import Callable, Coroutine
+from typing import Any, ParamSpec, TypeVar
+
+from .background_task import periodic
+from .redis import RedisClientSDK, exclusive
+
+P = ParamSpec("P")
+R = TypeVar("R")
+
+
+def exclusive_periodic(
+    redis_client: RedisClientSDK | Callable[..., RedisClientSDK],
+    *,
+    task_interval: datetime.timedelta,
+    retry_after: datetime.timedelta = datetime.timedelta(seconds=1),
+) -> Callable[
+    [Callable[P, Coroutine[Any, Any, None]]], Callable[P, Coroutine[Any, Any, None]]
+]:
+    """decorates a function to become exclusive and periodic.
+
+    Arguments:
+        client -- The Redis client
+        task_interval -- the task periodicity
+        retry_after -- in case the exclusive lock cannot be acquired or is lost, this is the retry interval
+
+    Returns:
+        Nothing, a periodic method does not return anything as it runs forever.
+    """
+
+    def _decorator(
+        func: Callable[P, Coroutine[Any, Any, None]],
+    ) -> Callable[P, Coroutine[Any, Any, None]]:
+        @periodic(interval=retry_after)
+        @exclusive(
+            redis_client,
+            lock_key=f"lock:exclusive_periodic_task:{func.__name__}",
+        )
+        @periodic(interval=task_interval)
+        @functools.wraps(func)
+        async def _wrapper(*args: P.args, **kwargs: P.kwargs) -> None:
+            return await func(*args, **kwargs)
+
+        return _wrapper
+
+    return _decorator

packages/service-library/src/servicelib/redis/__init__.py

@@ -1,7 +1,6 @@
 from ._client import RedisClientSDK
 from ._clients_manager import RedisClientsManager
 from ._decorators import exclusive
-from ._distributed_locks_utils import create_exclusive_periodic_task
 from ._errors import (
     CouldNotAcquireLockError,
     CouldNotConnectToRedisError,
@@ -19,7 +18,6 @@
     "RedisClientSDK",
     "RedisClientsManager",
     "RedisManagerDBConfig",
-    "create_exclusive_periodic_task",
 )
 
 # nopycln: file

packages/service-library/src/servicelib/redis/_decorators.py

@@ -2,10 +2,12 @@
 import contextlib
 import functools
 import logging
+import socket
 from collections.abc import Callable, Coroutine
 from datetime import timedelta
 from typing import Any, Final, ParamSpec, TypeVar
 
+import arrow
 import redis.exceptions
 from redis.asyncio.lock import Lock
 
@@ -77,6 +79,9 @@ async def _wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
                 else redis_client
             )
             assert isinstance(client, RedisClientSDK)  # nosec
+            nonlocal lock_value
+            if lock_value is None:
+                lock_value = f"locked since {arrow.utcnow().format()} by {client.client_name} on {socket.gethostname()}"
 
             lock = client.create_lock(redis_lock_key, ttl=DEFAULT_LOCK_TTL)
             if not await lock.acquire(

packages/service-library/src/servicelib/redis/_distributed_locks_utils.py

@@ -17,6 +17,14 @@
 from servicelib.async_utils import cancel_wait_task
 from servicelib.background_task import create_periodic_task, periodic, periodic_task
 
+pytest_simcore_core_services_selection = [
+    "redis",
+]
+pytest_simcore_ops_services_selection = [
+    "redis-commander",
+]
+
+
 _FAST_POLL_INTERVAL: Final[int] = 1
 _VERY_SLOW_POLL_INTERVAL: Final[int] = 100
 

packages/service-library/tests/test_background_task.py

@@ -1,19 +1,22 @@
-# pylint:disable=unused-variable
-# pylint:disable=unused-argument
-# pylint:disable=redefined-outer-name
-# pylint:disable=protected-access
+# pylint: disable=no-value-for-parameter
+# pylint: disable=redefined-outer-name
+# pylint: disable=unused-argument
+# pylint: disable=unused-variable
 
 
 import asyncio
-from datetime import timedelta
+import datetime
+from collections.abc import AsyncIterator, Callable
+from contextlib import AbstractAsyncContextManager
 from itertools import chain
-from unittest.mock import Mock
+from unittest import mock
 
 import arrow
+import pytest
 from servicelib.async_utils import cancel_wait_task
-from servicelib.redis._client import RedisClientSDK
-from servicelib.redis._distributed_locks_utils import create_exclusive_periodic_task
-from servicelib.utils import logged_gather
+from servicelib.background_task_utils import exclusive_periodic
+from servicelib.redis import RedisClientSDK
+from settings_library.redis import RedisDatabase
 from tenacity import (
     AsyncRetrying,
     retry_if_exception_type,
@@ -29,14 +32,17 @@
 ]
 
 
-async def _sleep_task(sleep_interval: float, on_sleep_events: Mock) -> None:
-    on_sleep_events(arrow.utcnow())
-    await asyncio.sleep(sleep_interval)
-    print("Slept for", sleep_interval)
-    on_sleep_events(arrow.utcnow())
+@pytest.fixture
+async def redis_client_sdk(
+    get_redis_client_sdk: Callable[
+        [RedisDatabase], AbstractAsyncContextManager[RedisClientSDK]
+    ],
+) -> AsyncIterator[RedisClientSDK]:
+    async with get_redis_client_sdk(RedisDatabase.RESOURCES) as client:
+        yield client
 
 
-async def _assert_on_sleep_done(on_sleep_events: Mock, *, stop_after: float):
+async def _assert_on_sleep_done(on_sleep_events: mock.Mock, *, stop_after: float):
     async for attempt in AsyncRetrying(
         wait=wait_fixed(0.1),
         stop=stop_after_delay(stop_after),
@@ -52,20 +58,20 @@ async def _assert_task_completes_once(
     redis_client_sdk: RedisClientSDK,
     stop_after: float,
 ) -> tuple[float, ...]:
-    sleep_events = Mock()
-
-    started_task = create_exclusive_periodic_task(
-        redis_client_sdk,
-        _sleep_task,
-        task_period=timedelta(seconds=1),
-        task_name="pytest_sleep_task",
-        sleep_interval=1,
-        on_sleep_events=sleep_events,
-    )
+    @exclusive_periodic(redis_client_sdk, task_interval=datetime.timedelta(seconds=1))
+    async def _sleep_task(sleep_interval: float, on_sleep_events: mock.Mock) -> None:
+        on_sleep_events(arrow.utcnow())
+        await asyncio.sleep(sleep_interval)
+        print("Slept for", sleep_interval)
+        on_sleep_events(arrow.utcnow())
+
+    sleep_events = mock.Mock()
+
+    task = asyncio.create_task(_sleep_task(1, sleep_events), name="pytest_sleep_task")
 
     await _assert_on_sleep_done(sleep_events, stop_after=stop_after)
 
-    await cancel_wait_task(started_task, max_delay=5)
+    await cancel_wait_task(task, max_delay=5)
 
     events_timestamps: tuple[float, ...] = tuple(
         x.args[0].timestamp() for x in sleep_events.call_args_list
@@ -86,33 +92,33 @@ def test__check_elements_lower():
     assert not _check_elements_lower([1, 2, 4, 3, 5])
 
 
-async def test_create_exclusive_periodic_task_single(
+async def test_exclusive_periodic_decorator_single(
     redis_client_sdk: RedisClientSDK,
 ):
     await _assert_task_completes_once(redis_client_sdk, stop_after=2)
 
 
-async def test_create_exclusive_periodic_task_parallel_all_finish(
+async def test_exclusive_periodic_decorator_parallel_all_finish(
     redis_client_sdk: RedisClientSDK,
 ):
     parallel_tasks = 10
-    results: list[tuple[float, float]] = await logged_gather(
+    results = await asyncio.gather(
         *[
             _assert_task_completes_once(redis_client_sdk, stop_after=60)
             for _ in range(parallel_tasks)
         ],
-        reraise=False,
+        return_exceptions=True,
     )
 
     # check no error occurred
     assert [isinstance(x, tuple) for x in results].count(True) == parallel_tasks
-    assert [x[0] < x[1] for x in results].count(True) == parallel_tasks
+    assert [isinstance(x, Exception) for x in results].count(True) == 0
+    valid_results = [x for x in results if isinstance(x, tuple)]
+    assert [x[0] < x[1] for x in valid_results].count(True) == parallel_tasks
 
     # sort by start time (task start order is not equal to the task lock acquisition order)
-    sorted_results: list[tuple[float, float]] = sorted(results, key=lambda x: x[0])
-
-    # pylint:disable=unnecessary-comprehension
-    flattened_results: list[float] = [x for x in chain(*sorted_results)]  # noqa: C416
+    sorted_results = sorted(valid_results, key=lambda x: x[0])
+    flattened_results = list(chain(*sorted_results))
 
     # NOTE all entries should be in increasing order;
     # this means that the `_sleep_task` ran sequentially