redis
Package redis.
A Redis client, written in slang over net like pg: commands waiting on the server park the task on the reactor instead of blocking a worker thread. Errors keep the server's own text through the same result[_, str] story as every other package.
The protocol core is usable now; pooling, cluster routing and pub/sub are all here too. connect dials with a deadline, runs AUTH and SELECT, and hands back a Conn that any task may share -- one round trip at a time, serialized by an internal lock. do runs one command; a reply that violates the protocol breaks the connection for good, while a server-side command error only fails that call. Heavier use goes through Pool (shared connections with idle probing), Cluster (slot routing with MOVED/ASK handling and same-slot checks), explicit MULTI transactions on direct Conns, and pull-model pub/sub where the caller drives sub_next (a 4-line pump feeds a chan when push suits better).
import "redis";
import "time";
import "log";
let dl = until_of(time.mono() + 5000000000);
let cr = redis.connect("redis://:secret@cache.internal:6380/2", dl);
guard let c = cr else let e = err_of(cr) {
log.error("connect: " + e);
exit(1);
}
// encode a command to wire bytes (binary-safe values pass through)
let wire: bytes = redis.encode([to_bytes("SET"), to_bytes("k"),
to_bytes("v")]);
// run it; at most one command is ever in flight per Conn
let r = redis.do(c, [to_bytes("PING")], dl);
// decode one reply; ok(none) means feed more bytes and retry,
// err means the bytes violate the protocol
let dr: result[opt[redis.Decoded], str] = redis.decode(wire);
// cluster hash slot of a key (0..16383), honouring {...} hash tags
let s: int = redis.slot("{user1000}.following");
// connection strings parse into Configs for pooling and clusters
let ur = redis.parse_url("redis://alice:secret@cache.internal:6380/2");
// pooled and clustered use underneath (see the table)
let pr = redis.new_pool("redis://cache.internal:6380", 8);
redis.close(c);| Function | Signature |
|---|---|
redis.encode(args) | bytes — one command as a RESP2 array of bulk strings |
redis.decode(buf) / redis.decode_at(buf, pos) | result[opt[Decoded], str] — one reply plus bytes consumed |
redis.slot(key) | int — cluster hash slot with {...} tag support |
redis.parse_url(url) | result[Config, str] — redis:// / rediss:// |
redis.connect(url, deadline) / redis.connect_config(cfg, deadline) | result[Conn, str] — dial, AUTH, SELECT |
redis.do(c, args, deadline) | result[Reply, str] — one command round trip |
redis.close(c) / redis.usable(c) | — shutdown / may commands still be attempted |
| strings | ping, echo, get, set, set_ex, set_nx, del_keys, exists, expire, pexpire, ttl, pttl, persist, incr, decr, incr_by, decr_by, append, strlen, mget, mset |
| hashes | hset, hget, hgetall, hdel, hexists, hkeys, hvals, hlen, hincr_by |
| lists | lpush, rpush, lpop, rpop, llen, lrange, ltrim, lindex, lrem |
| sets | sadd, smembers, srem, scard, sismember, spop |
| sorted sets | zadd, zrange, zrange_scores, zrank, zscore, zrem, zcard, zincr_by |
| keys | key_type, rename, rename_nx, scan (KEYS omitted on purpose) |
| pool | new_pool, new_pool_config, acquire, release, pool_do, pool_close |
| cluster | new_cluster, cluster_do, cluster_refresh, cluster_close, c* typed wrappers, same-slot multi-key checks |
| transactions | multi, queue, exec, discard, watch, unwatch (direct Conns; pool and cluster routing stay out) |
| scripting | eval, evalsha with NOSCRIPT fallback |
| pub/sub | subscribe, sub_add, sub_next, sub_remove, sub_close (pull model; pump pattern in docs) |
| streams | xadd, xadd_maxlen, xrange, xrevrange, xlen, xtrim, xdel, xread, cluster cxadd/cxlen/cxread (groups deferred) |
A reply is a redis.Reply: kind is one of REPLY_SIMPLE, REPLY_ERROR, REPLY_INT, REPLY_BULK or REPLY_ARRAY, with the payload in text, num, bulk (none for nil) or items (empty with is_nil for a nil array).
Replies default to a 256 MiB wire-byte limit and 1,000,000 aggregate array elements, including nested arrays. Config.max_reply_bytes and Config.max_reply_values can be raised for larger legitimate reads; each must remain positive, and the existing per-bulk, per-array, line, and nesting limits still apply. An over-limit reply fails and makes its connection unusable.
Not supported yet: consumer groups, RESP3, replica reads.
API#
The five RESP2 reply types. A reply is always one of exactly one.
A decoded reply plus how many bytes of the input it consumed, so a connection holding a read buffer knows what to keep.
Encode one command invocation as a RESP2 array of bulk strings: SET key value -> *3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$5\r\nvalue\r\n Takes already-encoded argument bytes, so binary-safe values pass through untouched. Pure: no I/O, no allocation beyond the result.
Decode one RESP2 reply from the front of buf. ok(none) means the buffer ends mid-message -- feed more bytes and retry. ok(some(d)) consumed d.consumed bytes; anything after them is the next message. err means the bytes violate the protocol: never retry on the same connection.
Decode one reply starting at pos instead of 0, for readers that keep one long-lived buffer and an offset into it (see Conn): no slicing, so no per-recv copy of everything already buffered. consumed counts from pos.
The hash slot of a key, 0..16383. {...} hash tags force colocation: only what sits between the first { and the next } is hashed. Empty tags hash the whole key, and a { with no } is literal.
Parse redis://[[username:]password@]host[:port][/db][?sslmode=...]. rediss:// forces sslmode=require. Percent-escapes in credentials and database decode the same way pg's URLs do.
One server connection: exactly one round trip in flight at a time, serialized by lock. A Conn is safe to share between tasks; every public call below takes the lock for its whole exchange.
Open a connection from a parsed Config and run the handshake (AUTH when a password is set, SELECT when db is not 0). The deadline covers everything: DNS, TCP connect, TLS upgrade, login.
Open a connection from a URL. See parse_url for the shape.
Run one command: args[0] is the command name. Exactly one reply is consumed, so at most one command is ever in flight per Conn; hold no lock of your own -- this takes c.lock for the round trip. Refused inside MULTI: queued commands answer +QUEUED, which no typed shape could read -- use queue there.
Shut the connection down. In-flight calls on other tasks finish (or hit their own deadlines) first; close waits for the lock.
True while commands may still be attempted: not closed, not broken. A server-side idle close is discovered on use, which marks the connection broken then.
PING, expecting PONG back.
ECHO, expecting the value back byte-identical.
GET: the value, or none when the key is absent. A missing key is absent data (opt), not an error.
fn set(c: Conn, key: str, val: bytes, deadline: until) -> result[bool, str] stdlib/redis/redis.sl:1335#SET: true when the server answers +OK.
fn set_ex(c: Conn, key: str, seconds: int, val: bytes, deadline: until) -> result[bool, str] stdlib/redis/redis.sl:1348#SET with a TTL in seconds. True on +OK.
fn set_nx(c: Conn, key: str, val: bytes, deadline: until) -> result[bool, str] stdlib/redis/redis.sl:1362#SETNX: true when the key was absent and is now set.
DEL_KEYS: how many of the keys existed. Named with a suffix because del itself removes map entries -- it is a builtin.
EXISTS: how many of the keys exist.
fn expire(c: Conn, key: str, seconds: int, deadline: until) -> result[bool, str] stdlib/redis/redis.sl:1399#EXPIRE: true when the timeout was set (false: missing key).
fn pexpire(c: Conn, key: str, ms: int, deadline: until) -> result[bool, str] stdlib/redis/redis.sl:1410#PEXPIRE: like EXPIRE with a millisecond TTL.
TTL in seconds: -2 missing key, -1 no TTL, else seconds left.
PTTL: like TTL in milliseconds.
PERSIST: true when a TTL was removed (false: missing key or none).
INCR/DECR: the value after the change.
fn incr_by(c: Conn, key: str, n: int, deadline: until) -> result[int, str] stdlib/redis/redis.sl:1464#fn decr_by(c: Conn, key: str, n: int, deadline: until) -> result[int, str] stdlib/redis/redis.sl:1474#fn append(c: Conn, key: str, val: bytes, deadline: until) -> result[int, str] stdlib/redis/redis.sl:1485#APPEND: the length after appending. STRLEN: the current length.
fn mget(c: Conn, keys: [str], deadline: until) -> result[[opt[bytes]], str] stdlib/redis/redis.sl:1503#MGET: one slot per key, none for the missing ones.
fn mset(c: Conn, kv: map[str]bytes, deadline: until) -> result[bool, str] stdlib/redis/redis.sl:1528#MSET: field iteration order is insertion order, so the wire order is deterministic for a literally-built map.
fn hset(c: Conn, key: str, field: str, val: bytes, deadline: until) -> result[int, str] stdlib/redis/redis.sl:1548#HSET: how many fields were newly added.
fn hget(c: Conn, key: str, field: str, deadline: until) -> result[opt[bytes], str] stdlib/redis/redis.sl:1559#HGET: the value, or none when key or field is absent.
fn hgetall(c: Conn, key: str, deadline: until) -> result[map[str]bytes, str] stdlib/redis/redis.sl:1574#HGETALL: the whole hash. Field names decode to str; binary field names a program did not put there itself come back lossy.
fn hdel(c: Conn, key: str, fields: [str], deadline: until) -> result[int, str] stdlib/redis/redis.sl:1605#HDEL: how many fields were removed.
fn hexists(c: Conn, key: str, field: str, deadline: until) -> result[bool, str] stdlib/redis/redis.sl:1618#fn hincr_by(c: Conn, key: str, field: str, n: int, deadline: until) -> result[int, str] stdlib/redis/redis.sl:1680#fn lpush(c: Conn, key: str, vals: [bytes], deadline: until) -> result[int, str] stdlib/redis/redis.sl:1706#LPUSH/RPUSH: the length after pushing.
fn rpush(c: Conn, key: str, vals: [bytes], deadline: until) -> result[int, str] stdlib/redis/redis.sl:1711#LPOP/RPOP: the element, or none when the list is absent or drained.
fn lrange(c: Conn, key: str, start: int, stop: int, deadline: until) -> result[[bytes], str] stdlib/redis/redis.sl:1749#LRANGE: elements from start to stop inclusive; negative indexes count from the tail, exactly as Redis documents.
fn ltrim(c: Conn, key: str, start: int, stop: int, deadline: until) -> result[bool, str] stdlib/redis/redis.sl:1773#fn lindex(c: Conn, key: str, i: int, deadline: until) -> result[opt[bytes], str] stdlib/redis/redis.sl:1787#fn lrem(c: Conn, key: str, count: int, val: bytes, deadline: until) -> result[int, str] stdlib/redis/redis.sl:1801#LREM: removes count occurrences of val, returns how many went.
fn sadd(c: Conn, key: str, members: [bytes], deadline: until) -> result[int, str] stdlib/redis/redis.sl:1814#SADD: how many members were newly added.
fn srem(c: Conn, key: str, members: [bytes], deadline: until) -> result[int, str] stdlib/redis/redis.sl:1842#SREM: how many members were removed.
fn sismember(c: Conn, key: str, member: bytes, deadline: until) -> result[bool, str] stdlib/redis/redis.sl:1855#SPOP: a removed member, or none when the set is absent or drained.
fn zadd(c: Conn, key: str, members: map[str]float, deadline: until) -> result[int, str] stdlib/redis/redis.sl:1885#ZADD: how many members were newly added.
fn zrange(c: Conn, key: str, start: int, stop: int, deadline: until) -> result[[bytes], str] stdlib/redis/redis.sl:1914#ZRANGE/ZREVRANGE without scores.
fn zrange_scores(c: Conn, key: str, start: int, stop: int, deadline: until) -> result[[ZMember], str] stdlib/redis/redis.sl:1939#ZRANGE WITHSCORES: member/score pairs in range order.
fn zrank(c: Conn, key: str, member: bytes, deadline: until) -> result[opt[int], str] stdlib/redis/redis.sl:1973#ZRANK: the rank, or none when key or member is absent.
fn zscore(c: Conn, key: str, member: bytes, deadline: until) -> result[opt[float], str] stdlib/redis/redis.sl:1989#ZSCORE: the score, or none when key or member is absent.
fn zrem(c: Conn, key: str, members: [bytes], deadline: until) -> result[int, str] stdlib/redis/redis.sl:2009#ZREM: how many members were removed.
fn zincr_by(c: Conn, key: str, n: float, member: bytes, deadline: until) -> result[float, str] stdlib/redis/redis.sl:2022#TYPE: the key's type name ("none" when absent).
fn rename(c: Conn, key: str, newkey: str, deadline: until) -> result[bool, str] stdlib/redis/redis.sl:2044#RENAME: true on +OK. RENAMENX: true only when newkey was absent.
fn rename_nx(c: Conn, key: str, newkey: str, deadline: until) -> result[bool, str] stdlib/redis/redis.sl:2057#fn scan(c: Conn, cursor: int, match: opt[str], count: opt[int], deadline: until) -> result[ScanOut, str] stdlib/redis/redis.sl:2075#SCAN: one cursor step. Thread cursor back in until it returns 0; match and count are server hints, both optional. KEYS is deliberately absent: it blocks the server for the whole keyspace.
Parses the url; connects nothing until the first acquire.
Same, from a Config built by hand (pool_size is ignored: max_open says it here, once, where the pool is made).
A connection for the caller's exclusive use, until release().
Returns a connection to the pool. One that is broken, closed, or inside MULTI is closed instead: handing those to the next caller would fail its first command, or run it inside someone else's uncommitted transaction.
fn pool_do(p: Pool, args: [bytes], deadline: until) -> result[Reply, str] stdlib/redis/redis.sl:2484#One command on a pooled connection: acquire, run, release. The connection goes back even when the command fails.
Closes every idle connection. Connections checked out are closed as they are released; acquire fails from now on. Closes every idle connection. Connections checked out are closed as they are released; acquire fails from now on.
fn new_cluster(cfg: Config, seeds: [str], deadline: until) -> result[Cluster, str] stdlib/redis/redis.sl:2700#Connect to a cluster: try each seed until one serves CLUSTER SLOTS. Only database 0 exists in cluster mode. The deadline covers the whole bootstrap.
Re-learn the whole slot map from a known node (any current pool will do; the first one wins) or a bootstrap seed. Manual recovery for outages the MOVED path cannot see.
fn cluster_do(cl: Cluster, key: str, args: [bytes], deadline: until) -> result[Reply, str] stdlib/redis/redis.sl:2787#Run args against the node owning key, following MOVED (map update plus retry, up to MAX_REDIRECTS) and ASK (one directed ASKING hop, returned directly). Every other error returns verbatim.
fn cget(cl: Cluster, key: str, deadline: until) -> result[opt[bytes], str] stdlib/redis/redis.sl:2921#fn cset(cl: Cluster, key: str, val: bytes, deadline: until) -> result[bool, str] stdlib/redis/redis.sl:2933#fn cset_ex(cl: Cluster, key: str, seconds: int, val: bytes, deadline: until) -> result[bool, str] stdlib/redis/redis.sl:2946#fn cset_nx(cl: Cluster, key: str, val: bytes, deadline: until) -> result[bool, str] stdlib/redis/redis.sl:2960#fn cdel_keys(cl: Cluster, keys: [str], deadline: until) -> result[int, str] stdlib/redis/redis.sl:2970#fn cexists(cl: Cluster, keys: [str], deadline: until) -> result[int, str] stdlib/redis/redis.sl:2986#fn cexpire(cl: Cluster, key: str, seconds: int, deadline: until) -> result[bool, str] stdlib/redis/redis.sl:3002#fn cpexpire(cl: Cluster, key: str, ms: int, deadline: until) -> result[bool, str] stdlib/redis/redis.sl:3012#fn cpersist(cl: Cluster, key: str, deadline: until) -> result[bool, str] stdlib/redis/redis.sl:3040#fn cincr_by(cl: Cluster, key: str, n: int, deadline: until) -> result[int, str] stdlib/redis/redis.sl:3068#fn cdecr_by(cl: Cluster, key: str, n: int, deadline: until) -> result[int, str] stdlib/redis/redis.sl:3078#fn cappend(cl: Cluster, key: str, val: bytes, deadline: until) -> result[int, str] stdlib/redis/redis.sl:3088#fn cmget(cl: Cluster, keys: [str], deadline: until) -> result[[opt[bytes]], str] stdlib/redis/redis.sl:3108#fn cmset(cl: Cluster, kv: map[str]bytes, deadline: until) -> result[bool, str] stdlib/redis/redis.sl:3134#fn chset(cl: Cluster, key: str, field: str, val: bytes, deadline: until) -> result[int, str] stdlib/redis/redis.sl:3161#fn chget(cl: Cluster, key: str, field: str, deadline: until) -> result[opt[bytes], str] stdlib/redis/redis.sl:3171#fn chdel(cl: Cluster, key: str, fields: [str], deadline: until) -> result[int, str] stdlib/redis/redis.sl:3184#fn chexists(cl: Cluster, key: str, field: str, deadline: until) -> result[bool, str] stdlib/redis/redis.sl:3197#fn chincr_by(cl: Cluster, key: str, field: str, n: int, deadline: until) -> result[int, str] stdlib/redis/redis.sl:3216#fn clpush(cl: Cluster, key: str, vals: [bytes], deadline: until) -> result[int, str] stdlib/redis/redis.sl:3227#fn crpush(cl: Cluster, key: str, vals: [bytes], deadline: until) -> result[int, str] stdlib/redis/redis.sl:3240#fn clpop(cl: Cluster, key: str, deadline: until) -> result[opt[bytes], str] stdlib/redis/redis.sl:3253#fn crpop(cl: Cluster, key: str, deadline: until) -> result[opt[bytes], str] stdlib/redis/redis.sl:3266#fn csadd(cl: Cluster, key: str, members: [bytes], deadline: until) -> result[int, str] stdlib/redis/redis.sl:3288#fn csrem(cl: Cluster, key: str, members: [bytes], deadline: until) -> result[int, str] stdlib/redis/redis.sl:3301#fn csismember(cl: Cluster, key: str, member: bytes, deadline: until) -> result[bool, str] stdlib/redis/redis.sl:3323#fn czadd(cl: Cluster, key: str, members: map[str]float, deadline: until) -> result[int, str] stdlib/redis/redis.sl:3333#fn czrem(cl: Cluster, key: str, members: [bytes], deadline: until) -> result[int, str] stdlib/redis/redis.sl:3347#fn czscore(cl: Cluster, key: str, member: bytes, deadline: until) -> result[opt[float], str] stdlib/redis/redis.sl:3369#fn ckey_type(cl: Cluster, key: str, deadline: until) -> result[str, str] stdlib/redis/redis.sl:3389#fn crename(cl: Cluster, key: str, newkey: str, deadline: until) -> result[bool, str] stdlib/redis/redis.sl:3399#fn cscan(cl: Cluster, cursor: int, match: opt[str], count: opt[int], deadline: until) -> result[ScanOut, str] stdlib/redis/redis.sl:3418#SCAN steps one node only (slot 0's owner): cluster-wide iteration fans out per node with cluster_refresh's map in hand.
MULTI: true on +OK. The connection leaves the pool from here until EXEC or DISCARD.
Queue one command inside MULTI. Anything but +QUEUED aborts the whole transaction server-side; that surfaces here as an err, and a later EXEC answers EXECABORT.
EXEC: the queued replies in order, error elements included. A nil array means nothing ran (watched keys changed): an err, since no caller could use an empty success. Always leaves MULTI, even on EXECABORT -- the server does too.
DISCARD: true on +OK, back outside MULTI either way the server answers.
WATCH/UNWATCH for optimistic locking: watch, read, MULTI, queue, EXEC; a nil EXEC (err here) means someone else wrote first, so retry the whole sequence.
fn eval(c: Conn, script: str, keys: [str], args: [bytes], deadline: until) -> result[Reply, str] stdlib/redis/redis.sl:3640#EVAL: the script's raw reply, whose shape depends on what it returns -- bulk, int, array, or error, decoded verbatim.
fn evalsha(c: Conn, script: str, keys: [str], args: [bytes], deadline: until) -> result[Reply, str] stdlib/redis/redis.sl:3649#EVALSHA with automatic EVAL fallback: the common path sends only 40 hex characters; a server that never saw the script answers NOSCRIPT and the call transparently re-sends the source.
A published message. kind is "message" or "pmessage"; pattern is set only for the latter (the glob that matched).
A live subscription: a dedicated subscriber-mode connection plus messages that arrived while someone else held the lock. Never use sub.c directly (do, close, anything): concurrent socket readers would split the stream mid-reply. Everything here serializes on the Sub lock instead.
fn subscribe(url: str, channels: [str], patterns: [str], deadline: until) -> result[Sub, str] stdlib/redis/redis.sl:3777#Subscribe, returning a live Sub. channels and patterns are the SUBSCRIBE and PSUBSCRIBE lists; at least one of them is nonempty.
fn sub_add(sub: Sub, channels: [str], patterns: [str], deadline: until) -> result[bool, str] stdlib/redis/redis.sl:3796#Add subscriptions to a live Sub.
Next message, or none when the deadline passes first. Confirms and pongs the server interleaves are skipped, never surfaced; a malformed reply is an error, since valid RESP has only the shapes as_message knows.
fn sub_remove(sub: Sub, channels: [str], patterns: [str], deadline: until) -> result[bool, str] stdlib/redis/redis.sl:3877#Remove subscriptions. Fully unsubscribing returns the connection to plain unicast mode; close it or subscribe again after.
Shut a subscription down. A task blocked in sub_next wakes with an error from the closed socket.
fn xadd(c: Conn, key: str, id: str, fields: map[str]bytes, deadline: until) -> result[str, str] stdlib/redis/redis.sl:3981#XADD: the new entry's ID.
fn xadd_maxlen(c: Conn, key: str, maxlen: int, approx: bool, id: str, fields: map[str]bytes, deadline: until) -> result[str, str] stdlib/redis/redis.sl:4002#XADD with MAXLEN trimming: approx picks ~ (cheap) over exact.
fn xrange(c: Conn, key: str, start: str, end: str, count: opt[int], deadline: until) -> result[[StreamEntry], str] stdlib/redis/redis.sl:4069#XRANGE/XREVRANGE: entries between two IDs ("-" and "+" are the ends), oldest first (XREVRANGE newest first). count caps the reply, none for no cap.
fn xrevrange(c: Conn, key: str, end: str, start: str, count: opt[int], deadline: until) -> result[[StreamEntry], str] stdlib/redis/redis.sl:4074#fn xtrim(c: Conn, key: str, maxlen: int, approx: bool, deadline: until) -> result[int, str] stdlib/redis/redis.sl:4088#XTRIM MAXLEN: how many entries were removed.
fn xdel(c: Conn, key: str, ids: [str], deadline: until) -> result[int, str] stdlib/redis/redis.sl:4103#fn xread(c: Conn, keys: [str], ids: [str], block_ms: opt[int], count: opt[int], deadline: until) -> result[opt[[StreamRead]], str] stdlib/redis/redis.sl:4120#XREAD: new entries per key since each id ("$" means "everything after now" on first call, then the last seen id after). block_ms waits that long for data (none => return at once); a wait that finds nothing is ok(none), not an error. count caps per call.
fn cxadd(cl: Cluster, key: str, id: str, fields: map[str]bytes, deadline: until) -> result[str, str] stdlib/redis/redis.sl:4197#fn cxread(cl: Cluster, keys: [str], ids: [str], block_ms: opt[int], count: opt[int], deadline: until) -> result[opt[[StreamRead]], str] stdlib/redis/redis.sl:4226#