Issue #633 move job actions (e.g. result download) after starting new jobs

(this is a squashed commit based on #711)

Author: HansVRP

openeo/extra/job_management/__init__.py

@@ -19,6 +19,7 @@
     Mapping,
     NamedTuple,
     Optional,
+    Tuple,
     Union,
 )
 
@@ -517,7 +518,7 @@ def _job_update_loop(
         stats = stats if stats is not None else collections.defaultdict(int)
 
         with ignore_connection_errors(context="get statuses"):
-            self._track_statuses(job_db, stats=stats)
+            jobs_done, jobs_error, jobs_cancel = self._track_statuses(job_db, stats=stats)
             stats["track_statuses"] += 1
 
         not_started = job_db.get_by_status(statuses=["not_started"], max=200).copy()
@@ -540,6 +541,17 @@ def _job_update_loop(
                         stats["job_db persist"] += 1
                         total_added += 1
 
+        # Act on jobs
+        for job, row in jobs_done:
+            self.on_job_done(job, row)
+
+        for job, row in jobs_error:
+            self.on_job_error(job, row)
+
+        for job, row in jobs_cancel:
+            self.on_job_cancel(job, row)
+
+
     def _launch_job(self, start_job, df, i, backend_name, stats: Optional[dict] = None):
         """Helper method for launching jobs
 
@@ -696,14 +708,19 @@ def ensure_job_dir_exists(self, job_id: str) -> Path:
         if not job_dir.exists():
             job_dir.mkdir(parents=True)
 
-    def _track_statuses(self, job_db: JobDatabaseInterface, stats: Optional[dict] = None):
+    def _track_statuses(self, job_db: JobDatabaseInterface, stats: Optional[dict] = None) -> Tuple[List, List, List]:
         """
         Tracks status (and stats) of running jobs (in place).
         Optionally cancels jobs when running too long.
         """
         stats = stats if stats is not None else collections.defaultdict(int)
 
         active = job_db.get_by_status(statuses=["created", "queued", "running"]).copy()
+
+        jobs_done = []
+        jobs_error = []
+        jobs_cancel = []
+
         for i in active.index:
             job_id = active.loc[i, "id"]
             backend_name = active.loc[i, "backend_name"]
@@ -722,20 +739,20 @@ def _track_statuses(self, job_db: JobDatabaseInterface, stats: Optional[dict] =
 
                 if new_status == "finished":
                     stats["job finished"] += 1
-                    self.on_job_done(the_job, active.loc[i])
+                    jobs_done.append((the_job, active.loc[i]))
 
                 if previous_status != "error" and new_status == "error":
                     stats["job failed"] += 1
-                    self.on_job_error(the_job, active.loc[i])
+                    jobs_error.append((the_job, active.loc[i]))
+
+                if new_status == "canceled":
+                    stats["job canceled"] += 1
+                    jobs_cancel.append((the_job, active.loc[i]))
 
                 if previous_status in {"created", "queued"} and new_status == "running":
                     stats["job started running"] += 1
                     active.loc[i, "running_start_time"] = rfc3339.utcnow()
 
-                if new_status == "canceled":
-                    stats["job canceled"] += 1
-                    self.on_job_cancel(the_job, active.loc[i])
-
                 if self._cancel_running_job_after and new_status == "running":
                     if  (not active.loc[i, "running_start_time"] or pd.isna(active.loc[i, "running_start_time"])):
                         _log.warning(
@@ -763,6 +780,8 @@ def _track_statuses(self, job_db: JobDatabaseInterface, stats: Optional[dict] =
         stats["job_db persist"] += 1
         job_db.persist(active)
 
+        return jobs_done, jobs_error, jobs_cancel
+
 
 def _format_usage_stat(job_metadata: dict, field: str) -> str:
     value = deep_get(job_metadata, "usage", field, "value", default=0)