Redis Pub/Sub and Keyspace Notifications: Sharded Pub/Sub, At-Most-Once and Cache Invalidation Signals

Version note: The examples in this post were run with Redis 8.10.3 (redis:8), Valkey 9.1.3, .NET 10 and StackExchange.Redis 3.4.35.

Guess First

At Bookstore, the online bookstore used as the shared example across this site's series, the pricing screen publishes a book's ISBN to the bookstore:events:price-changed channel whenever its price changes; the API instances listen on that channel and drop the book from their in-memory caches. In the flow below four price changes are published: one before the listener subscribes, one after it subscribes, one at the moment the server drops every pub/sub connection, and the last one three seconds later:

Log($"PUBLISH 9780441013593 -> receivers: {await pub.PublishAsync(PriceChanged, "9780441013593")}");

await listener.GetSubscriber().SubscribeAsync(PriceChanged, (_, isbn) => Log($"listener: received {isbn}"));

Log($"PUBLISH 9780451524935 -> receivers: {await pub.PublishAsync(PriceChanged, "9780451524935")}");
await Task.Delay(200);

// The server drops every pub/sub connection (as a network problem or a failover would)
long killed = await publisher.GetServers()[0].ClientKillAsync(null, ClientType.PubSub, null, skipMe: true);
Log($"CLIENT KILL TYPE pubsub -> {killed} connection(s)");
Log($"PUBLISH 9780141439518 -> receivers: {await pub.PublishAsync(PriceChanged, "9780141439518")}");
await Task.Delay(3000);
Log($"PUBLISH 9780547928227 -> receivers: {await pub.PublishAsync(PriceChanged, "9780547928227")}");

Which of the four messages does the listener get? Does it receive the missed ones later, once the connection is back?

dotnet run --project src/Bookstore.RedisLab -- pubsub-delivery

 0.00s PUBLISH 9780441013593 -> receivers: 0
 0.00s PUBLISH 9780451524935 -> receivers: 1
 0.01s listener: received 9780451524935
 0.21s CLIENT KILL TYPE pubsub -> 2 connection(s)
 0.21s PUBLISH 9780141439518 -> receivers: 0
 3.21s listener: received 9780547928227
 3.21s PUBLISH 9780547928227 -> receivers: 1

The listener got only the second and the fourth message. The first was lost because nobody was listening at that moment, the third because the connection had dropped and StackExchange.Redis had not resubscribed yet; the receiver count returned by PUBLISH is 0 for both. Redis keeps the lost message nowhere and never delivers it later. The delivery guarantee of pub/sub is at-most-once: a message is delivered at most once, and possibly not at all. This post covers where to use pub/sub with that limit in mind: subscriptions and message order in StackExchange.Redis, the separate pub/sub connection, a cache invalidation signal to several API instances, a "work available" signal for the queue from post 05, sharded pub/sub, keyspace notifications, and Redis 8.8's notifications at the hash field level.

PUBLISH, SUBSCRIBE and At-Most-Once

Pub/sub has no keys, only channels. SUBSCRIBE channel makes a connection a listener of that channel; PSUBSCRIBE bookstore:events:* listens to several channels with a glob pattern; PUBLISH channel message delivers the message to every listener connected at that moment and returns how many it reached. Redis writes the message nowhere; whatever the persistence settings, pub/sub messages never go into the RDB or the AOF. When a connection drops, StackExchange.Redis reconnects and renews the subscriptions itself; that is why the fourth message arrived in Guess First. But the messages in between are gone.

So pub/sub suits signals whose loss the system can recover from: "this book's price changed, drop it from your cache", "there is work in the queue, take a look". Not jobs where a lost message means a customer's e-mail is never sent; for those there are Streams, in post 10.

Subscribing in StackExchange.Redis: a Handler or a ChannelMessageQueue?

ISubscriber.SubscribeAsync comes in two forms: one that takes a callback and one that returns a ChannelMessageQueue. According to the library's docs, the callback form may process messages concurrently and gives no ordering guarantee; the queue form processes messages one at a time, in the order they arrive. I published ten messages to two channels, doing work of 1–20 ms in each:

await sub.SubscribeAsync(handlerChannel, (_, m) =>
{
    Thread.Sleep(random.Next(1, 20));                     // some work of varying length
    handlerSeen.Enqueue((int)m);
});
ChannelMessageQueue queue = await sub.SubscribeAsync(queueChannel);
queue.OnMessage(async message =>
{
    await Task.Delay(random.Next(1, 20));
    queueSeen.Enqueue((int)message.Message);
});

dotnet run --project src/Bookstore.RedisLab -- pubsub-order

handler            : 3 5 1 6 8 4 9 2 10 7
ChannelMessageQueue: 1 2 3 4 5 6 7 8 9 10

With the callback the order was completely scrambled. For cache invalidation order does not matter (it makes no difference which book is dropped first) and concurrent processing is faster; but for messages that carry state, like "the price is now 300 TL" followed by "the price is now 280 TL", a scrambled order lets the old value win. Use the queue in that case, or put a version number in the message.

As we saw in post 02, every multiplexer opens two connections: an interactive one for commands and a separate one for pub/sub. RESP3 could actually manage with a single connection, and StackExchange.Redis has the sharedSubscriptionConnection=true setting for that; since 3.4.0 the default is a separate connection. The library's configuration docs give the reason: Redis applies the much tighter pubsub output buffer limits to every connection that has a subscription. I tried it by reading a 40 MB report with both settings:

dotnet run --project src/Bookstore.RedisLab -- shared-subscription

shared=False  GET 40 MB: 40 MB received, connection failures=0
shared=True   GET 40 MB: RedisConnectionException: SocketClosed on localhost:6379/Interactive, Idle/RanToCompletion, last: GET, origin: ReadAllAsync, outstanding, connection failures=1

On the shared connection, even the multiplexer's own subscription to its configuration channel makes the connection a pub/sub client in Redis's eyes (the flags=P from post 02). The 40 MB reply exceeded the 32 MB pubsub limit and Redis closed the connection, taking every pending command on it along. Unless you have a good reason, leave the default alone.

Bookstore: L1 Invalidation When a Price Changes

Each Bookstore.Api instance keeps book prices in its own memory (L1, IMemoryCache) for 10 minutes; the real value is in a Redis hash (bookstore:prices, in kurus). When the price changes on one instance, the others keep showing the old price from their L1; that is the problem described in "HybridCache in Depth: L1/L2, Tags, Logical Invalidation and L1 Staleness Across Servers" in the Caching series. The solution side and the other options are in that post and in "Cache Invalidation Strategies"; here we build the Redis side. The endpoint that writes the price publishes the change:

// Write the new price, then tell every instance to drop its L1 copy
app.MapPut("/books/{isbn}/price", async (string isbn, PriceRequest request, IConnectionMultiplexer redis) =>
{
    IDatabase db = redis.GetDatabase();
    await db.HashSetAsync(RedisKeys.Prices, isbn, request.PriceInKurus);
    long receivers = await db.PublishAsync(RedisKeys.PriceChanged, isbn);            // PUBLISH
    return Results.Ok(new { isbn, request.PriceInKurus, notifiedInstances = receivers });
});

On every instance a BackgroundService listens to the channel and removes the incoming ISBN from its own L1. When the connection comes back, it clears the whole L1, because the messages sent while the connection was down are lost and there is no way to know which books changed:

// Listens for price changes and removes the book from this instance's L1 cache
public sealed class PriceInvalidationListener(IConnectionMultiplexer redis, IMemoryCache local,
    ILogger<PriceInvalidationListener> logger) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        // Messages sent while the connection was down are lost: start over with an empty L1
        redis.ConnectionRestored += (_, e) =>
        {
            if (e.ConnectionType == ConnectionType.Subscription && local is MemoryCache memory)
            {
                memory.Clear();
                logger.LogWarning("Subscription connection restored, L1 price cache cleared");
            }
        };

        ChannelMessageQueue queue = await redis.GetSubscriber().SubscribeAsync(RedisKeys.PriceChanged);
        try
        {
            await foreach (ChannelMessage message in queue.WithCancellation(stoppingToken))
            {
                local.Remove(BookPriceEndpoints.CacheKey(message.Message!));
                logger.LogInformation("Price of {Isbn} changed, L1 entry removed", (string?)message.Message);
            }
        }
        catch (OperationCanceledException)
        {
        }
        finally
        {
            await queue.UnsubscribeAsync();
        }
    }
}

I started two instances on ports 5080 and 5081 (dotnet run --project src/Bookstore.Api -- --urls http://localhost:5081 --InstanceName api-2):

curl http://localhost:5080/books/9780441013593/price
{"isbn":"9780441013593","priceInKurus":29990,"source":"redis","instance":"api-1"}
curl http://localhost:5081/books/9780441013593/price
{"isbn":"9780441013593","priceInKurus":29990,"source":"redis","instance":"api-2"}
curl http://localhost:5081/books/9780441013593/price
{"isbn":"9780441013593","priceInKurus":29990,"source":"l1","instance":"api-2"}
curl -X PUT http://localhost:5080/books/9780441013593/price -H "Content-Type: application/json" -d '{"priceInKurus":27990}'
{"isbn":"9780441013593","priceInKurus":27990,"notifiedInstances":2}
curl http://localhost:5081/books/9780441013593/price
{"isbn":"9780441013593","priceInKurus":27990,"source":"redis","instance":"api-2"}

info: Bookstore.Api.PriceInvalidationListener[0]
      Price of 9780441013593 changed, L1 entry removed
warn: Bookstore.Api.PriceInvalidationListener[0]
      Subscription connection restored, L1 price cache cleared

The price changed on 5080, and 5081 read the new price from Redis on its next request. The second log line was written at the end of the experiment, when I dropped the subscription with CLIENT KILL TYPE pubsub. Because a signal can be lost, the L1 lifetime has to stay short too: 10 minutes is the upper bound on how stale a price can be after a missed signal. SignalR's Redis backplane works with the same mechanism: every server subscribes to shared channels and forwards the messages to the clients connected to other servers. The details are in "Scaling Out and Running SignalR in Production" in the Real-time Applications series.

A "Work Available" Signal for the Queue

In post 05 we saw BLPOP freeze the shared connection in StackExchange.Redis and noted the docs' advice: keep the data in the list and let pub/sub only say "there is work". The e-mail worker now waits on the channel to sleep, but for at most 2 seconds; if a signal is lost, polling takes over:

while (!stop.IsCancellationRequested)
{
    RedisValue job;
    while (!(job = await workerDb.ListMoveAsync(queue, processing, ListSide.Right, ListSide.Left)).IsNull)
    {
        Console.WriteLine($"  worker: {job} picked up after {sw.ElapsedMilliseconds - enqueuedAt[job!],5} ms");
        await workerDb.ListRemoveAsync(processing, job, count: 1);
    }
    // Sleep until a signal arrives, but never longer than the polling interval
    using CancellationTokenSource poll = CancellationTokenSource.CreateLinkedTokenSource(stop.Token);
    poll.CancelAfter(TimeSpan.FromSeconds(2));
    try
    {
        await wakeUps.ReadAsync(poll.Token);
    }
    catch (OperationCanceledException)
    {
    }
}

dotnet run --project src/Bookstore.RedisLab -- queue-signal

api: LPUSH order-1001, PUBLISH signal -> 1
  worker: order-1001 picked up after    12 ms
api: LPUSH order-1002, PUBLISH signal -> 1
  worker: order-1002 picked up after     2 ms
api: LPUSH order-1003, signal lost
  worker: order-1003 picked up after  1310 ms

Jobs with a signal were picked up within milliseconds; the job whose signal was lost was picked up at the next poll, 1.3 seconds later. The polling interval sets the maximum delay a lost signal can cause. No command on the shared multiplexer waits, and the empty queue does not get hundreds of LMOVE calls per second.

Sharded Pub/Sub

In Redis Cluster, classic PUBLISH spreads every message to every node in the cluster; that traffic grows with the number of nodes. Sharded pub/sub (SSUBSCRIBE, SPUBLISH), added in Redis 7.0, maps the channel name to a slot like a key and keeps the message within the shard of that slot. It also works on the single-server lab:

127.0.0.1:6379> SPUBLISH bookstore:events:price-changed:{9780441013593} 27990
(integer) 1
127.0.0.1:6379> PUBSUB SHARDNUMSUB bookstore:events:price-changed:{9780441013593}
1) "bookstore:events:price-changed:{9780441013593}"
2) (integer) 1
127.0.0.1:6379> PUBLISH bookstore:events:price-changed:{9780441013593} 27990
(integer) 0

A redis-cli listening in the background with SSUBSCRIBE received the message; a classic PUBLISH to the channel with the same name reached 0 receivers, because the two mechanisms have separate channels. In StackExchange.Redis a channel created with RedisChannel.Sharded("…") uses SSUBSCRIBE/SPUBLISH. You can use a hash tag in a channel name as well; the details of slots and shards are in post 13. Valkey 9.1.3 has the sharded pub/sub commands.

Keyspace Notifications

Redis can also publish changes to keys on pub/sub channels: the __keyspace@0__:key channel carries the events on that key, and the __keyevent@0__:event channel carries the keys on which that event happened. They are off by default and enabled with the notify-keyspace-events setting. Ex enables only expire events (x) on only the key-event channels (E). The idea is tempting: "tell me when a verification code expires", "put the stock back when a reservation expires". I measured when the expired events arrive for 20,000 keys that all expire at the same moment:

dotnet run --project src/Bookstore.RedisLab -- expired-events

expired events: 20,000; after the TTL reached zero: first 99 ms, median 129 ms, last 299 ms

The events arrive not when the TTL reaches zero but when Redis actually deletes the key. As described in post 03, that happens either when the key is accessed or in the active expire cycle in the background; here the first event was 99 ms late and the last 299 ms. According to the docs, with many keys with a TTL and no access to them this delay can become significant. Add pub/sub's at-most-once delivery on top, and the fact that in a cluster each node publishes events only for its own keys (you have to subscribe to every node separately to get all of them). So do not make the expired event the only trigger of business logic: if expired reservations have to go back into stock, write a job that scans a sorted set periodically, as in the Small Experiment of post 08; let the event be at most a signal that wakes that job up early.

Hash Field Notifications and Client-Side Caching

Classic keyspace notifications tell you an HSET happened on a hash, but not which field changed. Redis 8.8 introduced subkey notifications; the event message now carries the field names too. I tried them with the cart price locks from post 04 (per-field TTL with HSETEX):

127.0.0.1:6379> CONFIG SET notify-keyspace-events STh
OK
127.0.0.1:6379> HSETEX bookstore:cart:7:price-locks PX 1500 FIELDS 2 9780441013593 29990 9780141439518 18950
(integer) 1
127.0.0.1:6379> HSET bookstore:cart:7:price-locks 9780451524935 22500
(integer) 1
127.0.0.1:6379> HDEL bookstore:cart:7:price-locks 9780451524935
(integer) 1

$ docker compose -f lab/compose.yaml exec redis redis-cli --csv psubscribe "__subkey*"
"psubscribe","__subkey*",1
"pmessage","__subkey*","__subkeyspace@0__:bookstore:cart:7:price-locks","hset|13:9780441013593,13:9780141439518"
"pmessage","__subkey*","__subkeyevent@0__:hset","28:bookstore:cart:7:price-locks|13:9780441013593,13:9780141439518"
"pmessage","__subkey*","__subkeyspace@0__:bookstore:cart:7:price-locks","hexpire|13:9780441013593,13:9780141439518"
"pmessage","__subkey*","__subkeyevent@0__:hexpire","28:bookstore:cart:7:price-locks|13:9780441013593,13:9780141439518"
"pmessage","__subkey*","__subkeyspace@0__:bookstore:cart:7:price-locks","hset|13:9780451524935"
"pmessage","__subkey*","__subkeyevent@0__:hset","28:bookstore:cart:7:price-locks|13:9780451524935"
"pmessage","__subkey*","__subkeyspace@0__:bookstore:cart:7:price-locks","hdel|13:9780451524935"
"pmessage","__subkey*","__subkeyevent@0__:hdel","28:bookstore:cart:7:price-locks|13:9780451524935"
"pmessage","__subkey*","__subkeyspace@0__:bookstore:cart:7:price-locks","hexpired|13:9780141439518,13:9780441013593"
"pmessage","__subkey*","__subkeyevent@0__:hexpired","28:bookstore:cart:7:price-locks|13:9780141439518,13:9780441013593"

S enables the per-key channels (__subkeyspace) and T the per-event channels (__subkeyevent); field names come with a length prefix (13:9780441013593), so names containing commas can be parsed safely too. When both price locks expired, a single hexpired message reported them together. There is a difference between the docs and the measurement here: the example on the subkey notifications page sets only ST and receives messages for an HSET. On Redis 8.10.3 no message arrived with ST; they arrived once I added h (hash commands) from the same page's flag list (STh). Valkey 9.1.3 rejected STh with an invalid character error; subkey notifications do not exist on Valkey. StackExchange.Redis 3.4 has channel helpers such as RedisChannel.SubKeyEvent and ChannelMessage.TryParseKeyNotification, which parses the fields out of the message.

A feature closely related to notifications is client-side caching: for a connection that says CLIENT TRACKING ON, Redis remembers the keys that connection read and sends an invalidation message when a key changes (on the same connection in RESP3, redirected to another connection in RESP2). It is the server-side version of the L1 invalidation we built by hand above. I did not find a ready-made L1 cache layer using it in the public API of StackExchange.Redis 3.4.35; on the .NET side this belongs to HybridCache and the Caching series.

In Production

The most common production problem with pub/sub is a slow subscriber. If Redis cannot write a message to a listener's TCP connection, it accumulates it in that listener's output buffer. I opened a subscriber that never reads and published 16 KB messages:

dotnet run --project src/Bookstore.RedisLab -- slow-subscriber

client-output-buffer-limit: normal 0 0 0 slave 268435456 67108864 60 pubsub 33554432 8388608 60
published   500 x 16 KB  receivers=1  slow-subscriber: omem=5043984
published 1,000 x 16 KB  receivers=1  slow-subscriber: omem=13286592
published 1,500 x 16 KB  receivers=1  slow-subscriber: omem=21529200
published 2,000 x 16 KB  receivers=1  slow-subscriber: omem=29771808
published 2,231 x 16 KB  receivers=0  slow-subscriber: gone

docker compose -f lab/compose.yaml logs redis
... name=slow-subscriber ... flags=P ... omem=33565048 ... cmd=subscribe ... scheduled to be closed ASAP for overcoming of output buffer limits.

pubsub 33554432 8388608 60: if the buffer exceeds 32 MB the connection is closed immediately, and also if it stays above 8 MB for 60 seconds. At the 2,231st message the buffer passed 32 MB and Redis dropped the connection; the messages the subscriber missed will not come back. The limit protects the server's memory: without it a single client that never reads would bloat memory like the MONITOR in post 02. On the prevention side: do not do heavy work in the listener the moment a message arrives; put it on an in-memory queue (ChannelMessageQueue or System.Threading.Channels) and process it separately; do not send large objects in messages, send only an identity (like an ISBN); and watch the omem of connections with flags=P in CLIENT LIST. Enable keyspace notifications only for the event classes you need; KEA produces a message for every write command and puts load on the CPU.

Common Pitfalls

Using pub/sub as a job queue: in Guess First two of four messages were lost and nobody noticed. For work that cannot be lost, use Streams (post 10).

Processing messages that carry state with a callback subscription: ten messages were processed in scrambled order; if order matters, use a ChannelMessageQueue.

Doing nothing after a reconnect: the messages in between are lost; you have to clear the L1 or read the source again.

Reading large replies with sharedSubscriptionConnection=true: a 40 MB GET dropped the command connection.

Treating the expired event as a timer: it arrives when the key is deleted, not when the TTL ends, and it can be lost.

A slow subscriber: after 32 MB Redis closed the connection.

Assuming Redis 8.8's subkey notifications on Valkey, or setting only ST and waiting for messages.

A Small Experiment

★☆☆ Open redis-cli in two terminals; run PSUBSCRIBE bookstore:events:* in one and PUBLISH to two different channels from the other. What do PUBSUB NUMPAT and PUBSUB CHANNELS show?

★★☆ Change PriceInvalidationListener to a callback subscription and add the price to the message (isbn:price); the listener updates the L1 with the new price instead of removing it. When does a wrong price remain if you publish two prices for the same book back to back? Fix it by adding a version number to the message.

★★★ Repeat the expired-events experiment with 200,000 keys and with hz raised from 10 to 100 (only in the lab). How do the delays of the first and last events change? Compare with expired_keys in INFO stats.

In the next post we move on to messages that must not be lost: Redis Streams, consumer groups, the pending entry list, taking over a dead worker's messages with XAUTOCLAIM, and Redis 8.6's idempotent XADD; we will move the e-mail queue from post 05 to a stream.