Build Redis from Scratch in Rust — Part 2: Expiry, Persistence, and Pub/Sub

In Part 1, we built a TCP server that speaks the Redis protocol. We implemented SET, GET, and DEL commands with in-memory storage. But real Redis has many more features.

In this part, we add three important features:

Key expiry — keys that delete themselves after a timeout
Persistence — saving data to disk so it survives restarts
Pub/Sub — publish and subscribe messaging between clients

Key Expiry

In real Redis, you can set a key with an expiration time. After that time, the key disappears. This is useful for caches, sessions, and rate limiting.

We need to change our store to track expiry times. Update src/store.rs:

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

#[derive(Clone)]
struct Entry {
    value: String,
    expires_at: Option<Instant>,
}

#[derive(Clone)]
pub struct Store {
    data: Arc<Mutex<HashMap<String, Entry>>>,
}

impl Store {
    pub fn new() -> Self {
        Store {
            data: Arc::new(Mutex::new(HashMap::new())),
        }
    }

    pub fn set(&self, key: String, value: String) {
        let mut data = self.data.lock().unwrap();
        data.insert(
            key,
            Entry {
                value,
                expires_at: None,
            },
        );
    }

    pub fn set_with_expiry(&self, key: String, value: String, ttl: Duration) {
        let mut data = self.data.lock().unwrap();
        data.insert(
            key,
            Entry {
                value,
                expires_at: Some(Instant::now() + ttl),
            },
        );
    }

    pub fn get(&self, key: &str) -> Option<String> {
        let mut data = self.data.lock().unwrap();

        // Check if the key has expired
        if let Some(entry) = data.get(key) {
            if let Some(expires_at) = entry.expires_at {
                if Instant::now() > expires_at {
                    data.remove(key);
                    return None;
                }
            }
            Some(entry.value.clone())
        } else {
            None
        }
    }

    pub fn del(&self, key: &str) -> bool {
        let mut data = self.data.lock().unwrap();
        data.remove(key).is_some()
    }

    pub fn expire(&self, key: &str, ttl: Duration) -> bool {
        let mut data = self.data.lock().unwrap();
        if let Some(entry) = data.get_mut(key) {
            entry.expires_at = Some(Instant::now() + ttl);
            true
        } else {
            false
        }
    }

    pub fn ttl(&self, key: &str) -> i64 {
        let data = self.data.lock().unwrap();
        match data.get(key) {
            None => -2, // key does not exist
            Some(entry) => match entry.expires_at {
                None => -1, // key exists but has no expiry
                Some(expires_at) => {
                    let now = Instant::now();
                    if now > expires_at {
                        -2 // expired
                    } else {
                        (expires_at - now).as_secs() as i64
                    }
                }
            },
        }
    }

    /// Get all key-value pairs for persistence. Skips expired keys.
    pub fn get_all(&self) -> Vec<(String, String)> {
        let data = self.data.lock().unwrap();
        let now = Instant::now();
        data.iter()
            .filter(|(_, entry)| {
                entry.expires_at.map_or(true, |exp| now <= exp)
            })
            .map(|(k, v)| (k.clone(), v.value.clone()))
            .collect()
    }

    /// Load data from persistence. Sets keys without expiry.
    pub fn load(&self, pairs: Vec<(String, String)>) {
        let mut data = self.data.lock().unwrap();
        for (key, value) in pairs {
            data.insert(
                key,
                Entry {
                    value,
                    expires_at: None,
                },
            );
        }
    }
}

The key change is the Entry struct. Each entry now has an optional expires_at time. When we read a key, we check if it has expired. This is called lazy expiration — we only check when someone accesses the key. Real Redis also uses this strategy, along with periodic cleanup.

The ttl method returns:

-2 if the key does not exist
-1 if the key exists but has no expiry
The remaining seconds otherwise

This matches the real Redis TTL command behavior.

New Commands

Now update src/command.rs to add EXPIRE, TTL, and SET with EX option:

use std::time::Duration;

use crate::resp::RespValue;
use crate::store::Store;

pub fn handle_command(value: RespValue, store: &Store) -> RespValue {
    let args = match value {
        RespValue::Array(args) => args,
        _ => return RespValue::Error("ERR expected array".to_string()),
    };

    if args.is_empty() {
        return RespValue::Error("ERR empty command".to_string());
    }

    let command = match &args[0] {
        RespValue::BulkString(s) => s.to_uppercase(),
        _ => return RespValue::Error("ERR invalid command format".to_string()),
    };

    match command.as_str() {
        "PING" => handle_ping(&args),
        "ECHO" => handle_echo(&args),
        "SET" => handle_set(&args, store),
        "GET" => handle_get(&args, store),
        "DEL" => handle_del(&args, store),
        "EXPIRE" => handle_expire(&args, store),
        "TTL" => handle_ttl(&args, store),
        "SAVE" => handle_save(store),
        "COMMAND" => RespValue::SimpleString("OK".to_string()),
        _ => RespValue::Error(format!("ERR unknown command '{}'", command)),
    }
}

fn handle_ping(args: &[RespValue]) -> RespValue {
    if args.len() > 1 {
        if let RespValue::BulkString(msg) = &args[1] {
            return RespValue::BulkString(msg.clone());
        }
    }
    RespValue::SimpleString("PONG".to_string())
}

fn handle_echo(args: &[RespValue]) -> RespValue {
    if args.len() < 2 {
        return RespValue::Error("ERR wrong number of arguments for 'echo'".to_string());
    }
    if let RespValue::BulkString(msg) = &args[1] {
        RespValue::BulkString(msg.clone())
    } else {
        RespValue::Error("ERR invalid argument".to_string())
    }
}

fn handle_set(args: &[RespValue], store: &Store) -> RespValue {
    if args.len() < 3 {
        return RespValue::Error(
            "ERR wrong number of arguments for 'set'".to_string(),
        );
    }

    let key = match &args[1] {
        RespValue::BulkString(s) => s.clone(),
        _ => return RespValue::Error("ERR invalid key".to_string()),
    };

    let value = match &args[2] {
        RespValue::BulkString(s) => s.clone(),
        _ => return RespValue::Error("ERR invalid value".to_string()),
    };

    // Check for EX option: SET key value EX seconds
    if args.len() >= 5 {
        if let RespValue::BulkString(option) = &args[3] {
            if option.to_uppercase() == "EX" {
                if let RespValue::BulkString(seconds_str) = &args[4] {
                    if let Ok(seconds) = seconds_str.parse::<u64>() {
                        store.set_with_expiry(
                            key,
                            value,
                            Duration::from_secs(seconds),
                        );
                        return RespValue::SimpleString("OK".to_string());
                    }
                }
                return RespValue::Error("ERR invalid expire time".to_string());
            }
        }
    }

    store.set(key, value);
    RespValue::SimpleString("OK".to_string())
}

fn handle_get(args: &[RespValue], store: &Store) -> RespValue {
    if args.len() < 2 {
        return RespValue::Error(
            "ERR wrong number of arguments for 'get'".to_string(),
        );
    }

    let key = match &args[1] {
        RespValue::BulkString(s) => s.as_str(),
        _ => return RespValue::Error("ERR invalid key".to_string()),
    };

    match store.get(key) {
        Some(value) => RespValue::BulkString(value),
        None => RespValue::Null,
    }
}

fn handle_del(args: &[RespValue], store: &Store) -> RespValue {
    if args.len() < 2 {
        return RespValue::Error(
            "ERR wrong number of arguments for 'del'".to_string(),
        );
    }

    let mut count = 0;
    for arg in &args[1..] {
        if let RespValue::BulkString(key) = arg {
            if store.del(key) {
                count += 1;
            }
        }
    }
    RespValue::Integer(count)
}

fn handle_expire(args: &[RespValue], store: &Store) -> RespValue {
    if args.len() < 3 {
        return RespValue::Error(
            "ERR wrong number of arguments for 'expire'".to_string(),
        );
    }

    let key = match &args[1] {
        RespValue::BulkString(s) => s.as_str(),
        _ => return RespValue::Error("ERR invalid key".to_string()),
    };

    let seconds = match &args[2] {
        RespValue::BulkString(s) => match s.parse::<u64>() {
            Ok(n) => n,
            Err(_) => return RespValue::Error("ERR invalid seconds".to_string()),
        },
        _ => return RespValue::Error("ERR invalid argument".to_string()),
    };

    if store.expire(key, Duration::from_secs(seconds)) {
        RespValue::Integer(1)
    } else {
        RespValue::Integer(0)
    }
}

fn handle_ttl(args: &[RespValue], store: &Store) -> RespValue {
    if args.len() < 2 {
        return RespValue::Error(
            "ERR wrong number of arguments for 'ttl'".to_string(),
        );
    }

    let key = match &args[1] {
        RespValue::BulkString(s) => s.as_str(),
        _ => return RespValue::Error("ERR invalid key".to_string()),
    };

    RespValue::Integer(store.ttl(key))
}

fn handle_save(store: &Store) -> RespValue {
    match crate::persistence::save_to_disk(store) {
        Ok(_) => RespValue::SimpleString("OK".to_string()),
        Err(e) => RespValue::Error(format!("ERR save failed: {}", e)),
    }
}

Now you can set keys with expiry:

127.0.0.1:6379> SET session abc123 EX 60
OK
127.0.0.1:6379> TTL session
(integer) 59
127.0.0.1:6379> EXPIRE session 10
(integer) 1
127.0.0.1:6379> TTL session
(integer) 9

Persistence

Redis can save data to disk. The simplest approach is called snapshotting — write all key-value pairs to a file. Real Redis uses a binary format called RDB. We keep it simple and use a line-based text format.

Create src/persistence.rs:

use std::fs;
use std::io;
use std::path::Path;

use crate::store::Store;

const DATA_FILE: &str = "mini-redis.db";

pub fn save_to_disk(store: &Store) -> io::Result<()> {
    let pairs = store.get_all();
    let mut content = String::new();

    for (key, value) in &pairs {
        // Simple format: key\tvalue\n
        // We escape tabs and newlines in keys/values
        let escaped_key = key.replace('\\', "\\\\").replace('\t', "\\t").replace('\n', "\\n");
        let escaped_value = value.replace('\\', "\\\\").replace('\t', "\\t").replace('\n', "\\n");
        content.push_str(&escaped_key);
        content.push('\t');
        content.push_str(&escaped_value);
        content.push('\n');
    }

    // Write to a temp file first, then rename for atomic write
    let tmp_file = format!("{}.tmp", DATA_FILE);
    fs::write(&tmp_file, &content)?;
    fs::rename(&tmp_file, DATA_FILE)?;

    println!("Saved {} keys to {}", pairs.len(), DATA_FILE);
    Ok(())
}

pub fn load_from_disk(store: &Store) -> io::Result<()> {
    let path = Path::new(DATA_FILE);
    if !path.exists() {
        println!("No data file found, starting fresh");
        return Ok(());
    }

    let content = fs::read_to_string(path)?;
    let mut pairs = Vec::new();

    for line in content.lines() {
        if line.is_empty() {
            continue;
        }
        let parts: Vec<&str> = line.splitn(2, '\t').collect();
        if parts.len() == 2 {
            let key = parts[0].replace("\\t", "\t").replace("\\n", "\n").replace("\\\\", "\\");
            let value = parts[1].replace("\\t", "\t").replace("\\n", "\n").replace("\\\\", "\\");
            pairs.push((key, value));
        }
    }

    let count = pairs.len();
    store.load(pairs);
    println!("Loaded {} keys from {}", count, DATA_FILE);
    Ok(())
}

Notice the atomic write pattern. We write to a .tmp file first, then rename it. This prevents data corruption if the process crashes during the write. Real Redis uses the same strategy.

Adding Pub/Sub

Pub/Sub lets clients subscribe to channels and receive messages published by other clients. This is one of the most interesting Redis features.

We need a new structure to manage subscriptions. Create src/pubsub.rs:

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use tokio::sync::broadcast;

const CHANNEL_CAPACITY: usize = 100;

#[derive(Clone)]
pub struct PubSub {
    channels: Arc<Mutex<HashMap<String, broadcast::Sender<String>>>>,
}

impl PubSub {
    pub fn new() -> Self {
        PubSub {
            channels: Arc::new(Mutex::new(HashMap::new())),
        }
    }

    pub fn subscribe(&self, channel: &str) -> broadcast::Receiver<String> {
        let mut channels = self.channels.lock().unwrap();
        let sender = channels
            .entry(channel.to_string())
            .or_insert_with(|| {
                let (tx, _) = broadcast::channel(CHANNEL_CAPACITY);
                tx
            });
        sender.subscribe()
    }

    pub fn publish(&self, channel: &str, message: String) -> usize {
        let channels = self.channels.lock().unwrap();
        if let Some(sender) = channels.get(channel) {
            // send returns Err if there are no receivers, which is fine
            sender.send(message).unwrap_or(0)
        } else {
            0
        }
    }
}

We use Tokio’s broadcast channel. It lets one sender deliver messages to multiple receivers. Each subscriber gets their own receiver.

Now we need to handle SUBSCRIBE and PUBLISH commands. Update src/main.rs to support pub/sub mode:

mod command;
mod persistence;
mod pubsub;
mod resp;
mod store;

use command::handle_command;
use pubsub::PubSub;
use resp::{RespParser, RespValue};
use store::Store;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let store = Store::new();
    let pubsub = PubSub::new();

    // Load data from disk on startup
    persistence::load_from_disk(&store)?;

    let listener = TcpListener::bind("127.0.0.1:6379").await?;
    println!("Mini-Redis server listening on 127.0.0.1:6379");

    loop {
        let (mut socket, addr) = listener.accept().await?;
        let store = store.clone();
        let pubsub = pubsub.clone();

        tokio::spawn(async move {
            println!("New connection from {}", addr);
            let mut buf = vec![0u8; 4096];

            loop {
                let n = match socket.read(&mut buf).await {
                    Ok(0) => {
                        println!("Connection closed: {}", addr);
                        return;
                    }
                    Ok(n) => n,
                    Err(e) => {
                        eprintln!("Read error from {}: {}", addr, e);
                        return;
                    }
                };

                let data = buf[..n].to_vec();
                let mut parser = RespParser::new(data);

                let value = match parser.parse() {
                    Ok(v) => v,
                    Err(e) => {
                        let err = RespValue::Error(format!("ERR parse error: {}", e));
                        let _ = socket.write_all(err.serialize().as_bytes()).await;
                        continue;
                    }
                };

                // Check if this is a SUBSCRIBE command
                if is_subscribe_command(&value) {
                    handle_subscribe(&mut socket, &value, &pubsub).await;
                    return; // subscribe takes over the connection
                }

                // Check if this is a PUBLISH command
                if is_publish_command(&value) {
                    let response = handle_publish(&value, &pubsub);
                    let _ = socket.write_all(response.serialize().as_bytes()).await;
                    continue;
                }

                let response = handle_command(value, &store);
                if let Err(e) = socket.write_all(response.serialize().as_bytes()).await {
                    eprintln!("Write error to {}: {}", addr, e);
                    return;
                }
            }
        });
    }
}

fn is_subscribe_command(value: &RespValue) -> bool {
    if let RespValue::Array(args) = value {
        if let Some(RespValue::BulkString(cmd)) = args.first() {
            return cmd.to_uppercase() == "SUBSCRIBE";
        }
    }
    false
}

fn is_publish_command(value: &RespValue) -> bool {
    if let RespValue::Array(args) = value {
        if let Some(RespValue::BulkString(cmd)) = args.first() {
            return cmd.to_uppercase() == "PUBLISH";
        }
    }
    false
}

async fn handle_subscribe(
    socket: &mut tokio::net::TcpStream,
    value: &RespValue,
    pubsub: &PubSub,
) {
    let args = match value {
        RespValue::Array(args) => args,
        _ => return,
    };

    if args.len() < 2 {
        let err = RespValue::Error("ERR wrong number of arguments for 'subscribe'".to_string());
        let _ = socket.write_all(err.serialize().as_bytes()).await;
        return;
    }

    let channel = match &args[1] {
        RespValue::BulkString(s) => s.clone(),
        _ => return,
    };

    let mut receiver = pubsub.subscribe(&channel);

    // Send subscribe confirmation
    let confirm = RespValue::Array(vec![
        RespValue::BulkString("subscribe".to_string()),
        RespValue::BulkString(channel.clone()),
        RespValue::Integer(1),
    ]);
    let _ = socket.write_all(confirm.serialize().as_bytes()).await;

    // Forward messages to the client
    loop {
        match receiver.recv().await {
            Ok(message) => {
                let msg = RespValue::Array(vec![
                    RespValue::BulkString("message".to_string()),
                    RespValue::BulkString(channel.clone()),
                    RespValue::BulkString(message),
                ]);
                if socket.write_all(msg.serialize().as_bytes()).await.is_err() {
                    return; // client disconnected
                }
            }
            Err(_) => return,
        }
    }
}

fn handle_publish(value: &RespValue, pubsub: &PubSub) -> RespValue {
    let args = match value {
        RespValue::Array(args) => args,
        _ => return RespValue::Error("ERR expected array".to_string()),
    };

    if args.len() < 3 {
        return RespValue::Error(
            "ERR wrong number of arguments for 'publish'".to_string(),
        );
    }

    let channel = match &args[1] {
        RespValue::BulkString(s) => s.as_str(),
        _ => return RespValue::Error("ERR invalid channel".to_string()),
    };

    let message = match &args[2] {
        RespValue::BulkString(s) => s.clone(),
        _ => return RespValue::Error("ERR invalid message".to_string()),
    };

    let count = pubsub.publish(channel, message);
    RespValue::Integer(count as i64)
}

Testing Pub/Sub

Start the server and open three terminals.

Terminal 1 — subscribe:

redis-cli
127.0.0.1:6379> SUBSCRIBE news
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "news"
3) (integer) 1

Terminal 2 — publish:

redis-cli
127.0.0.1:6379> PUBLISH news "Hello from Rust!"
(integer) 1

Terminal 1 receives:

1) "message"
2) "news"
3) "Hello from Rust!"

Terminal 3 — you can subscribe too, and both terminals will receive the message.

Test Persistence

127.0.0.1:6379> SET language Rust
OK
127.0.0.1:6379> SET framework Tokio
OK
127.0.0.1:6379> SAVE
OK

Stop the server with Ctrl+C. Start it again. Your data is still there:

127.0.0.1:6379> GET language
"Rust"
127.0.0.1:6379> GET framework
"Tokio"

Project Structure

mini-redis/
  Cargo.toml
  src/
    main.rs          — TCP server with pub/sub handling
    resp.rs          — RESP protocol parser
    store.rs         — key-value storage with expiry
    command.rs       — command handlers (SET, GET, DEL, EXPIRE, TTL, SAVE)
    persistence.rs   — save/load data to disk
    pubsub.rs        — publish/subscribe messaging

What’s Next?

In Part 3, we add:

List commands: LPUSH, LPOP, LRANGE
INCR command for atomic counters
Benchmarks against real Redis
Error handling and graceful shutdown

We also measure how our Rust implementation compares to real Redis in performance.

Part 1: TCP Server and Commands — the foundation we built on
Rust Tutorial: Async/Await and Tokio — understanding the async runtime
Rust Tutorial: Error Handling — Result patterns used throughout
Rust Tutorial: Modules and Packages — how we organize our code into modules

Key Expiry#

New Commands#

Persistence#

Adding Pub/Sub#

Testing Pub/Sub#

Test Persistence#

Project Structure#

What’s Next?#

Related Articles#