Almost every backend I have built needed some version of “do this later”. Send a reminder tomorrow morning. Retry a failed webhook in five minutes. Close an unpaid order after an hour. The usual answer is another moving part: a cron table, a scheduler service, or a polling loop that wakes up every second and asks the database what is due.
If you already run Ruvio for queues or caching, you can skip that part. There are two ways to schedule work, and which one fits depends on whether you want consumer groups or the simplest thing that works.
- DELAY
- one option on XADD; the entry is stored at once and delivered later
- 10,000
- due jobs ZMOVEDUE can hand to a list queue in one call
- 0
- extra WAL records when you do not use a delay
Option 1: a stream entry that waits
Streams with consumer groups are the solid way to run workers: every entry is delivered to one consumer, unacknowledged entries stay pending, and a crashed worker’s entries can be claimed by another one. Ruvio adds one option to XADD:
XGROUP CREATE jobs workers $ MKSTREAM
XADD jobs DELAY 30000 * kind send-reminder user 42
# 1793295045919-0 -> stored now
XREADGROUP GROUP workers w1 COUNT 10 BLOCK 5000 STREAMS jobs >
# (nil) until the 30 s have passed
XACK jobs workers 1793295045919-0use 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);
}The entry is written right away, so it is in the log and survives a restart. XREADGROUP simply skips it until the delay has passed; after that it is delivered like any other entry. Pending lists, XCLAIM and XAUTOCLAIM work as usual, so your retry and dead-letter logic does not change. A plain XREAD still sees the entry, which is handy when you want to look at what is scheduled.
The worker side has nothing special in it: pick your language in the tabs above.
Option 2: a sorted set as a timetable
Sometimes a stream is more than you need. Put each job in a sorted set with its run time as the score, and take what is due:
ZADD {jobs}:due 1700000300 job:9 1700000360 job:10
ZPOPBYSCORE {jobs}:due 1700000330 COUNT 10
# job:9 -> due and removed in the same callZPOPBYSCORE returns the members whose score is at or below the limit and removes them in the same step. Two workers calling it at the same moment never get the same job, which is the part that used to need a script.
If your workers already read from a list, ZMOVEDUE moves the due jobs onto it for you:
ZMOVEDUE {jobs}:due {jobs}:ready 1700000400 COUNT 100
# 1 -> one job moved to the tail of {jobs}:readyIt is one recorded step, up to 10,000 members per call. Run it from a small loop, or from any worker before it blocks on the list.
Which one to pick
Stream with DELAY | Sorted set | |
|---|---|---|
| Delivery | one consumer per entry, with acks | whoever pops first |
| Crashed worker | entry stays pending, XAUTOCLAIM it | gone once popped, add your own retry |
| Payload | fields on the entry | member name, details in another key |
| Change the time later | no, add a new entry | yes, ZADD the new score |
| Best for | reminders, retries, anything that must not be lost | timetables, rescheduling, simple queues |
My own rule: if losing a job would be a bug report, use the stream. If jobs get rescheduled a lot, or the “job” is just an id you look up anyway, the sorted set is simpler and easier to inspect.
Things worth knowing
- One time unit
DELAYtakes milliseconds; sorted-set scores take whatever you choose. I use Unix milliseconds everywhere. - Trimming counts themA delayed entry is part of the stream’s length, so
MAXLENcan remove it before it is delivered. - Ordinary writesWith
everysecoralwaysdurability both options are in the WAL and replicate like everything else.
If you also want to see jobs as they are written, without polling, CHANGES SUBSCRIBE MATCH jobs* gives you a resumable feed of every write when the server runs with a WAL. That is a story for another post. The full syntax for everything here is in the command reference and on the use cases page.