handling of processed and scheduled

Author: sanderegg

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

@@ -1,7 +1,6 @@
 import logging
 from typing import Final
 
-import arrow
 import networkx as nx
 from aiopg.sa import Engine
 from fastapi import FastAPI
@@ -126,13 +125,13 @@ async def schedule_all_pipelines(app: FastAPI) -> None:
         runs_to_schedule = await CompRunsRepository.instance(db_engine).list(
             filter_by_state=SCHEDULED_STATES,
             never_scheduled=True,
-            processed_before=arrow.utcnow().datetime - SCHEDULER_INTERVAL,
+            processed_since=SCHEDULER_INTERVAL,
         )
         possibly_lost_scheduled_pipelines = await CompRunsRepository.instance(
             db_engine
         ).list(
             filter_by_state=SCHEDULED_STATES,
-            scheduled_after=SCHEDULER_INTERVAL * _LOST_TASKS_FACTOR,
+            scheduled_since=SCHEDULER_INTERVAL * _LOST_TASKS_FACTOR,
         )
         if possibly_lost_scheduled_pipelines:
             _logger.error(

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

@@ -65,7 +65,7 @@ async def list(
         filter_by_state: set[RunningState] | None = None,
         never_scheduled: bool = False,
         processed_since: datetime.timedelta | None = None,
-        scheduled_after: datetime.timedelta | None = None,
+        scheduled_since: datetime.timedelta | None = None,
     ) -> list[CompRunsAtDB]:
         conditions = []
         if filter_by_state:
@@ -81,13 +81,31 @@ async def list(
         scheduling_or_conditions = []
         if never_scheduled:
             scheduling_or_conditions.append(comp_runs.c.scheduled.is_(None))
-        if scheduled_after is not None:
-            scheduled_cutoff = arrow.utcnow().datetime - scheduled_after
-            scheduling_or_conditions.append(comp_runs.c.scheduled <= scheduled_cutoff)
+        if scheduled_since is not None:
+            # a scheduled run is a run that has been scheduled but not processed yet
+            # e.g. the processing timepoint is either null or before the scheduling timepoint
+            scheduled_cutoff = arrow.utcnow().datetime - scheduled_since
+            scheduling_filter = (
+                comp_runs.c.scheduled.is_not(None)
+                & (
+                    comp_runs.c.processed.is_(None)
+                    | (comp_runs.c.scheduled > comp_runs.c.processed)
+                )
+                & (comp_runs.c.scheduled <= scheduled_cutoff)
+            )
+            scheduling_or_conditions.append(scheduling_filter)
 
         if processed_since is not None:
+            # a processed run is a run that has been scheduled and processed
+            # and the processing timepoint is after the scheduling timepoint
             processed_cutoff = arrow.utcnow().datetime - processed_since
-            scheduling_or_conditions.append(comp_runs.c.processed <= processed_cutoff)
+            processed_filter = (
+                comp_runs.c.processed.is_not(None)
+                & (comp_runs.c.processed > comp_runs.c.scheduled)
+                & (comp_runs.c.processed <= processed_cutoff)
+            )
+
+            scheduling_or_conditions.append(processed_filter)
 
         if scheduling_or_conditions:
             conditions.append(sa.or_(*scheduling_or_conditions))

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

@@ -181,17 +181,25 @@ async def test_list(
         )
         == []
     )
-    # now we artificially change the processed time and set it 2x the scheduler interval
-    fake_processed_time = arrow.utcnow().datetime - 2 * SCHEDULER_INTERVAL
-    comp_runs_marked_as_processed = await asyncio.gather(
-        *(
-            CompRunsRepository(aiopg_engine).update(
-                user_id=comp_run.user_id,
-                project_id=comp_run.project_uuid,
-                iteration=comp_run.iteration,
-                processed=fake_processed_time,
-            )
-            for comp_run in comp_runs_marked_as_processed
+    # now we artificially change the scheduled/processed time and set it 2x the scheduler interval
+    # these are correctly processed ones, so we should get them back
+    fake_scheduled_time = arrow.utcnow().datetime - 2 * SCHEDULER_INTERVAL
+    fake_processed_time = fake_scheduled_time + 0.5 * SCHEDULER_INTERVAL
+    comp_runs_marked_as_processed = (
+        cast(  # NOTE: the cast here is ok since gather will raise if there is an error
+            list[CompRunsAtDB],
+            await asyncio.gather(
+                *(
+                    CompRunsRepository(aiopg_engine).update(
+                        user_id=comp_run.user_id,
+                        project_id=comp_run.project_uuid,
+                        iteration=comp_run.iteration,
+                        scheduled=fake_scheduled_time,
+                        processed=fake_processed_time,
+                    )
+                    for comp_run in comp_runs_marked_as_processed
+                )
+            ),
         )
     )
     # now we should get them
@@ -200,8 +208,48 @@ async def test_list(
             never_scheduled=False, processed_since=SCHEDULER_INTERVAL
         ),
         key=lambda x: x.iteration,
+    ) == sorted(comp_runs_marked_as_processed, key=lambda x: x.iteration)
+
+    # now some of them were never processed (e.g. processed time is either null or before schedule time)
+    comp_runs_waiting_for_processing_or_never_processed = random.sample(
+        comp_runs_marked_as_processed, k=6
+    )
+    comp_runs_marked_as_processed = [
+        r
+        for r in comp_runs_marked_as_processed
+        if r not in comp_runs_waiting_for_processing_or_never_processed
+    ]
+    # now we artificially change the processed time to be before the scheduled time
+    comp_runs_waiting_for_processing_or_never_processed = cast(
+        list[CompRunsAtDB],
+        await asyncio.gather(
+            *(
+                CompRunsRepository(aiopg_engine).update(
+                    user_id=comp_run.user_id,
+                    project_id=comp_run.project_uuid,
+                    iteration=comp_run.iteration,
+                    scheduled=fake_processed_time,  # NOTE: we invert here the timings
+                    processed=random.choice([fake_scheduled_time, None]),  # noqa: S311
+                )
+                for comp_run in comp_runs_waiting_for_processing_or_never_processed
+            )
+        ),
+    )
+    # so the processed ones shall remain
+    assert sorted(
+        await CompRunsRepository(aiopg_engine).list(
+            never_scheduled=False, processed_since=SCHEDULER_INTERVAL
+        ),
+        key=lambda x: x.iteration,
+    ) == sorted(comp_runs_marked_as_processed, key=lambda x: x.iteration)
+    # the ones waiting for scheduling now
+    assert sorted(
+        await CompRunsRepository(aiopg_engine).list(
+            never_scheduled=False, scheduled_since=SCHEDULER_INTERVAL
+        ),
+        key=lambda x: x.iteration,
     ) == sorted(
-        comp_runs_marked_as_processed, key=lambda x: cast(CompRunsAtDB, x).iteration
+        comp_runs_waiting_for_processing_or_never_processed, key=lambda x: x.iteration
     )
 
 

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

@@ -178,6 +178,7 @@ async def test_schedule_all_pipelines(
     assert comp_run.metadata == run_metadata
     assert comp_run.result is RunningState.PUBLISHED
     assert comp_run.scheduled is not None
+    assert comp_run.processed is None
     start_schedule_time = comp_run.scheduled
     start_modified_time = comp_run.modified
 
@@ -186,15 +187,18 @@ async def test_schedule_all_pipelines(
     scheduler_rabbit_client_parser.assert_not_called()
     comp_runs = await assert_comp_runs(sqlalchemy_async_engine, expected_total=1)
     comp_run = comp_runs[0]
+    assert comp_run.scheduled
     assert comp_run.scheduled == start_schedule_time, "scheduled time changed!"
     assert comp_run.cancelled is None
     assert comp_run.modified == start_modified_time
 
-    # once the worker is done, the schedule time is set back to None
-    await CompRunsRepository(aiopg_engine).mark_as_processed(
+    # to simulate that the worker did its job we will set times in the past
+    await CompRunsRepository(aiopg_engine).update(
         user_id=comp_run.user_id,
         project_id=comp_run.project_uuid,
         iteration=comp_run.iteration,
+        scheduled=comp_run.scheduled - 1.5 * SCHEDULER_INTERVAL,
+        processed=comp_run.scheduled - 1.1 * SCHEDULER_INTERVAL,
     )
 
     # now we schedule a pipeline again, but we wait for the scheduler interval to pass

services/director-v2/tests/unit/with_dbs/conftest.py

@@ -123,7 +123,7 @@ async def _(
                 ),
                 "node_class": to_node_class(node_data.key),
                 "internal_id": internal_id + 1,
-                "submit": datetime.datetime.now(),
+                "submit": datetime.datetime.now(datetime.UTC),
                 "job_id": generate_dask_job_id(
                     service_key=node_data.key,
                     service_version=node_data.version,