🐛 Ensure proper Redis client shutdown in Celery

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

@@ -2,13 +2,9 @@
 from typing import Any
 
 from celery import Celery  # type: ignore[import-untyped]
-from servicelib.redis import RedisClientSDK
 from settings_library.celery import CelerySettings
 from settings_library.redis import RedisDatabase
 
-from .backends._redis import RedisTaskInfoStore
-from .task_manager import CeleryTaskManager
-
 
 def _celery_configure(celery_settings: CelerySettings) -> dict[str, Any]:
     base_config = {
@@ -36,22 +32,3 @@ def create_app(settings: CelerySettings) -> Celery:
         ),
         **_celery_configure(settings),
     )
-
-
-async def create_task_manager(
-    app: Celery, settings: CelerySettings
-) -> CeleryTaskManager:
-    redis_client_sdk = RedisClientSDK(
-        settings.CELERY_REDIS_RESULT_BACKEND.build_redis_dsn(
-            RedisDatabase.CELERY_TASKS
-        ),
-        client_name="celery_tasks",
-    )
-    await redis_client_sdk.setup()
-    # GCR please address https://github.com/ITISFoundation/osparc-simcore/issues/8159
-
-    return CeleryTaskManager(
-        app,
-        settings,
-        RedisTaskInfoStore(redis_client_sdk),
-    )

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

@@ -6,17 +6,14 @@
 from celery.worker.worker import WorkController  # type: ignore[import-untyped]
 from servicelib.celery.app_server import BaseAppServer
 from servicelib.logging_utils import log_context
-from settings_library.celery import CelerySettings
 
-from .common import create_task_manager
 from .utils import get_app_server, set_app_server
 
 _logger = logging.getLogger(__name__)
 
 
 def on_worker_init(
     app_server: BaseAppServer,
-    celery_settings: CelerySettings,
     sender: WorkController,
     **_kwargs,
 ) -> None:
@@ -26,21 +23,14 @@ def _init(startup_complete_event: threading.Event) -> None:
         loop = asyncio.new_event_loop()
         asyncio.set_event_loop(loop)
 
-        async def _setup_task_manager():
-            assert sender.app  # nosec
-            assert isinstance(sender.app, Celery)  # nosec
+        assert sender.app  # nosec
+        assert isinstance(sender.app, Celery)  # nosec
 
-            app_server.task_manager = await create_task_manager(
-                sender.app,
-                celery_settings,
-            )
-
-            set_app_server(sender.app, app_server)
+        set_app_server(sender.app, app_server)
 
         app_server.event_loop = loop
 
-        loop.run_until_complete(_setup_task_manager())
-        loop.run_until_complete(app_server.lifespan(startup_complete_event))
+        loop.run_until_complete(app_server.start_and_hold(startup_complete_event))
 
     thread = threading.Thread(
         group=None,

packages/celery-library/tests/conftest.py

@@ -2,25 +2,30 @@
 # pylint: disable=unused-argument
 
 import datetime
+import logging
 import threading
 from collections.abc import AsyncIterator, Callable
-from functools import partial
 from typing import Any
 
 import pytest
 from celery import Celery  # type: ignore[import-untyped]
-from celery.contrib.testing.worker import TestWorkController, start_worker
+from celery.contrib.testing.worker import (
+    TestWorkController,
+    start_worker,
+)
 from celery.signals import worker_init, worker_shutdown
 from celery.worker.worker import WorkController
-from celery_library.common import create_task_manager
+from celery_library.backends._redis import RedisTaskInfoStore
 from celery_library.signals import on_worker_init, on_worker_shutdown
 from celery_library.task_manager import CeleryTaskManager
 from celery_library.types import register_celery_types
 from pytest_simcore.helpers.monkeypatch_envs import setenvs_from_dict
 from pytest_simcore.helpers.typing_env import EnvVarsDict
 from servicelib.celery.app_server import BaseAppServer
+from servicelib.celery.task_manager import TaskManager
+from servicelib.redis import RedisClientSDK
 from settings_library.celery import CelerySettings
-from settings_library.redis import RedisSettings
+from settings_library.redis import RedisDatabase, RedisSettings
 
 pytest_plugins = [
     "pytest_simcore.docker_compose",
@@ -33,11 +38,40 @@
 ]
 
 
+_logger = logging.getLogger(__name__)
+
+
 class FakeAppServer(BaseAppServer):
-    async def lifespan(self, startup_completed_event: threading.Event) -> None:
+    def __init__(self, app: Celery, settings: CelerySettings):
+        super().__init__(app)
+        self._settings = settings
+        self._task_manager: CeleryTaskManager | None = None
+
+    @property
+    def task_manager(self) -> TaskManager:
+        assert self._task_manager, "Task manager is not initialized"
+        return self._task_manager
+
+    async def start_and_hold(self, startup_completed_event: threading.Event) -> None:
+        redis_client_sdk = RedisClientSDK(
+            self._settings.CELERY_REDIS_RESULT_BACKEND.build_redis_dsn(
+                RedisDatabase.CELERY_TASKS
+            ),
+            client_name="pytest_celery_tasks",
+        )
+        await redis_client_sdk.setup()
+
+        self._task_manager = CeleryTaskManager(
+            self._app,
+            self._settings,
+            RedisTaskInfoStore(redis_client_sdk),
+        )
+
         startup_completed_event.set()
         await self.shutdown_event.wait()  # wait for shutdown
 
+        await redis_client_sdk.shutdown()
+
 
 @pytest.fixture
 def register_celery_tasks() -> Callable[[Celery], None]:
@@ -51,17 +85,12 @@ def _(celery_app: Celery) -> None: ...
 @pytest.fixture
 def app_environment(
     monkeypatch: pytest.MonkeyPatch,
-    redis_service: RedisSettings,
     env_devel_dict: EnvVarsDict,
 ) -> EnvVarsDict:
     return setenvs_from_dict(
         monkeypatch,
         {
             **env_devel_dict,
-            "REDIS_SECURE": redis_service.REDIS_SECURE,
-            "REDIS_HOST": redis_service.REDIS_HOST,
-            "REDIS_PORT": f"{redis_service.REDIS_PORT}",
-            "REDIS_PASSWORD": redis_service.REDIS_PASSWORD.get_secret_value(),
         },
     )
 
@@ -74,8 +103,8 @@ def celery_settings(
 
 
 @pytest.fixture
-def app_server() -> BaseAppServer:
-    return FakeAppServer(app=None)
+def app_server(celery_app: Celery, celery_settings: CelerySettings) -> BaseAppServer:
+    return FakeAppServer(app=celery_app, settings=celery_settings)
 
 
 @pytest.fixture(scope="session")
@@ -98,11 +127,10 @@ def celery_config() -> dict[str, Any]:
 async def with_celery_worker(
     celery_app: Celery,
     app_server: BaseAppServer,
-    celery_settings: CelerySettings,
     register_celery_tasks: Callable[[Celery], None],
 ) -> AsyncIterator[TestWorkController]:
     def _on_worker_init_wrapper(sender: WorkController, **_kwargs):
-        return partial(on_worker_init, app_server, celery_settings)(sender, **_kwargs)
+        return on_worker_init(app_server, sender, **_kwargs)
 
     worker_init.connect(_on_worker_init_wrapper)
     worker_shutdown.connect(on_worker_shutdown)
@@ -111,24 +139,39 @@ def _on_worker_init_wrapper(sender: WorkController, **_kwargs):
 
     with start_worker(
         celery_app,
-        pool="threads",
         concurrency=1,
+        pool="threads",
         loglevel="info",
         perform_ping_check=False,
         queues="default",
     ) as worker:
         yield worker
 
 
+@pytest.fixture
+async def mock_celery_app(celery_config: dict[str, Any]) -> Celery:
+    return Celery(**celery_config)
+
+
 @pytest.fixture
 async def celery_task_manager(
-    celery_app: Celery,
+    mock_celery_app: Celery,
     celery_settings: CelerySettings,
-    with_celery_worker: TestWorkController,
-) -> CeleryTaskManager:
+    use_in_memory_redis: RedisSettings,
+) -> AsyncIterator[CeleryTaskManager]:
     register_celery_types()
 
-    return await create_task_manager(
-        celery_app,
-        celery_settings,
-    )
+    try:
+        redis_client_sdk = RedisClientSDK(
+            use_in_memory_redis.build_redis_dsn(RedisDatabase.CELERY_TASKS),
+            client_name="pytest_celery_tasks",
+        )
+        await redis_client_sdk.setup()
+
+        yield CeleryTaskManager(
+            mock_celery_app,
+            celery_settings,
+            RedisTaskInfoStore(redis_client_sdk),
+        )
+    finally:
+        await redis_client_sdk.shutdown()

packages/celery-library/tests/unit/test_async_jobs.py

@@ -301,6 +301,8 @@ async def test_async_jobs_cancel(
         payload=60 * 10,  # test hangs if not cancelled properly
     )
 
+    await asyncio.sleep(3.0)    # wait a bit before cancelling
+
     await async_jobs.cancel(
         async_jobs_rabbitmq_rpc_client,
         rpc_namespace=ASYNC_JOBS_RPC_NAMESPACE,

packages/celery-library/tests/unit/test_tasks.py

@@ -13,6 +13,7 @@
 import pytest
 from celery import Celery, Task
 from celery.contrib.abortable import AbortableTask
+from celery.worker.worker import WorkController
 from celery_library.errors import TransferrableCeleryError
 from celery_library.task import register_task
 from celery_library.task_manager import CeleryTaskManager
@@ -92,6 +93,7 @@ def _(celery_app: Celery) -> None:
 
 async def test_submitting_task_calling_async_function_results_with_success_state(
     celery_task_manager: CeleryTaskManager,
+    with_celery_worker: WorkController,
 ):
     task_filter = TaskFilter(user_id=42)
 
@@ -122,6 +124,7 @@ async def test_submitting_task_calling_async_function_results_with_success_state
 
 async def test_submitting_task_with_failure_results_with_error(
     celery_task_manager: CeleryTaskManager,
+    with_celery_worker: WorkController,
 ):
     task_filter = TaskFilter(user_id=42)
 
@@ -150,6 +153,7 @@ async def test_submitting_task_with_failure_results_with_error(
 
 async def test_cancelling_a_running_task_aborts_and_deletes(
     celery_task_manager: CeleryTaskManager,
+    with_celery_worker: WorkController,
 ):
     task_filter = TaskFilter(user_id=42)
 
@@ -182,6 +186,7 @@ async def test_cancelling_a_running_task_aborts_and_deletes(
 
 async def test_listing_task_uuids_contains_submitted_task(
     celery_task_manager: CeleryTaskManager,
+    with_celery_worker: WorkController,
 ):
     task_filter = TaskFilter(user_id=42)
 
@@ -201,5 +206,5 @@ async def test_listing_task_uuids_contains_submitted_task(
             tasks = await celery_task_manager.list_tasks(task_filter)
             assert any(task.uuid == task_uuid for task in tasks)
 
-        tasks = await celery_task_manager.list_tasks(task_filter)
-        assert any(task.uuid == task_uuid for task in tasks)
+    tasks = await celery_task_manager.list_tasks(task_filter)
+    assert any(task.uuid == task_uuid for task in tasks)

packages/service-library/src/servicelib/celery/app_server.py

@@ -31,15 +31,12 @@ def shutdown_event(self) -> asyncio.Event:
         return self._shutdown_event
 
     @property
+    @abstractmethod
     def task_manager(self) -> TaskManager:
-        return self._task_manager
-
-    @task_manager.setter
-    def task_manager(self, manager: TaskManager) -> None:
-        self._task_manager = manager
+        raise NotImplementedError
 
     @abstractmethod
-    async def lifespan(
+    async def start_and_hold(
         self,
         startup_completed_event: threading.Event,
     ) -> None:

packages/service-library/src/servicelib/celery/models.py

@@ -58,7 +58,7 @@ async def get_task_metadata(self, task_id: TaskID) -> TaskMetadata | None: ...
 
     async def get_task_progress(self, task_id: TaskID) -> ProgressReport | None: ...
 
-    async def list_tasks(self, task_context: TaskFilter) -> list[Task]: ...
+    async def list_tasks(self, task_filter: TaskFilter) -> list[Task]: ...
 
     async def remove_task(self, task_id: TaskID) -> None: ...
 

packages/service-library/src/servicelib/celery/task_manager.py

@@ -1,4 +1,4 @@
-from typing import Any, Protocol
+from typing import Any, Protocol, runtime_checkable
 
 from models_library.progress_bar import ProgressReport
 
@@ -12,6 +12,7 @@
 )
 
 
+@runtime_checkable
 class TaskManager(Protocol):
     async def submit_task(
         self, task_metadata: TaskMetadata, *, task_filter: TaskFilter, **task_param

packages/service-library/src/servicelib/fastapi/celery/app_server.py

@@ -7,6 +7,7 @@
 from fastapi import FastAPI
 
 from ...celery.app_server import BaseAppServer
+from ...celery.task_manager import TaskManager
 
 _SHUTDOWN_TIMEOUT: Final[float] = datetime.timedelta(seconds=10).total_seconds()
 
@@ -18,7 +19,14 @@ def __init__(self, app: FastAPI):
         super().__init__(app)
         self._lifespan_manager: LifespanManager | None = None
 
-    async def lifespan(self, startup_completed_event: threading.Event) -> None:
+    @property
+    def task_manager(self) -> TaskManager:
+        task_manager = self.app.state.task_manager
+        assert task_manager, "Task manager is not initialized"  # nosec
+        assert isinstance(task_manager, TaskManager)
+        return task_manager
+
+    async def start_and_hold(self, startup_completed_event: threading.Event) -> None:
         async with LifespanManager(
             self.app,
             startup_timeout=None,  # waits for full app initialization (DB migrations, etc.)

services/storage/src/simcore_service_storage/core/application.py

@@ -36,7 +36,7 @@
 from ..dsm import setup_dsm
 from ..dsm_cleaner import setup_dsm_cleaner
 from ..exceptions.handlers import set_exception_handlers
-from ..modules.celery import setup_task_manager
+from ..modules.celery import setup_celery
 from ..modules.db import setup_db
 from ..modules.long_running_tasks import setup_rest_api_long_running_tasks_for_uploads
 from ..modules.rabbitmq import setup as setup_rabbitmq
@@ -71,12 +71,13 @@ def create_app(settings: ApplicationSettings) -> FastAPI:  # noqa: C901
     setup_s3(app)
     setup_client_session(app, tracing_settings=settings.STORAGE_TRACING)
 
-    if settings.STORAGE_CELERY and not settings.STORAGE_WORKER_MODE:
-        setup_rabbitmq(app)
+    if settings.STORAGE_CELERY:
+        setup_celery(app, settings=settings.STORAGE_CELERY)
 
-        setup_task_manager(app, celery_settings=settings.STORAGE_CELERY)
+        if not settings.STORAGE_WORKER_MODE:
+            setup_rabbitmq(app)
+            setup_rpc_routes(app)
 
-        setup_rpc_routes(app)
     setup_rest_api_long_running_tasks_for_uploads(app)
     setup_rest_api_routes(app, API_VTAG)
     set_exception_handlers(app)