From 25d9b2a41efc69bb3aad793291a2cdafb295f2da Mon Sep 17 00:00:00 2001 From: Alexey Shalaev <75322386+AlexeyShalaev@users.noreply.github.com> Date: Fri, 16 Oct 2026 18:39:39 +0400 Subject: [PATCH] feat: let the coordinator fail closed when the idempotency store is unavailable With the store unreachable the coordinator always ran the action unprotected, which is the right default for most operations and the wrong one for an effect nothing downstream deduplicates. A new coordinator option, on_storage_error="run" | "raise" (default "run"), also on BaseIdempotencySettings and the Dishka provider, lets a service refuse instead: with "raise", a storage failure on the reservation or the read is still counted and logged, then raises IdempotencyStorageError before the action runs. A driver exception a custom repository lets through is wrapped in one. A failure after the action, and a record the store returned but cannot be read, behave as before in both modes. Closes #53 --- docs/agents.md | 61 +++-- docs/api_reference.md | 3 +- docs/architecture.md | 5 +- docs/user_guide.md | 45 +++- idempotency_kit/core/constants.py | 8 + .../core/services/aio/coordinator.py | 44 +++- idempotency_kit/dishka/aio/coordinator.py | 7 +- idempotency_kit/dishka/protocols.py | 10 +- idempotency_kit/settings.py | 9 + tests/unit/core/test_on_storage_error.py | 227 ++++++++++++++++++ tests/unit/dishka/test_providers.py | 26 +- 11 files changed, 405 insertions(+), 40 deletions(-) create mode 100644 tests/unit/core/test_on_storage_error.py 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)."""