[Core] Add jitter to token refresh intervals (#43720)

pvaneck · web-flow · commit 4c4c8698d61f · 2026-03-04T14:40:51.000-08:00
Reduce thundering herds of refresh requests by adding some jitter.

Signed-off-by: Paul Van Eck &lt;paulvaneck@microsoft.com&gt;
diff --git a/sdk/core/azure-core/CHANGELOG.md b/sdk/core/azure-core/CHANGELOG.md
@@ -12,6 +12,8 @@
 
 ### Other Changes
 
+- Added jitter to token refresh timing in `BearerTokenCredentialPolicy` and `AsyncBearerTokenCredentialPolicy` to prevent simultaneous token refresh attempts across multiple processes. This helps mitigate the thundering herd problem during token refresh operations. #43720
+
 ## 1.38.2 (2026-02-18)
 
 ### Bugs Fixed
diff --git a/sdk/core/azure-core/azure/core/pipeline/policies/_authentication.py b/sdk/core/azure-core/azure/core/pipeline/policies/_authentication.py
@@ -5,6 +5,7 @@
 # -------------------------------------------------------------------------
 import time
 import base64
+import random
 from typing import TYPE_CHECKING, Optional, TypeVar, MutableMapping, Any, Union, cast
 
 from azure.core.credentials import (
@@ -36,6 +37,38 @@
 HTTPResponseType = TypeVar("HTTPResponseType", HttpResponse, LegacyHttpResponse)
 HTTPRequestType = TypeVar("HTTPRequestType", HttpRequest, LegacyHttpRequest)
 
+DEFAULT_REFRESH_WINDOW_SECONDS = 300  # 5 minutes
+MAX_REFRESH_JITTER_SECONDS = 60  # 1 minute
+
+
+def _should_refresh_token(token: Optional[Union["AccessToken", "AccessTokenInfo"]], refresh_jitter: int) -> bool:
+    """Check if a new token is needed based on expiry and refresh logic.
+
+    :param token: The current token or None if no token exists
+    :type token: Optional[Union[~azure.core.credentials.AccessToken, ~azure.core.credentials.AccessTokenInfo]]
+    :param int refresh_jitter: The jitter to apply to refresh timing
+    :return: True if a new token is needed, False otherwise
+    :rtype: bool
+    """
+    if not token:
+        return True
+
+    now = time.time()
+    if token.expires_on <= now:
+        return True
+
+    refresh_on = getattr(token, "refresh_on", None)
+
+    if refresh_on:
+        # Apply jitter, but ensure that adding it doesn't push the refresh time past the actual expiration.
+        # This is a safeguard, as refresh_on is typically well before expires_on.
+        effective_refresh_time = min(refresh_on + refresh_jitter, token.expires_on)
+        return effective_refresh_time <= now
+
+    time_until_expiry = token.expires_on - now
+    # Reduce refresh window by jitter to delay refresh and distribute load
+    return time_until_expiry < (DEFAULT_REFRESH_WINDOW_SECONDS - refresh_jitter)
+
 
 # pylint:disable=too-few-public-methods
 class _BearerTokenCredentialPolicyBase:
@@ -54,6 +87,7 @@ def __init__(self, credential: TokenProvider, *scopes: str, **kwargs: Any) -> No
         self._credential = credential
         self._token: Optional[Union["AccessToken", "AccessTokenInfo"]] = None
         self._enable_cae: bool = kwargs.get("enable_cae", False)
+        self._refresh_jitter = 0
 
     @staticmethod
     def _enforce_https(request: PipelineRequest[HTTPRequestType]) -> None:
@@ -82,9 +116,7 @@ def _update_headers(headers: MutableMapping[str, str], token: str) -> None:
 
     @property
     def _need_new_token(self) -> bool:
-        now = time.time()
-        refresh_on = getattr(self._token, "refresh_on", None)
-        return not self._token or (refresh_on and refresh_on <= now) or self._token.expires_on - now < 300
+        return _should_refresh_token(self._token, self._refresh_jitter)
 
     def _get_token(self, *scopes: str, **kwargs: Any) -> Union["AccessToken", "AccessTokenInfo"]:
         if self._enable_cae:
@@ -108,6 +140,7 @@ def _request_token(self, *scopes: str, **kwargs: Any) -> None:
         :param str scopes: The type of access needed.
         """
         self._token = self._get_token(*scopes, **kwargs)
+        self._refresh_jitter = random.randint(0, MAX_REFRESH_JITTER_SECONDS)
 
 
 class BearerTokenCredentialPolicy(_BearerTokenCredentialPolicyBase, HTTPPolicy[HTTPRequestType, HTTPResponseType]):
diff --git a/sdk/core/azure-core/azure/core/pipeline/policies/_authentication_async.py b/sdk/core/azure-core/azure/core/pipeline/policies/_authentication_async.py
@@ -3,7 +3,7 @@
 # Licensed under the MIT License. See LICENSE.txt in the project root for
 # license information.
 # -------------------------------------------------------------------------
-import time
+import random
 import base64
 from typing import Any, Awaitable, Optional, cast, TypeVar, Union
 
@@ -18,6 +18,8 @@
 from azure.core.pipeline.policies import AsyncHTTPPolicy
 from azure.core.pipeline.policies._authentication import (
     _BearerTokenCredentialPolicyBase,
+    _should_refresh_token,
+    MAX_REFRESH_JITTER_SECONDS,
 )
 from azure.core.pipeline.transport import (
     AsyncHttpResponse as LegacyAsyncHttpResponse,
@@ -50,6 +52,7 @@ def __init__(self, credential: AsyncTokenProvider, *scopes: str, **kwargs: Any)
         self._lock_instance = None
         self._token: Optional[Union["AccessToken", "AccessTokenInfo"]] = None
         self._enable_cae: bool = kwargs.get("enable_cae", False)
+        self._refresh_jitter = 0
 
     @property
     def _lock(self):
@@ -192,9 +195,7 @@ def on_exception(self, request: PipelineRequest[HTTPRequestType]) -> None:
         return
 
     def _need_new_token(self) -> bool:
-        now = time.time()
-        refresh_on = getattr(self._token, "refresh_on", None)
-        return not self._token or (refresh_on and refresh_on <= now) or self._token.expires_on - now < 300
+        return _should_refresh_token(self._token, self._refresh_jitter)
 
     async def _get_token(self, *scopes: str, **kwargs: Any) -> Union["AccessToken", "AccessTokenInfo"]:
         if self._enable_cae:
@@ -226,3 +227,4 @@ async def _request_token(self, *scopes: str, **kwargs: Any) -> None:
         :param str scopes: The type of access needed.
         """
         self._token = await self._get_token(*scopes, **kwargs)
+        self._refresh_jitter = random.randint(0, MAX_REFRESH_JITTER_SECONDS)
diff --git a/sdk/core/azure-core/tests/async_tests/test_authentication_async.py b/sdk/core/azure-core/tests/async_tests/test_authentication_async.py
@@ -20,6 +20,7 @@
     AsyncRedirectPolicy,
     SensitiveHeaderCleanupPolicy,
 )
+from azure.core.pipeline.policies._authentication import MAX_REFRESH_JITTER_SECONDS
 from azure.core.pipeline.transport import AsyncHttpTransport, HttpRequest
 import pytest
 import trio
@@ -244,7 +245,9 @@ async def test_bearer_policy_access_token_info_caching(http_request):
     await pipeline.run(http_request("GET", "https://spam.eggs"))
     assert credential.get_token_info.call_count == 2  # token is expired -> policy should call get_token_info again
 
-    refreshable_token = AccessTokenInfo("token", int(time.time() + 3600), refresh_on=int(time.time() - 1))
+    refreshable_token = AccessTokenInfo(
+        "token", int(time.time() + 3600), refresh_on=int(time.time() - (MAX_REFRESH_JITTER_SECONDS + 5))
+    )
     credential.get_token_info.reset_mock()
     credential.get_token_info.return_value = refreshable_token
     pipeline = AsyncPipeline(transport=AsyncMock(), policies=[AsyncBearerTokenCredentialPolicy(credential, "scope")])
@@ -735,3 +738,33 @@ async def mock_transport_send(request, **kwargs):
     # Verify the exception chaining
     assert exc_info.value.__cause__ is not None
     assert isinstance(exc_info.value.__cause__, HttpResponseError)
+
+
+@pytest.mark.asyncio
+async def test_jitter_set_on_token_request_async():
+    """Test that _refresh_jitter is set when _request_token is called on the async policy."""
+    token = AccessToken("test_token", int(time.time()) + 3600)
+
+    credential = AsyncMock(spec_set=["get_token"])
+    credential.get_token.return_value = token
+    policy = AsyncBearerTokenCredentialPolicy(credential, "scope")
+
+    # Initially jitter should be 0
+    assert policy._refresh_jitter == 0
+
+    with patch("azure.core.pipeline.policies._authentication_async.random.randint") as mock_randint:
+        mock_randint.return_value = 42
+
+        await policy._request_token("scope")
+
+        assert policy._refresh_jitter == 42
+        mock_randint.assert_called_once_with(0, MAX_REFRESH_JITTER_SECONDS)
+
+    # Test that jitter is updated on subsequent token requests
+    with patch("azure.core.pipeline.policies._authentication_async.random.randint") as mock_randint:
+        mock_randint.return_value = 25
+
+        await policy._request_token("scope")
+
+        assert policy._refresh_jitter == 25
+        mock_randint.assert_called_once_with(0, MAX_REFRESH_JITTER_SECONDS)
diff --git a/sdk/core/azure-core/tests/test_authentication.py b/sdk/core/azure-core/tests/test_authentication.py
