diff --git a/docs/agents.md b/docs/agents.md index 9b8252f..aabb62a 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -77,7 +77,8 @@ Four nouns and one flow. has passed is a miss even if the repository handed it back; an expired lease is an abandoned reservation. An action that raises deletes its reservation. Storage and decode failures are swallowed and the action runs, which is the deliberate trade of exactly-once - for availability. + for availability; `on_storage_error="raise"` turns a storage failure before the action + into `IdempotencyStorageError` instead. `IdempotencyDomainService` sits between the coordinator and the record: it applies the TTL bounds, builds the pending record for a reservation, and turns Pydantic validation errors @@ -155,7 +156,7 @@ result = await coordinator.coordinate( |---|---|---| | `async_idempotent` | `(operation, adapter, ttl_seconds=None, key_param="idempotency_key", infra_param=None, fingerprint_params=None)` | decorator for an async function or method; `fingerprint_params` names the parameters whose values identify the request | | `fingerprint_of` | `(**values)` | SHA-256 hex of the JSON form of the named values, keys sorted; what the decorator computes from `fingerprint_params`, for a caller of `coordinate()` | -| `AsyncIdempotencyCoordinator` | `(repository, domain_service, operation_ttls=None, metrics=None, enabled=True, in_flight="wait", in_flight_lease_seconds=30)` | the flow; `operation_ttls` is `dict[str, int]` in seconds; `enabled=False` runs the action and nothing else; `in_flight` is `"wait"`, `"raise"` or `"run"` | +| `AsyncIdempotencyCoordinator` | `(repository, domain_service, operation_ttls=None, metrics=None, enabled=True, in_flight="wait", in_flight_lease_seconds=30, on_storage_error="run")` | the flow; `operation_ttls` is `dict[str, int]` in seconds; `enabled=False` runs the action and nothing else; `in_flight` is `"wait"`, `"raise"` or `"run"`; `on_storage_error` is `"run"` or `"raise"` | | `IdempotencyDomainService` | `(*, default_ttl_minutes=60, min_ttl_seconds=60, max_ttl_seconds=2592000)` | record factory and TTL bounds; keyword-only, defaults from `core.constants` | | `IdempotencyRecord` | frozen Pydantic model | the cached result | | `IdempotencyIdentifiers` | Pydantic model | `operation` + `idempotency_key`, and the rules they obey | @@ -176,13 +177,13 @@ result = await coordinator.coordinate( | `PrometheusIdempotencyMetrics`, `get_idempotency_metrics` | `idempotency_kit.infra.metrics.prometheus` | | `BaseIdempotencySettings` | `idempotency_kit.settings` | | `IdempotencyProvider`, `AsyncIdempotencyCoordinatorProvider`, `AsyncRedisIdempotencyProvider`, `IdempotencySettingsProtocol` | `idempotency_kit.dishka` | -| `MAX_KEY_LENGTH`, `MAX_OPERATION_LENGTH`, `DEFAULT_TTL_MINUTES`, `MIN_TTL_SECONDS`, `MAX_TTL_SECONDS`, `InFlightMode`, `DEFAULT_IN_FLIGHT_MODE`, `DEFAULT_IN_FLIGHT_LEASE_SECONDS`, `IN_FLIGHT_POLL_INTERVAL_SECONDS` | `idempotency_kit.core.constants` | +| `MAX_KEY_LENGTH`, `MAX_OPERATION_LENGTH`, `DEFAULT_TTL_MINUTES`, `MIN_TTL_SECONDS`, `MAX_TTL_SECONDS`, `InFlightMode`, `DEFAULT_IN_FLIGHT_MODE`, `DEFAULT_IN_FLIGHT_LEASE_SECONDS`, `IN_FLIGHT_POLL_INTERVAL_SECONDS`, `StorageErrorMode`, `DEFAULT_ON_STORAGE_ERROR` | `idempotency_kit.core.constants` | ### Coordinator and domain service | Method | Returns | Notes | |---|---|---| -| `AsyncIdempotencyCoordinator.coordinate(operation, idempotency_key, ttl_seconds, adapter, action, /, *args, idempotency_fingerprint=None, **kwargs)` | `T` | never raises for storage or decode trouble; raises `IdempotencyValidationError` when the operation or a non-`None` key is invalid, `IdempotencyInProgressError` when the key is in flight and `in_flight` says so, `IdempotencyKeyReuseError` when the record's fingerprint differs from `idempotency_fingerprint` | +| `AsyncIdempotencyCoordinator.coordinate(operation, idempotency_key, ttl_seconds, adapter, action, /, *args, idempotency_fingerprint=None, **kwargs)` | `T` | never raises for decode trouble, nor for storage trouble unless `on_storage_error="raise"` (then `IdempotencyStorageError`, before the action runs); raises `IdempotencyValidationError` when the operation or a non-`None` key is invalid, `IdempotencyInProgressError` when the key is in flight and `in_flight` says so, `IdempotencyKeyReuseError` when the record's fingerprint differs from `idempotency_fingerprint` | | `IdempotencyDomainService.create_record(operation, idempotency_key, result, *, ttl_minutes=None, fingerprint=None)` | `IdempotencyRecord` | raises `IdempotencyInvalidTTLError`, `IdempotencyValidationError` | | `IdempotencyDomainService.create_pending_record(operation, idempotency_key, *, lease_seconds, fingerprint=None)` | `IdempotencyRecord` | the reservation; not held to the TTL bounds; raises `IdempotencyValidationError` | | `IdempotencyDomainService.validate_record(record)` | `None` | raises `IdempotencyRecordExpiredError`; the coordinator calls it on every record it reads | @@ -249,8 +250,9 @@ are `storage_get_error`, `storage_reserve_error`, `storage_save_error`, `BaseIdempotencySettings` is a plain Pydantic model with `enabled=True`, `key_prefix` (required, no default), `metrics_enabled=False`, `default_ttl_minutes=60`, -`min_ttl_seconds=60`, `max_ttl_seconds=2592000`, `operation_ttls={}`, `in_flight="wait"` -and `in_flight_lease_seconds=30` — the TTL and in-flight fields default to the +`min_ttl_seconds=60`, `max_ttl_seconds=2592000`, `operation_ttls={}`, `in_flight="wait"`, +`in_flight_lease_seconds=30` and `on_storage_error="run"` — the TTL, in-flight and +storage-error fields default to the `core.constants` values, which is also what `IdempotencyDomainService` and the coordinator use. It satisfies `IdempotencySettingsProtocol`, which is what the providers ask for. @@ -282,9 +284,10 @@ process, under the key `IdempotencyMetricsProtocol` the other two ask for — `get_idempotency_metrics()` when `settings.metrics_enabled`, a no-op otherwise — so a container rebuilt per test gets the instance the first one registered. Another backend, or a metric prefix, goes in with `@provide(override=True)` in a provider listed after it. -`settings.enabled`, `settings.in_flight` and `settings.in_flight_lease_seconds` reach the -coordinator: `enabled=False` makes it a pass-through. A settings object written before those -fields existed is read as enabled, `"wait"` and 30 seconds. +`settings.enabled`, `settings.in_flight`, `settings.in_flight_lease_seconds` and +`settings.on_storage_error` reach the coordinator: `enabled=False` makes it a pass-through. +A settings object written before those fields existed is read as enabled, `"wait"`, 30 +seconds and `"run"`. ## Rules that hold or break the code @@ -333,13 +336,18 @@ fields existed is read as enabled, `"wait"` and 30 seconds. `idempotency_kit.core.decorators.aio.idempotent` and calls the function anyway — the operation stays available, unprotected. Pass `infra_param=` so a renamed attribute is one grep away, and alert on that warning. -6. **`coordinate()` never raises for storage trouble.** A Redis failure on read is counted - as `storage_get_error` and treated as a miss; a failure on the reservation is - `storage_reserve_error` and the action runs unreserved; a failure on write is - `storage_save_error` and the fresh result is returned uncached. Availability over - exactly-once, deliberately. Repository methods called directly do raise — only the - coordinator and the decorator swallow. The one thing they do raise is - `IdempotencyInProgressError`, which is about the caller's request, not about storage. +6. **`coordinate()` does not raise for storage trouble unless you ask it to.** A Redis + failure on read is counted as `storage_get_error` and treated as a miss; a failure on the + reservation is `storage_reserve_error` and the action runs unreserved; a failure on write + is `storage_save_error` and the fresh result is returned uncached. Availability over + exactly-once, deliberately, and the default. `on_storage_error="raise"` fails closed + instead: a failure on the reservation or the read, before the action runs, is still + counted and logged, and then raises `IdempotencyStorageError` — the repository's own, + or one wrapping a driver exception a custom repository let through — and the action does + not run; 503 in HTTP terms. A failure once the action has started (a renewal, the + write, the release) is swallowed in both modes, because the effect may have happened. A record the store returned + but nobody can read is not an outage in either mode (rule 7). Repository methods called + directly always raise. 7. **A decode failure is a miss, every time.** If the adapter cannot decode a stored record, the coordinator logs and runs the action again. Changing the adapter or the DTO's shape while records are live re-executes the operation for every caller until those records @@ -507,15 +515,19 @@ except IdempotencyKeyReuseError: ``` ```python -# WRONG — expecting the coordinator to tell you Redis is down +# WRONG — expecting the default coordinator to tell you Redis is down try: result = await coordinator.coordinate("order.create", key, 3600, adapter, action, dto) except IdempotencyStorageError: ... # never reached -# RIGHT — the coordinator degrades to running the action; watch the metrics and the -# logs for it, or call the repository yourself when you need the failure -record = await repository.get("order.create", key) # this one raises +# RIGHT — ask for it where running unprotected is worse than refusing +coordinator = AsyncIdempotencyCoordinator(repository, service, on_storage_error="raise") + +try: + return await coordinator.coordinate("payment.charge", key, 3600, adapter, action, dto) +except IdempotencyStorageError: + raise HTTPException(503, detail="try again later") # the action did not run ``` ```python @@ -546,15 +558,16 @@ All derive from `IdempotencyError`, which is exported alongside them. | `IdempotencyRecordExpiredError` | `(operation, key)` | `validate_record` was given an expired record. The coordinator raises it internally and turns it into a miss; through the repository or your own call to `validate_record` you meet it directly | | `IdempotencyInProgressError` | `(operation, key)` | another call with the same key is still running its action. `coordinate()` and the decorator raise it at once with `in_flight="raise"`, and after a whole lease of waiting with `"wait"`. Carries `.operation` and `.key` | | `IdempotencyKeyReuseError` | `(operation, key, stored_fingerprint, fingerprint)` | the record under the key was made for a different request. `coordinate()` and the decorator raise it before the action runs, from a completed record and from a pending one. Carries all four | -| `IdempotencyStorageError` | `(message, operation=None, original_error=None)` | the backend failed. Carries `.operation` and `.original_error` | +| `IdempotencyStorageError` | `(message, operation=None, original_error=None)` | the backend failed. `coordinate()` and the decorator raise it only with `on_storage_error="raise"`, before the action runs. Carries `.operation` and `.original_error` | | `IdempotencyValidationError` | `(message, errors=None)` | an identifier or a result failed validation. `coordinate()` and the decorator raise it for an invalid operation or key before the action runs. `.errors` holds the Pydantic error list when there is one | | `IdempotencyInvalidTTLError` | `(ttl_seconds, min_ttl, max_ttl)` | the TTL is outside the domain service's range. Carries all three | Through `coordinate()` and the decorator only `IdempotencyValidationError` for the identifiers, `IdempotencyInProgressError` and `IdempotencyKeyReuseError` reach the caller — -all three are about the request, not about storage: -the collision is resolved into a wait or the winner's result, and every other one is logged, -counted and swallowed. The rest are the contract of the repository and the domain +all three are about the request, not about storage — and `IdempotencyStorageError` when +`on_storage_error="raise"` asks for it: the collision is resolved into a wait or the +winner's result, and every other one is logged, counted and swallowed. The rest are the +contract of the repository and the domain service, which is where you meet them if you drive those directly. ## Documentation map diff --git a/docs/api_reference.md b/docs/api_reference.md index ac4ffde..eb3fb0f 100644 --- a/docs/api_reference.md +++ b/docs/api_reference.md @@ -54,9 +54,10 @@ The flow: reserve the key, run the action, store the result; replay a stored res - `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 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. + - `on_storage_error` (`"run" | "raise"`, default: `"run"`): What a call gets when the store fails before the action runs, on the reservation or the read — the action run unprotected, or `IdempotencyStorageError` and no run. A failure once the action has started (renewing the reservation, saving the result, releasing it) is logged in both modes. - *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. A `None` key runs the action and nothing else. Never raises for storage or decode trouble; raises `IdempotencyValidationError` when the operation or a non-`None` key (the empty string included) is invalid, before the action runs, `IdempotencyInProgressError` when the key is in flight and `in_flight` says so, and `IdempotencyKeyReuseError` when the record's fingerprint differs from the caller's. + - `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. A `None` key runs the action and nothing else. Never raises for decode trouble, nor for storage trouble unless `on_storage_error="raise"`, which raises `IdempotencyStorageError` before the action runs; raises `IdempotencyValidationError` when the operation or a non-`None` key (the empty string included) is invalid, before the action runs, `IdempotencyInProgressError` when the key is in flight and `in_flight` says so, and `IdempotencyKeyReuseError` when the record's fingerprint differs from the caller's. ### async_idempotent The same flow as a decorator: it finds the key in the call's arguments and the coordinator in the call's arguments or on `self`, then delegates to `coordinate()`. diff --git a/docs/architecture.md b/docs/architecture.md index faab5e9..3fb09d5 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -184,8 +184,9 @@ The library follows the **Idempotency Key Pattern**: 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. +action unreserved rather than failing the request — or, with `on_storage_error="raise"`, +no execution and an `IdempotencyStorageError` the caller turns into a 503 (see Failing +closed in the User Guide). It is not exactly-once: the reservation is a lease, not a lock. ## Redis Key Format diff --git a/docs/user_guide.md b/docs/user_guide.md index 0de64e8..ad0b790 100644 --- a/docs/user_guide.md +++ b/docs/user_guide.md @@ -145,9 +145,8 @@ result, and the key is freed rather than holding retries for a record that never **Storage trouble.** A reservation that cannot be written is logged and counted as `storage_reserve_error`, and the action runs unreserved — the same trade of exactly-once for -availability the coordinator makes everywhere else. `IdempotencyInProgressError` is the one -exception `coordinate()` and the decorator do raise: it is about the caller's request, not -about storage. +availability the coordinator makes everywhere else, unless `on_storage_error="raise"` says +otherwise (see [Failing closed](#failing-closed-when-the-store-is-down)). **Metrics.** The first caller is a miss; the second caller records a collision when it finds the reservation, then a hit when it gets the record (`"wait"`) or nothing more (`"raise"`, @@ -305,6 +304,38 @@ deleted_count = await idempotency_repo.delete_many("user.create", [key1, key2, k Distinguishing between different error types allows for robust integration. +### Failing closed when the store is down + +By default the coordinator runs the action when it cannot reach the store, so a Redis +outage costs exactly-once, not availability. That is the wrong trade for an effect that is +expensive or impossible to repeat when nothing downstream deduplicates it: a charge at a +provider without its own idempotency keys, an email, a ledger write in another system. For +those, build the coordinator with `on_storage_error="raise"`: + +```python +coordinator = AsyncIdempotencyCoordinator(repo, service, on_storage_error="raise") + +@app.post("/charges") +async def charge(dto: ChargeRequest, idempotency_key: str | None = Header(None, alias="Idempotency-Key")): + try: + return await use_case.execute(dto, idempotency_key=idempotency_key) + except IdempotencyStorageError: + raise HTTPException(503, detail="try again later") +``` + +| Where the store fails | `on_storage_error="run"` (default) | `on_storage_error="raise"` | +|---|---|---| +| reserving the key, or reading it | logged, counted, the action runs unprotected | logged, counted, `IdempotencyStorageError`; the action does not run | +| renewing the reservation, saving the result, or releasing the reservation | logged, counted; the action carries on and its result is returned | the same: the action has started, and raising would make the client retry it | + +What is raised is the repository's own `IdempotencyStorageError`, with `operation` and +`original_error`; a custom repository that lets a driver exception through gets it wrapped +in one. Out of `coordinate()` it always means the action did not run. A record the store +returned but cannot be read (corrupted, or an adapter that no longer decodes it) is not an +outage: in both modes the action runs and its result replaces the record. The option is +per coordinator, so a service with both kinds of operation builds two. Settings and the +Dishka provider carry it as `on_storage_error`. + ### Graceful Degradation (IdempotencyStorageError) If Redis is down, you might want to proceed with the operation anyway. @@ -421,6 +452,7 @@ coordinator = AsyncIdempotencyCoordinator( 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 outlives a dead owner; renewed while the action runs + on_storage_error="run", # "run" | "raise", see Failing closed when the store is down ) ``` @@ -438,6 +470,7 @@ from settings and you get the same numbers. | `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 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 | +| `DEFAULT_ON_STORAGE_ERROR` | `"run"` | what a call gets when the store fails before the action | `MAX_KEY_LENGTH`, `MAX_OPERATION_LENGTH` and the `InFlightMode` type live beside them. @@ -480,7 +513,7 @@ The library defines several exceptions to handle various idempotency scenarios: - **`IdempotencyRecordExpiredError`**: Raised by `service.validate_record()` if the record exists but its TTL has passed. - **`IdempotencyInvalidTTLError`**: Raised by `service.create_record()` if the requested TTL is outside the allowed range (configured in `IdempotencyDomainService`). - **`IdempotencyValidationError`**: Raised by `coordinator.coordinate()` and the decorator before the action runs, and by `service.create_record()`, if validation of `operation` or `idempotency_key` fails (e.g., empty string or too long). Map it to HTTP 400. -- **`IdempotencyStorageError`**: Raised for critical infrastructure failures like Redis connection errors. This allows you to distinguish between a cache miss and a storage backend failure. +- **`IdempotencyStorageError`**: Raised by the repository for critical infrastructure failures like Redis connection errors. This allows you to distinguish between a cache miss and a storage backend failure. `coordinator.coordinate()` and the decorator raise it only with `on_storage_error="raise"`, before the action runs. Map it to HTTP 503. - **`IdempotencyError`**: Base exception for all library errors. Also raised for data corruption during deserialization. ## Best Practices @@ -490,7 +523,7 @@ The library defines several exceptions to handle various idempotency scenarios: 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). +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), which is the default; for an effect that must not repeat, pass `on_storage_error="raise"` and answer 503. ## Production Examples @@ -766,7 +799,7 @@ def create_api_container(settings: Settings) -> AsyncContainer: Your own providers supply the Redis client and the settings object. The client is requested as `redis.asyncio.Redis | RedisCluster` — the key `redis_client_kit.AsyncRedisClient` names, so `redis_client_kit.providers.AsyncRedisProvider` stands in for `RedisProvider()` above with no adapter in between. A provider of your own must use that exact annotation: Dishka matches keys exactly, and `Redis` alone is a different one. The settings object must satisfy `IdempotencySettingsProtocol`, which `BaseIdempotencySettings` does. `IdempotencyProvider` also provides the metrics collector, under the key `IdempotencyMetricsProtocol` that the repository and the coordinator ask for: `PrometheusIdempotencyMetrics` when `metrics_enabled` is true (install the `prometheus` extra), a no-op collector otherwise. The Prometheus one comes from `get_idempotency_metrics()` — one instance per prefix on the default registry — so a container rebuilt per test, as a test suite does, gets the instance the first one registered instead of a `DuplicateTimeseries` from `prometheus_client`. To use another metrics backend, or a metric name prefix, provide `IdempotencyMetricsProtocol` yourself with `@provide(override=True)` in a provider listed after `IdempotencyProvider()`. -`enabled` on the settings object is the kill switch: with `enabled=False` the coordinator these providers build runs the action and nothing else — no read, no write, no metric. `in_flight` and `in_flight_lease_seconds` reach the coordinator the same way. A settings object that predates these fields is read as enabled, `"wait"` and 30 seconds. +`enabled` on the settings object is the kill switch: with `enabled=False` the coordinator these providers build runs the action and nothing else — no read, no write, no metric. `in_flight`, `in_flight_lease_seconds` and `on_storage_error` reach the coordinator the same way. A settings object that predates these fields is read as enabled, `"wait"`, 30 seconds and `"run"`. ## Migration Guide diff --git a/idempotency_kit/core/constants.py b/idempotency_kit/core/constants.py index c09a7c0..73ddbfa 100644 --- a/idempotency_kit/core/constants.py +++ b/idempotency_kit/core/constants.py @@ -39,3 +39,11 @@ # How often a waiting caller re-reads the key. IN_FLIGHT_POLL_INTERVAL_SECONDS: float = 0.05 + +# Storage failures +# What the coordinator does when the store fails before the action runs, on the +# reservation or the read: run the action unprotected, or raise IdempotencyStorageError. +StorageErrorMode = Literal["run", "raise"] + +# Run by default: availability over exactly-once. +DEFAULT_ON_STORAGE_ERROR: StorageErrorMode = "run" diff --git a/idempotency_kit/core/services/aio/coordinator.py b/idempotency_kit/core/services/aio/coordinator.py index 7bfe056..052c229 100644 --- a/idempotency_kit/core/services/aio/coordinator.py +++ b/idempotency_kit/core/services/aio/coordinator.py @@ -13,15 +13,19 @@ from idempotency_kit.core.constants import ( DEFAULT_IN_FLIGHT_LEASE_SECONDS, DEFAULT_IN_FLIGHT_MODE, + DEFAULT_ON_STORAGE_ERROR, IN_FLIGHT_POLL_INTERVAL_SECONDS, InFlightMode, + StorageErrorMode, ) from idempotency_kit.core.exceptions import ( + IdempotencyError, IdempotencyInProgressError, IdempotencyInvalidTTLError, IdempotencyKeyCollisionError, IdempotencyKeyReuseError, IdempotencyRecordExpiredError, + IdempotencyStorageError, IdempotencyValidationError, ) from idempotency_kit.core.models.entities import IdempotencyIdentifiers, IdempotencyRecord @@ -35,6 +39,7 @@ logger = logging.getLogger(__name__) _IN_FLIGHT_MODES: tuple[InFlightMode, ...] = ("wait", "raise", "run") +_STORAGE_ERROR_MODES: tuple[StorageErrorMode, ...] = ("run", "raise") @dataclass(frozen=True) @@ -76,6 +81,10 @@ class AsyncIdempotencyCoordinator: 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. + on_storage_error: What a call gets when the store fails before the action runs, on + the reservation or the read. ``"run"`` (the default) runs the action unprotected, + ``"raise"`` raises ``IdempotencyStorageError`` and does not run it. A failure once + the action has started, renewing, saving or releasing, is logged either way. """ def __init__( @@ -87,11 +96,16 @@ def __init__( enabled: bool = True, in_flight: InFlightMode = DEFAULT_IN_FLIGHT_MODE, in_flight_lease_seconds: int = DEFAULT_IN_FLIGHT_LEASE_SECONDS, + on_storage_error: StorageErrorMode = DEFAULT_ON_STORAGE_ERROR, ) -> None: if in_flight not in _IN_FLIGHT_MODES: raise IdempotencyValidationError(f"in_flight must be one of {_IN_FLIGHT_MODES}, got {in_flight!r}") if in_flight_lease_seconds < 1: raise IdempotencyValidationError(f"in_flight_lease_seconds must be >= 1, got {in_flight_lease_seconds}") + if on_storage_error not in _STORAGE_ERROR_MODES: + raise IdempotencyValidationError( + f"on_storage_error must be one of {_STORAGE_ERROR_MODES}, got {on_storage_error!r}" + ) if in_flight != "run" and not callable(getattr(repository, "replace", None)): # A repository written before reservations existed would take the pending # record through save() and never get it replaced, so every retry within the @@ -107,6 +121,7 @@ def __init__( self._enabled = enabled self._in_flight = in_flight self._in_flight_lease_seconds = in_flight_lease_seconds + self._on_storage_error = on_storage_error async def coordinate( self, @@ -135,8 +150,10 @@ async def coordinate( With ``enabled=False`` the action is simply run: nothing is validated, read or written, and no metric is recorded. - Storage and decode trouble never raise: the coordinator degrades to running the - action. What does raise is ``IdempotencyInProgressError``, when another call with + Storage and decode trouble do not raise by default: the coordinator degrades to + running the action. With ``on_storage_error="raise"`` a store that fails before the + action runs raises ``IdempotencyStorageError`` instead, and the action does not run. + What also raises is ``IdempotencyInProgressError``, when another call with the same key is still running its action and ``in_flight`` is ``"raise"``, or ``"wait"`` and a whole lease has passed; and ``IdempotencyKeyReuseError``, when the record under the key was made for a different request. @@ -296,12 +313,13 @@ async def _try_reserve( extra={"operation": operation, "idempotency_key": idempotency_key}, ) return None - except Exception: + except Exception as e: self._metrics.record_error(operation, "storage_reserve_error") logger.exception( "Idempotency coordinator error while reserving", extra={"operation": operation, "idempotency_key": idempotency_key}, ) + self._raise_if_failing_closed(operation, e) return None else: return pending @@ -356,6 +374,23 @@ async def _try_renew(self, reservation: IdempotencyRecord) -> None: extra={"operation": reservation.operation, "idempotency_key": reservation.idempotency_key}, ) + def _raise_if_failing_closed(self, operation: str, error: Exception) -> None: + """Raise ``IdempotencyStorageError`` for a failed store when ``on_storage_error`` is ``"raise"``. + + A store that answered with a record nobody can read (an ``IdempotencyError`` other + than a storage one) is not an outage: the action runs and its result replaces the + record. Anything outside the library's hierarchy is a driver error a custom + repository let through, and counts as the store failing. + """ + if self._on_storage_error != "raise": + return + if isinstance(error, IdempotencyStorageError): + raise error + if not isinstance(error, IdempotencyError): + raise IdempotencyStorageError( + f"Idempotency storage failure for {operation}", operation=operation, original_error=error + ) from error + 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: @@ -389,12 +424,13 @@ async def _try_get_cached( extra={"operation": operation, "idempotency_key": idempotency_key}, ) raise - except Exception: + except Exception as e: self._metrics.record_error(operation, "storage_get_error") logger.exception( "Idempotency coordinator error while fetching", extra={"operation": operation, "idempotency_key": idempotency_key}, ) + self._raise_if_failing_closed(operation, e) return _Lookup.UNUSABLE finally: self._metrics.record_latency(operation, "get", time.perf_counter() - start_time) diff --git a/idempotency_kit/dishka/aio/coordinator.py b/idempotency_kit/dishka/aio/coordinator.py index a7ca27d..8d8dd19 100644 --- a/idempotency_kit/dishka/aio/coordinator.py +++ b/idempotency_kit/dishka/aio/coordinator.py @@ -8,7 +8,11 @@ IdempotencyDomainService, IdempotencyMetricsProtocol, ) -from idempotency_kit.core.constants import DEFAULT_IN_FLIGHT_LEASE_SECONDS, DEFAULT_IN_FLIGHT_MODE +from idempotency_kit.core.constants import ( + DEFAULT_IN_FLIGHT_LEASE_SECONDS, + DEFAULT_IN_FLIGHT_MODE, + DEFAULT_ON_STORAGE_ERROR, +) from ..protocols import IdempotencySettingsProtocol @@ -37,4 +41,5 @@ def get_coordinator( enabled=getattr(settings, "enabled", True), in_flight=getattr(settings, "in_flight", DEFAULT_IN_FLIGHT_MODE), in_flight_lease_seconds=getattr(settings, "in_flight_lease_seconds", DEFAULT_IN_FLIGHT_LEASE_SECONDS), + on_storage_error=getattr(settings, "on_storage_error", DEFAULT_ON_STORAGE_ERROR), ) diff --git a/idempotency_kit/dishka/protocols.py b/idempotency_kit/dishka/protocols.py index 2c59429..ec992d7 100644 --- a/idempotency_kit/dishka/protocols.py +++ b/idempotency_kit/dishka/protocols.py @@ -2,7 +2,7 @@ from typing import Protocol, runtime_checkable -from idempotency_kit.core.constants import InFlightMode +from idempotency_kit.core.constants import InFlightMode, StorageErrorMode @runtime_checkable @@ -64,3 +64,11 @@ def in_flight_lease_seconds(self) -> int: A settings object without the attribute is read as the default lease. """ ... + + @property + def on_storage_error(self) -> StorageErrorMode: + """What a call gets when the store fails before the action runs. + + A settings object without the attribute is read as ``"run"``. + """ + ... diff --git a/idempotency_kit/settings.py b/idempotency_kit/settings.py index 2ded4f4..9dcdb31 100644 --- a/idempotency_kit/settings.py +++ b/idempotency_kit/settings.py @@ -5,10 +5,12 @@ from .core.constants import ( DEFAULT_IN_FLIGHT_LEASE_SECONDS, DEFAULT_IN_FLIGHT_MODE, + DEFAULT_ON_STORAGE_ERROR, DEFAULT_TTL_MINUTES, MAX_TTL_SECONDS, MIN_TTL_SECONDS, InFlightMode, + StorageErrorMode, ) @@ -41,3 +43,10 @@ class BaseIdempotencySettings(BaseModel): default=DEFAULT_IN_FLIGHT_LEASE_SECONDS, description="How long an in-flight reservation is held before it counts as abandoned (30 seconds)", ) + on_storage_error: StorageErrorMode = Field( + default=DEFAULT_ON_STORAGE_ERROR, + description=( + "What a call gets when the store fails before the action runs: run the action unprotected, " + "or raise IdempotencyStorageError" + ), + ) diff --git a/tests/unit/core/test_on_storage_error.py b/tests/unit/core/test_on_storage_error.py new file mode 100644 index 0000000..b74a3e0 --- /dev/null +++ b/tests/unit/core/test_on_storage_error.py @@ -0,0 +1,227 @@ +"""``on_storage_error``: what a call gets when the store fails before the action runs. + +Regression cover for issue #53: with the store unreachable, every call ran the action +unprotected, and there was no way to refuse instead. +""" + +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest +from redis.asyncio import Redis +from redis.asyncio.retry import Retry +from redis.backoff import NoBackoff + +from idempotency_kit import ( + AsyncIdempotencyCoordinator, + IdempotencyDomainService, + IdempotencyError, + IdempotencyKeyCollisionError, + IdempotencyMetricsProtocol, + IdempotencyRecord, + IdempotencyStorageError, + IdempotencyValidationError, + JsonResultAdapter, +) +from idempotency_kit.core.constants import InFlightMode +from idempotency_kit.infra.storage.redis.aio import RedisAsyncIdempotencyRepository + + +@pytest.mark.asyncio +async def test__coordinator__on_storage_error_raise__unreachable_redis__raises_and_never_runs_the_action() -> None: + """The reporter's scenario: two calls with the same key and Redis down charge nobody.""" + # Arrange + unreachable = Redis(host="127.0.0.1", port=1, socket_connect_timeout=0.2, retry=Retry(NoBackoff(), 0)) + coordinator = AsyncIdempotencyCoordinator( + RedisAsyncIdempotencyRepository(unreachable), + IdempotencyDomainService(), + in_flight="raise", + on_storage_error="raise", + ) + action = AsyncMock(return_value={"charge_id": "ch_1"}) + + # Act & Assert + try: + for _ in range(2): + with pytest.raises(IdempotencyStorageError) as raised: + await coordinator.coordinate("payment.charge", "same-key", 3600, JsonResultAdapter(), action) + assert raised.value.operation == "payment.charge" + assert raised.value.original_error is not None + finally: + await unreachable.aclose() + action.assert_not_called() + + +@pytest.mark.asyncio +async def test__coordinator__on_storage_error_default__unreachable_redis__runs_the_action() -> None: + """The default stays fail-open: availability over exactly-once.""" + # Arrange + unreachable = Redis(host="127.0.0.1", port=1, socket_connect_timeout=0.2, retry=Retry(NoBackoff(), 0)) + coordinator = AsyncIdempotencyCoordinator(RedisAsyncIdempotencyRepository(unreachable), IdempotencyDomainService()) + action = AsyncMock(return_value={"charge_id": "ch_1"}) + + # Act + try: + for _ in range(2): + await coordinator.coordinate("payment.charge", "same-key", 3600, JsonResultAdapter(), action) + finally: + await unreachable.aclose() + + # Assert + assert action.await_count == 2 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("in_flight", ["wait", "raise"]) +async def test__coordinator__on_storage_error_raise__read_after_a_contested_reservation_fails__raises( + mock_repo: AsyncMock, mock_adapter: MagicMock, in_flight: InFlightMode +) -> None: + """The key is taken and the store fails on the read that would say by what: nothing runs.""" + # Arrange + coordinator = AsyncIdempotencyCoordinator( + repository=mock_repo, domain_service=IdempotencyDomainService(), in_flight=in_flight, on_storage_error="raise" + ) + mock_repo.save.side_effect = IdempotencyKeyCollisionError("op", "key") + failure = IdempotencyStorageError("redis down", operation="op") + mock_repo.get.side_effect = failure + action = AsyncMock() + + # Act & Assert + with pytest.raises(IdempotencyStorageError) as raised: + await coordinator.coordinate("op", "key", 60, mock_adapter, action) + assert raised.value is failure + action.assert_not_called() + mock_repo.delete.assert_not_called() + + +@pytest.mark.asyncio +async def test__coordinator__on_storage_error_raise__in_flight_run__failed_read_raises( + mock_repo: AsyncMock, mock_adapter: MagicMock +) -> None: + """The unreserved flow's first touch of the store is the read, and that is where it refuses.""" + # Arrange + coordinator = AsyncIdempotencyCoordinator( + repository=mock_repo, domain_service=IdempotencyDomainService(), in_flight="run", on_storage_error="raise" + ) + mock_repo.get.side_effect = IdempotencyStorageError("redis down", operation="op") + action = AsyncMock() + + # Act & Assert + with pytest.raises(IdempotencyStorageError): + await coordinator.coordinate("op", "key", 60, mock_adapter, action) + action.assert_not_called() + + +@pytest.mark.asyncio +async def test__coordinator__on_storage_error_raise__failed_reservation__counts_and_raises( + mock_repo: AsyncMock, mock_adapter: MagicMock +) -> None: + """The refusal is still a storage error on the dashboards.""" + # Arrange + metrics = MagicMock(spec=IdempotencyMetricsProtocol) + coordinator = AsyncIdempotencyCoordinator( + repository=mock_repo, domain_service=IdempotencyDomainService(), metrics=metrics, on_storage_error="raise" + ) + mock_repo.save.side_effect = IdempotencyStorageError("redis down", operation="op") + action = AsyncMock() + + # Act & Assert + with pytest.raises(IdempotencyStorageError): + await coordinator.coordinate("op", "key", 60, mock_adapter, action) + action.assert_not_called() + metrics.record_error.assert_called_once_with("op", "storage_reserve_error") + + +@pytest.mark.asyncio +async def test__coordinator__on_storage_error_raise__driver_error_from_a_custom_repository__is_wrapped( + mock_repo: AsyncMock, mock_adapter: MagicMock +) -> None: + """A repository that lets the driver's exception through still fails closed, as IdempotencyStorageError.""" + # Arrange + coordinator = AsyncIdempotencyCoordinator( + repository=mock_repo, domain_service=IdempotencyDomainService(), on_storage_error="raise" + ) + failure = ConnectionError("connection refused") + mock_repo.save.side_effect = failure + action = AsyncMock() + + # Act & Assert + with pytest.raises(IdempotencyStorageError) as raised: + await coordinator.coordinate("op", "key", 60, mock_adapter, action) + assert raised.value.original_error is failure + assert raised.value.__cause__ is failure + action.assert_not_called() + + +@pytest.mark.asyncio +async def test__coordinator__on_storage_error_raise__unreadable_record__runs_the_action_and_replaces_it( + mock_repo: AsyncMock, mock_adapter: MagicMock +) -> None: + """A store that answered with a corrupted record is not an outage: the key heals instead of failing every retry.""" + # Arrange + coordinator = AsyncIdempotencyCoordinator( + repository=mock_repo, domain_service=IdempotencyDomainService(), on_storage_error="raise" + ) + mock_repo.save.side_effect = IdempotencyKeyCollisionError("op", "key") + mock_repo.get.side_effect = IdempotencyError("Corrupted idempotency record for op:key") + action = AsyncMock(return_value={"data": "fresh"}) + + # Act + result = await coordinator.coordinate("op", "key", 60, mock_adapter, action) + + # Assert + assert result == {"data": "fresh"} + action.assert_awaited_once() + mock_repo.replace.assert_awaited_once() + + +@pytest.mark.asyncio +async def test__coordinator__on_storage_error_raise__save_fails_after_the_action__returns_the_result( + mock_repo: AsyncMock, mock_adapter: MagicMock +) -> None: + """The effect has happened; raising now would make the client retry a completed operation.""" + # Arrange + coordinator = AsyncIdempotencyCoordinator( + repository=mock_repo, domain_service=IdempotencyDomainService(), on_storage_error="raise" + ) + mock_repo.replace.side_effect = IdempotencyStorageError("redis down", operation="op") + mock_repo.delete.side_effect = IdempotencyStorageError("redis down", operation="op") + action = AsyncMock(return_value={"data": "done"}) + + # Act + result = await coordinator.coordinate("op", "key", 60, mock_adapter, action) + + # Assert + assert result == {"data": "done"} + action.assert_awaited_once() + + +@pytest.mark.asyncio +async def test__coordinator__on_storage_error_raise__healthy_store__replays_as_usual( + redis_repository: RedisAsyncIdempotencyRepository, +) -> None: + """The mode changes nothing while the store answers.""" + # Arrange + coordinator = AsyncIdempotencyCoordinator( + repository=redis_repository, domain_service=IdempotencyDomainService(), on_storage_error="raise" + ) + action = AsyncMock(return_value={"charge_id": "ch_1"}) + + # Act + first = await coordinator.coordinate("payment.charge", "order-42", 3600, JsonResultAdapter(), action) + second = await coordinator.coordinate("payment.charge", "order-42", 3600, JsonResultAdapter(), action) + + # Assert + assert first == second == {"charge_id": "ch_1"} + action.assert_awaited_once() + stored = await redis_repository.get("payment.charge", "order-42") + assert isinstance(stored, IdempotencyRecord) + + +def test__coordinator__unknown_on_storage_error__raises_validation_error(mock_repo: AsyncMock) -> None: + # Arrange + kwargs: dict[str, Any] = {"on_storage_error": "fail"} + + # Act & Assert + with pytest.raises(IdempotencyValidationError, match="on_storage_error must be one of"): + AsyncIdempotencyCoordinator(repository=mock_repo, domain_service=IdempotencyDomainService(), **kwargs) diff --git a/tests/unit/dishka/test_providers.py b/tests/unit/dishka/test_providers.py index 65847c2..cbd8f72 100644 --- a/tests/unit/dishka/test_providers.py +++ b/tests/unit/dishka/test_providers.py @@ -15,7 +15,7 @@ JsonResultAdapter, NoOpIdempotencyMetrics, ) -from idempotency_kit.core.constants import InFlightMode +from idempotency_kit.core.constants import InFlightMode, StorageErrorMode from idempotency_kit.dishka import ( AsyncIdempotencyCoordinatorProvider, AsyncRedisIdempotencyProvider, @@ -39,12 +39,14 @@ def __init__( enabled: bool = True, in_flight: InFlightMode = "wait", in_flight_lease_seconds: int = 30, + on_storage_error: StorageErrorMode = "run", ) -> None: super().__init__() self._metrics_enabled = metrics_enabled self._enabled = enabled self._in_flight = in_flight self._in_flight_lease_seconds = in_flight_lease_seconds + self._on_storage_error = on_storage_error @provide def settings(self) -> IdempotencySettingsProtocol: @@ -55,6 +57,7 @@ def settings(self) -> IdempotencySettingsProtocol: enabled=self._enabled, in_flight=self._in_flight, in_flight_lease_seconds=self._in_flight_lease_seconds, + on_storage_error=self._on_storage_error, ) @provide @@ -284,6 +287,27 @@ async def test__shipped_providers__in_flight_settings__reach_the_coordinator() - await container.close() +@pytest.mark.asyncio +async def test__shipped_providers__on_storage_error_setting__reaches_the_coordinator() -> None: + """A service that wants to fail closed says so on the settings object.""" + # Arrange + container = make_async_container( + _AppProvider(metrics_enabled=False, on_storage_error="raise"), + IdempotencyProvider(), + AsyncRedisIdempotencyProvider(), + AsyncIdempotencyCoordinatorProvider(), + ) + + # Act + try: + coordinator = await container.get(AsyncIdempotencyCoordinator) + + # Assert + assert coordinator._on_storage_error == "raise" + finally: + await container.close() + + @pytest.mark.asyncio async def test__shipped_providers__redis_client_kit_client__container_builds_and_resolves_repository() -> None: """The client redis-client-kit provides is the one the Redis provider asks for: no adapter in between (#32)."""