Merge branch 'master' into feature/listenToProjectDocumentWS

Author: odeimaiz

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

@@ -118,7 +118,7 @@ def handle_aiohttp_web_http_error(
     exception.content_type = MIMETYPE_APPLICATION_JSON
     if exception.reason:
         exception.set_status(
-            exception.status, safe_status_message(message=exception.reason)
+            exception.status, reason=safe_status_message(message=exception.reason)
         )
 
     if not exception.text or not is_enveloped_from_text(exception.text):
@@ -165,7 +165,7 @@ def _handle_aiohttp_web_http_successful(
     exception.content_type = MIMETYPE_APPLICATION_JSON
     if exception.reason:
         exception.set_status(
-            exception.status, safe_status_message(message=exception.reason)
+            exception.status, reason=safe_status_message(message=exception.reason)
         )
 
     if exception.text and not is_enveloped_from_text(exception.text):

services/director-v2/src/simcore_service_director_v2/api/routes/computations.py

@@ -58,7 +58,6 @@
     ProjectNotFoundError,
     WalletNotEnoughCreditsError,
 )
-from ...models.comp_pipelines import CompPipelineAtDB
 from ...models.comp_runs import CompRunsAtDB, ProjectMetadataDict, RunMetadataDict
 from ...models.comp_tasks import CompTaskAtDB
 from ...modules.catalog import CatalogClient
@@ -537,10 +536,8 @@ async def stop_computation(
         # check the project exists
         await project_repo.get_project(project_id)
         # get the project pipeline
-        pipeline_at_db: CompPipelineAtDB = await comp_pipelines_repo.get_pipeline(
-            project_id
-        )
-        pipeline_dag: nx.DiGraph = pipeline_at_db.get_graph()
+        pipeline_at_db = await comp_pipelines_repo.get_pipeline(project_id)
+        pipeline_dag = pipeline_at_db.get_graph()
         # get the project task states
         tasks: list[CompTaskAtDB] = await comp_tasks_repo.list_tasks(project_id)
         # create the complete DAG graph

services/director-v2/src/simcore_service_director_v2/utils/dags.py

@@ -42,7 +42,7 @@ def create_complete_dag(workbench: NodesDict) -> nx.DiGraph:
         )
         if node.input_nodes:
             for input_node_id in node.input_nodes:
-                predecessor_node = workbench.get(NodeIDStr(input_node_id))
+                predecessor_node = workbench.get(f"{input_node_id}")
                 if predecessor_node:
                     dag_graph.add_edge(str(input_node_id), node_id)
 
@@ -95,19 +95,18 @@ async def get_node_io_payload_cb(node_id: NodeID) -> dict[str, Any]:
         return result
 
     computed_hash = await compute_node_hash(node_id, get_node_io_payload_cb)
-    if computed_hash != node["run_hash"]:
-        return True
-    return False
+    return bool(computed_hash != node["run_hash"])
 
 
 async def _compute_node_dependencies_state(graph_data, node_id) -> set[NodeID]:
     node = graph_data[f"{node_id}"]
     # check if the previous node is outdated or waits for dependencies... in which case this one has to wait
     non_computed_dependencies: set[NodeID] = set()
     for input_port in node.get("inputs", {}).values():
-        if isinstance(input_port, PortLink):
-            if _node_needs_computation(graph_data, input_port.node_uuid):
-                non_computed_dependencies.add(input_port.node_uuid)
+        if isinstance(input_port, PortLink) and _node_needs_computation(
+            graph_data, input_port.node_uuid
+        ):
+            non_computed_dependencies.add(input_port.node_uuid)
     # all good. ready
     return non_computed_dependencies
 
@@ -188,14 +187,14 @@ def compute_pipeline_started_timestamp(
     if not pipeline_dag.nodes:
         return None
     node_id_to_comp_task: dict[NodeIDStr, CompTaskAtDB] = {
-        NodeIDStr(f"{task.node_id}"): task for task in comp_tasks
+        f"{task.node_id}": task for task in comp_tasks
     }
-    TOMORROW = arrow.utcnow().shift(days=1).datetime
+    tomorrow = arrow.utcnow().shift(days=1).datetime
     pipeline_started_at: datetime.datetime | None = min(
-        node_id_to_comp_task[node_id].start or TOMORROW
+        node_id_to_comp_task[node_id].start or tomorrow
         for node_id in pipeline_dag.nodes
     )
-    if pipeline_started_at == TOMORROW:
+    if pipeline_started_at == tomorrow:
         pipeline_started_at = None
     return pipeline_started_at
 
@@ -206,13 +205,13 @@ def compute_pipeline_stopped_timestamp(
     if not pipeline_dag.nodes:
         return None
     node_id_to_comp_task: dict[NodeIDStr, CompTaskAtDB] = {
-        NodeIDStr(f"{task.node_id}"): task for task in comp_tasks
+        f"{task.node_id}": task for task in comp_tasks
     }
-    TOMORROW = arrow.utcnow().shift(days=1).datetime
+    tomorrow = arrow.utcnow().shift(days=1).datetime
     pipeline_stopped_at: datetime.datetime | None = max(
-        node_id_to_comp_task[node_id].end or TOMORROW for node_id in pipeline_dag.nodes
+        node_id_to_comp_task[node_id].end or tomorrow for node_id in pipeline_dag.nodes
     )
-    if pipeline_stopped_at == TOMORROW:
+    if pipeline_stopped_at == tomorrow:
         pipeline_stopped_at = None
     return pipeline_stopped_at
 
@@ -227,15 +226,15 @@ async def compute_pipeline_details(
 
     # NOTE: the latest progress is available in comp_tasks only
     node_id_to_comp_task: dict[NodeIDStr, CompTaskAtDB] = {
-        NodeIDStr(f"{task.node_id}"): task for task in comp_tasks
+        f"{task.node_id}": task for task in comp_tasks
     }
     pipeline_progress = None
     if len(pipeline_dag.nodes) > 0:
-
         pipeline_progress = sum(
             (node_id_to_comp_task[node_id].progress or 0) / len(pipeline_dag.nodes)
             for node_id in pipeline_dag.nodes
-            if node_id_to_comp_task[node_id].progress is not None
+            if node_id in node_id_to_comp_task
+            and node_id_to_comp_task[node_id].progress is not None
         )
         pipeline_progress = max(0.0, min(pipeline_progress, 1.0))
 
@@ -246,10 +245,15 @@ async def compute_pipeline_details(
             node_id: NodeState(
                 modified=node_data.get(kNODE_MODIFIED_STATE, False),
                 dependencies=node_data.get(kNODE_DEPENDENCIES_TO_COMPUTE, set()),
-                current_status=node_id_to_comp_task[node_id].state,
+                current_status=(
+                    node_id_to_comp_task[node_id].state
+                    if node_id in node_id_to_comp_task
+                    else RunningState.UNKNOWN
+                ),
                 progress=(
                     node_id_to_comp_task[node_id].progress
-                    if node_id_to_comp_task[node_id].progress is not None
+                    if node_id in node_id_to_comp_task
+                    and node_id_to_comp_task[node_id].progress is not None
                     else None
                 ),
             )
@@ -261,12 +265,13 @@ async def compute_pipeline_details(
 
 def find_computational_node_cycles(dag: nx.DiGraph) -> list[list[str]]:
     """returns a list of nodes part of a cycle and computational, which is currently forbidden."""
-    computational_node_cycles = []
+
     list_potential_cycles = nx.algorithms.cycles.simple_cycles(dag)
-    for cycle in list_potential_cycles:
+    return [
+        deepcopy(cycle)
+        for cycle in list_potential_cycles
         if any(
             dag.nodes[node_id]["node_class"] is NodeClass.COMPUTATIONAL
             for node_id in cycle
-        ):
-            computational_node_cycles.append(deepcopy(cycle))
-    return computational_node_cycles
+        )
+    ]

services/director-v2/tests/unit/test_utils_dags.py

@@ -390,13 +390,6 @@ def pipeline_test_params(
     list_comp_tasks: list[CompTaskAtDB],
     expected_pipeline_details_output: PipelineDetails,
 ) -> PipelineDetailsTestParams:
-    # check the inputs make sense
-    assert len(set(dag_adjacency)) == len(node_keys) == len(list_comp_tasks)
-    assert dag_adjacency.keys() == node_keys.keys()
-    assert len(
-        {t.node_id for t in list_comp_tasks}.intersection(node_keys.keys())
-    ) == len(set(dag_adjacency))
-
     # resolve the naming
     node_name_to_uuid_map = {}
     resolved_dag_adjacency: dict[str, list[str]] = {}
@@ -596,3 +589,95 @@ async def test_compute_pipeline_details(
         received_details.model_dump()
         == pipeline_test_params.expected_pipeline_details.model_dump()
     )
+
+
+@pytest.mark.parametrize(
+    "dag_adjacency, node_keys, list_comp_tasks, expected_pipeline_details_output",
+    [
+        pytest.param(
+            {"node_1": ["node_2", "node_3"], "node_2": ["node_3"], "node_3": []},
+            {
+                "node_1": {
+                    "key": "simcore/services/comp/fake",
+                    "node_class": NodeClass.COMPUTATIONAL,
+                    "state": RunningState.NOT_STARTED,
+                    "outputs": None,
+                },
+                "node_2": {
+                    "key": "simcore/services/comp/fake",
+                    "node_class": NodeClass.COMPUTATIONAL,
+                    "state": RunningState.NOT_STARTED,
+                    "outputs": None,
+                },
+                "node_3": {
+                    "key": "simcore/services/comp/fake",
+                    "node_class": NodeClass.COMPUTATIONAL,
+                    "state": RunningState.NOT_STARTED,
+                    "outputs": None,
+                },
+            },
+            [
+                # NOTE: we use construct here to be able to use non uuid names to simplify test setup
+                CompTaskAtDB.model_construct(
+                    project_id=uuid4(),
+                    node_id="node_1",
+                    schema=NodeSchema(inputs={}, outputs={}),
+                    inputs=None,
+                    image=Image(name="simcore/services/comp/fake", tag="1.3.4"),
+                    state=RunningState.NOT_STARTED,
+                    internal_id=2,
+                    node_class=NodeClass.COMPUTATIONAL,
+                    created=datetime.datetime.now(tz=datetime.UTC),
+                    modified=datetime.datetime.now(tz=datetime.UTC),
+                    last_heartbeat=None,
+                ),
+                CompTaskAtDB.model_construct(
+                    project_id=uuid4(),
+                    node_id="node_2",
+                    schema=NodeSchema(inputs={}, outputs={}),
+                    inputs=None,
+                    image=Image(name="simcore/services/comp/fake", tag="1.3.4"),
+                    state=RunningState.NOT_STARTED,
+                    internal_id=3,
+                    node_class=NodeClass.COMPUTATIONAL,
+                    created=datetime.datetime.now(tz=datetime.UTC),
+                    modified=datetime.datetime.now(tz=datetime.UTC),
+                    last_heartbeat=None,
+                ),
+            ],
+            PipelineDetails.model_construct(
+                adjacency_list={
+                    "node_1": ["node_2", "node_3"],
+                    "node_2": ["node_3"],
+                    "node_3": [],
+                },
+                progress=0.0,
+                node_states={
+                    "node_1": NodeState(modified=True, progress=None),
+                    "node_2": NodeState(modified=True, progress=None),
+                    "node_3": NodeState(
+                        modified=True,
+                        progress=None,
+                        current_status=RunningState.UNKNOWN,
+                    ),
+                },
+            ),
+            id="dag with missing tasks (node 3 is missing, so it is not skipped in the pipeline details)",
+        )
+    ],
+)
+@pytest.mark.acceptance_test(
+    "For https://github.com/ITISFoundation/osparc-simcore/issues/8172"
+)
+async def test_compute_pipeline_details_with_missing_tasks(
+    pipeline_test_params: PipelineDetailsTestParams,
+):
+    received_details = await compute_pipeline_details(
+        pipeline_test_params.complete_dag,
+        pipeline_test_params.pipeline_dag,
+        pipeline_test_params.comp_tasks,
+    )
+    assert (
+        received_details.model_dump()
+        == pipeline_test_params.expected_pipeline_details.model_dump()
+    )

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/notifier/_notifier.py

@@ -11,7 +11,7 @@
 )
 from models_library.api_schemas_webserver.projects_nodes import NodeGet, NodeGetIdle
 from models_library.api_schemas_webserver.socketio import SocketIORoomStr
-from models_library.users import UserID
+from models_library.projects import ProjectID
 from servicelib.fastapi.app_state import SingletonInAppStateMixin
 from servicelib.services_utils import get_status_as_dict
 
@@ -23,20 +23,22 @@ def __init__(self, sio_manager: socketio.AsyncAioPikaManager):
         self._sio_manager = sio_manager
 
     async def notify_service_status(
-        self, user_id: UserID, status: NodeGet | DynamicServiceGet | NodeGetIdle
+        self, project_id: ProjectID, status: NodeGet | DynamicServiceGet | NodeGetIdle
     ) -> None:
         await self._sio_manager.emit(
             SOCKET_IO_SERVICE_STATUS_EVENT,
             data=jsonable_encoder(get_status_as_dict(status)),
-            room=SocketIORoomStr.from_user_id(user_id),
+            room=SocketIORoomStr.from_project_id(project_id),
         )
 
 
 async def notify_service_status_change(
-    app: FastAPI, user_id: UserID, status: NodeGet | DynamicServiceGet | NodeGetIdle
+    app: FastAPI,
+    project_id: ProjectID,
+    status: NodeGet | DynamicServiceGet | NodeGetIdle,
 ) -> None:
     notifier: Notifier = Notifier.get_from_app_state(app)
-    await notifier.notify_service_status(user_id=user_id, status=status)
+    await notifier.notify_service_status(project_id=project_id, status=status)
 
 
 async def lifespan(app: FastAPI) -> AsyncIterator[State]:

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/service_tracker/__init__.py

@@ -1,8 +1,8 @@
 from ._api import (
     NORMAL_RATE_POLL_INTERVAL,
     get_all_tracked_services,
+    get_project_id_for_service,
     get_tracked_service,
-    get_user_id_for_service,
     remove_tracked_service,
     set_frontend_notified_for_service,
     set_if_status_changed_for_service,
@@ -17,11 +17,11 @@
 
 __all__: tuple[str, ...] = (
     "get_all_tracked_services",
+    "get_project_id_for_service",
     "get_tracked_service",
-    "get_user_id_for_service",
-    "service_tracker_lifespan",
     "NORMAL_RATE_POLL_INTERVAL",
     "remove_tracked_service",
+    "service_tracker_lifespan",
     "set_frontend_notified_for_service",
     "set_if_status_changed_for_service",
     "set_request_as_running",

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/service_tracker/_api.py

@@ -11,12 +11,16 @@
     DynamicServiceStop,
 )
 from models_library.api_schemas_webserver.projects_nodes import NodeGet, NodeGetIdle
+from models_library.projects import ProjectID
 from models_library.projects_nodes_io import NodeID
 from models_library.services_enums import ServiceState
-from models_library.users import UserID
 from servicelib.deferred_tasks import TaskUID
 
-from ._models import SchedulerServiceState, TrackedServiceModel, UserRequestedState
+from ._models import (
+    SchedulerServiceState,
+    TrackedServiceModel,
+    UserRequestedState,
+)
 from ._setup import get_tracker
 
 _logger = logging.getLogger(__name__)
@@ -242,7 +246,7 @@ async def get_all_tracked_services(app: FastAPI) -> dict[NodeID, TrackedServiceM
     return await get_tracker(app).all()
 
 
-async def get_user_id_for_service(app: FastAPI, node_id: NodeID) -> UserID | None:
-    """returns user_id for the service"""
+async def get_project_id_for_service(app: FastAPI, node_id: NodeID) -> ProjectID | None:
+    """returns project_id for the service"""
     model: TrackedServiceModel | None = await get_tracker(app).load(node_id)
-    return model.user_id if model else None
+    return model.project_id if model else None

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/status_monitor/_deferred_get_status.py

@@ -7,8 +7,8 @@
     RunningDynamicServiceDetails,
 )
 from models_library.api_schemas_webserver.projects_nodes import NodeGet, NodeGetIdle
+from models_library.projects import ProjectID
 from models_library.projects_nodes_io import NodeID
-from models_library.users import UserID
 from servicelib.deferred_tasks import BaseDeferredHandler, TaskUID
 from servicelib.deferred_tasks._base_deferred_handler import DeferredContext
 
@@ -69,15 +69,15 @@ async def on_result(
         if await service_tracker.should_notify_frontend_for_service(
             app, node_id, status_changed=status_changed
         ):
-            user_id: UserID | None = await service_tracker.get_user_id_for_service(
-                app, node_id
+            project_id: ProjectID | None = (
+                await service_tracker.get_project_id_for_service(app, node_id)
             )
-            if user_id:
-                await notify_service_status_change(app, user_id, result)
+            if project_id:
+                await notify_service_status_change(app, project_id, result)
                 await service_tracker.set_frontend_notified_for_service(app, node_id)
             else:
                 _logger.info(
-                    "Did not find a user for '%s', skipping status delivery of: %s",
+                    "Did not find a project for '%s', skipping status delivery of: %s",
                     node_id,
                     result,
                 )