♻️ add rabbitmq interface to long_running_tasks ⚠️🚨 (#8198)

Co-authored-by: Andrei Neagu <[email protected]>

Author: GitHK

.env-devel

@@ -147,6 +147,7 @@ LICENSES_ITIS_VIP_API_URL=https://replace-with-itis-api/{category}
 LICENSES_ITIS_VIP_CATEGORIES='{"HumanWholeBody": "Humans", "HumanBodyRegion": "Humans (Region)", "AnimalWholeBody": "Animal"}'
 LICENSES_SPEAG_PHANTOMS_API_URL=https://replace-with-speag-api/{category}
 LICENSES_SPEAG_PHANTOMS_CATEGORIES='{"ComputationalPhantom": "Phantom of the Opera"}'
+LONG_RUNNING_TASKS_NAMESPACE_SUFFIX=development
 
 # Can use 'docker run -it itisfoundation/invitations:latest simcore-service-invitations generate-dotenv --auto-password'
 INVITATIONS_DEFAULT_PRODUCT=osparc

.github/workflows/ci-testing-deploy.yml

@@ -1252,7 +1252,7 @@ jobs:
   unit-test-service-library:
     needs: changes
     if: ${{ needs.changes.outputs.service-library == 'true' || github.event_name == 'push' || github.event.inputs.force_all_builds == 'true' }}
-    timeout-minutes: 18 # if this timeout gets too small, then split the tests
+    timeout-minutes: 20 # if this timeout gets too small, then split the tests
     name: "[unit] service-library"
     runs-on: ${{ matrix.os }}
     strategy:

api/specs/web-server/_long_running_tasks.py

@@ -32,44 +32,40 @@
 @router.get(
     "/tasks",
     response_model=Envelope[list[TaskGet]],
-    name="list_tasks",
-    description="Lists all long running tasks",
     responses=_export_data_responses,
 )
-def get_async_jobs(): ...
+def get_async_jobs():
+    """Lists all long running tasks"""
 
 
 @router.get(
     "/tasks/{task_id}",
     response_model=Envelope[TaskStatus],
-    name="get_task_status",
-    description="Retrieves the status of a task",
     responses=_export_data_responses,
 )
 def get_async_job_status(
     _path_params: Annotated[_PathParam, Depends()],
-): ...
+):
+    """Retrieves the status of a task"""
 
 
 @router.delete(
     "/tasks/{task_id}",
-    name="cancel_and_delete_task",
-    description="Cancels and deletes a task",
     responses=_export_data_responses,
     status_code=status.HTTP_204_NO_CONTENT,
 )
 def cancel_async_job(
     _path_params: Annotated[_PathParam, Depends()],
-): ...
+):
+    """Cancels and removes a task"""
 
 
 @router.get(
     "/tasks/{task_id}/result",
     response_model=Any,
-    name="get_task_result",
-    description="Retrieves the result of a task",
     responses=_export_data_responses,
 )
 def get_async_job_result(
     _path_params: Annotated[_PathParam, Depends()],
-): ...
+):
+    """Retrieves the result of a task"""

api/specs/web-server/_long_running_tasks_legacy.py

@@ -42,11 +42,11 @@ async def get_task_status(
 
 @router.delete(
     "/{task_id}",
-    name="cancel_and_delete_task",
-    description="Cancels and deletes a task",
+    name="remove_task",
+    description="Cancels and removes a task",
     status_code=status.HTTP_204_NO_CONTENT,
 )
-async def cancel_and_delete_task(
+async def remove_task(
     _path_params: Annotated[_PathParam, Depends()],
 ): ...
 

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

@@ -0,0 +1,14 @@
+from datetime import timedelta
+
+import pytest
+from pytest_mock import MockerFixture
+
+
+@pytest.fixture
+async def fast_long_running_tasks_cancellation(
+    mocker: MockerFixture,
+) -> None:
+    mocker.patch(
+        "servicelib.long_running_tasks.task._CANCEL_TASKS_CHECK_INTERVAL",
+        new=timedelta(seconds=1),
+    )

packages/service-library/src/servicelib/aiohttp/long_running_tasks/_manager.py

@@ -1,41 +1,12 @@
-import datetime
-
 from aiohttp import web
-from settings_library.redis import RedisSettings
 
 from ...long_running_tasks.base_long_running_manager import BaseLongRunningManager
 from ...long_running_tasks.models import TaskContext
-from ...long_running_tasks.task import RedisNamespace, TasksManager
 from ._constants import APP_LONG_RUNNING_MANAGER_KEY
 from ._request import get_task_context
 
 
 class AiohttpLongRunningManager(BaseLongRunningManager):
-    def __init__(
-        self,
-        app: web.Application,
-        stale_task_check_interval: datetime.timedelta,
-        stale_task_detect_timeout: datetime.timedelta,
-        redis_settings: RedisSettings,
-        redis_namespace: RedisNamespace,
-    ):
-        self._app = app
-        self._tasks_manager = TasksManager(
-            stale_task_check_interval=stale_task_check_interval,
-            stale_task_detect_timeout=stale_task_detect_timeout,
-            redis_settings=redis_settings,
-            redis_namespace=redis_namespace,
-        )
-
-    @property
-    def tasks_manager(self) -> TasksManager:
-        return self._tasks_manager
-
-    async def setup(self) -> None:
-        await self._tasks_manager.setup()
-
-    async def teardown(self) -> None:
-        await self._tasks_manager.teardown()
 
     @staticmethod
     def get_task_context(request: web.Request) -> TaskContext:

packages/service-library/src/servicelib/aiohttp/long_running_tasks/_routes.py

@@ -1,12 +1,16 @@
-from typing import Any
+from typing import Annotated, Any
 
 from aiohttp import web
-from pydantic import BaseModel
+from models_library.rest_base import RequestParameters
+from pydantic import BaseModel, Field
 
 from ...aiohttp import status
 from ...long_running_tasks import lrt_api
 from ...long_running_tasks.models import TaskGet, TaskId
-from ..requests_validation import parse_request_path_parameters_as
+from ..requests_validation import (
+    parse_request_path_parameters_as,
+    parse_request_query_parameters_as,
+)
 from ..rest_responses import create_data_response
 from ._manager import get_long_running_manager
 
@@ -26,10 +30,11 @@ async def list_tasks(request: web.Request) -> web.Response:
                 task_id=t.task_id,
                 status_href=f"{request.app.router['get_task_status'].url_for(task_id=t.task_id)}",
                 result_href=f"{request.app.router['get_task_result'].url_for(task_id=t.task_id)}",
-                abort_href=f"{request.app.router['cancel_and_delete_task'].url_for(task_id=t.task_id)}",
+                abort_href=f"{request.app.router['remove_task'].url_for(task_id=t.task_id)}",
             )
             for t in await lrt_api.list_tasks(
-                long_running_manager.tasks_manager,
+                long_running_manager.rpc_client,
+                long_running_manager.lrt_namespace,
                 long_running_manager.get_task_context(request),
             )
         ]
@@ -42,7 +47,8 @@ async def get_task_status(request: web.Request) -> web.Response:
     long_running_manager = get_long_running_manager(request.app)
 
     task_status = await lrt_api.get_task_status(
-        long_running_manager.tasks_manager,
+        long_running_manager.rpc_client,
+        long_running_manager.lrt_namespace,
         long_running_manager.get_task_context(request),
         path_params.task_id,
     )
@@ -56,20 +62,36 @@ async def get_task_result(request: web.Request) -> web.Response | Any:
 
     # NOTE: this might raise an exception that will be catched by the _error_handlers
     return await lrt_api.get_task_result(
-        long_running_manager.tasks_manager,
+        long_running_manager.rpc_client,
+        long_running_manager.lrt_namespace,
         long_running_manager.get_task_context(request),
         path_params.task_id,
     )
 
 
-@routes.delete("/{task_id}", name="cancel_and_delete_task")
-async def cancel_and_delete_task(request: web.Request) -> web.Response:
+class _RemoveTaskQueryParams(RequestParameters):
+    wait_for_removal: Annotated[
+        bool,
+        Field(
+            description=(
+                "when True waits for the task to be removed "
+                "completly instead of returning immediately"
+            )
+        ),
+    ] = True
+
+
+@routes.delete("/{task_id}", name="remove_task")
+async def remove_task(request: web.Request) -> web.Response:
     path_params = parse_request_path_parameters_as(_PathParam, request)
+    query_params = parse_request_query_parameters_as(_RemoveTaskQueryParams, request)
     long_running_manager = get_long_running_manager(request.app)
 
     await lrt_api.remove_task(
-        long_running_manager.tasks_manager,
+        long_running_manager.rpc_client,
+        long_running_manager.lrt_namespace,
         long_running_manager.get_task_context(request),
         path_params.task_id,
+        wait_for_removal=query_params.wait_for_removal,
     )
     return web.json_response(status=status.HTTP_204_NO_CONTENT)

packages/service-library/src/servicelib/aiohttp/long_running_tasks/_server.py

@@ -8,20 +8,25 @@
 from aiohttp.web import HTTPException
 from common_library.json_serialization import json_dumps
 from pydantic import AnyHttpUrl, TypeAdapter
+from settings_library.rabbit import RabbitSettings
 from settings_library.redis import RedisSettings
 
 from ...aiohttp import status
 from ...long_running_tasks import lrt_api
-from ...long_running_tasks._redis_serialization import (
+from ...long_running_tasks._serialization import (
     BaseObjectSerializer,
     register_custom_serialization,
 )
 from ...long_running_tasks.constants import (
     DEFAULT_STALE_TASK_CHECK_INTERVAL,
     DEFAULT_STALE_TASK_DETECT_TIMEOUT,
 )
-from ...long_running_tasks.models import TaskContext, TaskGet
-from ...long_running_tasks.task import RedisNamespace, RegisteredTaskName
+from ...long_running_tasks.models import (
+    LRTNamespace,
+    RegisteredTaskName,
+    TaskContext,
+    TaskGet,
+)
 from ..typing_extension import Handler
 from . import _routes
 from ._constants import (
@@ -63,7 +68,8 @@ async def start_long_running_task(
     task_id = None
     try:
         task_id = await lrt_api.start_task(
-            long_running_manager.tasks_manager,
+            long_running_manager.rpc_client,
+            long_running_manager.lrt_namespace,
             registerd_task_name,
             fire_and_forget=fire_and_forget,
             task_context=task_context,
@@ -81,7 +87,7 @@ async def start_long_running_task(
             f"http://{ip_addr}:{port}{request_.app.router['get_task_result'].url_for(task_id=task_id)}"  # NOSONAR
         )
         abort_url = TypeAdapter(AnyHttpUrl).validate_python(
-            f"http://{ip_addr}:{port}{request_.app.router['cancel_and_delete_task'].url_for(task_id=task_id)}"  # NOSONAR
+            f"http://{ip_addr}:{port}{request_.app.router['remove_task'].url_for(task_id=task_id)}"  # NOSONAR
         )
         task_get = TaskGet(
             task_id=task_id,
@@ -98,7 +104,11 @@ async def start_long_running_task(
         # remove the task, the client was disconnected
         if task_id:
             await lrt_api.remove_task(
-                long_running_manager.tasks_manager, task_context, task_id
+                long_running_manager.rpc_client,
+                long_running_manager.lrt_namespace,
+                task_context,
+                task_id,
+                wait_for_removal=True,
             )
         raise
 
@@ -143,20 +153,23 @@ def setup(
     *,
     router_prefix: str,
     redis_settings: RedisSettings,
-    redis_namespace: RedisNamespace,
-    handler_check_decorator: Callable = _no_ops_decorator,
-    task_request_context_decorator: Callable = _no_task_context_decorator,
+    rabbit_settings: RabbitSettings,
+    lrt_namespace: LRTNamespace,
     stale_task_check_interval: datetime.timedelta = DEFAULT_STALE_TASK_CHECK_INTERVAL,
     stale_task_detect_timeout: datetime.timedelta = DEFAULT_STALE_TASK_DETECT_TIMEOUT,
+    handler_check_decorator: Callable = _no_ops_decorator,
+    task_request_context_decorator: Callable = _no_task_context_decorator,
 ) -> None:
     """
     - `router_prefix` APIs are mounted on `/...`, this
         will change them to be mounted as `{router_prefix}/...`
-    - `stale_task_check_interval_s` interval at which the
+    - `redis_settings` settings for Redis connection
+    - `rabbit_settings` settings for RabbitMQ connection
+    - `lrt_namespace` namespace for the long-running tasks
+    - `stale_task_check_interval` interval at which the
         TaskManager checks for tasks which are no longer being
         actively monitored by a client
-    - `stale_task_detect_timeout_s` interval after which a
-        task is considered stale
+    - `stale_task_detect_timeout` interval after which atask is considered stale
     """
 
     async def on_cleanup_ctx(app: web.Application) -> AsyncGenerator[None, None]:
@@ -168,11 +181,11 @@ async def on_cleanup_ctx(app: web.Application) -> AsyncGenerator[None, None]:
         # add components to state
         app[APP_LONG_RUNNING_MANAGER_KEY] = long_running_manager = (
             AiohttpLongRunningManager(
-                app=app,
                 stale_task_check_interval=stale_task_check_interval,
                 stale_task_detect_timeout=stale_task_detect_timeout,
                 redis_settings=redis_settings,
-                redis_namespace=redis_namespace,
+                rabbit_settings=rabbit_settings,
+                lrt_namespace=lrt_namespace,
             )
         )
 

packages/service-library/src/servicelib/aiohttp/profiler_middleware.py

@@ -1,9 +1,6 @@
 from aiohttp.web import HTTPInternalServerError, Request, StreamResponse, middleware
-from servicelib.mimetype_constants import (
-    MIMETYPE_APPLICATION_JSON,
-    MIMETYPE_APPLICATION_ND_JSON,
-)
 
+from ..mimetype_constants import MIMETYPE_APPLICATION_JSON, MIMETYPE_APPLICATION_ND_JSON
 from ..utils_profiling_middleware import _is_profiling, _profiler, append_profile
 
 

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

@@ -2,9 +2,10 @@
 
 import httpx
 from fastapi import FastAPI
-from servicelib.fastapi.tracing import setup_httpx_client_tracing
 from settings_library.tracing import TracingSettings
 
+from .tracing import setup_httpx_client_tracing
+
 
 def setup_client_session(
     app: FastAPI,