♻️ add rabbitmq interface to long_running_tasks ⚠️🚨

.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

@@ -53,8 +53,8 @@ def get_async_job_status(
 
 @router.delete(
     "/tasks/{task_id}",
-    name="cancel_and_delete_task",
-    description="Cancels and deletes a task",
+    name="remove_task",
+    description="Cancels and removes a task",
     responses=_export_data_responses,
     status_code=status.HTTP_204_NO_CONTENT,
 )

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

@@ -26,10 +26,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,
                 long_running_manager.get_task_context(request),
             )
         ]
@@ -42,7 +43,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,
         long_running_manager.get_task_context(request),
         path_params.task_id,
     )
@@ -56,19 +58,21 @@ 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,
         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:
+@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)
     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,
         long_running_manager.get_task_context(request),
         path_params.task_id,
     )

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

@@ -8,20 +8,21 @@
 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, TaskContext, TaskGet
+from ...long_running_tasks.task import RegisteredTaskName
 from ..typing_extension import Handler
 from . import _routes
 from ._constants import (
@@ -63,7 +64,7 @@ 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,
             registerd_task_name,
             fire_and_forget=fire_and_forget,
             task_context=task_context,
@@ -81,7 +82,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 +99,10 @@ 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,
+                task_context,
+                task_id,
             )
         raise
 
@@ -143,7 +147,8 @@ def setup(
     *,
     router_prefix: str,
     redis_settings: RedisSettings,
-    redis_namespace: RedisNamespace,
+    rabbit_settings: RabbitSettings,
+    lrt_namespace: LRTNamespace,
     handler_check_decorator: Callable = _no_ops_decorator,
     task_request_context_decorator: Callable = _no_task_context_decorator,
     stale_task_check_interval: datetime.timedelta = DEFAULT_STALE_TASK_CHECK_INTERVAL,
@@ -168,11 +173,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/long_running_tasks/_client.py

@@ -171,7 +171,7 @@ async def get_task_result(
         return result.json()
 
     @retry_on_http_errors
-    async def cancel_and_delete_task(
+    async def remove_task(
         self, task_id: TaskId, *, timeout: PositiveFloat | None = None  # noqa: ASYNC109
     ) -> None:
         timeout = timeout or self._client_configuration.default_timeout

packages/service-library/src/servicelib/fastapi/long_running_tasks/_context_manager.py

@@ -121,7 +121,7 @@ async def _wait_for_task_result() -> Any:
 
         yield result
     except TimeoutError as e:
-        await client.cancel_and_delete_task(task_id)
+        await client.remove_task(task_id)
         raise TaskClientTimeoutError(
             task_id=task_id,
             timeout=task_timeout,

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

@@ -1,35 +1,5 @@
-import datetime
-
-from fastapi import FastAPI
-from settings_library.redis import RedisSettings
-
 from ...long_running_tasks.base_long_running_manager import BaseLongRunningManager
-from ...long_running_tasks.task import RedisNamespace, TasksManager
 
 
 class FastAPILongRunningManager(BaseLongRunningManager):
-    def __init__(
-        self,
-        app: FastAPI,
-        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()
+    pass

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

@@ -25,12 +25,10 @@ async def list_tasks(
             task_id=t.task_id,
             status_href=str(request.url_for("get_task_status", task_id=t.task_id)),
             result_href=str(request.url_for("get_task_result", task_id=t.task_id)),
-            abort_href=str(
-                request.url_for("cancel_and_delete_task", task_id=t.task_id)
-            ),
+            abort_href=str(request.url_for("remove_task", task_id=t.task_id)),
         )
         for t in await lrt_api.list_tasks(
-            long_running_manager.tasks_manager, task_context={}
+            long_running_manager.rpc_client, long_running_manager, task_context={}
         )
     ]
 
@@ -52,7 +50,10 @@ async def get_task_status(
 ) -> TaskStatus:
     assert request  # nosec
     return await lrt_api.get_task_status(
-        long_running_manager.tasks_manager, task_context={}, task_id=task_id
+        long_running_manager.rpc_client,
+        long_running_manager,
+        task_context={},
+        task_id=task_id,
     )
 
 
@@ -75,21 +76,24 @@ async def get_task_result(
 ) -> TaskResult | Any:
     assert request  # nosec
     return await lrt_api.get_task_result(
-        long_running_manager.tasks_manager, task_context={}, task_id=task_id
+        long_running_manager.rpc_client,
+        long_running_manager,
+        task_context={},
+        task_id=task_id,
     )
 
 
 @router.delete(
     "/{task_id}",
-    summary="Cancel and deletes a task",
+    summary="Cancels and removes a task",
     response_model=None,
     status_code=status.HTTP_204_NO_CONTENT,
     responses={
         status.HTTP_404_NOT_FOUND: {"description": "Task does not exist"},
     },
 )
 @cancel_on_disconnect
-async def cancel_and_delete_task(
+async def remove_task(
     request: Request,
     task_id: TaskId,
     long_running_manager: Annotated[
@@ -98,5 +102,8 @@ async def cancel_and_delete_task(
 ) -> None:
     assert request  # nosec
     await lrt_api.remove_task(
-        long_running_manager.tasks_manager, task_context={}, task_id=task_id
+        long_running_manager.rpc_client,
+        long_running_manager,
+        task_context={},
+        task_id=task_id,
     )