🐛Introduce timeout in project lock (a la aioredlock) (ITISFoundation#3675)

Author: sanderegg

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

@@ -2,7 +2,7 @@
 import contextlib
 import datetime
 import logging
-from typing import Awaitable, Callable, Optional
+from typing import AsyncIterator, Awaitable, Callable, Optional
 
 from servicelib.logging_utils import log_catch, log_context
 from tenacity import TryAgain
@@ -64,3 +64,22 @@ async def stop_periodic_task(
         asyncio_task.cancel()
         with log_catch(logger, reraise=False):
             await asyncio.wait((asyncio_task,), timeout=timeout)
+
+
+@contextlib.asynccontextmanager
+async def periodic_task(
+    task: Callable[..., Awaitable[None]],
+    *,
+    interval: datetime.timedelta,
+    task_name: str,
+    **kwargs,
+) -> AsyncIterator[asyncio.Task]:
+    asyncio_task = None
+    try:
+        asyncio_task = await start_periodic_task(
+            task, interval=interval, task_name=task_name, **kwargs
+        )
+        yield asyncio_task
+    finally:
+        if asyncio_task is not None:
+            await stop_periodic_task(asyncio_task)

packages/service-library/tests/test_background_task.py

@@ -13,7 +13,11 @@
 from faker import Faker
 from pytest import FixtureRequest
 from pytest_mock.plugin import MockerFixture
-from servicelib.background_task import start_periodic_task, stop_periodic_task
+from servicelib.background_task import (
+    periodic_task,
+    start_periodic_task,
+    stop_periodic_task,
+)
 
 _FAST_POLL_INTERVAL = 1
 
@@ -107,3 +111,20 @@ async def test_dynamic_scaling_task_correctly_cancels(
     await asyncio.sleep(5 * task_interval.total_seconds())
     # the task will be called once, and then stop
     mock_background_task.assert_called_once()
+
+
+async def test_periodic_task_context_manager(
+    mock_background_task: mock.AsyncMock,
+    task_interval: datetime.timedelta,
+    faker: Faker,
+):
+    task_name = faker.pystr()
+    async with periodic_task(
+        mock_background_task, interval=task_interval, task_name=task_name
+    ) as asyncio_task:
+        assert asyncio_task.get_name() == task_name
+        assert asyncio_task.cancelled() is False
+        await asyncio.sleep(5 * task_interval.total_seconds())
+        assert asyncio_task.cancelled() is False
+        assert asyncio_task.done() is False
+    assert asyncio_task.cancelled() is True

services/web/server/src/simcore_service_webserver/projects/project_lock.py

@@ -1,22 +1,30 @@
+import datetime
 from asyncio.log import logger
 from contextlib import asynccontextmanager
-from typing import Optional, Union
+from typing import Final, Optional, Union
 
 import redis
 from aiohttp import web
 from models_library.projects import ProjectID
 from models_library.projects_state import Owner, ProjectLocked, ProjectStatus
 from redis.asyncio.lock import Lock
+from servicelib.background_task import periodic_task
+from servicelib.logging_utils import log_catch
 
 from ..redis import get_redis_lock_manager_client
 from ..users_api import UserNameDict
 from .projects_exceptions import ProjectLockError
 
 PROJECT_REDIS_LOCK_KEY: str = "project_lock:{}"
-
+PROJECT_LOCK_TIMEOUT: Final[datetime.timedelta] = datetime.timedelta(seconds=10)
 ProjectLock = Lock
 
 
+async def _auto_extend_project_lock(project_lock: Lock) -> None:
+    with log_catch(logger, reraise=False):
+        await project_lock.reacquire()
+
+
 @asynccontextmanager
 async def lock_project(
     app: web.Application,
@@ -38,7 +46,8 @@ async def lock_project(
 
     """
     redis_lock = get_redis_lock_manager_client(app).lock(
-        PROJECT_REDIS_LOCK_KEY.format(project_uuid)
+        PROJECT_REDIS_LOCK_KEY.format(project_uuid),
+        timeout=PROJECT_LOCK_TIMEOUT.total_seconds(),
     )
     try:
         if not await redis_lock.acquire(
@@ -52,7 +61,12 @@ async def lock_project(
             raise ProjectLockError(
                 f"Lock for project {project_uuid!r} user {user_id!r} could not be acquired"
             )
-        yield
+        async with periodic_task(
+            _auto_extend_project_lock,
+            interval=0.6 * PROJECT_LOCK_TIMEOUT,
+            task_name=f"{PROJECT_REDIS_LOCK_KEY.format(project_uuid)}_lock_auto_extend",
+        ):
+            yield
     finally:
         # let's ensure we release that stuff
         try:

services/web/server/tests/unit/with_dbs/03/test_redis.py

@@ -3,39 +3,71 @@
 # pylint:disable=redefined-outer-name
 
 import asyncio
+import datetime
 
 import pytest
 import redis
 import redis.asyncio as aioredis
+from faker import Faker
+from redis.asyncio.lock import Lock
+from servicelib.background_task import periodic_task
 
 
-async def test_aioredis(redis_client: aioredis.Redis):
-    await redis_client.set("my-key", "value")
-    val = await redis_client.get("my-key")
-    assert val == "value"
+async def test_aioredis(redis_client: aioredis.Redis, faker: Faker):
+    key = faker.pystr()
+    value = faker.pystr()
+    await redis_client.set(key, value)
+    val = await redis_client.get(key)
+    assert val == value
 
 
-async def test_redlocks_features(redis_client: aioredis.Redis):
-    # Check wether a resourece acquired by any other redlock instance:
-    lock = redis_client.lock("resource_name")
+async def test_lock_acquisition(redis_client: aioredis.Redis, faker: Faker):
+    lock_name = faker.pystr()
+    lock = redis_client.lock(lock_name)
     assert not await lock.locked()
 
     # Try to acquire the lock:
     lock_acquired = await lock.acquire(blocking=False)
-    assert lock_acquired, "Lock not acquired"
+    assert lock_acquired is True
+    assert await lock.locked() is True
+    assert await lock.owned() is True
+    with pytest.raises(redis.exceptions.LockError):
+        # a lock with no timeout cannot be reacquired
+        await lock.reacquire()
+    with pytest.raises(redis.exceptions.LockError):
+        # a lock with no timeout cannot be extended
+        await lock.extend(2)
+
+    # try to acquire the lock a second time
+    same_lock = redis_client.lock(lock_name)
+    assert await same_lock.locked() is True
+    assert await same_lock.owned() is False
+    assert await same_lock.acquire(blocking=False) is False
+
+    # now release the lock
     await lock.release()
     assert not await lock.locked()
     assert not await lock.owned()
 
-    # use as context manager
+
+async def test_lock_context_manager(redis_client: aioredis.Redis, faker: Faker):
+    lock_name = faker.pystr()
+    lock = redis_client.lock(lock_name)
+    assert not await lock.locked()
+
     async with lock:
         assert await lock.locked()
         assert await lock.owned()
-        # a lock with no timeout cannot be extended
         with pytest.raises(redis.exceptions.LockError):
+            # a lock with no timeout cannot be reacquired
+            await lock.reacquire()
+
+        with pytest.raises(redis.exceptions.LockError):
+            # a lock with no timeout cannot be extended
             await lock.extend(2)
+
         # try to acquire the lock a second time
-        same_lock = redis_client.lock("resource_name", blocking_timeout=1)
+        same_lock = redis_client.lock(lock_name, blocking_timeout=1)
         assert await same_lock.locked()
         assert not await same_lock.owned()
         assert await same_lock.acquire() == False
@@ -44,13 +76,46 @@ async def test_redlocks_features(redis_client: aioredis.Redis):
                 ...
     assert not await lock.locked()
 
-    # now create a lock with a ttl
-    ttl_lock = redis_client.lock("ttl_resource", timeout=2, blocking_timeout=1)
+
+@pytest.fixture
+def lock_timeout() -> datetime.timedelta:
+    return datetime.timedelta(seconds=2)
+
+
+async def test_lock_with_ttl(
+    redis_client: aioredis.Redis, faker: Faker, lock_timeout: datetime.timedelta
+):
+    ttl_lock = redis_client.lock(faker.pystr(), timeout=lock_timeout.total_seconds())
     assert not await ttl_lock.locked()
+
     with pytest.raises(redis.exceptions.LockNotOwnedError):
         # this raises as the lock is lost
         async with ttl_lock:
             assert await ttl_lock.locked()
             assert await ttl_lock.owned()
-            await asyncio.sleep(3)
+            await asyncio.sleep(2 * lock_timeout.total_seconds())
             assert not await ttl_lock.locked()
+
+
+async def test_lock_with_auto_extent(
+    redis_client: aioredis.Redis, faker: Faker, lock_timeout: datetime.timedelta
+):
+    ttl_lock = redis_client.lock(faker.pystr(), timeout=lock_timeout.total_seconds())
+    assert not await ttl_lock.locked()
+
+    async def _auto_extend_lock(lock: Lock) -> None:
+        assert await lock.reacquire() is True
+
+    async with ttl_lock, periodic_task(
+        _auto_extend_lock,
+        interval=0.6 * lock_timeout,
+        task_name=f"{ttl_lock.name}_auto_extend",
+        lock=ttl_lock,
+    ):
+        assert await ttl_lock.locked() is True
+        assert await ttl_lock.owned() is True
+        await asyncio.sleep(5 * lock_timeout.total_seconds())
+        assert await ttl_lock.locked() is True
+        assert await ttl_lock.owned() is True
+    assert await ttl_lock.locked() is False
+    assert await ttl_lock.owned() is False