MultiBackendJobManager: refresh bearer before _JobStartTask #817

Author: soxofaan

CHANGELOG.md

@@ -17,6 +17,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Proactively refresh access/bearer token in `MultiBackendJobManager` before launching a job start thread ([#817](https://github.com/Open-EO/openeo-python-client/issues/817))
+
 
 ## [0.45.0] - 2025-09-17
 

openeo/extra/job_management/__init__.py

@@ -244,6 +244,8 @@ def __init__(
         )
         self._thread = None
         self._worker_pool = None
+        # Generic cache
+        self._cache = {}
 
     def add_backend(
         self,
@@ -650,6 +652,8 @@ def _launch_job(self, start_job, df, i, backend_name, stats: Optional[dict] = No
                         # 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,
@@ -670,6 +674,19 @@ def _launch_job(self, start_job, df, i, backend_name, stats: Optional[dict] = No
                 df.loc[i, "status"] = "skipped"
                 stats["start_job skipped"] += 1
 
+    def _refresh_bearer_token(self, connection: Connection, *, max_age: float = 60):
+        """
+        Helper to proactively refresh access token of connection
+        (but not too often, based on `max_age`).
+        """
+        # TODO: be smarter about timing, e.g. by inspecting expiry of current token?
+        now = time.time()
+        key = f"connection-{id(connection)}-refresh-time"
+        if self._cache.get(key, 0) + max_age < now:
+            refreshed = connection.try_access_token_refresh()
+            if refreshed:
+                self._cache[key] = now
+
     def _process_threadworker_updates(
         self,
         worker_pool: _JobManagerWorkerThreadPool,

openeo/rest/auth/testing.py

@@ -143,6 +143,11 @@ def token_callback_resource_owner_password_credentials(self, params: dict, conte
         assert params["scope"] == self.expected_fields["scope"]
         return self._build_token_response()
 
+    def token_callback_block_400(self, params: dict, context):
+        """Failing callback with 400 Bad Request"""
+        context.status_code = 400
+        return "block_400"
+
     def device_code_callback(self, request: requests_mock.request._RequestObjectProxy, context):
         params = self._get_query_params(query=request.text)
         assert params["client_id"] == self.expected_client_id

openeo/rest/connection.py

@@ -665,6 +665,28 @@ def authenticate_bearer_token(self, bearer_token: str) -> Connection:
         self._oidc_auth_renewer = None
         return self
 
+    def try_access_token_refresh(self, *, reason: Optional[str] = None) -> bool:
+        """
+        Try to get a fresh access token if possible.
+        Returns whether a new access token was obtained.
+        """
+        reason = f" Reason: {reason}" if reason else ""
+        if isinstance(self.auth, OidcBearerAuth) and self._oidc_auth_renewer:
+            try:
+                self._authenticate_oidc(
+                    authenticator=self._oidc_auth_renewer,
+                    provider_id=self._oidc_auth_renewer.provider_info.id,
+                    store_refresh_token=False,
+                    oidc_auth_renewer=self._oidc_auth_renewer,
+                )
+                _log.info(f"Obtained new access token (grant {self._oidc_auth_renewer.grant_type!r}).{reason}")
+                return True
+            except OpenEoClientException as auth_exc:
+                _log.error(
+                    f"Failed to obtain new access token (grant {self._oidc_auth_renewer.grant_type!r}): {auth_exc!r}.{reason}"
+                )
+        return False
+
     def request(
         self,
         method: str,
@@ -690,24 +712,11 @@ def _request():
                 api_exc.http_status_code in {HTTP_401_UNAUTHORIZED, HTTP_403_FORBIDDEN}
                 and api_exc.code == "TokenInvalid"
             ):
-                # Auth token expired: can we refresh?
-                if isinstance(self.auth, OidcBearerAuth) and self._oidc_auth_renewer:
-                    msg = f"OIDC access token expired ({api_exc.http_status_code} {api_exc.code})."
-                    try:
-                        self._authenticate_oidc(
-                            authenticator=self._oidc_auth_renewer,
-                            provider_id=self._oidc_auth_renewer.provider_info.id,
-                            store_refresh_token=False,
-                            oidc_auth_renewer=self._oidc_auth_renewer,
-                        )
-                        _log.info(f"{msg} Obtained new access token (grant {self._oidc_auth_renewer.grant_type!r}).")
-                    except OpenEoClientException as auth_exc:
-                        _log.error(
-                            f"{msg} Failed to obtain new access token (grant {self._oidc_auth_renewer.grant_type!r}): {auth_exc!r}."
-                        )
-                    else:
-                        # Retry request.
-                        return _request()
+                # Retry if we can refresh the access token
+                if self.try_access_token_refresh(
+                    reason=f"OIDC access token expired ({api_exc.http_status_code} {api_exc.code})."
+                ):
+                    return _request()
             raise
 
     def describe_account(self) -> dict: