CacheInvalidation.org

Implementing Redis Pub/Sub for Real-Time Cache Invalidation

You run several services that each keep a local Redis cache, and a write in one service must evict the matching key everywhere else within milliseconds. Polling for changes wastes CPU and still leaves a stale window, while synchronous HTTP fan-out couples every service to the availability of every other. Redis Publish/Subscribe solves the propagation problem with a single asynchronous broadcast: one PUBLISH reaches every subscribed service instantly. The catch is that Pub/Sub is fire-and-forget — a subscriber that is disconnected, throttled, or blocked when the message is sent never sees it. This page implements a subscriber and publisher that survive that reality: a heartbeat-aware receive loop, exponential-backoff reconnection, and the diagnostics that expose silent message loss before it corrupts your caches. It is the runtime layer beneath Pub/Sub Routing for Cross-Service Invalidation, and a lighter-weight alternative to the durable queue in Building Async Invalidation Queues with Celery when you need speed over guaranteed delivery.

Prerequisites

  • Redis 6.2+ (7.x recommended for sharded Pub/Sub via SSUBSCRIBE on a Redis cluster).
  • redis-py 5.0+ and Python 3.10+ (the examples use redis.asyncio).
  • A deterministic channel naming scheme, e.g. cache:invalidate:<domain>:<entity>, agreed across all services.
  • Reviewed client-output-buffer-limit pubsub in redis.conf (default 32mb 8mb 60); subscribers that exceed it are force-disconnected.
  • Invalidation payloads kept under a few KB — Pub/Sub is a signal, never a data channel.

Step-by-Step Implementation

Each step below is independently runnable: paste it into a REPL or module, point REDIS_URL at your instance, and it works on its own before you compose them into a service.

1. Build a resilient async connection pool. Tune the pool for long-lived signaling rather than short request/response traffic, so idle subscriber connections survive load balancers and NAT timeouts.

import redis.asyncio as redis

REDIS_URL = "redis://localhost:6379/0"

def make_client() -> redis.Redis:
    pool = redis.ConnectionPool.from_url(
        REDIS_URL,
        max_connections=10,
        socket_timeout=2.0,          # fail fast on a dead socket
        socket_keepalive=True,       # detect drops behind LB/NAT
        health_check_interval=30,    # periodic PING keeps the connection warm
        decode_responses=True,
    )
    return redis.Redis(connection_pool=pool)

2. Subscribe to a deterministic channel hierarchy. Use a namespaced pattern so a subscriber receives only the domains it owns, avoiding backpressure from unrelated traffic.

async def subscribe(client: redis.Redis, pattern: str = "cache:invalidate:*"):
    pubsub = client.pubsub(ignore_subscribe_messages=True)
    await pubsub.psubscribe(pattern)   # pattern match; use subscribe() for exact channels
    return pubsub

3. Run a heartbeat-aware receive loop with exponential backoff. Replace the blocking listen() generator with a timed get_message() poll so you can validate the connection and reconnect cleanly instead of hanging on a half-open socket.

import asyncio, logging
from redis.exceptions import ConnectionError, TimeoutError

log = logging.getLogger("cache.invalidation")

async def run(client: redis.Redis, pattern: str, handler):
    attempt = 0
    while True:
        try:
            pubsub = await subscribe(client, pattern)
            attempt = 0  # reset backoff after a clean subscribe
            while True:
                await client.ping()  # surface silent TCP drops
                msg = await pubsub.get_message(timeout=1.0)
                if msg and msg["type"] in ("message", "pmessage"):
                    await handler(msg["data"])
        except (ConnectionError, TimeoutError) as e:
            delay = min(30.0, 1.0 * (2 ** attempt))  # capped exponential backoff
            attempt += 1
            log.warning("pubsub reconnect in %.1fs: %s", delay, e)
            await asyncio.sleep(delay)

4. Publish invalidation events with a small structured payload. Publishers are stateless: a single PUBLISH reaches every current subscriber, so keep the message to the key or key-pattern being evicted.

import json

async def invalidate(client: redis.Redis, domain: str, entity: str, key: str):
    channel = f"cache:invalidate:{domain}:{entity}"
    payload = json.dumps({"key": key, "op": "del"})
    receivers = await client.publish(channel, payload)  # PUBLISH returns subscriber count
    log.info("published %s -> %d subscribers", channel, receivers)
    return receivers

5. Make the handler idempotent. Reconnects and retries can redeliver or overlap, so an invalidation handler must be safe to run more than once for the same key.

async def handle(client: redis.Redis, data: str):
    msg = json.loads(data)
    # DEL is naturally idempotent — deleting an absent key is a no-op.
    await client.delete(msg["key"])

How the Signal Propagates

The critical path is a single asynchronous hop from writer to every reader. If a subscriber is offline at that instant, the message is gone — reconnection restores the stream, not the missed events, which is why Step 3 caps backoff and Step 5 stays idempotent.

Fire-and-forget Pub/Sub invalidation fan-out with one offline subscriber Service A commits a write and issues a single PUBLISH to the channel cache:invalidate:users:profile. Redis broadcasts the message to all currently connected subscribers with no per-message buffer and no replay. Service B and Service C are connected: each receives the message and runs DEL on its local key. Service D is offline and reconnecting with capped exponential backoff, so the broadcast never reaches it and that eviction is lost until the key's TTL backstop expires. Service A write commits Redis cache:invalidate:* fire-and-forget broadcast no buffer · no replay Service B → DEL users:profile:42 delivered ✓ Service C → DEL users:profile:42 delivered ✓ Service D reconnecting · backoff message lost ✗ ① PUBLISH users:profile:42 ② deliver ② deliver ✗ lost One PUBLISH reaches only the subscribers connected at that instant — a disconnected service gets no replay, so its eviction waits on the TTL backstop.

Channel naming should stay deterministic (cache:invalidate:users:profile) so pattern subscribers match precisely without waking every consumer. This targeted routing is the core concern of the parent Pub/Sub Routing for Cross-Service Invalidation page; here it prevents one noisy domain from filling another subscriber's output buffer.

Failure Modes

Silent subscriber drop from output-buffer overflow. A slow handler lets Redis queue messages until the subscriber crosses client-output-buffer-limit pubsub, at which point Redis force-closes the connection to protect its event loop — publishing continues but that service silently stops evicting. Diagnose it by watching the subscriber count fall to zero while writes continue, and by inspecting the buffer high-water mark:

redis-cli PUBSUB NUMSUB cache:invalidate:users:profile   # 0 during active writes => dropped
redis-cli INFO clients | grep client_recent_max_output_buffer

Fix it by moving slow work out of the receive loop (dispatch to a task/queue), shrinking payloads to a bare key, and only then raising the limit if genuinely needed.

Lost messages across a reconnect window. Because delivery is fire-and-forget, any event published while a subscriber is reconnecting is never re-sent, leaving one cache stale until its next natural expiry. Diagnose by correlating reconnect log lines with an unexpectedly high total_net_input_bytes gap or user-visible staleness:

redis-cli INFO stats | grep -E "pubsub_channels|rejected_connections"

Fix it with a bounded TTL floor on every cached key as a backstop, and for critical data pair Pub/Sub with a durable reconciliation layer such as Redis Streams or the queue described in the asynchronous invalidation workflows guide.

Cluster redirection breaks non-sharded Pub/Sub. On Redis Cluster, classic PUBLISH/SUBSCRIBE is broadcast cluster-wide but a CLUSTERDOWN or a mid-MOVED topology change can interrupt subscribers during a resharding operation. Diagnose it by checking cluster health while migrations run:

redis-cli CLUSTER INFO | grep cluster_state   # expect cluster_state:ok

Fix it by using sharded Pub/Sub (SSUBSCRIBE/SPUBLISH, Redis 7+) so the channel is pinned to a slot, and by scheduling invalidation-sensitive work away from active resharding as covered in the Step-by-Step Redis Cluster Slot Migration Guide.

Verification

Confirm the pipeline end to end before trusting it in production.

Check that subscribers are actually attached to the channel a publisher uses:

redis-cli PUBSUB NUMSUB cache:invalidate:users:profile
# => 1) "cache:invalidate:users:profile" 2) (integer) 3

Assert propagation latency under load in an ephemeral container, gating merges on a p99 threshold:

import asyncio, time, redis.asyncio as redis

async def test_propagation_under_50ms():
    client = redis.Redis.from_url("redis://localhost:6379/0", decode_responses=True)
    ps = client.pubsub(ignore_subscribe_messages=True)
    await ps.subscribe("cache:invalidate:test")
    await asyncio.sleep(0.05)  # let the subscription register
    t0 = time.perf_counter()
    await client.publish("cache:invalidate:test", "k")
    msg = None
    while msg is None:
        msg = await ps.get_message(timeout=1.0)
    assert (time.perf_counter() - t0) < 0.05

Validate that peak publish volume does not trip the output-buffer limit:

redis-benchmark -c 500 -n 100000 -t publish

Confirm the subscriber's backoff recovers without duplication under injected latency:

tc qdisc add dev eth0 root netem delay 100ms jitter 20ms

FAQ

Can I send the updated value over Pub/Sub instead of just an eviction signal? No. Pub/Sub has no delivery guarantee, no ordering across reconnects, and a per-subscriber output buffer. Publish only the key or pattern to invalidate and let each service re-read from the source of truth. Shipping payloads risks buffer overflow and inconsistent state.

Why use get_message(timeout=...) instead of the blocking listen() loop? listen() blocks indefinitely on a half-open socket, so a dropped connection behind a load balancer goes undetected. A timed poll lets you interleave a ping() heartbeat, honor shutdown signals, and drive reconnection with backoff.

Do I still need TTLs if Pub/Sub invalidation is instant? Yes. Fire-and-forget means any message published while a subscriber is reconnecting is lost forever. A bounded TTL caps how long a missed eviction can serve stale data, acting as a safety net beneath the real-time path.

Does Pub/Sub work correctly on Redis Cluster? Classic Pub/Sub broadcasts across the whole cluster, which works but ignores slots and can be disrupted during resharding. On Redis 7+, prefer sharded Pub/Sub (SSUBSCRIBE/SPUBLISH) so channels are slot-scoped and behave predictably during slot migration.

When should I choose a durable queue over Pub/Sub? When missing an invalidation is unacceptable — financial balances, permission changes. Pub/Sub optimizes for latency; a queue like Celery or Redis Streams optimizes for guaranteed, replayable delivery. Many teams run both: Pub/Sub for speed, a queue for reconciliation.

Up one level: Pub/Sub Routing for Cross-Service Invalidation