From 8ee7639cc1e88b586b3515dbea235d5e5fff0871 Mon Sep 17 00:00:00 2001 From: Aditya Jain Date: Fri, 14 Aug 2026 17:52:14 -0700 Subject: [PATCH 1/2] fix(idempotency): Redis persistence layer reclaims live in-progress records as orphans, allowing concurrent double-execution RedisCachePersistenceLayer._put_in_progress_record() only guarded against a competing invocation when the existing record's in_progress_expiry_timestamp was set AND still in the future. When that field is None -- which is the normal case for idempotent_function, since it never calls config.register_lambda_context(), unlike the idempotent() handler decorator -- the guard was skipped entirely and a genuinely still-running invocation's record fell through to the "orphan record" branch, which unconditionally overwrites the record with no NX guard. A second concurrent invocation with the same idempotency key would then proceed to execute the underlying function too, defeating idempotency (e.g. double-charging a customer). This is exactly what persistence/base.py's own warning at that call site flags ("Couldn't determine the remaining time left. Did you call register_lambda_context on IdempotencyConfig?") -- it warns, but nothing downstream actually failed closed on it. The DynamoDB persistence layer does not have this gap: its conditional expression requires attribute_exists(#in_progress_expiry) before allowing an expired-in-progress reclaim, so a missing attribute correctly blocks the competing writer instead of granting a reclaim. Fix: when status is INPROGRESS and in_progress_expiry_timestamp is None, treat the record as still in progress (fail closed) instead of falling through to the orphan-reclaim path, mirroring the DynamoDB layer's behavior. --- .../idempotency/persistence/redis.py | 15 +++++--- .../idempotency/_redis/test_redis_layer.py | 34 +++++++++++++++++++ 2 files changed, 45 insertions(+), 4 deletions(-) diff --git a/aws_lambda_powertools/utilities/idempotency/persistence/redis.py b/aws_lambda_powertools/utilities/idempotency/persistence/redis.py index 82a44e079de..c0490a2b671 100644 --- a/aws_lambda_powertools/utilities/idempotency/persistence/redis.py +++ b/aws_lambda_powertools/utilities/idempotency/persistence/redis.py @@ -425,10 +425,17 @@ def _put_in_progress_record(self, data_record: DataRecord) -> None: # (meaning the timestamp is greater than the current timestamp in milliseconds), then we have encountered # a valid in-progress record. This indicates that another process is currently handling the request, and # to maintain idempotency, we raise an error to prevent concurrent processing of the same request. - if ( - idempotency_record.status == STATUS_CONSTANTS["INPROGRESS"] - and idempotency_record.in_progress_expiry_timestamp - and idempotency_record.in_progress_expiry_timestamp > int(now.timestamp() * 1000) + # + # If the record is INPROGRESS but in_progress_expiry_timestamp was never set (e.g. the caller never + # invoked config.register_lambda_context(), so remaining_time_in_millis was None when the record was + # created), we cannot determine whether the in-progress invocation has actually timed out. Fail closed + # and treat it as still in progress, rather than reclaiming it as an "orphan" below -- otherwise a + # second concurrent invocation would wrongly conclude the first one has expired and execute the + # function a second time. This mirrors the DynamoDB persistence layer, which requires + # attribute_exists(#in_progress_expiry) before allowing an expired-in-progress reclaim. + if idempotency_record.status == STATUS_CONSTANTS["INPROGRESS"] and ( + idempotency_record.in_progress_expiry_timestamp is None + or idempotency_record.in_progress_expiry_timestamp > int(now.timestamp() * 1000) ): raise IdempotencyItemAlreadyExistsError diff --git a/tests/functional/idempotency/_redis/test_redis_layer.py b/tests/functional/idempotency/_redis/test_redis_layer.py index 22c3b9a6d83..8a3a9830367 100644 --- a/tests/functional/idempotency/_redis/test_redis_layer.py +++ b/tests/functional/idempotency/_redis/test_redis_layer.py @@ -198,6 +198,18 @@ def valid_record(): ) +@pytest.fixture +def in_progress_record_missing_expiry(): + # Simulates a record created via idempotent_function without register_lambda_context() + # having been called: in_progress_expiry_timestamp was never set. This record represents + # a genuinely still-running invocation, NOT an orphan. + return DataRecord( + idempotency_key="test_orphan_key", + status=STATUS_CONSTANTS["INPROGRESS"], + in_progress_expiry_timestamp=None, + ) + + @mock.patch("aws_lambda_powertools.utilities.idempotency.persistence.redis.redis", MockRedis()) def test_redis_connection_standalone(): # when RedisCachePersistenceLayer is init with the following params @@ -303,6 +315,28 @@ def test_redis_orphan_record_lock(orphan_record, valid_record): ) +@mock.patch("aws_lambda_powertools.utilities.idempotency.persistence.redis.redis", MockRedis()) +def test_redis_in_progress_record_missing_expiry_is_not_treated_as_orphan(in_progress_record_missing_expiry): + """Regression test: an INPROGRESS record whose in_progress_expiry_timestamp is None (e.g. because + idempotent_function was used without register_lambda_context()) must NOT be reclaimed as an orphan. + Doing so lets a second concurrent invocation execute the underlying function while the first is + still genuinely running, defeating idempotency (e.g. double-charging a customer). + """ + layer = RedisCachePersistenceLayer(host="host") + # Given a genuinely still-running in-progress record with no expiry info + layer._put_in_progress_record(in_progress_record_missing_expiry) + + # When a second, concurrent invocation tries to claim the same idempotency key + # Then it must be rejected as "already in progress", not treated as an orphan and overwritten + with pytest.raises(IdempotencyItemAlreadyExistsError): + layer._put_in_progress_record(in_progress_record_missing_expiry) + + # And the original record must remain untouched + assert layer._get_record(in_progress_record_missing_expiry.idempotency_key).status == STATUS_CONSTANTS[ + "INPROGRESS" + ] + + @mock.patch("aws_lambda_powertools.utilities.idempotency.persistence.redis.redis", MockRedis()) def test_redis_error_in_progress(valid_record): layer = RedisCachePersistenceLayer(host="host", mode="standalone") From 7187e9bf1d499e069ac14896c6177aeae0414411 Mon Sep 17 00:00:00 2001 From: Leandro Date: Fri, 28 Aug 2026 20:08:07 +0100 Subject: [PATCH 2/2] test(idempotency): cover concurrent cache execution --- .../idempotency/persistence/redis.py | 9 +-- .../idempotency/_redis/test_redis_layer.py | 77 +++++++++++++++---- 2 files changed, 63 insertions(+), 23 deletions(-) diff --git a/aws_lambda_powertools/utilities/idempotency/persistence/redis.py b/aws_lambda_powertools/utilities/idempotency/persistence/redis.py index c0490a2b671..ce7100bf215 100644 --- a/aws_lambda_powertools/utilities/idempotency/persistence/redis.py +++ b/aws_lambda_powertools/utilities/idempotency/persistence/redis.py @@ -426,13 +426,8 @@ def _put_in_progress_record(self, data_record: DataRecord) -> None: # a valid in-progress record. This indicates that another process is currently handling the request, and # to maintain idempotency, we raise an error to prevent concurrent processing of the same request. # - # If the record is INPROGRESS but in_progress_expiry_timestamp was never set (e.g. the caller never - # invoked config.register_lambda_context(), so remaining_time_in_millis was None when the record was - # created), we cannot determine whether the in-progress invocation has actually timed out. Fail closed - # and treat it as still in progress, rather than reclaiming it as an "orphan" below -- otherwise a - # second concurrent invocation would wrongly conclude the first one has expired and execute the - # function a second time. This mirrors the DynamoDB persistence layer, which requires - # attribute_exists(#in_progress_expiry) before allowing an expired-in-progress reclaim. + # Without an in-progress expiry, we cannot safely distinguish an active invocation from a timed-out one. + # Fail closed until the record TTL expires, consistent with the DynamoDB persistence layer. if idempotency_record.status == STATUS_CONSTANTS["INPROGRESS"] and ( idempotency_record.in_progress_expiry_timestamp is None or idempotency_record.in_progress_expiry_timestamp > int(now.timestamp() * 1000) diff --git a/tests/functional/idempotency/_redis/test_redis_layer.py b/tests/functional/idempotency/_redis/test_redis_layer.py index 8a3a9830367..6adb97a64a4 100644 --- a/tests/functional/idempotency/_redis/test_redis_layer.py +++ b/tests/functional/idempotency/_redis/test_redis_layer.py @@ -3,6 +3,7 @@ import datetime import json import time as t +from threading import Event, Lock as ThreadLock, Thread from unittest import mock import pytest @@ -26,6 +27,7 @@ STATUS_CONSTANTS, DataRecord, ) +from aws_lambda_powertools.utilities.idempotency.persistence.cache import CachePersistenceLayer from aws_lambda_powertools.utilities.idempotency.persistence.redis import ( RedisCachePersistenceLayer, ) @@ -200,12 +202,10 @@ def valid_record(): @pytest.fixture def in_progress_record_missing_expiry(): - # Simulates a record created via idempotent_function without register_lambda_context() - # having been called: in_progress_expiry_timestamp was never set. This record represents - # a genuinely still-running invocation, NOT an orphan. return DataRecord( idempotency_key="test_orphan_key", status=STATUS_CONSTANTS["INPROGRESS"], + expiry_timestamp=int(datetime.datetime.now().timestamp()) + 60, in_progress_expiry_timestamp=None, ) @@ -317,24 +317,69 @@ def test_redis_orphan_record_lock(orphan_record, valid_record): @mock.patch("aws_lambda_powertools.utilities.idempotency.persistence.redis.redis", MockRedis()) def test_redis_in_progress_record_missing_expiry_is_not_treated_as_orphan(in_progress_record_missing_expiry): - """Regression test: an INPROGRESS record whose in_progress_expiry_timestamp is None (e.g. because - idempotent_function was used without register_lambda_context()) must NOT be reclaimed as an orphan. - Doing so lets a second concurrent invocation execute the underlying function while the first is - still genuinely running, defeating idempotency (e.g. double-charging a customer). - """ layer = RedisCachePersistenceLayer(host="host") - # Given a genuinely still-running in-progress record with no expiry info layer._put_in_progress_record(in_progress_record_missing_expiry) - # When a second, concurrent invocation tries to claim the same idempotency key - # Then it must be rejected as "already in progress", not treated as an orphan and overwritten + contender = DataRecord( + idempotency_key=in_progress_record_missing_expiry.idempotency_key, + status=STATUS_CONSTANTS["INPROGRESS"], + expiry_timestamp=in_progress_record_missing_expiry.expiry_timestamp + 60, + in_progress_expiry_timestamp=None, + ) + with pytest.raises(IdempotencyItemAlreadyExistsError): - layer._put_in_progress_record(in_progress_record_missing_expiry) + layer._put_in_progress_record(contender) + + stored_record = layer._get_record(in_progress_record_missing_expiry.idempotency_key) + assert stored_record.status == STATUS_CONSTANTS["INPROGRESS"] + assert stored_record.expiry_timestamp == in_progress_record_missing_expiry.expiry_timestamp + + +@pytest.mark.filterwarnings("ignore:Couldn't determine the remaining time left") +def test_idempotent_function_blocks_concurrent_invocation_without_lambda_context(): + layer = CachePersistenceLayer(client=MockRedis(host="localhost")) + first_invocation_started = Event() + release_first_invocation = Event() + execution_lock = ThreadLock() + execution_count = 0 + first_result = [] + first_errors = [] + + @idempotent_function(data_keyword_argument="record", persistence_store=layer) + def process(record): + nonlocal execution_count + with execution_lock: + execution_count += 1 + current_execution = execution_count + + if current_execution == 1: + first_invocation_started.set() + if not release_first_invocation.wait(timeout=5): + raise TimeoutError("Timed out waiting to release the first invocation") + + return {"execution": current_execution} + + def invoke_first(): + try: + first_result.append(process(record={"id": "same"})) + except Exception as exc: + first_errors.append(exc) + + first_invocation = Thread(target=invoke_first) + first_invocation.start() + assert first_invocation_started.wait(timeout=2) - # And the original record must remain untouched - assert layer._get_record(in_progress_record_missing_expiry.idempotency_key).status == STATUS_CONSTANTS[ - "INPROGRESS" - ] + try: + with pytest.raises(IdempotencyAlreadyInProgressError): + process(record={"id": "same"}) + finally: + release_first_invocation.set() + first_invocation.join(timeout=5) + + assert not first_invocation.is_alive() + assert first_errors == [] + assert first_result == [{"execution": 1}] + assert execution_count == 1 @mock.patch("aws_lambda_powertools.utilities.idempotency.persistence.redis.redis", MockRedis())