From 6c1a33435c7f7d00252b19f296168f9e5d056c21 Mon Sep 17 00:00:00 2001
From: Alexey Shalaev <75322386+AlexeyShalaev@users.noreply.github.com>
Date: Mon, 7 Sep 2026 01:47:54 +0300
Subject: [PATCH] feat!: reserve the key while the action runs

Two callers with the same key, the second arriving while the first one's
action is still running, both ran the action: the check was a GET and the
write a SET NX after the action, with nothing marking the key as taken in
between. The coordinator now reserves the key first, with a pending record
written under SET NX and the in-flight lease as its TTL, and writes the
result over it afterwards with the new repository method replace().

What a second caller gets is the coordinator's in_flight mode: "wait" (the
default) polls until the first caller's record lands, "raise" raises the new
IdempotencyInProgressError at once, and "run" is the previous flow for
actions that are safe to repeat. An action that raises or is cancelled
deletes its reservation, as does a result that cannot be stored, so the
retry runs again. A pending record past its lease counts as absent.

BREAKING CHANGE: a second concurrent caller with the same key now waits for the first caller's result instead of running the action too, and coordinate() and the decorator can raise IdempotencyInProgressError. Pass in_flight="run" to the coordinator (or set it on the settings object) to keep the previous behaviour. AsyncIdempotencyRepository gained replace(record); a custom repository needs it, and the coordinator raises TypeError at construction without it unless in_flight="run". IdempotencyRecord gained status, which records written before this change read as "completed".

Closes #26
---
 README.md                                     |  10 +-
 docs/agents.md                                | 185 +++++---
 docs/api_reference.md                         |  28 +-
 docs/architecture.md                          |  89 +++-
 docs/index.md                                 |   8 +-
 docs/quickstart.md                            |   6 +
 docs/user_guide.md                            | 103 ++++-
 idempotency_kit/__init__.py                   |   2 +
 idempotency_kit/core/constants.py             |  19 +
 idempotency_kit/core/exceptions.py            |   9 +
 idempotency_kit/core/models/entities.py       |  35 +-
 .../core/protocols/aio/repository.py          |  16 +
 .../core/services/aio/coordinator.py          | 319 ++++++++++++--
 idempotency_kit/core/services/domain.py       |  35 ++
 idempotency_kit/dishka/aio/coordinator.py     |   5 +-
 idempotency_kit/dishka/protocols.py           |  18 +
 .../infra/storage/redis/aio/repository.py     |  89 ++--
 idempotency_kit/settings.py                   |  20 +-
 tests/integration/test_redis_integration.py   |  41 +-
 tests/unit/core/conftest.py                   |  15 +-
 tests/unit/core/test_coordinator.py           |   3 +-
 tests/unit/core/test_entities.py              |  41 ++
 tests/unit/core/test_idempotent.py            |  18 +
 tests/unit/core/test_in_flight.py             | 396 ++++++++++++++++++
 tests/unit/core/test_services.py              |  55 ++-
 tests/unit/dishka/test_providers.py           |  36 +-
 .../unit/storage/redis/aio/test_repository.py |  70 +++-
 27 files changed, 1496 insertions(+), 175 deletions(-)
 create mode 100644 tests/unit/core/test_in_flight.py

diff --git a/README.md b/README.md
index 76b2d2c..e19e183 100644
--- a/README.md
+++ b/README.md
@@ -27,7 +27,7 @@ Ensure operations execute exactly once, even when called multiple times with the
 - **Type-Safe** — full type hints with Pydantic validation
 - **Async First** — built for asyncio applications
 - **Graceful Degradation** — high availability over strict exactly-once
-- **Collision Handling** — automatic resolution of concurrent requests
+- **In-flight Reservation** — a retry that arrives while the original is still running waits for its result or gets a 409; the action runs once
 - **Observability** — built-in metrics (hits, misses, collisions, latency)
 - **Bulk Operations** — efficient `get_many`, `save_many`, `delete_many`
 - **Redis Cluster Compatible** — non-transactional pipelines
@@ -110,12 +110,12 @@ return order
 
 ### 4. Concurrent requests handled
 
-If two requests arrive simultaneously:
+If two requests arrive while the first is still executing:
 
-- First request: cache miss → execute → save ✅
-- Second request: collision on save → fetch first result → return ✅
+- First request: reserves the key → execute → write the result over the reservation ✅
+- Second request: finds the reservation → waits for the first result → return ✅ (or a 409 with `in_flight="raise"`)
 
-Both requests get the **same result** - idempotency guaranteed!
+Both requests get the **same result**, and the business logic ran once.
 
 ## Use cases
 
diff --git a/docs/agents.md b/docs/agents.md
index d5ecfcf..1a68139 100644
--- a/docs/agents.md
+++ b/docs/agents.md
@@ -32,39 +32,49 @@ method that sounds plausible.
 ## Scope
 
 **It does** cache the result of an async operation under a caller-supplied key, replay that
-result on a repeat call, and resolve the race between two callers that arrive with the same
-key: the loser of the write reads the winner's record and returns it. It ships a Redis
-repository, a metrics protocol with a Prometheus implementation, a decorator that hides the
-whole flow, and Dishka providers that wire the pieces together.
+result on a repeat call, and hold the key while the first caller's action runs: a second
+caller that arrives with the same key in that window waits for the first one's result, or is
+refused, instead of running the action too. It ships a Redis repository, a metrics protocol
+with a Prometheus implementation, a decorator that hides the whole flow, and Dishka providers
+that wire the pieces together.
 
 **It does not** derive the key — the caller supplies it, and the request body is not part of
-it; it does not lock, reserve or otherwise prevent two concurrent callers from both running
-your business logic; it does not roll anything back; it does not cache failures; it does not
-retry; it has no sync API; and it stores nothing but JSON. It is a result cache with
-race-aware writes, not a distributed transaction.
+it; it does not roll anything back; it does not cache failures; it does not retry; it has no
+sync API; and it stores nothing but JSON. The reservation is a lease, not a lock: an action
+that outlives its lease can run twice, and when storage is down the action runs unreserved.
+It is a result cache with an in-flight reservation, not a distributed transaction.
 
 ## Mental model
 
 Four nouns and one flow.
 
 * **`IdempotencyRecord`** — a frozen Pydantic model: `operation`, `idempotency_key`, the
-  JSON `result`, `created_at`, `expires_at`. `expires_at` is what decides whether a record
-  is still a hit.
-* **`AsyncIdempotencyRepository`** — the storage protocol: `get` / `save` / `delete` and
-  their bulk twins. `save` is a write that fails if the key is already there; that failure,
-  `IdempotencyKeyCollisionError`, is the whole concurrency mechanism.
-  `RedisAsyncIdempotencyRepository` is the only implementation that ships.
+  JSON `result`, `created_at`, `expires_at`, and a `status` that is `"completed"` for a
+  stored result and `"pending"` for an in-flight reservation. `expires_at` is what decides
+  whether a record is still a hit; on a pending record it is the lease.
+* **`AsyncIdempotencyRepository`** — the storage protocol: `get` / `save` / `replace` /
+  `delete` and the bulk twins of `get`, `save` and `delete`. `save` is a write that fails if
+  the key is already there; that failure, `IdempotencyKeyCollisionError`, is how a
+  reservation is contested. `replace` writes whether or not the key is there; it is how a
+  reservation becomes a result. `RedisAsyncIdempotencyRepository` is the only
+  implementation that ships.
 * **`ResultAdapter`** — `encode` a return value into JSON, `decode` it back. Three ship:
   `PydanticResultAdapter(model_class)`, `JsonResultAdapter()`, `VoidResultAdapter()`.
-* **`AsyncIdempotencyCoordinator`** — the flow: read the record, decode it, return it if it
-  is there; otherwise run the action, encode the result, write it under `SET NX`, and on a
-  collision re-read and return the winner's result instead. A record whose `expires_at`
-  has passed is a miss even if the repository handed it back. It swallows every storage and
-  decode failure and degrades to running the action, which is the deliberate trade of
-  exactly-once for availability.
+* **`AsyncIdempotencyCoordinator`** — the flow: reserve the key by writing a pending
+  record under `SET NX` with the lease as its TTL. If that succeeds, run the action, encode
+  the result and `replace` the reservation with the completed record. If it fails, read
+  what holds the key: a completed record is decoded and returned; a pending one means
+  another caller is in flight, and `in_flight` decides — `"wait"` polls until the record
+  arrives, `"raise"` raises `IdempotencyInProgressError`, `"run"` is the old flow of read,
+  run, `SET NX` and adopt the winner's result on a collision. A record whose `expires_at`
+  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.
 
 `IdempotencyDomainService` sits between the coordinator and the record: it applies the TTL
-bounds and turns Pydantic validation errors into `IdempotencyValidationError`.
+bounds, builds the pending record for a reservation, and turns Pydantic validation errors
+into `IdempotencyValidationError`.
 `@async_idempotent` is 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.
 
@@ -133,7 +143,7 @@ result = await coordinator.coordinate(
 | Name | Signature | What it is |
 |---|---|---|
 | `async_idempotent` | `(operation, adapter, ttl_seconds=None, key_param="idempotency_key", infra_param=None)` | decorator for an async function or method |
-| `AsyncIdempotencyCoordinator` | `(repository, domain_service, operation_ttls=None, metrics=None, enabled=True)` | the flow; `operation_ttls` is `dict[str, int]` in seconds; `enabled=False` runs the action and nothing else |
+| `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"` |
 | `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 |
@@ -144,7 +154,7 @@ result = await coordinator.coordinate(
 | `VoidResultAdapter` | `()` | stores JSON `null`, decodes back to `None` |
 | `IdempotencyMetricsProtocol` | runtime-checkable `Protocol` | metrics contract |
 | `NoOpIdempotencyMetrics` | `()` | the default collector |
-| `IdempotencyError` and its five subclasses | | see [Errors](#errors) |
+| `IdempotencyError` and its six subclasses | | see [Errors](#errors) |
 
 ### Not exported from the root
 
@@ -154,14 +164,15 @@ result = await coordinator.coordinate(
 | `PrometheusIdempotencyMetrics` | `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` | `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` | `idempotency_kit.core.constants` |
 
 ### Coordinator and domain service
 
 | Method | Returns | Notes |
 |---|---|---|
-| `AsyncIdempotencyCoordinator.coordinate(operation, idempotency_key, ttl_seconds, adapter, action, /, *args, **kwargs)` | `T` | never raises for storage or decode trouble |
+| `AsyncIdempotencyCoordinator.coordinate(operation, idempotency_key, ttl_seconds, adapter, action, /, *args, **kwargs)` | `T` | never raises for storage or decode trouble; raises `IdempotencyInProgressError` when the key is in flight and `in_flight` says so |
 | `IdempotencyDomainService.create_record(operation, idempotency_key, result, *, ttl_minutes=None)` | `IdempotencyRecord` | raises `IdempotencyInvalidTTLError`, `IdempotencyValidationError` |
+| `IdempotencyDomainService.create_pending_record(operation, idempotency_key, *, lease_seconds)` | `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 |
 
 ### Record
@@ -171,8 +182,11 @@ result = await coordinator.coordinate(
 | `operation` | `str` | 1-100 chars, stripped, no `:` |
 | `idempotency_key` | `str` | 1-255 chars, stripped, no `:` |
 | `result` | `JsonValue` | whatever the adapter encoded; `null` for a void result |
-| `created_at` / `expires_at` | `datetime` | UTC, set by `create` |
+| `created_at` / `expires_at` | `datetime` | UTC, set by `create` and `pending`; on a pending record `expires_at` is the lease |
+| `status` | `"pending" \| "completed"` | `"completed"` unless said otherwise, which is how a record written before the field existed reads |
 | `IdempotencyRecord.create(operation, idempotency_key, result, ttl_seconds)` | `IdempotencyRecord` | classmethod; `ttl_seconds` is a `float` here |
+| `IdempotencyRecord.pending(operation, idempotency_key, lease_seconds)` | `IdempotencyRecord` | classmethod; the reservation, `result` is `null` |
+| `.is_pending` | `bool` | `status == "pending"` |
 | `.is_expired` | `bool` | `now >= expires_at` |
 | `.ttl_seconds` | `float` | remaining, `0.0` once expired |
 
@@ -182,15 +196,18 @@ result = await coordinator.coordinate(
 |---|---|---|
 | `get(operation, idempotency_key)` | `IdempotencyRecord | None` | `IdempotencyValidationError`, `IdempotencyStorageError`, `IdempotencyError` |
 | `save(record)` | `None` | `IdempotencyKeyCollisionError`, `IdempotencyValidationError`, `IdempotencyStorageError`, `IdempotencyError` |
+| `replace(record)` | `None` | as `save` minus the collision: it writes over whatever is there |
 | `delete(operation, idempotency_key)` | `bool` | `IdempotencyValidationError`, `IdempotencyStorageError` |
 | `get_many(operation, idempotency_keys)` | `dict[str, IdempotencyRecord]` | as `get`; only found keys appear |
 | `save_many(records, *, rollback_on_error=False)` | `None` | as `save`; the collision carries `list[str]` |
 | `delete_many(operation, idempotency_keys)` | `int` | as `delete` |
 
 `RedisAsyncIdempotencyRepository(redis, *, key_prefix="idempotency:", metrics=None)` is the
-implementation: `SET key value EX ceil(record.ttl_seconds) NX` for `save`, `GET` for `get`,
-`MGET` for `get_many`, and a non-transactional pipeline for `save_many` so it works on Redis
-Cluster. It deletes any record it reads back expired and reports that as a miss.
+implementation: `SET key value EX ceil(record.ttl_seconds) NX` for `save`, the same without
+`NX` for `replace`, `GET` for `get`, `MGET` for `get_many`, and a non-transactional pipeline
+for `save_many` so it works on Redis Cluster. It deletes any record it reads back expired and
+reports that as a miss. A repository of your own needs `replace` too: the coordinator raises
+`TypeError` at construction without it, unless `in_flight="run"`.
 
 ### Metrics
 
@@ -202,17 +219,25 @@ and `record_bulk_miss(operation, count)`. `PrometheusIdempotencyMetrics(prefix=N
 `idempotency_operation_duration_seconds{operation,method}`.
 
 Each metric has one owner, so the same collector can go to both layers: the coordinator
-records hit, miss, collision and the latency of `get` and `save`; the repository records
-errors, the bulk hit and miss counts of `get_many`, and the latency of `delete` and
+records hit, miss, collision and the latency of `get`, `reserve` and `save`; the repository
+records errors, the bulk hit and miss counts of `get_many`, and the latency of `delete` and
 `get_many`.
 
+A collision is two callers on one key at the same time. With a reservation it is what the
+second caller records on finding the pending record: a waiter then records a hit when the
+result arrives, a refused caller records nothing more and raises. In `"run"` mode it is the
+loser's `SET NX` failing after both ran, as before. The error types the coordinator reports
+are `storage_get_error`, `storage_reserve_error`, `storage_save_error`,
+`storage_release_error` and `record_validation_error`.
+
 ### Settings and Dishka
 
 `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` and `operation_ttls={}` — the three TTL
-fields default to the `core.constants` values, which is also what `IdempotencyDomainService`
-uses. It satisfies `IdempotencySettingsProtocol`, which is what the providers ask for.
+`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
+`core.constants` values, which is also what `IdempotencyDomainService` and the coordinator
+use. It satisfies `IdempotencySettingsProtocol`, which is what the providers ask for.
 
 ```python
 from dishka import make_async_container
@@ -235,8 +260,9 @@ container = make_async_container(
 All three are `Scope.APP`. `IdempotencyProvider` gives one metrics collector to the whole
 process — `PrometheusIdempotencyMetrics` when `settings.metrics_enabled`, a no-op otherwise.
 Another backend goes in with `@provide(override=True)` in a provider listed after it.
-`settings.enabled` reaches the coordinator: `False` makes it a pass-through. A settings
-object written before that field existed is read as enabled.
+`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.
 
 ## Rules that hold or break the code
 
@@ -244,11 +270,18 @@ object written before that field existed is read as enabled.
    body. Two calls with the same `operation` and `idempotency_key` and different payloads
    replay the first result. A key must be unique per intended effect, and one client request
    must not reuse a key across two different operations' worth of work.
-2. **Two concurrent callers both run your business logic.** There is no lock and no
-   in-flight reservation: the check is a `GET`, the write is `SET NX`. Both miss, both
-   execute, the loser's `save` collides, and it re-reads and returns the winner's record.
-   The callers see one result; the side effects happened twice. Make the action safe to run
-   twice — a database upsert, an outbox row keyed by the same key — or take your own lock.
+2. **A second caller with the same key does not run your business logic while the first is
+   in flight — unless you ask for that.** The coordinator reserves the key with a pending
+   record (`SET NX`, TTL `in_flight_lease_seconds`, 30 s by default) before the action and
+   writes the result over it after. A second caller that finds the reservation waits for the
+   result with `in_flight="wait"`, the default — polling every 50 ms, giving up with
+   `IdempotencyInProgressError` after a whole lease — or is refused at once with
+   `in_flight="raise"`, which an HTTP layer maps to 409. `in_flight="run"` is the flow from
+   before reservations existed: both run, the loser's `SET NX` collides and it adopts the
+   winner's result; the callers see one result and the side effect happened twice, so keep
+   it for actions that are genuinely safe to repeat. The reservation is a lease, not a lock:
+   a pending record past its lease counts as absent and the next caller runs the action, so
+   `in_flight_lease_seconds` has to be longer than the action can ever take.
 3. **Declare the key keyword-only anyway.** The decorator reads `key_param` from the
    keyword arguments, and from the positional arguments when the parameter can be passed
    that way. Keyword-only is still the shape to write: nothing can land in it by position,
@@ -263,16 +296,20 @@ object written before that field existed is read as enabled.
    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 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.
+   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.
 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
    expire.
-8. **Failures are not cached.** An exception from the action propagates and nothing is
-   written, so the next call with the same key runs the action again.
+8. **Failures are not cached.** An exception from the action propagates, nothing is
+   written, and the reservation is deleted — cancellation included — so the next call with
+   the same key runs the action again. If the result cannot be stored (rules 11, 13, 14) the
+   reservation is deleted too, rather than holding retries for a record that never comes.
 9. **TTLs are given in seconds and truncated to whole minutes, with a floor of one.**
    `max(1, ttl_seconds // 60)`. `ttl_seconds=3600` is an hour; `ttl_seconds=90` is one
    minute; `ttl_seconds=30` is one minute. There is no sub-minute record.
@@ -306,13 +343,24 @@ object written before that field existed is read as enabled.
     collectors in the constructor; a second instance with the same prefix raises from
     `prometheus_client`.
 18. **Each metric has one owner.** The coordinator records hit, miss, collision and the
-    latency of `get` and `save`; the repository records errors, the bulk hit and miss counts
-    of `get_many`, and the latency of `delete` and `get_many`. One collector shared between
-    them — which is what the Dishka providers wire — counts every operation once.
+    latency of `get`, `reserve` and `save`; the repository records errors, the bulk hit and
+    miss counts of `get_many`, and the latency of `delete` and `get_many`. One collector
+    shared between them — which is what the Dishka providers wire — counts every operation
+    once: a waited call is one collision and one hit, a reserved run is one miss.
 19. **Async only.** There is no sync mirror, and no `__init__.py` name that gives you one.
 20. **`enabled=False` switches the whole thing off.** The coordinator runs the action and
     nothing else: no read, no write, no metric. The Dishka providers pass `settings.enabled`
     through to it.
+21. **A repository of your own needs `replace`.** The reservation goes in through `save`
+    and comes out through `replace`; without it every retry within the lease would wait for,
+    or be refused over, a result that is never written. The coordinator raises `TypeError`
+    at construction when the repository lacks it and `in_flight` is not `"run"`.
+22. **Finish a rolling upgrade before relying on the reservation.** An instance on a version
+    without `status` reads a pending record as a completed one with a `null` result:
+    `PydanticResultAdapter` cannot decode that and runs the action, which is the old
+    behaviour, but `JsonResultAdapter` and `VoidResultAdapter` replay the `None`. Roll out
+    with `in_flight="run"` and switch once every instance is on the new version, or accept
+    the window.
 
 ## Common mistakes
 
@@ -346,14 +394,34 @@ async def find(self, *, idempotency_key: str | None = None) -> dict | None: ...
 ```
 
 ```python
-# WRONG — treating the decorator as a mutex: both concurrent callers reach this body
+# WRONG — in_flight="run" on an action that must not happen twice: both concurrent callers
+# reach this body, and only the responses are deduplicated
+coordinator = AsyncIdempotencyCoordinator(repository, service, in_flight="run")
+
 @async_idempotent(operation="payment.charge", adapter=PydanticResultAdapter(ChargeDTO))
 async def charge(self, dto, *, idempotency_key: str | None = None) -> ChargeDTO:
     return await self._psp.charge(dto)            # charged twice
 
-# RIGHT — the action is safe to run twice, or the provider deduplicates on the same key
+# RIGHT — leave the default, and the second caller gets the first one's charge; or refuse it
+coordinator = AsyncIdempotencyCoordinator(repository, service)                    # waits
+coordinator = AsyncIdempotencyCoordinator(repository, service, in_flight="raise") # 409
+
+try:
+    return await use_case.charge(dto, idempotency_key=key)
+except IdempotencyInProgressError:
+    raise HTTPException(409, detail="a request with this Idempotency-Key is still being processed")
+```
+
+```python
+# WRONG — a lease the action can outlive: the reservation expires mid-run, a retry at
+# second six finds the key free, and the provider is charged twice after all
+coordinator = AsyncIdempotencyCoordinator(repository, service, in_flight_lease_seconds=5)
+
 async def charge(self, dto, *, idempotency_key: str | None = None) -> ChargeDTO:
-    return await self._psp.charge(dto, idempotency_key=idempotency_key)
+    return await self._psp.charge(dto, timeout=20)
+
+# RIGHT — the lease outlives the action's worst case, timeouts and retries included
+coordinator = AsyncIdempotencyCoordinator(repository, service, in_flight_lease_seconds=60)
 ```
 
 ```python
@@ -394,14 +462,15 @@ All derive from `IdempotencyError`, which is exported alongside them.
 | `IdempotencyError` | `(message)` | the base, and what a corrupted stored payload raises |
 | `IdempotencyKeyCollisionError` | `(operation, key)` | `save` found the key already there; `key` is a `str`, or a `list[str]` from `save_many`. Carries `.operation` and `.key` |
 | `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` |
 | `IdempotencyStorageError` | `(message, operation=None, original_error=None)` | the backend failed. Carries `.operation` and `.original_error` |
 | `IdempotencyValidationError` | `(message, errors=None)` | an identifier or a result failed validation. `.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 none of these reach the caller: the collision is
-resolved into the winner's result, and every other one is logged, counted and swallowed.
-They are the contract of the repository and the domain service, which is where you meet them
-if you drive those directly.
+Through `coordinate()` and the decorator only `IdempotencyInProgressError` reaches the
+caller: 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
 
@@ -411,7 +480,7 @@ Fetch a page when the task is the one named beside it.
 |---|---|
 | [Home](index.md) | placing the library — what it is for, the four shapes of caller |
 | [Quick Start](quickstart.md) | the first integration, and what makes a good key |
-| [User Guide](user_guide.md) | bulk operations, graceful degradation, Dishka wiring, worked services |
+| [User Guide](user_guide.md) | in-flight handling and the lease, bulk operations, graceful degradation, Dishka wiring, worked services |
 | [Architecture](architecture.md) | the layers, the request-flow diagrams, the cluster reasoning |
 | [API Reference](api_reference.md) | an exact field, default or constructor argument |
 | [Testing](testing_conventions.md) | writing tests against this library, or contributing to it |
diff --git a/docs/api_reference.md b/docs/api_reference.md
index 12129a8..1351127 100644
--- a/docs/api_reference.md
+++ b/docs/api_reference.md
@@ -19,9 +19,12 @@ A frozen Pydantic model representing an idempotency result. Inherits from `Idemp
   - `idempotency_key` (str): Unique key for this instance.
   - `result` (JsonValue): The cached result — any JSON value (mapping, list, scalar); `null` for a void result.
   - `created_at` (datetime): When the record was created.
-  - `expires_at` (datetime): When the record will expire.
+  - `expires_at` (datetime): When the record will expire. On a pending record this is the in-flight lease.
+  - `status` (`"pending" | "completed"`, default `"completed"`): `"pending"` while the action runs under an in-flight reservation, `"completed"` once the result is stored. A record written before the field existed reads as completed.
 - **Methods**:
   - `create(operation: str, idempotency_key: str, result: JsonValue, ttl_seconds: float) -> IdempotencyRecord`: Class method to create a new record.
+  - `pending(operation: str, idempotency_key: str, lease_seconds: float) -> IdempotencyRecord`: Class method to create the in-flight reservation; `result` is `null`.
+  - `is_pending`: Property returning `True` for an in-flight reservation.
   - `is_expired`: Property returning `True` if current time is after `expires_at`.
   - `ttl_seconds`: Property returning remaining TTL in seconds.
 
@@ -36,14 +39,31 @@ Service for creating and validating records.
   - *Note*: Constructor validates that `default_ttl_minutes` (converted to seconds) is within the `[min_ttl_seconds, max_ttl_seconds]` range.
 - **Methods**:
   - `create_record(operation, idempotency_key, result, *, ttl_minutes=None)`: Creates a new `IdempotencyRecord` with validation and TTL management.
+  - `create_pending_record(operation, idempotency_key, *, lease_seconds)`: Creates the in-flight reservation. The lease is not held to the TTL bounds; it must be at least a second.
   - `validate_record(record)`: Validates that a record is still usable, raising `IdempotencyRecordExpiredError` if not. The coordinator calls it on every record it reads.
 
+### AsyncIdempotencyCoordinator
+The flow: reserve the key, run the action, store the result; replay a stored result; hand a second concurrent caller the first one's result.
+
+- **Constructor**:
+  - `repository` (AsyncIdempotencyRepository): Storage for records.
+  - `domain_service` (IdempotencyDomainService): Record factory and TTL bounds.
+  - `operation_ttls` (dict[str, int], optional): Per-operation TTLs in seconds; win over the decorator's `ttl_seconds`.
+  - `metrics` (IdempotencyMetricsProtocol, optional): Metrics collector.
+  - `enabled` (bool, default: True): `False` runs the action and nothing else.
+  - `in_flight` (`"wait" | "raise" | "run"`, default: `"wait"`): What a second caller gets while the first one's action is still running under the same key — the first caller's result once it lands, `IdempotencyInProgressError` at once, or a run of its own with the winner's result adopted on collision.
+  - `in_flight_lease_seconds` (int, default: 30): How long a reservation is held before it counts as abandoned, and the longest a waiting caller waits. Must be at least 1 and longer than the action can take.
+  - *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, **kwargs)`: Runs the flow and returns the action's result type. Never raises for storage or decode trouble; raises `IdempotencyInProgressError` when the key is in flight and `in_flight` says so.
+
 ### AsyncIdempotencyRepository (Protocol)
 Interface for idempotency storage.
 
 - **Methods**:
   - `get(operation, idempotency_key)`: Returns `IdempotencyRecord` or `None`.
   - `save(record)`: Saves a record with NX (not exists) guarantee.
+  - `replace(record)`: Writes a record whether or not the key is there; how a reservation becomes a result.
   - `delete(operation, idempotency_key)`: Deletes a record. Returns `True` if deleted.
   - `get_many(operation, idempotency_keys)`: Returns `dict[str, IdempotencyRecord]`.
   - `save_many(records, *, rollback_on_error=False)`: Saves multiple records. NOT atomic unless `rollback_on_error=True`.
@@ -55,12 +75,12 @@ Interface for metrics collection.
 - **Methods**:
   - `record_hit(operation)`
   - `record_miss(operation)`
-  - `record_collision(operation)`
+  - `record_collision(operation)`: Two callers on one key at the same time.
   - `record_error(operation, error_type)`
   - `record_latency(operation, method, duration_seconds)`
   - `record_bulk_hit(operation, count)`: For bulk operations.
   - `record_bulk_miss(operation, count)`: For bulk operations.
-- **Who records what**: the coordinator records hit, miss, collision and the latency of `get` and `save`; the repository
+- **Who records what**: the coordinator records hit, miss, collision and the latency of `get`, `reserve` and `save`; the repository
   records errors, the bulk hit and miss counts of `get_many`, and the latency of `delete` and `get_many`. One collector
   handed to both — which is what the Dishka providers do — therefore counts each operation once.
 
@@ -73,6 +93,7 @@ Redis implementation of the repository protocol.
   - `redis` (`redis.asyncio.Redis`): Any async Redis client, including a subclass such as an instrumented or fake one.
   - `key_prefix` (str, default: "idempotency:"): Prefix for all Redis keys. (**keyword-only**)
   - `metrics` (IdempotencyMetricsProtocol, optional): Metrics collector. (**keyword-only**)
+- **Storage**: `SET key value EX ttl NX` for `save`, the same without `NX` for `replace`, `GET` for `get`.
 
 ## Exceptions
 
@@ -80,6 +101,7 @@ Redis implementation of the repository protocol.
 - **`IdempotencyKeyCollisionError(operation, key)`**: Raised when a key already exists.
   - `key` can be a single `str` or a `list[str]` for bulk operations.
 - **`IdempotencyRecordExpiredError(operation, key)`**: Raised when record exists but is expired.
+- **`IdempotencyInProgressError(operation, key)`**: Raised by `coordinate()` and the decorator when another call with the same key is still running its action.
 - **`IdempotencyStorageError(message, operation, original_error)`**: Raised on storage failure.
 - **`IdempotencyValidationError(message, errors=None)`**: Raised for invalid input (e.g. empty key, too long string).
   - `errors` (list, optional): Detailed Pydantic validation errors.
diff --git a/docs/architecture.md b/docs/architecture.md
index 80b7876..0003c87 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -43,34 +43,77 @@ sequenceDiagram
     UseCase-->>Client: Cached Result
 ```
 
-### Cache Miss and Save
+### Cache Miss: Reserve, Execute, Complete
+
+The coordinator reserves the key before the action runs. The reservation is the record in
+its pending state, written with `SET NX` and the in-flight lease as its TTL; the completed
+record is written over it afterwards with a plain `SET`.
 
 ```mermaid
 sequenceDiagram
     participant Client
-    participant UseCase
+    participant Coordinator
     participant Repository
     participant DomainService
     participant Storage
 
-    Client->>UseCase: execute(idempotency_key)
-    UseCase->>Repository: get(operation, key)
-    Repository->>Storage: GET key
-    Storage-->>Repository: None
-    Repository-->>UseCase: None
+    Client->>Coordinator: coordinate(operation, key, ...)
+    Coordinator->>DomainService: create_pending_record(operation, key, lease)
+    DomainService-->>Coordinator: IdempotencyRecord(status="pending")
+    Coordinator->>Repository: save(pending)
+    Repository->>Storage: SET key pending NX EX lease
+    Storage-->>Repository: OK
 
-    UseCase->>UseCase: execute business logic
-    UseCase->>DomainService: create_record(operation, key, result)
-    DomainService-->>UseCase: IdempotencyRecord
+    Coordinator->>Coordinator: execute business logic
+    Coordinator->>DomainService: create_record(operation, key, result)
+    DomainService-->>Coordinator: IdempotencyRecord(status="completed")
 
-    UseCase->>Repository: save(record)
-    Repository->>Storage: SET key NX EX
+    Coordinator->>Repository: replace(record)
+    Repository->>Storage: SET key record EX ttl
     Storage-->>Repository: OK
 
-    UseCase-->>Client: New Result
+    Coordinator-->>Client: New Result
 ```
 
-### Concurrent Collision Scenario
+When the reservation fails, the coordinator reads the key: a completed record is a hit and is
+returned. An action that raises deletes its reservation, so the retry runs it again.
+
+### Concurrent Callers
+
+With `in_flight="wait"` (the default) the second caller finds the reservation and polls the
+key until the first caller's record replaces it. With `in_flight="raise"` it raises
+`IdempotencyInProgressError` as soon as it sees the reservation, which an HTTP layer maps to
+409. Either way the action ran once.
+
+```mermaid
+sequenceDiagram
+    participant Caller1
+    participant Caller2
+    participant Storage
+
+    Caller1->>Storage: SET key pending NX EX lease -> OK
+    Caller1->>Caller1: execute logic
+
+    Caller2->>Storage: SET key pending NX EX lease -> Fails (Already exists)
+    Caller2->>Storage: GET key -> pending
+    Caller2->>Caller2: wait 50 ms (or raise IdempotencyInProgressError)
+    Caller2->>Storage: GET key -> pending
+
+    Caller1->>Storage: SET key record EX ttl -> OK
+    Caller2->>Storage: GET key -> record
+    Caller2-->>Caller2: Use result from Caller 1
+```
+
+The wait is bounded by the lease: a pending record past its lease counts as absent, so a
+worker that crashed mid-action cannot hold the key for longer, and a waiter that has spent a
+whole lease raises `IdempotencyInProgressError` rather than waiting forever. The other side of
+that bound is that the lease has to outlive the action; a reservation that expires mid-run
+lets the next caller run the action again.
+
+### Concurrent Callers with `in_flight="run"`
+
+The flow from before reservations existed, for actions that are safe to repeat: both callers
+run the action, the loser's `SET NX` collides, and it adopts the winner's result.
 
 ```mermaid
 sequenceDiagram
@@ -112,16 +155,16 @@ The library provides `IdempotencyMetricsProtocol` for observability:
 
 - `record_hit` - cache hit
 - `record_miss` - cache miss or not found
-- `record_collision` - duplicate key on save
+- `record_collision` - two callers on one key at the same time: the second one found a reservation, or (`in_flight="run"`) its save collided
 - `record_error` - storage/serialization error
 - `record_latency` - operation duration
 - `record_bulk_hit` - multiple hits in bulk operation
 - `record_bulk_miss` - multiple misses in bulk operation
 
 Each metric has one owner, so a single collector can be handed to both layers without
-double counting: the coordinator records hit, miss, collision and the latency of `get` and
-`save`; the repository records errors, the bulk hit and miss counts of `get_many`, and the
-latency of `delete` and `get_many`.
+double counting: the coordinator records hit, miss, collision and the latency of `get`,
+`reserve` and `save`; the repository records errors, the bulk hit and miss counts of
+`get_many`, and the latency of `delete` and `get_many`.
 
 ```python
 metrics = PrometheusMetrics()
@@ -131,11 +174,13 @@ coordinator = AsyncIdempotencyCoordinator(repo, IdempotencyDomainService(), metr
 
 The library follows the **Idempotency Key Pattern**:
 1. Client provides a unique key for an operation.
-2. Server checks if a result for this key is already cached.
-3. If found, returns the cached result immediately.
-4. If not found, executes the operation, caches the result, and returns it.
+2. Server reserves the key; if the key is already taken, it returns the stored result, or waits for the caller that holds it.
+3. If the reservation succeeds, it executes the operation, writes the result over the reservation, and returns it.
 
-This ensures **at-most-once** or **exactly-once** semantics depending on how the application handles storage failures (see Graceful Degradation in User Guide).
+What this gives is one execution per key while storage is up and the action finishes within
+its lease, and at-least-once when storage is down, because the coordinator then runs the
+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.
 
 ## Redis Key Format
 
diff --git a/docs/index.md b/docs/index.md
index 3f4d6b0..142c207 100644
--- a/docs/index.md
+++ b/docs/index.md
@@ -157,12 +157,12 @@ return order
 
 ### 4. Concurrent Requests Handled
 
-If two requests arrive simultaneously:
+If two requests arrive while the first is still executing:
 
-- First request: cache miss → execute → save ✅
-- Second request: collision on save → fetch first result → return ✅
+- First request: reserves the key → execute → write the result over the reservation ✅
+- Second request: finds the reservation → waits for the first result → return ✅ (or a 409 with `in_flight="raise"`)
 
-Both requests get the **same result** - idempotency guaranteed!
+Both requests get the **same result**, and the business logic ran once.
 
 ## Why idempotency-kit?
 
diff --git a/docs/quickstart.md b/docs/quickstart.md
index bb8baec..cf49e6d 100644
--- a/docs/quickstart.md
+++ b/docs/quickstart.md
@@ -121,6 +121,12 @@ Request B --key--> Check (miss) --> Execute --> Save ❌ (collision)
                                              Get A's result
 ```
 
+That is the hand-rolled flow above, where both requests execute and only the responses are
+deduplicated. `AsyncIdempotencyCoordinator` and `@async_idempotent` reserve the key before
+executing, so request B waits for A's result — or is refused with
+`IdempotencyInProgressError` — instead of executing too. See
+[In-flight requests](user_guide.md#in-flight-requests) in the User Guide.
+
 ## Key Concepts
 
 ### Idempotency Key
diff --git a/docs/user_guide.md b/docs/user_guide.md
index dc970d1..66ea580 100644
--- a/docs/user_guide.md
+++ b/docs/user_guide.md
@@ -85,6 +85,80 @@ class CreateOrderUseCase:
         return result
 ```
 
+This hand-rolled flow reads, executes and then writes with `SET NX`, so two requests with the
+same key that overlap both execute; only their responses are deduplicated.
+`AsyncIdempotencyCoordinator` reserves the key before executing, which is why the decorator
+and the coordinator are the recommended shape — see [In-flight requests](#in-flight-requests).
+
+## In-flight requests
+
+The case an idempotency key exists for is a client that timed out and retried while the
+original request is still being processed. The coordinator handles it by reserving the key
+before the action runs: a pending record goes in under `SET NX` with a lease as its TTL, the
+action runs, and the completed record is written over the reservation. What a second caller
+gets while the key is reserved is the coordinator's `in_flight` mode:
+
+| `in_flight` | The second caller... | Use it when |
+|---|---|---|
+| `"wait"` (default) | polls the key every 50 ms and returns the first caller's result when it lands | the client wants an answer, not an error |
+| `"raise"` | raises `IdempotencyInProgressError` at once | the client can retry later; map it to HTTP 409 |
+| `"run"` | runs the action too and adopts the first caller's result when its own write collides | the action is genuinely safe to repeat and you want the pre-reservation flow |
+
+```python
+coordinator = AsyncIdempotencyCoordinator(
+    repo,
+    IdempotencyDomainService(),
+    in_flight="raise",
+    in_flight_lease_seconds=60,
+)
+```
+
+```python
+@app.post("/charges")
+async def charge(dto: ChargeDTO, idempotency_key: str | None = Header(None, alias="Idempotency-Key")):
+    try:
+        return await use_case.execute(dto, idempotency_key=idempotency_key)
+    except IdempotencyInProgressError:
+        raise HTTPException(409, detail="a request with this Idempotency-Key is still being processed")
+```
+
+**The lease.** `in_flight_lease_seconds` (default 30) is how long the reservation is held.
+A pending record past its lease counts as absent, so a worker that crashed mid-action cannot
+wedge the key; it is also how long a waiting caller waits before it gives up with
+`IdempotencyInProgressError`. The flip side is that the lease has to be longer than the action
+can ever take, timeouts and internal retries included: a reservation that expires while the
+action is still running lets the next caller run it again.
+
+**Failures.** An action that raises, or is cancelled, deletes its reservation before the
+exception propagates, so the retry runs the action again — failures are not cached, and
+neither is the reservation of a failed action. The same happens when the result cannot be
+stored (an out-of-range TTL, a result the adapter cannot encode): the caller still gets the
+result, and the key is freed rather than holding retries for a record that never comes.
+
+**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.
+
+**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"`,
+the exception is the signal). The reservation is timed under `method="reserve"`.
+
+### Upgrading
+
+Records written before this change carry no `status` and read as completed, so nothing has
+to be migrated. Two things to know before turning the default on across a fleet:
+
+- **A custom repository needs `replace`** — `save` without `NX`. The coordinator raises
+  `TypeError` at construction when the repository lacks it and `in_flight` is not `"run"`.
+- **During a rolling upgrade**, an instance still on a version without `status` reads a
+  pending record as a completed one with a `null` result. With `PydanticResultAdapter` that
+  is a decode failure and the action runs, which is the old behaviour; with
+  `JsonResultAdapter` or `VoidResultAdapter` it replays as `None`. Roll out with
+  `in_flight="run"` and switch to `"wait"` once every instance is on the new version, or
+  accept that window.
+
 ## Advanced Use Cases
 
 ### Custom TTL
@@ -184,7 +258,7 @@ The `RedisAsyncIdempotencyRepository` is fully compatible with Redis Cluster. It
 
 ## Metrics and Observability
 
-You can inject a metrics collector into the repository and the coordinator to track hits, misses, collisions, errors and latency. Each metric has one owner, so the same collector goes to both without double counting: the coordinator records hit, miss, collision and the latency of `get` and `save`; the repository records errors, the bulk hit and miss counts of `get_many`, and the latency of `delete` and `get_many`.
+You can inject a metrics collector into the repository and the coordinator to track hits, misses, collisions, errors and latency. Each metric has one owner, so the same collector goes to both without double counting: the coordinator records hit, miss, collision and the latency of `get`, `reserve` and `save`; the repository records errors, the bulk hit and miss counts of `get_many`, and the latency of `delete` and `get_many`. A collision is two callers on one key at the same time — the second one found the first one's reservation, or, with `in_flight="run"`, its own save collided.
 
 ```python
 from idempotency_kit.core.protocols.metrics import IdempotencyMetricsProtocol
@@ -245,6 +319,20 @@ repo = RedisAsyncIdempotencyRepository(
 )
 ```
 
+### AsyncIdempotencyCoordinator
+
+```python
+coordinator = AsyncIdempotencyCoordinator(
+    repo,
+    service,
+    operation_ttls={"order.create": 3600},  # Per-operation TTLs in seconds; win over the decorator
+    metrics=custom_metrics,                 # Optional metrics collector
+    enabled=True,                           # False runs the action and nothing else
+    in_flight="wait",                       # "wait" | "raise" | "run", see In-flight requests
+    in_flight_lease_seconds=30,             # How long a reservation is held; must outlive the action
+)
+```
+
 ### Constants
 
 `idempotency_kit.core.constants` is the single source for the TTL defaults, and
@@ -256,8 +344,11 @@ from settings and you get the same numbers.
 | `DEFAULT_TTL_MINUTES` | 60 | how long a record is kept when no TTL is given |
 | `MIN_TTL_SECONDS` | 60 | the floor every TTL is raised to |
 | `MAX_TTL_SECONDS` | 2592000 | the ceiling, thirty days |
+| `DEFAULT_IN_FLIGHT_MODE` | `"wait"` | what a second caller gets while the first is in flight |
+| `DEFAULT_IN_FLIGHT_LEASE_SECONDS` | 30 | how long a reservation is held, and the longest a caller waits |
+| `IN_FLIGHT_POLL_INTERVAL_SECONDS` | 0.05 | how often a waiting caller re-reads the key |
 
-`MAX_KEY_LENGTH` and `MAX_OPERATION_LENGTH` live beside them.
+`MAX_KEY_LENGTH`, `MAX_OPERATION_LENGTH` and the `InFlightMode` type live beside them.
 
 #### Upgrading
 
@@ -287,6 +378,7 @@ The same `idempotency_key` can be used for different operations (e.g., `user.cre
 The library defines several exceptions to handle various idempotency scenarios:
 
 - **`IdempotencyKeyCollisionError`**: Raised by `repository.save()` when you try to save a result for a key that already exists. This typically means another identical request is either being processed or has already finished.
+- **`IdempotencyInProgressError`**: Raised by `coordinator.coordinate()` and the decorator when another call with the same key is still running its action — at once with `in_flight="raise"`, after a whole lease of waiting with `in_flight="wait"`. Map it to HTTP 409.
 - **`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 `service.create_record()` if validation of `operation` or `idempotency_key` fails (e.g., empty string or too long).
@@ -297,8 +389,9 @@ The library defines several exceptions to handle various idempotency scenarios:
 
 1. **Natural Keys**: Use natural unique identifiers as idempotency keys if possible (e.g., `order_id`, `message_id`).
 2. **Atomic Operations**: Always save the result to the cache *after* the business logic has successfully completed.
-3. **Pydantic Support**: The library works best with Pydantic models. Use `model_dump(mode="json")` when saving and `**cached.result` when restoring.
-4. **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).
+3. **Lease Longer Than the Action**: Set `in_flight_lease_seconds` above the longest the action can take, timeouts and retries included; a reservation that expires mid-run lets the next caller run the action again.
+4. **Pydantic Support**: The library works best with Pydantic models. Use `model_dump(mode="json")` when saving and `**cached.result` when restoring.
+5. **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).
 
 ## Production Examples
 
@@ -574,7 +667,7 @@ def create_api_container(settings: Settings) -> AsyncContainer:
 
 Your own providers supply the `Redis` client and the settings object; the settings object must satisfy `IdempotencySettingsProtocol`, which `BaseIdempotencySettings` does. `IdempotencyProvider` also provides the metrics collector: `PrometheusIdempotencyMetrics` when `metrics_enabled` is true (install the `prometheus` extra), a no-op collector otherwise. To use another metrics backend, 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. A settings object that predates the field is read as enabled.
+`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.
 
 ## Migration Guide
 
diff --git a/idempotency_kit/__init__.py b/idempotency_kit/__init__.py
index c072996..9d8d843 100644
--- a/idempotency_kit/__init__.py
+++ b/idempotency_kit/__init__.py
@@ -8,6 +8,7 @@
 from .core.decorators.aio.idempotent import async_idempotent
 from .core.exceptions import (
     IdempotencyError,
+    IdempotencyInProgressError,
     IdempotencyInvalidTTLError,
     IdempotencyKeyCollisionError,
     IdempotencyRecordExpiredError,
@@ -27,6 +28,7 @@
     "IdempotencyDomainService",
     "IdempotencyError",
     "IdempotencyIdentifiers",
+    "IdempotencyInProgressError",
     "IdempotencyInvalidTTLError",
     "IdempotencyKeyCollisionError",
     "IdempotencyMetricsProtocol",
diff --git a/idempotency_kit/core/constants.py b/idempotency_kit/core/constants.py
index 7395fc3..b426f37 100644
--- a/idempotency_kit/core/constants.py
+++ b/idempotency_kit/core/constants.py
@@ -1,5 +1,7 @@
 """Core constants for idempotency."""
 
+from typing import Literal
+
 # Key constraints
 # Maximum allowed length for idempotency key.
 MAX_KEY_LENGTH: int = 255
@@ -20,3 +22,20 @@
 
 # Maximum allowed TTL in seconds (30 days).
 MAX_TTL_SECONDS: int = 30 * 24 * 3600
+
+# In-flight handling
+# What the coordinator does with a second caller that arrives while the first one's
+# action is still running under the same key: wait for the first caller's result, raise
+# IdempotencyInProgressError at once, or run the action too.
+InFlightMode = Literal["wait", "raise", "run"]
+
+# Wait by default: a second caller in the retry window gets the first caller's result
+# instead of producing a second one.
+DEFAULT_IN_FLIGHT_MODE: InFlightMode = "wait"
+
+# How long a reservation is held before it counts as abandoned (30 seconds). It has to
+# outlive the action; a waiting caller gives up after the same span.
+DEFAULT_IN_FLIGHT_LEASE_SECONDS: int = 30
+
+# How often a waiting caller re-reads the key.
+IN_FLIGHT_POLL_INTERVAL_SECONDS: float = 0.05
diff --git a/idempotency_kit/core/exceptions.py b/idempotency_kit/core/exceptions.py
index 97a7651..24e59b8 100644
--- a/idempotency_kit/core/exceptions.py
+++ b/idempotency_kit/core/exceptions.py
@@ -54,3 +54,12 @@ def __init__(self, ttl_seconds: float, min_ttl: float, max_ttl: float) -> None:
         self.min_ttl = min_ttl
         self.max_ttl = max_ttl
         super().__init__(f"Invalid TTL {ttl_seconds}s. Must be between {min_ttl}s and {max_ttl}s")
+
+
+class IdempotencyInProgressError(IdempotencyError):
+    """Raised when another call with the same key is still running its action."""
+
+    def __init__(self, operation: str, key: str) -> None:
+        self.operation = operation
+        self.key = key
+        super().__init__(f"Idempotency record for operation '{operation}', key '{key}' is still in flight")
diff --git a/idempotency_kit/core/models/entities.py b/idempotency_kit/core/models/entities.py
index a59fe83..a4cc256 100644
--- a/idempotency_kit/core/models/entities.py
+++ b/idempotency_kit/core/models/entities.py
@@ -1,7 +1,7 @@
 """Idempotency record entity."""
 
 from datetime import UTC, datetime, timedelta
-from typing import Annotated, Self
+from typing import Annotated, Literal, Self
 
 from pydantic import BaseModel, ConfigDict, Field, JsonValue, StringConstraints, field_validator
 
@@ -34,6 +34,10 @@ class IdempotencyRecord(IdempotencyIdentifiers):
 
     Stores the result of an operation so that repeated calls with the same key
     return the cached result without re-executing business logic.
+
+    A *pending* record is an in-flight reservation: the key is taken, the action is still
+    running, and the result is not there yet. It is written before the action and replaced
+    by the completed record after it; its ``expires_at`` is the lease.
     """
 
     model_config = ConfigDict(frozen=True)
@@ -45,6 +49,12 @@ class IdempotencyRecord(IdempotencyIdentifiers):
     created_at: datetime = Field(description="When this record was created")
     expires_at: datetime = Field(description="When this record should expire from cache")
 
+    # Records written before the field existed carry no status and read as completed.
+    status: Literal["pending", "completed"] = Field(
+        default="completed",
+        description="'pending' while the action runs under a reservation, 'completed' once the result is stored",
+    )
+
     @classmethod
     def create(
         cls,
@@ -63,6 +73,29 @@ def create(
             expires_at=now + timedelta(seconds=ttl_seconds),
         )
 
+    @classmethod
+    def pending(
+        cls,
+        operation: str,
+        idempotency_key: str,
+        lease_seconds: float,
+    ) -> Self:
+        """Create the in-flight reservation for an action that is about to run."""
+        now = datetime.now(UTC)
+        return cls(
+            operation=operation,
+            idempotency_key=idempotency_key,
+            result=None,
+            created_at=now,
+            expires_at=now + timedelta(seconds=lease_seconds),
+            status="pending",
+        )
+
+    @property
+    def is_pending(self) -> bool:
+        """Whether this record is an in-flight reservation rather than a stored result."""
+        return self.status == "pending"
+
     @property
     def is_expired(self) -> bool:
         """Check if the record has expired."""
diff --git a/idempotency_kit/core/protocols/aio/repository.py b/idempotency_kit/core/protocols/aio/repository.py
index 2f22c0e..a14efd5 100644
--- a/idempotency_kit/core/protocols/aio/repository.py
+++ b/idempotency_kit/core/protocols/aio/repository.py
@@ -43,6 +43,22 @@ async def save(self, record: IdempotencyRecord) -> None:
         """
         ...
 
+    async def replace(self, record: IdempotencyRecord) -> None:
+        """Write a record whether or not the key is already there.
+
+        The coordinator completes an in-flight reservation with it: the pending record
+        under the key gives way to the final one.
+
+        Args:
+            record: Record to write
+
+        Raises:
+            IdempotencyValidationError: If record fails validation
+            IdempotencyStorageError: If storage operation fails
+            IdempotencyError: If internal error (e.g. serialization) occurs
+        """
+        ...
+
     async def delete(self, operation: str, idempotency_key: str) -> bool:
         """Delete an idempotency record.
 
diff --git a/idempotency_kit/core/services/aio/coordinator.py b/idempotency_kit/core/services/aio/coordinator.py
index a394e83..f482f15 100644
--- a/idempotency_kit/core/services/aio/coordinator.py
+++ b/idempotency_kit/core/services/aio/coordinator.py
@@ -1,10 +1,19 @@
+import asyncio
 import logging
 import time
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
+from enum import Enum, auto
 from typing import Any, Generic, TypeVar
 
+from idempotency_kit.core.constants import (
+    DEFAULT_IN_FLIGHT_LEASE_SECONDS,
+    DEFAULT_IN_FLIGHT_MODE,
+    IN_FLIGHT_POLL_INTERVAL_SECONDS,
+    InFlightMode,
+)
 from idempotency_kit.core.exceptions import (
+    IdempotencyInProgressError,
     IdempotencyInvalidTTLError,
     IdempotencyKeyCollisionError,
     IdempotencyRecordExpiredError,
@@ -19,6 +28,8 @@
 
 logger = logging.getLogger(__name__)
 
+_IN_FLIGHT_MODES: tuple[InFlightMode, ...] = ("wait", "raise", "run")
+
 
 @dataclass(frozen=True)
 class _Hit(Generic[T]):
@@ -32,6 +43,17 @@ class _Hit(Generic[T]):
     value: T
 
 
+class _Lookup(Enum):
+    """What a read found under the key when it was not a usable result."""
+
+    # No record, or an expired one: the key is free.
+    ABSENT = auto()
+    # A pending record: another caller holds the key and its action has not finished.
+    IN_FLIGHT = auto()
+    # A record this adapter cannot decode, or a read that failed: run the action.
+    UNUSABLE = auto()
+
+
 class AsyncIdempotencyCoordinator:
     """Coordinator for asynchronous idempotent operations.
 
@@ -41,6 +63,12 @@ class AsyncIdempotencyCoordinator:
         operation_ttls: Per-operation TTL overrides in seconds; wins over the decorator.
         metrics: Metrics collector for hits, misses, collisions, errors and latency.
         enabled: Set to False to make every call a pass-through to the action.
+        in_flight: What a second caller gets while the first one's action is still running
+            under the same key. ``"wait"`` (the default) waits for the first caller's result,
+            ``"raise"`` raises ``IdempotencyInProgressError`` at once, and ``"run"`` runs the
+            action too, as the coordinator did before reservations existed.
+        in_flight_lease_seconds: How long a reservation is held before it counts as
+            abandoned, and the longest a waiting caller waits. It has to outlive the action.
     """
 
     def __init__(
@@ -50,12 +78,28 @@ def __init__(
         operation_ttls: dict[str, int] | None = None,
         metrics: IdempotencyMetricsProtocol | None = None,
         enabled: bool = True,
+        in_flight: InFlightMode = DEFAULT_IN_FLIGHT_MODE,
+        in_flight_lease_seconds: int = DEFAULT_IN_FLIGHT_LEASE_SECONDS,
     ) -> 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 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
+            # lease would wait or be refused. Fail at construction instead.
+            raise TypeError(
+                f"{type(repository).__name__} has no replace(); in_flight={in_flight!r} completes a reservation "
+                "with it. Add the method, or pass in_flight='run'."
+            )
         self._repo = repository
         self._svc = domain_service
         self._operation_ttls = operation_ttls or {}
         self._metrics = metrics or NoOpIdempotencyMetrics()
         self._enabled = enabled
+        self._in_flight = in_flight
+        self._in_flight_lease_seconds = in_flight_lease_seconds
 
     async def coordinate(
         self,
@@ -72,13 +116,55 @@ async def coordinate(
 
         With ``enabled=False`` the action is simply run: nothing is read, nothing is
         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
+        the same key is still running its action and ``in_flight`` is ``"raise"``, or
+        ``"wait"`` and a whole lease has passed.
         """
         if not self._enabled or not idempotency_key:
             return await action(*args, **kwargs)
 
+        if self._in_flight == "run":
+            return await self._coordinate_unreserved(
+                operation, idempotency_key, ttl_seconds, adapter, action, *args, **kwargs
+            )
+
+        claim = await self._claim(operation, idempotency_key, adapter)
+        if isinstance(claim, _Hit):
+            return claim.value
+
+        try:
+            result = await action(*args, **kwargs)
+        except BaseException:
+            # Failures are not cached, so the reservation goes too and the retry runs
+            # again. Cancellation counts: nothing says the action completed.
+            if claim:
+                await self._try_release(operation, idempotency_key)
+            raise
+
+        ttl_minutes = self._resolve_ttl_minutes(operation, ttl_seconds)
+        completed = await self._try_complete(operation, idempotency_key, result, adapter, ttl_minutes)
+        if claim and not completed:
+            # Our reservation with no result behind it would make every retry within the
+            # lease wait for, or be refused over, a record that is never coming.
+            await self._try_release(operation, idempotency_key)
+        return result
+
+    async def _coordinate_unreserved(
+        self,
+        operation: str,
+        idempotency_key: str,
+        ttl_seconds: int | None,
+        adapter: ResultAdapter[T],
+        action: Callable[..., Awaitable[T]],
+        *args: Any,
+        **kwargs: Any,
+    ) -> T:
+        """The flow without a reservation: read, run, ``SET NX``, and adopt the winner's result on a collision."""
         # 1. Try to get from storage
         hit = await self._try_get_cached(operation, idempotency_key, adapter)
-        if hit is not None:
+        if isinstance(hit, _Hit):
             return hit.value
 
         # 2. Execute business logic
@@ -88,6 +174,64 @@ async def coordinate(
         ttl_minutes = self._resolve_ttl_minutes(operation, ttl_seconds)
         return await self._try_save_result(operation, idempotency_key, result, adapter, ttl_minutes)
 
+    async def _claim(
+        self,
+        operation: str,
+        idempotency_key: str,
+        adapter: ResultAdapter[T],
+    ) -> _Hit[T] | bool:
+        """Reserve the key, or replay the record that holds it.
+
+        Returns the hit when a completed record is there, ``True`` when the reservation is
+        ours, and ``False`` when the key holds nothing usable and the action has to run
+        unreserved. Raises ``IdempotencyInProgressError`` for a key another caller holds,
+        at once in ``"raise"`` mode and after a whole lease of waiting in ``"wait"`` mode.
+        """
+        deadline = time.monotonic() + self._in_flight_lease_seconds
+        read = False
+        waiting = False
+        while True:
+            reserved = await self._try_reserve(operation, idempotency_key)
+            if reserved is None:
+                return False
+            if reserved:
+                # A read that came before us has counted the miss already.
+                if not read:
+                    self._metrics.record_miss(operation)
+                return True
+
+            found = await self._try_get_cached(operation, idempotency_key, adapter)
+            read = True
+            if isinstance(found, _Hit):
+                return found
+            if found is _Lookup.UNUSABLE:
+                return False
+            if found is _Lookup.IN_FLIGHT:
+                if not waiting:
+                    waiting = True
+                    self._metrics.record_collision(operation)
+                    logger.info(
+                        "Idempotency key in flight",
+                        extra={
+                            "operation": operation,
+                            "idempotency_key": idempotency_key,
+                            "in_flight": self._in_flight,
+                        },
+                    )
+                if self._in_flight == "raise":
+                    raise IdempotencyInProgressError(operation, idempotency_key)
+            # ABSENT after a collision: the holder gave the key up between our write and
+            # our read, or its lease ran out. Reserve again, as a waiter does when the
+            # marker disappears; the deadline bounds both.
+            if time.monotonic() >= deadline:
+                logger.warning(
+                    "Idempotency key still in flight after the lease; refusing the call",
+                    extra={"operation": operation, "idempotency_key": idempotency_key},
+                )
+                raise IdempotencyInProgressError(operation, idempotency_key)
+            if found is _Lookup.IN_FLIGHT:
+                await asyncio.sleep(IN_FLIGHT_POLL_INTERVAL_SECONDS)
+
     def _resolve_ttl_minutes(self, operation: str, ttl_seconds: int | None) -> int | None:
         """Determine TTL in minutes based on settings and overrides."""
         effective_ttl_seconds = self._operation_ttls.get(operation) or ttl_seconds
@@ -95,33 +239,81 @@ def _resolve_ttl_minutes(self, operation: str, ttl_seconds: int | None) -> int |
             return None
         return max(1, effective_ttl_seconds // 60)
 
+    async def _try_reserve(self, operation: str, idempotency_key: str) -> bool | None:
+        """Write the pending record under ``SET NX``.
+
+        ``True`` when the reservation is ours, ``False`` when the key is already taken,
+        ``None`` when the write could not be made at all.
+        """
+        start_time = time.perf_counter()
+        try:
+            pending = self._svc.create_pending_record(
+                operation,
+                idempotency_key,
+                lease_seconds=self._in_flight_lease_seconds,
+            )
+            await self._repo.save(pending)
+        except IdempotencyKeyCollisionError:
+            return False
+        except IdempotencyValidationError:
+            self._metrics.record_error(operation, "record_validation_error")
+            logger.exception(
+                "Idempotency reservation rejected; this operation will not be cached",
+                extra={"operation": operation, "idempotency_key": idempotency_key},
+            )
+            return None
+        except Exception:
+            self._metrics.record_error(operation, "storage_reserve_error")
+            logger.exception(
+                "Idempotency coordinator error while reserving",
+                extra={"operation": operation, "idempotency_key": idempotency_key},
+            )
+            return None
+        else:
+            return True
+        finally:
+            self._metrics.record_latency(operation, "reserve", time.perf_counter() - start_time)
+
+    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:
+            await self._repo.delete(operation, idempotency_key)
+        except Exception:
+            self._metrics.record_error(operation, "storage_release_error")
+            logger.exception(
+                "Idempotency coordinator error while releasing a reservation",
+                extra={"operation": operation, "idempotency_key": idempotency_key},
+            )
+
     async def _try_get_cached(
         self,
         operation: str,
         idempotency_key: str,
         adapter: ResultAdapter[T],
-    ) -> _Hit[T] | None:
-        """Try to fetch and decode result from storage. Returns None on miss or error."""
+    ) -> _Hit[T] | _Lookup:
+        """Fetch and decode the record under the key; a read that fails is ``UNUSABLE``, never an exception."""
         start_time = time.perf_counter()
         try:
-            hit = await self._get_and_decode(operation, idempotency_key, adapter)
-            if hit is not None:
-                self._metrics.record_hit(operation)
-                logger.info(
-                    "Idempotency cache hit",
-                    extra={"operation": operation, "idempotency_key": idempotency_key},
-                )
-                return hit
-            self._metrics.record_miss(operation)
+            found = await self._get_and_decode(operation, idempotency_key, adapter)
         except Exception:
             self._metrics.record_error(operation, "storage_get_error")
             logger.exception(
                 "Idempotency coordinator error while fetching",
                 extra={"operation": operation, "idempotency_key": idempotency_key},
             )
+            return _Lookup.UNUSABLE
         finally:
             self._metrics.record_latency(operation, "get", time.perf_counter() - start_time)
-        return None
+
+        if isinstance(found, _Hit):
+            self._metrics.record_hit(operation)
+            logger.info(
+                "Idempotency cache hit",
+                extra={"operation": operation, "idempotency_key": idempotency_key},
+            )
+        elif found is not _Lookup.IN_FLIGHT:
+            self._metrics.record_miss(operation)
+        return found
 
     async def _try_save_result(
         self,
@@ -134,7 +326,7 @@ async def _try_save_result(
         """Try to save result to storage. Handles collisions and errors gracefully."""
         start_time = time.perf_counter()
         try:
-            await self._save_to_repo(operation, idempotency_key, result, adapter, ttl_minutes)
+            await self._save_to_repo(operation, idempotency_key, result, adapter, ttl_minutes, replace=False)
             logger.info(
                 "Idempotency result saved",
                 extra={"operation": operation, "idempotency_key": idempotency_key},
@@ -142,52 +334,88 @@ async def _try_save_result(
         except IdempotencyKeyCollisionError:
             return await self._handle_collision(operation, idempotency_key, result, adapter)
         except (IdempotencyValidationError, IdempotencyInvalidTTLError):
-            # The record itself is invalid — the adapter encoded something the
-            # storage format cannot hold, or the TTL is out of range. No retry can
-            # fix that, so it is reported as a contract violation rather than a
-            # storage blip. The result is still returned: the action has already
-            # run, and raising here would make the caller retry a completed
-            # operation — the one thing an idempotency layer must never cause.
-            self._metrics.record_error(operation, "record_validation_error")
-            logger.exception(
-                "Idempotency record rejected; this operation will not be cached",
-                extra={
-                    "operation": operation,
-                    "idempotency_key": idempotency_key,
-                    "adapter": type(adapter).__name__,
-                },
-            )
+            self._report_unstorable_record(operation, idempotency_key, adapter)
         except Exception:
-            self._metrics.record_error(operation, "storage_save_error")
-            logger.exception(
-                "Idempotency coordinator error while saving for operation",
+            self._report_save_failure(operation, idempotency_key)
+        finally:
+            self._metrics.record_latency(operation, "save", time.perf_counter() - start_time)
+
+        return result
+
+    async def _try_complete(
+        self,
+        operation: str,
+        idempotency_key: str,
+        result: T,
+        adapter: ResultAdapter[T],
+        ttl_minutes: int | None,
+    ) -> bool:
+        """Write the result over the reservation; ``False`` when it could not be stored, never an exception."""
+        start_time = time.perf_counter()
+        try:
+            await self._save_to_repo(operation, idempotency_key, result, adapter, ttl_minutes, replace=True)
+        except (IdempotencyValidationError, IdempotencyInvalidTTLError):
+            self._report_unstorable_record(operation, idempotency_key, adapter)
+            return False
+        except Exception:
+            self._report_save_failure(operation, idempotency_key)
+            return False
+        else:
+            logger.info(
+                "Idempotency result saved",
                 extra={"operation": operation, "idempotency_key": idempotency_key},
             )
+            return True
         finally:
             self._metrics.record_latency(operation, "save", time.perf_counter() - start_time)
 
-        return result
+    def _report_unstorable_record(self, operation: str, idempotency_key: str, adapter: ResultAdapter[T]) -> None:
+        # The record itself is invalid — the adapter encoded something the
+        # storage format cannot hold, or the TTL is out of range. No retry can
+        # fix that, so it is reported as a contract violation rather than a
+        # storage blip. The result is still returned: the action has already
+        # run, and raising here would make the caller retry a completed
+        # operation — the one thing an idempotency layer must never cause.
+        self._metrics.record_error(operation, "record_validation_error")
+        logger.exception(
+            "Idempotency record rejected; this operation will not be cached",
+            extra={
+                "operation": operation,
+                "idempotency_key": idempotency_key,
+                "adapter": type(adapter).__name__,
+            },
+        )
+
+    def _report_save_failure(self, operation: str, idempotency_key: str) -> None:
+        self._metrics.record_error(operation, "storage_save_error")
+        logger.exception(
+            "Idempotency coordinator error while saving for operation",
+            extra={"operation": operation, "idempotency_key": idempotency_key},
+        )
 
     async def _get_and_decode(
         self,
         operation: str,
         idempotency_key: str,
         adapter: ResultAdapter[T],
-    ) -> _Hit[T] | None:
-        """Fetch record from repository and decode it safely; ``None`` means no usable record."""
+    ) -> _Hit[T] | _Lookup:
+        """Fetch record from repository and decode it safely."""
         cached = await self._repo.get(operation, idempotency_key)
         if cached is None:
-            return None
+            return _Lookup.ABSENT
         try:
             # The protocol asks a repository not to return an expired record, but expiry is
             # the domain's rule to enforce, and a backend without native expiry cannot.
+            # An expired lease is an abandoned reservation, so it goes the same way.
             self._svc.validate_record(cached)
         except IdempotencyRecordExpiredError:
             logger.warning(
                 "Idempotency record expired; treating it as a miss",
                 extra={"operation": operation, "idempotency_key": idempotency_key},
             )
-            return None
+            return _Lookup.ABSENT
+        if cached.is_pending:
+            return _Lookup.IN_FLIGHT
         return self._decode_safely(adapter, cached.result, operation, idempotency_key)
 
     async def _save_to_repo(
@@ -197,15 +425,20 @@ async def _save_to_repo(
         result: T,
         adapter: ResultAdapter[T],
         ttl_minutes: int | None,
+        *,
+        replace: bool,
     ) -> None:
-        """Perform the actual save operation."""
+        """Perform the actual save operation: ``SET NX``, or a plain ``SET`` over our reservation."""
         record = self._svc.create_record(
             operation=operation,
             idempotency_key=idempotency_key,
             result=adapter.encode(result),
             ttl_minutes=ttl_minutes,
         )
-        await self._repo.save(record)
+        if replace:
+            await self._repo.replace(record)
+        else:
+            await self._repo.save(record)
 
     async def _handle_collision(
         self,
@@ -222,7 +455,7 @@ async def _handle_collision(
         )
         try:
             winner = await self._get_and_decode(operation, idempotency_key, adapter)
-            if winner is not None:
+            if isinstance(winner, _Hit):
                 return winner.value
         except Exception:
             logger.exception(
@@ -237,8 +470,8 @@ def _decode_safely(
         data: Any,
         operation: str,
         idempotency_key: str,
-    ) -> _Hit[T] | None:
-        """Try to decode data using adapter. Returns None and logs error on failure."""
+    ) -> _Hit[T] | _Lookup:
+        """Try to decode data using adapter. Returns ``UNUSABLE`` and logs error on failure."""
         try:
             return _Hit(adapter.decode(data))
         except Exception:
@@ -246,4 +479,4 @@ def _decode_safely(
                 "Idempotency decode error",
                 extra={"operation": operation, "idempotency_key": idempotency_key},
             )
-            return None
+            return _Lookup.UNUSABLE
diff --git a/idempotency_kit/core/services/domain.py b/idempotency_kit/core/services/domain.py
index e50193b..c44ca38 100644
--- a/idempotency_kit/core/services/domain.py
+++ b/idempotency_kit/core/services/domain.py
@@ -89,6 +89,41 @@ def create_record(
             # Re-map Pydantic validation error to domain validation error with detailed errors
             raise IdempotencyValidationError(str(e), errors=e.errors()) from e
 
+    def create_pending_record(
+        self,
+        operation: str,
+        idempotency_key: str,
+        *,
+        lease_seconds: int,
+    ) -> IdempotencyRecord:
+        """Create the in-flight reservation for an operation whose action is about to run.
+
+        The lease is not a record TTL and is not held to the TTL bounds: it only has to
+        outlive the action, and the coordinator decides it.
+
+        Args:
+            operation: Operation name (e.g., 'user.create')
+            idempotency_key: Unique key for this operation
+            lease_seconds: How long the reservation is held before it counts as abandoned
+
+        Returns:
+            A pending IdempotencyRecord ready to be saved
+
+        Raises:
+            IdempotencyValidationError: If validation fails, or the lease is below a second
+        """
+        if lease_seconds < 1:
+            raise IdempotencyValidationError(f"lease_seconds must be >= 1, got {lease_seconds}")
+
+        try:
+            return IdempotencyRecord.pending(
+                operation=operation,
+                idempotency_key=idempotency_key,
+                lease_seconds=lease_seconds,
+            )
+        except ValidationError as e:
+            raise IdempotencyValidationError(str(e), errors=e.errors()) from e
+
     def validate_record(self, record: IdempotencyRecord) -> None:
         """Validate that record is still usable.
 
diff --git a/idempotency_kit/dishka/aio/coordinator.py b/idempotency_kit/dishka/aio/coordinator.py
index 25db995..a7ca27d 100644
--- a/idempotency_kit/dishka/aio/coordinator.py
+++ b/idempotency_kit/dishka/aio/coordinator.py
@@ -8,6 +8,7 @@
     IdempotencyDomainService,
     IdempotencyMetricsProtocol,
 )
+from idempotency_kit.core.constants import DEFAULT_IN_FLIGHT_LEASE_SECONDS, DEFAULT_IN_FLIGHT_MODE
 
 from ..protocols import IdempotencySettingsProtocol
 
@@ -32,6 +33,8 @@ def get_coordinator(
             operation_ttls=settings.operation_ttls,
             metrics=metrics,
             # Read defensively: settings objects written against the protocol before
-            # ``enabled`` was part of it stay valid, and they mean enabled.
+            # these fields were part of it stay valid, and they mean the defaults.
             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),
         )
diff --git a/idempotency_kit/dishka/protocols.py b/idempotency_kit/dishka/protocols.py
index 7b5c850..2c59429 100644
--- a/idempotency_kit/dishka/protocols.py
+++ b/idempotency_kit/dishka/protocols.py
@@ -2,6 +2,8 @@
 
 from typing import Protocol, runtime_checkable
 
+from idempotency_kit.core.constants import InFlightMode
+
 
 @runtime_checkable
 class IdempotencySettingsProtocol(Protocol):
@@ -46,3 +48,19 @@ def max_ttl_seconds(self) -> int:
     def operation_ttls(self) -> dict[str, int]:
         """Specific TTLs for operations in seconds."""
         ...
+
+    @property
+    def in_flight(self) -> InFlightMode:
+        """What a second caller gets while the first one's action is still running.
+
+        A settings object without the attribute is read as ``"wait"``.
+        """
+        ...
+
+    @property
+    def in_flight_lease_seconds(self) -> int:
+        """How long an in-flight reservation is held before it counts as abandoned.
+
+        A settings object without the attribute is read as the default lease.
+        """
+        ...
diff --git a/idempotency_kit/infra/storage/redis/aio/repository.py b/idempotency_kit/infra/storage/redis/aio/repository.py
index 85ece48..d1fd6ae 100644
--- a/idempotency_kit/infra/storage/redis/aio/repository.py
+++ b/idempotency_kit/infra/storage/redis/aio/repository.py
@@ -177,21 +177,8 @@ async def get(self, operation: str, idempotency_key: str) -> IdempotencyRecord |
 
         return record
 
-    async def save(
-        self,
-        record: IdempotencyRecord,
-    ) -> None:
-        """Save record to Redis with NX (set if not exists).
-
-        Args:
-            record: Idempotency record to save
-
-        Raises:
-            IdempotencyKeyCollisionError: If key already exists in Redis
-            IdempotencyValidationError: If record is invalid or already expired
-            IdempotencyStorageError: If Redis operation fails
-            IdempotencyError: If serialization fails
-        """
+    def _prepare_write(self, record: IdempotencyRecord) -> tuple[str, bytes, int]:
+        """Validate and serialize a record for a write; returns the Redis key, the payload and the TTL in seconds."""
         operation = record.operation
         self._validate_inputs(operation, record.idempotency_key)
         key = self._make_key(operation, record.idempotency_key)
@@ -209,7 +196,11 @@ async def save(
 
         # Serialize record
         try:
-            data = orjson.dumps(record.model_dump(mode="json")) if _HAS_ORJSON else record.model_dump_json()
+            data = (
+                orjson.dumps(record.model_dump(mode="json"))
+                if _HAS_ORJSON
+                else record.model_dump_json().encode("utf-8")
+            )
         except Exception as e:
             self._metrics.record_error(operation, "serialization_error")
             logger.exception(
@@ -218,28 +209,74 @@ async def save(
             )
             raise IdempotencyError("Serialization failed") from e
 
-        # Use SET with NX (only if key doesn't exist) and EX (expiration)
+        return key, data, ttl_seconds
+
+    async def _set(
+        self, record: IdempotencyRecord, key: str, data: bytes, ttl_seconds: int, *, nx: bool, method: str
+    ) -> bool:
+        """``SET key data EX ttl_seconds``, with ``NX`` when asked; returns whether the key was written."""
         try:
-            was_set = await self._redis.set(key, data, ex=ttl_seconds, nx=True)
+            was_set = await self._redis.set(key, data, ex=ttl_seconds, nx=nx)
         except Exception as e:
-            self._metrics.record_error(operation, type(e).__name__)
+            self._metrics.record_error(record.operation, type(e).__name__)
             logger.exception(
-                "Redis error during save",
-                extra={"operation": operation, "key": record.idempotency_key},
+                f"Redis error during {method}",
+                extra={"operation": record.operation, "key": record.idempotency_key},
             )
             # In case of Redis error, we cannot guarantee idempotency.
             raise IdempotencyStorageError(
-                "Redis storage failure during save",
-                operation=operation,
+                f"Redis storage failure during {method}",
+                operation=record.operation,
                 original_error=e,
             ) from e
+        return bool(was_set)
 
-        if not was_set:
-            raise IdempotencyKeyCollisionError(operation, record.idempotency_key)
+    async def save(
+        self,
+        record: IdempotencyRecord,
+    ) -> None:
+        """Save record to Redis with NX (set if not exists).
+
+        Args:
+            record: Idempotency record to save
+
+        Raises:
+            IdempotencyKeyCollisionError: If key already exists in Redis
+            IdempotencyValidationError: If record is invalid or already expired
+            IdempotencyStorageError: If Redis operation fails
+            IdempotencyError: If serialization fails
+        """
+        key, data, ttl_seconds = self._prepare_write(record)
+
+        # Use SET with NX (only if key doesn't exist) and EX (expiration)
+        if not await self._set(record, key, data, ttl_seconds, nx=True, method="save"):
+            raise IdempotencyKeyCollisionError(record.operation, record.idempotency_key)
 
         logger.debug(
             "Saved idempotency record",
-            extra={"operation": operation, "key": record.idempotency_key, "ttl_seconds": ttl_seconds},
+            extra={"operation": record.operation, "key": record.idempotency_key, "ttl_seconds": ttl_seconds},
+        )
+
+    async def replace(self, record: IdempotencyRecord) -> None:
+        """Write record to Redis whether or not the key is already there.
+
+        This is how an in-flight reservation is completed: the pending record under the
+        key gives way to the final one. It is ``save`` without ``NX``.
+
+        Args:
+            record: Idempotency record to write
+
+        Raises:
+            IdempotencyValidationError: If record is invalid or already expired
+            IdempotencyStorageError: If Redis operation fails
+            IdempotencyError: If serialization fails
+        """
+        key, data, ttl_seconds = self._prepare_write(record)
+        await self._set(record, key, data, ttl_seconds, nx=False, method="replace")
+
+        logger.debug(
+            "Replaced idempotency record",
+            extra={"operation": record.operation, "key": record.idempotency_key, "ttl_seconds": ttl_seconds},
         )
 
     async def delete(self, operation: str, idempotency_key: str) -> bool:
diff --git a/idempotency_kit/settings.py b/idempotency_kit/settings.py
index bc01d44..2ded4f4 100644
--- a/idempotency_kit/settings.py
+++ b/idempotency_kit/settings.py
@@ -2,7 +2,14 @@
 
 from pydantic import BaseModel, Field
 
-from .core.constants import DEFAULT_TTL_MINUTES, MAX_TTL_SECONDS, MIN_TTL_SECONDS
+from .core.constants import (
+    DEFAULT_IN_FLIGHT_LEASE_SECONDS,
+    DEFAULT_IN_FLIGHT_MODE,
+    DEFAULT_TTL_MINUTES,
+    MAX_TTL_SECONDS,
+    MIN_TTL_SECONDS,
+    InFlightMode,
+)
 
 
 class BaseIdempotencySettings(BaseModel):
@@ -23,3 +30,14 @@ class BaseIdempotencySettings(BaseModel):
         default_factory=dict,
         description="Operation-specific TTLs in seconds (overrides decorator and default)",
     )
+    in_flight: InFlightMode = Field(
+        default=DEFAULT_IN_FLIGHT_MODE,
+        description=(
+            "What a second caller gets while the first one's action is still running under the same key: "
+            "wait for its result, raise IdempotencyInProgressError, or run the action too"
+        ),
+    )
+    in_flight_lease_seconds: int = Field(
+        default=DEFAULT_IN_FLIGHT_LEASE_SECONDS,
+        description="How long an in-flight reservation is held before it counts as abandoned (30 seconds)",
+    )
diff --git a/tests/integration/test_redis_integration.py b/tests/integration/test_redis_integration.py
index 1232acd..b2edb3f 100644
--- a/tests/integration/test_redis_integration.py
+++ b/tests/integration/test_redis_integration.py
@@ -7,7 +7,13 @@
 import pytest
 from redis.asyncio import Redis as AsyncRedisClient
 
-from idempotency_kit import IdempotencyDomainService, IdempotencyKeyCollisionError, IdempotencyRecord
+from idempotency_kit import (
+    AsyncIdempotencyCoordinator,
+    IdempotencyDomainService,
+    IdempotencyKeyCollisionError,
+    IdempotencyRecord,
+    JsonResultAdapter,
+)
 from idempotency_kit.infra.storage.redis.aio import RedisAsyncIdempotencyRepository
 
 
@@ -129,3 +135,36 @@ async def test__redis_repository__duplicate_save__raises_collision_error(redis_c
     # Act & Assert
     with pytest.raises(IdempotencyKeyCollisionError):
         await repo.save(record)
+
+
+@pytest.mark.asyncio
+async def test__coordinator__concurrent_callers_on_real_redis__run_the_action_once(
+    redis_client: AsyncRedisClient,
+) -> None:
+    """Two calls with one key, the second arriving while the first's action is still running: one side effect."""
+    # Arrange
+    coordinator = AsyncIdempotencyCoordinator(
+        RedisAsyncIdempotencyRepository(redis_client),
+        IdempotencyDomainService(),
+    )
+    key = str(uuid4())
+    charges: list[str] = []
+
+    async def charge_card(amount: int) -> dict[str, int | str]:
+        await asyncio.sleep(0.3)
+        charges.append(f"ch_{len(charges) + 1}")
+        return {"charge_id": charges[-1], "amount": amount}
+
+    # Act
+    first = asyncio.create_task(
+        coordinator.coordinate("payment.charge", key, 3600, JsonResultAdapter(), charge_card, 1999)
+    )
+    await asyncio.sleep(0.05)
+    second = asyncio.create_task(
+        coordinator.coordinate("payment.charge", key, 3600, JsonResultAdapter(), charge_card, 1999)
+    )
+    results = await asyncio.gather(first, second)
+
+    # Assert
+    assert charges == ["ch_1"]
+    assert results == [{"charge_id": "ch_1", "amount": 1999}, {"charge_id": "ch_1", "amount": 1999}]
diff --git a/tests/unit/core/conftest.py b/tests/unit/core/conftest.py
index 829840d..233ec1d 100644
--- a/tests/unit/core/conftest.py
+++ b/tests/unit/core/conftest.py
@@ -4,10 +4,12 @@
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
+from fakeredis import FakeAsyncRedis as AsyncRedisClient
 
 from idempotency_kit.core.models.entities import IdempotencyRecord
 from idempotency_kit.core.protocols.adapter import ResultAdapter
 from idempotency_kit.core.services.aio.coordinator import AsyncIdempotencyCoordinator
+from idempotency_kit.infra.storage.redis.aio import RedisAsyncIdempotencyRepository
 
 
 @pytest.fixture
@@ -41,13 +43,24 @@ def mock_adapter() -> MagicMock:
 
 @pytest.fixture
 def coordinator(mock_repo: AsyncMock, mock_domain_service: MagicMock) -> AsyncIdempotencyCoordinator:
-    """Create coordinator with mocked dependencies."""
+    """Create coordinator with mocked dependencies.
+
+    It runs the unreserved flow -- read, run, ``SET NX`` -- which is the choreography the
+    mock-based tests describe; the reservation has its own tests against a fake Redis.
+    """
     return AsyncIdempotencyCoordinator(
         repository=mock_repo,
         domain_service=mock_domain_service,
+        in_flight="run",
     )
 
 
+@pytest.fixture
+def redis_repository(fake_redis: AsyncRedisClient) -> RedisAsyncIdempotencyRepository:
+    """Create the shipped repository on a fake Redis, for tests that need real storage semantics."""
+    return RedisAsyncIdempotencyRepository(fake_redis, key_prefix="probe:")
+
+
 @pytest.fixture
 def mock_coordinator() -> MagicMock:
     """Create mock coordinator for decorator tests."""
diff --git a/tests/unit/core/test_coordinator.py b/tests/unit/core/test_coordinator.py
index b516dd3..078e616 100644
--- a/tests/unit/core/test_coordinator.py
+++ b/tests/unit/core/test_coordinator.py
@@ -296,7 +296,8 @@ async def test__coordinator__record_validation_error__returns_result_and_reports
 
     # Assert
     assert result == {"data": "ok"}
-    mock_repo.save.assert_not_called()
+    mock_repo.replace.assert_not_called()
+    mock_repo.delete.assert_called_once_with("op", "key")  # the reservation goes with the result that never came
     metrics.record_error.assert_called_once_with("op", "record_validation_error")
 
 
diff --git a/tests/unit/core/test_entities.py b/tests/unit/core/test_entities.py
index 2917d0f..586dec3 100644
--- a/tests/unit/core/test_entities.py
+++ b/tests/unit/core/test_entities.py
@@ -110,3 +110,44 @@ def test__idempotency_record__any_json_result__is_accepted(result: object) -> No
 
     # Assert
     assert record.result == result
+
+
+def test__idempotency_record__built_without_a_status__is_completed() -> None:
+    """A record is a stored result unless it says otherwise."""
+    # Act
+    record = IdempotencyRecord.create(operation="op", idempotency_key="key", result={"id": 1}, ttl_seconds=60)
+
+    # Assert
+    assert record.status == "completed"
+    assert not record.is_pending
+
+
+def test__idempotency_record__json_written_before_the_status_existed__still_decodes_as_completed() -> None:
+    """Records already in Redis carry no ``status``; they must read back as the results they are."""
+    # Arrange
+    stored = (
+        '{"operation": "op", "idempotency_key": "key", "result": {"id": 1}, '
+        '"created_at": "2026-09-06T00:00:00Z", "expires_at": "2126-09-06T00:00:00Z"}'
+    )
+
+    # Act
+    record = IdempotencyRecord.model_validate_json(stored)
+
+    # Assert
+    assert record.status == "completed"
+    assert record.result == {"id": 1}
+    assert not record.is_pending
+
+
+def test__idempotency_record_pending__lease__creates_an_in_flight_reservation() -> None:
+    """The reservation is the record in its pending state, with the lease as its expiry."""
+    # Act
+    record = IdempotencyRecord.pending(operation="op", idempotency_key="key", lease_seconds=30)
+
+    # Assert
+    assert record.is_pending
+    assert record.status == "pending"
+    assert record.result is None
+    assert not record.is_expired
+    assert 29 < record.ttl_seconds <= 30
+    assert record.model_dump(mode="json")["status"] == "pending"
diff --git a/tests/unit/core/test_idempotent.py b/tests/unit/core/test_idempotent.py
index a30e192..cb64813 100644
--- a/tests/unit/core/test_idempotent.py
+++ b/tests/unit/core/test_idempotent.py
@@ -7,6 +7,7 @@
 import pytest
 
 from idempotency_kit.core.decorators.aio.idempotent import async_idempotent
+from idempotency_kit.core.exceptions import IdempotencyInProgressError
 from idempotency_kit.core.services.aio.coordinator import AsyncIdempotencyCoordinator
 
 
@@ -245,3 +246,20 @@ async def my_func(**kwargs: Any) -> str:
     # Assert
     assert result == "ok"
     mock_coordinator.coordinate.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test__decorator__key_in_flight__lets_the_refusal_through(
+    mock_coordinator: MagicMock, mock_adapter: MagicMock
+) -> None:
+    """A second caller refused by the coordinator is refused by the decorated function too; it is the caller's 409."""
+    # Arrange
+    mock_coordinator.coordinate.side_effect = IdempotencyInProgressError("test.op", "test-key")
+
+    @async_idempotent(operation="test.op", adapter=mock_adapter)
+    async def my_func(*, idempotency_key: str | None = None, coord: AsyncIdempotencyCoordinator) -> str:
+        return "not used"
+
+    # Act & Assert
+    with pytest.raises(IdempotencyInProgressError):
+        await my_func(idempotency_key="test-key", coord=mock_coordinator)
diff --git a/tests/unit/core/test_in_flight.py b/tests/unit/core/test_in_flight.py
new file mode 100644
index 0000000..604d91b
--- /dev/null
+++ b/tests/unit/core/test_in_flight.py
@@ -0,0 +1,396 @@
+"""The in-flight reservation: what a second caller gets while the first one's action is still running.
+
+Regression cover for issue #26: two ``coordinate()`` calls with the same key, the second starting
+while the first's action was still running, both ran the action -- the check was a ``GET`` and the
+write a ``SET NX`` after the action, and nothing marked the key as taken in between.
+"""
+
+import asyncio
+from datetime import UTC, datetime, timedelta
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import orjson
+import pytest
+from fakeredis import FakeAsyncRedis as AsyncRedisClient
+
+from idempotency_kit import (
+    AsyncIdempotencyCoordinator,
+    IdempotencyDomainService,
+    IdempotencyInProgressError,
+    IdempotencyKeyCollisionError,
+    IdempotencyMetricsProtocol,
+    IdempotencyRecord,
+    IdempotencyValidationError,
+    JsonResultAdapter,
+)
+from idempotency_kit.core.constants import InFlightMode
+from idempotency_kit.infra.storage.redis.aio import RedisAsyncIdempotencyRepository
+
+COORDINATOR_MODULE = "idempotency_kit.core.services.aio.coordinator"
+
+
+class _Provider:
+    """A payment provider that remembers every charge it made, the way the reporter's script does."""
+
+    def __init__(self, delay: float = 0.2) -> None:
+        self.charges: list[str] = []
+        self._delay = delay
+
+    async def charge(self, amount: int) -> dict[str, Any]:
+        await asyncio.sleep(self._delay)
+        self.charges.append(f"ch_{len(self.charges) + 1}")
+        return {"charge_id": self.charges[-1], "amount": amount}
+
+
+async def _two_callers_50ms_apart(
+    coordinator: AsyncIdempotencyCoordinator, provider: _Provider
+) -> list[dict[str, Any] | BaseException]:
+    first = asyncio.create_task(
+        coordinator.coordinate("payment.charge", "order-42", 3600, JsonResultAdapter(), provider.charge, 1999)
+    )
+    await asyncio.sleep(0.05)
+    second = asyncio.create_task(
+        coordinator.coordinate("payment.charge", "order-42", 3600, JsonResultAdapter(), provider.charge, 1999)
+    )
+    return await asyncio.gather(first, second, return_exceptions=True)
+
+
+@pytest.mark.asyncio
+async def test__coordinator__second_caller_while_the_first_is_in_flight__waits_and_replays_without_running_the_action(
+    redis_repository: RedisAsyncIdempotencyRepository,
+) -> None:
+    """The reporter's scenario: the card is charged once and both callers get that charge."""
+    # Arrange
+    coordinator = AsyncIdempotencyCoordinator(repository=redis_repository, domain_service=IdempotencyDomainService())
+    provider = _Provider()
+
+    # Act
+    results = await _two_callers_50ms_apart(coordinator, provider)
+
+    # Assert
+    assert provider.charges == ["ch_1"]
+    assert results == [{"charge_id": "ch_1", "amount": 1999}, {"charge_id": "ch_1", "amount": 1999}]
+    stored = await redis_repository.get("payment.charge", "order-42")
+    assert stored is not None
+    assert stored.status == "completed"
+
+
+@pytest.mark.asyncio
+async def test__coordinator__in_flight_raise__second_caller_is_refused_and_the_action_runs_once(
+    redis_repository: RedisAsyncIdempotencyRepository,
+) -> None:
+    """The 409 shape: the retry is told the original is still running instead of waiting for it."""
+    # Arrange
+    coordinator = AsyncIdempotencyCoordinator(
+        repository=redis_repository, domain_service=IdempotencyDomainService(), in_flight="raise"
+    )
+    provider = _Provider()
+
+    # Act
+    results = await _two_callers_50ms_apart(coordinator, provider)
+
+    # Assert
+    assert provider.charges == ["ch_1"]
+    assert results[0] == {"charge_id": "ch_1", "amount": 1999}
+    assert isinstance(results[1], IdempotencyInProgressError)
+    assert (results[1].operation, results[1].key) == ("payment.charge", "order-42")
+
+
+@pytest.mark.asyncio
+async def test__coordinator__in_flight_run__both_callers_run_the_action_and_the_loser_adopts_the_winner_result(
+    redis_repository: RedisAsyncIdempotencyRepository,
+) -> None:
+    """The flow before reservations existed, kept for actions that are safe to repeat."""
+    # Arrange
+    coordinator = AsyncIdempotencyCoordinator(
+        repository=redis_repository, domain_service=IdempotencyDomainService(), in_flight="run"
+    )
+    provider = _Provider()
+
+    # Act
+    results = await _two_callers_50ms_apart(coordinator, provider)
+
+    # Assert
+    assert provider.charges == ["ch_1", "ch_2"]
+    assert results == [{"charge_id": "ch_1", "amount": 1999}, {"charge_id": "ch_1", "amount": 1999}]
+
+
+@pytest.mark.asyncio
+async def test__coordinator__in_flight_wait__one_miss_one_collision_one_hit(
+    redis_repository: RedisAsyncIdempotencyRepository,
+) -> None:
+    """The runner is a miss; the waiter records a collision on finding the marker and a hit on getting the record."""
+    # Arrange
+    metrics = MagicMock(spec=IdempotencyMetricsProtocol)
+    coordinator = AsyncIdempotencyCoordinator(
+        repository=redis_repository, domain_service=IdempotencyDomainService(), metrics=metrics
+    )
+    provider = _Provider()
+
+    # Act
+    await _two_callers_50ms_apart(coordinator, provider)
+
+    # Assert
+    assert metrics.record_miss.call_args_list == [(("payment.charge",),)]
+    assert metrics.record_collision.call_args_list == [(("payment.charge",),)]
+    assert metrics.record_hit.call_args_list == [(("payment.charge",),)]
+    assert metrics.record_error.call_count == 0
+    assert {call.args[1] for call in metrics.record_latency.call_args_list} == {"reserve", "save", "get"}
+
+
+@pytest.mark.asyncio
+async def test__coordinator__action_raises__releases_the_reservation_so_the_retry_runs_again(
+    redis_repository: RedisAsyncIdempotencyRepository, fake_redis: AsyncRedisClient
+) -> None:
+    """Failures are not cached, and neither is the reservation of a failed action."""
+    # Arrange
+    coordinator = AsyncIdempotencyCoordinator(repository=redis_repository, domain_service=IdempotencyDomainService())
+    calls = 0
+
+    async def flaky() -> str:
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            raise RuntimeError("provider down")
+        return "charged"
+
+    # Act
+    with pytest.raises(RuntimeError, match="provider down"):
+        await coordinator.coordinate("op", "key", 600, JsonResultAdapter(), flaky)
+    left_behind = await fake_redis.get("probe:op:key")
+    retried = await coordinator.coordinate("op", "key", 600, JsonResultAdapter(), flaky)
+
+    # Assert
+    assert left_behind is None
+    assert retried == "charged"
+    assert calls == 2
+
+
+@pytest.mark.asyncio
+async def test__coordinator__action_cancelled__releases_the_reservation(
+    redis_repository: RedisAsyncIdempotencyRepository, fake_redis: AsyncRedisClient
+) -> None:
+    """A cancelled action did not complete as far as anyone knows; the key must not stay taken."""
+    # Arrange
+    coordinator = AsyncIdempotencyCoordinator(repository=redis_repository, domain_service=IdempotencyDomainService())
+
+    async def slow() -> str:
+        await asyncio.sleep(10)
+        return "never"
+
+    # Act
+    task = asyncio.create_task(coordinator.coordinate("op", "key", 600, JsonResultAdapter(), slow))
+    await asyncio.sleep(0.05)
+    assert await fake_redis.get("probe:op:key") is not None
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+
+    # Assert
+    assert await fake_redis.get("probe:op:key") is None
+
+
+@pytest.mark.asyncio
+async def test__coordinator__holder_fails_while_a_caller_waits__the_waiter_runs_the_action(
+    redis_repository: RedisAsyncIdempotencyRepository,
+) -> None:
+    """When the marker disappears, the waiter does not give up: it takes the key and runs the action itself."""
+    # Arrange
+    coordinator = AsyncIdempotencyCoordinator(repository=redis_repository, domain_service=IdempotencyDomainService())
+    calls = 0
+
+    async def action() -> str:
+        nonlocal calls
+        calls += 1
+        attempt = calls
+        await asyncio.sleep(0.1)
+        if attempt == 1:
+            raise RuntimeError("first attempt failed")
+        return "second attempt"
+
+    # Act
+    first = asyncio.create_task(coordinator.coordinate("op", "key", 600, JsonResultAdapter(), action))
+    await asyncio.sleep(0.05)
+    second = asyncio.create_task(coordinator.coordinate("op", "key", 600, JsonResultAdapter(), action))
+    results = await asyncio.gather(first, second, return_exceptions=True)
+
+    # Assert
+    assert isinstance(results[0], RuntimeError)
+    assert results[1] == "second attempt"
+    assert calls == 2
+
+
+@pytest.mark.asyncio
+async def test__coordinator__pending_record_past_its_lease__counts_as_absent_and_the_action_runs(
+    redis_repository: RedisAsyncIdempotencyRepository, fake_redis: AsyncRedisClient
+) -> None:
+    """A worker that crashed mid-action cannot wedge the key beyond its lease."""
+    # Arrange
+    coordinator = AsyncIdempotencyCoordinator(repository=redis_repository, domain_service=IdempotencyDomainService())
+    now = datetime.now(UTC)
+    abandoned = IdempotencyRecord(
+        operation="op",
+        idempotency_key="key",
+        result=None,
+        created_at=now - timedelta(seconds=60),
+        expires_at=now - timedelta(seconds=30),
+        status="pending",
+    )
+    await fake_redis.set("probe:op:key", orjson.dumps(abandoned.model_dump(mode="json")))
+    action = AsyncMock(return_value="fresh")
+
+    # Act
+    result = await coordinator.coordinate("op", "key", 600, JsonResultAdapter(), action)
+
+    # Assert
+    assert result == "fresh"
+    action.assert_awaited_once()
+    stored = await redis_repository.get("op", "key")
+    assert stored is not None
+    assert (stored.status, stored.result) == ("completed", "fresh")
+
+
+@pytest.mark.asyncio
+async def test__coordinator__in_flight_wait__still_pending_after_a_whole_lease__raises_in_progress(
+    mock_repo: AsyncMock, mock_adapter: MagicMock
+) -> None:
+    """The wait is bounded by the lease; past it the caller is refused rather than kept forever."""
+    # Arrange
+    coordinator = AsyncIdempotencyCoordinator(
+        repository=mock_repo, domain_service=IdempotencyDomainService(), in_flight_lease_seconds=30
+    )
+    mock_repo.save.side_effect = IdempotencyKeyCollisionError("op", "key")
+    mock_repo.get.return_value = IdempotencyRecord.pending("op", "key", lease_seconds=3600)
+    action = AsyncMock()
+
+    # Act & Assert
+    with (
+        patch(f"{COORDINATOR_MODULE}.time.monotonic", side_effect=[0.0, 10.0, 31.0]),
+        patch(f"{COORDINATOR_MODULE}.asyncio.sleep", new=AsyncMock()) as sleep,
+        pytest.raises(IdempotencyInProgressError),
+    ):
+        await coordinator.coordinate("op", "key", 60, mock_adapter, action)
+    action.assert_not_called()
+    sleep.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test__coordinator__storage_error_on_reserve__runs_the_action_unreserved(
+    mock_repo: AsyncMock, mock_adapter: MagicMock
+) -> None:
+    """Availability over exactly-once, as on every other storage failure; and nothing is released that was not taken."""
+    # Arrange
+    metrics = MagicMock(spec=IdempotencyMetricsProtocol)
+    coordinator = AsyncIdempotencyCoordinator(
+        repository=mock_repo, domain_service=IdempotencyDomainService(), metrics=metrics
+    )
+    mock_repo.save.side_effect = Exception("redis down")
+    action = AsyncMock(return_value={"data": "ok"})
+
+    # Act
+    result = await coordinator.coordinate("op", "key", 60, mock_adapter, action)
+
+    # Assert
+    assert result == {"data": "ok"}
+    action.assert_awaited_once()
+    mock_repo.replace.assert_awaited_once()
+    mock_repo.delete.assert_not_called()
+    metrics.record_error.assert_called_once_with("op", "storage_reserve_error")
+
+
+@pytest.mark.asyncio
+async def test__coordinator__undecodable_record_under_the_key__runs_the_action_and_replaces_it(
+    mock_repo: AsyncMock, mock_adapter: MagicMock
+) -> None:
+    """A record this adapter cannot read is a miss, and the fresh result heals it instead of re-running until expiry."""
+    # Arrange
+    coordinator = AsyncIdempotencyCoordinator(repository=mock_repo, domain_service=IdempotencyDomainService())
+    mock_repo.save.side_effect = IdempotencyKeyCollisionError("op", "key")
+    mock_repo.get.return_value = IdempotencyRecord.create("op", "key", {"old": "shape"}, ttl_seconds=600)
+    mock_adapter.decode.side_effect = Exception("decode failed")
+    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()
+    mock_repo.delete.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test__coordinator__release_fails__the_action_error_still_propagates(
+    mock_repo: AsyncMock, mock_adapter: MagicMock
+) -> None:
+    """Storage trouble on the way out is logged and counted; the caller sees the action's own failure."""
+    # Arrange
+    metrics = MagicMock(spec=IdempotencyMetricsProtocol)
+    coordinator = AsyncIdempotencyCoordinator(
+        repository=mock_repo, domain_service=IdempotencyDomainService(), metrics=metrics
+    )
+    mock_repo.delete.side_effect = Exception("redis down")
+    action = AsyncMock(side_effect=RuntimeError("provider down"))
+
+    # Act & Assert
+    with pytest.raises(RuntimeError, match="provider down"):
+        await coordinator.coordinate("op", "key", 60, mock_adapter, action)
+    metrics.record_error.assert_called_once_with("op", "storage_release_error")
+
+
+@pytest.mark.asyncio
+async def test__coordinator__in_flight_run__pending_record_from_a_reserving_peer__runs_the_action(
+    mock_repo: AsyncMock, mock_adapter: MagicMock
+) -> None:
+    """A coordinator in ``run`` mode meeting a marker (a mixed rollout) runs and keeps its own result."""
+    # Arrange
+    coordinator = AsyncIdempotencyCoordinator(
+        repository=mock_repo, domain_service=IdempotencyDomainService(), in_flight="run"
+    )
+    mock_repo.get.return_value = IdempotencyRecord.pending("op", "key", lease_seconds=30)
+    mock_repo.save.side_effect = IdempotencyKeyCollisionError("op", "key")
+    action = AsyncMock(return_value={"data": "mine"})
+
+    # Act
+    result = await coordinator.coordinate("op", "key", 60, mock_adapter, action)
+
+    # Assert
+    assert result == {"data": "mine"}
+    action.assert_awaited_once()
+    mock_adapter.decode.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    ("kwargs", "match"),
+    [
+        ({"in_flight": "block"}, "in_flight must be one of"),
+        ({"in_flight_lease_seconds": 0}, "in_flight_lease_seconds must be >= 1"),
+    ],
+    ids=["unknown_mode", "zero_lease"],
+)
+def test__coordinator__invalid_in_flight_settings__raise_validation_error(
+    mock_repo: AsyncMock, kwargs: dict[str, Any], match: str
+) -> None:
+    # Act & Assert
+    with pytest.raises(IdempotencyValidationError, match=match):
+        AsyncIdempotencyCoordinator(repository=mock_repo, domain_service=IdempotencyDomainService(), **kwargs)
+
+
+@pytest.mark.parametrize("in_flight", ["wait", "raise"])
+def test__coordinator__repository_without_replace__raises_type_error_unless_in_flight_is_run(
+    in_flight: InFlightMode,
+) -> None:
+    """A repository written against the old protocol fails at construction, not as a 30 s hang on every retry."""
+    # Arrange
+    legacy_repository = MagicMock(spec=["get", "save", "delete", "get_many", "save_many", "delete_many"])
+
+    # Act & Assert
+    with pytest.raises(TypeError, match="has no replace"):
+        AsyncIdempotencyCoordinator(
+            repository=legacy_repository, domain_service=IdempotencyDomainService(), in_flight=in_flight
+        )
+    AsyncIdempotencyCoordinator(
+        repository=legacy_repository, domain_service=IdempotencyDomainService(), in_flight="run"
+    )
diff --git a/tests/unit/core/test_services.py b/tests/unit/core/test_services.py
index 7f9e5c6..9c31f1c 100644
--- a/tests/unit/core/test_services.py
+++ b/tests/unit/core/test_services.py
@@ -12,7 +12,13 @@
     IdempotencyRecordExpiredError,
     IdempotencyValidationError,
 )
-from idempotency_kit.core.constants import DEFAULT_TTL_MINUTES, MAX_TTL_SECONDS, MIN_TTL_SECONDS
+from idempotency_kit.core.constants import (
+    DEFAULT_IN_FLIGHT_LEASE_SECONDS,
+    DEFAULT_IN_FLIGHT_MODE,
+    DEFAULT_TTL_MINUTES,
+    MAX_TTL_SECONDS,
+    MIN_TTL_SECONDS,
+)
 from idempotency_kit.core.protocols.metrics import NoOpIdempotencyMetrics
 from idempotency_kit.settings import BaseIdempotencySettings
 
@@ -268,3 +274,50 @@ def test__domain_service_and_settings__agree_on_the_ttl_defaults() -> None:
     # Assert
     assert from_service == from_settings
     assert from_settings == (DEFAULT_TTL_MINUTES, MIN_TTL_SECONDS, MAX_TTL_SECONDS)
+
+
+def test__domain_service__create_pending_record__is_not_held_to_the_ttl_bounds() -> None:
+    """A lease is shorter than the record floor of a minute, and that is fine: it only has to outlive the action."""
+    # Arrange
+    service = IdempotencyDomainService()
+
+    # Act
+    record = service.create_pending_record("payment.charge", "order-42", lease_seconds=30)
+
+    # Assert
+    assert record.is_pending
+    assert record.operation == "payment.charge"
+    assert record.idempotency_key == "order-42"
+    assert 29 < record.ttl_seconds <= 30
+
+
+def test__domain_service__create_pending_record__sub_second_lease__raises_validation_error() -> None:
+    # Arrange
+    service = IdempotencyDomainService()
+
+    # Act & Assert
+    with pytest.raises(IdempotencyValidationError, match="lease_seconds must be >= 1"):
+        service.create_pending_record("op", "key", lease_seconds=0)
+
+
+def test__domain_service__create_pending_record__colon_in_key__raises_validation_error() -> None:
+    """The identifiers obey the same rules whichever state the record is in."""
+    # Arrange
+    service = IdempotencyDomainService()
+
+    # Act & Assert
+    with pytest.raises(IdempotencyValidationError, match="cannot contain ':'"):
+        service.create_pending_record("op", "key:123", lease_seconds=30)
+
+
+def test__coordinator_and_settings__agree_on_the_in_flight_defaults() -> None:
+    """One canonical mode and lease, so a coordinator built by hand or from settings behaves the same."""
+    # Arrange
+    settings = BaseIdempotencySettings(key_prefix="probe:")
+
+    # Assert
+    assert (settings.in_flight, settings.in_flight_lease_seconds) == (
+        DEFAULT_IN_FLIGHT_MODE,
+        DEFAULT_IN_FLIGHT_LEASE_SECONDS,
+    )
+    assert (DEFAULT_IN_FLIGHT_MODE, DEFAULT_IN_FLIGHT_LEASE_SECONDS) == ("wait", 30)
diff --git a/tests/unit/dishka/test_providers.py b/tests/unit/dishka/test_providers.py
index 00bf069..52b19aa 100644
--- a/tests/unit/dishka/test_providers.py
+++ b/tests/unit/dishka/test_providers.py
@@ -13,6 +13,7 @@
     JsonResultAdapter,
     NoOpIdempotencyMetrics,
 )
+from idempotency_kit.core.constants import InFlightMode
 from idempotency_kit.dishka import (
     AsyncIdempotencyCoordinatorProvider,
     AsyncRedisIdempotencyProvider,
@@ -28,10 +29,19 @@ class _AppProvider(Provider):
 
     scope = Scope.APP
 
-    def __init__(self, *, metrics_enabled: bool, enabled: bool = True) -> None:
+    def __init__(
+        self,
+        *,
+        metrics_enabled: bool,
+        enabled: bool = True,
+        in_flight: InFlightMode = "wait",
+        in_flight_lease_seconds: int = 30,
+    ) -> 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
 
     @provide
     def settings(self) -> IdempotencySettingsProtocol:
@@ -40,6 +50,8 @@ def settings(self) -> IdempotencySettingsProtocol:
             key_prefix="probe:",
             metrics_enabled=self._metrics_enabled,
             enabled=self._enabled,
+            in_flight=self._in_flight,
+            in_flight_lease_seconds=self._in_flight_lease_seconds,
         )
 
     @provide
@@ -187,3 +199,25 @@ async def action() -> str:
         assert await repository.get("op.disabled", "key") is None
     finally:
         await container.close()
+
+
+@pytest.mark.asyncio
+async def test__shipped_providers__in_flight_settings__reach_the_coordinator() -> None:
+    """The mode and the lease on the settings object are what the provided coordinator runs with."""
+    # Arrange
+    container = make_async_container(
+        _AppProvider(metrics_enabled=False, in_flight="raise", in_flight_lease_seconds=5),
+        IdempotencyProvider(),
+        AsyncRedisIdempotencyProvider(),
+        AsyncIdempotencyCoordinatorProvider(),
+    )
+
+    # Act
+    try:
+        coordinator = await container.get(AsyncIdempotencyCoordinator)
+
+        # Assert
+        assert coordinator._in_flight == "raise"
+        assert coordinator._in_flight_lease_seconds == 5
+    finally:
+        await container.close()
diff --git a/tests/unit/storage/redis/aio/test_repository.py b/tests/unit/storage/redis/aio/test_repository.py
index e58ca7d..f1db14a 100644
--- a/tests/unit/storage/redis/aio/test_repository.py
+++ b/tests/unit/storage/redis/aio/test_repository.py
@@ -508,7 +508,13 @@ async def action() -> dict[str, int]:
     # Assert
     assert metrics_mock.record_miss.call_count == 1
     assert metrics_mock.record_hit.call_count == 1
-    assert [call.args[1] for call in metrics_mock.record_latency.call_args_list] == ["get", "save", "get"]
+    assert metrics_mock.record_collision.call_count == 0
+    assert [call.args[1] for call in metrics_mock.record_latency.call_args_list] == [
+        "reserve",
+        "save",
+        "reserve",
+        "get",
+    ]
 
 
 @pytest.mark.asyncio
@@ -545,3 +551,65 @@ async def test_redis_save_subsecond_ttl(fake_redis: AsyncRedisClient) -> None:
     # fake-redis might return bytes
     key_found = any(k.decode() == f"{custom_prefix}test:key" for k in keys)
     assert key_found is True
+
+
+@pytest.mark.asyncio
+async def test_redis_replace_overwrites_the_pending_record(fake_redis: AsyncRedisClient) -> None:
+    """replace() is save() without NX: it is how a reservation becomes a result."""
+    repo = RedisAsyncIdempotencyRepository(fake_redis)
+    service = IdempotencyDomainService()
+    await repo.save(service.create_pending_record("test", "key", lease_seconds=30))
+
+    await repo.replace(service.create_record("test", "key", {"foo": "bar"}, ttl_minutes=10))
+
+    stored = await repo.get("test", "key")
+    assert stored is not None
+    assert stored.status == "completed"
+    assert stored.result == {"foo": "bar"}
+    assert 500 < await fake_redis.ttl("idempotency:test:key") <= 600  # type: ignore[attr-defined]
+
+
+@pytest.mark.asyncio
+async def test_redis_replace_writes_when_the_key_is_free(fake_redis: AsyncRedisClient) -> None:
+    """A reservation whose lease ran out leaves nothing behind; the result is still written."""
+    repo = RedisAsyncIdempotencyRepository(fake_redis)
+    service = IdempotencyDomainService()
+
+    await repo.replace(service.create_record("test", "key", {"foo": "bar"}))
+
+    stored = await repo.get("test", "key")
+    assert stored is not None
+    assert stored.result == {"foo": "bar"}
+
+
+@pytest.mark.asyncio
+async def test_redis_replace_exception(fake_redis: AsyncRedisClient) -> None:
+    """Test replace() when Redis raises an exception."""
+    repo = RedisAsyncIdempotencyRepository(fake_redis)
+    service = IdempotencyDomainService()
+    record = service.create_record("test", "key", {})
+    original_exc = Exception("Redis down")
+    with (
+        patch.object(fake_redis, "set", side_effect=original_exc),
+        pytest.raises(IdempotencyStorageError, match="Redis storage failure during replace") as exc_info,
+    ):
+        await repo.replace(record)
+    assert exc_info.value.original_error is original_exc
+
+
+@pytest.mark.asyncio
+async def test_redis_replace_expired_record(fake_redis: AsyncRedisClient) -> None:
+    """Test that replacing with an already expired record raises an error."""
+    repo = RedisAsyncIdempotencyRepository(fake_redis)
+
+    now = datetime.now(UTC)
+    record = IdempotencyRecord(
+        operation="test",
+        idempotency_key="key",
+        result={},
+        created_at=now - timedelta(minutes=20),
+        expires_at=now - timedelta(minutes=10),
+    )
+
+    with pytest.raises(IdempotencyValidationError, match="Cannot save already expired record"):
+        await repo.replace(record)