🐛♻️Computational pipelines: Align computation pipeline HTTP codes, fix dynamic cycles bug (#8963)

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: sanderegg

api/specs/web-server/_computations.py

@@ -45,13 +45,16 @@ async def get_computation(_path: Annotated[ComputationPathParams, Depends()]): .
 @router.post(
     "/computations/{project_id}:start",
     response_model=Envelope[ComputationStarted],
+    status_code=status.HTTP_201_CREATED,
     responses={
+        status.HTTP_200_OK: {
+            "description": "Pipeline is up-to-date, nothing was started",
+            "model": Envelope[ComputationGet],
+        },
         status.HTTP_402_PAYMENT_REQUIRED: {"description": "Insufficient credits to run computation"},
-        status.HTTP_404_NOT_FOUND: {"description": "Project/wallet/pricing details were not found"},
-        status.HTTP_406_NOT_ACCEPTABLE: {"description": "Cluster not found"},
-        status.HTTP_409_CONFLICT: {"description": "Project already started"},
-        status.HTTP_422_UNPROCESSABLE_ENTITY: {"description": "Configuration error"},
-        status.HTTP_503_SERVICE_UNAVAILABLE: {"description": "Service not available"},
+        status.HTTP_404_NOT_FOUND: {"description": "Project/wallet/pricing/cluster details were not found"},
+        status.HTTP_409_CONFLICT: {"description": "Project already started or contains deprecated services"},
+        status.HTTP_503_SERVICE_UNAVAILABLE: {"description": "Service not available or configuration error"},
     },
 )
 async def start_computation(

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

@@ -2,6 +2,7 @@
 # pylint:disable=unused-argument
 # pylint:disable=redefined-outer-name
 
+import warnings
 from collections.abc import Iterator
 from dataclasses import dataclass
 from pathlib import Path
@@ -90,7 +91,10 @@ def dask_client_security(
 def dask_client(dask_scheduler_service: str, dask_client_security: distributed.Security) -> Iterator[Client]:
     client = Client(dask_scheduler_service, security=dask_client_security)
     yield client
-    client.close()
+    # SEE https://github.com/dask/distributed/issues/2507
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore", RuntimeWarning)
+        client.close()
 
 
 @pytest.fixture

services/api-server/src/simcore_service_api_server/exceptions/backend_errors.py

@@ -88,14 +88,11 @@ class SolverOutputNotFoundError(BaseBackEndError):
     status_code = status.HTTP_404_NOT_FOUND
 
 
-class ClusterNotFoundError(BaseBackEndError):
-    msg_template = user_message("The requested cluster could not be found.", _version=1)
-    status_code = status.HTTP_406_NOT_ACCEPTABLE
 
 
 class ConfigurationError(BaseBackEndError):
     msg_template = user_message("A configuration error occurred.", _version=1)
-    status_code = status.HTTP_422_UNPROCESSABLE_ENTITY
+    status_code = status.HTTP_503_SERVICE_UNAVAILABLE
 
 
 class ProductPriceNotFoundError(BaseBackEndError):
@@ -145,7 +142,8 @@ class InsufficientNumberOfSeatsError(BaseBackEndError):
 
 class CanNotCheckoutServiceIsNotRunningError(BaseBackEndError):
     msg_template = user_message(
-        "Unable to check out license item {licensed_item_id} because the dynamic service is not running. Current service ID: {service_run_id}.",
+        "Unable to check out license item {licensed_item_id} because the dynamic service is not running. "
+        "Current service ID: {service_run_id}.",
         _version=1,
     )
     status_code = status.HTTP_422_UNPROCESSABLE_ENTITY
@@ -171,7 +169,8 @@ class CeleryTaskNotFoundError(BaseBackEndError):
 
 class SolverJobOutputRequestButNotSucceededError(BaseBackEndError):
     msg_template = user_message(
-        "Cannot retrieve output for solver job '{job_id}' because it has not completed successfully. Current state: {state}.",
+        "Cannot retrieve output for solver job '{job_id}' because it has not completed successfully. "
+        "Current state: {state}.",
         _version=1,
     )
     status_code = status.HTTP_409_CONFLICT
@@ -187,7 +186,8 @@ class SolverJobNotStoppedYetError(BaseBackEndError):
 
 class StudyJobOutputRequestButNotSucceededError(BaseBackEndError):
     msg_template = user_message(
-        "Cannot retrieve output for project job '{job_id}' because it has not completed successfully. Current state: {state}.",
+        "Cannot retrieve output for project job '{job_id}' because it has not completed successfully. "
+        "Current state: {state}.",
         _version=1,
     )
     status_code = status.HTTP_409_CONFLICT

services/api-server/src/simcore_service_api_server/services_http/webserver.py

@@ -57,7 +57,6 @@
 
 from ..core.settings import WebServerSettings
 from ..exceptions.backend_errors import (
-    ClusterNotFoundError,
     ConfigurationError,
     ForbiddenWalletError,
     JobNotFoundError,
@@ -265,7 +264,10 @@ async def _wait_for_long_running_task_results(self, lrt_response: httpx.Response
                 task_status = Envelope[TaskStatus].model_validate_json(get_response.text).data
                 assert task_status is not None  # nosec
                 if not task_status.done:
-                    msg = "Timed out creating project. TIP: Try again, or contact oSparc support if this is happening repeatedly"
+                    msg = (
+                        "Timed out creating project. "
+                        "TIP: Try again, or contact oSparc support if this is happening repeatedly"
+                    )
                     raise TryAgain(msg)
 
         result_response = await self.long_running_task_client.get(
@@ -492,8 +494,7 @@ async def connect_pricing_unit_to_project_node(
         http_status_map=_JOB_STATUS_MAP
         | {
             status.HTTP_409_CONFLICT: ProjectAlreadyStartedError,
-            status.HTTP_406_NOT_ACCEPTABLE: ClusterNotFoundError,
-            status.HTTP_422_UNPROCESSABLE_ENTITY: ConfigurationError,
+            status.HTTP_503_SERVICE_UNAVAILABLE: ConfigurationError,
         }
     )
     async def start_project(

services/director-v2/openapi.json

@@ -89,23 +89,23 @@
               }
             }
           },
-          "404": {
-            "description": "Project or pricing details not found"
+          "200": {
+            "description": "Pipeline is up-to-date, nothing was started"
           },
-          "406": {
-            "description": "Cluster not found"
+          "404": {
+            "description": "Project/cluster/pricing details not found"
           },
           "503": {
-            "description": "Service not available"
-          },
-          "422": {
-            "description": "Configuration error"
+            "description": "Service not available or configuration error"
           },
           "402": {
             "description": "Payment required"
           },
           "409": {
-            "description": "Project already started"
+            "description": "Project already started or contains deprecated services"
+          },
+          "422": {
+            "description": "Invalid computation request (e.g. missing collection_run_id)"
           }
         }
       }

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

@@ -21,7 +21,7 @@
 from typing import Annotated, Any, Final
 
 import networkx as nx
-from fastapi import APIRouter, Depends, FastAPI, HTTPException
+from fastapi import APIRouter, Depends, FastAPI, HTTPException, Response
 from models_library.api_schemas_directorv2.computations import (
     ComputationCreate,
     ComputationDelete,
@@ -53,6 +53,7 @@
     ComputationalRunNotFoundError,
     ComputationalSchedulerError,
     ConfigurationError,
+    EC2InstanceTypeNotFoundError,
     PipelineTaskMissingError,
     PricingPlanUnitNotFoundError,
     ProjectNotFoundError,
@@ -119,7 +120,7 @@ async def _check_pipeline_startable(
         catalog_client,
     ):
         raise HTTPException(
-            status_code=status.HTTP_406_NOT_ACCEPTABLE,
+            status_code=status.HTTP_409_CONFLICT,
             detail=f"Project {computation.project_id} cannot run since "
             f"it contains deprecated tasks {jsonable_encoder(deprecated_tasks)}",
         )
@@ -187,25 +188,14 @@ async def _try_start_pipeline(
     *,
     project_repo: ProjectsRepository,
     computation: ComputationCreate,
-    complete_dag: nx.DiGraph,
     minimal_dag: nx.DiGraph,
     project: ProjectAtDB,
     users_repo: UsersRepository,
     projects_metadata_repo: ProjectsMetadataRepository,
-) -> None:
+) -> bool:
     if not minimal_dag.nodes():
-        # 2 options here: either we have cycles in the graph or it's really done
-        if find_computational_node_cycles(complete_dag):
-            raise HTTPException(
-                status_code=status.HTTP_409_CONFLICT,
-                detail=f"Project {computation.project_id} contains cycles with "
-                "computational services which are currently not supported! Please remove them.",
-            )
-        # there is nothing else to be run here, so we are done
-        raise HTTPException(
-            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
-            detail=f"Project {computation.project_id} has no computational services",
-        )
+        # there is nothing else to be run here (up-to-date or no computational services)
+        return False
 
     # Billing info
     wallet_id = None
@@ -239,6 +229,7 @@ async def _try_start_pipeline(
         use_on_demand_clusters=computation.use_on_demand_clusters,
         collection_run_id=computation.collection_run_id,
     )
+    return True
 
 
 @router.post(
@@ -247,20 +238,20 @@ async def _try_start_pipeline(
     response_model=ComputationGet,
     status_code=status.HTTP_201_CREATED,
     responses={
-        status.HTTP_404_NOT_FOUND: {
-            "description": "Project or pricing details not found",
+        status.HTTP_200_OK: {
+            "description": "Pipeline is up-to-date, nothing was started",
         },
-        status.HTTP_406_NOT_ACCEPTABLE: {
-            "description": "Cluster not found",
+        status.HTTP_404_NOT_FOUND: {
+            "description": "Project/cluster/pricing details not found",
         },
         status.HTTP_503_SERVICE_UNAVAILABLE: {
-            "description": "Service not available",
+            "description": "Service not available or configuration error",
         },
+        status.HTTP_402_PAYMENT_REQUIRED: {"description": "Payment required"},
+        status.HTTP_409_CONFLICT: {"description": "Project already started or contains deprecated services"},
         status.HTTP_422_UNPROCESSABLE_ENTITY: {
-            "description": "Configuration error",
+            "description": "Invalid computation request (e.g. missing collection_run_id)",
         },
-        status.HTTP_402_PAYMENT_REQUIRED: {"description": "Payment required"},
-        status.HTTP_409_CONFLICT: {"description": "Project already started"},
     },
 )
 # NOTE: in case of a burst of calls to that endpoint, we might end up in a weird state.
@@ -269,6 +260,7 @@ async def _try_start_pipeline(
 async def create_or_update_or_start_computation(  # noqa: PLR0913 # pylint: disable=too-many-positional-arguments
     computation: ComputationCreate,
     request: Request,
+    response: Response,
     project_repo: Annotated[ProjectsRepository, Depends(get_repository(ProjectsRepository))],
     comp_pipelines_repo: Annotated[CompPipelinesRepository, Depends(get_repository(CompPipelinesRepository))],
     comp_tasks_repo: Annotated[CompTasksRepository, Depends(get_repository(CompTasksRepository))],
@@ -293,6 +285,19 @@ async def create_or_update_or_start_computation(  # noqa: PLR0913 # pylint: disa
 
         # create the complete DAG graph
         complete_dag = create_complete_dag(project.workbench)
+
+        # reject cycles involving computational nodes early (before catalog checks)
+        if computation.start_pipeline and (computational_cycles := find_computational_node_cycles(complete_dag)):
+            cycle_descriptions = [
+                [complete_dag.nodes[n].get("name", n) for n in cycle] for cycle in computational_cycles
+            ]
+            raise HTTPException(
+                status_code=status.HTTP_409_CONFLICT,
+                detail=f"Project {computation.project_id} contains cycles with "
+                "computational services which are currently not supported: "
+                f"{cycle_descriptions}. Please remove them.",
+            )
+
         # find the minimal viable graph to be run
         minimal_computational_dag: nx.DiGraph = await create_minimal_computational_graph_based_on_selection(
             complete_dag=complete_dag,
@@ -322,17 +327,19 @@ async def create_or_update_or_start_computation(  # noqa: PLR0913 # pylint: disa
             rabbitmq_rpc_client=rpc_client,
         )
 
+        pipeline_started = False
         if computation.start_pipeline:
-            await _try_start_pipeline(
+            pipeline_started = await _try_start_pipeline(
                 request.app,
                 project_repo=project_repo,
                 computation=computation,
-                complete_dag=complete_dag,
                 minimal_dag=minimal_computational_dag,
                 project=project,
                 users_repo=users_repo,
                 projects_metadata_repo=projects_metadata_repo,
             )
+            if not pipeline_started:
+                response.status_code = status.HTTP_200_OK
 
         # get run details if any
         last_run: CompRunsAtDB | None = None
@@ -352,7 +359,7 @@ async def create_or_update_or_start_computation(  # noqa: PLR0913 # pylint: disa
                 TypeAdapter(AnyHttpUrl).validate_python(
                     f"{request.url}/{computation.project_id}:stop?user_id={computation.user_id}",
                 )
-                if computation.start_pipeline
+                if computation.start_pipeline and pipeline_started
                 else None
             ),
             iteration=last_run.iteration if last_run else None,
@@ -362,16 +369,20 @@ async def create_or_update_or_start_computation(  # noqa: PLR0913 # pylint: disa
             submitted=last_run.created if last_run else None,
         )
 
-    except ProjectNotFoundError as e:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"{e}") from e
-    except ClusterNotFoundError as e:
-        raise HTTPException(status_code=status.HTTP_406_NOT_ACCEPTABLE, detail=f"{e}") from e
-    except PricingPlanUnitNotFoundError as e:
+    except (
+        ProjectNotFoundError,
+        ClusterNotFoundError,
+        PricingPlanUnitNotFoundError,
+        EC2InstanceTypeNotFoundError,
+    ) as e:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"{e}") from e
-    except ClustersKeeperNotAvailableError as e:
+
+    except (
+        ClustersKeeperNotAvailableError,
+        ConfigurationError,
+    ) as e:
         raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=f"{e}") from e
-    except ConfigurationError as e:
-        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=f"{e}") from e
+
     except WalletNotEnoughCreditsError as e:
         raise HTTPException(status_code=status.HTTP_402_PAYMENT_REQUIRED, detail=f"{e}") from e
 

services/director-v2/src/simcore_service_director_v2/core/errors.py

@@ -51,6 +51,10 @@ class PricingPlanUnitNotFoundError(DirectorError):
     msg_template: str = "pricing plan not found {msg}"
 
 
+class EC2InstanceTypeNotFoundError(DirectorError):
+    msg_template: str = "invalid EC2 instance type selected {ec2_instance_types}. TIP: adjust product configuration"
+
+
 class PipelineNotFoundError(DirectorError):
     msg_template: str = "pipeline {pipeline_id} not found"
 
@@ -115,7 +119,8 @@ class MissingComputationalResourcesError(TaskSchedulingError):  # pylint: disabl
 
 class InsufficientComputationalResourcesError(TaskSchedulingError):  # pylint: disable=too-many-ancestors
     msg_template: str = (
-        "Insufficient computational resources to run {service_name}:{service_version} with {service_requested_resources} on cluster."
+        "Insufficient computational resources to run {service_name}:{service_version} "
+        "with {service_requested_resources} on cluster."
         "Cluster available workers: {cluster_available_resources}"
         "TIP: Reduce service required resources or contact oSparc support"
     )

services/director-v2/src/simcore_service_director_v2/modules/db/repositories/comp_tasks/_utils.py

@@ -52,7 +52,7 @@
 
 from .....core.errors import (
     ClustersKeeperNotAvailableError,
-    ConfigurationError,
+    EC2InstanceTypeNotFoundError,
     WalletNotEnoughCreditsError,
 )
 from .....models.comp_tasks import CompTaskAtDB, Image, NodeSchema
@@ -286,10 +286,11 @@ def _by_type_name(ec2: EC2InstanceTypeGet) -> bool:
         else:
             _logger.warning("Services resource override not implemented yet for multi-container services!!!")
     except StopIteration as exc:
-        msg = (
-            f"invalid EC2 type name selected {set(hardware_info.aws_ec2_instances)}. TIP: adjust product configuration"
-        )
-        raise ConfigurationError(msg=msg) from exc
+        raise EC2InstanceTypeNotFoundError(
+            ec2_instance_types=f"{set(hardware_info.aws_ec2_instances)}",
+            node_id=f"{node_id}",
+            project_id=f"{project_id}",
+        ) from exc
     except (
         RemoteMethodNotRegisteredError,
         RPCServerError,

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

@@ -126,9 +126,16 @@ def _node_needs_computation(graph_data: nx.classes.reportviews.NodeDataView, nod
 
 async def _set_computational_nodes_states(complete_dag: nx.DiGraph) -> None:
     graph_data: nx.classes.reportviews.NodeDataView = complete_dag.nodes.data()
-    for node_id in nx.algorithms.dag.topological_sort(complete_dag):
-        if graph_data[node_id]["node_class"] is NodeClass.COMPUTATIONAL:
-            await _compute_node_states(graph_data, node_id)
+    # Build a subgraph of only computational nodes for topological ordering.
+    # The complete_dag may contain cycles among dynamic services (which is valid),
+    # but those cycles prevent topological_sort on the full graph. Since we only
+    # process computational nodes, we only need *their* ordering to be acyclic.
+    # State lookups (hash, dependencies) still use graph_data from the complete DAG,
+    # so dynamic-node data remains accessible.
+    comp_nodes = [n for n, d in graph_data if d["node_class"] is NodeClass.COMPUTATIONAL]
+    comp_subgraph = complete_dag.subgraph(comp_nodes)
+    for node_id in nx.algorithms.dag.topological_sort(comp_subgraph):
+        await _compute_node_states(graph_data, node_id)
 
 
 async def create_minimal_computational_graph_based_on_selection(