fixed redis client sdk being stuck

Author: Andrei Neagu

packages/celery-library/src/celery_library/common.py

@@ -47,6 +47,8 @@ async def create_task_manager(
         ),
         client_name="celery_tasks",
     )
+    await redis_client_sdk.setup()
+    #
 
     return CeleryTaskManager(
         app,

packages/service-library/src/servicelib/fastapi/redis_lifespan.py

@@ -51,6 +51,7 @@ async def redis_client_sdk_lifespan(_: FastAPI, state: State) -> AsyncIterator[S
                 redis_dsn_with_secrets,
                 client_name=redis_state.REDIS_CLIENT_NAME,
             )
+            await redis_client.setup()
 
         try:
             yield {"REDIS_CLIENT_SDK": redis_client, **called_state}

packages/service-library/src/servicelib/long_running_tasks/_store/base.py

@@ -36,5 +36,5 @@ async def setup(self) -> None:
         """Setup the store, if needed."""
 
     @abstractmethod
-    async def teardown(self) -> None:
-        """Teardown the store, if needed."""
+    async def shutdown(self) -> None:
+        """Shutdown the store, if needed."""

packages/service-library/src/servicelib/long_running_tasks/_store/in_memory.py

@@ -12,7 +12,7 @@ def __init__(self, *args, **kwargs):
     async def setup(self) -> None:
         pass
 
-    async def teardown(self) -> None:
+    async def shutdown(self) -> None:
         pass
 
     async def get_task_data(self, task_id: TaskId) -> TaskData | None:

packages/service-library/src/servicelib/long_running_tasks/_store/redis.py

@@ -25,8 +25,9 @@ async def setup(self) -> None:
             self.redis_settings.build_redis_dsn(RedisDatabase.LONG_RUNNING_TASKS),
             client_name=f"long_running_tasks_store_{self.namespace}",
         )
+        await self._client.setup()
 
-    async def teardown(self) -> None:
+    async def shutdown(self) -> None:
         if self._client:
             await self._client.shutdown()
 

packages/service-library/src/servicelib/long_running_tasks/task.py

@@ -132,6 +132,7 @@ async def setup(self) -> None:
             self.redis_settings.build_redis_dsn(RedisDatabase.LOCKS),
             client_name=f"long_running_tasks_store_{self.namespace}_lock",
         )
+        await self.redis_client_sdk.setup()
 
         self._stale_tasks_monitor_task = create_periodic_task(
             task=exclusive(
@@ -148,7 +149,6 @@ async def setup(self) -> None:
         )
 
     async def teardown(self) -> None:
-
         for tracked_task in await self._tasks_data.list_tasks_data():
             # when closing we do not care about pending errors
             await self.remove_task(
@@ -167,11 +167,11 @@ async def teardown(self) -> None:
                     self._cancelled_tasks_removal_task, max_delay=_CANCEL_TASK_TIMEOUT
                 )
 
-        await self._tasks_data.teardown()
-
-        if self.redis_client_sdk:
+        if self.redis_client_sdk is not None:
             await self.redis_client_sdk.shutdown()
 
+        await self._tasks_data.shutdown()
+
     async def _stale_tasks_monitor_worker(self) -> None:
         """
         A task is considered stale, if the task status is not queried

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

@@ -63,6 +63,7 @@ def __post_init__(self) -> None:
         self._is_healthy = False
         self._health_check_task_started_event = asyncio.Event()
 
+    async def setup(self) -> None:
         @periodic(interval=self.health_check_interval)
         async def _periodic_check_health() -> None:
             assert self._health_check_task_started_event  # nosec
@@ -74,6 +75,9 @@ async def _periodic_check_health() -> None:
             name=f"redis_service_health_check_{self.redis_dsn}__{uuid4()}",
         )
 
+        assert self._health_check_task_started_event  # nosec
+        await self._health_check_task_started_event.wait()
+
         _logger.info(
             "Connection to %s succeeded with %s",
             f"redis at {self.redis_dsn=}",

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

@@ -27,6 +27,7 @@ async def setup(self) -> None:
                 health_check_interval=config.health_check_interval,
                 client_name=f"{self.client_name}",
             )
+            await self._client_sdks[config.database].setup()
 
     async def shutdown(self) -> None:
         await asyncio.gather(

packages/service-library/tests/conftest.py

@@ -86,6 +86,7 @@ async def _(
         client = RedisClientSDK(
             redis_resources_dns, decode_responses=decode_response, client_name="pytest"
         )
+        await client.setup()
         assert client
         assert client.redis_dsn == redis_resources_dns
         assert client.client_name == "pytest"

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

@@ -95,6 +95,7 @@ def __init__(
         )
 
     async def setup(self) -> None:
+        await self._redis_client.setup()
         await self._manager.setup()