🐛Refactoring of project locking using decorator (#7044)

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

@@ -5,19 +5,29 @@
     CouldNotAcquireLockError,
     CouldNotConnectToRedisError,
     LockLostError,
+    ProjectLockError,
 )
 from ._models import RedisManagerDBConfig
+from ._project_lock import (
+    get_project_locked_state,
+    is_project_locked,
+    with_project_locked,
+)
 from ._utils import handle_redis_returns_union_types
 
 __all__: tuple[str, ...] = (
     "CouldNotAcquireLockError",
     "CouldNotConnectToRedisError",
     "exclusive",
+    "get_project_locked_state",
     "handle_redis_returns_union_types",
+    "is_project_locked",
     "LockLostError",
+    "ProjectLockError",
     "RedisClientSDK",
     "RedisClientsManager",
     "RedisManagerDBConfig",
+    "with_project_locked",
 )
 
 # nopycln: file

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

@@ -1,3 +1,6 @@
+from typing import TypeAlias
+
+import redis.exceptions
 from common_library.errors_classes import OsparcErrorMixin
 
 
@@ -19,3 +22,6 @@ class LockLostError(BaseRedisError):
         "TIP: check connection to Redis DBs or look for Synchronous "
         "code that might block the auto-extender task. Somehow the distributed lock disappeared!"
     )
+
+
+ProjectLockError: TypeAlias = redis.exceptions.LockError  # NOTE: backwards compatible

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

@@ -0,0 +1,99 @@
+import functools
+from collections.abc import Awaitable, Callable, Coroutine
+from typing import Any, Final, ParamSpec, TypeVar
+
+from models_library.projects import ProjectID
+from models_library.projects_access import Owner
+from models_library.projects_state import ProjectLocked, ProjectStatus
+
+from ._client import RedisClientSDK
+from ._decorators import exclusive
+from ._errors import CouldNotAcquireLockError, ProjectLockError
+
+_PROJECT_REDIS_LOCK_KEY: Final[str] = "project_lock:{}"
+
+
+P = ParamSpec("P")
+R = TypeVar("R")
+
+
+def with_project_locked(
+    redis_client: RedisClientSDK | Callable[..., RedisClientSDK],
+    *,
+    project_uuid: str | ProjectID,
+    status: ProjectStatus,
+    owner: Owner | None,
+    notification_cb: Callable[[], Awaitable[None]] | None,
+) -> Callable[
+    [Callable[P, Coroutine[Any, Any, R]]], Callable[P, Coroutine[Any, Any, R]]
+]:
+    """creates a distributed auto sustained Redis lock for project with project_uuid, keeping its status and owner in the lock data
+
+    Arguments:
+        redis_client -- the client to use to access redis
+        project_uuid -- the project UUID
+        status -- the project status
+        owner -- the owner of the lock (default: {None})
+        notification_cb -- an optional notification callback that will be called AFTER the project is locked and AFTER it was unlocked
+
+    Returns:
+        the decorated function return value
+
+    Raises:
+        raises anything from the decorated function and from the optional notification callback
+    """
+
+    def _decorator(
+        func: Callable[P, Coroutine[Any, Any, R]],
+    ) -> Callable[P, Coroutine[Any, Any, R]]:
+        @functools.wraps(func)
+        async def _wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
+            @exclusive(
+                redis_client,
+                lock_key=_PROJECT_REDIS_LOCK_KEY.format(project_uuid),
+                lock_value=ProjectLocked(
+                    value=True,
+                    owner=owner,
+                    status=status,
+                ).model_dump_json(),
+            )
+            async def _exclusive_func(*args, **kwargs) -> R:
+                if notification_cb is not None:
+                    await notification_cb()
+                return await func(*args, **kwargs)
+
+            try:
+                result = await _exclusive_func(*args, **kwargs)
+                # we are now unlocked
+                if notification_cb is not None:
+                    await notification_cb()
+                return result
+            except CouldNotAcquireLockError as e:
+                raise ProjectLockError from e
+
+        return _wrapper
+
+    return _decorator
+
+
+async def is_project_locked(
+    redis_client: RedisClientSDK, project_uuid: str | ProjectID
+) -> bool:
+    redis_lock = redis_client.create_lock(_PROJECT_REDIS_LOCK_KEY.format(project_uuid))
+    return await redis_lock.locked()
+
+
+async def get_project_locked_state(
+    redis_client: RedisClientSDK, project_uuid: str | ProjectID
+) -> ProjectLocked | None:
+    """
+    Returns:
+        ProjectLocked object if the project project_uuid is locked or None otherwise
+    """
+    if await is_project_locked(redis_client, project_uuid=project_uuid) and (
+        lock_value := await redis_client.redis.get(
+            _PROJECT_REDIS_LOCK_KEY.format(project_uuid)
+        )
+    ):
+        return ProjectLocked.model_validate_json(lock_value)
+    return None

packages/service-library/tests/redis/test_decorators.py

@@ -75,7 +75,7 @@ async def _():
             pass
 
 
-async def test_exclusive_decorator(
+async def test_exclusive_decorator_runs_original_method(
     redis_client_sdk: RedisClientSDK,
     lock_name: str,
     sleep_duration: float,

packages/service-library/tests/redis/test_project_lock.py

@@ -0,0 +1,144 @@
+# pylint: disable=no-value-for-parameter
+# pylint: disable=protected-access
+# pylint: disable=redefined-outer-name
+# pylint: disable=unused-argument
+# pylint: disable=unused-variable
+
+import asyncio
+from typing import cast
+from unittest import mock
+from uuid import UUID
+
+import pytest
+from faker import Faker
+from models_library.projects import ProjectID
+from models_library.projects_access import Owner
+from models_library.projects_state import ProjectLocked, ProjectStatus
+from servicelib.async_utils import cancel_wait_task
+from servicelib.redis import (
+    ProjectLockError,
+    RedisClientSDK,
+    get_project_locked_state,
+    is_project_locked,
+    with_project_locked,
+)
+from servicelib.redis._project_lock import _PROJECT_REDIS_LOCK_KEY
+
+pytest_simcore_core_services_selection = [
+    "redis",
+]
+pytest_simcore_ops_services_selection = [
+    "redis-commander",
+]
+
+
+@pytest.fixture()
+def project_uuid(faker: Faker) -> ProjectID:
+    return cast(UUID, faker.uuid4(cast_to=None))
+
+
+assert "json_schema_extra" in Owner.model_config
+assert isinstance(Owner.model_config["json_schema_extra"], dict)
+assert isinstance(Owner.model_config["json_schema_extra"]["examples"], list)
+
+
+@pytest.fixture(params=Owner.model_config["json_schema_extra"]["examples"])
+def owner(request: pytest.FixtureRequest) -> Owner:
+    return Owner(**request.param)
+
+
+@pytest.fixture
+def mocked_notification_cb() -> mock.AsyncMock:
+    return mock.AsyncMock()
+
+
+@pytest.mark.parametrize(
+    "project_status",
+    [
+        ProjectStatus.CLOSING,
+        ProjectStatus.CLONING,
+        ProjectStatus.EXPORTING,
+        ProjectStatus.OPENING,
+        ProjectStatus.MAINTAINING,
+    ],
+)
+async def test_with_project_locked(
+    redis_client_sdk: RedisClientSDK,
+    project_uuid: ProjectID,
+    owner: Owner,
+    project_status: ProjectStatus,
+    mocked_notification_cb: mock.AsyncMock,
+):
+    @with_project_locked(
+        redis_client_sdk,
+        project_uuid=project_uuid,
+        status=project_status,
+        owner=owner,
+        notification_cb=mocked_notification_cb,
+    )
+    async def _locked_fct() -> None:
+        mocked_notification_cb.assert_called_once()
+        assert await is_project_locked(redis_client_sdk, project_uuid) is True
+        locked_state = await get_project_locked_state(redis_client_sdk, project_uuid)
+        assert locked_state is not None
+        assert locked_state == ProjectLocked(
+            value=True,
+            owner=owner,
+            status=project_status,
+        )
+        # check lock name formatting is correct
+        redis_lock = await redis_client_sdk.redis.get(
+            _PROJECT_REDIS_LOCK_KEY.format(project_uuid)
+        )
+        assert redis_lock
+        assert ProjectLocked.model_validate_json(redis_lock) == ProjectLocked(
+            value=True,
+            owner=owner,
+            status=project_status,
+        )
+
+    mocked_notification_cb.assert_not_called()
+    assert await get_project_locked_state(redis_client_sdk, project_uuid) is None
+    assert await is_project_locked(redis_client_sdk, project_uuid) is False
+    await _locked_fct()
+    assert await is_project_locked(redis_client_sdk, project_uuid) is False
+    assert await get_project_locked_state(redis_client_sdk, project_uuid) is None
+    mocked_notification_cb.assert_called()
+    assert mocked_notification_cb.call_count == 2
+
+
+@pytest.mark.parametrize(
+    "project_status",
+    [
+        ProjectStatus.CLOSING,
+        ProjectStatus.CLONING,
+        ProjectStatus.EXPORTING,
+        ProjectStatus.OPENING,
+        ProjectStatus.MAINTAINING,
+    ],
+)
+async def test_lock_already_locked_project_raises(
+    redis_client_sdk: RedisClientSDK,
+    project_uuid: ProjectID,
+    owner: Owner,
+    project_status: ProjectStatus,
+):
+    started_event = asyncio.Event()
+
+    @with_project_locked(
+        redis_client_sdk,
+        project_uuid=project_uuid,
+        status=project_status,
+        owner=owner,
+        notification_cb=None,
+    )
+    async def _locked_fct() -> None:
+        started_event.set()
+        await asyncio.sleep(10)
+
+    task1 = asyncio.create_task(_locked_fct(), name="pytest_task_1")
+    await started_event.wait()
+    with pytest.raises(ProjectLockError):
+        await _locked_fct()
+
+    await cancel_wait_task(task1)

services/clusters-keeper/src/simcore_service_clusters_keeper/rpc/clusters.py

@@ -7,8 +7,7 @@
 from models_library.users import UserID
 from models_library.wallets import WalletID
 from servicelib.rabbitmq import RPCRouter
-from servicelib.redis._client import RedisClientSDK
-from servicelib.redis._decorators import exclusive
+from servicelib.redis import RedisClientSDK, exclusive
 
 from ..core.settings import get_application_settings
 from ..modules import clusters