Redis in .NET routes: cache, Pub/Sub, Streams with consumer groups and Claim Check in one connector

redb.Route

In an integration project Redis is rarely just one thing. It caches answers from external systems, holds counters and locks, carries events over Pub/Sub, keeps a log of messages in Streams, and works as a job queue on a list. Usually each of those roles lives in its own piece of code, with its own connection, its own retries and its own logging.

In redb.Route they all live in one connector, redb.Route.Redis. Any Redis operation becomes a step of a route, and a subscription, a stream or a list becomes its input. The connection, the telemetry and the graceful shutdown are the ones every other connector uses. Underneath it is StackExchange.Redis 2.11.8.

The consumers are described here as of version 4.1.0: pattern subscriptions, a stream without a group keeping a position of its own, claiming back stream entries that got stuck, and a reliable list queue all arrived in it.

Installing the package, and the address

dotnet add package redb.Route.Redis

services.AddRedbRoute(route =>
{
    route.Services.AddRedbRouteRedis();
    route.AddRouteBuilder<OrderRoutes>();
});

An endpoint address reads redis:OPERATION:resource?parameters. The first path segment is the Redis operation (case does not matter), everything after the first colon is the key, the channel or the stream name. Colons inside a key are kept, so redis:SET:session:42?ttl=300 writes the key session:42 with a 300 second time to live.

The fluent builder produces the same address, and the rest of this post uses it:

.To(Redis.Set("session:42").Ttl(300))   // redis:SET:session:42?ttl=300

Redis has ready factories for keys, Pub/Sub, streams and lists (Set, Get, Incr, Publish, Subscribe, XAdd, XRead, LPush and the rest). Any other operation comes from Redis.Command(operation, key), for example Redis.Command("HSET", "customer:42").Field("email").

Every data structure as a route step

The producer covers the operations of all the main Redis structures:

Structure Operations
Strings and keys SET, GET, DEL, EXISTS, EXPIRE, INCR, DECR, SETNX
Lists LPUSH, RPUSH, LPOP, RPOP, LLEN, LRANGE
Hashes HSET, HGET, HMSET, HMGET, HGETALL, HDEL, HLEN
Sets SADD, SREM, SMEMBERS, SCARD, SISMEMBER
Sorted sets ZADD, ZREM, ZRANGE, ZCARD, ZSCORE, ZRANGEBYSCORE
Geo GEOADD, GEODIST, GEORADIUS
HyperLogLog PFADD, PFCOUNT, PFMERGE
Bitmaps SETBIT, GETBIT, BITCOUNT
Messaging PUBLISH, XADD
Any command COMMAND, with the command name in CustomCommand(...) and its arguments in a header

Where an operation takes its data from is the same rule everywhere:

  • the value or the element is the message body. In lists, hashes, sets, sorted sets and streams a byte[] body is written as it is, without being turned into a string;
  • scalar parameters live in the address: field for hashes, score, minScore and maxScore for sorted sets, start and stop for ranges, longitude, latitude, member1, member2 and geoUnit for geo, offset and bit for bitmaps;
  • sets of values come in headers: the field map for HMSET (redbRedis.HashFields), the field names for HMGET (redbRedis.FieldNames), the source keys for PFMERGE (redbRedis.SourceKeys), the centre and the radius for GEORADIUS, the arguments for COMMAND.

The key, the field, the channel and the stream name may carry ${...} templates, which are resolved per message.

The result of an operation becomes the body of the next step. GET returns a string or null, INCR the new value of the counter, LRANGE and SMEMBERS an array of strings, HGETALL a dictionary, GEORADIUS the members it found:

From("direct://order-created")
    .To(Redis.Incr("stats:orders:today"))
    .Log("orders today: ${body}");

SETNX writes the value only when the key is not there yet and returns "OK" or null. Together with Ttl(...) that is a lock for the duration of the work, or a way to drop repeats.

Pub/Sub: events for whoever is listening right now

From("direct://publish-event")
    .To(Redis.Publish("orders.events"));

From(Redis.Subscribe("orders.events"))
    .To("direct://notify");

After a publish the number of recipients is in the body and in the redbRedis.Publish.Recipients header. On the subscriber side the body is the message text, and the headers carry the channel (redbRedis.Channel), the type (PubSub) and the time it arrived.

A pattern subscription is PSUBSCRIBE: From(Redis.PSubscribe("orders.*")) receives messages from every channel matching the pattern, and redbRedis.Channel holds the channel the message actually came from. The usePattern option turns a plain SUBSCRIBE into a pattern subscription too.

Redis Pub/Sub delivers a message only to those connected at the moment of the publish, and stores nothing. That is the right fit for notifications and cache invalidation. When you need delivery you can rely on, use Streams.

Streams: a log of messages with consumer groups

Writing to a stream:

From("direct://order-created")
    .To(Redis.XAdd("orders").StreamMaxLength(100_000));

The body goes into a data field, alongside a timestamp field in milliseconds. For a field map of your own, pass a dictionary in the redbRedis.StreamFields header. StreamMaxLength trims the stream by length, approximately by default (MAXLEN ~); StreamApproximate(false) makes the trimming exact. The id of the new entry comes back in the body and in the redbRedis.Stream.MessageId header.

Reading with a consumer group:

From(Redis.XRead("orders")
        .ConsumerGroup("billing")
        .ConsumerName("node-1")
        .StreamReadCount(50)
        .StreamClaimMinIdle(60_000))
    .To("direct://bill");

What happens underneath:

  • the group is created for you at start, together with the stream if it isn't there yet. By default it starts at the end of the stream and gets the entries added after it was created; StreamStartPosition("0") starts at the first entry. If the group already exists, the consumer simply joins it;
  • entries are read in batches of streamReadCount. With nothing new to read the consumer waits streamBlockTimeMs (1000 ms by default) and polls again;
  • the body is the dictionary of the entry's fields, each field is also a redbRedis.Stream.<field> header, and the entry id is in redbRedis.MessageId;
  • the consumer name defaults to the machine name.

Several nodes sharing a group, each with its own consumer name, split the stream between them: every entry goes to exactly one of them. That is how one stream is processed by a cluster without duplicates.

Acknowledgement and a second attempt

A group consumer acknowledges an entry (XACK) once the whole unit of work has finished well, the route transaction included. An entry whose route failed is not acknowledged and stays in the group's pending list, which is what makes this at-least-once.

StreamClaimMinIdle(ms) brings those entries back into work. On every poll the consumer claims (XAUTOCLAIM) the group's entries that have been pending for at least that long and processes them again: its own failed ones, and the ones left behind by a consumer on a node that died. Pick an idle time longer than your slowest route, or an entry will be claimed while it is still being worked on.

Where losing an entry is acceptable, there is StreamNoAck(): reading goes with NOACK, an entry counts as delivered the moment it is read and never enters the pending list (at-most-once). It does not combine with StreamClaimMinIdle, and an endpoint asking for both is refused.

A stream without a group

Without a group every consumer reads the whole stream and keeps the position itself, moving it past each entry it has read. That is how one log is handed to several independent readers, or replayed from the beginning:

From(Redis.XRead("orders").StreamStartPosition("0"))
    .To("direct://rebuild-projection");

With no StreamStartPosition the consumer starts at the entries added after it started, "0" reads from the first entry, and an entry id continues after that entry. The > position means "not yet delivered to this consumer group" and only means anything to a group: without one the consumer is not created. Nothing holds a failed entry back when there is no group, and reading moves on.

A list as a job queue

From("direct://enqueue-job")
    .To(Redis.LPush("jobs"));

From(Redis.Command("BRPOP", "jobs")
        .ProcessingList("jobs:processing")
        .PollDelay(500))
    .To("direct://run-job");

LPUSH at the head and reading from the tail give a queue in arrival order. The consumer takes items one at a time by polling and waits pollDelayMs while the list is empty. BLPOP and BRPOP here are the names of the operations rather than blocking commands: a blocking read would hold the shared connection the whole process works through.

With ProcessingList the queue is reliable. The item is moved (LMOVE) into the processing list and leaves it only after the route succeeded. A failed item goes back to the queue atomically, to the side it was taken from, and is processed again; whatever a previous run left in the processing list is returned to the queue at start. One consumer per processing list. LMOVE is available in Redis 6.2 and later.

Without ProcessingList the item is popped before processing and a failed one is gone. That suits work you can afford to lose, such as warming a cache.

Connections and secrets

The shortest path is parameters in the address: connectionString (localhost:6379 by default), database and password. The password is marked as a secret and is redacted in logs.

In production a named connection factory in the context registry is nicer, because secrets stay out of the route address:

context.AddToRegistry("prod", new RedisConnectionFactory
{
    ConnectionString = "redis-1:6379,redis-2:6379",
    User = "route",
    Password = secrets.RedisPassword,
    Ssl = true,
    SslProtocols = "Tls12, Tls13",
});

From("direct://rates")
    .To(Redis.Set("rates:usd").ConnectionFactory("prod").Ttl(300));

The factory knows about Redis 6+ ACL users, TLS with a choice of protocols, Sentinel (ServiceName), connect and operation timeouts, keep-alive, the reconnect policy (exponential or linear) and a channel prefix. Configuration mistakes are not masked: a factory name that is not in the registry, or a typo in SslProtocols, gives a clear error on connect instead of a silent fall back to the defaults. The same goes for address parameters: an option the endpoint does not know, or a value of the wrong type, is refused and named, not quietly dropped.

Every endpoint keeps one StackExchange.Redis connection and reconnects on its own. A lost and a restored connection are both logged.

Claim Check on Redis

The Claim Check pattern from the Enterprise Integration Patterns catalog takes a heavy body out of the message for the length of a route and brings it back when it is needed again. RedisClaimCheckRepository is the store for it:

var claims = new RedisClaimCheckRepository(
    new RedisConnectionFactory { ConnectionString = "redis.internal:6379" },
    defaultTtl: TimeSpan.FromHours(1));

From("direct://inbound-invoice")
    .ClaimCheck(claims, ClaimCheckOperation.Push)   // body into Redis, the key in the message
    .To("direct://route-by-headers")
    .ClaimCheck(claims, ClaimCheckOperation.Pop)    // body back, the entry removed
    .To("direct://archive");

The store writes the data as it is, in binary, and puts Redis's own time to live on the entry. Keys get the redb:claimcheck: prefix, which you can change. Read-and-remove is atomic: it runs as a single Lua script, so it works on Redis older than 6.2, where there is no GETDEL yet.

Observability and shutdown

Every producer call opens an OpenTelemetry span named redis <OPERATION>, tagged db.system=redis, with the resource and the operation on it. Redis calls show up in the same trace as the rest of the route.

Consumers stop gracefully, as they do in every other redb.Route connector: they stop taking new messages and finish the ones already in flight.

Getting it

dotnet add package redb.Route
dotnet add package redb.Route.Redis

Source: github.com/redbase-app/redb-route. The package is on NuGet.

If this was useful — a ⭐ on GitHub helps others find it.

More of my writing: redbase.app/articles, and on dev.to.