kv-server

A Redis-compatible key-value store written from scratch in Rust, extended with a Raft consensus layer for distributed operation across multiple nodes.

Clients connect using any Redis client (redis-cli, any Redis library) over TCP. In cluster mode, writes are replicated to a quorum of nodes before being acknowledged — data survives individual node failures.

---

Running it

Standalone (single node)

cargo run

Connect with the Redis CLI:

redis-cli -p 6379

To require a password:

KV_PASSWORD=secret cargo run
redis-cli -p 6379 -a secret

Cluster mode (3-node example)

Each node is a separate process. Open three terminals:

Node 1

RAFT_NODE_ID=1 \
RAFT_ADDR=127.0.0.1:7001 \
RAFT_CLIENT_PORT=6379 \
RAFT_PEERS=2,127.0.0.1:7002;3,127.0.0.1:7003 \
cargo run

Node 2

RAFT_NODE_ID=2 \
RAFT_ADDR=127.0.0.1:7002 \
RAFT_CLIENT_PORT=6380 \
RAFT_PEERS=1,127.0.0.1:7001;3,127.0.0.1:7003 \
cargo run

Node 3

RAFT_NODE_ID=3 \
RAFT_ADDR=127.0.0.1:7003 \
RAFT_CLIENT_PORT=6381 \
RAFT_PEERS=1,127.0.0.1:7001;2,127.0.0.1:7002 \
cargo run

The cluster elects a leader automatically within ~300ms. Connect to whichever node is the leader to issue writes:

redis-cli -p 6379   # connect to node 1
SET foo bar         # replicated to quorum before OK is returned
GET foo             # served locally (may be slightly stale on followers)

Followers reject writes with ERR MOVED — connect to the leader. Watch the server stdout to see which node won the election.

---

Supported commands

Command	Example	Description
SET key value	SET name Alice	Store a string value
GET key	GET name	Retrieve a value, or nil if missing/expired
DEL key [key ...]	DEL name age	Delete one or more keys, returns count removed
EXPIRE key seconds	EXPIRE name 60	Set a TTL on an existing key
TTL key	TTL name	Seconds remaining; -1 = no expiry, -2 = missing
INCR key	INCR counter	Increment an integer value by 1 (creates at 0 if missing)
APPEND key value	APPEND log " world"	Append to a string, returns new length
KEYS pattern	KEYS user:*	List all keys matching a glob pattern
SCAN cursor [MATCH pattern] [COUNT n]	SCAN 0 MATCH user:* COUNT 5	Cursor-based key iteration
PUBLISH channel message	PUBLISH events "hello"	Send a message to all subscribers on a channel
SUBSCRIBE channel [...]	SUBSCRIBE events	Subscribe to one or more channels
UNSUBSCRIBE [channel ...]	UNSUBSCRIBE events	Unsubscribe from channels (or all if none given)
AUTH password	AUTH secret	Authenticate when server has KV_PASSWORD set
PING	PING	Returns PONG

---

Architecture

TCP server

The entry point binds a TCP listener. For each incoming connection it spawns a tokio task — a lightweight async unit of work, not a thread. Thousands of concurrent clients are handled with a small fixed thread pool rather than one thread per connection.

RESP protocol (src/resp.rs)

Redis clients don't send plain text. They use RESP (REdis Serialization Protocol), a simple binary-safe wire format. A command like SET name Alice arrives as:

*3\r\n$3\r\nSET\r\n$4\r\nname\r\n$5\r\nAlice\r\n

• *3 — array of 3 elements
• $3\r\nSET — bulk string, 3 bytes
• $4\r\nname — bulk string, 4 bytes
• $5\r\nAlice — bulk string, 5 bytes

src/resp.rs parses this from the TCP stream and serialises responses back into it. Implementing this from scratch is what makes any Redis client in any language able to connect.

The parser is recursive (arrays contain elements which may themselves be arrays). Because Rust cannot size a recursive async fn future at compile time, recursive calls are heap-allocated with Box::pin.

Shared state

All connections share a single Store struct wrapped in Arc<Mutex<...>>:

• Arc — atomic reference counting; multiple async tasks hold a reference to the same data without copying
• Mutex — ensures only one task reads or writes at a time

The tradeoff is a single global lock per command. A production system would shard the map across multiple locks to reduce contention.

TTL expiration

Two-layered, same strategy Redis uses:

1. Lazy expiry — on every GET, check if the key has passed its deadline. Zero overhead for keys never accessed after expiry.
2. Active sweep — a background tokio task wakes every second and calls retain() to drop all expired entries. Reclaims memory even for keys nobody requests.

Expiry deadlines are stored as std::time::Instant (monotonic clock), so host clock adjustments don't affect behaviour.

Persistence — append-only file (kv.aof)

Every write command is appended to kv.aof as a tab-separated line before the response is sent. On startup, the file is replayed line by line to restore state.

EXPIRE records the absolute UNIX deadline rather than relative seconds, so a key set to expire in 60 seconds and replayed 30 seconds later expires at the right time — not 60 seconds from the replay.

In cluster mode the Raft log is the source of truth for writes. The AOF file is still written by INCR, APPEND and EXPIRE (which bypass Raft), but SET and DEL are not written there — those go through the log.

Pub-sub

PUBLISH and SUBSCRIBE are implemented with tokio::sync::broadcast channels:

• Each named channel has one broadcast::Sender<String> in the Store.
• When a client sends SUBSCRIBE, it calls sender.subscribe() to get a Receiver, then a dedicated task is spawned to forward messages into a per-connection mpsc channel.
• The connection handler uses tokio::select! to race between incoming pub-sub messages and new commands — this is how a subscriber can still send UNSUBSCRIBE while waiting for messages.

AUTH

Optional password via the KV_PASSWORD environment variable. Each connection tracks an authenticated flag. If a password is set, all commands except AUTH are rejected until the client authenticates. If no password is configured, AUTH is accepted as a no-op — matches Redis behaviour.

---

Raft consensus layer (src/cluster/)

The cluster module implements a simplified but architecturally honest Raft consensus protocol. No external Raft crate is used — the point is to understand and demonstrate the implementation.

src/cluster/
  mod.rs    — StateMachine trait, ClusterConfig, ClusterHandle, start_cluster
  log.rs    — LogEntry, Command, RaftLog
  node.rs   — wire protocol, RaftMsg, rpc_listener, outbound RPC helpers
  raft.rs   — RaftNode state machine (Follower / Candidate / Leader)

Why Raft

Raft is a consensus algorithm that keeps a cluster of nodes in agreement on a replicated log of commands. Once a command is written to that log and acknowledged by a majority (a quorum), it is committed — it will survive any single node failure and remain in the log forever.

The three things Raft guarantees:

• Leader election — exactly one leader at a time, always with the most up-to-date log
• Log replication — the leader's log is the truth; followers copy it exactly
• Safety — a committed entry is never lost, even if the leader crashes immediately after committing

cluster/log.rs — the replicated log

entries[0]   sentinel  (term=0, index=0)   always present, never committed
entries[1]   index 1
entries[2]   index 2
...
entries[N]   last_index
             ↑ commit_index  (entries up to here are durable on a quorum)
             ↑ last_applied  (entries up to here have been sent to the state machine)

Key invariant: last_applied ≤ commit_index ≤ last_index

The sentinel at index 0 means prevLogIndex = 0 is always a valid reference, which simplifies the AppendEntries consistency check significantly.

Key methods:

Method	Purpose
append(term, cmd)	Leader appends a new entry, returns its index
append_entry(entry)	Follower copies an entry from AppendEntries
truncate_from(index)	Removes a conflicting suffix before overwriting
slice_from(next_index)	Returns entries to send to a specific follower
advance_commit(leader_commit)	Follower advances its commit index
take_unapplied()	Returns committed-but-not-yet-applied entries and advances last_applied

cluster/node.rs — network transport

Two separate TCP ports per node:

• Client port (e.g. 6379) — RESP protocol, Redis clients connect here
• Raft port (e.g. 7001) — internal RPC protocol, peers connect here

The internal RPC format is length-prefix framing over TCP: 4-byte big-endian length followed by a JSON body. JSON is not the most efficient encoding but it's debuggable with nc or Wireshark.

┌────────────────┬──────────────────────────────────────────────┐
│  4 bytes (len) │  JSON body (RequestVote / AppendEntries / …) │
└────────────────┴──────────────────────────────────────────────┘

Wire types (WireMsg, serialised):

• RequestVote(RequestVoteArgs) — candidate asks for a vote
• VoteReply(RequestVoteReply) — voter responds
• AppendEntries(AppendEntriesArgs) — leader replicates entries / heartbeat
• AppendReply(AppendEntriesReply) — follower confirms or rejects

Internal channel type (RaftMsg, never serialised):

The RaftNode receives all inbound events through a single mpsc::Receiver<RaftMsg>. Inbound requests from peers include a oneshot::Sender so RaftNode can reply without knowing how the reply gets back on the wire. The listener task bridges the gap: it sends the request to RaftNode, awaits the oneshot reply, and writes it back on the TCP connection.

Peer A                      rpc_listener task              RaftNode
  │── RequestVote ──────────►│                               │
  │                          │── RaftMsg::VoteRequest ──────►│
  │                          │   (with oneshot reply_tx)     │
  │                          │                    processes  │
  │                          │◄── oneshot reply_tx.send() ───│
  │◄── VoteReply ────────────│                               │

Outbound RPCs (from RaftNode to peers) are fire-and-forget: a task dials the peer, sends the message, reads one reply, and routes it back into RaftNode's channel as a *Reply variant. If the peer is unreachable the reply is silently dropped — the heartbeat timer or election timeout handles the retry.

cluster/raft.rs — the state machine

RaftNode runs as a single long-lived tokio task. Its run() method loops over three async sub-loops, one per role:

      ┌──────────────────────────────────────────────────────┐
      │                   RaftNode::run()                    │
      │                                                      │
      │   ┌─────────────┐   timeout    ┌─────────────────┐  │
      │   │   Follower  │─────────────►│    Candidate    │  │
      │   │             │◄─────────────│                 │  │
      │   └──────┬──────┘  AppendEntries  └──────┬────────┘  │
      │          │          from valid leader     │ quorum    │
      │          │ higher term                    │ votes     │
      │          │                                ▼           │
      │          │                        ┌──────────────┐   │
      │          └───────────────────────►│    Leader    │   │
      │                higher term        └──────────────┘   │
      └──────────────────────────────────────────────────────┘

Follower

Waits for messages on msg_rx. On each iteration a fresh tokio::time::sleep(timeout) future is created — if a valid AppendEntries arrives before it fires, the loop restarts with a new sleep, which is how the election timer resets on heartbeats. If the sleep fires first the node becomes a Candidate.

Election timeout is randomised between 150 ms and 300 ms. The randomisation is why Raft elections resolve quickly: with high probability only one node times out first.

Candidate

1. Increments current_term, votes for itself
2. Broadcasts RequestVote to all peers (each in its own spawned task)
3. Races a fresh election timeout against incoming replies

Vote is granted to a candidate only if:

• Its term is ≥ the voter's current term
• The voter hasn't already voted in this term
• The candidate's log is at least as up-to-date as the voter's log (higher last term, or same term and longer log — §5.4.1)

The log-safety check is what prevents a stale node from winning an election and overwriting committed data.

Leader

On election:

1. Sets next_index[peer] = last_index + 1 for each peer (optimistic — assume they're caught up)
2. Sets match_index[peer] = 0 for each peer (conservative — we don't know yet)
3. Appends a Noop entry. This is Raft §5.4.2: a new leader cannot commit entries from previous terms directly, but by appending a Noop in its own term it causes those entries to be committed as a side effect. Without this, a leader might serve stale reads indefinitely.
4. Sends an AppendEntries immediately (carries the Noop)

On the 50 ms heartbeat: broadcasts AppendEntries to all peers. An empty entries list is a valid heartbeat — it resets followers' election timers and updates their commit indices.

On AppendEntriesReply:

• Success: advance match_index[peer] and next_index[peer]
• Failure: use the follower's match_index hint to jump next_index to the right place rather than decrementing one at a time

Advancing the commit index (§5.4.2):

// Collect match indices for all nodes (including leader itself).
let mut indices = match_index.values() + [last_index];
indices.sort();
// The quorum-th largest — the highest index replicated on a majority.
let quorum_index = indices[n - quorum];
// Only commit if the entry is from the current term.
if log[quorum_index].term == current_term {
    commit_index = quorum_index;
}

The current-term restriction is critical. Consider: a leader from term 2 replicates an entry at index 3 to a majority, then crashes before committing it. A new leader in term 3 must not directly commit that term-2 entry — instead it appends a term-3 Noop, which drives the commit of both the old entry and the Noop together. This prevents a class of safety violations described in the Raft paper (Figure 8).

Handling stale terms

Any RPC with term > current_term causes an immediate step-down regardless of role. step_down():

1. Sets current_term = new_term
2. Clears voted_for
3. Fails any pending client proposes with "lost leadership"
4. Sets role to Follower

This is the mechanism that prevents split-brain: the moment a partitioned leader sees a newer term from a node that was part of a new quorum, it surrenders immediately.

cluster/mod.rs — the public interface

StateMachine trait: The existing Store struct implements this. The applier task calls sm.apply(cmd) for each committed log entry, which is the only path that mutates store.data in cluster mode.

ClusterHandle::propose(cmd): Called by the command handler for every write. Sends a RaftMsg::Propose to RaftNode with a oneshot receiver, then awaits the reply. The oneshot is resolved inside resolve_pending() once commit_index advances past the entry's log index. The client only gets OK after the entry is durable on a quorum.

start_cluster wires three tasks:

start_cluster()
    │
    ├── tokio::spawn  rpc_listener(raft_addr, msg_tx)
    │       Accepts peer TCP connections, routes to RaftNode
    │
    ├── tokio::spawn  RaftNode::run()
    │       The Raft state machine — reads msg_rx, writes apply_tx
    │
    └── tokio::spawn  applier loop
            Reads apply_rx, calls StateMachine::apply()
            ← only path that mutates the KV store in cluster mode

Write path in cluster mode

redis-cli SET foo bar
      │
      ▼
handle_connection()
      │  ch.propose(Command::Set { key: "foo", value: "bar" })
      │  (awaits oneshot)
      ▼
RaftNode (Leader)
  log.append(current_term, Set { key, value })
  broadcast AppendEntries to all peers
      │
      ▼  (majority of peers reply success)
maybe_advance_commit()
resolve_pending()          ← oneshot resolved, propose() returns Ok(())
      │
      ▼
apply_tx.send(entry)
      │
      ▼
applier task
  store.data.insert(key, Entry { value, .. })
      │
      ▼
redis-cli ← "+OK\r\n"

What's simplified vs production Raft

This is an honest learning implementation, not production software. Known gaps:

Gap	What a production system does
Log is in-memory	Persist log entries to disk (WAL) before responding to RPCs — crash recovery requires this
No log compaction	Without snapshots the log grows forever; real systems snapshot state and truncate the log
No cluster membership changes	Adding/removing nodes requires a joint-consensus or single-server change protocol (§6)
Followers reject writes	A production system proxies the write to the leader rather than returning ERR MOVED
Ephemeral TCP connections	One connection per RPC is simple but wasteful; production uses persistent multiplexed connections
No linearisable reads	Local reads may be stale; linearisable reads require either going through the log or a lease mechanism
Single global Mutex	A production store shards the key space across multiple locks

---

Project structure

src/
  main.rs           — server, command dispatch, AOF persistence, pub-sub
  resp.rs           — RESP protocol parser and serialiser
  cluster/
    mod.rs          — StateMachine, ClusterConfig, ClusterHandle, start_cluster
    log.rs          — LogEntry, Command, RaftLog
    node.rs         — wire protocol, RaftMsg, rpc_listener, outbound RPC helpers
    raft.rs         — RaftNode: Follower / Candidate / Leader state machine