From d9ac077eed875008ebdad156c32f2e26cab4330c Mon Sep 17 00:00:00 2001 From: Vidit Shah Date: Sun, 18 Oct 2026 22:24:07 +0530 Subject: [PATCH 1/3] fix(parameters): recover AppConfig sessions after rejected tokens --- .../utilities/parameters/appconfig.py | 46 +++- .../_boto3/test_appconfig_token_recovery.py | 196 ++++++++++++++++++ 2 files changed, 234 insertions(+), 8 deletions(-) create mode 100644 tests/functional/parameters/_boto3/test_appconfig_token_recovery.py diff --git a/aws_lambda_powertools/utilities/parameters/appconfig.py b/aws_lambda_powertools/utilities/parameters/appconfig.py index 1b52cd96698..fb79ef5cb39 100644 --- a/aws_lambda_powertools/utilities/parameters/appconfig.py +++ b/aws_lambda_powertools/utilities/parameters/appconfig.py @@ -30,6 +30,13 @@ class AppConfigProvider(BaseProvider): """ AWS App Config Provider + A rejected poll token starts a new configuration session and retries retrieval once. + If the replacement token is rejected, retrieval raises GetParameterError and the next + call can start a fresh session. Other service errors do not trigger session recovery. + If reading the response body fails, retrieval raises GetParameterError and discards + that configuration's token so the next retrieval can recover. Caching and transforms + still apply, and recovery does not reset other configuration profiles. + Parameters ---------- environment: str @@ -125,17 +132,40 @@ def _get(self, name: str, **sdk_options) -> bytes: SDK options to propagate to `start_configuration_session` API call """ if name not in self._next_token: - sdk_options["ConfigurationProfileIdentifier"] = name - sdk_options["ApplicationIdentifier"] = self.application - sdk_options["EnvironmentIdentifier"] = self.environment - response_configuration = self.client.start_configuration_session(**sdk_options) - self._next_token[name] = response_configuration["InitialConfigurationToken"] - + self._start_configuration_session(name, **sdk_options) + + try: + return self._get_configuration(name) + except self.client.exceptions.BadRequestException: + # Rejected tokens cannot be reused. Restart once, without recursively retrying + # a rejected replacement or treating unrelated service errors as token expiry. + self._start_configuration_session(name, **sdk_options) + return self._get_configuration(name) + + def _start_configuration_session(self, name: str, **sdk_options) -> None: + sdk_options["ConfigurationProfileIdentifier"] = name + sdk_options["ApplicationIdentifier"] = self.application + sdk_options["EnvironmentIdentifier"] = self.environment + response = self.client.start_configuration_session(**sdk_options) + self._next_token[name] = response["InitialConfigurationToken"] + + def _get_configuration(self, name: str) -> bytes: # The new AppConfig APIs require two API calls to return the configuration # First we start the session and after that we retrieve the configuration # We need to store the token to use in the next execution - response = self.client.get_latest_configuration(ConfigurationToken=self._next_token[name]) - return_value = response["Configuration"].read() + try: + response = self.client.get_latest_configuration(ConfigurationToken=self._next_token[name]) + except self.client.exceptions.BadRequestException: + self._next_token.pop(name, None) + raise + + try: + return_value = response["Configuration"].read() + except Exception: + # The request consumed the token, but we did not receive a complete value. + # A new session on the next call retrieves the configuration again. + self._next_token.pop(name, None) + raise self._next_token[name] = response["NextPollConfigurationToken"] # The return of get_latest_configuration can be null because this value is supposed to be cached diff --git a/tests/functional/parameters/_boto3/test_appconfig_token_recovery.py b/tests/functional/parameters/_boto3/test_appconfig_token_recovery.py new file mode 100644 index 00000000000..6df02800286 --- /dev/null +++ b/tests/functional/parameters/_boto3/test_appconfig_token_recovery.py @@ -0,0 +1,196 @@ +from __future__ import annotations + +from io import BytesIO + +import boto3 +import pytest +from botocore.config import Config +from botocore.response import StreamingBody +from botocore.stub import Stubber + +from aws_lambda_powertools.utilities.parameters import AppConfigProvider +from aws_lambda_powertools.utilities.parameters.exceptions import GetParameterError + + +@pytest.fixture +def provider_and_stubber(): + client = boto3.client( + "appconfigdata", + config=Config(region_name="us-east-1", proxies={}), + aws_access_key_id="testing", + aws_secret_access_key="testing", + aws_session_token="testing", + ) + provider = AppConfigProvider(environment="dev", application="app", boto3_client=client) + with Stubber(client) as stubber: + yield provider, stubber + stubber.assert_no_pending_responses() + + +def add_session(stubber, token, name="cfg", **sdk_options): + stubber.add_response( + "start_configuration_session", + {"InitialConfigurationToken": token}, + { + "ConfigurationProfileIdentifier": name, + "ApplicationIdentifier": "app", + "EnvironmentIdentifier": "dev", + **sdk_options, + }, + ) + + +def add_poll(stubber, token, next_token, value, declared_length=None): + stubber.add_response( + "get_latest_configuration", + { + "NextPollConfigurationToken": next_token, + "Configuration": StreamingBody(BytesIO(value), len(value) if declared_length is None else declared_length), + }, + {"ConfigurationToken": token}, + ) + + +def add_rejected_token(stubber, token): + stubber.add_client_error( + "get_latest_configuration", + "BadRequestException", + "Token expired", + expected_params={"ConfigurationToken": token}, + ) + + +def test_appconfig_recovers_rejected_poll_token_and_caches_new_value(provider_and_stubber): + provider, stubber = provider_and_stubber + add_session(stubber, "initial") + add_poll(stubber, "initial", "expired", b"old") + add_rejected_token(stubber, "expired") + add_session(stubber, "fresh") + add_poll(stubber, "fresh", "next", b"new") + + assert provider.get("cfg", max_age=300) == b"old" + assert provider.get("cfg", force_fetch=True, max_age=300) == b"new" + assert provider.get("cfg", max_age=300) == b"new" + assert provider._next_token["cfg"] == "next" + + +def test_appconfig_retries_once_then_allows_recovery_on_a_later_call(provider_and_stubber): + provider, stubber = provider_and_stubber + add_session(stubber, "initial") + add_rejected_token(stubber, "initial") + add_session(stubber, "retry") + add_rejected_token(stubber, "retry") + + with pytest.raises(GetParameterError, match="Token expired"): + provider.get("cfg") + stubber.assert_no_pending_responses() + assert "cfg" not in provider._next_token + + add_session(stubber, "fresh") + add_poll(stubber, "fresh", "next", b"recovered") + assert provider.get("cfg") == b"recovered" + + +def test_appconfig_starts_new_session_after_response_body_read_fails(provider_and_stubber): + provider, stubber = provider_and_stubber + add_session(stubber, "initial") + add_poll(stubber, "initial", "consumed", b"old") + # A truncated response consumed the token but did not deliver usable configuration. + add_poll(stubber, "consumed", "unread", b"partial", declared_length=100) + + assert provider.get("cfg", max_age=300) == b"old" + with pytest.raises(GetParameterError, match="read, but total bytes expected"): + provider.get("cfg", force_fetch=True) + assert "cfg" not in provider._next_token + assert provider.last_returned_value["cfg"] == b"old" + + add_session(stubber, "fresh") + add_poll(stubber, "fresh", "next", b"complete") + assert provider.get("cfg", force_fetch=True) == b"complete" + + +@pytest.mark.parametrize("error_code", ["ThrottlingException", "InternalServerException", "ResourceNotFoundException"]) +def test_appconfig_does_not_restart_for_unrelated_service_errors(provider_and_stubber, error_code): + provider, stubber = provider_and_stubber + add_session(stubber, "initial") + stubber.add_client_error( + "get_latest_configuration", + error_code, + "service failure", + expected_params={"ConfigurationToken": "initial"}, + ) + with pytest.raises(GetParameterError, match="service failure"): + provider.get("cfg") + assert provider._next_token["cfg"] == "initial" + stubber.assert_no_pending_responses() + + +def test_appconfig_preserves_session_options_and_configuration_isolation(provider_and_stubber): + provider, stubber = provider_and_stubber + options = {"RequiredMinimumPollIntervalInSeconds": 30} + add_session(stubber, "other-initial", name="other") + add_poll(stubber, "other-initial", "other-next", b"other-value") + add_session(stubber, "initial", **options) + add_rejected_token(stubber, "initial") + add_session(stubber, "fresh", **options) + add_poll(stubber, "fresh", "next", b'{"enabled":true}') + add_poll(stubber, "next", "next-empty", b"") + add_poll(stubber, "other-next", "other-last", b"") + + assert provider.get("other", max_age=0) == b"other-value" + assert provider.get("cfg", transform="json", max_age=0, **options) == {"enabled": True} + assert provider.get("cfg", transform="json", max_age=0) == {"enabled": True} + assert provider.get("other", max_age=0) == b"other-value" + assert provider._next_token == {"other": "other-last", "cfg": "next-empty"} + assert options == {"RequiredMinimumPollIntervalInSeconds": 30} + + +def test_appconfig_can_recover_after_restart_session_fails(provider_and_stubber): + provider, stubber = provider_and_stubber + add_session(stubber, "initial") + add_rejected_token(stubber, "initial") + stubber.add_client_error("start_configuration_session", "ResourceNotFoundException", "session unavailable") + + with pytest.raises(GetParameterError, match="session unavailable"): + provider.get("cfg") + assert "cfg" not in provider._next_token + + add_session(stubber, "fresh") + add_poll(stubber, "fresh", "next", b"recovered") + assert provider.get("cfg") == b"recovered" + + +def test_appconfig_does_not_retry_initial_session_errors(provider_and_stubber): + provider, stubber = provider_and_stubber + stubber.add_client_error("start_configuration_session", "BadRequestException", "invalid session options") + with pytest.raises(GetParameterError, match="invalid session options"): + provider.get("cfg") + assert "cfg" not in provider._next_token + + +def test_appconfig_does_not_retry_replacement_token_throttling(provider_and_stubber): + provider, stubber = provider_and_stubber + add_session(stubber, "initial") + add_rejected_token(stubber, "initial") + add_session(stubber, "fresh") + stubber.add_client_error( + "get_latest_configuration", + "ThrottlingException", + "slow down", + expected_params={"ConfigurationToken": "fresh"}, + ) + with pytest.raises(GetParameterError, match="slow down"): + provider.get("cfg") + assert provider._next_token["cfg"] == "fresh" + + +def test_appconfig_discards_replacement_token_if_its_response_is_truncated(provider_and_stubber): + provider, stubber = provider_and_stubber + add_session(stubber, "initial") + add_rejected_token(stubber, "initial") + add_session(stubber, "fresh") + add_poll(stubber, "fresh", "unread", b"partial", declared_length=100) + with pytest.raises(GetParameterError, match="read, but total bytes expected"): + provider.get("cfg") + assert "cfg" not in provider._next_token + assert "cfg" not in provider.last_returned_value From cf271854c99f75ef380fc8cf35c8cb17510e0250 Mon Sep 17 00:00:00 2001 From: Vidit Shah Date: Mon, 19 Oct 2026 20:53:02 +0530 Subject: [PATCH 2/3] fix(parameters): preserve valid AppConfig poll tokens --- .../utilities/parameters/appconfig.py | 30 ++++-- .../_boto3/test_appconfig_token_recovery.py | 101 ++++++++++++++++-- 2 files changed, 117 insertions(+), 14 deletions(-) diff --git a/aws_lambda_powertools/utilities/parameters/appconfig.py b/aws_lambda_powertools/utilities/parameters/appconfig.py index fb79ef5cb39..bf8ebf652e7 100644 --- a/aws_lambda_powertools/utilities/parameters/appconfig.py +++ b/aws_lambda_powertools/utilities/parameters/appconfig.py @@ -21,6 +21,7 @@ if TYPE_CHECKING: from botocore.config import Config + from botocore.exceptions import ClientError from mypy_boto3_appconfigdata.client import AppConfigDataClient from aws_lambda_powertools.utilities.parameters.types import TransformOptions @@ -30,9 +31,10 @@ class AppConfigProvider(BaseProvider): """ AWS App Config Provider - A rejected poll token starts a new configuration session and retries retrieval once. - If the replacement token is rejected, retrieval raises GetParameterError and the next - call can start a fresh session. Other service errors do not trigger session recovery. + An expired or corrupted poll token starts a new configuration session and retries retrieval once. + If the replacement token is expired or corrupted, retrieval raises GetParameterError and the next + call can start a fresh session. Other service errors, including polling too early, retain + the token and do not trigger session recovery. If reading the response body fails, retrieval raises GetParameterError and discards that configuration's token so the next retrieval can recover. Caching and transforms still apply, and recovery does not reset other configuration profiles. @@ -136,9 +138,11 @@ def _get(self, name: str, **sdk_options) -> bytes: try: return self._get_configuration(name) - except self.client.exceptions.BadRequestException: - # Rejected tokens cannot be reused. Restart once, without recursively retrying - # a rejected replacement or treating unrelated service errors as token expiry. + except self.client.exceptions.BadRequestException as exc: + if not self._is_token_expired_or_corrupted(exc): + raise + # Expired or corrupted tokens cannot be reused. Restart once, without recursively + # retrying a rejected replacement or bypassing the configured polling interval. self._start_configuration_session(name, **sdk_options) return self._get_configuration(name) @@ -155,8 +159,9 @@ def _get_configuration(self, name: str) -> bytes: # We need to store the token to use in the next execution try: response = self.client.get_latest_configuration(ConfigurationToken=self._next_token[name]) - except self.client.exceptions.BadRequestException: - self._next_token.pop(name, None) + except self.client.exceptions.BadRequestException as exc: + if self._is_token_expired_or_corrupted(exc): + self._next_token.pop(name, None) raise try: @@ -177,6 +182,15 @@ def _get_configuration(self, name: str) -> bytes: return self.last_returned_value[name] + @staticmethod + def _is_token_expired_or_corrupted(exc: ClientError) -> bool: + problem: object = exc.response + for key in ("Details", "InvalidParameters", "ConfigurationToken", "Problem"): + if not isinstance(problem, dict): + return False + problem = problem.get(key) + return problem in ("Expired", "Corrupted") + def _get_multiple(self, path: str, **sdk_options) -> dict[str, str]: """ Retrieving multiple parameter values is not supported with AWS App Config Provider diff --git a/tests/functional/parameters/_boto3/test_appconfig_token_recovery.py b/tests/functional/parameters/_boto3/test_appconfig_token_recovery.py index 6df02800286..31baea6e21f 100644 --- a/tests/functional/parameters/_boto3/test_appconfig_token_recovery.py +++ b/tests/functional/parameters/_boto3/test_appconfig_token_recovery.py @@ -51,20 +51,25 @@ def add_poll(stubber, token, next_token, value, declared_length=None): ) -def add_rejected_token(stubber, token): +def add_rejected_token(stubber, token, problem="Expired"): stubber.add_client_error( "get_latest_configuration", "BadRequestException", "Token expired", expected_params={"ConfigurationToken": token}, + modeled_fields={ + "Reason": "InvalidParameters", + "Details": {"InvalidParameters": {"ConfigurationToken": {"Problem": problem}}}, + }, ) -def test_appconfig_recovers_rejected_poll_token_and_caches_new_value(provider_and_stubber): +@pytest.mark.parametrize("problem", ["Expired", "Corrupted"]) +def test_appconfig_recovers_rejected_poll_token_and_caches_new_value(provider_and_stubber, problem): provider, stubber = provider_and_stubber add_session(stubber, "initial") add_poll(stubber, "initial", "expired", b"old") - add_rejected_token(stubber, "expired") + add_rejected_token(stubber, "expired", problem) add_session(stubber, "fresh") add_poll(stubber, "fresh", "next", b"new") @@ -74,12 +79,18 @@ def test_appconfig_recovers_rejected_poll_token_and_caches_new_value(provider_an assert provider._next_token["cfg"] == "next" -def test_appconfig_retries_once_then_allows_recovery_on_a_later_call(provider_and_stubber): +@pytest.mark.parametrize("problem", ["Expired", "Corrupted"]) +@pytest.mark.parametrize("replacement_problem", ["Expired", "Corrupted"]) +def test_appconfig_retries_once_then_allows_recovery_on_a_later_call( + provider_and_stubber, + problem, + replacement_problem, +): provider, stubber = provider_and_stubber add_session(stubber, "initial") - add_rejected_token(stubber, "initial") + add_rejected_token(stubber, "initial", problem) add_session(stubber, "retry") - add_rejected_token(stubber, "retry") + add_rejected_token(stubber, "retry", replacement_problem) with pytest.raises(GetParameterError, match="Token expired"): provider.get("cfg") @@ -91,6 +102,84 @@ def test_appconfig_retries_once_then_allows_recovery_on_a_later_call(provider_an assert provider.get("cfg") == b"recovered" +@pytest.mark.parametrize("recovery_problem", [None, "Expired", "Corrupted"]) +def test_appconfig_preserves_token_and_last_value_for_poll_interval_errors(provider_and_stubber, recovery_problem): + provider, stubber = provider_and_stubber + options = {"RequiredMinimumPollIntervalInSeconds": 30} + add_session(stubber, "initial", **options) + add_poll(stubber, "initial", "next", b"old") + token = "next" + if recovery_problem: + add_rejected_token(stubber, token, recovery_problem) + add_session(stubber, "replacement", **options) + token = "replacement" + add_rejected_token(stubber, token, "PollIntervalNotSatisfied") + add_rejected_token(stubber, token, "PollIntervalNotSatisfied") + add_poll(stubber, token, "last", b"") + + assert provider.get("cfg", max_age=300, **options) == b"old" + for _ in range(2): + with pytest.raises(GetParameterError, match="BadRequestException"): + provider.get("cfg", force_fetch=True, **options) + assert provider._next_token["cfg"] == token + assert provider.last_returned_value["cfg"] == b"old" + assert provider.get("cfg", max_age=300) == b"old" + + assert provider.get("cfg", force_fetch=True) == b"old" + assert provider._next_token["cfg"] == "last" + + +@pytest.mark.parametrize("recovery_problem", [None, "Expired", "Corrupted"]) +def test_appconfig_preserves_token_for_poll_interval_errors_before_first_value(provider_and_stubber, recovery_problem): + provider, stubber = provider_and_stubber + add_session(stubber, "initial") + token = "initial" + if recovery_problem: + add_rejected_token(stubber, token, recovery_problem) + add_session(stubber, "replacement") + token = "replacement" + add_rejected_token(stubber, token, "PollIntervalNotSatisfied") + add_poll(stubber, token, "next", b"new") + + with pytest.raises(GetParameterError, match="BadRequestException"): + provider.get("cfg") + assert provider._next_token["cfg"] == token + assert "cfg" not in provider.last_returned_value + assert provider.get("cfg") == b"new" + + +@pytest.mark.parametrize( + "modeled_fields", + [ + None, + {"Reason": "InvalidParameters"}, + {"Details": {"InvalidParameters": {}}}, + {"Details": {"InvalidParameters": {"ConfigurationToken": {}}}}, + {"Details": {"InvalidParameters": {"ConfigurationToken": {"Problem": "Unknown"}}}}, + {"Details": {"InvalidParameters": {"OtherParameter": {"Problem": "Expired"}}}}, + ], +) +def test_appconfig_preserves_token_for_unrecognized_bad_requests(provider_and_stubber, modeled_fields): + provider, stubber = provider_and_stubber + add_session(stubber, "initial") + add_poll(stubber, "initial", "next", b"old") + stubber.add_client_error( + "get_latest_configuration", + "BadRequestException", + "unrecognized bad request", + expected_params={"ConfigurationToken": "next"}, + modeled_fields=modeled_fields, + ) + add_poll(stubber, "next", "last", b"") + + assert provider.get("cfg", max_age=300) == b"old" + with pytest.raises(GetParameterError, match="unrecognized bad request"): + provider.get("cfg", force_fetch=True) + assert provider._next_token["cfg"] == "next" + assert provider.last_returned_value["cfg"] == b"old" + assert provider.get("cfg", force_fetch=True) == b"old" + + def test_appconfig_starts_new_session_after_response_body_read_fails(provider_and_stubber): provider, stubber = provider_and_stubber add_session(stubber, "initial") From c0a4cdcd62767000367783d464d3809a05f746eb Mon Sep 17 00:00:00 2001 From: Leandro Damascena Date: Thu, 22 Oct 2026 14:39:58 +0000 Subject: [PATCH 3/3] test(parameters): read the cached value with the same sdk_options Since #8619 sdk_options are part of the cache key, so the cached read must pass the same RequiredMinimumPollIntervalInSeconds option as the call that populated it. --- .../parameters/_boto3/test_appconfig_token_recovery.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/functional/parameters/_boto3/test_appconfig_token_recovery.py b/tests/functional/parameters/_boto3/test_appconfig_token_recovery.py index 31baea6e21f..02b99ead25f 100644 --- a/tests/functional/parameters/_boto3/test_appconfig_token_recovery.py +++ b/tests/functional/parameters/_boto3/test_appconfig_token_recovery.py @@ -123,7 +123,7 @@ def test_appconfig_preserves_token_and_last_value_for_poll_interval_errors(provi provider.get("cfg", force_fetch=True, **options) assert provider._next_token["cfg"] == token assert provider.last_returned_value["cfg"] == b"old" - assert provider.get("cfg", max_age=300) == b"old" + assert provider.get("cfg", max_age=300, **options) == b"old" assert provider.get("cfg", force_fetch=True) == b"old" assert provider._next_token["cfg"] == "last"