fix(idempotency): Redis persistence layer reclaims live in-progress records as orphans, allowing concurrent double-execution

aws_lambda_powertools/utilities/idempotency/persistence/redis.py

@@ -425,10 +425,12 @@ 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)
+            #
+            # 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)
             ):
                 raise IdempotencyItemAlreadyExistsError
 

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,
 )
@@ -198,6 +200,16 @@ def valid_record():
     )
 
 
+@pytest.fixture
+def in_progress_record_missing_expiry():
+    return DataRecord(
+        idempotency_key="test_orphan_key",
+        status=STATUS_CONSTANTS["INPROGRESS"],
+        expiry_timestamp=int(datetime.datetime.now().timestamp()) + 60,
+        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,73 @@ 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):
+    layer = RedisCachePersistenceLayer(host="host")
+    layer._put_in_progress_record(in_progress_record_missing_expiry)
+
+    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(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)
+
+    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())
 def test_redis_error_in_progress(valid_record):
     layer = RedisCachePersistenceLayer(host="host", mode="standalone")