Redis

Redis is where a microservice keeps the state that is shared between replicas but doesn't belong in the primary database: cached reads, locks, rate-limit counters, messages on their way to a websocket, the chunks of a streamed LLM answer. The redis feature of wakuwaku turns Redis into a data source like PostgreSQL: every command is a typed query, and a Db handle only runs the queries its capability allows.

On top of the data source, wakuwaku ships ready-made data structures, each one a small const value whose methods build queries:

The collections are queried on demand. The dynamic structures can also be followed as they change.

If you come from Spring, this page covers what Spring Data Redis' connection setup does, and the subpages replace the usual mix of RedisCacheManager, a distributed lock registry and a rate-limiting library, with one typing discipline across all of them.

SetupCopy Anchor Link

Enable the redis feature. Bodies stored in Redis are serialized with kanau's MessageSer / MessageDe; the JSON derives need a direct kanau dependency with its serde_json feature, plus serde:

[dependencies]
wakuwaku = { version = "1.0.0-rc3", features = ["redis"] }
kanau = { version = "0.6", features = ["serde_json"] }  # JsonMessageSer / JsonMessageDe
serde = { version = "1", features = ["derive"] }
tokio = { version = "1", features = ["full"] }

Cargo unifies this kanau with the one wakuwaku uses, so wakuwaku::deps::kanau and kanau are the same crate. See Serializing the body for how the derives work and how to write MessageSer / MessageDe by hand instead.

The Redis client itself is re-exported as wakuwaku::deps::redis, so you don't add redis to Cargo.toml.

ConnectingCopy Anchor Link

A RedisSource wraps a cloneable async connection and runs each query on a clone of it. With the default MultiplexedConnection the clones share one socket, so this is cheap. Wrap the source in a Db with a capability:

use wakuwaku::deps::redis;
use wakuwaku::entity::redis::RedisSource;
use wakuwaku::entity::{Db, ReadWrite};

/// The Redis handle every component of this service shares.
pub type Redis = Db<RedisSource, ReadWrite>;

pub async fn connect(client: &redis::Client) -> redis::RedisResult<Redis> {
    let connection = client.get_multiplexed_async_connection().await?;
    Ok(Db::new(RedisSource::new(connection)))
}

RedisSource<C> accepts any C: ConnectionLike + Clone + Send + Sync, so a cluster connection or a ConnectionManager works the same way.

Registering itCopy Anchor Link

Db is Clone whenever its source is, so register it in the ServiceBuilder and build components from it:

use wakuwaku::{ServiceBuilder, ServiceCreation};

pub struct SessionService {
    redis: Redis,
}

impl ServiceCreation for SessionService {
    type Dep = Redis;

    fn create_service(redis: Redis) -> Self {
        Self { redis }
    }
}

pub async fn run() -> Result<(), Box<dyn std::error::Error>> {
    let client = redis::Client::open("redis://127.0.0.1/")?;
    let services = ServiceBuilder::new(connect(&client).await?);
    let redis: &Redis = services.provide::<Redis, _>();
    Ok(())
}

Every query of this section is then run with redis.process(query).await, the Processor method that Db implements for each query its capability covers.

Your own Redis queriesCopy Anchor Link

The built-in structures are queries like any other. When you need a command they don't cover, write the query yourself: Query declares which entity it touches, and Execute<RedisSource<C>> runs it on the connection.

use wakuwaku::deps::redis::aio::ConnectionLike;
use wakuwaku::deps::redis::{self, AsyncCommands};
use wakuwaku::entity::redis::RedisSource;
use wakuwaku::entity::{Entity, Execute, Query, Write};

/// The entity: login sessions, wherever they are stored.
pub struct Session;
impl Entity for Session {}

pub struct StoreSession {
    pub token: String,
    pub user: u64,
}

impl Query for StoreSession {
    type Effect = Write<Session>;
}

impl<C: ConnectionLike + Clone + Send + Sync> Execute<RedisSource<C>> for StoreSession {
    type Output = ();
    type Error = redis::RedisError;

    async fn execute(self, conn: &mut C) -> redis::RedisResult<()> {
        conn.set_ex(format!("session:{}", self.token), self.user, 3600).await
    }
}

Every query declares its error type. It must convert from RedisError, the error of RedisSource, so a failure to talk to Redis can always be reported. The built-in structures use RedisError itself, and also report a body that fails to (de)serialize, and a key whose contents don't match what the structure wrote ("corrupted item"), as a RedisError.

Queries that can refuseCopy Anchor Link

A query that can refuse on its own, because a seat is already taken or a business rule says no, declares its own error type instead. The refusal then reaches the caller as an Err next to the Redis failures, and a single ? handles both:

use wakuwaku::deps::redis::aio::ConnectionLike;
use wakuwaku::deps::redis::{self, AsyncCommands};
use wakuwaku::entity::redis::RedisSource;
use wakuwaku::entity::{Entity, Execute, Query, Write};

pub struct Seat;
impl Entity for Seat {}

#[derive(Debug)]
pub enum ClaimError {
    /// Someone else holds the seat.
    Taken,
    Redis(redis::RedisError),
}

impl From<redis::RedisError> for ClaimError {
    fn from(error: redis::RedisError) -> Self {
        Self::Redis(error)
    }
}

pub struct ClaimSeat {
    pub seat: u32,
    pub user: u64,
}

impl Query for ClaimSeat {
    type Effect = Write<Seat>;
}

impl<C: ConnectionLike + Clone + Send + Sync> Execute<RedisSource<C>> for ClaimSeat {
    type Output = ();
    type Error = ClaimError;

    async fn execute(self, conn: &mut C) -> Result<(), ClaimError> {
        let claimed: bool = conn.set_nx(format!("seat:{}", self.seat), self.user).await?;
        if claimed { Ok(()) } else { Err(ClaimError::Taken) }
    }
}

redis.process(ClaimSeat { seat, user }).await then outputs Result<(), ClaimError>: no Result nested in the output, and no separate check after the query.

Names and keysCopy Anchor Link

Each structure is generic over the type N that names its items: Cache<u64, Profile> is keyed by user ID, Lock<str> by a string. A name must implement Name, which strings, byte strings and every integer type already do. For an identifier newtype, write its canonical form:

use wakuwaku::entity::redis::Name;

pub struct UserId(pub u64);

impl Name for UserId {
    fn write_name(&self, key: &mut Vec<u8>) {
        self.0.write_name(key);
    }
}

An item lives under the key

@@wakuwaku-entity:<kind>:<namespace>:<name>

where <kind> is the structure type (Cache, Lock, TokenBucket, …) and <namespace> is the first argument of its new. Items made of several keys (RwLock, Pipe, LiveView) wrap <namespace>:<name> in braces and append a suffix per key, so all their keys share one Redis Cluster hash slot and their Lua scripts and MULTI blocks work on a cluster.

CapabilitiesCopy Anchor Link

Each structure type is also the entity its queries read and write, so Cache<u64, Profile> and Cache<str, Page> are different entities. Which queries count as writes follows what they do in Redis, not what they look like:

A Db<_, ReadOnly> therefore can't take a lock or consume a rate-limit token, and putting a body through it doesn't compile:

use wakuwaku::deps::kanau::processor::Processor;
use wakuwaku::deps::redis::RedisResult;
use wakuwaku::entity::redis::RedisSource;
use wakuwaku::entity::{Db, ReadOnly};

async fn warm(replica: &Db<RedisSource, ReadOnly>, user: u64, profile: Profile) -> RedisResult<()> {
    // error: `ReadOnly: CanWrite<Cache<u64, Profile>>` is not satisfied
    replica.process(PROFILES.put(&user, profile)).await
}

A capability can also grant access to some structures only. This gateway reads every cache but writes only the session cache:

use wakuwaku::entity::redis::collection::cache::Cache;
use wakuwaku::entity::{CanRead, CanWrite};

pub struct SessionToken(String);
pub struct Session {
    user: u64,
}

/// Reads every cache, writes only the session cache.
pub enum Gateway {}
impl<N: ?Sized, B> CanRead<Cache<N, B>> for Gateway {}
impl CanWrite<Cache<SessionToken, Session>> for Gateway {}

Following changes: dedicated connectionsCopy Anchor Link

Receiving from a pipe and subscribing to a live view or a stream buffer are queries too, so the capability is checked as usual. They output a receiver or subscription that waits with blocking Redis commands (BLPOP, XREAD BLOCK) on a connection you pass to the query.

A blocking command stalls every other command sent on the same socket, so that connection must be used for nothing else: never pass a clone of the multiplexed connection inside your RedisSource. It must also not time out a response before the block ends. Open one per receiver with the response timeout turned off:

use wakuwaku::deps::redis::AsyncConnectionConfig;
use wakuwaku::deps::redis::aio::MultiplexedConnection;

/// A connection for one receiver or subscription, used for nothing else.
pub async fn dedicated_connection(
    client: &redis::Client,
) -> redis::RedisResult<MultiplexedConnection> {
    let config = AsyncConnectionConfig::new().set_response_timeout(None);
    client.get_multiplexed_async_connection_with_config(&config).await
}

The examples in the following pages use this helper. Each receiver and subscription has an async fn recv and an into_stream that turns it into a futures::Stream.

Server requirementsCopy Anchor Link

• Locks and the rate limiter read the server clock with TIME inside Lua scripts. This needs effects replication of scripts, the default since Redis 5. Because the clock is the server's, clock skew between your replicas doesn't matter.
• Pipe receivers block with fractional BLPOP timeouts, which need Redis 6.
• Structures that span several keys keep them in one hash slot, so Redis Cluster is supported.

Next stepsCopy Anchor Link