Merge branch 'feature/node-in-use' of github.com:odeimaiz/osparc-simcore into feature/node-in-use

Author: odeimaiz

.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/_conversations.py

@@ -32,7 +32,6 @@
 )
 from simcore_service_webserver.conversations._controller._conversations_rest import (
     _ConversationsCreateBodyParams,
-    _GetConversationsQueryParams,
     _ListConversationsQueryParams,
 )
 
@@ -56,7 +55,6 @@
 )
 async def create_conversation(
     _body: _ConversationsCreateBodyParams,
-    _query: Annotated[_GetConversationsQueryParams, Depends()],
 ): ...
 
 
@@ -76,7 +74,6 @@ async def list_conversations(
 async def update_conversation(
     _params: Annotated[ConversationPathParams, Depends()],
     _body: ConversationPatch,
-    _query: Annotated[as_query(_GetConversationsQueryParams), Depends()],
 ): ...
 
 
@@ -86,7 +83,6 @@ async def update_conversation(
 )
 async def delete_conversation(
     _params: Annotated[ConversationPathParams, Depends()],
-    _query: Annotated[as_query(_GetConversationsQueryParams), Depends()],
 ): ...
 
 
@@ -96,7 +92,6 @@ async def delete_conversation(
 )
 async def get_conversation(
     _params: Annotated[ConversationPathParams, Depends()],
-    _query: Annotated[as_query(_GetConversationsQueryParams), Depends()],
 ): ...
 
 

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/models-library/src/models_library/api_schemas_directorv2/dynamic_services_service.py

@@ -1,5 +1,6 @@
 from functools import cached_property
 from pathlib import Path
+from typing import Annotated
 
 from pydantic import BaseModel, ConfigDict, Field
 from pydantic.config import JsonDict
@@ -89,6 +90,11 @@ class RunningDynamicServiceDetails(ServiceDetails):
         alias="service_message",
     )
 
+    is_collaborative: Annotated[
+        bool,
+        Field(description="True if service allows collaboration (multi-tenant access)"),
+    ] = False
+
     @staticmethod
     def _update_json_schema_extra(schema: JsonDict) -> None:
         schema.update(

packages/models-library/tests/test_docker.py

@@ -112,9 +112,9 @@ def test_docker_generic_tag(image_name: str, valid: bool):
 def test_simcore_service_docker_label_keys(obj_data: dict[str, Any]):
     simcore_service_docker_label_keys = SimcoreContainerLabels.model_validate(obj_data)
     exported_dict = simcore_service_docker_label_keys.to_simcore_runtime_docker_labels()
-    assert all(
-        isinstance(v, str) for v in exported_dict.values()
-    ), "docker labels must be strings!"
+    assert all(isinstance(v, str) for v in exported_dict.values()), (
+        "docker labels must be strings!"
+    )
     assert all(
         key.startswith(_SIMCORE_RUNTIME_DOCKER_LABEL_PREFIX) for key in exported_dict
     )

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)