✨ Store computational services progress in comp_tasks table, decouple webserver from api-server 🗃️

packages/dask-task-models-library/src/dask_task_models_library/container_tasks/events.py

@@ -3,7 +3,7 @@
 from typing import TypeAlias, Union
 
 from distributed.worker import get_worker
-from pydantic import BaseModel, Extra, NonNegativeFloat
+from pydantic import BaseModel, Extra, validator
 
 
 class BaseTaskEvent(BaseModel, ABC):
@@ -20,7 +20,7 @@ class Config:
 
 
 class TaskProgressEvent(BaseTaskEvent):
-    progress: NonNegativeFloat
+    progress: float
 
     @staticmethod
     def topic_name() -> str:
@@ -44,6 +44,13 @@ class Config(BaseTaskEvent.Config):
             ]
         }
 
+    @validator("progress", always=True)
+    @classmethod
+    def ensure_between_0_1(cls, v):
+        if 0 <= v <= 1:
+            return v
+        return min(max(0, v), 1)
+
 
 LogMessageStr: TypeAlias = str
 LogLevelInt: TypeAlias = int

packages/models-library/src/models_library/projects_nodes.py

@@ -85,6 +85,12 @@ class NodeState(BaseModel):
         description="the node's current state",
         alias="currentStatus",
     )
+    progress: float | None = Field(
+        default=0,
+        ge=0.0,
+        le=1.0,
+        description="current progress of the task if available (None if not started or not a computational task)",
+    )
 
     class Config:
         extra = Extra.forbid
@@ -133,7 +139,11 @@ class Node(BaseModel):
         ..., description="The short name of the node", examples=["JupyterLab"]
     )
     progress: float | None = Field(
-        default=None, ge=0, le=100, description="the node progress value"
+        default=None,
+        ge=0,
+        le=100,
+        description="the node progress value",
+        deprecated=True,
     )
     thumbnail: HttpUrlWithCustomMinLength | None = Field(
         default=None,

packages/models-library/src/models_library/projects_pipeline.py

@@ -1,4 +1,3 @@
-from typing import Dict, List, Optional
 from uuid import UUID
 
 from pydantic import BaseModel, Field, PositiveInt
@@ -9,11 +8,17 @@
 
 
 class PipelineDetails(BaseModel):
-    adjacency_list: Dict[NodeID, List[NodeID]] = Field(
+    adjacency_list: dict[NodeID, list[NodeID]] = Field(
         ...,
         description="The adjacency list of the current pipeline in terms of {NodeID: [successor NodeID]}",
     )
-    node_states: Dict[NodeID, NodeState] = Field(
+    progress: float | None = Field(
+        ...,
+        ge=0,
+        le=1.0,
+        description="the progress of the pipeline (None if there are no computational tasks)",
+    )
+    node_states: dict[NodeID, NodeState] = Field(
         ..., description="The states of each of the computational nodes in the pipeline"
     )
 
@@ -24,17 +29,15 @@ class PipelineDetails(BaseModel):
 class ComputationTask(BaseModel):
     id: TaskID = Field(..., description="the id of the computation task")
     state: RunningState = Field(..., description="the state of the computational task")
-    result: Optional[str] = Field(
-        None, description="the result of the computational task"
-    )
+    result: str | None = Field(None, description="the result of the computational task")
     pipeline_details: PipelineDetails = Field(
         ..., description="the details of the generated pipeline"
     )
-    iteration: Optional[PositiveInt] = Field(
+    iteration: PositiveInt | None = Field(
         ...,
         description="the iteration id of the computation task (none if no task ran yet)",
     )
-    cluster_id: Optional[ClusterID] = Field(
+    cluster_id: ClusterID | None = Field(
         ...,
         description="the cluster on which the computaional task runs/ran (none if no task ran yet)",
     )
@@ -56,14 +59,17 @@ class Config:
                             "2fb4808a-e403-4a46-b52c-892560d27862": {
                                 "modified": True,
                                 "dependencies": [],
+                                "progress": 0.0,
                             },
                             "19a40c7b-0a40-458a-92df-c77a5df7c886": {
                                 "modified": False,
                                 "dependencies": [
                                     "2fb4808a-e403-4a46-b52c-892560d27862"
                                 ],
+                                "progress": 0.0,
                             },
                         },
+                        "progress": 0.0,
                     },
                     "iteration": None,
                     "cluster_id": None,
@@ -82,14 +88,17 @@ class Config:
                             "2fb4808a-e403-4a46-b52c-892560d27862": {
                                 "modified": False,
                                 "dependencies": [],
+                                "progress": 1.0,
                             },
                             "19a40c7b-0a40-458a-92df-c77a5df7c886": {
                                 "modified": False,
                                 "dependencies": [
                                     "2fb4808a-e403-4a46-b52c-892560d27862"
                                 ],
+                                "progress": 1.0,
                             },
                         },
+                        "progress": 1.0,
                     },
                     "iteration": 2,
                     "cluster_id": 0,

packages/postgres-database/src/simcore_postgres_database/migration/versions/0c084cb1091c_add_progress_to_comp_tasks.py

@@ -0,0 +1,30 @@
+"""add progress to comp_tasks
+
+Revision ID: 0c084cb1091c
+Revises: 432aa859098b
+Create Date: 2023-05-05 08:00:18.951040+00:00
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "0c084cb1091c"
+down_revision = "432aa859098b"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column(
+        "comp_tasks",
+        sa.Column("progress", sa.Numeric(precision=3, scale=2), nullable=True),
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column("comp_tasks", "progress")
+    # ### end Alembic commands ###

packages/postgres-database/src/simcore_postgres_database/models/comp_tasks.py

@@ -56,7 +56,7 @@ class NodeClass(enum.Enum):
         sa.Enum(StateType),
         nullable=False,
         server_default=StateType.NOT_STARTED.value,
-        doc="Current state in the task lifecicle",
+        doc="Current state in the task lifecycle",
     ),
     sa.Column(
         "errors",
@@ -65,6 +65,12 @@ class NodeClass(enum.Enum):
         doc="List[models_library.errors.ErrorDict] with error information"
         " for a failing state, otherwise set to None",
     ),
+    sa.Column(
+        "progress",
+        sa.Numeric(precision=3, scale=2),  # numbers from 0.00 and 1.00
+        nullable=True,
+        doc="current progress of the task if available",
+    ),
     # utc timestamps for submission/start/end
     sa.Column("submit", sa.DateTime, doc="UTC timestamp for task submission"),
     sa.Column("start", sa.DateTime, doc="UTC timestamp when task started"),

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

@@ -39,13 +39,13 @@ def creator(**user_kwargs) -> dict[str, Any]:
             )
             # this is needed to get the primary_gid correctly
             result = con.execute(
-                sa.select([users]).where(users.c.id == user_config["id"])
+                sa.select(users).where(users.c.id == user_config["id"])
             )
             user = result.first()
             assert user
             print(f"--> created {user=}")
             created_user_ids.append(user["id"])
-        return dict(user)
+        return dict(user._asdict())
 
     yield creator
 
@@ -81,7 +81,7 @@ def creator(user: dict[str, Any], **overrides) -> ProjectAtDB:
                 .returning(sa.literal_column("*"))
             )
 
-            inserted_project = ProjectAtDB.parse_obj(result.first())
+            inserted_project = ProjectAtDB.from_orm(result.first())
         print(f"--> created {inserted_project=}")
         created_project_ids.append(f"{inserted_project.uuid}")
         return inserted_project

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

@@ -44,7 +44,7 @@ def _is_docker_swarm_init(docker_client: docker.client.DockerClient) -> bool:
 
 
 @retry(
-    wait=wait_fixed(5),
+    wait=wait_fixed(1),
     stop=stop_after_delay(8 * MINUTE),
     before_sleep=before_sleep_log(log, logging.WARNING),
     reraise=True,
@@ -183,25 +183,26 @@ def docker_swarm(
     retry=retry_if_exception_type(AssertionError),
     stop=stop_after_delay(30),
 )
-def _wait_for_new_task_to_be_started(service: Any, old_task_ids: set[str]) -> None:
-    service.reload()
-    new_task_ids = {t["ID"] for t in service.tasks()}
-    assert len(new_task_ids.difference(old_task_ids)) == 1
+def _wait_for_migration_service_to_be_removed(
+    docker_client: docker.client.DockerClient,
+) -> None:
+    for service in docker_client.services.list():
+        if "migration" in service.name:  # type: ignore
+            raise TryAgain
 
 
-def _force_restart_migration_service(docker_client: docker.client.DockerClient) -> None:
+def _force_remove_migration_service(docker_client: docker.client.DockerClient) -> None:
     for migration_service in (
         service
         for service in docker_client.services.list()
-        if "migration" in service.name
+        if "migration" in service.name  # type: ignore
     ):
         print(
-            "WARNING: migration service detected before updating stack, it will be force-updated"
+            "WARNING: migration service detected before updating stack, it will be force-removed now and re-deployed to ensure DB update"
         )
-        before_update_task_ids = {t["ID"] for t in migration_service.tasks()}
-        migration_service.force_update()
-        _wait_for_new_task_to_be_started(migration_service, before_update_task_ids)
-        print(f"forced updated {migration_service.name}.")
+        migration_service.remove()  # type: ignore
+        _wait_for_migration_service_to_be_removed(docker_client)
+        print(f"forced updated {migration_service.name}.")  # type: ignore
 
 
 def _deploy_stack(compose_file: Path, stack_name: str) -> None:
@@ -273,7 +274,7 @@ def docker_stack(
 
     # NOTE: if the migration service was already running prior to this call it must
     # be force updated so that it does its job. else it remains and tests will fail
-    _force_restart_migration_service(docker_client)
+    _force_remove_migration_service(docker_client)
 
     # make up-version
     stacks_deployed: dict[str, dict] = {}

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

@@ -70,7 +70,6 @@
 
 
 def create_computation_cb(url, **kwargs) -> CallbackResult:
-
     assert "json" in kwargs, f"missing body in call to {url}"
     body = kwargs["json"]
     for param in ["user_id", "project_id"]:
@@ -113,6 +112,7 @@ def create_computation_cb(url, **kwargs) -> CallbackResult:
             "pipeline_details": {
                 "adjacency_list": pipeline,
                 "node_states": node_states,
+                "progress": 0,
             },
         },
     )
@@ -131,6 +131,7 @@ def get_computation_cb(url, **kwargs) -> CallbackResult:
             "pipeline_details": {
                 "adjacency_list": pipeline,
                 "node_states": node_states,
+                "progress": 0,
             },
             "iteration": 2,
             "cluster_id": 23,
@@ -350,7 +351,6 @@ def get_upload_link_cb(url: URL, **kwargs) -> CallbackResult:
     scheme = {LinkType.PRESIGNED: "http", LinkType.S3: "s3"}
 
     if file_size := kwargs["params"].get("file_size") is not None:
-
         upload_schema = FileUploadSchema(
             chunk_size=parse_obj_as(ByteSize, "5GiB"),
             urls=[parse_obj_as(AnyUrl, f"{scheme[link_type]}://{file_id}")],

packages/service-library/src/servicelib/long_running_tasks/_models.py

@@ -2,13 +2,13 @@
 import urllib.parse
 from asyncio import Task
 from datetime import datetime
-from typing import Any, Awaitable, Callable, Coroutine, Optional
+from typing import Any, Awaitable, Callable, Coroutine
 
 from pydantic import (
     BaseModel,
+    ConstrainedFloat,
     Field,
     PositiveFloat,
-    confloat,
     validate_arguments,
     validator,
 )
@@ -20,7 +20,13 @@
 TaskType = Callable[..., Coroutine[Any, Any, Any]]
 
 ProgressMessage = str
-ProgressPercent = confloat(ge=0.0, le=1.0)
+
+
+class ProgressPercent(ConstrainedFloat):
+    ge = 0.0
+    le = 1.0
+
+
 ProgressCallback = Callable[[ProgressMessage, ProgressPercent, TaskId], Awaitable[None]]
 
 
@@ -41,8 +47,8 @@ class TaskProgress(BaseModel):
     def update(
         self,
         *,
-        message: Optional[ProgressMessage] = None,
-        percent: Optional[ProgressPercent] = None,
+        message: ProgressMessage | None = None,
+        percent: ProgressPercent | None = None,
     ) -> None:
         """`percent` must be between 0.0 and 1.0 otherwise ValueError is raised"""
         if message:
@@ -77,7 +83,7 @@ class TrackedTask(BaseModel):
     )
 
     started: datetime = Field(default_factory=datetime.utcnow)
-    last_status_check: Optional[datetime] = Field(
+    last_status_check: datetime | None = Field(
         default=None,
         description=(
             "used to detect when if the task is not actively "
@@ -96,8 +102,8 @@ class TaskStatus(BaseModel):
 
 
 class TaskResult(BaseModel):
-    result: Optional[Any]
-    error: Optional[Any]
+    result: Any | None
+    error: Any | None
 
 
 class ClientConfiguration(BaseModel):

services/api-server/src/simcore_service_api_server/api/routes/solvers_jobs.py

@@ -46,7 +46,7 @@ def _compose_job_resource_name(solver_key, solver_version, job_id) -> str:
     )
 
 
-## JOBS ---------------
+# JOBS ---------------
 #
 # - Similar to docker container's API design (container = job and image = solver)
 #
@@ -224,7 +224,7 @@ async def inspect_job(
     job_id: UUID,
     user_id: PositiveInt = Depends(get_current_user_id),
     director2_api: DirectorV2Api = Depends(get_api_client(DirectorV2Api)),
-):
+) -> JobStatus:
     job_name = _compose_job_resource_name(solver_key, version, job_id)
     _logger.debug("Inspecting Job '%s'", job_name)