fix(oauth): bound token refresh retries for REFRESH_TOKEN_THEN_RETRY

airbyte_cdk/sources/streams/http/http_client.py

@@ -175,6 +175,7 @@ def __init__(
             self._backoff_strategies = [DefaultBackoffStrategy()]
         self._error_message_parser = error_message_parser or JsonErrorMessageParser()
         self._request_attempt_count: Dict[requests.PreparedRequest, int] = {}
+        self._token_refresh_outcomes: Dict[requests.PreparedRequest, bool] = {}
         self._disable_retries = disable_retries
         self._message_repository = message_repository
         self._authenticator_update_failed = False
@@ -537,6 +538,26 @@ def _evict_key(self, prepared_request: requests.PreparedRequest) -> None:
         """
         if prepared_request in self._request_attempt_count:
             del self._request_attempt_count[prepared_request]
+        self._token_refresh_outcomes.pop(prepared_request, None)
+
+    def _auth_header_changed_since(self, request: requests.PreparedRequest) -> bool:
+        """Whether the authenticator's current Authorization header differs from the one the request was sent with."""
+        # request.headers is None on an unprepared request
+        sent = request.headers.get("Authorization") if request.headers else None
+        if not sent or not hasattr(self._session.auth, "get_auth_header"):
+            return False
+        current = self._session.auth.get_auth_header().get("Authorization")  # type: ignore[union-attr]
+        return current is not None and current != sent
+
+    @staticmethod
+    def _is_token_endpoint_rejection(error: requests.exceptions.RequestException) -> bool:
+        """Whether the token endpoint answered with a 4xx other than 429, i.e. it rejected the credentials rather than failing transiently."""
+        response = error.response
+        return (
+            response is not None
+            and 400 <= response.status_code < 500
+            and response.status_code != 429
+        )
 
     def _handle_error_resolution(
         self,
@@ -575,20 +596,81 @@ def _handle_error_resolution(
             # backoff retry loop. Adding `\n` to the message and ignore 'end' ensure that few messages are printed at the same time.
             print(f"{message}\n", end="", flush=True)
 
-        # Handle REFRESH_TOKEN_THEN_RETRY: Force refresh the OAuth token before retry
-        # This is useful when the API returns 401 but the stored token expiry hasn't been reached yet
-        # Only OAuth authenticators have refresh_and_set_access_token method
-        # Non-OAuth auth types (e.g., BearerAuthenticator) will fall through to normal retry
+        # Handle REFRESH_TOKEN_THEN_RETRY: force refresh the OAuth token before retrying,
+        # at most once per request. A config error from the refresh (e.g. bad credentials,
+        # including a 4xx rejection from the token endpoint) or a request rejected again
+        # after a refresh attempt fails fast instead of refreshing in a loop; the failure
+        # type reflects whether the refresh succeeded. Transient refresh failures (network
+        # errors, 5xx, 429) keep the retry transient. Non-OAuth auth types (e.g.,
+        # BearerAuthenticator) fall through to normal retry.
         if error_resolution.response_action == ResponseAction.REFRESH_TOKEN_THEN_RETRY:
+            status = (
+                f"status code '{response.status_code}'"
+                if response is not None
+                else f"exception '{exc}'"
+            )
+            if request in self._token_refresh_outcomes:
+                refreshed = self._token_refresh_outcomes[request]
+                self._evict_key(request)
+                if refreshed:
+                    internal_message = f"'{request.method}' request to '{request.url}' was rejected with {status} again after the OAuth token was refreshed; not refreshing again."
+                    failure_type = FailureType.config_error
+                    message = "Refreshed OAuth access token is rejected by the API."
+                else:
+                    internal_message = f"'{request.method}' request to '{request.url}' was rejected with {status} and the OAuth token could not be refreshed; not refreshing again."
+                    failure_type = FailureType.transient_error
+                    message = (
+                        "API rejects the current OAuth access token and the token refresh failed."
+                    )
+                if error_resolution.error_message:
+                    internal_message += (
+                        f" Error handler message: '{error_resolution.error_message}'"
+                    )
+                self._logger.error(internal_message)
+                raise AirbyteTracedException(
+                    internal_message=internal_message,
+                    message=message,
+                    failure_type=failure_type,
+                )
             if (
                 hasattr(self._session, "auth")
                 and self._session.auth is not None
                 and hasattr(self._session.auth, "refresh_and_set_access_token")
             ):
+                self._token_refresh_outcomes[request] = False
                 try:
-                    self._session.auth.refresh_and_set_access_token()  # type: ignore[union-attr]
-                    self._logger.info(
-                        "Refreshed OAuth token due to REFRESH_TOKEN_THEN_RETRY response action"
+                    if self._auth_header_changed_since(request):
+                        self._logger.info(
+                            "OAuth token was already replaced since this request was sent; retrying with the current token without refreshing again."
+                        )
+                    else:
+                        self._session.auth.refresh_and_set_access_token()  # type: ignore[union-attr]
+                        self._logger.info(
+                            "Refreshed OAuth token due to REFRESH_TOKEN_THEN_RETRY response action"
+                        )
+                    self._token_refresh_outcomes[request] = True
+                except AirbyteTracedException as refresh_error:
+                    if refresh_error.failure_type == FailureType.config_error:
+                        self._evict_key(request)
+                        raise
+                    self._logger.warning(
+                        f"Failed to refresh OAuth token: {refresh_error}. Proceeding with retry using existing token."
+                    )
+                except requests.exceptions.RequestException as refresh_error:
+                    if self._is_token_endpoint_rejection(refresh_error):
+                        self._evict_key(request)
+                        internal_message = (
+                            f"'{request.method}' request to '{request.url}' was rejected with {status} and the OAuth token endpoint "
+                            f"rejected the refresh request: {refresh_error}"
+                        )
+                        self._logger.error(internal_message)
+                        raise AirbyteTracedException(
+                            internal_message=internal_message,
+                            message="OAuth token refresh request is rejected by the token endpoint.",
+                            failure_type=FailureType.config_error,
+                        ) from refresh_error
+                    self._logger.warning(
+                        f"Failed to refresh OAuth token: {refresh_error}. Proceeding with retry using existing token."
                     )
                 except Exception as refresh_error:
                     self._logger.warning(

airbyte_cdk/sources/streams/http/requests_native_auth/abstract_oauth.py

@@ -61,7 +61,7 @@ class AbstractOauth2Authenticator(AuthBase):
     # through the connector config. Without this lock, concurrent refresh attempts can cause race
     # conditions where one stream successfully refreshes the token while others fail because the
     # refresh token has been invalidated (especially for single-use refresh tokens).
-    _token_refresh_lock: threading.Lock = threading.Lock()
+    _token_refresh_lock = threading.RLock()
 
     def __init__(
         self,
@@ -124,16 +124,36 @@ def get_access_token(self) -> str:
 
         return self.access_token
 
+    def _current_access_token_or_none(self) -> Optional[str]:
+        """The current access token, or None when the implementation has none to report
+        (e.g., a declarative authenticator whose token has not been initialized yet)."""
+        try:
+            return self.access_token
+        except Exception:
+            return None
+
     def refresh_and_set_access_token(self) -> None:
         """Force refresh the access token and update internal state.
 
-        This method refreshes the access token regardless of whether it has expired,
-        and updates the internal token and expiry date. Subclasses may override this
-        to handle additional state updates (e.g., persisting new refresh tokens).
-        """
-        token, expires_in = self.refresh_access_token()
-        self.access_token = token
-        self.set_token_expiry_date(expires_in)
+        Refreshes regardless of expiry, serialized on the class-level refresh lock. If another
+        thread using this same authenticator instance replaced the access token while this one
+        waited for the lock, the refresh is skipped and the request is retried with that token.
+        `SingleUseRefreshTokenOauth2Authenticator` reads `access_token` from the connector config
+        shared by all stream instances, so the early return also covers separate instances there.
+        Only per-instance token authenticators (base and declarative) are limited to same-instance
+        detection; the Authorization-header check in `HttpClient._handle_error_resolution` covers
+        the rest by skipping the forced refresh when the rejected request's token was already
+        replaced.
+        Subclasses may override this to handle additional state updates (e.g., persisting new
+        refresh tokens).
+        """
+        token_before_waiting = self._current_access_token_or_none()
+        with self._token_refresh_lock:
+            if self._current_access_token_or_none() != token_before_waiting:
+                return
+            token, expires_in = self.refresh_access_token()
+            self.access_token = token
+            self.set_token_expiry_date(expires_in)
 
     def token_has_expired(self) -> bool:
         """Returns True if the token is expired"""

airbyte_cdk/sources/streams/http/requests_native_auth/oauth.py

@@ -351,12 +351,18 @@ def refresh_and_set_access_token(self) -> None:
         and emits a control message to update the connector config. If the
         response omits a refresh token, the existing one is preserved.
         """
-        new_access_token, access_token_expires_in, new_refresh_token = self.refresh_access_token()
-        self.access_token = new_access_token
-        if new_refresh_token is not None:
-            self.set_refresh_token(new_refresh_token)
-        self.set_token_expiry_date(access_token_expires_in)
-        self._emit_control_message()
+        token_before_waiting = self._current_access_token_or_none()
+        with self._token_refresh_lock:
+            if self._current_access_token_or_none() != token_before_waiting:
+                return
+            new_access_token, access_token_expires_in, new_refresh_token = (
+                self.refresh_access_token()
+            )
+            self.access_token = new_access_token
+            if new_refresh_token is not None:
+                self.set_refresh_token(new_refresh_token)
+            self.set_token_expiry_date(access_token_expires_in)
+            self._emit_control_message()
 
     def refresh_access_token(self) -> Tuple[str, AirbyteDateTime, Optional[str]]:  # type: ignore[override]
         """

unit_tests/sources/streams/http/requests_native_auth/test_requests_native_auth.py

@@ -1414,3 +1414,197 @@ def get_token():
         assert len(results) == 5
         assert all(token == "new_access_token" for token in results)
         assert refresh_call_count == 1, f"Expected 1 refresh call, got {refresh_call_count}"
+
+    def test_refresh_and_set_access_token_skips_refresh_when_another_thread_already_refreshed(
+        self, mocker
+    ):
+        """
+        A forced refresh that was queued behind the class-level lock must not refresh
+        again when the access token changed while it waited: the request is retried
+        with the token the other thread already obtained.
+        """
+        oauth = Oauth2Authenticator(
+            token_refresh_endpoint="https://refresh_endpoint.com",
+            client_id="client_id",
+            client_secret="client_secret",
+            refresh_token="refresh_token",
+            token_expiry_date=ab_datetime_now() + timedelta(hours=1),
+        )
+        oauth.access_token = "old"
+
+        mocked_refresh = mocker.patch.object(oauth, "refresh_access_token")
+
+        snapshot_taken = threading.Event()
+        real_snapshot = oauth._current_access_token_or_none
+
+        def record_snapshot():
+            value = real_snapshot()
+            snapshot_taken.set()
+            return value
+
+        mocker.patch.object(oauth, "_current_access_token_or_none", side_effect=record_snapshot)
+
+        with Oauth2Authenticator._token_refresh_lock:
+            thread = threading.Thread(target=oauth.refresh_and_set_access_token)
+            thread.start()
+            # The worker took its pre-lock token snapshot and is now blocked on the lock.
+            assert snapshot_taken.wait(5)
+            oauth.access_token = "winner"
+        thread.join(timeout=5)
+
+        assert not thread.is_alive()
+        mocked_refresh.assert_not_called()
+        assert oauth.access_token == "winner"
+
+    def test_refresh_and_set_access_token_refreshes_when_token_unchanged(self, mocker):
+        oauth = Oauth2Authenticator(
+            token_refresh_endpoint="https://refresh_endpoint.com",
+            client_id="client_id",
+            client_secret="client_secret",
+            refresh_token="refresh_token",
+            token_expiry_date=ab_datetime_now() + timedelta(hours=1),
+        )
+        oauth.access_token = "old"
+
+        mocked_refresh = mocker.patch.object(
+            oauth,
+            "refresh_access_token",
+            return_value=("new", ab_datetime_now() + timedelta(hours=1)),
+        )
+
+        oauth.refresh_and_set_access_token()
+
+        mocked_refresh.assert_called_once()
+        assert oauth.access_token == "new"
+
+    def test_get_access_token_refresh_lock_is_reentrant(self, requests_mock):
+        """
+        get_access_token() holds the class-level refresh lock while calling
+        refresh_and_set_access_token, which re-acquires it; a non-reentrant lock
+        would deadlock. The worker thread must finish and report the new token.
+        """
+        requests_mock.post(
+            "https://refresh_endpoint.com",
+            json={"access_token": "new_access_token", "expires_in": 3600},
+        )
+        oauth = Oauth2Authenticator(
+            token_refresh_endpoint="https://refresh_endpoint.com",
+            client_id="client_id",
+            client_secret="client_secret",
+            refresh_token="refresh_token",
+            token_expiry_date=ab_datetime_now() - timedelta(hours=1),
+        )
+
+        results, errors = [], []
+
+        def get_token():
+            try:
+                results.append(oauth.get_access_token())
+            except Exception as e:
+                errors.append(e)
+
+        thread = threading.Thread(target=get_token)
+        thread.start()
+        thread.join(timeout=5)
+
+        assert not thread.is_alive(), "get_access_token deadlocked on a non-reentrant lock"
+        assert errors == []
+        assert results == ["new_access_token"]
+
+    def test_refresh_and_set_access_token_holds_lock_during_refresh(self, mocker):
+        """
+        The class-level lock must be held for the whole refresh: a probe from another
+        thread cannot acquire it while refresh_access_token is in flight.
+        """
+        oauth = Oauth2Authenticator(
+            token_refresh_endpoint="https://refresh_endpoint.com",
+            client_id="client_id",
+            client_secret="client_secret",
+            refresh_token="refresh_token",
+            token_expiry_date=ab_datetime_now() + timedelta(hours=1),
+        )
+        oauth.access_token = "old"
+
+        probe_results = []
+
+        def spy_refresh_access_token(self):
+            def probe():
+                acquired = Oauth2Authenticator._token_refresh_lock.acquire(blocking=False)
+                if acquired:
+                    Oauth2Authenticator._token_refresh_lock.release()
+                probe_results.append(acquired)
+
+            probe_thread = threading.Thread(target=probe)
+            probe_thread.start()
+            probe_thread.join()
+            return ("new", ab_datetime_now() + timedelta(hours=1))
+
+        mocker.patch.object(Oauth2Authenticator, "refresh_access_token", spy_refresh_access_token)
+
+        oauth.refresh_and_set_access_token()
+
+        assert probe_results == [False]
+        assert oauth.access_token == "new"
+
+
+class TestSingleUseRefreshTokenInstanceSkip:
+    """
+    Two SingleUseRefreshTokenOauth2Authenticator instances over one shared connector
+    config: a forced refresh queued behind the lock sees the token the first instance
+    already wrote back into the shared config and skips its own refresh.
+    """
+
+    def test_second_instance_skips_refresh_after_first_refreshes(self, requests_mock, mocker):
+        requests_mock.post(
+            "https://refresh_endpoint.com",
+            json={
+                "access_token": "new_access_token",
+                "refresh_token": "new_refresh_token",
+                "expires_in": 3600,
+            },
+        )
+        connector_config = {
+            "credentials": {
+                "client_id": "client_id",
+                "client_secret": "client_secret",
+                "refresh_token": "refresh_token",
+                "access_token": "old_access_token",
+                "token_expiry_date": str(ab_datetime_now() + timedelta(hours=1)),
+            }
+        }
+        auth1 = SingleUseRefreshTokenOauth2Authenticator(
+            connector_config=connector_config,
+            token_refresh_endpoint="https://refresh_endpoint.com",
+        )
+        auth2 = SingleUseRefreshTokenOauth2Authenticator(
+            connector_config=connector_config,
+            token_refresh_endpoint="https://refresh_endpoint.com",
+        )
+
+        mocked_emit = mocker.patch.object(
+            SingleUseRefreshTokenOauth2Authenticator, "_emit_control_message"
+        )
+
+        snapshot_taken = threading.Event()
+        real_snapshot = auth2._current_access_token_or_none
+
+        def record_snapshot():
+            value = real_snapshot()
+            snapshot_taken.set()
+            return value
+
+        mocker.patch.object(auth2, "_current_access_token_or_none", side_effect=record_snapshot)
+
+        with SingleUseRefreshTokenOauth2Authenticator._token_refresh_lock:
+            thread = threading.Thread(target=auth2.refresh_and_set_access_token)
+            thread.start()
+            # auth2 took its pre-lock snapshot ("old_access_token") and now waits on the lock.
+            assert snapshot_taken.wait(5)
+            auth1.refresh_and_set_access_token()
+        thread.join(timeout=5)
+
+        assert not thread.is_alive()
+        assert len(requests_mock.request_history) == 1
+        assert mocked_emit.call_count == 1
+        assert auth2.access_token == "new_access_token"
+        assert connector_config["credentials"]["refresh_token"] == "new_refresh_token"