Replicated Key-Value Store¶

Putting the pieces together¶

Raft gives an ordered, replicated log. A useful service needs more: a state machine that applies the log, clients that find the leader and retry, exactly-once semantics despite retries, and reads that are both correct and cheap. labs/raftkv is that service, built from labs/raft and the LSM engine labs/kv (Build a Storage Engine in Go): etcd and TiKV, in miniature.

Requests and the write path¶

type Request struct {
    Client uint64
    Seq    uint64 // per-client, increasing: makes retries idempotent
    Kind   OpKind // Get, Put, Append
    Key    string
    Value  string
}

func (s *Server) Do(r Request) (string, error) {
    if r.Kind == OpGet && s.mode != ReadThroughLog {
        return s.fastRead(r.Key)
    }
    idx, _, isLeader := s.rf.Start(r.encode())
    if !isLeader {
        return "", ErrWrongLeader
    }
    ch := make(chan result, 1)
    s.mu.Lock()
    s.waiters[idx] = ch
    s.mu.Unlock()
    select {
    case res := <-ch:
        // Another leader's entry may have been committed at our index.
        if res.client != r.Client || res.seq != r.Seq {
            return "", ErrWrongLeader
        }
        return res.value, nil
    case <-time.After(500 * time.Millisecond):
        s.mu.Lock()
        delete(s.waiters, idx)
        s.mu.Unlock()
        return "", ErrTimeout
    case <-s.done:
        return "", ErrWrongLeader
    }
}

The check res.client != r.Client || res.seq != r.Seq is the consequence of Start being only a hope (Raft: Log Replication): if this server lost leadership, the entry committed at idx may be someone else's.

Exactly-once: the dedup table¶

A client that times out cannot know whether its request was applied — the request may have committed and only the reply was lost (Failure and Time). So it retries, and without care an Append is applied twice. Each client numbers its requests; the state machine remembers, per client, the highest sequence number it has applied, and skips anything at or below it:

func (s *Server) applyLocked(r Request, b *kv.Batch) result {
    // ... Get returns the current value ...
    last := s.lastSeqFor(r.Client) // from memory, or from the db after a restart
    if r.Seq <= last {
        return res // a retry of a write already applied: do not apply twice
    }
    switch r.Kind {
    case OpPut:
        b.Put([]byte(r.Key), []byte(r.Value))
    case OpAppend:
        b.Put([]byte(r.Key), append(cur, r.Value...))
    }
    s.lastSeq[r.Client] = r.Seq
    b.Put([]byte(clientKey+strconv.FormatUint(r.Client, 10)), strconv.AppendUint(nil, r.Seq, 10))
    return res
}

The dedup decision is made in the state machine, at apply time — not when the request arrives at a server. Two copies of the same request can both be in the log (the client retried through a new leader); only applying them in log order on every replica gives every replica the same answer.

Durable state machine: one atomic batch per entry¶

After a restart, Raft re-delivers committed entries from index 1 (commitIndex is volatile). A state machine with persistent storage must know which entries it already contains. The apply loop writes the data change, the dedup record, and the applied index in one atomic batch of the storage engine:

for m := range ch {
    s.mu.Lock()
    if m.Index <= s.applied {
        s.mu.Unlock()
        continue // already in the db from before a restart
    }
    var res result
    var b kv.Batch
    if req, ok := decodeRequest(m.Command); ok && m.Command != nil {
        res = s.applyLocked(req, &b)
    }
    b.Put([]byte(appliedKey), strconv.AppendUint(nil, m.Index, 10))
    if err := s.db.Write(&b, kv.WriteOptions{}); err != nil {
        panic(err)
    }
    s.applied = m.Index
    // ... wake the waiting request, if any ...
}

Because the batch is atomic (Write-Ahead Log), a crash leaves the store either before the entry (it will be replayed) or after it (it will be skipped) — never with the effect applied but the index not recorded, which would apply it twice.

TestExactlyOnceAcrossCrashes: 5 servers, 3 clients appending unique tokens to one key as fast as they can, while every second a server is crashed (often the leader, so in-flight requests are retried through a new leader) and restarted from its persisted Raft state and kv directory, 10 times. Result: 1,991 appends, each present exactly once in the final value.

Reads: three ways, measured¶

A read is where correctness is most often traded away for speed.

ReadIndex, in labs/raft:

func (n *Node) ReadIndex() (uint64, bool) {
    n.mu.Lock()
    // The commit index is only known to be current once an entry of the
    // leader's own term (the no-op) has committed.
    if n.state != Leader || n.log[n.commitIndex].Term != n.term {
        n.mu.Unlock()
        return 0, false
    }
    readIdx, term := n.commitIndex, n.term
    // ... send one heartbeat to every follower, without holding mu ...
    n.mu.Unlock()
    // Wait for a majority (counting ourselves) to answer with our term.
    // ... if any reply carries a higher term, we step down and fail ...
    return readIdx, true
}

The heartbeat round proves that, at some moment after the read arrived, a majority still accepted this node as leader — so no newer leader could have committed writes this node doesn't know about. Waiting until the state machine has applied readIdx ensures the local read reflects everything committed before the read started.

Correctness under partitions¶

Each run: 5 servers, 5 clients for about 25 seconds of simulated time (clients 0–1 only read, 2–4 write), with the current leader and one follower cut off into a minority partition every other round. Every operation's call and return time is recorded and the history is checked by labs/lincheck (Testing Distributed Systems).

The unsafe mode fails for a precise reason: the old leader isolated in the minority still believes it is leader, and the read-only clients keep asking it, because it keeps answering successfully. Meanwhile the majority elects a new leader and commits new writes, which the old leader never sees. The readers see the past.

Check-quorum (Raft: Leader Election) makes the isolated leader step down after about one election timeout (310 ms measured), which shrinks the window: 2 of 5 runs still produced a stale read, against 5 of 5 without it. A shorter window is not safety. Between the partition and the step-down, the old leader still serves stale values; only ReadIndex (or a log read) closes the gap, because it asks a majority at the time of each read.

Note how the first version of this test did not catch the bug: every client did both reads and writes, so all of them got stuck on the old leader's writes (which can't commit), timed out, and moved to the new leader together — nobody was left reading from the stale node. Real read-only clients have no reason to move. A test that cannot fail proves nothing; this one was checked by making sure it fails on the known-broken mode.

Cost¶

One client, no concurrent writes, 200 reads; counting the Raft RPCs and persists the reads themselves cause (heartbeats subtracted):

A log read appends an entry on the leader and all four followers — five fsyncs on a real disk — and grows the log. ReadIndex costs one heartbeat to each follower and touches no disk. In this simulation persistence is free, so both modes show the same ~21 ms read latency (two client hops plus one round trip to a majority); on real hardware the log read adds the fsync latency of the slowest node in the fastest majority. This is why etcd serves linearizable reads with ReadIndex, and offers lease-based reads (Leases and Fencing) and serializable (possibly stale) reads as explicit opt-ins.

The client¶

func (k *clerk) do(kind OpKind, key, value string) string {
    k.seq++
    r := Request{Client: k.id, Seq: k.seq, Kind: kind, Key: key, Value: value}
    for {
        v, err := k.c.call(k.leader, r) // same Seq on every retry
        if err == nil {
            return v
        }
        k.leader = (k.leader + 1) % k.c.n
        time.Sleep(10 * time.Millisecond)
    }
}

The client remembers the last known leader, retries the same request (same sequence number) on any error, and rotates servers. Real clients use the leaderID hint in rejection replies instead of rotating blindly, and give each request an overall deadline.

Snapshots and batching¶

Two production features are implemented and measured in Raft in Production:

• Snapshots: every SnapshotEvery applied entries the server makes its db durable (one synced write), dumps it, and lets Raft discard the log up to that index; a lagging follower receives the dump through InstallSnapshot and replaces its db in one atomic batch. With snapshots every 50 entries, the exactly-once test keeps at most 33 entries in memory instead of 1,549.
• Batched proposals: a proposer goroutine groups concurrent client writes into one StartBatch call, one Raft log write for many requests. Waiters are registered under the server mutex before the batch is proposed — the first version registered them after Start returned, so a fast commit could be applied before anyone was waiting for it, and the request timed out and was retried (harmless thanks to deduplication, but slow and wasteful).

Still missing compared to etcd or TiKV: multiple Raft groups (one per key range), dedup-table expiry via client sessions, and pipelining proposals while the leader's own fsync is in progress.

Common pitfall¶

Doing duplicate detection at the server's front door. A team adds an idempotency check in the request handler: "if I've already seen (client, seq), return the cached response". It works in tests with a stable leader. During a leader change, the client's first attempt is committed through the old leader but its reply is lost; the retry goes to the new leader, whose front-door cache has never seen the request, so it proposes it again, and the log now contains the append twice. Every replica applies both. Deduplication must happen where every replica sees the same sequence of events — in the state machine, at apply time, persisted together with the state it protects.

Review questions¶

1. Why must the server check that the entry applied at its index is its own request?
2. Why is duplicate detection done at apply time rather than when a request arrives?
3. What would go wrong if the applied index were written in a separate batch from the data?
4. How does ReadIndex prove that no newer leader exists, and why must the leader wait until it has applied the read index?
5. Why did every unsafe run fail, and why did the first version of the test not detect the bug?