Coordination in one command

Use cases

Ruvio speaks RESP, so any Redis client can talk to it. What sets it apart is a set of commands that finish a coordination job on the shard that owns the key, in one call. Below are the jobs people reach for, with the commands as you would type them in redis-cli and a link to the reference.

Rate limits and quotas

A fixed window per API key, per user, or per tenant, in one command on the shard that owns the key.

LIMIT key max window_ms returns the remaining allowance, or a null bulk when the window is full. A rejected call is not counted. The window is the key's deadline, so a quiet key disappears by itself. In WAL-backed modes the accepted call is a log record.

Keep in mindFixed window, not sliding. If you need per-second smoothing on top, keep a short second window next to the long one.

ReferenceLIMIT

.NET packageRuvio.AspNetCore.RateLimiting

LIMIT api:key-7 100 60000
# 99    -> 99 calls left in this 60 s window
# (nil) -> window is full, nothing was counted

use ruvio_client::Client;

let mut db = Client::connect("127.0.0.1", 6379)?;
let left = db.limit("api:key-7", 100, 60_000)?;
// Some(99) -> 99 calls left in this 60 s window
// None     -> window is full, nothing was counted

await using var connection = await RuvioConnection.ConnectAsync("127.0.0.1", 6379);
var db = await connection.GetDefaultDatabaseAsync();

long? left = await db.LimitAsync("api:key-7", 100, 60_000);
// 99   -> 99 calls left in this 60 s window
// null -> window is full, nothing was counted

c, err := ruvio.Connect(ctx, ruvio.Options{Addr: "127.0.0.1:6379"})
if err != nil { return err }

left, ok, err := c.Limit(ctx, "api:key-7", 100, time.Minute)
if err != nil { return err }
fmt.Println(left, ok)
// 99 true -> 99 calls left in this 60 s window
// 0 false -> window is full, nothing was counted

var c = RuvioClient.connect("127.0.0.1", 6379);

Long left = c.limit("api:key-7", 100, 60_000);
// 99   -> 99 calls left in this 60 s window
// null -> window is full, nothing was counted

import ruvio

r = ruvio.Ruvio("127.0.0.1", 6379)
left = r.limit("api:key-7", 100, 60_000)
# 99   -> 99 calls left in this 60 s window
# None -> window is full, nothing was counted

import { createClient } from 'ruvio-client';

const c = createClient({ host: '127.0.0.1', port: 6379 });
const left = await c.limit('api:key-7', 100, 60_000);
// 99   -> 99 calls left in this 60 s window
// null -> window is full, nothing was counted

builder.Services.AddRuvioClient(builder.Configuration.GetSection("Ruvio"));
builder.Services.AddRuvioRateLimiting(options =>
{
    options.KeyPrefix = "orders:rate:v1:";
    options.PartitionKey = context =>
        context.User.FindFirst("tenant_id")?.Value
        ?? throw new InvalidOperationException("An authenticated tenant is required.");
});

app.UseRouting();
app.UseAuthentication();
app.UseAuthorization();
app.UseRuvioRateLimiting();
app.MapPost("/orders", HandleOrder)
    .RequireAuthorization()
    .WithMetadata(new RuvioRateLimitAttribute("create-order", permits: 100, windowSeconds: 60));

Locks, leader election and fencing

One worker at a time holds a named resource; others get a null reply until it frees or expires.

Each grant returns a token that grows. Send it with writes to the thing you protect and have that side reject older tokens. Expiry and RELEASE keep the sequence while the key exists.

Keep in mindDeleting the key restarts the token at 1, and data loss can do the same. A token alone does not fence an external system after that.

ReferenceLEASE and RELEASE

.NET packageRuvio.AspNetCore.DistributedLock

LEASE job:nightly-report 30000
# 17    -> you hold it, token 17
# (nil) -> someone else holds it
RELEASE job:nightly-report 17
# 1     -> released (0 if the token no longer matches)

use ruvio_client::Client;

let mut db = Client::connect("127.0.0.1", 6379)?;
let token = db.lease("job:nightly-report", 30_000)?;
// Some(17) -> you hold it, token 17
// None     -> someone else holds it
if let Some(token) = token {
    let released = db.release("job:nightly-report", token as u64)?;
    // 1 -> released (0 if the token no longer matches)
}

await using var connection = await RuvioConnection.ConnectAsync("127.0.0.1", 6379);
var db = await connection.GetDefaultDatabaseAsync();

long? token = await db.LeaseAsync("job:nightly-report", 30_000);
// 17   -> you hold it, token 17
// null -> someone else holds it
if (token is long held)
{
    long released = await db.ReleaseAsync("job:nightly-report", held);
    // 1 -> released (0 if the token no longer matches)
}

c, err := ruvio.Connect(ctx, ruvio.Options{Addr: "127.0.0.1:6379"})
if err != nil { return err }

token, ok, err := c.Lease(ctx, "job:nightly-report", 30*time.Second)
if err != nil { return err }
if !ok { return nil } // someone else holds it
// you hold it, token 17
released, err := c.Release(ctx, "job:nightly-report", token)
if err != nil { return err }
fmt.Println(released) // true -> released (false if the token no longer matches)

var c = RuvioClient.connect("127.0.0.1", 6379);

Long token = c.lease("job:nightly-report", 30_000);
// 17   -> you hold it, token 17
// null -> someone else holds it
if (token != null) {
    Boolean released = c.release("job:nightly-report", token);
    // true -> released (false if the token no longer matches)
}

import ruvio

r = ruvio.Ruvio("127.0.0.1", 6379)
token = r.lease("job:nightly-report", 30_000)
# 17   -> you hold it, token 17
# None -> someone else holds it
if token is not None:
    released = r.release("job:nightly-report", token)
    # True -> released (False if the token no longer matches)

import { createClient } from 'ruvio-client';

const c = createClient({ host: '127.0.0.1', port: 6379 });
const token = await c.lease('job:nightly-report', 30_000);
// 17   -> you hold it, token 17
// null -> someone else holds it
if (token !== null) {
  const released = await c.release('job:nightly-report', token);
  // true -> released (false if the token no longer matches)
}

services.AddRuvioDistributedLock(options => options.KeyPrefix = "myapp:lock:");

await using IRuvioLock? held = await locks.TryAcquireAsync(
    "nightly-report", lease: TimeSpan.FromSeconds(30), wait: TimeSpan.FromSeconds(5));

if (held is null) return;               // someone else runs it

await RunReportAsync(held.LockLost);    // stops if the lock is lost mid-run

// Or guard an endpoint: one request per order at a time, across every instance.
app.UseRuvioLock();
app.MapPost("/orders/{id}/pay", PayAsync).WithMetadata(new RuvioLockAttribute("order:{id}"));

Concurrency limits for workers

At most N holders at once: image resizers, export jobs, calls to a slow upstream.

Every holder has its own TTL, so a crashed worker frees its slot when the TTL ends. The key is removed when the last holder leaves.

Keep in mindPick a TTL longer than the longest normal job, and release as soon as you finish.

ReferenceSEMAPHORE

SEMAPHORE upstream:billing 8 15000
# 1700000000123 -> a slot, this is your token
# (nil)         -> all 8 slots are taken
RELEASE upstream:billing 1700000000123

use ruvio_client::Client;

let mut db = Client::connect("127.0.0.1", 6379)?;
let token = db.semaphore("upstream:billing", 8, 15_000)?;
// Some(1700000000123) -> a slot, this is your token
// None                -> all 8 slots are taken
if let Some(token) = token {
    db.release("upstream:billing", token as u64)?;
}

await using var connection = await RuvioConnection.ConnectAsync("127.0.0.1", 6379);
var db = await connection.GetDefaultDatabaseAsync();

long? token = await db.AcquireSemaphoreAsync("upstream:billing", 8, 15_000);
// 1700000000123 -> a slot, this is your token
// null          -> all 8 slots are taken
if (token is long held)
{
    await db.ReleaseAsync("upstream:billing", held);
}

c, err := ruvio.Connect(ctx, ruvio.Options{Addr: "127.0.0.1:6379"})
if err != nil { return err }

token, ok, err := c.Semaphore(ctx, "upstream:billing", 8, 15*time.Second)
if err != nil { return err }
if !ok { return nil } // all 8 slots are taken
// a slot, token is 1700000000123
if _, err := c.Release(ctx, "upstream:billing", token); err != nil { return err }

var c = RuvioClient.connect("127.0.0.1", 6379);

Long token = c.semaphore("upstream:billing", 8, 15_000);
// 1700000000123 -> a slot, this is your token
// null          -> all 8 slots are taken
if (token != null) {
    c.release("upstream:billing", token);
}

import ruvio

r = ruvio.Ruvio("127.0.0.1", 6379)
token = r.semaphore("upstream:billing", 8, 15_000)
# 1700000000123 -> a slot, this is your token
# None          -> all 8 slots are taken
if token is not None:
    r.release("upstream:billing", token)

import { createClient } from 'ruvio-client';

const c = createClient({ host: '127.0.0.1', port: 6379 });
const token = await c.semaphore('upstream:billing', 8, 15_000);
// '1700000000123' -> a slot, this is your token (a decimal string)
// null            -> all 8 slots are taken
if (token !== null) {
  await c.release('upstream:billing', token);
}

Inventory, tickets and seat counts

Sell from a counter without overselling, with no script and no read-then-write race.

TAKE subtracts only when the key holds enough. INCRBY … MAX adds only while the result stays at or below the cap. Both answer with a null bulk and write nothing when they refuse.

Keep in mindA refused call writes nothing, neither to the key nor to the WAL.

ReferenceTAKE INCRBY MAX

SET stock:ticket-42 500
TAKE stock:ticket-42 2
# 498   -> took 2, 498 left
# (nil) -> not enough, the key is unchanged

INCRBY seats:room-9 1 MAX 120
# 57    -> joined
# (nil) -> room is full, nothing was written

use ruvio_client::Client;

let mut db = Client::connect("127.0.0.1", 6379)?;
db.set("stock:ticket-42", "500")?;
let left = db.take("stock:ticket-42", 2)?;
// Some(498) -> took 2, 498 left
// None      -> not enough, the key is unchanged

let seats = db.incr_by_max("seats:room-9", 1, 120)?;
// Some(57) -> joined
// None     -> room is full, nothing was written

await using var connection = await RuvioConnection.ConnectAsync("127.0.0.1", 6379);
var db = await connection.GetDefaultDatabaseAsync();

await db.SetAsync("stock:ticket-42", "500");
long? left = await db.TakeAsync("stock:ticket-42", 2);
// 498  -> took 2, 498 left
// null -> not enough, the key is unchanged

long? seats = await db.IncrementWithLimitAsync("seats:room-9", 1, 120);
// 57   -> joined
// null -> room is full, nothing was written

c, err := ruvio.Connect(ctx, ruvio.Options{Addr: "127.0.0.1:6379"})
if err != nil { return err }

if err := c.Set(ctx, "stock:ticket-42", "500"); err != nil { return err }
left, took, err := c.Take(ctx, "stock:ticket-42", 2)
if err != nil { return err }
fmt.Println(left, took) // 498 true -> took 2; 0 false -> not enough, the key is unchanged

seats, joined, err := c.IncrByMax(ctx, "seats:room-9", 1, 120)
if err != nil { return err }
fmt.Println(seats, joined) // 57 true -> joined; 0 false -> room is full, nothing was written

var c = RuvioClient.connect("127.0.0.1", 6379);

c.set("stock:ticket-42", "500");
Long left = c.take("stock:ticket-42", 2);
// 498  -> took 2, 498 left
// null -> not enough, the key is unchanged

Long seats = c.incrbyMax("seats:room-9", 1, 120);
// 57   -> joined
// null -> room is full, nothing was written

import ruvio

r = ruvio.Ruvio("127.0.0.1", 6379)
r.set("stock:ticket-42", 500)
left = r.take("stock:ticket-42", 2)
# 498  -> took 2, 498 left
# None -> not enough, the key is unchanged

seats = r.incrby("seats:room-9", 1, max=120)
# 57   -> joined
# None -> room is full, nothing was written

import { createClient } from 'ruvio-client';

const c = createClient({ host: '127.0.0.1', port: 6379 });
await c.set('stock:ticket-42', 500);
const left = await c.take('stock:ticket-42', 2);
// 498  -> took 2, 498 left
// null -> not enough, the key is unchanged

const seats = await c.incrbyMax('seats:room-9', 1, 120);
// 57   -> joined
// null -> room is full, nothing was written

Idempotent APIs and payments

A retry of the same request replays the stored answer instead of running the handler again.

BEGIN answers acquired, pending, conflict (same key, different fingerprint), or completed with the saved result. For plain once-only work such as a webhook delivery id, ONCE key ttl_ms is the lighter tool: 1 the first time, 0 until the deadline.

Keep in mindThis does not make an outside effect exactly-once. A handler can crash before COMPLETE, so the payment provider still needs its own idempotency key.

ReferenceIDEM ONCE

.NET packageRuvio.AspNetCore.Idempotency

IDEM order:42 BEGIN request-digest attempt-1 60000
# ["acquired"]                 -> run the handler once
IDEM order:42 COMPLETE request-digest attempt-1 saved-result
# 1
IDEM order:42 BEGIN request-digest attempt-2 60000
# ["completed","saved-result"]

use std::time::Duration;
use ruvio_client::Client;

let mut db = Client::connect("127.0.0.1", 6379)?;
let first = db.idem_begin("order:42", "request-digest", "attempt-1", Duration::from_secs(60))?;
// IdemState::Acquired -> run the handler once
db.idem_complete("order:42", "request-digest", "attempt-1", b"saved-result")?; // true
let retry = db.idem_begin("order:42", "request-digest", "attempt-2", Duration::from_secs(60))?;
// IdemState::Completed(b"saved-result".to_vec())

await using var connection = await RuvioConnection.ConnectAsync("127.0.0.1", 6379);
var db = await connection.GetDefaultDatabaseAsync();

var first = await db.TryBeginIdempotencyAsync("order:42", "request-digest", "attempt-1", TimeSpan.FromSeconds(60));
// first.State == IdempotencyState.Acquired -> run the handler once
await db.CompleteIdempotencyAsync("order:42", "request-digest", "attempt-1", "saved-result"); // true
var retry = await db.TryBeginIdempotencyAsync("order:42", "request-digest", "attempt-2", TimeSpan.FromSeconds(60));
// retry.State == IdempotencyState.Completed, retry.Result holds "saved-result" as bytes

c, err := ruvio.Connect(ctx, ruvio.Options{Addr: "127.0.0.1:6379"})
if err != nil { return err }

first, err := c.IdemBegin(ctx, "order:42", "request-digest", "attempt-1", time.Minute)
if err != nil { return err }
fmt.Println(first.State) // acquired -> run the handler once
n, err := c.IdemComplete(ctx, "order:42", "request-digest", "attempt-1", []byte("saved-result"))
if err != nil { return err }
fmt.Println(n) // 1
retry, err := c.IdemBegin(ctx, "order:42", "request-digest", "attempt-2", time.Minute)
if err != nil { return err }
fmt.Println(retry.State, string(retry.Result)) // completed saved-result

var c = RuvioClient.connect("127.0.0.1", 6379);

var first = c.idemBegin("order:42", "request-digest", "attempt-1", 60_000);
// first.state() == IdemResult.State.ACQUIRED -> run the handler once
c.idemComplete("order:42", "request-digest", "attempt-1", "saved-result"); // 1
var retry = c.idemBegin("order:42", "request-digest", "attempt-2", 60_000);
// retry.state() == COMPLETED, retry.resultString() == "saved-result"

import ruvio

r = ruvio.Ruvio("127.0.0.1", 6379)
first = r.idem_begin("order:42", "request-digest", "attempt-1", 60_000)
# IdemResult(state='acquired', result=None) -> run the handler once
r.idem_complete("order:42", "request-digest", "attempt-1", "saved-result")  # 1
retry = r.idem_begin("order:42", "request-digest", "attempt-2", 60_000)
# IdemResult(state='completed', result='saved-result')

import { createClient } from 'ruvio-client';

const c = createClient({ host: '127.0.0.1', port: 6379 });
const first = await c.idemBegin('order:42', 'request-digest', 'attempt-1', 60_000);
// { state: 'acquired', result: null } -> run the handler once
await c.idemComplete('order:42', 'request-digest', 'attempt-1', 'saved-result'); // 1
const retry = await c.idemBegin('order:42', 'request-digest', 'attempt-2', 60_000);
// { state: 'completed', result: 'saved-result' }

builder.Services.AddRuvioClient(builder.Configuration.GetSection("Ruvio"));
builder.Services.AddRuvioIdempotency(options =>
{
    options.Namespace = "orders-production-v1"; // stable across replicas
    options.ScopeSelector = context => context.User.FindFirst("tenant_id")?.Value;
    options.Retention = TimeSpan.FromHours(24);
});

app.UseRouting();
app.UseAuthentication();
app.UseAuthorization(); // must run before idempotency, including on replays
app.UseRuvioIdempotency();
app.MapPost("/orders", HandleOrder)
    .RequireAuthorization()
    .WithMetadata(new RuvioIdempotencyAttribute());

Lost-update protection

Two editors open the same record; only the one holding the current version may save.

Optimistic concurrency without WATCH or a script. Only keys written with SETV carry a version.

Keep in mindA plain SET replaces a versioned string.

ReferenceSETV

SETV profile:ada "{...}"
# 1     -> version 1 created
SETV profile:ada "{...}" 1
# 2     -> accepted, now version 2
SETV profile:ada "{...}" 1
# error -> stale version, value unchanged

use ruvio_client::Client;

let mut db = Client::connect("127.0.0.1", 6379)?;
db.setv("profile:ada", "{...}", None)?;    // 1 -> version 1 created
db.setv("profile:ada", "{...}", Some(1))?; // 2 -> accepted, now version 2
let stale = db.setv("profile:ada", "{...}", Some(1));
// Err(Error::Server("ERR version mismatch")) -> stale version, value unchanged

await using var connection = await RuvioConnection.ConnectAsync("127.0.0.1", 6379);
var db = await connection.GetDefaultDatabaseAsync();

await db.SetVersionAsync("profile:ada", "{...}");    // 1 -> version 1 created
await db.SetVersionAsync("profile:ada", "{...}", 1); // 2 -> accepted, now version 2
try
{
    await db.SetVersionAsync("profile:ada", "{...}", 1);
}
catch (RuvioException)
{
    // stale version, value unchanged
}

c, err := ruvio.Connect(ctx, ruvio.Options{Addr: "127.0.0.1:6379"})
if err != nil { return err }

v, err := c.SetV(ctx, "profile:ada", "{...}", 0) // 1 -> version 1 created (0 sends no version)
if err != nil { return err }
v, err = c.SetV(ctx, "profile:ada", "{...}", 1) // 2 -> accepted, now version 2
if err != nil { return err }
_, err = c.SetV(ctx, "profile:ada", "{...}", 1)
fmt.Println(v, err) // 2 ERR version mismatch -> stale version, value unchanged

var c = RuvioClient.connect("127.0.0.1", 6379);

c.setv("profile:ada", "{...}");    // 1 -> version 1 created
c.setv("profile:ada", "{...}", 1); // 2 -> accepted, now version 2
try {
    c.setv("profile:ada", "{...}", 1);
} catch (RuvioServerException e) {
    // stale version, value unchanged
}

import ruvio

r = ruvio.Ruvio("127.0.0.1", 6379)
r.setv("profile:ada", "{...}")     # 1 -> version 1 created
r.setv("profile:ada", "{...}", 1)  # 2 -> accepted, now version 2
try:
    r.setv("profile:ada", "{...}", 1)
except ruvio.ResponseError:
    pass  # stale version, value unchanged

import { createClient } from 'ruvio-client';

const c = createClient({ host: '127.0.0.1', port: 6379 });
await c.setv('profile:ada', '{...}');    // 1 -> version 1 created
await c.setv('profile:ada', '{...}', 1); // 2 -> accepted, now version 2
try {
  await c.setv('profile:ada', '{...}', 1);
} catch (err) {
  // RuvioError: stale version, value unchanged
}

Sessions and sliding expiry

Reading a session keeps it alive; an idle one expires on its own.

GETEX returns the string and moves its deadline in one step, so there is no separate EXPIRE call. GET does not move a deadline.

ReferenceGETEX

.NET packagesRuvio.Extensions.CachingRuvio.AspNetCore.DataProtection

SET session:9f3 "{...}" PX 1800000
GETEX session:9f3 1800000
# "{...}" and the deadline is now 30 min from now

use std::time::Duration;
use ruvio_client::{Client, SetMode};

let mut db = Client::connect("127.0.0.1", 6379)?;
db.set_with("session:9f3", "{...}", Some(Duration::from_millis(1_800_000)), SetMode::Always)?;
let session = db.getex("session:9f3", 1_800_000)?;
// Some("{...}") and the deadline is now 30 min from now

await using var connection = await RuvioConnection.ConnectAsync("127.0.0.1", 6379);
var db = await connection.GetDefaultDatabaseAsync();

await db.SetAsync("session:9f3", "{...}", TimeSpan.FromMilliseconds(1_800_000));
string? session = await db.GetExpireAsync("session:9f3", 1_800_000);
// "{...}" and the deadline is now 30 min from now

c, err := ruvio.Connect(ctx, ruvio.Options{Addr: "127.0.0.1:6379"})
if err != nil { return err }

if _, err := c.SetArgs(ctx, "session:9f3", "{...}", ruvio.SetOptions{TTL: 30 * time.Minute}); err != nil { return err }
session, ok, err := c.GetEx(ctx, "session:9f3", 30*time.Minute)
if err != nil { return err }
fmt.Println(session, ok) // {...} true, and the deadline is now 30 min from now

var c = RuvioClient.connect("127.0.0.1", 6379);

c.set("session:9f3", "{...}", SetOptions.px(1_800_000));
String session = c.getex("session:9f3", 1_800_000);
// "{...}" and the deadline is now 30 min from now

import ruvio

r = ruvio.Ruvio("127.0.0.1", 6379)
r.set("session:9f3", "{...}", px=1_800_000)
session = r.getex("session:9f3", 1_800_000)
# '{...}' and the deadline is now 30 min from now

import { createClient } from 'ruvio-client';

const c = createClient({ host: '127.0.0.1', port: 6379 });
await c.set('session:9f3', '{...}', { px: 1_800_000 });
const session = await c.getex('session:9f3', 1_800_000);
// '{...}' and the deadline is now 30 min from now

builder.Services.AddRuvioDistributedCache(o => o.InstanceName = "orders-session-v1");
builder.Services.AddSession(o => o.IdleTimeout = TimeSpan.FromMinutes(20));

var app = builder.Build();
app.UseRouting();
app.UseSession();
app.MapGet("/visits", async (HttpContext context) =>
{
    await context.Session.LoadAsync(context.RequestAborted);
    var visits = (context.Session.GetInt32("visits") ?? 0) + 1;
    context.Session.SetInt32("visits", visits);
    return visits;
});

Cache fills without stampedes

When a hot key expires, one caller refills it and the others wait or serve the stale copy.

One round trip decides who loads from the database. WAIT holds a caller that would get BUSY until the filler writes, so there is no retry loop in the client.

Keep in mindFill markers live in memory only and lapse after fill_ms. WAIT is refused inside MULTI and scripts.

ReferenceFILLGET and FILLSET

.NET packagesRuvio.Extensions.CachingRuvio.AspNetCore.OutputCaching

FILLGET product:42 5000 WAIT 200
# [HIT, value]          -> served from memory
# [FILL, token]         -> you refill it
# [BUSY, nil]           -> someone else is filling
# [STALE, value, token] -> stale copy, and you refresh it (SWR window)
# [STALE, value, nil]   -> stale copy, another caller is refreshing
FILLSET product:42 "{...}" 60000 <token> SWR 30000
# 1 -> stored (0 if the token is no longer live)

use std::time::Duration;
use ruvio_client::{Client, FillOutcome};

let mut db = Client::connect("127.0.0.1", 6379)?;
let outcome = db.fill_get_wait("product:42", Duration::from_millis(5000), Duration::from_millis(200))?;
// Hit(value)  -> served from memory      Fill(token)            -> you refill it
// Busy        -> someone else is filling Stale { value, token } -> inside a stale-while-revalidate window
if let FillOutcome::Fill(token) | FillOutcome::Stale { token: Some(token), .. } = outcome {
    let stored = db.fill_set("product:42", "{...}", Duration::from_secs(60), &token, Some(Duration::from_secs(30)))?;
    // true -> stored (false if the token is no longer live)
}

await using var connection = await RuvioConnection.ConnectAsync("127.0.0.1", 6379);
var db = await connection.GetDefaultDatabaseAsync();

FillResult fill = await db.FillGetAsync("product:42", TimeSpan.FromMilliseconds(5000), TimeSpan.FromMilliseconds(200));
// fill.State: Hit (fill.Value), Fill (fill.Token), Busy, or Stale (fill.Value, and fill.Token when you refresh)
if (fill.Token is long token)
{
    bool stored = await db.FillSetAsync("product:42", "{...}", TimeSpan.FromSeconds(60), token, TimeSpan.FromSeconds(30));
    // true -> stored (false if the token is no longer live)
}

c, err := ruvio.Connect(ctx, ruvio.Options{Addr: "127.0.0.1:6379"})
if err != nil { return err }

f, err := c.FillGetWait(ctx, "product:42", 5*time.Second, 200*time.Millisecond)
if err != nil { return err }
// f.State: ruvio.FillHit (f.Value), FillFill (f.Token), FillBusy, or FillStale (f.Value, f.Token)
if f.Token != "" { // you refill it
    stored, err := c.FillSet(ctx, "product:42", "{...}", time.Minute, f.Token, 30*time.Second)
    if err != nil { return err }
    fmt.Println(stored) // true -> stored (false if the token is no longer live)
}

var c = RuvioClient.connect("127.0.0.1", 6379);

var fill = c.fillgetWait("product:42", 5000, 200);
// fill.state(): HIT (fill.value()), FILL (fill.token()), BUSY, or STALE (fill.value(), fill.token())
if (fill.token() != null) { // you refill it
    Boolean stored = c.fillset("product:42", "{...}", 60_000, fill.token(), 30_000);
    // true -> stored (false if the token is no longer live)
}

import ruvio

r = ruvio.Ruvio("127.0.0.1", 6379)
fill = r.fillget("product:42", 5000, wait_ms=200)
# FillResult(state, value, token): state is HIT, FILL, BUSY or STALE
if fill.token is not None:  # you refill it
    stored = r.fillset("product:42", "{...}", 60_000, fill.token, swr_ms=30_000)
    # True -> stored (False if the token is no longer live)

import { createClient } from 'ruvio-client';

const c = createClient({ host: '127.0.0.1', port: 6379 });
const fill = await c.fillget('product:42', 5000, 200);
// { state: 'hit' | 'fill' | 'busy' | 'stale', value, token }
if (fill.token !== null) { // you refill it
  const stored = await c.fillset('product:42', '{...}', 60_000, fill.token, 30_000);
  // true -> stored (false if the token is no longer live)
}

builder.Services.AddRuvioClient(builder.Configuration.GetSection("Ruvio"));
builder.Services.AddRuvioDistributedCache(options => options.InstanceName = "orders");

// One caller across every instance runs the factory; the others receive its result.
byte[] page = await cache.GetOrCreateAsync(
    "invoice:42",
    async token => await RenderAsync(token),
    new DistributedCacheEntryOptions { AbsoluteExpirationRelativeToNow = TimeSpan.FromMinutes(10) });

Delayed jobs and queues

Schedule work for later with a stream and a consumer group, without a separate scheduler.

The entry is stored right away. XREADGROUP skips it until the delay passes, and XREAD can still see it. Pending entries, XCLAIM and XAUTOCLAIM work as in Redis streams.

Keep in mindThe hold is a WAL record and is written only when a delay is set.

ReferenceXADD DELAY

XGROUP CREATE jobs workers $ MKSTREAM
XADD jobs DELAY 30000 * kind send-reminder user 42
XREADGROUP GROUP workers w1 COUNT 10 BLOCK 5000 STREAMS jobs >
# nil until the 30 s delay has passed; the worker keeps reading in a loop
XACK jobs workers <id>

use std::time::Duration;
use ruvio_client::{Client, StreamReadOptions};

let mut db = Client::connect("127.0.0.1", 6379)?;
db.xgroup_create("jobs", "workers", "$", true)?; // true adds MKSTREAM
db.xadd_delay("jobs", 30_000, "*", &[("kind", "send-reminder"), ("user", "42")])?;
let options = StreamReadOptions { count: Some(10), block: Some(Duration::from_millis(5000)), ..Default::default() };
for stream in db.xreadgroup("workers", "w1", &options, &[("jobs", ">")])? {
    for entry in stream.entries { // empty until the 30 s delay has passed; read in a loop
        db.xack("jobs", "workers", &[&entry.id])?;
    }
}

await using var connection = await RuvioConnection.ConnectAsync("127.0.0.1", 6379);
var db = await connection.GetDefaultDatabaseAsync();

await db.CreateStreamGroupAsync("jobs", "workers", "$", true); // true adds MKSTREAM
await db.AddStreamEntryDelayedAsync("jobs", 30_000, "*", ("kind", "send-reminder"), ("user", "42"));
var read = await db.GetStreamGroupUpdatesAsync("workers", "w1", "jobs", ">",
    new StreamReadOptions { Count = 10, Block = TimeSpan.FromMilliseconds(5000) });
foreach (var entry in read.SelectMany(stream => stream.Entries)) // empty until the 30 s delay has passed; read in a loop
{
    await db.AcknowledgeStreamAsync("jobs", "workers", entry.Id);
}

c, err := ruvio.Connect(ctx, ruvio.Options{Addr: "127.0.0.1:6379"})
if err != nil { return err }

if err := c.XGroupCreate(ctx, "jobs", "workers", "$", true); err != nil { return err } // true adds MKSTREAM
_, _, err = c.XAdd(ctx, ruvio.XAddArgs{Stream: "jobs", Delay: 30 * time.Second, ID: "*",
    Values: []string{"kind", "send-reminder", "user", "42"}})
if err != nil { return err }
msgs, err := c.XReadGroup(ctx, ruvio.XReadGroupArgs{Group: "workers", Consumer: "w1", Stream: "jobs",
    Count: 10, Block: 5 * time.Second})
if err != nil { return err }
for _, m := range msgs { // empty until the 30 s delay has passed; read in a loop
    if _, err := c.XAck(ctx, "jobs", "workers", m.ID); err != nil { return err }
}

var c = RuvioClient.connect("127.0.0.1", 6379);

c.xgroupCreate("jobs", "workers", "$", true); // true adds MKSTREAM
c.xadd("jobs", XAddArgs.builder().delay(30_000), Map.of("kind", "send-reminder", "user", "42"));
var read = c.xreadgroup("workers", "w1", XReadArgs.builder().count(10).block(5000), "jobs", ">");
for (var entries : read.values()) {
    for (var entry : entries) { // empty until the 30 s delay has passed; read in a loop
        c.xack("jobs", "workers", entry.id());
    }
}

import ruvio

r = ruvio.Ruvio("127.0.0.1", 6379)
r.xgroup_create("jobs", "workers", "$", mkstream=True)
r.xadd("jobs", {"kind": "send-reminder", "user": 42}, delay_ms=30_000)
read = r.xreadgroup("workers", "w1", {"jobs": ">"}, count=10, block=5000)
for _stream, entries in read or []:
    for entry in entries:  # the entry appears here after 30 s
        r.xack("jobs", "workers", entry.id)

import { createClient } from 'ruvio-client';

const c = createClient({ host: '127.0.0.1', port: 6379 });
await c.xgroupCreate('jobs', 'workers', '$', { mkStream: true });
await c.xadd('jobs', '*', { kind: 'send-reminder', user: 42 }, { delayMs: 30_000 });
const read = await c.xreadgroup('workers', 'w1', 'jobs', { count: 10, block: 5000 });
for (const entry of read?.[0]?.entries ?? []) { // empty until the 30 s delay has passed; read in a loop
  await c.xack('jobs', 'workers', entry.id);
}

Change feeds

React to writes (cache invalidation, search indexing, audit) by reading a feed instead of polling.

The durable feed is built from the WAL and can resume from a saved position, with optional key patterns, types and values. CHANGES START is the lighter in-memory feed of up to 1024 notes per shard.

Keep in mindThe stream needs a durability profile with a WAL and is refused on a replica. It takes over its connection, so give it a connection of its own. Limits and retention are under Change stream in the configuration reference.

ReferenceCHANGES Change stream settings

CHANGES SUBSCRIBE FROM NOW MATCH user:* TYPES put,del
# a durable stream of changes, resumable with FROM <shard>:<seq>

use ruvio_client::Client;

let mut db = Client::connect("127.0.0.1", 6379)?;
// No typed method for CHANGES SUBSCRIBE: execute sends it, read_message reads each event.
let start = db.execute(&["CHANGES", "SUBSCRIBE", "FROM", "NOW", "MATCH", "user:*", "TYPES", "put,del"])?;
// start: the cursor the stream begins after, "<shard>:<seq>"
loop {
    let event = db.read_message()?;
    // ["change", cursor, db, type, key, value]; save the cursor to resume with FROM
}

await using var connection = await RuvioConnection.ConnectAsync("127.0.0.1", 6379);
var db = await connection.GetDefaultDatabaseAsync();

// Ruvio.Client cannot read the CHANGES SUBSCRIBE push stream, so this polls the lighter
// in-memory feed from CHANGES START instead (this shard only, no MATCH or TYPES filter).
long cursor = await db.StartChangesAsync();
while (true)
{
    foreach (ChangeNote note in await db.ReadChangesAsync(cursor))
    {
        cursor = note.Sequence; // note.Text names the command and key, such as "SET user:7"
    }
    await Task.Delay(TimeSpan.FromSeconds(1));
}

c, err := ruvio.Connect(ctx, ruvio.Options{Addr: "127.0.0.1:6379", Protocol: 3,
    OnPush: func(event ruvio.Value) {
        // ["change", cursor, db, type, key, value]; save the cursor to resume with FROM
    }})
if err != nil { return err }

// No typed method for CHANGES SUBSCRIBE: Do sends it, and the events arrive in OnPush.
start, err := c.Do(ctx, "CHANGES", "SUBSCRIBE", "FROM", "NOW", "MATCH", "user:*", "TYPES", "put,del")
if err != nil { return err }
fmt.Println(start.Str()) // the cursor the stream begins after, "<shard>:<seq>"

var c = RuvioClient.connect(RuvioOptions.builder().host("127.0.0.1").port(6379)
    .pushListener(event -> {
        // ["change", cursor, db, type, key, value]; save the cursor to resume with FROM
    })
    .build());
c.hello(3); // events arrive as RESP3 pushes

// No typed method for CHANGES SUBSCRIBE: executeText sends it.
Object start = c.executeText("CHANGES", "SUBSCRIBE", "FROM", "NOW", "MATCH", "user:*", "TYPES", "put,del");
// the cursor the stream begins after, "<shard>:<seq>"

from ruvio.asyncio import Ruvio

def on_change(event):
    ...  # ["change", cursor, db, type, key, value]; save the cursor to resume with FROM

r = Ruvio("127.0.0.1", 6379, on_push=on_change)
await r.hello(3)  # events arrive as RESP3 pushes
# No typed method for CHANGES SUBSCRIBE: execute_command sends it.
start = await r.execute_command("CHANGES", "SUBSCRIBE", "FROM", "NOW", "MATCH", "user:*", "TYPES", "put,del")
# the cursor the stream begins after, "<shard>:<seq>"

import { createClient } from 'ruvio-client';

const c = createClient({ host: '127.0.0.1', port: 6379, protocol: 3 });
c.on('push', (event) => {
  // ['change', cursor, db, type, key, value]; save the cursor to resume with FROM
});
// No typed method for CHANGES SUBSCRIBE: call sends it.
const start = await c.call('CHANGES', 'SUBSCRIBE', 'FROM', 'NOW', 'MATCH', 'user:*', 'TYPES', 'put,del');
// the cursor the stream begins after, '<shard>:<seq>'

Leaderboards, geo and documents

The data types you already use, plus JSON documents with an exact-match index.

Strings, lists, sets, hashes, sorted sets, streams, geo, Bloom filters and JSON are covered. JSON.INDEX is an exact-equality index on one scalar field; it works on database 0 and keeps its postings in memory.

ReferenceData types Commands

ZADD leaderboard 1200 ada 980 grace
ZRANGE leaderboard 0 9 REV WITHSCORES
# cities was filled earlier with GEOADD
GEOSEARCH cities FROMLONLAT 13.36 38.11 BYRADIUS 50 km COUNT 10
JSON.SET user:7 $ '{"plan":"pro","country":"TR"}'
JSON.INDEX CREATE by_plan PREFIX user: PATH $.plan TYPE string
JSON.FIND by_plan VALUE pro COUNT 20
# ERR index is building -> the first build runs in the background; retry

use ruvio_client::{Client, GeoFrom, GeoSearch, JsonIndexType};

let mut db = Client::connect("127.0.0.1", 6379)?;
db.zadd("leaderboard", &[("ada", 1200.0), ("grace", 980.0)])?;
let top = db.zrevrange_with_scores("leaderboard", 0, 9)?; // highest score first
let search = GeoSearch { count: Some(10), ..GeoSearch::radius(GeoFrom::LonLat(13.36, 38.11), 50.0, "km") };
let near = db.geo_search_with("cities", &search)?;
db.json_set("user:7", "$", r#"{"plan":"pro","country":"TR"}"#)?;
db.json_index_create("by_plan", "user:", "$.plan", Some(JsonIndexType::String))?;
let pro = db.json_find("by_plan", "pro", Some(20), None)?; // ["user:7"]

await using var connection = await RuvioConnection.ConnectAsync("127.0.0.1", 6379);
var db = await connection.GetDefaultDatabaseAsync();

await db.AddSortedSetAsync("leaderboard", ("ada", 1200), ("grace", 980));
var top = await db.GetSortedSetRangeByRankWithScoresAsync("leaderboard", 0, 9, descending: true);
var near = (await db.GeoSearchAsync("cities", 13.36, 38.11, 50, "km")).Take(10); // no COUNT argument here
await db.JsonSetAsync("user:7", "$", """{"plan":"pro","country":"TR"}""");
await db.JsonCreateIndexAsync("by_plan", "user:", "$.plan", JsonIndexType.String);
var pro = await db.JsonFindAsync("by_plan", "pro", count: 20); // ["user:7"]

c, err := ruvio.Connect(ctx, ruvio.Options{Addr: "127.0.0.1:6379"})
if err != nil { return err }

if _, err := c.ZAdd(ctx, "leaderboard", ruvio.Z{Member: "ada", Score: 1200}, ruvio.Z{Member: "grace", Score: 980}); err != nil { return err }
top, err := c.ZRangeArgsWithScores(ctx, ruvio.ZRangeArgs{Key: "leaderboard", Start: "0", Stop: "9", Rev: true})
if err != nil { return err }
near, err := c.GeoSearch(ctx, "cities", ruvio.GeoSearchQuery{Longitude: 13.36, Latitude: 38.11, Radius: 50, Unit: "km", Count: 10})
if err != nil { return err }
if err := c.JSONSet(ctx, "user:7", "$", `{"plan":"pro","country":"TR"}`); err != nil { return err }
if err := c.JSONIndexCreate(ctx, "by_plan", "user:", "$.plan", "string"); err != nil { return err }
pro, err := c.JSONFind(ctx, "by_plan", "pro", 20, "")
if err != nil { return err }
fmt.Println(top, near, pro) // pro: [user:7]

var c = RuvioClient.connect("127.0.0.1", 6379);

c.zadd("leaderboard", Map.of("ada", 1200.0, "grace", 980.0));
var top = c.zrangeWithScores("leaderboard", "0", "9", ZRangeArgs.builder().rev());
var near = c.geosearch("cities", GeoSearchArgs.fromLonLat(13.36, 38.11).byRadius(50, GeoUnit.KM).count(10));
c.jsonSet("user:7", "$", "{\"plan\":\"pro\",\"country\":\"TR\"}");
c.jsonIndexCreate("by_plan", "user:", "$.plan", "string");
var pro = c.jsonFind("by_plan", "pro", 20, null); // [user:7]

import ruvio

r = ruvio.Ruvio("127.0.0.1", 6379)
r.zadd("leaderboard", {"ada": 1200, "grace": 980})
top = r.zrange("leaderboard", 0, 9, rev=True, withscores=True)  # [('ada', 1200.0), ('grace', 980.0)]
near = r.geosearch("cities", longitude=13.36, latitude=38.11, radius=50, unit="km", count=10)
r.json_set("user:7", "$", '{"plan":"pro","country":"TR"}')
r.json_index_create("by_plan", "user:", "$.plan", type="string")
pro = r.json_find("by_plan", "pro", count=20)  # ['user:7']

import { createClient } from 'ruvio-client';

const c = createClient({ host: '127.0.0.1', port: 6379 });
await c.zadd('leaderboard', [[1200, 'ada'], [980, 'grace']]);
const top = await c.zrangeWithScores('leaderboard', 0, 9, { rev: true }); // [{ member: 'ada', score: 1200 }, ...]
const near = await c.geosearch('cities', { fromLonLat: [13.36, 38.11], byRadius: { radius: 50, unit: 'km' }, count: 10 });
await c.jsonSet('user:7', '$', '{"plan":"pro","country":"TR"}');
await c.jsonIndexCreate('by_plan', 'user:', '$.plan', 'string');
const pro = await c.jsonFind('by_plan', 'pro', { count: 20 }); // ['user:7']

Moving from Redis

Ruvio is not a full Redis clone. It speaks RESP2 and covers the commands listed in the command reference. Check that the commands you use are there before you switch.

ruvio-ctl import copies keys with their TTLs from a running Redis and can verify the copy. It does not move consumer groups, Lua scripts, ACL users or module types.

Start with Get started, then pick a client. For durability modes and what each one promises, see Configuration and Recovery.

ReferenceImport Compatibility Commands

Try it on your machine

One Docker command or a prebuilt binary, then redis-cli. Or try three of these commands in the browser.

Get started Command reference