gcp-pubsub-idempotency 0.1.0

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.

gcp-pubsub-idempotency

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.

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:

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) · lock vs mutex · DECISIONS.md · 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.
• 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).
• Typed (py.typed, mypy --strict clean) and pure-stdlib at the core; each backend pulls its driver only via an extra.

---

Install

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 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.

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):

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/:

You're deploying to	Example
A Pub/Sub-triggered Cloud Function (gen2)	examples/cloud_function_datastore.py
A Cloud Run push service (Flask/gunicorn)	examples/cloud_run_worker.py
Local dev / unit tests (no GCP)	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.

@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:

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 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:

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.

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.

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). 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):

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

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):

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

License

MIT — see LICENSE.