improve error messaging on processing futures

Author: HansVRP

openeo/extra/job_management/__init__.py

@@ -677,36 +677,31 @@ def _process_threadworker_updates(
         """
         # Retrieve completed task results immediately
         results, _ = worker_pool.process_futures(timeout=0)
-        if not isinstance(results, list):
-            raise TypeError(f"Expected list of TaskResult, got {results}")
 
         # Collect update dicts
         updates: List[Dict[str, Any]] = []
         for res in results:
             # Process database updates
             if res.db_update:
                 try:
-                    if 'id' in res.db_update or 'df_idx' in res.db_update:
-                        raise KeyError("db_update must not override 'id' or 'df_idx'")
                     updates.append({
                         'id': res.job_id,
                         'df_idx': res.df_idx,
                         **res.db_update,
                     })
                 except Exception as e:
-                    _log.error(f"Skipping invalid db_update for job '{res.job_id}': {e}")
+                    _log.error(f"Skipping invalid db_update '{res.db_update}' for job '{res.job_id}': {e}", )
 
             # Process stats updates
             if res.stats_update:
-                for key, val in res.stats_update.items():
-                    try:
+                try:
+                    for key, val in res.stats_update.items():
                         count = int(val)
                         stats[key] = stats.get(key, 0) + count
-                    except Exception:
-                        _log.error(
-                            f"Skipping invalid stats_update for job '{res.job_id}': "
-                            f"key={key!r}, val={val!r}"
-                        )
+                except Exception as e:
+                    _log.error(
+                        f"Skipping invalid stats_update {res.stats_update} for job '{res.job_id}': {e}"
+                    )
 
         # No valid updates: nothing to persist
         if not updates:
@@ -982,7 +977,7 @@ def get_by_status(self, statuses, max=None) -> pd.DataFrame:
 
     def _merge_into_df(self, df: pd.DataFrame):
         if self._df is not None:
-            self._df.update(df, overwrite=True) #TODO index is not consistent so this creates an issue. --> when we get a row, we need to give the right index to the row. Best solution when creating a task; pass this one along
+            self._df.update(df, overwrite=True) 
         else:
             self._df = df
 

tests/extra/job_management/test_job_management.py

@@ -803,7 +803,7 @@ def test_no_results_leaves_db_and_stats_untouched(self, tmp_path, caplog):
         assert stats == {}
 
 
-    def test_logs_on_invalid_db_update(self, tmp_path, caplog):
+    def test_logs_on_invalid_update(self, tmp_path, caplog):
         pool = _JobManagerWorkerThreadPool(max_workers=2)
         stats = collections.defaultdict(int)
 
@@ -812,7 +812,7 @@ class BadTask:
             job_id = "bad-task"
             df_idx = 0
             db_update = "invalid"  # invalid
-            stats_update = None
+            stats_update = "a"
 
             def execute(self):
                 return self
@@ -834,7 +834,8 @@ def execute(self):
         assert stats == {}
 
         # Assert log about invalid db update
-        assert any("Skipping invalid db_update for job" in msg for msg in caplog.messages)
+        assert any("Skipping invalid db_update" in msg for msg in caplog.messages)
+        assert any("Skipping invalid stats_update" in msg for msg in caplog.messages)
 
 JOB_DB_DF_BASICS = pd.DataFrame(
     {

tests/extra/job_management/test_thread_worker.py

@@ -84,7 +84,7 @@ def execute(self) -> _TaskResult:
             raise ValueError("Oh no!")
         return _TaskResult(
             job_id=self.job_id,
-            df_idx= self.df_idx,
+            df_idx=self.df_idx,
             db_update={"status": "dummified"},
             stats_update={"dummy": 1},
         )
@@ -104,7 +104,7 @@ def execute(self) -> _TaskResult:
             raise TimeoutError("Waiting for event timed out")
         if not self.success:
             raise ValueError("Oh no!")
-        return _TaskResult(job_id=self.job_id, df_idx= self.df_idx, db_update={"status": "all fine"})
+        return _TaskResult(job_id=self.job_id, df_idx=self.df_idx, db_update={"status": "all fine"})
 
 
 
@@ -154,16 +154,16 @@ def test_submit_and_process_with_error(self, worker_pool):
         assert remaining == 0
 
     def test_submit_and_process_iterative(self, worker_pool):
-        worker_pool.submit_task(NopTask(job_id="j-1", df_idx= 1))
+        worker_pool.submit_task(NopTask(job_id="j-1", df_idx=1))
         results, remaining = worker_pool.process_futures(timeout=1)
-        assert results == [_TaskResult(job_id="j-1", df_idx= 1)]
+        assert results == [_TaskResult(job_id="j-1", df_idx=1)]
         assert remaining == 0
 
         # Add some more
-        worker_pool.submit_task(NopTask(job_id="j-22", df_idx= 22))
-        worker_pool.submit_task(NopTask(job_id="j-222", df_idx= 222))
+        worker_pool.submit_task(NopTask(job_id="j-22", df_idx=22))
+        worker_pool.submit_task(NopTask(job_id="j-222", df_idx=222))
         results, remaining = worker_pool.process_futures(timeout=1)
-        assert results == [_TaskResult(job_id="j-22", df_idx= 22), _TaskResult(job_id="j-222",  df_idx= 222)]
+        assert results == [_TaskResult(job_id="j-22", df_idx=22), _TaskResult(job_id="j-222",  df_idx=222)]
         assert remaining == 0
 
     def test_submit_multiple_simple(self, worker_pool):