Redis recipes

These recipes use the Respire package; leader election also needs Respire.Coordination. Copy the classes you need from Respire.Samples.Recipes. The code blocks below display those source files directly. The existing CI integration jobs compile and execute them against Redis 8.4 in both RESP2 and RESP3 through RecipeIntegrationTests; they are not separate, untested copies of the examples.

Recipe	Server requirement	Data layout
Leaderboard	Redis 7+ or compatible Valkey	Sorted set of scores and hash of member names
Sessions	Redis 8.0+ (HSETEX, HGETEX)	One hash per user, one expiring field per session
Outbox	Redis 8.4+ (XREADGROUP CLAIM)	Business state hash and consumer-group stream in one slot
Leader election	Redis 7+ or compatible Valkey	Expiring lease key and persistent fencing counter
Environment connection	Any server supported by Respire	REDIS_URL, then VALKEY_URL

Keep one client for the application lifetime and dispose it at shutdown. Examples use fixed keys for clarity; use a key prefix to isolate applications and tenants. Select server versions explicitly: a Valkey connection URL does not imply support for the Redis 8.x commands used by sessions and the outbox.

Connect from the environment

Set REDIS_URL=redis://localhost:6379/0 (or VALKEY_URL) in the process environment, then call EnvironmentConnection.ConnectAsync(). No configuration file is needed. This is a recipe helper: the core client does not automatically read these variables. The first nonblank value wins, with REDIS_URL taking precedence. Missing configuration fails at startup instead of silently connecting to localhost. An invalid primary URL fails; it does not trigger fallback to a different server.

EnvironmentConnection.cs

using Respire;

namespace Respire.Samples.Recipes;

public static class EnvironmentConnection
{
    public static ValueTask<RespireClient> ConnectAsync(CancellationToken cancellationToken = default)
        => ConnectAsync(Environment.GetEnvironmentVariable, cancellationToken);

    public static ValueTask<RespireClient> ConnectAsync(Func<string, string?> lookup,
        CancellationToken cancellationToken = default)
        => RespireClient.ConnectAsync(Resolve(lookup), cancellationToken);

    // Accepting a lookup also lets tests exercise precedence without changing process globals.
    public static string Resolve(Func<string, string?> lookup)
    {
        foreach (var name in new[] { "REDIS_URL", "VALKEY_URL" })
        {
            var value = lookup(name);
            if (!string.IsNullOrWhiteSpace(value)) return value;
        }

        throw new InvalidOperationException("Set REDIS_URL or VALKEY_URL before starting the application.");
    }
}

Use rediss:// for TLS and percent-encode reserved characters in URI credentials. Supply secrets through your deployment's secret store; do not log the URL. See connection strings for authentication and timeout options. The tests inject the lookup function so they can check both variable names and precedence without changing environment variables for other tests.

Leaderboard with member metadata

Create new Leaderboard(client), call SaveAsync("ada", "Ada", 1200), and read TopAsync(10). The sorted set stores stable member IDs as members and numeric scores as scores; the hash stores the display name under the same ID. Repeating a save replaces the score and name, so retrying the same values does not increment a score twice.

Leaderboard.cs

using Respire;

namespace Respire.Samples.Recipes;

public sealed class Leaderboard(IRespireClient client)
{
    public const string ScoresKey = "{leaderboard:season-1}:scores";
    public const string NamesKey = "{leaderboard:season-1}:names";

    public async Task SaveAsync(string memberId, string displayName, double score,
        CancellationToken cancellationToken = default)
    {
        // Both keys share a Cluster slot. All writers must reserve these key types.
        await using var transaction = client.CreateTransaction();
        var name = transaction.Hashes.Set(NamesKey, (memberId, (RespireValue)displayName));
        var rank = transaction.SortedSets.Add(ScoresKey, memberId, score);
        await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
        _ = name.Result;
        _ = rank.Result;
    }

    public async Task<Row[]> TopAsync(int count, CancellationToken cancellationToken = default)
    {
        ArgumentOutOfRangeException.ThrowIfNegativeOrZero(count);
        var entries = await client.SortedSets.RangeWithScoresAsync(
            ScoresKey, 0, count - 1, descending: true, cancellationToken).ConfigureAwait(false);
        if (entries.Length == 0) return [];

        var names = await client.Hashes.GetManyAsync(
            NamesKey, entries.Select(entry => entry.Member).ToArray(), cancellationToken).ConfigureAwait(false);
        return entries.Select((entry, index) =>
            new Row(entry.Member, names[index] ?? entry.Member, entry.Score)).ToArray();
    }

    public sealed record Row(string MemberId, string DisplayName, double Score);
}

Both keys carry the same hash tag for Redis Cluster. MULTI/EXEC prevents other clients from interleaving the two writes. Reserve these key types: Redis transactions do not roll back earlier commands when a later command has a runtime error. Both pending results are observed so such failures reach the caller.

Reads fetch descending scores followed by one HMGET for names. They are not a snapshot across those two commands; a concurrent name update may be visible. A missing name falls back to the member ID. Equal scores use Redis's reverse lexicographic member order, so the recipe does not invent a tie-breaking rank.

The board has no implicit expiry. Use a new pair of keys per season, then delete both old keys when that season's retention ends. For rolling retention, apply the same absolute deadline to both keys rather than expiring scores and metadata independently. Delete a departing member from both structures in one transaction. See sorted sets for score increments and range queries.

Session store with independent expiry

Create new SessionStore(client). Save an opaque, cryptographically random session ID with SaveAsync(userId, sessionId, payload, TimeSpan.FromMinutes(30)). On authenticated use, call ReadAndRefreshAsync with the same idle timeout; a null result means the user needs to sign in again. RevokeAsync deletes just that session field.

SessionStore.cs

using Respire;

namespace Respire.Samples.Recipes;

public sealed class SessionStore(IRespireClient client)
{
    // One hash per user; every opaque session ID has an independent sliding lifetime.
    public ValueTask<bool> SaveAsync(string userId, string sessionId, string payload,
        TimeSpan idleTimeout, CancellationToken cancellationToken = default)
        => client.Hashes.SetExpireAsync(Key(userId), idleTimeout,
            (sessionId, (RespireValue)payload), cancellationToken); // HSETEX

    public async Task<string?> ReadAndRefreshAsync(string userId, string sessionId,
        TimeSpan idleTimeout, CancellationToken cancellationToken = default)
    {
        var values = await client.Hashes.GetAndExpireAsync(
            Key(userId), idleTimeout, sessionId, cancellationToken).ConfigureAwait(false); // HGETEX
        return values[0]; // Missing or expired sessions stay missing.
    }

    public ValueTask<long> RevokeAsync(string userId, string sessionId,
        CancellationToken cancellationToken = default)
        => client.Hashes.RemoveAsync(Key(userId), sessionId, cancellationToken);

    public static string Key(string userId) => $"sessions:{userId}";
}

HSETEX writes and expires a field atomically. HGETEX returns it and slides only that field's expiry. Reading an expired or absent session never recreates it; another device's session keeps its own deadline. Do not set a key-level TTL, because that would remove all of the user's sessions together. Do not update the field with ordinary HSET, which can clear its field expiry; use this save path consistently.

Treat the payload as application data, not proof of authorization. Validate the session's user binding and access rights on the server. For an absolute maximum lifetime, include the creation/expiry time in the payload, validate it before authorizing the request, and revoke the session when the limit is reached. A sliding idle timeout alone allows a continuously active session to live indefinitely. Fail closed on Redis errors and never turn a timeout into a newly authenticated session. See hash expiry for the underlying commands.

Transactional outbox relay over streams

This recipe stores both business state and the outbox in Redis. Call SaveOrderAsync("order-42", "paid", "event-42") with a stable, application-assigned event ID. The transaction saves the order state and uses the job queue to enqueue the event in the same Cluster slot. Respire.Streaming supplies the typed producer; the job ID is the event ID and its JSON payload contains the order ID and state. At relay startup call InitializeRelayAsync; it starts the group at 0-0, so entries written before the group existed are included. Calling it again is safe.

Outbox.cs

using Respire;
using Respire.Streaming;
using System.Text.Json;
using System.Text.Json.Serialization;

namespace Respire.Samples.Recipes;

public sealed class Outbox(IRespireClient client)
{
    public const string StateKey = "{respire-job:orders}:state";
    public const string StreamKey = "{respire-job:orders}:ready";
    public const string Group = "publishers";
    private readonly RespireJobQueue<OrderEvent> _queue = new(client, "orders",
        payload => JsonSerializer.SerializeToUtf8Bytes(payload, OutboxJson.Default.OrderEvent));

    public async Task<RespireStreamId> SaveOrderAsync(string orderId, string state,
        string eventId, CancellationToken cancellationToken = default)
    {
        // This transaction is for business state stored in Redis, not an external SQL database.
        await using var transaction = client.CreateTransaction();
        var saved = transaction.Hashes.Set(StateKey, (orderId, (RespireValue)state));
        var added = _queue.Enqueue(transaction, eventId, new OrderEvent(orderId, state));
        await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
        _ = saved.Result;
        return added.Result.EntryId;
    }

    public ValueTask<bool> InitializeRelayAsync(CancellationToken cancellationToken = default)
        => client.Streams.CreateGroupAsync(StreamKey, Group, "0-0", cancellationToken: cancellationToken);

    public async Task<int> RelayPageAsync(string consumer, TimeSpan claimMinIdle,
        Func<RespireStreamEntry, CancellationToken, Task> publish,
        CancellationToken cancellationToken = default)
    {
        // Redis 8.4: recover abandoned deliveries and read new entries in the same command.
        var entries = await client.Streams.ReadGroupOnceAsync(StreamKey, Group, consumer,
            new RespireStreamReadOptions { Count = 10, ClaimMinIdle = claimMinIdle },
            cancellationToken: cancellationToken).ConfigureAwait(false);
        foreach (var entry in entries)
        {
            await publish(entry, cancellationToken).ConfigureAwait(false);
            await entry.AckAsync(cancellationToken).ConfigureAwait(false); // Only acknowledge confirmed publication.
        }
        return entries.Length;
    }
}

public sealed record OrderEvent(string OrderId, string State);

[JsonSerializable(typeof(OrderEvent))]
internal partial class OutboxJson : JsonSerializerContext;

Give each relay process a unique consumer name. Call RelayPageAsync repeatedly from a cancellable background service, passing a publisher that completes only when the downstream system confirms acceptance. Back off after an empty page or a transient error. Choose claimMinIdle longer than the maximum processing time for the whole ten-entry page, including downstream retries; otherwise another worker can claim an entry still being processed. XREADGROUP CLAIM delivers eligible abandoned entries as well as new entries.

Publication failures leave entries pending. A process crash after publication but before XACK causes redelivery, so this is at-least-once delivery. The receiver must atomically deduplicate by _respire.job.id with its own business update. Read the JSON event from _respire.job.payload. Retrying an uncertain producer transaction with the same event ID deduplicates enqueue within the default one-day window; business writes must also be idempotent. After that window expires, the event can be appended again. Do not retry with a new ID. XACK removes the pending record, not the stream entry.

Reserve the business hash and the queue's stream/deduplication key namespace: MULTI/EXEC has no rollback for runtime command errors. Inspect both pending results. Configure persistence, replication, and eviction for durable business data. Do not apply a TTL or arbitrary MAXLEN trim to unprocessed outbox entries. Set a retention policy that accounts for every consumer group's progress, monitor pending age/delivery counts, and quarantine poison messages durably before acknowledging them. This small relay stops the page on a publication failure; it does not silently discard a bad entry.

An SQL transaction cannot atomically commit this Redis transaction. For SQL-owned orders, write the outbox row in the same SQL transaction, then relay it to Redis with stable IDs and receiver-side deduplication. See stream workers for a hosted worker with recovery and dead-letter handling.

Leader election with a renewable lease

For hosted applications, use the RespireLeaderElection API for automatic campaigning, renewal, revocation callbacks, and leadership metrics. The recipe below shows a single acquisition for applications that own their retry loop.

Every contender calls LeaderElection.TryRunAsync(client, leaderWork, stoppingToken). A false result means another contender holds the lease; the caller can retry with jitter and a cancellation-aware delay. A successful contender runs its callback under a renewable 30-second fenced lease, then releases it on completion or failure.

LeaderElection.cs

using Respire;
using Respire.Coordination;

namespace Respire.Samples.Recipes;

public static class LeaderElection
{
    public static async Task<bool> TryRunAsync(IRespireClient client,
        Func<long, CancellationToken, Task> leaderWork, CancellationToken cancellationToken = default)
    {
        var coordination = new RespireCoordination(client);
        await using var attempt = await coordination.TryAcquireFencedLockAsync(
            "{scheduler}:leader", "{scheduler}:fence", TimeSpan.FromSeconds(30), cancellationToken)
            .ConfigureAwait(false);
        if (!attempt.Acquired) return false;

        await using var renewal = await attempt.Lock.KeepAliveAsync(cancellationToken).ConfigureAwait(false);
        using var lifetime = CancellationTokenSource.CreateLinkedTokenSource(
            cancellationToken, renewal.CancellationToken);
        // The protected resource must reject writes with an older fencing token.
        await leaderWork(attempt.Lock.FencingToken, lifetime.Token).ConfigureAwait(false);
        lifetime.Token.ThrowIfCancellationRequested();
        return true;
    }
}

The callback must honor its cancellation token, including during downstream I/O. Renewal failure cancels that token; cancellation is cooperative and cannot stop a paused or unresponsive process. Pass the fencing token to the protected resource on every write. That resource must atomically reject a token older than its last accepted token. Equal tokens allow repeated writes by the same leader. Do not start detached work that outlives the callback.

Only the lease expires. Never expire, delete, reset, or evict the fencing counter. An old leader's release compares ownership, so it cannot release a replacement's lease. The counter's monotonicity depends on retaining Redis history: failover or restore can roll it back. Use a consensus-backed authority if election must remain exclusive across those events. See coordination and fencing for the full guarantees.

Run the same samples locally

Docker is required. From the repository root:

& scripts/Invoke-AgentDotNet.ps1 -SingleNode -DotNetArguments @(
    'build', 'tests/Respire.IntegrationTests', '-c', 'Release', '-f', 'net10.0')
& scripts/Invoke-AgentDotNet.ps1 -DotNetArguments @(
    'run', '--project', 'tests/Respire.IntegrationTests', '-c', 'Release', '-f', 'net10.0',
    '--no-build', '--', '--treenode-filter', '/*/*/RecipeIntegrationTests/*',
    '--minimum-expected-tests', '14')

Testcontainers starts and removes its own Redis 8.4 instance. Tests cover score updates and metadata, independent session refresh and expiry, failed outbox publication and recovery, competing leaders and release after failure, and both environment variable names.