From 6169ae97b6c189a3be79d2ab780dc63ac7f11ce0 Mon Sep 17 00:00:00 2001 From: Alexey Shalaev <75322386+AlexeyShalaev@users.noreply.github.com> Date: Fri, 16 Oct 2026 18:53:33 +0400 Subject: [PATCH] fix: renew the reservation while the action runs A pending record expired a lease after its owner reserved the key, however alive the owner was, so a duplicate that arrived meanwhile ("wait" mode included) found the key free and ran the action a second time. The coordinator now renews its reservation while the action runs: a task beside it rewrites the pending record every third of a lease with its expiry a lease ahead. The task is stopped and awaited, not cancelled, before the result is written or the reservation released, on every exit including cancellation, so a renewal still on its way to the store cannot land after either. A failed renewal is counted as storage_renew_error and logged, and the next one tries again; a store down for a whole lease loses the reservation, as before, and the action carries on. Expiry now means the owner is gone, and the lease no longer has to outlive the action. A waiter still waits at most a lease from its arrival and then raises IdempotencyInProgressError. Closes #52 --- docs/agents.md | 37 +-- docs/api_reference.md | 2 +- docs/architecture.md | 13 +- docs/user_guide.md | 26 +- idempotency_kit/core/constants.py | 4 +- .../core/services/aio/coordinator.py | 97 +++++-- idempotency_kit/core/services/domain.py | 4 +- tests/integration/test_reservation_renewal.py | 238 ++++++++++++++++++ tests/unit/core/test_in_flight.py | 131 ++++++++++ 9 files changed, 498 insertions(+), 54 deletions(-) create mode 100644 tests/integration/test_reservation_renewal.py diff --git a/docs/agents.md b/docs/agents.md index fb9a3fe..a305d0e 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -42,8 +42,10 @@ that hides the whole flow, and Dishka providers that wire the pieces together. **It does not** derive the key — the caller supplies it, and the request body is not part of it unless you name the parameters that fingerprint the request; it does not roll anything back; it does not cache failures; it does not retry; it has no sync API; and it stores -nothing but JSON. The reservation is a lease, not a lock: an action -that outlives its lease can run twice, and when storage is down the action runs unreserved. +nothing but JSON. The reservation is a lease, not a lock: the owner renews it while the +action runs, but an owner that cannot renew it for a whole lease (a blocked event loop, a +store that is down) loses it and the action can run twice, and when storage is down the +action runs unreserved. It is a result cache with an in-flight reservation, not a distributed transaction. ## Mental model @@ -64,8 +66,9 @@ Four nouns and one flow. * **`ResultAdapter`** — `encode` a return value into JSON, `decode` it back. Three ship: `PydanticResultAdapter(model_class)`, `JsonResultAdapter()`, `VoidResultAdapter()`. * **`AsyncIdempotencyCoordinator`** — the flow: reserve the key by writing a pending - record under `SET NX` with the lease as its TTL. If that succeeds, run the action, encode - the result and `replace` the reservation with the completed record. If it fails, read + record under `SET NX` with the lease as its TTL. If that succeeds, run the action while + renewing the reservation every third of a lease, encode the result and `replace` the + reservation with the completed record. If it fails, read what holds the key: a record whose `fingerprint` differs from the caller's raises `IdempotencyKeyReuseError`, pending or not; a completed record is decoded and returned; a pending one means another caller is in flight, and `in_flight` decides — `"wait"` polls @@ -240,7 +243,7 @@ second caller records on finding the pending record: a waiter then records a hit result arrives, a refused caller records nothing more and raises. In `"run"` mode it is the loser's `SET NX` failing after both ran, as before. The error types the coordinator reports are `storage_get_error`, `storage_reserve_error`, `storage_save_error`, -`storage_release_error`, `record_validation_error` and `key_reuse`. +`storage_release_error`, `storage_renew_error`, `record_validation_error` and `key_reuse`. ### Settings and Dishka @@ -306,8 +309,13 @@ fields existed is read as enabled, `"wait"` and 30 seconds. before reservations existed: both run, the loser's `SET NX` collides and it adopts the winner's result; the callers see one result and the side effect happened twice, so keep it for actions that are genuinely safe to repeat. The reservation is a lease, not a lock: - a pending record past its lease counts as absent and the next caller runs the action, so - `in_flight_lease_seconds` has to be longer than the action can ever take. + while the action runs, the owner rewrites the pending record every third of a lease with + its expiry a lease ahead, and a pending record past its lease counts as absent and the + next caller runs the action. So the lease does not have to outlive the action; it is how + long a dead owner holds the key, and how long a waiter waits. Renewal runs on the event + loop: an action that blocks the loop for a whole lease loses the key. A failed renewal is + counted as `storage_renew_error` and the next one tries again; a store down for a whole + lease loses the reservation, and the action carries on. 3. **Declare the key keyword-only anyway.** The decorator reads `key_param` from the keyword arguments, and from the positional arguments when the parameter can be passed that way. Keyword-only is still the shape to write: nothing can land in it by position, @@ -459,15 +467,16 @@ except IdempotencyInProgressError: ``` ```python -# WRONG — a lease the action can outlive: the reservation expires mid-run, a retry at -# second six finds the key free, and the provider is charged twice after all -coordinator = AsyncIdempotencyCoordinator(repository, service, in_flight_lease_seconds=5) - +# WRONG — a blocking call inside the action: the event loop cannot renew the reservation, +# it expires after a lease, a retry finds the key free, and the provider is charged twice +@async_idempotent(operation="payment.charge", adapter=PydanticResultAdapter(ChargeDTO)) async def charge(self, dto, *, idempotency_key: str | None = None) -> ChargeDTO: - return await self._psp.charge(dto, timeout=20) + return self._psp_sync_client.charge(dto, timeout=60) -# RIGHT — the lease outlives the action's worst case, timeouts and retries included -coordinator = AsyncIdempotencyCoordinator(repository, service, in_flight_lease_seconds=60) +# RIGHT — keep the loop free while the action waits +@async_idempotent(operation="payment.charge", adapter=PydanticResultAdapter(ChargeDTO)) +async def charge(self, dto, *, idempotency_key: str | None = None) -> ChargeDTO: + return await asyncio.to_thread(self._psp_sync_client.charge, dto, timeout=60) ``` ```python diff --git a/docs/api_reference.md b/docs/api_reference.md index 5a82cc3..08db9d6 100644 --- a/docs/api_reference.md +++ b/docs/api_reference.md @@ -53,7 +53,7 @@ The flow: reserve the key, run the action, store the result; replay a stored res - `metrics` (IdempotencyMetricsProtocol, optional): Metrics collector. - `enabled` (bool, default: True): `False` runs the action and nothing else. - `in_flight` (`"wait" | "raise" | "run"`, default: `"wait"`): What a second caller gets while the first one's action is still running under the same key — the first caller's result once it lands, `IdempotencyInProgressError` at once, or a run of its own with the winner's result adopted on collision. - - `in_flight_lease_seconds` (int, default: 30): How long a reservation is held before it counts as abandoned, and the longest a waiting caller waits. Must be at least 1 and longer than the action can take. + - `in_flight_lease_seconds` (int, default: 30): How long a reservation outlives its owner before it counts as abandoned, and the longest a waiting caller waits. Must be at least 1. The owner renews the reservation every third of a lease while the action runs, so the action may take longer. - *Note*: The constructor raises `TypeError` when the repository has no `replace` and `in_flight` is not `"run"`. - **Methods**: - `coordinate(operation, idempotency_key, ttl_seconds, adapter, action, /, *args, idempotency_fingerprint=None, **kwargs)`: Runs the flow and returns the action's result type. Everything after the five positional-only arguments goes to the action except `idempotency_fingerprint`, the fingerprint of the request, which is stored with the record and compared on a hit. Never raises for storage or decode trouble; raises `IdempotencyInProgressError` when the key is in flight and `in_flight` says so, and `IdempotencyKeyReuseError` when the record's fingerprint differs from the caller's. diff --git a/docs/architecture.md b/docs/architecture.md index 4a50656..faab5e9 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -106,9 +106,12 @@ sequenceDiagram The wait is bounded by the lease: a pending record past its lease counts as absent, so a worker that crashed mid-action cannot hold the key for longer, and a waiter that has spent a -whole lease raises `IdempotencyInProgressError` rather than waiting forever. The other side of -that bound is that the lease has to outlive the action; a reservation that expires mid-run -lets the next caller run the action again. +whole lease raises `IdempotencyInProgressError` rather than waiting forever. A live owner +does not lose the key that way: while its action runs, a task beside it rewrites the pending +record every third of a lease with its expiry a lease ahead. The task is stopped and waited +for, not cancelled, before the result is written or the reservation deleted, so a renewal +still on its way to the store cannot land after either. Expiry therefore means the owner is +gone, or could not reach the store or run its event loop for a whole lease. ### Concurrent Callers with `in_flight="run"` @@ -179,8 +182,8 @@ The library follows the **Idempotency Key Pattern**: 2. Server reserves the key; if the key is already taken, it returns the stored result, or waits for the caller that holds it. 3. If the reservation succeeds, it executes the operation, writes the result over the reservation, and returns it. -What this gives is one execution per key while storage is up and the action finishes within -its lease, and at-least-once when storage is down, because the coordinator then runs the +What this gives is one execution per key while storage is up and the owner keeps renewing +its reservation, and at-least-once when storage is down, because the coordinator then runs the action unreserved rather than failing the request (see Graceful Degradation in the User Guide). It is not exactly-once: the reservation is a lease, not a lock. diff --git a/docs/user_guide.md b/docs/user_guide.md index 473cfc9..1ab677b 100644 --- a/docs/user_guide.md +++ b/docs/user_guide.md @@ -122,12 +122,20 @@ async def charge(dto: ChargeDTO, idempotency_key: str | None = Header(None, alia raise HTTPException(409, detail="a request with this Idempotency-Key is still being processed") ``` -**The lease.** `in_flight_lease_seconds` (default 30) is how long the reservation is held. -A pending record past its lease counts as absent, so a worker that crashed mid-action cannot -wedge the key; it is also how long a waiting caller waits before it gives up with -`IdempotencyInProgressError`. The flip side is that the lease has to be longer than the action -can ever take, timeouts and internal retries included: a reservation that expires while the -action is still running lets the next caller run it again. +**The lease.** `in_flight_lease_seconds` (default 30) is how long the reservation outlives +its owner. While the action runs, the owner renews it every third of a lease, pushing its +expiry a lease ahead, so an action may take longer than the lease. A pending record past its +lease counts as absent, so a worker that crashed mid-action holds the key for at most one +lease; it is also how long a waiting caller waits before it gives up with +`IdempotencyInProgressError`. Pick it for those two: how long a retry should be blocked by a +dead worker, and how long a duplicate should wait for the original's result. + +Renewal is a task on the event loop. An action that blocks the loop (a synchronous client, a +CPU-bound step) for a whole lease stops it, the reservation expires, and the next caller runs +the action again; run such work with `asyncio.to_thread`. A renewal the store rejects is +logged and counted as `storage_renew_error`, and the next one, a third of a lease later, +tries again; a store down for a whole lease loses the reservation, and the action carries on +regardless. **Failures.** An action that raises, or is cancelled, deletes its reservation before the exception propagates, so the retry runs the action again — failures are not cached, and @@ -412,7 +420,7 @@ coordinator = AsyncIdempotencyCoordinator( metrics=custom_metrics, # Optional metrics collector enabled=True, # False runs the action and nothing else in_flight="wait", # "wait" | "raise" | "run", see In-flight requests - in_flight_lease_seconds=30, # How long a reservation is held; must outlive the action + in_flight_lease_seconds=30, # How long a reservation outlives a dead owner; renewed while the action runs ) ``` @@ -428,7 +436,7 @@ from settings and you get the same numbers. | `MIN_TTL_SECONDS` | 60 | the floor every TTL is raised to | | `MAX_TTL_SECONDS` | 2592000 | the ceiling, thirty days | | `DEFAULT_IN_FLIGHT_MODE` | `"wait"` | what a second caller gets while the first is in flight | -| `DEFAULT_IN_FLIGHT_LEASE_SECONDS` | 30 | how long a reservation is held, and the longest a caller waits | +| `DEFAULT_IN_FLIGHT_LEASE_SECONDS` | 30 | how long a reservation outlives its owner, and the longest a caller waits | | `IN_FLIGHT_POLL_INTERVAL_SECONDS` | 0.05 | how often a waiting caller re-reads the key | `MAX_KEY_LENGTH`, `MAX_OPERATION_LENGTH` and the `InFlightMode` type live beside them. @@ -473,7 +481,7 @@ The library defines several exceptions to handle various idempotency scenarios: 1. **Natural Keys**: Use natural unique identifiers as idempotency keys if possible (e.g., `order_id`, `message_id`). 2. **Atomic Operations**: Always save the result to the cache *after* the business logic has successfully completed. -3. **Lease Longer Than the Action**: Set `in_flight_lease_seconds` above the longest the action can take, timeouts and retries included; a reservation that expires mid-run lets the next caller run the action again. +3. **Keep the Event Loop Free**: The reservation is renewed on the event loop while the action runs; an action that blocks the loop for a whole lease loses it and lets the next caller run the action again. Push blocking work to `asyncio.to_thread`. 4. **Fingerprint the Request**: Name the parameters that identify the request in `fingerprint_params`, so a key reused by mistake for a different request is refused with `IdempotencyKeyReuseError` rather than answered with the first result. 5. **Pydantic Support**: The library works best with Pydantic models. Use `model_dump(mode="json")` when saving and `**cached.result` when restoring. 6. **Graceful Degradation**: Decide whether your service should fail if idempotency storage is down. For most high-availability services, it's better to log an error and proceed (at-least-once delivery) than to crash (exactly-once requirement). diff --git a/idempotency_kit/core/constants.py b/idempotency_kit/core/constants.py index b426f37..c09a7c0 100644 --- a/idempotency_kit/core/constants.py +++ b/idempotency_kit/core/constants.py @@ -33,8 +33,8 @@ # instead of producing a second one. DEFAULT_IN_FLIGHT_MODE: InFlightMode = "wait" -# How long a reservation is held before it counts as abandoned (30 seconds). It has to -# outlive the action; a waiting caller gives up after the same span. +# How long a reservation outlives its owner before it counts as abandoned (30 seconds). +# The owner renews it while the action runs; a waiting caller gives up after the same span. DEFAULT_IN_FLIGHT_LEASE_SECONDS: int = 30 # How often a waiting caller re-reads the key. diff --git a/idempotency_kit/core/services/aio/coordinator.py b/idempotency_kit/core/services/aio/coordinator.py index 29d4ccb..cf5ee06 100644 --- a/idempotency_kit/core/services/aio/coordinator.py +++ b/idempotency_kit/core/services/aio/coordinator.py @@ -1,10 +1,12 @@ import asyncio import logging import time -from collections.abc import Awaitable, Callable +from collections.abc import AsyncIterator, Awaitable, Callable +from contextlib import asynccontextmanager from dataclasses import dataclass +from datetime import UTC, datetime, timedelta from enum import Enum, auto -from typing import Any, Generic, TypeVar +from typing import Any, Generic, Literal, TypeVar from idempotency_kit.core.constants import ( DEFAULT_IN_FLIGHT_LEASE_SECONDS, @@ -69,8 +71,9 @@ class AsyncIdempotencyCoordinator: under the same key. ``"wait"`` (the default) waits for the first caller's result, ``"raise"`` raises ``IdempotencyInProgressError`` at once, and ``"run"`` runs the action too, as the coordinator did before reservations existed. - in_flight_lease_seconds: How long a reservation is held before it counts as - abandoned, and the longest a waiting caller waits. It has to outlive the action. + in_flight_lease_seconds: How long a reservation outlives its owner before it counts + as abandoned, and the longest a waiting caller waits. The owner renews it every + third of a lease while the action runs, so it does not have to outlive the action. """ def __init__( @@ -144,11 +147,12 @@ async def coordinate( return claim.value try: - result = await action(*args, **kwargs) + async with self._renewing(claim): + result = await action(*args, **kwargs) except BaseException: # Failures are not cached, so the reservation goes too and the retry runs # again. Cancellation counts: nothing says the action completed. - if claim: + if claim is not None: await self._try_release(operation, idempotency_key) raise @@ -156,7 +160,7 @@ async def coordinate( completed = await self._try_complete( operation, idempotency_key, result, adapter, ttl_minutes, idempotency_fingerprint ) - if claim and not completed: + if claim is not None and not completed: # Our reservation with no result behind it would make every retry within the # lease wait for, or be refused over, a record that is never coming. await self._try_release(operation, idempotency_key) @@ -192,14 +196,15 @@ async def _claim( idempotency_key: str, adapter: ResultAdapter[T], fingerprint: str | None, - ) -> _Hit[T] | bool: + ) -> _Hit[T] | IdempotencyRecord | None: """Reserve the key, or replay the record that holds it. - Returns the hit when a completed record is there, ``True`` when the reservation is - ours, and ``False`` when the key holds nothing usable and the action has to run - unreserved. Raises ``IdempotencyInProgressError`` for a key another caller holds, - at once in ``"raise"`` mode and after a whole lease of waiting in ``"wait"`` mode, - and ``IdempotencyKeyReuseError`` for a key that holds a different request. + Returns the hit when a completed record is there, our pending record when the + reservation is ours, and ``None`` when the key holds nothing usable and the action + has to run unreserved. Raises ``IdempotencyInProgressError`` for a key another + caller holds, at once in ``"raise"`` mode and after a whole lease of waiting in + ``"wait"`` mode, and ``IdempotencyKeyReuseError`` for a key that holds a different + request. """ deadline = time.monotonic() + self._in_flight_lease_seconds read = False @@ -207,19 +212,19 @@ async def _claim( while True: reserved = await self._try_reserve(operation, idempotency_key, fingerprint) if reserved is None: - return False - if reserved: + return None + if reserved is not False: # A read that came before us has counted the miss already. if not read: self._metrics.record_miss(operation) - return True + return reserved found = await self._try_get_cached(operation, idempotency_key, adapter, fingerprint) read = True if isinstance(found, _Hit): return found if found is _Lookup.UNUSABLE: - return False + return None if found is _Lookup.IN_FLIGHT: if not waiting: waiting = True @@ -253,11 +258,13 @@ def _resolve_ttl_minutes(self, operation: str, ttl_seconds: int | None) -> int | return None return max(1, effective_ttl_seconds // 60) - async def _try_reserve(self, operation: str, idempotency_key: str, fingerprint: str | None) -> bool | None: + async def _try_reserve( + self, operation: str, idempotency_key: str, fingerprint: str | None + ) -> IdempotencyRecord | Literal[False] | None: """Write the pending record under ``SET NX``. - ``True`` when the reservation is ours, ``False`` when the key is already taken, - ``None`` when the write could not be made at all. + The pending record when the reservation is ours, ``False`` when the key is already + taken, ``None`` when the write could not be made at all. """ start_time = time.perf_counter() try: @@ -285,10 +292,58 @@ async def _try_reserve(self, operation: str, idempotency_key: str, fingerprint: ) return None else: - return True + return pending finally: self._metrics.record_latency(operation, "reserve", time.perf_counter() - start_time) + @asynccontextmanager + async def _renewing(self, reservation: IdempotencyRecord | None) -> AsyncIterator[None]: + """Keep our reservation from expiring while the body runs; nothing to do without one. + + On the way out, by any exit, the renewal is told to stop and waited for, not + cancelled: a renewal write cancelled in flight could still land after the result or + the release that follows, and put a pending record back over it. + """ + if reservation is None: + yield + return + stop = asyncio.Event() + renewal = asyncio.create_task(self._renew(reservation, stop)) + try: + yield + finally: + stop.set() + await asyncio.wait({renewal}) + + async def _renew(self, reservation: IdempotencyRecord, stop: asyncio.Event) -> None: + """Push the reservation's expiry a lease ahead every third of a lease, until told to stop.""" + interval = self._in_flight_lease_seconds / 3 + while True: + try: + await asyncio.wait_for(stop.wait(), interval) + except TimeoutError: + await self._try_renew(reservation) + else: + return + + async def _try_renew(self, reservation: IdempotencyRecord) -> None: + """Rewrite our pending record with its expiry a lease from now; storage trouble is logged, not raised. + + One failure is absorbed by the next attempt, two by the rest of the lease; a store + down for longer loses the reservation, and the action runs on regardless. + """ + renewed = reservation.model_copy( + update={"expires_at": datetime.now(UTC) + timedelta(seconds=self._in_flight_lease_seconds)} + ) + try: + await self._repo.replace(renewed) + except Exception: + self._metrics.record_error(reservation.operation, "storage_renew_error") + logger.exception( + "Idempotency coordinator error while renewing a reservation", + extra={"operation": reservation.operation, "idempotency_key": reservation.idempotency_key}, + ) + async def _try_release(self, operation: str, idempotency_key: str) -> None: """Delete our pending record so the retry runs the action again; storage trouble is logged, not raised.""" try: diff --git a/idempotency_kit/core/services/domain.py b/idempotency_kit/core/services/domain.py index 95ede18..b9eadec 100644 --- a/idempotency_kit/core/services/domain.py +++ b/idempotency_kit/core/services/domain.py @@ -102,8 +102,8 @@ def create_pending_record( ) -> IdempotencyRecord: """Create the in-flight reservation for an operation whose action is about to run. - The lease is not a record TTL and is not held to the TTL bounds: it only has to - outlive the action, and the coordinator decides it. + The lease is not a record TTL and is not held to the TTL bounds: the coordinator + decides it, and renews the reservation while the action runs. Args: operation: Operation name (e.g., 'user.create') diff --git a/tests/integration/test_reservation_renewal.py b/tests/integration/test_reservation_renewal.py new file mode 100644 index 0000000..43f3384 --- /dev/null +++ b/tests/integration/test_reservation_renewal.py @@ -0,0 +1,238 @@ +"""Reservation renewal against real Redis, with a lease short enough to run out during the test. + +Regression cover for issue #52: the pending record expired a lease after the owner started, +however alive the owner was, and a duplicate that arrived meanwhile reserved the key and ran +the action a second time. +""" + +import asyncio +import subprocess +import sys +import textwrap +from typing import Any +from uuid import uuid4 + +import pytest +from redis.asyncio import Redis as AsyncRedisClient +from testcontainers.redis import RedisContainer + +from idempotency_kit import ( + AsyncIdempotencyCoordinator, + IdempotencyDomainService, + IdempotencyInProgressError, + JsonResultAdapter, +) +from idempotency_kit.core.constants import InFlightMode +from idempotency_kit.infra.storage.redis.aio import RedisAsyncIdempotencyRepository + +LEASE_SECONDS = 1 + +# A process that reserves a key with a one-second lease and runs an action far longer than +# that. It prints a line once the action has started, then either sleeps until killed or +# returns from main() while the action is still running, so that asyncio.run() cancels it. +_OWNER_SCRIPT = textwrap.dedent( + """ + import asyncio + import sys + + from redis.asyncio import Redis + + from idempotency_kit import AsyncIdempotencyCoordinator, IdempotencyDomainService, JsonResultAdapter + from idempotency_kit.infra.storage.redis.aio import RedisAsyncIdempotencyRepository + + + async def main(host, port, key, mode): + redis = Redis(host=host, port=int(port)) + coordinator = AsyncIdempotencyCoordinator( + RedisAsyncIdempotencyRepository(redis), IdempotencyDomainService(), in_flight_lease_seconds=1 + ) + started = asyncio.Event() + + async def charge(): + started.set() + await asyncio.sleep(30) + return "owner" + + task = asyncio.create_task(coordinator.coordinate("payment.charge", key, 3600, JsonResultAdapter(), charge)) + await started.wait() + print("started", flush=True) + if mode == "crash": + await asyncio.sleep(30) + else: + await asyncio.sleep(0.5) # past one renewal, then shut down with the action running + + + asyncio.run(main(*sys.argv[1:])) + """ +) + + +def _start_owner(redis_container: RedisContainer, key: str, mode: str) -> subprocess.Popen[str]: + host = redis_container.get_container_host_ip() + port = redis_container.get_exposed_port(6379) + owner = subprocess.Popen( # noqa: S603 -- our own interpreter and script + [sys.executable, "-c", _OWNER_SCRIPT, host, str(port), key, mode], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + assert owner.stdout is not None + assert owner.stdout.readline().strip() == "started" + return owner + + +def _coordinator(redis_client: AsyncRedisClient, in_flight: InFlightMode = "wait") -> AsyncIdempotencyCoordinator: + return AsyncIdempotencyCoordinator( + RedisAsyncIdempotencyRepository(redis_client), + IdempotencyDomainService(), + in_flight=in_flight, + in_flight_lease_seconds=LEASE_SECONDS, + ) + + +@pytest.mark.asyncio +async def test__coordinator__waiter_outlives_the_first_lease_on_real_redis__is_refused_and_the_action_runs_once( + redis_client: AsyncRedisClient, +) -> None: + """The reporter's scenario: a one-second lease, a three-second charge, and a duplicate 0.1 s in.""" + # Arrange + coordinator = _coordinator(redis_client) + key = str(uuid4()) + charges: list[str] = [] + + async def charge(tag: str, hold: float) -> dict[str, str]: + charges.append(tag) + await asyncio.sleep(hold) + return {"charge_id": tag} + + # Act + first = asyncio.create_task( + coordinator.coordinate("payment.charge", key, 3600, JsonResultAdapter(), charge, "first", 3) + ) + await asyncio.sleep(0.1) + with pytest.raises(IdempotencyInProgressError): + await coordinator.coordinate("payment.charge", key, 3600, JsonResultAdapter(), charge, "second", 0) + result = await first + replay = await coordinator.coordinate("payment.charge", key, 3600, JsonResultAdapter(), charge, "third", 0) + + # Assert + assert charges == ["first"] + assert result == replay == {"charge_id": "first"} + + +@pytest.mark.asyncio +async def test__coordinator__callers_after_the_first_lease_on_real_redis__find_the_key_still_reserved( + redis_client: AsyncRedisClient, +) -> None: + """Each lease the owner outlives is renewed: callers at 1.5 s and 2.5 s still see it in flight.""" + # Arrange + owner = _coordinator(redis_client) + latecomer = _coordinator(redis_client, in_flight="raise") + key = str(uuid4()) + runs: list[str] = [] + + async def action(tag: str, hold: float) -> str: + runs.append(tag) + await asyncio.sleep(hold) + return tag + + # Act + first = asyncio.create_task(owner.coordinate("op", key, 3600, JsonResultAdapter(), action, "owner", 3)) + await asyncio.sleep(1.5) + with pytest.raises(IdempotencyInProgressError): + await latecomer.coordinate("op", key, 3600, JsonResultAdapter(), action, "at 1.5 s", 0) + await asyncio.sleep(1) + with pytest.raises(IdempotencyInProgressError): + await latecomer.coordinate("op", key, 3600, JsonResultAdapter(), action, "at 2.5 s", 0) + result = await first + + # Assert + assert runs == ["owner"] + assert result == "owner" + assert await redis_client.ttl(f"idempotency:op:{key}") > 60 # the result's TTL, not a lease + + +@pytest.mark.asyncio +async def test__coordinator__owner_process_killed_on_real_redis__frees_the_key_within_a_lease( + redis_client: AsyncRedisClient, redis_container: RedisContainer +) -> None: + """Renewal lives and dies with the owner: kill it, and the next caller runs the action a lease later.""" + # Arrange + key = str(uuid4()) + owner = _start_owner(redis_container, key, "crash") + coordinator = _coordinator(redis_client, in_flight="raise") + action_ran = False + + async def action() -> str: + nonlocal action_ran + action_ran = True + return "recovered" + + # Act + await asyncio.sleep(1.5) # past the first lease: the owner is renewing + with pytest.raises(IdempotencyInProgressError): + await coordinator.coordinate("payment.charge", key, 3600, JsonResultAdapter(), action) + owner.kill() + owner.wait() + await asyncio.sleep(LEASE_SECONDS + 0.2) + result = await coordinator.coordinate("payment.charge", key, 3600, JsonResultAdapter(), action) + + # Assert + assert action_ran + assert result == "recovered" + + +@pytest.mark.asyncio +async def test__coordinator__owner_process_shuts_down_mid_action_on_real_redis__releases_the_key_cleanly( + redis_client: AsyncRedisClient, redis_container: RedisContainer +) -> None: + """asyncio.run() cancelling the action and the renewal together: the key is freed and nothing leaks.""" + # Arrange + key = str(uuid4()) + + # Act + owner = _start_owner(redis_container, key, "shutdown") + _, stderr = owner.communicate(timeout=10) + await asyncio.sleep(LEASE_SECONDS / 3 + 0.2) # long enough for a stray renewal to land + + # Assert + assert owner.returncode == 0, stderr + assert "Task was destroyed" not in stderr + assert "never retrieved" not in stderr + assert await redis_client.exists(f"idempotency:payment.charge:{key}") == 0 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("outcome", ["returns", "raises", "is_cancelled"]) +async def test__coordinator__action_ends_on_real_redis__no_renewal_lands_afterwards( + redis_client: AsyncRedisClient, outcome: str +) -> None: + """The renewal is over before the result is written or the key released, whichever way the action ends.""" + # Arrange + coordinator = _coordinator(redis_client) + key = str(uuid4()) + storage_key = f"idempotency:op:{key}" + + async def action() -> str: + await asyncio.sleep(LEASE_SECONDS * 1.2) # several renewals + if outcome == "raises": + raise RuntimeError("provider down") + return "done" + + # Act + task = asyncio.create_task(coordinator.coordinate("op", key, 3600, JsonResultAdapter(), action)) + if outcome == "is_cancelled": + await asyncio.sleep(LEASE_SECONDS * 0.8) + task.cancel() + outcomes: list[Any] = await asyncio.gather(task, return_exceptions=True) + await asyncio.sleep(LEASE_SECONDS / 3 + 0.2) + + # Assert + if outcome == "returns": + assert outcomes == ["done"] + stored = await RedisAsyncIdempotencyRepository(redis_client).get("op", key) + assert stored is not None + assert stored.status == "completed" + else: + assert isinstance(outcomes[0], RuntimeError | asyncio.CancelledError) + assert await redis_client.exists(storage_key) == 0 diff --git a/tests/unit/core/test_in_flight.py b/tests/unit/core/test_in_flight.py index 604d91b..04fc709 100644 --- a/tests/unit/core/test_in_flight.py +++ b/tests/unit/core/test_in_flight.py @@ -21,6 +21,7 @@ IdempotencyKeyCollisionError, IdempotencyMetricsProtocol, IdempotencyRecord, + IdempotencyStorageError, IdempotencyValidationError, JsonResultAdapter, ) @@ -394,3 +395,133 @@ def test__coordinator__repository_without_replace__raises_type_error_unless_in_f AsyncIdempotencyCoordinator( repository=legacy_repository, domain_service=IdempotencyDomainService(), in_flight="run" ) + + +class _RenewalFailingRepository(RedisAsyncIdempotencyRepository): + """Fails the first ``fail`` renewals -- a ``replace`` of a pending record; completions go through.""" + + def __init__(self, redis: AsyncRedisClient, *, fail: int) -> None: + super().__init__(redis, key_prefix="probe:") + self.renewals = 0 + self._fail = fail + + async def replace(self, record: IdempotencyRecord) -> None: + if record.is_pending: + self.renewals += 1 + if self.renewals <= self._fail: + raise IdempotencyStorageError("redis down", operation=record.operation) + await super().replace(record) + + +async def _owner_and_a_latecomer( + repository: RedisAsyncIdempotencyRepository, *, hold: float, late_by: float, metrics: Any = None +) -> tuple[list[str], list[str | BaseException]]: + """An owner with a one-second lease holding the key for ``hold`` seconds, and a caller arriving ``late_by`` in.""" + owner = AsyncIdempotencyCoordinator( + repository=repository, domain_service=IdempotencyDomainService(), in_flight_lease_seconds=1, metrics=metrics + ) + latecomer = AsyncIdempotencyCoordinator( + repository=repository, domain_service=IdempotencyDomainService(), in_flight="raise", in_flight_lease_seconds=1 + ) + runs: list[str] = [] + + async def action(tag: str, seconds: float) -> str: + runs.append(tag) + await asyncio.sleep(seconds) + return tag + + first = asyncio.create_task(owner.coordinate("op", "key", 600, JsonResultAdapter(), action, "owner", hold)) + await asyncio.sleep(late_by) + second = asyncio.create_task(latecomer.coordinate("op", "key", 600, JsonResultAdapter(), action, "latecomer", 0)) + results: list[str | BaseException] = await asyncio.gather(first, second, return_exceptions=True) + return runs, results + + +@pytest.mark.asyncio +async def test__coordinator__action_outlives_its_lease__the_reservation_is_renewed_and_a_latecomer_is_refused( + redis_repository: RedisAsyncIdempotencyRepository, +) -> None: + """Issue #52: a live owner's reservation no longer expires a lease after it started.""" + # Act + runs, results = await _owner_and_a_latecomer(redis_repository, hold=1.6, late_by=1.3) + + # Assert + assert runs == ["owner"] + assert results[0] == "owner" + assert isinstance(results[1], IdempotencyInProgressError) + + +@pytest.mark.asyncio +async def test__coordinator__a_renewal_fails__it_is_counted_and_the_next_one_keeps_the_reservation( + fake_redis: AsyncRedisClient, +) -> None: + """One lost renewal is absorbed: the next comes a third of a lease later, well before the expiry.""" + # Arrange + repository = _RenewalFailingRepository(fake_redis, fail=1) + metrics = MagicMock(spec=IdempotencyMetricsProtocol) + + # Act + runs, results = await _owner_and_a_latecomer(repository, hold=1.6, late_by=1.3, metrics=metrics) + + # Assert + assert runs == ["owner"] + assert results[0] == "owner" + assert isinstance(results[1], IdempotencyInProgressError) + metrics.record_error.assert_called_once_with("op", "storage_renew_error") + + +@pytest.mark.asyncio +async def test__coordinator__every_renewal_fails__the_reservation_lapses_and_the_owner_still_completes( + fake_redis: AsyncRedisClient, +) -> None: + """A store down for a whole lease loses the reservation, as before renewal; the owner's action is not cut short.""" + # Arrange + repository = _RenewalFailingRepository(fake_redis, fail=100) + + # Act + runs, results = await _owner_and_a_latecomer(repository, hold=1.6, late_by=1.3) + + # Assert + assert runs == ["owner", "latecomer"] + assert results == ["owner", "latecomer"] + assert repository.renewals >= 3 + + +class _SlowRenewalRepository(RedisAsyncIdempotencyRepository): + """A renewal that is already on the wire: it lands 0.3 s later even if the caller stops waiting for it.""" + + async def replace(self, record: IdempotencyRecord) -> None: + if record.is_pending: + await asyncio.shield(self._land_later(record)) + return + await super().replace(record) + + async def _land_later(self, record: IdempotencyRecord) -> None: + await asyncio.sleep(0.3) + await super().replace(record) + + +@pytest.mark.asyncio +async def test__coordinator__renewal_in_flight_when_the_action_ends__lands_before_the_result_not_over_it( + fake_redis: AsyncRedisClient, +) -> None: + """Cancelling the renewal would let its write land after the completed record and put the key back in flight.""" + # Arrange + repository = _SlowRenewalRepository(fake_redis, key_prefix="probe:") + coordinator = AsyncIdempotencyCoordinator( + repository=repository, domain_service=IdempotencyDomainService(), in_flight_lease_seconds=1 + ) + + async def action() -> str: + await asyncio.sleep(0.5) # ends while the renewal sent at 1/3 s is still on its way + return "done" + + # Act + result = await coordinator.coordinate("op", "key", 600, JsonResultAdapter(), action) + await asyncio.sleep(0.4) + + # Assert + assert result == "done" + stored = await repository.get("op", "key") + assert stored is not None + assert (stored.status, stored.result) == ("completed", "done")