removed socket timeout on redis clients as this is a dangerous setting

Author: sanderegg

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__)
@@ -65,7 +64,7 @@ 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,

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

@@ -7,8 +7,10 @@
     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:"
 

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

@@ -31,7 +31,6 @@
 from ._constants import (
     DEFAULT_EXPECTED_LOCK_OVERALL_TIME,
     DEFAULT_SEMAPHORE_TTL,
-    DEFAULT_SOCKET_TIMEOUT,
     SEMAPHORE_KEY_PREFIX,
 )
 from ._errors import (
@@ -97,7 +96,7 @@ class DistributedSemaphore(BaseModel):
     blocking_timeout: Annotated[
         datetime.timedelta | None,
         Field(description="Maximum time to wait when blocking"),
-    ] = DEFAULT_SOCKET_TIMEOUT
+    ] = None
     instance_id: Annotated[
         str,
         Field(
@@ -208,7 +207,7 @@ async def _acquire_forever_on_socket_timeout() -> list[str] | None:
             return tokens_key_token
 
         try:
-            # NOTE: redis-py library timeouts when the defined socket timeout triggers (e.g. DEFAULT_SOCKET_TIMEOUT)
+            # NOTE: redis-py library timeouts when the defined socket timeout triggers
             # The BRPOP command itself could timeout but the redis-py socket timeout defeats the purpose
             # so we always block forever on BRPOP, tenacity takes care of retrying when a socket timeout happens
             # and we use asyncio.timeout to enforce the blocking_timeout if defined
@@ -424,7 +423,7 @@ async def distributed_semaphore(  # noqa: C901
     capacity: PositiveInt,
     ttl: datetime.timedelta = DEFAULT_SEMAPHORE_TTL,
     blocking: bool = True,
-    blocking_timeout: datetime.timedelta | None = DEFAULT_SOCKET_TIMEOUT,
+    blocking_timeout: datetime.timedelta | None = None,
     expected_lock_overall_time: datetime.timedelta = DEFAULT_EXPECTED_LOCK_OVERALL_TIME,
 ) -> AsyncIterator[DistributedSemaphore]:
     """

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

@@ -9,7 +9,6 @@
 from ._constants import (
     DEFAULT_EXPECTED_LOCK_OVERALL_TIME,
     DEFAULT_SEMAPHORE_TTL,
-    DEFAULT_SOCKET_TIMEOUT,
 )
 from ._semaphore import distributed_semaphore
 
@@ -27,7 +26,7 @@ def with_limited_concurrency(
     capacity: int | Callable[..., int],
     ttl: datetime.timedelta = DEFAULT_SEMAPHORE_TTL,
     blocking: bool = True,
-    blocking_timeout: datetime.timedelta | None = DEFAULT_SOCKET_TIMEOUT,
+    blocking_timeout: datetime.timedelta | None = None,
     expected_lock_overall_time: datetime.timedelta = DEFAULT_EXPECTED_LOCK_OVERALL_TIME,
 ) -> Callable[
     [Callable[P, Coroutine[Any, Any, R]]], Callable[P, Coroutine[Any, Any, R]]
@@ -107,7 +106,7 @@ def with_limited_concurrency_cm(
     capacity: int | Callable[..., int],
     ttl: datetime.timedelta = DEFAULT_SEMAPHORE_TTL,
     blocking: bool = True,
-    blocking_timeout: datetime.timedelta | None = DEFAULT_SOCKET_TIMEOUT,
+    blocking_timeout: datetime.timedelta | None = None,
     expected_lock_overall_time: datetime.timedelta = DEFAULT_EXPECTED_LOCK_OVERALL_TIME,
 ) -> Callable[
     [Callable[P, AbstractAsyncContextManager[R]]],

packages/service-library/tests/conftest.py

@@ -112,7 +112,7 @@ async def _cleanup_redis_data(clients_manager: RedisClientsManager) -> None:
 
 @pytest.fixture
 async def get_redis_client_sdk(
-    mock_redis_socket_timeout: None, use_in_memory_redis: RedisSettings
+    use_in_memory_redis: RedisSettings,
 ) -> AsyncIterable[
     Callable[[RedisDatabase], AbstractAsyncContextManager[RedisClientSDK]]
 ]:
@@ -122,7 +122,7 @@ async def get_redis_client_sdk(
 
 @pytest.fixture
 async def get_in_process_redis_client_sdk(
-    mock_redis_socket_timeout: None, redis_service: RedisSettings
+    redis_service: RedisSettings,
 ) -> AsyncIterable[
     Callable[[RedisDatabase], AbstractAsyncContextManager[RedisClientSDK]]
 ]:

packages/service-library/tests/deferred_tasks/test_deferred_tasks.py

@@ -3,7 +3,6 @@
 
 import asyncio
 import contextlib
-import datetime
 import itertools
 import json
 import random
@@ -19,7 +18,6 @@
 from common_library.json_serialization import json_dumps
 from common_library.serialization import model_dump_with_secrets
 from pydantic import NonNegativeFloat, NonNegativeInt
-from pytest_mock import MockerFixture
 from servicelib.rabbitmq import RabbitMQClient
 from servicelib.redis import RedisClientSDK
 from servicelib.sequences_utils import partition_gen
@@ -385,19 +383,10 @@ async def pause_redis(self) -> AsyncIterator[None]:
             yield
 
 
-@pytest.fixture
-def mock_default_socket_timeout(mocker: MockerFixture) -> None:
-    mocker.patch(
-        "servicelib.redis._client.DEFAULT_SOCKET_TIMEOUT",
-        datetime.timedelta(seconds=0.25),
-    )
-
-
 @pytest.mark.parametrize("max_workers", [10])
 @pytest.mark.parametrize("deferred_tasks_to_start", [100])
 @pytest.mark.parametrize("service", ["rabbit", "redis"])
 async def test_workflow_with_third_party_services_outages(
-    mock_default_socket_timeout: None,
     paused_container: Callable[[str], AbstractAsyncContextManager[None]],
     redis_client_sdk_deferred_tasks: RedisClientSDK,
     rabbit_client: RabbitMQClient,

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

@@ -104,9 +104,7 @@ async def test_redis_lock_with_ttl(
             assert not await ttl_lock.locked()
 
 
-async def test_redis_client_sdk_setup_shutdown(
-    mock_redis_socket_timeout: None, redis_service: RedisSettings
-):
+async def test_redis_client_sdk_setup_shutdown(redis_service: RedisSettings):
     # setup
     redis_resources_dns = redis_service.build_redis_dsn(RedisDatabase.RESOURCES)
     client = RedisClientSDK(redis_resources_dns, client_name="pytest")
@@ -130,7 +128,6 @@ async def test_redis_client_sdk_setup_shutdown(
 
 
 async def test_regression_fails_on_redis_service_outage(
-    mock_redis_socket_timeout: None,
     paused_container: Callable[[str], AbstractAsyncContextManager[None]],
     redis_client_sdk: RedisClientSDK,
 ):

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

@@ -16,7 +16,6 @@
 
 
 async def test_redis_client_sdks_manager(
-    mock_redis_socket_timeout: None,
     redis_service: RedisSettings,
 ):
     all_redis_configs: set[RedisManagerDBConfig] = {

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

@@ -354,19 +354,7 @@ async def test_semaphore_multiple_instances_capacity_limit(
     await semaphores[2].release()
 
 
-@pytest.fixture
-def with_slow_redis_socket_timeout(
-    mock_redis_socket_timeout: None, mocker: MockerFixture
-) -> None:
-    # put back to higher value to allow normal operations
-    mocker.patch(
-        "servicelib.redis._client.DEFAULT_SOCKET_TIMEOUT",
-        datetime.timedelta(seconds=30),
-    )
-
-
 async def test_semaphore_with_timeout(
-    with_slow_redis_socket_timeout,
     redis_client_sdk: RedisClientSDK,
     semaphore_name: str,
 ):