Retry starts after concurrent job limits

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Retry job starts in a later job manager cycle when a backend reports `ConcurrentJobLimit`. ([#838](https://github.com/Open-EO/openeo-python-client/issues/838))
 - `DataCube.resample_spatial()` now supports parameterized `resolution` and `projection` arguments.  ([#897](https://github.com/Open-EO/openeo-python-client/issues/897))
 - Sanitize asset download filenames (e.g. strip slashes, (semi)colon, hash, ...), instead of blindly using the asset key as filename. ([#820](https://github.com/Open-EO/openeo-python-client/issues/820))
 - Support parameters in `DataCube` apply- and band-math operations ([#903](https://github.com/Open-EO/openeo-python-client/issues/903))

openeo/extra/job_management/_manager.py

@@ -30,9 +30,9 @@
 from openeo import BatchJob, Connection
 from openeo.extra.job_management._interface import JobDatabaseInterface
 from openeo.extra.job_management._thread_worker import (
+    _JobDownloadTask,
     _JobManagerWorkerThreadPool,
     _JobStartTask,
-    _JobDownloadTask
 )
 from openeo.rest import OpenEoApiError
 from openeo.rest.auth.auth import BearerAuth
@@ -512,16 +512,11 @@ def run_jobs(
 
         self._worker_pool = _JobManagerWorkerThreadPool()
 
-
-        while (
-            sum(
-                job_db.count_by_status(
-                    statuses=["not_started", "created", "queued_for_start", "queued", "running"]
-                ).values()) > 0
-
-            or (self._worker_pool is not None and self._worker_pool.has_unprocessed_tasks()) 
-
-        ):
+        while sum(
+            job_db.count_by_status(
+                statuses=["not_started", "created", "queued_for_start", "queued", "running"]
+            ).values()
+        ) > 0 or (self._worker_pool is not None and self._worker_pool.has_unprocessed_tasks()):
             self._job_update_loop(job_db=job_db, start_job=start_job, stats=stats)
             stats["run_jobs loop"] += 1
 
@@ -530,8 +525,6 @@ def run_jobs(
             time.sleep(self.poll_sleep)
             stats["sleep"] += 1
 
-
-
         self._worker_pool.shutdown()
         self._worker_pool = None
 
@@ -554,6 +547,11 @@ def _job_update_loop(
             jobs_done, jobs_error, jobs_cancel = self._track_statuses(job_db, stats=stats)
             stats["track_statuses"] += 1
 
+        # A start task that hits the backend's concurrent-job limit moves the
+        # job back to "created". Re-submit such jobs here, in a later manager
+        # cycle, instead of failing them or blocking a worker with sleep/retry.
+        self._queue_created_jobs(job_db=job_db, stats=stats)
+
         not_started = job_db.get_by_status(statuses=["not_started"], max=200).copy()
         if len(not_started) > 0:
             # Check number of jobs queued/running at each backend
@@ -581,7 +579,7 @@ def _job_update_loop(
 
         if self._worker_pool is not None:
             self._process_threadworker_updates(worker_pool=self._worker_pool, job_db=job_db, stats=stats)
-            
+
 
         # TODO: move this back closer to the `_track_statuses` call above, once job done/error handling is also handled in threads?
         for job, row in jobs_done:
@@ -648,29 +646,53 @@ def _launch_job(self, start_job, df, i, backend_name, stats: Optional[dict] = No
                     if status == "created":
                         # start job if not yet done by callback
                         try:
-                            job_con = job.connection
-                            # Proactively refresh bearer token (because task in thread will not be able to do that)
-                            self._refresh_bearer_token(connection=job_con)
-                            task = _JobStartTask(
-                                root_url=job_con.root_url,
-                                bearer_token=job_con.auth.bearer if isinstance(job_con.auth, BearerAuth) else None,
-                                job_id=job.job_id,
-                                df_idx=i,
-                            )
-                            _log.info(f"Submitting task {task} to thread pool")
-                            self._worker_pool.submit_task(task=task, pool_name="job_start")
-
-                            stats["job_queued_for_start"] += 1
+                            self._submit_job_start(job=job, df_idx=i, stats=stats)
                             df.loc[i, "status"] = "queued_for_start"
                         except OpenEoApiError as e:
-                            _log.info(f"Failed submitting task {task} to thread pool with error: {e}")
+                            _log.info(f"Failed to submit start task for job {job.job_id!r} with error: {e}")
                             df.loc[i, "status"] = "queued_for_start_failed"
                             stats["job queued for start failed"] += 1
             else:
                 # TODO: what is this "skipping" about actually?
                 df.loc[i, "status"] = "skipped"
                 stats["start_job skipped"] += 1
 
+    def _submit_job_start(self, job: BatchJob, df_idx: int, stats: dict) -> None:
+        """Submit a non-blocking start task for an already created batch job."""
+        job_con = job.connection
+        # Proactively refresh bearer token because the worker cannot refresh it.
+        self._refresh_bearer_token(connection=job_con)
+        task = _JobStartTask(
+            root_url=job_con.root_url,
+            bearer_token=job_con.auth.bearer if isinstance(job_con.auth, BearerAuth) else None,
+            job_id=job.job_id,
+            df_idx=df_idx,
+        )
+        _log.info(f"Submitting task {task} to thread pool")
+        self._worker_pool.submit_task(task=task, pool_name="job_start")
+        stats["job_queued_for_start"] += 1
+
+    def _queue_created_jobs(self, *, job_db: JobDatabaseInterface, stats: dict) -> None:
+        """Re-submit created jobs whose previous start attempt was deferred."""
+        created = job_db.get_by_status(statuses=["created"]).copy()
+        if created.empty:
+            return
+
+        for i in created.index:
+            job_id = created.loc[i, "id"]
+            backend_name = created.loc[i, "backend_name"]
+            try:
+                job = self._get_connection(backend_name).job(job_id)
+                self._submit_job_start(job=job, df_idx=i, stats=stats)
+                created.loc[i, "status"] = "queued_for_start"
+            except OpenEoApiError as e:
+                _log.info(f"Failed to re-submit start task for job {job_id!r} with error: {e}")
+                created.loc[i, "status"] = "queued_for_start_failed"
+                stats["job queued for start failed"] += 1
+
+        job_db.persist(created)
+        stats["job_db persist"] += 1
+
     def _refresh_bearer_token(self, connection: Connection, *, max_age: float = 60) -> None:
         """
         Helper to proactively refresh the bearer (access) token of the connection
@@ -758,19 +780,19 @@ def on_job_done(self, job: BatchJob, row):
             #Proactively refresh bearer token
             job_con =  job.connection
             self._refresh_bearer_token(connection=job_con)
-            
+
             task = _JobDownloadTask(
                 job_id=job.job_id,
-                df_idx=row.name, 
+                df_idx=row.name,
                 root_url=job_con.root_url,
                 bearer_token=job_con.auth.bearer if isinstance(job_con.auth, BearerAuth) else None,
                 download_dir=job_dir,
             )
             _log.info(f"Submitting download task {task} to download thread pool")
-            
+
             if self._worker_pool is None:
                 self._worker_pool = _JobManagerWorkerThreadPool()
-                
+
             self._worker_pool.submit_task(task=task, pool_name="job_download")
 
     def on_job_error(self, job: BatchJob, row):

openeo/extra/job_management/_thread_worker.py

@@ -3,16 +3,17 @@
 """
 
 import concurrent.futures
+import json
 import logging
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
-from typing import Any, Dict, List, Optional, Tuple, Union
 from pathlib import Path
+from typing import Any, Dict, List, Optional, Tuple, Union
 
-import json
 import urllib3.util
 
 import openeo
+from openeo.rest import OpenEoApiError
 from openeo.utils.http import HTTP_429_TOO_MANY_REQUESTS, retry_configuration
 
 _log = logging.getLogger(__name__)
@@ -132,16 +133,34 @@ def execute(self) -> _TaskResult:
                 db_update={"status": "queued"},
                 stats_update={"job start": 1},
             )
+        except OpenEoApiError as e:
+            if e.code == "ConcurrentJobLimit":
+                _log.warning(
+                    "Backend concurrent job limit reached while starting job %r; "
+                    "scheduling another attempt in a later job manager cycle.",
+                    self.job_id,
+                )
+                return _TaskResult(
+                    job_id=self.job_id,
+                    df_idx=self.df_idx,
+                    db_update={"status": "created"},
+                    stats_update={"job start retry": 1},
+                )
+            return self._start_failure_result(e)
         except Exception as e:
-            _log.error(f"Failed to start job {self.job_id!r}: {e!r}")
-            # TODO: more insights about the failure (e.g. the exception) are just logged, but lost from the result
-            return _TaskResult(
-                job_id=self.job_id,
-                df_idx=self.df_idx,
-                db_update={"status": "start_failed"},
-                stats_update={"start_job error": 1},
-            )
-
+            return self._start_failure_result(e)
+
+    def _start_failure_result(self, error: Exception) -> _TaskResult:
+        """Build the terminal result for a non-retryable job start failure."""
+        _log.error(f"Failed to start job {self.job_id!r}: {error!r}")
+        # TODO: more insights about the failure (e.g. the exception) are just logged, but lost from the result
+        return _TaskResult(
+            job_id=self.job_id,
+            df_idx=self.df_idx,
+            db_update={"status": "start_failed"},
+            stats_update={"start_job error": 1},
+        )
+
 @dataclass(frozen=True)
 class _JobDownloadTask(ConnectedTask):
     """
@@ -159,7 +178,7 @@ def execute(self) -> _TaskResult:
 
             # Count assets (files to download)
             file_count = len(job.get_results().get_assets())
-            
+
             # Download results
             job.get_results().download_files(target=self.download_dir)
 
@@ -168,7 +187,7 @@ def execute(self) -> _TaskResult:
             metadata_path = self.download_dir / f"job_{self.job_id}.json"
             with metadata_path.open("w", encoding="utf-8") as f:
                 json.dump(job_metadata, f, ensure_ascii=False)
-            
+
             _log.info(f"Job {self.job_id!r} results downloaded successfully")
             return _TaskResult(
                 job_id=self.job_id,
@@ -184,7 +203,8 @@ def execute(self) -> _TaskResult:
                 db_update={},
                 stats_update={"job download error": 1, "files downloaded": 0},
             )
-
+
+
 class _TaskThreadPool:
     """
     Thread pool-based worker that manages the execution of asynchronous tasks.
@@ -261,11 +281,11 @@ def process_futures(self, timeout: Union[float, None] = 0) -> Tuple[List[_TaskRe
         self._total_processed += len(results)
 
         return results, len(to_keep)
-    
+
     def get_unprocessed_count(self) -> int:
         """Get number of tasks that haven't been processed yet."""
         return self._total_submitted - self._total_processed
-    
+
     def has_unprocessed_tasks(self) -> bool:
         """Check if there are tasks that haven't been processed yet."""
         return self._total_submitted > self._total_processed
@@ -280,47 +300,47 @@ class _JobManagerWorkerThreadPool:
     """
     Generic wrapper that manages multiple thread pools with a dict.
     """
-    
+
     def __init__(self, pool_configs: Optional[Dict[str, int]] = None):
         self._pools: Dict[str, _TaskThreadPool] = {}
         self._pool_configs = pool_configs or {}
 
     def list_pools(self) -> List[str]:
         """List all active pool names."""
         return list(self._pools.keys())
-    
+
     def submit_task(self, task: Task, pool_name: str = "default") -> None:
         if pool_name not in self._pools:
             max_workers = self._pool_configs.get(pool_name, 2)
             self._pools[pool_name] = _TaskThreadPool(max_workers=max_workers)
             _log.info(f"Created pool '{pool_name}' with {max_workers} workers")
-        
+
         self._pools[pool_name].submit_task(task)
 
     def get_unprocessed_counts(self) -> Dict[str, int]:
         """Get unprocessed (submitted but not processed) task counts per pool."""
         return {name: pool.get_unprocessed_count() for name, pool in self._pools.items()}
-    
+
     def has_unprocessed_tasks(self) -> bool:
         """Check if any pool has unprocessed (submitted but not processed) tasks."""
         return any(pool.has_unprocessed_tasks() for pool in self._pools.values())
-    
+
     def process_futures(self, timeout: Union[float, None] = 0) -> Tuple[List[_TaskResult], Dict[str, int]]:
         """
         Process updates from ALL pools.
         Returns: (all_results, dict of remaining tasks per pool)
         """
         all_results = []
         to_keep = {}
-        
+
         for pool_name, pool in self._pools.items():
             results, remaining = pool.process_futures(timeout)
             all_results.extend(results)
 
             to_keep[pool_name] = remaining
 
         return all_results, to_keep
-    
+
     def shutdown(self, pool_name: Optional[str] = None) -> None:
         """
         Shutdown pools.
@@ -334,6 +354,3 @@ def shutdown(self, pool_name: Optional[str] = None) -> None:
             for pool_name, pool in list(self._pools.items()):
                 pool.shutdown()
                 del self._pools[pool_name]
-
-
-

tests/extra/job_management/test_concurrent_job_limit.py

@@ -0,0 +1,31 @@
+import logging
+
+from openeo.extra.job_management._thread_worker import _JobStartTask, _TaskResult
+from openeo.rest._testing import DummyBackend
+
+
+def test_concurrent_job_limit_is_retryable(requests_mock, caplog):
+    caplog.set_level(logging.WARNING)
+    backend = DummyBackend.at_url("https://foo.test", requests_mock=requests_mock)
+    job = backend.connection.create_job(process_graph={})
+    backend.setup_job_start_failure(
+        status_code=400,
+        response_body={"code": "ConcurrentJobLimit", "message": "Concurrent job limit reached."},
+    )
+
+    result = _JobStartTask(
+        job_id=job.job_id,
+        df_idx=0,
+        root_url=backend.connection.root_url,
+    ).execute()
+
+    assert result == _TaskResult(
+        job_id="job-000",
+        df_idx=0,
+        db_update={"status": "created"},
+        stats_update={"job start retry": 1},
+    )
+    assert caplog.messages == [
+        "Backend concurrent job limit reached while starting job 'job-000'; "
+        "scheduling another attempt in a later job manager cycle."
+    ]