🎨Computational backend: stability improvements step8 🚨

packages/celery-library/Makefile

@@ -27,6 +27,7 @@ tests: ## runs unit tests
 		--durations=10 \
 		--exitfirst \
 		--failed-first \
+		--keep-docker-up \
 		--pdb \
 		-vv \
 		$(CURDIR)/tests
@@ -41,6 +42,7 @@ tests-ci: ## runs unit tests
 		--cov-report=term-missing \
 		--cov-report=xml \
 		--junitxml=junit.xml -o junit_family=legacy \
+		--keep-docker-up \
 		--cov=celery_library \
 		--durations=10 \
 		--log-date-format="%Y-%m-%d %H:%M:%S" \

packages/pytest-simcore/src/pytest_simcore/redis_service.py

@@ -4,7 +4,6 @@
 
 import logging
 from collections.abc import AsyncIterator
-from datetime import timedelta
 
 import pytest
 import tenacity
@@ -116,14 +115,6 @@ async def wait_till_redis_responsive(redis_url: URL | str) -> None:
         await client.aclose(close_connection_pool=True)
 
 
-@pytest.fixture
-def mock_redis_socket_timeout(mocker: MockerFixture) -> None:
-    # lowered to allow CI to properly shutdown RedisClientSDK instances
-    mocker.patch(
-        "servicelib.redis._client.DEFAULT_SOCKET_TIMEOUT", timedelta(seconds=0.25)
-    )
-
-
 @pytest.fixture
 async def use_in_memory_redis(mocker: MockerFixture) -> RedisSettings:
     mocker.patch("redis.asyncio.from_url", FakeAsyncRedis)

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

@@ -20,7 +20,6 @@
     DEFAULT_DECODE_RESPONSES,
     DEFAULT_HEALTH_CHECK_INTERVAL,
     DEFAULT_LOCK_TTL,
-    DEFAULT_SOCKET_TIMEOUT,
 )
 
 _logger = logging.getLogger(__name__)
@@ -49,6 +48,7 @@ class RedisClientSDK:
     _client: aioredis.Redis = field(init=False)
     _task_health_check: Task | None = None
     _started_event_task_health_check: asyncio.Event | None = None
+    _cancelled_event_task_health_check: asyncio.Event | None = None
     _is_healthy: bool = False
 
     @property
@@ -65,20 +65,24 @@ def __post_init__(self) -> None:
                 redis.exceptions.ConnectionError,
             ],
             retry_on_timeout=True,
-            socket_timeout=DEFAULT_SOCKET_TIMEOUT.total_seconds(),
+            socket_timeout=None,  # NOTE: setting a timeout here can lead to issues with long running commands
             encoding="utf-8",
             decode_responses=self.decode_responses,
             client_name=self.client_name,
         )
         self._is_healthy = False
         self._started_event_task_health_check = asyncio.Event()
+        self._cancelled_event_task_health_check = asyncio.Event()
 
     async def setup(self) -> None:
         @periodic(interval=self.health_check_interval)
         async def _periodic_check_health() -> None:
             assert self._started_event_task_health_check  # nosec
+            assert self._cancelled_event_task_health_check  # nosec
             self._started_event_task_health_check.set()
             self._is_healthy = await self.ping()
+            if self._cancelled_event_task_health_check.is_set():
+                raise asyncio.CancelledError
 
         self._task_health_check = asyncio.create_task(
             _periodic_check_health(),
@@ -100,10 +104,9 @@ async def shutdown(self) -> None:
             if self._task_health_check:
                 assert self._started_event_task_health_check  # nosec
                 await self._started_event_task_health_check.wait()
-
-                await cancel_wait_task(
-                    self._task_health_check, max_delay=_HEALTHCHECK_TIMEOUT_S
-                )
+                assert self._cancelled_event_task_health_check  # nosec
+                self._cancelled_event_task_health_check.set()
+                await cancel_wait_task(self._task_health_check, max_delay=None)
 
             await self._client.aclose(close_connection_pool=True)
 

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

@@ -7,11 +7,12 @@
     seconds=30
 )
 DEFAULT_LOCK_TTL: Final[datetime.timedelta] = datetime.timedelta(seconds=10)
-DEFAULT_SOCKET_TIMEOUT: Final[datetime.timedelta] = datetime.timedelta(seconds=30)
 
+DEFAULT_SEMAPHORE_BLOCK_TIMEOUT: Final[datetime.timedelta] = datetime.timedelta(
+    seconds=30
+)
 DEFAULT_SEMAPHORE_TTL: Final[datetime.timedelta] = datetime.timedelta(seconds=10)
 SEMAPHORE_KEY_PREFIX: Final[str] = "semaphores:"
-SEMAPHORE_HOLDER_KEY_PREFIX: Final[str] = "semaphores:holders:"
 
 DEFAULT_DECODE_RESPONSES: Final[bool] = True
 DEFAULT_HEALTH_CHECK_INTERVAL: Final[datetime.timedelta] = datetime.timedelta(seconds=5)

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

@@ -26,13 +26,23 @@ class LockLostError(BaseRedisError):
 ProjectLockError: TypeAlias = redis.exceptions.LockError  # NOTE: backwards compatible
 
 
-class SemaphoreAcquisitionError(BaseRedisError):
-    msg_template: str = "Could not acquire semaphore '{name}' (capacity: {capacity})"
+class SemaphoreError(BaseRedisError):
+    msg_template: str = (
+        "Unexpected error with semaphore '{name}' by this instance `{instance_id}`"
+    )
+
+
+class SemaphoreAcquisitionError(SemaphoreError):
+    msg_template: str = (
+        "Could not acquire semaphore '{name}' by this instance `{instance_id}`"
+    )
 
 
-class SemaphoreNotAcquiredError(BaseRedisError):
-    msg_template: str = "Semaphore '{name}' was not acquired by this instance"
+class SemaphoreNotAcquiredError(SemaphoreError):
+    msg_template: str = (
+        "Semaphore '{name}' was not acquired by this instance `{instance_id}`"
+    )
 
 
-class SemaphoreLostError(BaseRedisError):
+class SemaphoreLostError(SemaphoreError):
     msg_template: str = "Semaphore '{name}' was lost by this instance `{instance_id}`"