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.

Author: Adityaj0

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
 

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")