Fine tuned the RedisBackend to fully support distributed systems (#16)

Author: Attakay78

docs/adapters/custom.md

@@ -219,6 +219,296 @@ Everything works from here — decorators, CLI, dashboard, audit log — with SQ
 
 ---
 
+## Distributed support
+
+The six abstract methods give you persistence. To unlock full distributed behaviour — live dashboard updates, cross-instance global maintenance sync, and webhook deduplication — implement three additional optional methods. Each one has a default that works correctly for single-instance deployments, so you can add them incrementally.
+
+---
+
+### The three distributed methods
+
+```python
+from collections.abc import AsyncIterator
+from shield.core.backends.base import ShieldBackend
+
+class MyDistributedBackend(ShieldBackend):
+
+    async def subscribe(self) -> AsyncIterator[RouteState]:
+        """Stream every per-route state change as it happens.
+
+        Used by the dashboard SSE endpoint to push live updates to browsers
+        without polling.  Yield a RouteState every time set_state() is called
+        by any instance.  If your store does not support pub/sub, leave this
+        unimplemented — the dashboard falls back to polling list_states()
+        every few seconds automatically.
+        """
+        ...
+
+    async def subscribe_global_config(self) -> AsyncIterator[None]:
+        """Yield None whenever any instance writes a new global maintenance config.
+
+        ShieldEngine keeps GlobalMaintenanceConfig in an in-process cache to
+        avoid a storage round-trip on every request.  When another instance
+        enables or disables global maintenance, it writes to the shared store
+        and your implementation of this method should yield a signal so the
+        engine drops its local cache and re-fetches on the next request.
+
+        Yield None for each change signal — the content does not matter,
+        only the arrival of the message.
+        """
+        ...
+
+    async def try_claim_webhook_dispatch(
+        self, dedup_key: str, ttl_seconds: int = 60
+    ) -> bool:
+        """Claim exclusive right to fire webhooks for one event.
+
+        When a scheduled maintenance window activates, every instance
+        independently calls set_maintenance() and would each fire all
+        registered webhooks — producing N deliveries for one event.
+
+        Before firing, ShieldEngine calls this method with a deterministic
+        key derived from event + path + serialised RouteState (identical
+        across all instances for the same event).  The first instance to
+        win the claim fires; all others return False and skip.
+
+        Use an atomic conditional write — "set this key only if it does not
+        already exist" — and return True if you wrote it, False if it was
+        already present.  Set the key to expire after ttl_seconds so that a
+        crashed instance does not permanently suppress re-delivery.
+
+        Return True unconditionally if your store does not support atomic
+        conditional writes — webhooks will be over-delivered rather than
+        silently dropped.
+        """
+        ...
+```
+
+All three raise `NotImplementedError` by default. The engine handles each gracefully:
+
+| Method | What happens if not implemented |
+|---|---|
+| `subscribe()` | Dashboard SSE falls back to polling `list_states()` every few seconds |
+| `subscribe_global_config()` | Global maintenance cache is per-process; stale until the process writes its own update |
+| `try_claim_webhook_dispatch()` | Always returns `True` — every instance fires webhooks (over-delivery) |
+
+---
+
+### PostgreSQL example
+
+PostgreSQL's `LISTEN` / `NOTIFY` is a built-in pub/sub mechanism that works across connections and processes — no extra broker needed.
+
+```python
+"""PostgreSQL distributed backend using asyncpg + LISTEN/NOTIFY.
+
+pip install asyncpg
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from collections.abc import AsyncIterator
+from datetime import UTC, datetime
+
+import asyncpg
+
+from shield.core.backends.base import ShieldBackend
+from shield.core.models import AuditEntry, GlobalMaintenanceConfig, RouteState, RouteStatus
+
+
+class PostgresBackend(ShieldBackend):
+
+    def __init__(self, dsn: str) -> None:
+        self._dsn = dsn
+        self._pool: asyncpg.Pool | None = None
+
+    # ------------------------------------------------------------------
+    # Lifecycle
+    # ------------------------------------------------------------------
+
+    async def startup(self) -> None:
+        self._pool = await asyncpg.create_pool(self._dsn)
+        async with self._pool.acquire() as conn:
+            await conn.execute("""
+                CREATE TABLE IF NOT EXISTS shield_states (
+                    path TEXT PRIMARY KEY,
+                    state_json TEXT NOT NULL
+                );
+                CREATE TABLE IF NOT EXISTS shield_audit (
+                    id TEXT PRIMARY KEY,
+                    ts TIMESTAMPTZ NOT NULL,
+                    path TEXT NOT NULL,
+                    entry_json TEXT NOT NULL
+                );
+                CREATE TABLE IF NOT EXISTS shield_webhook_dedup (
+                    dedup_key TEXT PRIMARY KEY,
+                    claimed_at TIMESTAMPTZ NOT NULL DEFAULT now()
+                );
+            """)
+
+    async def shutdown(self) -> None:
+        if self._pool:
+            await self._pool.close()
+
+    # ------------------------------------------------------------------
+    # Core interface
+    # ------------------------------------------------------------------
+
+    async def get_state(self, path: str) -> RouteState:
+        async with self._pool.acquire() as conn:
+            row = await conn.fetchrow(
+                "SELECT state_json FROM shield_states WHERE path = $1", path
+            )
+        if row is None:
+            raise KeyError(path)
+        return RouteState.model_validate_json(row["state_json"])
+
+    async def set_state(self, path: str, state: RouteState) -> None:
+        payload = state.model_dump_json()
+        async with self._pool.acquire() as conn:
+            await conn.execute(
+                """
+                INSERT INTO shield_states (path, state_json) VALUES ($1, $2)
+                ON CONFLICT (path) DO UPDATE SET state_json = EXCLUDED.state_json
+                """,
+                path, payload,
+            )
+            # Notify all listening instances of the per-route state change.
+            await conn.execute("SELECT pg_notify('shield_changes', $1)", payload)
+
+    async def delete_state(self, path: str) -> None:
+        async with self._pool.acquire() as conn:
+            await conn.execute(
+                "DELETE FROM shield_states WHERE path = $1", path
+            )
+
+    async def list_states(self) -> list[RouteState]:
+        async with self._pool.acquire() as conn:
+            rows = await conn.fetch("SELECT state_json FROM shield_states")
+        return [RouteState.model_validate_json(r["state_json"]) for r in rows]
+
+    async def write_audit(self, entry: AuditEntry) -> None:
+        async with self._pool.acquire() as conn:
+            await conn.execute(
+                """
+                INSERT INTO shield_audit (id, ts, path, entry_json)
+                VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING
+                """,
+                entry.id, entry.timestamp, entry.path, entry.model_dump_json(),
+            )
+
+    async def get_audit_log(
+        self, path: str | None = None, limit: int = 100
+    ) -> list[AuditEntry]:
+        async with self._pool.acquire() as conn:
+            if path:
+                rows = await conn.fetch(
+                    "SELECT entry_json FROM shield_audit WHERE path = $1"
+                    " ORDER BY ts DESC LIMIT $2",
+                    path, limit,
+                )
+            else:
+                rows = await conn.fetch(
+                    "SELECT entry_json FROM shield_audit ORDER BY ts DESC LIMIT $1",
+                    limit,
+                )
+        return [AuditEntry.model_validate_json(r["entry_json"]) for r in rows]
+
+    # ------------------------------------------------------------------
+    # Distributed: per-route live updates (dashboard SSE)
+    # ------------------------------------------------------------------
+
+    async def subscribe(self) -> AsyncIterator[RouteState]:
+        """Stream RouteState changes via PostgreSQL LISTEN/NOTIFY."""
+        queue: asyncio.Queue[RouteState] = asyncio.Queue()
+
+        def _on_notify(conn, pid, channel, payload):
+            try:
+                state = RouteState.model_validate_json(payload)
+                queue.put_nowait(state)
+            except Exception:
+                pass
+
+        async with self._pool.acquire() as conn:
+            await conn.add_listener("shield_changes", _on_notify)
+            try:
+                while True:
+                    yield await queue.get()
+            finally:
+                await conn.remove_listener("shield_changes", _on_notify)
+
+    # ------------------------------------------------------------------
+    # Distributed: global maintenance cache invalidation
+    # ------------------------------------------------------------------
+
+    async def set_global_config(self, config: GlobalMaintenanceConfig) -> None:
+        """Persist config and notify all instances to drop their cache."""
+        await super().set_global_config(config)
+        async with self._pool.acquire() as conn:
+            # Empty string payload — only the arrival of the notification
+            # matters, not its content.
+            await conn.execute(
+                "SELECT pg_notify('shield_global_invalidate', '1')"
+            )
+
+    async def subscribe_global_config(self) -> AsyncIterator[None]:
+        """Yield None on each global config change via LISTEN/NOTIFY."""
+        queue: asyncio.Queue[None] = asyncio.Queue()
+
+        def _on_notify(conn, pid, channel, payload):
+            queue.put_nowait(None)
+
+        async with self._pool.acquire() as conn:
+            await conn.add_listener("shield_global_invalidate", _on_notify)
+            try:
+                while True:
+                    yield await queue.get()
+            finally:
+                await conn.remove_listener("shield_global_invalidate", _on_notify)
+
+    # ------------------------------------------------------------------
+    # Distributed: webhook deduplication
+    # ------------------------------------------------------------------
+
+    async def try_claim_webhook_dispatch(
+        self, dedup_key: str, ttl_seconds: int = 60
+    ) -> bool:
+        """Claim webhook dispatch rights using an INSERT ... ON CONFLICT DO NOTHING.
+
+        PostgreSQL's INSERT with ON CONFLICT is atomic — only one instance
+        succeeds.  A background cleanup query removes expired rows so the
+        table does not grow indefinitely.
+        """
+        async with self._pool.acquire() as conn:
+            # Purge rows older than ttl_seconds first (best-effort cleanup).
+            await conn.execute(
+                "DELETE FROM shield_webhook_dedup"
+                " WHERE claimed_at < now() - ($1 || ' seconds')::interval",
+                str(ttl_seconds),
+            )
+            result = await conn.execute(
+                "INSERT INTO shield_webhook_dedup (dedup_key)"
+                " VALUES ($1) ON CONFLICT DO NOTHING",
+                dedup_key,
+            )
+        # asyncpg returns "INSERT 0 1" when a row was inserted,
+        # "INSERT 0 0" when ON CONFLICT suppressed the insert.
+        return result == "INSERT 0 1"
+```
+
+---
+
+### What your store needs to support each method
+
+| Method | Minimum capability required |
+|---|---|
+| `subscribe()` | Pub/sub or change-data-capture (PostgreSQL `LISTEN/NOTIFY`, MySQL binlog, Kafka, NATS) |
+| `subscribe_global_config()` | Same pub/sub as above — just a separate channel/topic |
+| `try_claim_webhook_dispatch()` | Atomic conditional write — "insert only if absent" (SQL `INSERT … ON CONFLICT DO NOTHING`, DynamoDB `PutItem` with `attribute_not_exists`, etcd transactions, Zookeeper ephemeral nodes, Memcached `add`) |
+
+---
+
 ## Building a framework adapter
 
 If you want to support a framework other than FastAPI, the pattern is:

docs/changelog.md

@@ -28,10 +28,27 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - `_ShieldCallable` extended with an optional `signature=` override and updated `__call__` to forward the `response` kwarg to `dep_raise` when present — fully backward compatible with all existing decorators
 - `examples/fastapi/dependency_injection.py` updated to include `@deprecated` as a `Depends()` example and a clear explanation of why `@force_active` cannot be used as a dependency
 
+#### Webhook Deduplication
+- `ShieldBackend` ABC gains `try_claim_webhook_dispatch(dedup_key, ttl_seconds)` — returns `True` if this instance should fire webhooks, `False` if another instance already claimed the right for this event. Default implementation always returns `True` (single-instance backends never need dedup).
+- `RedisBackend` overrides `try_claim_webhook_dispatch()` using `SET NX EX` — the first instance to win the atomic write fires webhooks; all others skip. Fails open: a Redis error returns `True` so webhooks are over-delivered rather than silently dropped.
+- `ShieldEngine._fire_webhooks` refactored — now schedules a single `_dispatch_webhooks` task instead of one task per URL. The task computes a deterministic SHA-256 dedup key from `event + path + serialised RouteState`, claims dispatch rights via the backend, then fans out to individual webhook URLs only if the claim succeeds.
+- Dedup key is deterministic across instances: because the scheduler produces an identical `RouteState` on all instances for the same window activation, the key is the same fleet-wide and only one instance wins.
+- TTL on the dedup key defaults to 60 seconds — if the winning instance crashes mid-dispatch the key expires and re-delivery is possible on the next activation cycle.
+
+#### Distributed Global Maintenance
+- `RedisBackend` now publishes a lightweight invalidation signal to `shield:global_invalidate` whenever `set_global_config()` is called — any other instance subscribed to this channel immediately drops its in-process `GlobalMaintenanceConfig` cache
+- `ShieldBackend` ABC gains `subscribe_global_config()` — an async generator that yields `None` on each remote global config change; default implementation raises `NotImplementedError` (no-op for `MemoryBackend` and `FileBackend`)
+- `ShieldEngine.start()` — starts a background `asyncio.Task` that listens for global config invalidation signals and calls `_invalidate_global_config_cache()` on each one; idempotent, safe to call multiple times
+- `ShieldEngine.stop()` — cancels and awaits the listener task; called automatically by `__aexit__`
+- `ShieldEngine.__aenter__` / `__aexit__` updated to call `start()` / `stop()` so CLI scripts using `async with ShieldEngine(...)` get distributed invalidation automatically
+- `ShieldRouter.register_shield_routes()` calls `engine.start()` at application startup so FastAPI apps also start the listener without requiring the context manager
+- For `MemoryBackend` / `FileBackend` the new code path is a transparent no-op — `NotImplementedError` is caught, the task exits immediately, and single-instance cache behaviour is unchanged
+
 #### Documentation & Communication
 - Early Access notice added to README and docs homepage — communicates that the library is fully functional and actively developed, and invites community feedback via GitHub Issues
 - Webhooks and Custom Responses added to the Key Features table in the docs homepage
 - Key Features section added to `README.md`
+- New guide: **Distributed Deployments** (`docs/guides/distributed.md`) — covers backend capability matrix, the request lifecycle across instances, global maintenance cache invalidation architecture, scheduler behaviour and webhook deduplication in multi-instance setups, OpenAPI schema staleness, the fail-open guarantee, and a production checklist. Explains why `FileBackend` intentionally does not support cross-instance sync and when to use each backend.
 
 ### Changed
 - `@deprecated` docstring updated — no longer described as decorator-only; documents the `Depends()` usage pattern