Caching with Redis

Redis is an in-memory data store that services use as a shared cache, so every instance sees the same entries. This lesson caches Storefront products from PostgreSQL with redis-rs (the redis crate, version 0.27) and covers cache-aside reads, invalidation, stampede protection, locks, rate-limit counters, and a sorted-set ranking. The database side is in PostgreSQL with SQLx.

Project setup

Wait a few seconds after docker run until PostgreSQL accepts connections, then run the psql line.

Terminal

docker run --name pg -e POSTGRES_PASSWORD=pw -e POSTGRES_DB=storefront -p 5432:5432 -d postgres:16
docker run --name redis -p 6379:6379 -d redis:7
docker exec -i pg psql -U postgres -d storefront < migrations/0001_products.sql
export DATABASE_URL=postgres://postgres:pw@127.0.0.1:5432/storefront

Cargo.toml

TOML

# Cargo.toml
[package]
name = "storefront-cache"
version = "0.1.0"
edition = "2021"

[dependencies]
axum = "0.8"
rand = "0.8"
redis = { version = "0.27", features = ["tokio-comp", "connection-manager"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
sqlx = { version = "0.8", features = ["runtime-tokio", "postgres"] }
tokio = { version = "1", features = ["full"] }
uuid = { version = "1", features = ["v4"] }

migrations/0001_products.sql

SQL

-- migrations/0001_products.sql
CREATE TABLE products (
    id          BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    sku         TEXT    NOT NULL UNIQUE,
    name        TEXT    NOT NULL,
    price_cents INTEGER NOT NULL CHECK (price_cents > 0)
);

INSERT INTO products (sku, name, price_cents)
VALUES ('PEN-1', 'Pen', 250), ('BOOK-1', 'Notebook', 900);

Example 1: Connect and run commands

ConnectionManager is a cloneable handle to one multiplexed connection. It reconnects after a dropped connection, so it is stored in the shared state. The async commands come from the AsyncCommands trait, and the annotated return type selects how the reply is converted.

Rust

// src/main.rs
use redis::AsyncCommands;

#[tokio::main]
async fn main() -> redis::RedisResult<()> {
    let client = redis::Client::open("redis://127.0.0.1/")?;
    let mut conn = redis::aio::ConnectionManager::new(client).await?;

    let _: () = conn.set("greeting", "hello").await?;
    let value: Option<String> = conn.get("greeting").await?;
    println!("{value:?}");

    let ttl: i64 = conn.ttl("greeting").await?;
    println!("{ttl}");

    let deleted: i64 = conn.del("greeting").await?;
    println!("{deleted}");

    let missing: Option<String> = conn.get("greeting").await?;
    println!("{missing:?}");
    Ok(())
}

Output

Some("hello")
-1
1
None

get returns Option<String>. None means the key does not exist.
ttl returns -1 for a key with no expiry. A cache entry always has one.
del returns the number of keys removed.
Redis runs one command at a time, so a single command is atomic. Commands from different clients can interleave.

Example 2: Cache-aside for a product

Cache-aside means the application reads the cache first, and on a miss loads from the database and fills the cache with a TTL. The TTL (time to live) is the number of seconds after which Redis removes the key. The next files replace src/main.rs and add the service.

src/error.rs

Rust

// src/error.rs
use axum::{
    http::StatusCode,
    response::{IntoResponse, Response},
    Json,
};
use serde_json::json;

pub enum ApiError {
    NotFound,
    Internal,
}

impl From<sqlx::Error> for ApiError {
    fn from(err: sqlx::Error) -> Self {
        eprintln!("database error: {err}");
        ApiError::Internal
    }
}

impl From<redis::RedisError> for ApiError {
    fn from(err: redis::RedisError) -> Self {
        eprintln!("redis error: {err}");
        ApiError::Internal
    }
}

impl IntoResponse for ApiError {
    fn into_response(self) -> Response {
        let (status, message) = match self {
            ApiError::NotFound => (StatusCode::NOT_FOUND, "not found"),
            ApiError::Internal => (StatusCode::INTERNAL_SERVER_ERROR, "internal error"),
        };
        (status, Json(json!({ "error": message }))).into_response()
    }
}

src/cache.rs

Rust

// src/cache.rs
use redis::{aio::ConnectionManager, AsyncCommands};
use serde::{de::DeserializeOwned, Serialize};

/// Returns `None` on a miss, on a Redis error, and on an undecodable value.
pub async fn get_json<T: DeserializeOwned>(redis: &mut ConnectionManager, key: &str) -> Option<T> {
    match redis.get::<_, Option<String>>(key).await {
        Ok(Some(json)) => serde_json::from_str(&json).ok(),
        Ok(None) => None,
        Err(err) => {
            eprintln!("cache read failed: {err}");
            None
        }
    }
}

/// Stores the value and its TTL in one command. A failure is logged, not returned.
pub async fn set_json<T: Serialize>(
    redis: &mut ConnectionManager,
    key: &str,
    value: &T,
    ttl_secs: u64,
) {
    let Ok(json) = serde_json::to_string(value) else {
        return;
    };
    if let Err(err) = redis.set_ex::<_, _, ()>(key, json, ttl_secs).await {
        eprintln!("cache write failed: {err}");
    }
}

pub async fn delete(redis: &mut ConnectionManager, key: &str) {
    if let Err(err) = redis.del::<_, ()>(key).await {
        eprintln!("cache delete failed: {err}");
    }
}

src/products.rs

Rust

// src/products.rs
use axum::{
    extract::{Path, State},
    response::{IntoResponse, Response},
    Json,
};
use serde::{Deserialize, Serialize};

use crate::{cache, error::ApiError, AppState};

const TTL_SECS: u64 = 300;
const MISSING_TTL_SECS: u64 = 30;

#[derive(Clone, Serialize, Deserialize, sqlx::FromRow)]
pub struct Product {
    pub id: i64,
    pub sku: String,
    pub name: String,
    pub price_cents: i32,
}

pub fn product_key(id: i64) -> String {
    format!("storefront:v1:product:{id}")
}

async fn find_product(state: &AppState, id: i64) -> Result<Option<Product>, ApiError> {
    let row = sqlx::query_as::<_, Product>(
        "SELECT id, sku, name, price_cents FROM products WHERE id = $1",
    )
    .bind(id)
    .fetch_optional(&state.pool)
    .await?;
    Ok(row)
}

fn respond(product: Option<Product>, source: &'static str) -> Result<Response, ApiError> {
    let product = product.ok_or(ApiError::NotFound)?;
    Ok(([("x-cache", source)], Json(product)).into_response())
}

pub async fn get_one(
    State(state): State<AppState>,
    Path(id): Path<i64>,
) -> Result<Response, ApiError> {
    let key = product_key(id);
    let mut redis = state.redis.clone();

    if let Some(cached) = cache::get_json::<Option<Product>>(&mut redis, &key).await {
        return respond(cached, "hit");
    }

    let product = find_product(&state, id).await?;
    let ttl = if product.is_some() { TTL_SECS } else { MISSING_TTL_SECS };
    cache::set_json(&mut redis, &key, &product, ttl).await;
    respond(product, "miss")
}

src/main.rs

Rust

// src/main.rs
mod cache;
mod error;
mod products;

use axum::{routing::get, Router};
use redis::aio::ConnectionManager;
use sqlx::{postgres::PgPoolOptions, PgPool};

#[derive(Clone)]
pub struct AppState {
    pub pool: PgPool,
    pub redis: ConnectionManager,
}

#[tokio::main]
async fn main() {
    let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL must be set");
    let redis_url =
        std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1/".to_owned());

    let pool = PgPoolOptions::new()
        .max_connections(5)
        .connect(&database_url)
        .await
        .expect("database connection");
    let client = redis::Client::open(redis_url).expect("redis url");
    let redis = ConnectionManager::new(client).await.expect("redis connection");
    let state = AppState { pool, redis };

    let app = Router::new()
        .route("/products/{id}", get(products::get_one))
        .with_state(state);

    let listener = tokio::net::TcpListener::bind("127.0.0.1:3000").await.unwrap();
    axum::serve(listener, app).await.unwrap();
}

The cache stores Option<Product> as JSON. A missing product is cached as null for 30 seconds (negative caching), so requests for ids that do not exist stop reaching the database.
A Redis error counts as a miss. The cache is an optimization, so a Redis outage slows the service and does not stop it.
The key has a namespace and a version (storefront:v1:product:7). Changing the stored format means changing v1 to v2, and old entries expire unread.
set_ex writes the value and TTL in one command. A separate SET and EXPIRE can leave a key with no expiry if the process stops between them.
The x-cache response header reports hit or miss.

Example 3: Invalidate on write

Cached data goes stale when the source changes. Three write strategies exist.

Strategy	Write path	Trade-off
Invalidate on write	Update the database, delete the key	The next read reloads. Fewest failure modes
Write-through	Update the database, then write the new value to the cache	Reads stay warm. A rolled-back write can leave a wrong cached value
Write-behind	Write the cache, persist later	Data is lost if Redis fails first. Not for orders or payments

The example uses invalidate on write. src/products.rs gains update. src/main.rs registers PUT. Other files are unchanged.

src/products.rs

Rust

// src/products.rs
use axum::{
    extract::{Path, State},
    response::{IntoResponse, Response},
    Json,
};
use serde::{Deserialize, Serialize};

use crate::{cache, error::ApiError, AppState};

const TTL_SECS: u64 = 300;
const MISSING_TTL_SECS: u64 = 30;

#[derive(Clone, Serialize, Deserialize, sqlx::FromRow)]
pub struct Product {
    pub id: i64,
    pub sku: String,
    pub name: String,
    pub price_cents: i32,
}

#[derive(Deserialize)]
pub struct UpdateProduct {
    name: String,
    price_cents: i32,
}

pub fn product_key(id: i64) -> String {
    format!("storefront:v1:product:{id}")
}

async fn find_product(state: &AppState, id: i64) -> Result<Option<Product>, ApiError> {
    let row = sqlx::query_as::<_, Product>(
        "SELECT id, sku, name, price_cents FROM products WHERE id = $1",
    )
    .bind(id)
    .fetch_optional(&state.pool)
    .await?;
    Ok(row)
}

fn respond(product: Option<Product>, source: &'static str) -> Result<Response, ApiError> {
    let product = product.ok_or(ApiError::NotFound)?;
    Ok(([("x-cache", source)], Json(product)).into_response())
}

pub async fn get_one(
    State(state): State<AppState>,
    Path(id): Path<i64>,
) -> Result<Response, ApiError> {
    let key = product_key(id);
    let mut redis = state.redis.clone();

    if let Some(cached) = cache::get_json::<Option<Product>>(&mut redis, &key).await {
        return respond(cached, "hit");
    }

    let product = find_product(&state, id).await?;
    let ttl = if product.is_some() { TTL_SECS } else { MISSING_TTL_SECS };
    cache::set_json(&mut redis, &key, &product, ttl).await;
    respond(product, "miss")
}

pub async fn update(
    State(state): State<AppState>,
    Path(id): Path<i64>,
    Json(body): Json<UpdateProduct>,
) -> Result<Json<Product>, ApiError> {
    let updated = sqlx::query_as::<_, Product>(
        "UPDATE products SET name = $2, price_cents = $3 WHERE id = $1
         RETURNING id, sku, name, price_cents",
    )
    .bind(id)
    .bind(&body.name)
    .bind(body.price_cents)
    .fetch_optional(&state.pool)
    .await?
    .ok_or(ApiError::NotFound)?;

    // The database change is committed. Delete the key after it, never before.
    let mut redis = state.redis.clone();
    cache::delete(&mut redis, &product_key(id)).await;
    Ok(Json(updated))
}

src/main.rs

Rust

// src/main.rs
mod cache;
mod error;
mod products;

use axum::{routing::get, Router};
use redis::aio::ConnectionManager;
use sqlx::{postgres::PgPoolOptions, PgPool};

#[derive(Clone)]
pub struct AppState {
    pub pool: PgPool,
    pub redis: ConnectionManager,
}

#[tokio::main]
async fn main() {
    let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL must be set");
    let redis_url =
        std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1/".to_owned());

    let pool = PgPoolOptions::new()
        .max_connections(5)
        .connect(&database_url)
        .await
        .expect("database connection");
    let client = redis::Client::open(redis_url).expect("redis url");
    let redis = ConnectionManager::new(client).await.expect("redis connection");
    let state = AppState { pool, redis };

    let app = Router::new()
        .route("/products/{id}", get(products::get_one).put(products::update))
        .with_state(state);

    let listener = tokio::net::TcpListener::bind("127.0.0.1:3000").await.unwrap();
    axum::serve(listener, app).await.unwrap();
}

The delete comes after the database commit. Deleting first lets a reader load the old row between the delete and the commit and cache it with a full TTL.
If the delete fails, the stale entry lives until its TTL ends. The TTL bounds every such mistake.
Deleting keys by pattern with KEYS blocks Redis. For list caches, put a version counter in the key and increment it to invalidate every list at once.

Example 4: Stampede protection

A stampede is many requests recomputing the same key after it expires. This example applies two defenses. TTL jitter adds up to 10 percent to each TTL so keys written together do not expire together. A lock lets one request per key load from the database while the others wait for the cache.

SET key token NX PX ms sets the key only if it does not exist and gives it an expiry, in one command. The token identifies the holder. Release runs a Lua script that deletes the key only if the token still matches, because the lock may have expired and been taken by another instance.

src/cache.rs gains try_lock, unlock, and jittered. src/products.rs gains fill and a new get_one.

src/cache.rs

Rust

// src/cache.rs
use redis::{aio::ConnectionManager, AsyncCommands};
use serde::{de::DeserializeOwned, Serialize};

/// Returns `None` on a miss, on a Redis error, and on an undecodable value.
pub async fn get_json<T: DeserializeOwned>(redis: &mut ConnectionManager, key: &str) -> Option<T> {
    match redis.get::<_, Option<String>>(key).await {
        Ok(Some(json)) => serde_json::from_str(&json).ok(),
        Ok(None) => None,
        Err(err) => {
            eprintln!("cache read failed: {err}");
            None
        }
    }
}

/// Stores the value and its TTL in one command. A failure is logged, not returned.
pub async fn set_json<T: Serialize>(
    redis: &mut ConnectionManager,
    key: &str,
    value: &T,
    ttl_secs: u64,
) {
    let Ok(json) = serde_json::to_string(value) else {
        return;
    };
    if let Err(err) = redis.set_ex::<_, _, ()>(key, json, ttl_secs).await {
        eprintln!("cache write failed: {err}");
    }
}

pub async fn delete(redis: &mut ConnectionManager, key: &str) {
    if let Err(err) = redis.del::<_, ()>(key).await {
        eprintln!("cache delete failed: {err}");
    }
}

/// `SET key token NX PX ttl_ms`: sets the key only if it does not exist.
/// When Redis fails, the caller proceeds without a lock.
pub async fn try_lock(
    redis: &mut ConnectionManager,
    key: &str,
    token: &str,
    ttl_ms: u64,
) -> bool {
    let reply: redis::RedisResult<Option<String>> = redis::cmd("SET")
        .arg(key)
        .arg(token)
        .arg("NX")
        .arg("PX")
        .arg(ttl_ms)
        .query_async(redis)
        .await;
    match reply {
        Ok(reply) => reply.is_some(),
        Err(err) => {
            eprintln!("lock failed: {err}");
            true
        }
    }
}

/// Deletes the lock only if it still holds this token. The check and the
/// delete run as one Lua script, so no other command runs between them.
pub async fn unlock(redis: &mut ConnectionManager, key: &str, token: &str) {
    let script = redis::Script::new(
        r#"
        if redis.call("GET", KEYS[1]) == ARGV[1] then
            return redis.call("DEL", KEYS[1])
        else
            return 0
        end
        "#,
    );
    let result: redis::RedisResult<i32> = script.key(key).arg(token).invoke_async(redis).await;
    if let Err(err) = result {
        eprintln!("unlock failed: {err}");
    }
}

/// Adds up to 10 percent to a TTL so keys written together do not expire together.
pub fn jittered(base_secs: u64) -> u64 {
    base_secs + rand::random::<u64>() % (base_secs / 10 + 1)
}

src/products.rs

Rust

// src/products.rs
use std::time::Duration;

use axum::{
    extract::{Path, State},
    response::{IntoResponse, Response},
    Json,
};
use redis::aio::ConnectionManager;
use serde::{Deserialize, Serialize};
use uuid::Uuid;

use crate::{cache, error::ApiError, AppState};

const TTL_SECS: u64 = 300;
const MISSING_TTL_SECS: u64 = 30;

#[derive(Clone, Serialize, Deserialize, sqlx::FromRow)]
pub struct Product {
    pub id: i64,
    pub sku: String,
    pub name: String,
    pub price_cents: i32,
}

#[derive(Deserialize)]
pub struct UpdateProduct {
    name: String,
    price_cents: i32,
}

pub fn product_key(id: i64) -> String {
    format!("storefront:v1:product:{id}")
}

async fn find_product(state: &AppState, id: i64) -> Result<Option<Product>, ApiError> {
    let row = sqlx::query_as::<_, Product>(
        "SELECT id, sku, name, price_cents FROM products WHERE id = $1",
    )
    .bind(id)
    .fetch_optional(&state.pool)
    .await?;
    Ok(row)
}

fn respond(product: Option<Product>, source: &'static str) -> Result<Response, ApiError> {
    let product = product.ok_or(ApiError::NotFound)?;
    Ok(([("x-cache", source)], Json(product)).into_response())
}

/// Loads from the database and fills the cache. A missing row is cached briefly.
async fn fill(
    state: &AppState,
    redis: &mut ConnectionManager,
    key: &str,
    id: i64,
) -> Result<Option<Product>, ApiError> {
    let product = find_product(state, id).await?;
    let ttl = if product.is_some() {
        cache::jittered(TTL_SECS)
    } else {
        MISSING_TTL_SECS
    };
    cache::set_json(redis, key, &product, ttl).await;
    Ok(product)
}

pub async fn get_one(
    State(state): State<AppState>,
    Path(id): Path<i64>,
) -> Result<Response, ApiError> {
    let key = product_key(id);
    let mut redis = state.redis.clone();

    if let Some(cached) = cache::get_json::<Option<Product>>(&mut redis, &key).await {
        return respond(cached, "hit");
    }

    // One request per key loads from the database. The others wait for the cache.
    let lock_key = format!("{key}:lock");
    let token = Uuid::new_v4().to_string();
    if cache::try_lock(&mut redis, &lock_key, &token, 5_000).await {
        let loaded = fill(&state, &mut redis, &key, id).await;
        cache::unlock(&mut redis, &lock_key, &token).await;
        return respond(loaded?, "miss");
    }

    for _ in 0..20 {
        tokio::time::sleep(Duration::from_millis(50)).await;
        if let Some(cached) = cache::get_json::<Option<Product>>(&mut redis, &key).await {
            return respond(cached, "wait");
        }
    }
    respond(fill(&state, &mut redis, &key, id).await?, "miss")
}

pub async fn update(
    State(state): State<AppState>,
    Path(id): Path<i64>,
    Json(body): Json<UpdateProduct>,
) -> Result<Json<Product>, ApiError> {
    let updated = sqlx::query_as::<_, Product>(
        "UPDATE products SET name = $2, price_cents = $3 WHERE id = $1
         RETURNING id, sku, name, price_cents",
    )
    .bind(id)
    .bind(&body.name)
    .bind(body.price_cents)
    .fetch_optional(&state.pool)
    .await?
    .ok_or(ApiError::NotFound)?;

    let mut redis = state.redis.clone();
    cache::delete(&mut redis, &product_key(id)).await;
    Ok(Json(updated))
}

try_lock treats a Redis error as "proceed", so a Redis outage does not block reads.
A waiting request polls the cache every 50 ms for up to one second, then loads from the database itself.
The lock is safe for avoiding duplicate work. It cannot guarantee mutual exclusion: if the holder pauses longer than the TTL, a second holder starts. Work that must never run twice needs a database constraint or a fencing token, a number that grows with each grant and that the protected resource checks.
Each instance can also hold an in-process cache such as the moka crate. It avoids the network round trip, but an update invalidates only the instance that handled it.

Example 5: Rate limiting and ranking

Redis values have types, and each type has commands that change part of a value atomically.

Type	Commands	Storefront use
String	`SET`, `GET`, `INCR`	Cached JSON, counters
Hash	`HSET`, `HINCRBY`	One object with independent fields, such as a cart
Set	`SADD`, `SISMEMBER`	Unique members
Sorted set	`ZADD`, `ZINCRBY`, `ZREVRANGE`	Rankings ordered by score

Two new files use them. The rate limiter is a fixed window: INCR counts requests per client, and the key expires when the window ends. A client can send the limit at the end of one window and again at the start of the next, up to twice the limit across the boundary. The ranking is a sorted set where each sale adds to a product score.

src/rate_limit.rs

Rust

// src/rate_limit.rs
use axum::{
    extract::{Request, State},
    http::{header, StatusCode},
    middleware::Next,
    response::{IntoResponse, Response},
    Json,
};
use redis::aio::ConnectionManager;
use serde_json::json;

use crate::AppState;

const LIMIT: i64 = 20;
const WINDOW_SECS: i64 = 60;

/// Counts one request for the client in the current window and returns the count.
/// INCR and EXPIRE run in one script, so a crash cannot leave a counter without an expiry.
async fn count_request(redis: &mut ConnectionManager, client: &str) -> redis::RedisResult<i64> {
    let script = redis::Script::new(
        r#"
        local count = redis.call("INCR", KEYS[1])
        if count == 1 then
            redis.call("EXPIRE", KEYS[1], ARGV[1])
        end
        return count
        "#,
    );
    let key = format!("storefront:v1:rate:{client}");
    script.key(key).arg(WINDOW_SECS).invoke_async(redis).await
}

pub async fn limit(State(state): State<AppState>, request: Request, next: Next) -> Response {
    let client = request
        .headers()
        .get("x-client-id")
        .and_then(|value| value.to_str().ok())
        .unwrap_or("anonymous")
        .to_owned();

    let mut redis = state.redis.clone();
    match count_request(&mut redis, &client).await {
        Ok(count) if count > LIMIT => (
            StatusCode::TOO_MANY_REQUESTS,
            [(header::RETRY_AFTER, WINDOW_SECS.to_string())],
            Json(json!({ "error": "rate limit exceeded" })),
        )
            .into_response(),
        Ok(_) => next.run(request).await,
        Err(err) => {
            // Fail open: a Redis outage does not block traffic.
            eprintln!("rate limiter unavailable: {err}");
            next.run(request).await
        }
    }
}

src/ranking.rs

Rust

// src/ranking.rs
use axum::{
    extract::{Path, State},
    Json,
};
use redis::AsyncCommands;
use serde::{Deserialize, Serialize};

use crate::{error::ApiError, AppState};

const KEY: &str = "storefront:v1:best_sellers";

#[derive(Deserialize)]
pub struct Sale {
    qty: i64,
}

#[derive(Serialize)]
pub struct Ranked {
    product_id: i64,
    sold: f64,
}

pub async fn record_sale(
    State(state): State<AppState>,
    Path(id): Path<i64>,
    Json(body): Json<Sale>,
) -> Result<Json<Ranked>, ApiError> {
    let mut redis = state.redis.clone();
    let sold: f64 = redis.zincr(KEY, id, body.qty).await?;
    Ok(Json(Ranked { product_id: id, sold }))
}

pub async fn best_sellers(State(state): State<AppState>) -> Result<Json<Vec<Ranked>>, ApiError> {
    let mut redis = state.redis.clone();
    let rows: Vec<(i64, f64)> = redis.zrevrange_withscores(KEY, 0, 2).await?;
    let ranked = rows
        .into_iter()
        .map(|(product_id, sold)| Ranked { product_id, sold })
        .collect();
    Ok(Json(ranked))
}

src/main.rs

Rust

// src/main.rs
mod cache;
mod error;
mod products;
mod ranking;
mod rate_limit;

use axum::{middleware, routing::{get, post}, Router};
use redis::aio::ConnectionManager;
use sqlx::{postgres::PgPoolOptions, PgPool};

#[derive(Clone)]
pub struct AppState {
    pub pool: PgPool,
    pub redis: ConnectionManager,
}

#[tokio::main]
async fn main() {
    let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL must be set");
    let redis_url =
        std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1/".to_owned());

    let pool = PgPoolOptions::new()
        .max_connections(5)
        .connect(&database_url)
        .await
        .expect("database connection");
    let client = redis::Client::open(redis_url).expect("redis url");
    let redis = ConnectionManager::new(client).await.expect("redis connection");
    let state = AppState { pool, redis };

    let app = Router::new()
        .route("/products/{id}", get(products::get_one).put(products::update))
        .route("/products/{id}/sales", post(ranking::record_sale))
        .route("/best-sellers", get(ranking::best_sellers))
        .layer(middleware::from_fn_with_state(state.clone(), rate_limit::limit))
        .with_state(state);

    let listener = tokio::net::TcpListener::bind("127.0.0.1:3000").await.unwrap();
    axum::serve(listener, app).await.unwrap();
}

The limiter identifies clients by the x-client-id header for this example. A real service uses the authenticated identity (Authentication and authorization).
When Redis is unavailable, the limiter fails open. Failing closed rejects all traffic during an outage.
ZINCRBY creates the member on first use and adds to its score in one atomic command.
Redis holds data in memory. Data that must not be lost stays in PostgreSQL. A cache is rebuildable from it. Other limiter algorithms are in Resilience patterns.

Complete example

File tree:

Text

storefront-cache/
  Cargo.toml
  migrations/0001_products.sql
  src/
    main.rs
    error.rs
    cache.rs
    products.rs
    rate_limit.rs
    ranking.rs

Cargo.toml

TOML

# Cargo.toml
[package]
name = "storefront-cache"
version = "0.1.0"
edition = "2021"

[dependencies]
axum = "0.8"
rand = "0.8"
redis = { version = "0.27", features = ["tokio-comp", "connection-manager"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
sqlx = { version = "0.8", features = ["runtime-tokio", "postgres"] }
tokio = { version = "1", features = ["full"] }
uuid = { version = "1", features = ["v4"] }

migrations/0001_products.sql

SQL

-- migrations/0001_products.sql
CREATE TABLE products (
    id          BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    sku         TEXT    NOT NULL UNIQUE,
    name        TEXT    NOT NULL,
    price_cents INTEGER NOT NULL CHECK (price_cents > 0)
);

INSERT INTO products (sku, name, price_cents)
VALUES ('PEN-1', 'Pen', 250), ('BOOK-1', 'Notebook', 900);

src/main.rs

Rust

// src/main.rs
mod cache;
mod error;
mod products;
mod ranking;
mod rate_limit;

use axum::{middleware, routing::{get, post}, Router};
use redis::aio::ConnectionManager;
use sqlx::{postgres::PgPoolOptions, PgPool};

#[derive(Clone)]
pub struct AppState {
    pub pool: PgPool,
    pub redis: ConnectionManager,
}

#[tokio::main]
async fn main() {
    let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL must be set");
    let redis_url =
        std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1/".to_owned());

    let pool = PgPoolOptions::new()
        .max_connections(5)
        .connect(&database_url)
        .await
        .expect("database connection");
    let client = redis::Client::open(redis_url).expect("redis url");
    let redis = ConnectionManager::new(client).await.expect("redis connection");
    let state = AppState { pool, redis };

    let app = Router::new()
        .route("/products/{id}", get(products::get_one).put(products::update))
        .route("/products/{id}/sales", post(ranking::record_sale))
        .route("/best-sellers", get(ranking::best_sellers))
        .layer(middleware::from_fn_with_state(state.clone(), rate_limit::limit))
        .with_state(state);

    let listener = tokio::net::TcpListener::bind("127.0.0.1:3000").await.unwrap();
    axum::serve(listener, app).await.unwrap();
}

src/error.rs

Rust

// src/error.rs
use axum::{
    http::StatusCode,
    response::{IntoResponse, Response},
    Json,
};
use serde_json::json;

pub enum ApiError {
    NotFound,
    Internal,
}

impl From<sqlx::Error> for ApiError {
    fn from(err: sqlx::Error) -> Self {
        eprintln!("database error: {err}");
        ApiError::Internal
    }
}

impl From<redis::RedisError> for ApiError {
    fn from(err: redis::RedisError) -> Self {
        eprintln!("redis error: {err}");
        ApiError::Internal
    }
}

impl IntoResponse for ApiError {
    fn into_response(self) -> Response {
        let (status, message) = match self {
            ApiError::NotFound => (StatusCode::NOT_FOUND, "not found"),
            ApiError::Internal => (StatusCode::INTERNAL_SERVER_ERROR, "internal error"),
        };
        (status, Json(json!({ "error": message }))).into_response()
    }
}

src/cache.rs

Rust

// src/cache.rs
use redis::{aio::ConnectionManager, AsyncCommands};
use serde::{de::DeserializeOwned, Serialize};

/// Returns `None` on a miss, on a Redis error, and on an undecodable value.
pub async fn get_json<T: DeserializeOwned>(redis: &mut ConnectionManager, key: &str) -> Option<T> {
    match redis.get::<_, Option<String>>(key).await {
        Ok(Some(json)) => serde_json::from_str(&json).ok(),
        Ok(None) => None,
        Err(err) => {
            eprintln!("cache read failed: {err}");
            None
        }
    }
}

/// Stores the value and its TTL in one command. A failure is logged, not returned.
pub async fn set_json<T: Serialize>(
    redis: &mut ConnectionManager,
    key: &str,
    value: &T,
    ttl_secs: u64,
) {
    let Ok(json) = serde_json::to_string(value) else {
        return;
    };
    if let Err(err) = redis.set_ex::<_, _, ()>(key, json, ttl_secs).await {
        eprintln!("cache write failed: {err}");
    }
}

pub async fn delete(redis: &mut ConnectionManager, key: &str) {
    if let Err(err) = redis.del::<_, ()>(key).await {
        eprintln!("cache delete failed: {err}");
    }
}

/// `SET key token NX PX ttl_ms`: sets the key only if it does not exist.
/// When Redis fails, the caller proceeds without a lock.
pub async fn try_lock(
    redis: &mut ConnectionManager,
    key: &str,
    token: &str,
    ttl_ms: u64,
) -> bool {
    let reply: redis::RedisResult<Option<String>> = redis::cmd("SET")
        .arg(key)
        .arg(token)
        .arg("NX")
        .arg("PX")
        .arg(ttl_ms)
        .query_async(redis)
        .await;
    match reply {
        Ok(reply) => reply.is_some(),
        Err(err) => {
            eprintln!("lock failed: {err}");
            true
        }
    }
}

/// Deletes the lock only if it still holds this token. The check and the
/// delete run as one Lua script, so no other command runs between them.
pub async fn unlock(redis: &mut ConnectionManager, key: &str, token: &str) {
    let script = redis::Script::new(
        r#"
        if redis.call("GET", KEYS[1]) == ARGV[1] then
            return redis.call("DEL", KEYS[1])
        else
            return 0
        end
        "#,
    );
    let result: redis::RedisResult<i32> = script.key(key).arg(token).invoke_async(redis).await;
    if let Err(err) = result {
        eprintln!("unlock failed: {err}");
    }
}

/// Adds up to 10 percent to a TTL so keys written together do not expire together.
pub fn jittered(base_secs: u64) -> u64 {
    base_secs + rand::random::<u64>() % (base_secs / 10 + 1)
}

src/products.rs

Rust

// src/products.rs
use std::time::Duration;

use axum::{
    extract::{Path, State},
    response::{IntoResponse, Response},
    Json,
};
use redis::aio::ConnectionManager;
use serde::{Deserialize, Serialize};
use uuid::Uuid;

use crate::{cache, error::ApiError, AppState};

const TTL_SECS: u64 = 300;
const MISSING_TTL_SECS: u64 = 30;

#[derive(Clone, Serialize, Deserialize, sqlx::FromRow)]
pub struct Product {
    pub id: i64,
    pub sku: String,
    pub name: String,
    pub price_cents: i32,
}

#[derive(Deserialize)]
pub struct UpdateProduct {
    name: String,
    price_cents: i32,
}

pub fn product_key(id: i64) -> String {
    format!("storefront:v1:product:{id}")
}

async fn find_product(state: &AppState, id: i64) -> Result<Option<Product>, ApiError> {
    let row = sqlx::query_as::<_, Product>(
        "SELECT id, sku, name, price_cents FROM products WHERE id = $1",
    )
    .bind(id)
    .fetch_optional(&state.pool)
    .await?;
    Ok(row)
}

fn respond(product: Option<Product>, source: &'static str) -> Result<Response, ApiError> {
    let product = product.ok_or(ApiError::NotFound)?;
    Ok(([("x-cache", source)], Json(product)).into_response())
}

/// Loads from the database and fills the cache. A missing row is cached briefly.
async fn fill(
    state: &AppState,
    redis: &mut ConnectionManager,
    key: &str,
    id: i64,
) -> Result<Option<Product>, ApiError> {
    let product = find_product(state, id).await?;
    let ttl = if product.is_some() {
        cache::jittered(TTL_SECS)
    } else {
        MISSING_TTL_SECS
    };
    cache::set_json(redis, key, &product, ttl).await;
    Ok(product)
}

pub async fn get_one(
    State(state): State<AppState>,
    Path(id): Path<i64>,
) -> Result<Response, ApiError> {
    let key = product_key(id);
    let mut redis = state.redis.clone();

    if let Some(cached) = cache::get_json::<Option<Product>>(&mut redis, &key).await {
        return respond(cached, "hit");
    }

    // One request per key loads from the database. The others wait for the cache.
    let lock_key = format!("{key}:lock");
    let token = Uuid::new_v4().to_string();
    if cache::try_lock(&mut redis, &lock_key, &token, 5_000).await {
        let loaded = fill(&state, &mut redis, &key, id).await;
        cache::unlock(&mut redis, &lock_key, &token).await;
        return respond(loaded?, "miss");
    }

    for _ in 0..20 {
        tokio::time::sleep(Duration::from_millis(50)).await;
        if let Some(cached) = cache::get_json::<Option<Product>>(&mut redis, &key).await {
            return respond(cached, "wait");
        }
    }
    respond(fill(&state, &mut redis, &key, id).await?, "miss")
}

pub async fn update(
    State(state): State<AppState>,
    Path(id): Path<i64>,
    Json(body): Json<UpdateProduct>,
) -> Result<Json<Product>, ApiError> {
    let updated = sqlx::query_as::<_, Product>(
        "UPDATE products SET name = $2, price_cents = $3 WHERE id = $1
         RETURNING id, sku, name, price_cents",
    )
    .bind(id)
    .bind(&body.name)
    .bind(body.price_cents)
    .fetch_optional(&state.pool)
    .await?
    .ok_or(ApiError::NotFound)?;

    let mut redis = state.redis.clone();
    cache::delete(&mut redis, &product_key(id)).await;
    Ok(Json(updated))
}

src/rate_limit.rs

Rust

// src/rate_limit.rs
use axum::{
    extract::{Request, State},
    http::{header, StatusCode},
    middleware::Next,
    response::{IntoResponse, Response},
    Json,
};
use redis::aio::ConnectionManager;
use serde_json::json;

use crate::AppState;

const LIMIT: i64 = 20;
const WINDOW_SECS: i64 = 60;

/// Counts one request for the client in the current window and returns the count.
/// INCR and EXPIRE run in one script, so a crash cannot leave a counter without an expiry.
async fn count_request(redis: &mut ConnectionManager, client: &str) -> redis::RedisResult<i64> {
    let script = redis::Script::new(
        r#"
        local count = redis.call("INCR", KEYS[1])
        if count == 1 then
            redis.call("EXPIRE", KEYS[1], ARGV[1])
        end
        return count
        "#,
    );
    let key = format!("storefront:v1:rate:{client}");
    script.key(key).arg(WINDOW_SECS).invoke_async(redis).await
}

pub async fn limit(State(state): State<AppState>, request: Request, next: Next) -> Response {
    let client = request
        .headers()
        .get("x-client-id")
        .and_then(|value| value.to_str().ok())
        .unwrap_or("anonymous")
        .to_owned();

    let mut redis = state.redis.clone();
    match count_request(&mut redis, &client).await {
        Ok(count) if count > LIMIT => (
            StatusCode::TOO_MANY_REQUESTS,
            [(header::RETRY_AFTER, WINDOW_SECS.to_string())],
            Json(json!({ "error": "rate limit exceeded" })),
        )
            .into_response(),
        Ok(_) => next.run(request).await,
        Err(err) => {
            // Fail open: a Redis outage does not block traffic.
            eprintln!("rate limiter unavailable: {err}");
            next.run(request).await
        }
    }
}

src/ranking.rs

Rust

// src/ranking.rs
use axum::{
    extract::{Path, State},
    Json,
};
use redis::AsyncCommands;
use serde::{Deserialize, Serialize};

use crate::{error::ApiError, AppState};

const KEY: &str = "storefront:v1:best_sellers";

#[derive(Deserialize)]
pub struct Sale {
    qty: i64,
}

#[derive(Serialize)]
pub struct Ranked {
    product_id: i64,
    sold: f64,
}

pub async fn record_sale(
    State(state): State<AppState>,
    Path(id): Path<i64>,
    Json(body): Json<Sale>,
) -> Result<Json<Ranked>, ApiError> {
    let mut redis = state.redis.clone();
    let sold: f64 = redis.zincr(KEY, id, body.qty).await?;
    Ok(Json(Ranked { product_id: id, sold }))
}

pub async fn best_sellers(State(state): State<AppState>) -> Result<Json<Vec<Ranked>>, ApiError> {
    let mut redis = state.redis.clone();
    let rows: Vec<(i64, f64)> = redis.zrevrange_withscores(KEY, 0, 2).await?;
    let ranked = rows
        .into_iter()
        .map(|(product_id, sold)| Ranked { product_id, sold })
        .collect();
    Ok(Json(ranked))
}

Start the databases and run:

Terminal

docker run --name pg -e POSTGRES_PASSWORD=pw -e POSTGRES_DB=storefront -p 5432:5432 -d postgres:16
docker run --name redis -p 6379:6379 -d redis:7
docker exec -i pg psql -U postgres -d storefront < migrations/0001_products.sql
export DATABASE_URL=postgres://postgres:pw@127.0.0.1:5432/storefront

Terminal

docker exec redis redis-cli flushall
cargo run

Requests:

Terminal

req() { curl -s -w ' [%{http_code}]\n' "$@"; }
J='content-type: application/json'

curl -si localhost:3000/products/1 | grep -i '^x-cache'
curl -si localhost:3000/products/1 | grep -i '^x-cache'
req -X PUT localhost:3000/products/1 -H "$J" -d '{"name":"Ballpoint Pen","price_cents":300}'
curl -si localhost:3000/products/1 | grep -i '^x-cache'
req localhost:3000/products/1
req localhost:3000/products/42
docker exec redis redis-cli get storefront:v1:product:42
req -X POST localhost:3000/products/1/sales -H "$J" -d '{"qty":3}'
req -X POST localhost:3000/products/2/sales -H "$J" -d '{"qty":5}'
req localhost:3000/best-sellers
for i in $(seq 1 22); do
  curl -s -o /dev/null -w '%{http_code}\n' -H 'x-client-id: demo' localhost:3000/products/1
done | sort | uniq -c

Responses on an empty cache:

Output

x-cache: miss
x-cache: hit
{"id":1,"sku":"PEN-1","name":"Ballpoint Pen","price_cents":300} [200]
x-cache: miss
{"id":1,"sku":"PEN-1","name":"Ballpoint Pen","price_cents":300} [200]
{"error":"not found"} [404]
null
{"product_id":1,"sold":3.0} [200]
{"product_id":2,"sold":5.0} [200]
[{"product_id":2,"sold":5.0},{"product_id":1,"sold":3.0}] [200]
     20 200
      2 429

The null line is the negative cache entry for product 42. The rate-limit counter expires after 60 seconds, so rerun the last loop with a new x-client-id value inside that window.

Summary

Redis is a shared in-memory cache reached through a cloneable ConnectionManager.
Cache-aside reads the cache, loads from the database on a miss, and stores the result with a TTL. A Redis error falls back to the database.
Keys carry a namespace and a version. A missing row is cached briefly (negative caching).
Invalidation deletes the key after the database commit. The TTL bounds any missed delete.
Stampede defenses are TTL jitter, a SET NX PX lock released by a token-checking script, and waiting on the cache.
A Redis lock avoids duplicate work and does not guarantee mutual exclusion under pauses.
INCR with EXPIRE in one Lua script makes a fixed-window rate limiter. ZINCRBY maintains a ranking.
Eviction can remove any key, so data that must survive stays in PostgreSQL.