September « 2026 « Shahzad Bhatti

September 1, 2026

Write a Redis Clone with Virtual Actors

Filed under: Computing — admin @ 11:27 am

I recently read Rust Projects – Write a Redis Clone book that builds a real Redis-compatible server from scratch in async Rust. It’s a good book that hand-rolled wire protocol using actor like abstractions. This inspired me to show how Redis clone can be built with PlexSpaces, the distributed actor framework I’ve been building. The code examples in the book used raw Tokio: an mpsc::channel for the actor mailbox, tokio::spawn for every connection, and replica examples used tokio::select! loops for fan-out. PlexSpaces abstracts that the kind of plumbing so I rebuilt the same Redis subset like storage, expiry, replication, transactions as PlexSpaces actors, once in Rust and once in Python compiled to WebAssembly, then benchmarked it against a real two-node gRPC cluster. This post walks through key abstractions I used to simplify the implementation of Redis clone.

What is Redis?

Redis is a single-threaded, in-memory key-value store. A single thread processes every command without parallelism and coordination inside the store, locking, and transactions. This makes it fast as nothing contends for anything. Here are its core capabilities:

The wire protocol. Redis speaks RESP (Redis Serialization Protocol), a binary format: type prefixes (+ for simple strings, $ for bulk strings, * for arrays), length prefixes, \r\n terminators. ECHO HELLO on the wire looks like *2\r\n$4\r\nECHO\r\n$5\r\nHELLO\r\n.
Persistence. It offers two durability modes. RDB takes periodic snapshots of the whole keyspace to disk. AOF (Append-Only File) logs every write command and replays the log on restart.
Replication. A master streams write commands to replicas. The handshake is three steps: the replica sends PING (master replies PONG), then REPLCONF (negotiates parameters), then PSYNC (triggers a full sync). After that, every write streams to replicas as it happens. WAIT blocks until N replicas confirm receipt.
Key expiry. Per-key TTLs, handled two ways: passively (check on GET, return nil if expired) and actively (a background scan periodically deletes expired keys).
Transactions. MULTI starts a queue; every command after it gets queued instead of executed. EXEC runs the whole queue atomically. DISCARD cancels. The guarantee is serialization but there’s no rollback on an individual command failure.

The Book: Working Redis Clone

Here’s a short overview of each chapter of the book:

Chapter	What it builds
Ch1: TCP bind	`TcpListener::bind`, `accept()` in a loop, spawn a task per connection.
Ch2: RESP parsing	A custom result type for partial parses, a test harness, scanning for `\r\n`, identifying type-prefix bytes, parsing simple strings, bulk strings, and arrays, etc.
Ch3: Storage	`GET`, `SET`, `DEL` against a `HashMap<String, String>` behind a `Mutex`.
Ch4: Key expiry	`SET key value EX seconds` / `PX milliseconds`. Store a creation timestamp per value; check it passively on GET; sweep expired keys actively with a background task.
Ch5: The actor pattern	This chapter swap the mutex-guarded `HashMap` for an `mpsc::channel(32)`: one storage actor owns the data and processes messages sequentially. Connection handlers send messages and wait for replies without locks.
Ch6: Command modules	Refactor the growing `match` in the connection handler into separate modules (strings, server commands, etc.).
Ch7–8: Replication	The three-step handshake (PING -> REPLCONF -> PSYNC). The master ships an RDB-equivalent snapshot to each new replica, then streams every write command to all replica senders in a fan-out loop. Chapter 8 adds `WAIT`: block until N replicas confirm their replication offset, via a `tokio::select!` loop.
Ch9: Transactions and INCR	`INCR` with create-if-missing and error-if-non-integer semantics. `MULTI` / `EXEC` / `DISCARD` with per-connection state. `EXEC` drains the queue atomically.

The Core Insight

The chapter 5 that showed how an actor owns one dataset, processes one message at a time without locking. However, it used fairly low-level APIs and only supported an architecture of one actor per machine. What if you had N actors across multiple nodes? That’s exactly what PlexSpaces‘ create_shard_group gives you. So I created an equivalent examples where PlexSpaces spins up N copies (StorageActors), hash-partitions the keyspace across them, and routes each operation to the shard that owns it. The application code never touches partitioning, routing, placement, or cross-node communication. It just calls set(key, value).

Here are five primitives in PlexSpaces that do all the distributed heavy lifting in this example:

create_shard_group: spins up N actor instances, hash-partitioned, placed across nodes.
bulk_update_shard_group: routes a batch of writes to the shard that owns each key.
scatter_gather: fans a query out to shards and collects responses, with a min_responses threshold and timeout.
broadcast_shard_group: sends the same message to every shard (replication, expiry sweeps, handshakes).
map_shard_group / reduce_shard_group: runs an operation on every shard in parallel and collects (map) or aggregates (reduce) the results.

Following diagram shows mapping of low-level Tokio implementation to above five calls:

The handler declarations stay almost identical in spirit but simpler (instead of a match-tree/HashMap):

Rust Implementation

#[plexspaces_handlers(gen_server)]
impl StorageActor {
#[handler(“get”)]
async fn handle_get(&mut self, _ctx: &ActorContext, msg: &Message)
-> Result<Value, BehaviorError> {
let key = msg.payload_json()?[“key”].as_str().unwrap_or(“”).to_string();
if let Some(entry) = self.store.get(&key) {
// passive expiry check
if let Some(exp) = entry.expires_at_ms {
if now_ms() > exp { self.store.remove(&key); return Ok(json!({“found”: false})); }
}
Ok(json!({“found”: true, “result”: entry.value}))
} else {
Ok(json!({“found”: false}))
}
}

/// SET key value [NX|XX] [EX seconds | PX millis] (Ch4).
#[handler(“set”)]
async fn handle_set(&mut self, _ctx: &ActorContext, msg: &Message) -> Result<Value, BehaviorError> {
#[derive(Deserialize)]
struct SetPayload {
key: String,
value: String,
#[serde(default)] nx: bool,
#[serde(default)] xx: bool,
#[serde(default)] ex: Option<u64>,
#[serde(default)] px: Option<u64>,
}
let p: SetPayload = serde_json::from_slice(&msg.payload)
.map_err(|e| BehaviorError::ProcessingError(format!(“bad payload: {}”, e)))?;

// NX: only if not exists
if p.nx && self.data.contains_key(&p.key) {
return Ok(json!({ “result”: null, “ok”: false }));
}
// XX: only if exists
if p.xx && !self.data.contains_key(&p.key) {
return Ok(json!({ “result”: null, “ok”: false }));
}

let expires_at_ms = if let Some(ex) = p.ex {
Some(now_ms() + ex * 1000)
} else if let Some(px) = p.px {
Some(now_ms() + px)
} else {
None
};

self.data.insert(p.key, StoredEntry { value: p.value, expires_at_ms });
self.replication_offset += 1;
Ok(json!({ “result”: “OK”, “ok”: true }))
}

/// INCR key — create with 1 if missing; error if value is not an integer (Ch9).
#[handler(“incr”)]
async fn handle_incr(&mut self, _ctx: &ActorContext, msg: &Message) -> Result<Value, BehaviorError> {
let payload: Value = serde_json::from_slice(&msg.payload)
.map_err(|e| BehaviorError::ProcessingError(format!(“bad payload: {}”, e)))?;
let key = payload.get(“key”).and_then(|v| v.as_str()).unwrap_or(“”).to_string();

// Passive expiry
if self.data.get(&key).map(is_expired).unwrap_or(false) {
self.data.remove(&key);
}

let new_val = match self.data.get(&key) {
None => 1i64,
Some(entry) => {
match entry.value.parse::<i64>() {
Ok(n) => n + 1,
Err(_) => return Ok(json!({
“result”: null,
“error”: “ERR value is not an integer or out of range”
})),
}
}
};

self.data.insert(key, StoredEntry { value: new_val.to_string(), expires_at_ms: None });
self.replication_offset += 1;
Ok(json!({ “result”: new_val, “error”: null }))
}

/// DEL key — remove key, return count deleted.
#[handler(“del”)]
async fn handle_del(&mut self, _ctx: &ActorContext, msg: &Message) -> Result<Value, BehaviorError> {
let payload: Value = serde_json::from_slice(&msg.payload)
.map_err(|e| BehaviorError::ProcessingError(format!(“bad payload: {}”, e)))?;
let key = payload.get(“key”).and_then(|v| v.as_str()).unwrap_or(“”);
let deleted = if self.data.remove(key).is_some() {
self.replication_offset += 1;
1
} else {
0
};
Ok(json!({ “result”: deleted }))
}

}

Python Implementation

@actor
class StorageActor:

    @handler("get")
    def handle_get(self, key: str = "") -> dict:
        entry = self.data.get(key)
        if entry is None:
            return {"result": None, "found": False}
        if is_expired(entry):
            del self.data[key]
            return {"result": None, "found": False}
        return {"result": entry["value"], "found": True}

    @handler("set")
    def handle_set(
        self,
        key: str = "",
        value: str = "",
        nx: bool = False,
        xx: bool = False,
        ex: Optional[int] = None,
        px: Optional[int] = None,
    ) -> dict:
        if self.num_shards > 1 and not self._owns_key(key):
            return {"result": None, "skip": True}
        if nx and key in self.data:
            return {"result": None, "ok": False}
        if xx and key not in self.data:
            return {"result": None, "ok": False}

        expires_at_ms: Optional[int] = None
        if ex is not None:
            expires_at_ms = now_ms() + ex * 1000
        elif px is not None:
            expires_at_ms = now_ms() + px

        self.data[key] = {"value": value, "expires_at_ms": expires_at_ms}
        self.replication_offset += 1
        return {"result": "OK", "ok": True}

    @handler("incr")
    def handle_incr(self, key: str = "") -> dict:
        if self.num_shards > 1 and not self._owns_key(key):
            return {"result": None, "skip": True}
        entry = self.data.get(key)
        if entry is not None and is_expired(entry):
            del self.data[key]
            entry = None

        if entry is None:
            new_val = 1
        else:
            try:
                new_val = int(entry["value"]) + 1
            except (ValueError, TypeError):
                return {
                    "result": None,
                    "error": "ERR value is not an integer or out of range",
                }

        self.data[key] = {"value": str(new_val), "expires_at_ms": None}
        self.replication_offset += 1
        return {"result": new_val, "error": None}

    @handler("del")
    def handle_del(self, key: str = "") -> dict:
        if self.num_shards > 1 and not self._owns_key(key):
            return {"result": 0, "skip": True}
        deleted = 1 if self.data.pop(key, None) is not None else 0
        if deleted:
            self.replication_offset += 1
        return {"result": deleted}

Chapter 2 Disappears

Chapter 2 is entirely about parsing RESP: 14 steps for byte-scanning and test harness. In PlexSpaces, this code disappears as the framework handles serialization, routing, and delivery.

In PlexSpaces, actors talk over JSON. A set command looks like so entire chapter disappears:

{"op": "set", "key": "user:1", "value": "alice", "ex": 300}

Replication

The book’s replication fan-out looks roughly like this:

// Book Ch7-8 — manual fan-out per write
for replica in &self.replicas {
    let tx = replica.sender.clone();
    let cmd = replication_event.clone();
    tokio::spawn(async move { tx.send(cmd).await.ok(); });
}

Each replica gets its own spawned task without retries, timeout or automated ACK tracker. With PlexSpaces:

// One call fans out to all replica shards, collects all ACKs
let ack_count = cluster.propagate_to_replicas("SET", "replicated:key", "hello", 1).await?;
// Replication: write propagated to all 3 replica shards via broadcast

Under the hood, broadcast_shard_group fans out to every shard in the replica group, collects responses, handles timeouts, and returns. Here is how the chapter 8 implements WAIT:

// Book Ch8 — manual WAIT implementation
let mut confirmed = 0;
let deadline = Instant::now() + Duration::from_millis(timeout_ms);
while confirmed < num_replicas && Instant::now() < deadline {
    tokio::select! {
        Some(ack) = rx.recv() => {
            if ack.offset >= required_offset { confirmed += 1; }
        }
        _ = tokio::time::sleep_until(deadline.into()) => break,
    }
}

Here is equivalent implementation in PlexSpaces:


let acks = cluster.wait(2, 5000).await?;
// scatter_gather collected ACKs from 3 replica shards

Transactions Without Locks (Ch9)

Chapter 9 introduces MULTI / EXEC / DISCARD via per-connection state:

// Book Ch9
struct ConnectionState {
    in_multi: bool,
    queue: Vec<Command>,
}

This is straightforward for a a single process but in a distributed environment, connections might route to different servers so you need to track transaction state. PlexSpaces solves this with virtual actors, which are inspired by Orleans Actors, i.e., one ConnectionActor per client, created lazily on first call.

Each actor processes one message at a time without locks, so in_multi and queue are just plain struct fields: MULTI -> SET -> EXEC arrive in order at the same actor without mutext or atomics. Virtual actors spin up on first message and get garbage-collected when idle, so there’s no connection map to maintain and no cleanup to do on disconnect.

Throughput Numbers

Here are numbers from rudimentary benchmarks that produced: 20 batches of 50 keys each via bulk_update_shard_group, plus 50 individual GETs, against a 3-shard group spread across two real gRPC nodes:

Throughput Benchmark Results

| Operation  |  TPS      |  p50 (µs) |  p95 (µs) |  p99 (µs) |
|------------|-----------|-----------|-----------|-----------|
|  SET (bulk) |      3200 |       420 |       890 |     1240 |
|  GET        |      1800 |       510 |       980 |     1450 |

1000 SET keys in 312ms ? 3200 SET/sec via bulk_update_shard_group

(each bulk_update fans out to 3 shards in parallel)

The Python WASM version reports the same shape of numbers through host.application_metrics_add():

Redis Cluster Throughput (3-shard group, 2-node gRPC cluster)

| Operation  |  TPS      |  p50 (ms) |  p99 (ms) |  Notes |
|------------|-----------|-----------|-----------|-----------|
|  SET (bulk) |      2100 |     0.45 |      1.30 |     50 keys/b |
|  GET        |      1200 |     0.55 |      1.60 |     individual |

Python WASM numbers are somewhat lower than Rust’s because WASM compilation adds overhead per handler invocation. But the architecture is identical: same PlexSpaces primitives, same shard group, same gRPC routing underneath.

The Python WASM Version

The same cluster logic also runs as Python actors compiled to WASM. No TCP socket without RESP parser or Tokio using the same broadcast_shard_group, scatter_gather, reduce_shard_group, and map_shard_group calls:

@actor
class RedisCoordinator:
    num_shards: int = state(default=3)
    total_coord_ms: float = state(default=0.0)

    @handler("replicate")
    def replicate(self, command: str = "", key: str = "", value: str = "", offset: int = 0) -> dict:
        t0 = time.time()
        resp = host.broadcast_shard_group({
            "group_id": "redis-replicas",
            "payload": {"op": "replicate", "command": command, "key": key, "value": value, "offset": offset},
            "timeout_ms": 5000,
        })
        coord_ms = (time.time() - t0) * 1000
        self.total_coord_ms += coord_ms
        host.application_metrics_add("redis-cluster", {
            "message_count": 1,
            "counter_metrics": {"replication_calls": 1},
            "latency_totals_ms": {"coord": int(self.total_coord_ms)},
            "latency_max_ms": {"coord": int(coord_ms)},
            "latency_samples": {"coord": 1},
        })
        return {"result": "OK", "acks": len(resp.get("shard_responses", []))}

This compiles to WASM, deploys to a running PlexSpaces node over HTTP, and runs against a live cluster. The host.* calls map to the exact same primitives the Rust version calls such as fan-out, collect, timeout, all identical in semantics. The full source, including StorageActor, ConnectionActor, RedisCoordinator, and the BenchmarkActor, is in the redis_cluster example.

The Lines That Disappeared

What	Book (~650 lines)	PlexSpaces Rust (~280 lines)	PlexSpaces Python (~300 lines)
RESP protocol parser	~120 lines (full Ch2)	0 (JSON messages)	0 (SON messages)
TCP accept loop	~40 lines	0 (actor mailbox)	0 (actor mailbox)
Connection tracking map	~30 lines	0 (virtual actor lifecycle)	0 (virtual actor lifecycle)
MPSC channel setup	~20 lines	0 (actor framework)	0 (actor framework)
Replica list management	~40 lines	0 (`broadcast_shard_group`)	0 (`host.broadcast_shard_group`)
WAIT loop (`tokio::select!`)	~50 lines	~3 lines (`scatter_gather`)	~8 lines (`host.scatter_gather`)
Manual fan-out per replica	~30 lines	~5 lines `(broadcast_shard_group`)	~8 lines
Shard routing / partitioning	0 (single node)	~1 line (`partition_strategy: hash`)	~1 line
Multi-node placement	0 (single node)	~2 lines ( `NodePlacement::Specific`)	~2 lines
Coordinated snapshot	Missing	~5 lines (`map_shard_group`)	~8 lines
Active expiry broadcast	Missing	~5 lines (`broadcast_shard_group`)	~8 lines

In PlexSpaces implementation includes the StorageActor handlers (get, set, incr, del, expiry logic, replication handlers), the ConnectionActor MULTI/EXEC/DISCARD state machine, and the cluster setup logic.

How to Test Everything

Rust embedded example

cd examples/rust/embedded/redis_cluster

# Run the demo directly — prints all 11 steps with coord_ms timing:
cargo run --bin redis_cluster

# Or run the full validated test suite:
./scripts/test.sh

Python WASM example

Prerequisites: a running PlexSpaces node on port 8091 (and optionally 8093 for multi-node), Python 3.10+, and the plexspaces-py CLI.

cd examples/python/apps/redis_cluster

# Build actors to WASM:
./build.sh

# Deploy, initialize, and test all 11 steps (single-node):
./test.sh
# or explicitly: ./test.sh 8091

# Multi-node (shards distributed across both nodes):
./test.sh 8091 8093

Above example runs two nodes on ports 8091 and 803. The create_shard_group uses from_registry placement to spread shard actors across both nodes automatically. Every collective operation like scatter_gather, reduce, map, broadcast routes cross-node over gRPC, using the same primitives whether the shards are local or remote.

Fixed ports, on purpose

The Rust embedded example starts two in-process nodes on fixed ports, :8091 and :8093, rather than ephemeral ones:

redis-master-node  ->  gRPC :8091
redis-replica-node ->  gRPC :8093

Both examples use the same ports, so you can point the Python test.sh at a cluster the Rust example already stood up, or vice versa. Run ./scripts/test.sh for the Rust side.

What the test scripts actually check

The Rust scripts/test.sh looks for: “Cluster ready”, “Basic operations”, “broadcast_shard_group”, “scatter_gather”, “reduce”, “map + concat”, “parallel map”, “Multi-Node”, “Throughput Benchmark”, “SET/sec”, “p50”, “Example Complete”.

The Python test.sh checks: a setup response with "status".*"ok", GET/SET/INCR semantics, an expired key returning "found".*false, replication returning "acks", a snapshot returning "shard_count" and "shards", and the benchmark returning "tps" and "p50_ms".

Learnings

The purpose of the book was to teach how Redis works using Rust. I rebuilt it on PlexSpaces to show how abstractions can simplify building complex distributed applications. For example, the RESP parser, the connection lifecycle, the replica list, and the WAIT loop are not your job. The redis book teaches you to hand-roll patterns like an mpsc mailbox, a spawn-per-connection loop, a tokio::select! deadline race. ThePlexSpaces uses many of same primitives to define high level abstractions but it provides an actor runtime that hides all complexity. This allows you to build the actual business logic like the storage logic, the replication semantics, the transaction model. Actors just give you a place to put them that scales horizontally without rewriting any of the logic itself. The throughput numbers show rough cost of the actor primitives, e.g., cluster setup is expensive, so you pay it once; individual operations are fast. With p50s in the hundreds of microseconds; bulk operations get much cheaper per key as the coordination overhead amortizes over more keys. This is how distributed systems work in general so you need to measure performance overhead.

GitHub: https://github.com/bhatti/PlexSpaces

Shahzad Bhatti Welcome to my ramblings and rants!