Metadata-Version: 2.4
Name: gcp-pubsub-idempotency
Version: 0.1.0
Summary: Exactly-once processing for at-least-once Google Cloud Pub/Sub — a tiny Python idempotency gateway for Cloud Functions & Cloud Run push handlers. Leases, fencing tokens; Datastore/Redis/SQL backends.
Project-URL: Homepage, https://github.com/vilyam/gcp-pubsub-idempotency
Project-URL: Repository, https://github.com/vilyam/gcp-pubsub-idempotency
Project-URL: Changelog, https://github.com/vilyam/gcp-pubsub-idempotency/blob/main/CHANGELOG.md
Project-URL: Issues, https://github.com/vilyam/gcp-pubsub-idempotency/issues
Project-URL: Documentation, https://github.com/vilyam/gcp-pubsub-idempotency#readme
Author-email: Ivan Viliamov <vilyamm@gmail.com>
License-Expression: MIT
License-File: LICENSE
Keywords: at-least-once,cloud-functions,cloud-run,datastore,deduplication,distributed-lock,eventarc,exactly-once,fencing-token,firestore,gcp,google-cloud,idempotency,idempotent-consumer,pubsub,redis
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Distributed Computing
Classifier: Typing :: Typed
Requires-Python: >=3.10
Provides-Extra: datastore
Requires-Dist: google-api-core>=2.17; extra == 'datastore'
Requires-Dist: google-cloud-datastore>=2.20; extra == 'datastore'
Provides-Extra: redis
Requires-Dist: redis>=5.0; extra == 'redis'
Provides-Extra: sql
Provides-Extra: test
Requires-Dist: functions-framework>=3.0; extra == 'test'
Requires-Dist: google-api-core>=2.17; extra == 'test'
Requires-Dist: google-cloud-datastore>=2.20; extra == 'test'
Requires-Dist: mypy>=1.10; extra == 'test'
Requires-Dist: psycopg[binary]>=3.1; extra == 'test'
Requires-Dist: pytest>=8.0; extra == 'test'
Requires-Dist: redis>=5.0; extra == 'test'
Requires-Dist: werkzeug>=2.0; extra == 'test'
Description-Content-Type: text/markdown

# gcp-pubsub-idempotency

[![CI](https://github.com/vilyam/gcp-pubsub-idempotency/actions/workflows/ci.yml/badge.svg)](https://github.com/vilyam/gcp-pubsub-idempotency/actions/workflows/ci.yml)
[![PyPI](https://img.shields.io/pypi/v/gcp-pubsub-idempotency.svg)](https://pypi.org/project/gcp-pubsub-idempotency/)
[![Python versions](https://img.shields.io/pypi/pyversions/gcp-pubsub-idempotency.svg)](https://pypi.org/project/gcp-pubsub-idempotency/)
[![License: MIT](https://img.shields.io/github/license/vilyam/gcp-pubsub-idempotency)](https://github.com/vilyam/gcp-pubsub-idempotency/blob/main/LICENSE)
[![mypy: strict](https://img.shields.io/badge/mypy-strict-blue.svg)](https://mypy-lang.org/)

**Exactly-once processing for at-least-once Google Cloud Pub/Sub consumers, in Python.** A tiny
idempotency gateway that turns a push handler into an idempotent consumer for **Cloud Functions
/ Cloud Run** (push or Eventarc): wrap the business action and it runs **at most once** per key —
a redelivery or a concurrent duplicate skips the body, a crashed holder self-heals, and a
resurrected "zombie" can't clobber the work that took over from it.

```python
from gcp_pubsub_idempotency import lock, DatastoreLock

backend = DatastoreLock(project="my-project")

with lock("po-123", backend=backend):
    send_order_to_supplier(...)      # runs once; a duplicate skips the body; a crash retries
```

### Why it exists

Pub/Sub push is **at-least-once**, so Pub/Sub duplicate messages are the norm: the same message
is redelivered, the same order is published twice by overlapping producers, and two instances
race the same entity. The usual fixes rot — a hand-rolled Datastore lock **copy-pasted across
services and quietly drifting**, or a `try/finally: return "ok"` that "de-duplicates" by
**swallowing every failure** so nothing ever retries. This library replaces that pattern, in
**one dependency**, with a transactional `claim → done` record with a self-healing **lease**, a
**fencing token** against zombies, and an ack/nack contract a handler can't accidentally defeat.

- **For whom** — any Python service consuming GCP Pub/Sub that does a **non-idempotent side
  effect**: placing an order, charging a card, sending an email, calling an external API.
- **What it prevents** — duplicate side effects from redelivery/republish · failures **silently
  dropped** (the `return "ok"` bug) · **interleaved** writes to one entity · **zombie** writes
  after a lease lapses · N drifting copies of the same lock.
- **How** — one decorator (or one `with`), no cron, no schema, payload-agnostic:

```python
from gcp_pubsub_idempotency import idempotent, DatastoreLock

@idempotent(backend=DatastoreLock(project="my-project"))   # de-dupe + ack/nack, handled for you
@functions_framework.cloud_event
def handle_order(event):
    send_order_to_supplier(event)      # at most once; a failure NACKs and retries, never vanishes
```

Design notes: [error handling (EAFP)](#error-handling--idiomatic-python-eafp) ·
[`lock` vs `mutex`](#serialize-without-de-duplicating-mutex) · [DECISIONS.md](https://github.com/vilyam/gcp-pubsub-idempotency/blob/main/DECISIONS.md) ·
[COMPARISON.md](https://github.com/vilyam/gcp-pubsub-idempotency/blob/main/COMPARISON.md) (alternatives & prior art — Powertools, idemkit, MassTransit, …).

The library never looks at your payload — the idempotency key is whatever string you pass.
No cron, no background sweeper: correctness comes from a lazy expiry check on every read.

- **One small port** — a backend is three methods (`claim` / `complete` / `release`). Ships
  with Datastore (production), Redis, SQL, and in-memory.
- **Two verbs** — `lock()` **de-duplicates** (the body runs at most once per key); `mutex()`
  **serializes** (every message for a key runs, one at a time, none dropped). Same backend,
  same port — see [Serialize without de-duplicating](#serialize-without-de-duplicating-mutex).
- **Two surfaces** — the context managers above, and an `@idempotent` decorator that maps the
  outcome to a Cloud Functions ack/nack for you.
- **Correct under the hard cases** — crash-before-release, work-outruns-lease, zombie
  resurrection, concurrent reclaim races, store errors after the side effect. Each is pinned
  by a test against every backend (see [Scenarios](#scenarios-handled)).
- **Typed** (`py.typed`, `mypy --strict` clean) and pure-stdlib at the core; each backend
  pulls its driver only via an extra.

---

## Install

```bash
pip install "gcp-pubsub-idempotency[datastore]"      # production: Firestore-in-Datastore-mode
pip install "gcp-pubsub-idempotency[redis]"          # Redis / Memorystore
pip install "gcp-pubsub-idempotency[sql]"            # any DB-API 2.0 driver (bring your own)
pip install "gcp-pubsub-idempotency"                 # core only (InMemoryLock, for tests/local)
```

Published to **PyPI** as `gcp-pubsub-idempotency` starting with `v0.1.0` — see [RELEASING.md](https://github.com/vilyam/gcp-pubsub-idempotency/blob/main/RELEASING.md) for how versions are cut.

## Quickstart

### As a Cloud Functions push handler — the `@idempotent` decorator

`@idempotent` owns the entrypoint: it resolves the key from the event (the Pub/Sub
`messageId` by default), runs the handler under `lock`, and maps the outcome to ack/nack — so
an error can never be swallowed into a silent ack.

```python
import functions_framework
from gcp_pubsub_idempotency import idempotent, DatastoreLock

@idempotent(backend=DatastoreLock(project="my-project"))   # key defaults to the messageId
@functions_framework.cloud_event
def handle_order(event):
    send_order_to_supplier(event)
```

| Outcome | HTTP | Pub/Sub |
|---|---|---|
| handler returns / duplicate | 200 / 204 | **ack** |
| live holder elsewhere (contention) | 503 (a warning log with the `deliveryAttempt` count — never a crash traceback) | nack, retry later |
| handler raises | 500 | nack, retry |

### As a context manager — `lock()`

Use `lock()` when you want to own the HTTP response yourself, or you're not on
functions_framework (plain Flask / Cloud Run / a worker):

```python
from gcp_pubsub_idempotency import lock, Duplicate, InProgress

try:
    with lock(key, backend=backend):
        do_irreversible_work()
except Duplicate:
    ...            # already done -> ack
except InProgress:
    ...            # a live holder owns it -> nack, retry later
```

### Where to run it

Three complete, copy-pasteable examples in [`examples/`](https://github.com/vilyam/gcp-pubsub-idempotency/tree/main/examples):

| You're deploying to | Example |
|---|---|
| A Pub/Sub-triggered **Cloud Function** (gen2) | [`examples/cloud_function_datastore.py`](https://github.com/vilyam/gcp-pubsub-idempotency/blob/main/examples/cloud_function_datastore.py) |
| A **Cloud Run** push service (Flask/gunicorn) | [`examples/cloud_run_worker.py`](https://github.com/vilyam/gcp-pubsub-idempotency/blob/main/examples/cloud_run_worker.py) |
| **Local dev / unit tests** (no GCP) | [`examples/local_dev.py`](https://github.com/vilyam/gcp-pubsub-idempotency/blob/main/examples/local_dev.py) |

## Choosing the key

The key is opaque — the library never parses it. What you choose decides *what* is deduped:

- **`messageId` (default)** guards **transport redelivery** only: the same publish delivered
  again. A producer *republish* of the same logical event gets a **new** messageId and is
  **not** caught.
- **A business id** (e.g. `key=lambda e: f"orders:{po_id}"`) dedupes the *entity*: "this order,
  once, however many times it's published". Namespace it (here `orders:`) if several
  subscriptions share one backend, or give each its own `kind`/`prefix`/`table`.

```python
@idempotent(backend=backend, key=lambda e: "orders:" + e.data["message"]["attributes"]["po_id"])
@functions_framework.cloud_event
def handle_order(event): ...
```

> A completed key is suppressed as a **`Duplicate` (ack)** for the whole `ttl`. So a business
> key means every message for that entity within the ttl is treated as the same occurrence —
> which is what you want for "one order once", but *not* if distinct events for one entity
> must each run. Pick the key at the grain you actually want deduped — or reach for `mutex()`
> (next), which serializes them instead of suppressing them.

## Serialize without de-duplicating: `mutex()`

> **Which do I need?** If a *repeat must disappear* — the same message redelivered, or the same
> operation republished (e.g. **one-shot order creation**, where a second send would place a
> second order) — that is **de-duplication → use `lock`/`@idempotent`**, keyed on the business
> entity so a republish is caught too. Reach for **`mutex` only** when *distinct, non-idempotent*
> messages for one entity must **each** run but never overlap. Guarding a one-shot create with
> `mutex` is a bug: it wouldn't stop the double order. **Repeat vanishes → `lock`. Repeats all
> run but not at once → `mutex`.**

`mutex` fits the second case: distinct messages for one entity must **each** run, just never at
the same time. Example: successive **status updates** for one order that must each be applied but
not interleave (two racing updates would corrupt the read-modify-write). The transport can *also*
redeliver the *same* update — a duplicate to drop. That is **two jobs on two keys**, so use **two
verbs**:

```python
import functions_framework
from gcp_pubsub_idempotency import idempotent, mutex, DatastoreLock

backend = DatastoreLock(project="my-project")

@idempotent(backend=backend)                        # verb 1 — dedup on the messageId (transport)
@functions_framework.cloud_event
def handle_po_update(event):
    po_id = event.data["message"]["attributes"]["po_id"]
    with mutex(f"po:{po_id}", backend=backend):     # verb 2 — serialize on the PO (per-entity)
        apply_status_update(po_id, event)           # one PO at a time; every update still runs
```

- `@idempotent` (keyed on the **messageId**) collapses a redelivery of the *same* message to a
  single run — transport de-duplication, with ack/nack handled for you.
- `mutex` (keyed on the **PO**) guarantees that, across every instance, at most one delivery
  touches that PO at a time — mutual exclusion — while letting *different* messages through.

**Why `lock()` is wrong for the PO, and `mutex()` right.** `lock()` marks its key **DONE** on
success. The first status update would complete the PO key, and every later update for that PO
would then be dropped as a `Duplicate` for the whole ttl — the updates would silently vanish.
`mutex()` is the sibling that **always `release()`s and never `complete()`s**: the key is never
DONE, so after each update it is immediately reclaimable and the next update runs — *serialized,
not suppressed*.

| | `lock(key)` | `mutex(key)` |
|---|---|---|
| On success | `complete()` → **DONE** | `release()` → **reclaimable** |
| A *later* message on the same key | `Duplicate` → body skipped | **runs** (once the holder releases) |
| A *concurrent* message on the same key | `InProgress` → nack/retry | `InProgress` → nack/retry |
| Reach for it when you mean | "this message / entity **once**" | "these messages **one at a time**" |

Both raise `InProgress` on live contention, so under `@idempotent` a contended delivery becomes
a quiet 503 nack and Pub/Sub retries it later — nothing is lost. A `mutex` used on a key that
some caller already completed with `lock()` (a DONE record) also raises `InProgress` rather than
running over a finished key.

> **Mutual exclusion, not ordering.** `mutex` guarantees no two deliveries for one key run at
> once; it does **not** guarantee they run in *publish order* — a nacked-then-redelivered update
> can land after a later one. If you need strict ordering, add a Pub/Sub **ordering key**; the
> mutex still gives you the no-interleaving guarantee on top of it.

## The invariants that keep it correct

Three sizing rules. The defaults are safe for Cloud Run; deviate deliberately.

1. **`lease` must exceed the maximum runtime of the guarded work.** The lease is the
   mutual-exclusion window. If the work outruns it, the claim becomes reclaimable and a
   concurrent redelivery could run the work twice. Default: `DEFAULT_LEASE` = **15 min**.
2. **`lease + ttl` must exceed the maximum lifetime of a holder process** (on Cloud Run, the
   request timeout — hard-capped at 3600 s). This is what keeps the fencing token meaningful:
   the record must outlive any holder that could still be alive, so a reclaim *bumps* the
   fence rather than a fresh record *resetting* it. Default: 15 min + **1 h** ttl = 75 min,
   which clears the 60 min Cloud Run cap. (Datastore's GC gives ~24 h of extra slack here;
   Redis reaps exactly at the retention window, so the rule is tightest for Redis.)
3. **Configure a Pub/Sub `dead_letter_policy` — and size `max_delivery_attempts` above the
   worst-case contention retries.** A *poison* message (one a handler can never process) is
   nacked and redelivered until the DLQ catches it — see
   [poison handling](#poison--terminal-failures) below. There is no built-in "terminal"
   outcome; poison handling is the subscription's job plus a `try/except` in your handler.
   **The sizing rule exists because Pub/Sub's dead-letter counter cannot tell contention from
   poison**: every nack — including the healthy 503 a delivery gets while another holder owns
   its key — increments `deliveryAttempt` identically. With the default 15-min lease and
   Pub/Sub's default retry backoff (10 s → 600 s), a message stuck behind a full lease can burn
   6–7 attempts before the key frees — above the `max_delivery_attempts` **minimum of 5** — so
   set it well clear of that (≥ 10 with the defaults; recompute if you lengthen `lease` or
   shorten the backoff), and alert on the decorator's contention warning, which includes the
   message's `deliveryAttempt` whenever the subscription has a dead-letter policy.

## Exactly-once outcome across a crash: key your sink

`lock` gives exactly-once **execution** under all producer chaos, across every instance. The
one thing it can't cover alone is a hard process crash (SIGKILL/OOM) in the window **after**
the send but **before** the DONE record is written: the send happened, the key isn't recorded,
and the reclaim re-runs it. No lock can make a network send atomic with a local commit across
a crash (Two Generals) — so *execution* is at-least-once there.

Close it at the sink — make the send **idempotent on the key you already hold**, so a re-send
is a no-op:

```python
with lock(order_key, backend=backend):
    # REST: carry the key so the vendor de-duplicates
    requests.post(url, json=order, headers={"Idempotency-Key": order_key})
    # SFTP: a deterministic remote filename — a re-drop overwrites, never duplicates
    sftp.put(local, f"/incoming/{order_key}.xml")
```

Now a crash-window re-send lands on the same key and the supplier holds the order exactly once
— **effectively-once outcome**. The multi-instance crash storm in `functional/` proves it:
`os._exit` crashes across a 4-instance fleet, yet every order reaches the (idempotent)
supplier once. For a sink with no dedup at all (plain email), at-least-once is the floor —
prefer a dedup-capable channel for irreversible sends.

## Backends

All implement the same three-method `LockBackend` port; swap freely.

```python
from gcp_pubsub_idempotency import DatastoreLock, RedisLock, SqlLock, InMemoryLock
```

| Backend | Notes | Extra |
|---|---|---|
| `DatastoreLock` | Firestore-in-Datastore-mode; **the production default**. Serializable transactional claim; a TTL policy on `expires_at` reclaims storage. | `[datastore]` |
| `RedisLock` | single-key Lua compare-and-set; server-side clock (no cross-client skew); Redis-Cluster-safe (one key per record). | `[redis]` |
| `SqlLock` | any DB-API 2.0 driver; epoch-ms timestamps (no datetime adapters). `SELECT … FOR UPDATE` + PK tiebreak. Pass a pooled `connect()`. | `[sql]` + driver |
| `InMemoryLock` | pure stdlib; tests and single-process use. Kept byte-for-byte aligned with `DatastoreLock`. | — |

Every backend gives you: an **atomic claim**, **fencing tokens** (a resurrected holder can't
clobber a newer claimant), a **lease** (a crashed holder self-heals once its lease lapses — no
cron), and a **ttl** (how long a completed key suppresses duplicates). Correctness is enforced
by a lazy expiry check on every read, never by a background sweep.

## Poison / terminal failures

There is **no built-in "terminal" outcome** — kept out by design, so the library never has to
guess whether an error is retryable. Two mechanisms cover poison:

- **Handle it in your handler.** Catch the fatal error, do whatever you need (log, raise an
  ticket, alert), and **return normally** — the key completes DONE, the message is acked, and
  it is never retried. Raising instead would nack and retry forever.
- **Back it with a `dead_letter_policy`.** For truly unexpected poison you don't catch, the
  subscription's `max_delivery_attempts` routes it to a dead-letter topic after N nacks.

## Error handling — idiomatic Python (EAFP)

This is a Python library, so it handles errors the Python way: **EAFP** — try the operation,
`try/except` where you can *meaningfully* handle the failure, and let everything else propagate —
plus **context managers** for guaranteed cleanup. (Python's glossary calls EAFP the "clean and
fast" Python style; it is also safer than look-before-you-leap under concurrency.) It is **not**
an Erlang "let it crash" system — that is a runtime architecture built on process isolation and
supervision trees a library doesn't have. The only real parallel is coarse and lives in the
*runtime*, not here: an unhandled error becomes a 5xx and **Pub/Sub redelivers** — and we get
there by ordinary exception propagation, not by catching.

The whole policy is one EAFP question at each point: **can the library meaningfully handle this
error here?**

- **No → let it propagate.** The business work's errors, a broken key function, an async misuse,
  live contention (a quiet 503) — the library can't fix these, so they reach the runtime
  boundary → nack → Pub/Sub retries. (The `finally: return 'ok'` force-ack anti-pattern is the
  canonical violation: it catches errors it cannot handle and silently hides them.)
- **Yes → catch it.** At three points in the delivery path, all where re-raising would be *worse than the error*:

| Point | Re-raising would… | So it… |
|---|---|---|
| `complete()` fails **after** the send happened | nack → redeliver → **repeat the send** | logs, **acks** |
| fenced out after the send (work outran its lease **and** a redelivery reclaimed the key) | same | logs, acks |
| `on_duplicate` hook raises (delivery already DONE) | nack → **loop a settled message to the DLQ** | logs, continues |

**Observation hooks never change control flow — the same choice the standard library makes.**
Python's `logging` never lets a handler's emit-time exception escape a `log.*()` call: the
handler's `emit()` catches it and routes it to `Handler.handleError`, which by default prints a
traceback to stderr (and, with `raiseExceptions` set to `False`, stays silent) but never
re-raises — because, as its docstring puts it, "most users will not care about errors in the
logging system." `on_duplicate` follows suit: it is an injected side effect (a warning, a
metric), and a hook that raises is logged and ignored. If a failure *should* nack, that is **work**, and it belongs in the
handler body — where it propagates correctly.

```python
from gcp_pubsub_idempotency import idempotent, DatastoreLock

@idempotent(
    backend=DatastoreLock(project="my-project"),
    key=lambda ev: "orders:" + ev.data["message"]["attributes"]["po_id"],
    on_duplicate=lambda key: log.warning("duplicate delivery for %s; skipping", key),
)
@functions_framework.cloud_event
def handle_order(event): ...
```

## How it works — the record state machine

Every key maps to at most one record with two stored states, **CLAIMED** and **DONE**, guarded
by two clocks: the **lease** (mutual exclusion while work is in flight) and the **ttl**
(duplicate suppression after success). "Reclaimable" is *not* stored — it's what `claim()` sees
when a lease or ttl has lapsed, evaluated lazily on every read.

```
                       complete()  [fence matches - even after the lease lapsed]
   claim()             ─────────────────────────────────────────────►
  ABSENT ───────────► CLAIMED ──────────────────────────────────────► DONE
     ▲                 │   ▲                                          │  │
     │                 │   │  claim() while lease live                │  │ claim() while
     │                 │   │   → HeldByOther  (no transition)         │  │  within ttl
     │       release() │   │                                          │  │  → AlreadyDone
     │       [fence    │   │  claim() after lease_expiry              │  │   (no transition)
     │        matches] ▼   │   → Won, fence+1  (reclaim / self-heal)  │  │
     │              CLAIMED*│                                         │  │ claim() after
     │        (lease_expiry └─────────────────────────────────────────┘  │  expires_at (ttl)
     │         set to now; record kept,                                  │ → Won, fence+1
     │         fence ADVANCED: the released                             ◄┘   (reclaim)
     │         claim is dead; reclaimable)
     │
     └── the record is physically deleted only AFTER expires_at (Redis PEXPIRE /
         Datastore TTL policy / an explicit purge_expired()). A claim() after that
         starts a fresh record at fence = 1.
```

**Transition table** (what every backend guarantees):

| From | Call | Guard | To | Returns | Fence |
|---|---|---|---|---|---|
| ABSENT | `claim` | — | CLAIMED | `Won` | = 1 |
| CLAIMED | `claim` | `now < lease_expiry` | CLAIMED | `HeldByOther` | unchanged |
| CLAIMED | `claim` | `now ≥ lease_expiry` (holder crashed) | CLAIMED | `Won` | **+1** |
| CLAIMED | `complete` | fence matches — no lease guard | DONE | `True` | unchanged |
| CLAIMED | `complete` | fence stale (reclaimed or released) | (no change) | `False` | unchanged |
| CLAIMED | `release` | fence matches | CLAIMED* (`lease_expiry=now`) | — | **+1** |
| DONE | `claim` | `now < expires_at` | DONE | `AlreadyDone` | unchanged |
| DONE | `claim` | `now ≥ expires_at` (ttl elapsed) | CLAIMED | `Won` | **+1** |
| DONE | `complete`/`release` | (idempotent) | DONE | `True`/no-op | unchanged |

**The fencing invariant.** `fencing_id` only ever rises while a record exists, and
`complete`/`release` require an *exact* fence match. So once a crashed holder's lease is
reclaimed (fence+1), the original — should it wake up — can no longer complete or release: its
fence is stale (Kleppmann fencing). This holds as long as the record physically outlives any
holder, which is what invariant #2 above guarantees.

**Why `complete` has no lease guard — ownership is the fence, not the clock.** Until someone
reclaims a lapsed key (bumping the fence), the lapsed holder is still the only party that ran
the work, so its late DONE **must** commit: refusing it would leave the key reclaimable *after
the side effect fired*, and the next delivery would run the irreversible action a second time.
The clock's job is deciding when a key may be **taken over** (`claim`); the fence's job is
deciding who may **write** (`complete`/`release`). `release` advances the fence for the same
reason from the other side: a released claim is abandoned, so it must never be completable
afterwards.

**Where `lock()` and `mutex()` travel on this diagram.** `lock()` walks the full path
`ABSENT → CLAIMED → DONE` (a duplicate is then suppressed for the ttl). `mutex()` walks only the
left half — `ABSENT → CLAIMED → (release) → reclaimable` — and cycles there forever: it calls
`claim()` then `release()` and **never** `complete()`, so a mutex key never enters DONE and has
no duplicate-suppression window. That is exactly what lets successive messages for one entity
each run (see [Serialize without de-duplicating](#serialize-without-de-duplicating-mutex)). The
fence still rises on every reclaim across the cycle, so ABA is closed for `mutex` too.

## Scenarios handled

Each row is a real at-least-once hazard and how the gateway resolves it, with the test that
pins it (`tests/` and `functional/`).

| Scenario | Behaviour | Proof |
|---|---|---|
| **Transport redelivery** — the same message again | 2nd delivery finds DONE → `Duplicate` → **ack**, body skipped | `test_lock_marks_done_…`, functional `test_ack_processes_once` |
| **Concurrent duplicate** — same key racing in parallel | exactly one wins; the rest get `HeldByOther` → `InProgress` → nack; on retry they see DONE → ack | `test_lock_concurrent_same_key_admits_one_body`, functional `test_concurrent_same_message_processed_once` |
| **Double subscription** — two subs, one topic, same `messageId` | identical key → deduped across both deliveries | functional `test_double_subscription_processes_once` |
| **Crashed holder** — SIGKILL mid-work, `release` never runs | CLAIMED record stays; after `lease_expiry` the next delivery reclaims (fence+1). No cron. | contract `test_crashed_holder_self_heals_after_lease_expiry` |
| **Zombie resurrection** — the crashed holder wakes after a reclaim | its fence is stale → `complete`/`release` refused | contract `test_zombie_cannot_complete/release_over_a_newer_claim`, `test_zombie_fenced_out_when_lease_ttl_exceeds_holder_life` (redis, server clock) |
| **Work outran the lease, key not reclaimed** — body slower than `lease`, no concurrent delivery | the fence still matches, so the late `complete` **commits DONE** — a redelivery is a `Duplicate`, nothing repeats | `test_lock_records_done_when_work_outruns_an_unreclaimed_lease`, contract `test_complete_commits_on_fence_match_even_after_the_lease_lapsed` |
| **Work outran the lease AND a redelivery reclaimed the key** | `complete` fenced out; the CM does **not** nack (that would replay an irreversible send) — it **acks** and logs an alertable error; the reclaimer may re-run the work, which only a longer `lease` prevents | `test_lock_acks_and_logs_when_fenced_out_by_a_reclaim` |
| **Store error after the side effect** — `complete` raises post-work | **ack** + loud log; never nack a side effect that already happened | `test_lock_acks_when_complete_raises` |
| **Transient work failure** | body raises → `release` (reclaimable) → propagates → **nack** → redelivered → succeeds once | functional `test_nack_retries_then_succeeds_once` |
| **Poison / permanent failure** | keeps retrying by default; catch it in the handler and return to ack (no built-in "terminal" — by design) | — (documented above) |
| **Business republish** — same PO, *new* `messageId` | the default `messageId` key does **not** dedup this; pass a business `key=` | `test_idempotent_custom_key_override` |
| **Per-entity serialization** — *different* messages for one entity that must each run, never concurrently | guard the entity with `mutex(entity_key)`: release-always, so each runs in turn (never a `Duplicate`); a concurrent one gets `InProgress` → nack → retry. Mutual exclusion, not ordering. | `tests/test_mutex.py`, contract `test_mutex_cycle_reclaims_without_ever_completing` (every backend), functional `test_mutex_serializes_different_messages_for_one_po` |
| **TTL expiry** — a genuinely new occurrence long after success | past `expires_at` the DONE record is reclaimable, so the new occurrence runs | `test_lock_ttl_expiry_makes_key_reclaimable` |
| **Unkeyed message** — no key resolvable | fail **closed**: refuse to run unguarded (nack), unless `allow_unguarded=True` | `test_idempotent_unkeyed_fails_closed_by_default` |
| **Lost commit response** (Datastore) — a retried claim after a dropped ack | re-affirms its own claim via a per-call token, never self-locks or double-bumps | `test_claim_is_retry_safe_via_claim_token`, `test_claim_reaffirm_under_retry_on_the_reclaim_path` |
| **Clock skew across instances** (Redis) | expiry is judged against Redis's own `TIME`, not any client wall clock | `test_redis_server_clock.py` |

## Writing a backend

Implement three methods (inherit `LockBackend` for the definition-time safety check, or conform
structurally):

```python
from gcp_pubsub_idempotency import LockBackend, Won, AlreadyDone, HeldByOther, Claim

class MyLock(LockBackend):
    def claim(self, key, lease): ...     # -> Won | AlreadyDone | HeldByOther  (atomic)
    def complete(self, claim): ...       # -> bool  (CLAIMED -> DONE on fence match — no lease guard)
    def release(self, claim): ...        # -> None  (expire the lease AND advance the fence; NEVER delete)
```

Contract: `claim` is atomic (create-if-absent / reclaim-if-expired) and checks expiry *at read
time*; `complete`/`release` re-validate `Claim.fencing_id` before mutating, and the fence is
`complete`'s **only** ownership test — a lapsed-but-unreclaimed claim must still commit (see
the state machine above); `release` expires the lease in place **and advances the fence**
(deleting would reset the fence → ABA; leaving the fence would let a released claim complete).
The same contract suite (`tests/test_backend_contract.py`) runs against every backend, so a
new one gets the full matrix for free.

## Testing your handler

Set `GCP_PUBSUB_IDEMPOTENCY_DISABLED=1` (or `true`/`yes`/`on`) and `@idempotent` becomes a
warning pass-through: the handler runs unguarded and the backend is never built, so unit tests
exercise the handler body with no backend at all. Never set it in production — a duplicate
delivery then re-runs the side effect. For tests that should exercise the gateway itself, use
`InMemoryLock` instead.

## Develop

```bash
pip install -e '.[test]'
pytest                          # unit + real-emulator/server matrix (Docker auto-starts what's missing)
mypy --strict src/gcp_pubsub_idempotency
```

`tests/conftest.py` starts Redis, PostgreSQL and the Datastore emulator in Docker for the
session (or reuses `REDIS_HOST` / `POSTGRES_DSN` / `DATASTORE_EMULATOR_HOST` if you set them),
so a plain `pytest` runs **every backend with no skips**. The end-to-end harness (real Pub/Sub
+ Datastore emulators, real push subscriptions):

```bash
cd functional && docker compose up --build --abort-on-container-exit --exit-code-from runner
```

## License

MIT — see [LICENSE](https://github.com/vilyam/gcp-pubsub-idempotency/blob/main/LICENSE).
