completed testing and added timeout

Author: sanderegg

services/director-v2/src/simcore_service_director_v2/modules/comp_scheduler/_scheduler_dask.py

@@ -363,11 +363,19 @@ async def _handle_computational_retrieval_error(
             )
         )
         task_errors: list[ErrorDict] = []
+        check_time = arrow.utcnow()
         if task.errors:
             for error in task.errors:
                 if error["type"] == _TASK_RETRIEVAL_ERROR_TYPE:
                     # already had a timeout error, let's keep it
                     task_errors.append(error)
+                    assert "ctx" in error  # nosec
+                    assert (
+                        _TASK_RETRIEVAL_ERROR_CONTEXT_TIME_KEY in error["ctx"]
+                    )  # nosec
+                    check_time = arrow.get(
+                        error["ctx"][_TASK_RETRIEVAL_ERROR_CONTEXT_TIME_KEY]
+                    )
                     break
         if not task_errors:
             # first time we have this error
@@ -377,14 +385,22 @@ async def _handle_computational_retrieval_error(
                     msg=f"{result}",
                     type=_TASK_RETRIEVAL_ERROR_TYPE,
                     ctx={
-                        _TASK_RETRIEVAL_ERROR_CONTEXT_TIME_KEY: f"{arrow.utcnow()}",
+                        _TASK_RETRIEVAL_ERROR_CONTEXT_TIME_KEY: f"{check_time}",
                         "user_id": user_id,
                         "project_id": f"{task.project_id}",
                         "node_id": f"{task.node_id}",
                         "job_id": task.job_id,
                     },
                 )
             )
+
+        # if the task has been running for too long, we consider it failed
+        elapsed_time = arrow.utcnow() - check_time
+        if (
+            elapsed_time
+            > self.settings.COMPUTATIONAL_BACKEND_MAX_WAITING_FOR_RETRIEVING_RESULTS
+        ):
+            return RunningState.FAILED, SimcorePlatformStatus.BAD, task_errors, True
         # state is kept as STARTED so it will be retried
         return RunningState.STARTED, SimcorePlatformStatus.BAD, task_errors, False
 
@@ -457,10 +473,6 @@ async def _process_task_result(
     ) -> tuple[bool, str]:
         """Returns True and the job ID if the task was successfully processed and can be released from the Dask cluster."""
         _logger.debug("received %s result: %s", f"{task=}", f"{result=}")
-        task_final_state = RunningState.FAILED
-        simcore_platform_status = SimcorePlatformStatus.OK
-        task_errors: list[ErrorDict] = []
-        task_completed = True
 
         assert task.job_id  # nosec
         (

services/director-v2/tests/unit/with_dbs/comp_scheduler/test_scheduler_dask.py

@@ -2387,7 +2387,7 @@ async def mocked_get_task_result(job_id: str) -> TaskOutputData:
             len(expected_timeouted_tasks)
         )
         mocked_dask_client.get_task_result.reset_mock()
-        await asyncio.sleep(0.5)  # wait a bit to ensure the retry decorator has reset
+
     comp_tasks, _ = await assert_comp_tasks_and_comp_run_snapshot_tasks(
         sqlalchemy_async_engine,
         project_uuid=running_project.project.uuid,
@@ -2417,3 +2417,25 @@ async def mocked_get_task_result(job_id: str) -> TaskOutputData:
             comp_runs.c.project_uuid == f"{running_project.project.uuid}",
         ),
     )
+
+    # now we wait for the max time and the task should be marked as FAILED
+    await asyncio.sleep(with_short_max_wait_for_retrieving_results.total_seconds() + 1)
+    await scheduler_api.apply(
+        user_id=running_project.project.prj_owner,
+        project_id=running_project.project.uuid,
+        iteration=1,
+    )
+    assert mocked_dask_client.get_task_result.call_count == len(
+        expected_timeouted_tasks
+    )
+    # NOTE: we do not check all tasks here as some are depending on random others
+    # so some are ABORTED and others are FAILED depending on the random sample above
+    await assert_comp_runs(
+        sqlalchemy_async_engine,
+        expected_total=1,
+        expected_state=RunningState.FAILED,
+        where_statement=and_(
+            comp_runs.c.user_id == running_project.project.prj_owner,
+            comp_runs.c.project_uuid == f"{running_project.project.uuid}",
+        ),
+    )