Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
34 changes: 34 additions & 0 deletions tests/functional/idempotency/_redis/test_redis_layer.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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")
Expand Down