Multi-Raft and Sharding¶

One Raft group is not a database¶

A single Raft group is limited by its leader: every write goes through one node's log, one CPU, one disk. Systems that store terabytes — CockroachDB, TiKV/TiDB, YugabyteDB — split the key space into ranges (tens to hundreds of MB each), make each range its own Raft group, and spread the groups' replicas and leaders over many machines. A cluster then runs thousands of Raft groups at once, and ranges split as they grow and move between machines to balance load. Partitioning in general is in Partitioning vs. Sharding; this page builds the Multi-Raft version in Go (labs/multiraft) and records the four bugs that testing found in it, because each one is a known trap in real systems.

Structure¶

Each store's replicas talk over a per-range simulated network, and partitions and machine failures are applied to all ranges at once, because they model machines.

Checking the range at apply time¶

A request is routed using the client's view of the descriptors, which may be stale: a split may have moved the key into a new range. The only safe place to check is the state machine, in log order:

func (r *Replica) applyLocked(c Command) result {
    // The range may have shrunk between routing and applying. Checking
    // here, at the request's position in the log, gives every replica
    // the same answer.
    if !r.st.Desc.Contains(c.Key) {
        return result{err: ErrWrongRange}
    }
    // ... apply Put/Append/Get, with per-client deduplication ...
}

The client treats ErrWrongRange as "refresh routes and retry". A check at the server's front door instead would be racy: the split could be applied between the check and the write.

Splits as Raft commands¶

A split is proposed to the range being split, and every replica applies it at the same log index:

func (r *Replica) applySplitLocked(c Command) {
    d := r.st.Desc
    // Decide ONLY from replicated state, never from anything local to
    // this store: every replica must make the same decision at this index.
    if !d.Contains(c.SplitKey) || c.SplitKey == d.Start {
        return
    }
    right := state{
        Desc: RangeDesc{ID: c.NewRange, Start: c.SplitKey, End: d.End,
            Replicas: slices.Clone(d.Replicas), Generation: d.Generation + 1},
        Data:    map[string]string{},
        LastSeq: map[uint64]uint64{},
    }
    for k, v := range r.st.Data { // move the upper half of the data
        if k >= c.SplitKey {
            right.Data[k] = v
            delete(r.st.Data, k)
        }
    }
    maps.Copy(right.LastSeq, r.st.LastSeq) // dedup state goes to both halves
    r.st.Desc.End = c.SplitKey
    r.st.Desc.Generation++
    r.store.createReplica(right.Desc, &right, right.Desc.Replicas) // no-op if it already exists
}

Because every replica applies the split at the same position in the log, the left range shrinks and the right range is created with identical contents on every store, and no request can be lost or applied to the wrong half: requests before the split in the log were applied to the whole range, requests after it are rejected with ErrWrongRange if their key moved. The deduplication table is copied to both halves, so a client retry that lands in the right-hand range after the split is still recognized.

Generation increases with every split, so a client that learns about the same range ID from two stores keeps the newer descriptor.

Moving replicas¶

Rebalancing moves a range from one set of stores to another, one membership change at a time (Raft in Production): add a replica on the target store, then remove one from the source. The new replica starts empty with no configuration; the leader compacts its log first, so its first message to the new replica is an InstallSnapshot carrying the whole range.

The four bugs, and what each one teaches¶

The tests run concurrent clients doing puts, appends, and gets on 100 keys while the range splits three times (or moves from stores {0,1,2} to {1,2,3}), check the full history with lincheck, and check that every range holds only keys within its bounds. They passed on the first runs. Running them a hundred times exposed the rest.

1. A removed replica never learns it was removed¶

After moving a range off store 0, the replica there lingered forever. The leader starts using a new configuration as soon as the entry is in its log, and the new configuration excludes store 0 — so the leader stops sending to store 0, including the very entry that removes it. Raft has no way to tell a removed member. Real systems run a replica GC queue: each store periodically checks the range's authoritative descriptor and destroys replicas that are no longer listed.

func (s *Store) GCReplicas(stores map[int]*Store) (removed int) {
    for _, d := range s.Descriptors() {
        // find the range's current leader; if its descriptor doesn't
        // list this store, destroy the local replica
    }
}

In the tests the lingering replica appeared in 5 of 6 runs; in the sixth, store 0 happened to be the leader, applied its own removal, and stepped down.

2. A lagging replica skipped the split entirely¶

In about 1 run in 20, one store ended with only 3 of the 4 ranges. Its replica of the left range was behind, and the leader — which had compacted its log — sent it a snapshot taken after the split. The replica jumped straight to the post-split state and never applied the split command, so it never created its replica of the right-hand range.

The fix is how CockroachDB does it, in two parts:

• When Raft traffic for a range arrives at a store that has no replica of it, the store creates an uninitialized replica on demand. It owns no keys, is excluded from routing, and cannot win an election (its log is empty).
• A range created by a split starts its Raft log from a snapshot at index 1 containing its initial state, not from an empty log. A replica that did not take part in the split therefore cannot replay the new range's log on top of nothing: the leader finds its log empty, sees that the needed entries are covered by a snapshot, and sends the snapshot.

if initial != nil {
    data, _ := json.Marshal(initial)
    storage = raft.NewMemStorageFromSnapshot(raft.Snapshot{
        Index: 1, Term: 1, Config: bootstrap, Data: data})
}

3. The split decision depended on local state¶

The first version of applySplitLocked skipped the split if the store already had a replica of the new range. Once replicas could be created on demand (fix 2), whether a store already had one depended on message timing — so two replicas of the same range could apply the same log entry differently: one shrank, the other did not. That is the most fundamental rule of replicated state machines: an entry's effect may depend only on replicated state and the entry itself, never on anything local (files present, maps of local objects, wall-clock time, random numbers). The fix decides from the descriptor alone and lets createReplica skip creation if the replica exists.

4. The simulator dropped the messages that would have fixed it¶

After fixes 2 and 3, the failure still appeared about once in 40 runs under the race detector. The simulated network checked "is the target reachable?" before looking up the target node, and an unregistered node counted as down — so messages for a range that did not yet exist on a store were dropped silently, before the on-demand hook could run. The test infrastructure had a bug that only mattered for this new code path. After the fix: 0 failures in 100 runs without the race detector and 0 in 60 with it.

The general lesson from all four: Multi-Raft's hard parts are not inside Raft, but between groups — splits, snapshots that cross splits, membership and garbage collection — and they surface only under repeated, randomized runs. A test that passes once says little.

Measured: why stores share one WAL¶

The point of many Raft groups is parallelism: different ranges can commit at the same time. With a separate log file per group, each group does its own fsync. Does the disk run them in parallel? Many goroutines, each appending 128 bytes and syncing its own file, for 3 seconds (Apple M1 Pro, internal NVMe):

Durable syncs did not parallelize at all: a device cache flush flushes the whole device, so independent logs queue behind each other, and 16 logs together managed the same ~250 durable appends per second as one. Without the device flush, parallel streams scaled about 2.5x before saturating.

This is why production Multi-Raft systems do not give each range its own WAL. CockroachDB writes the Raft logs of all ranges on a store into one storage engine with one WAL; TiKV's raft-engine is a single log shared by all regions on a store. One fsync then makes a batch of entries from many groups durable at once — group commit across groups, the same idea as the storage engine's WAL and Raft batching. The throughput of a store becomes (fsyncs per second) × (entries from all groups per fsync).

What production adds¶

Common pitfall¶

Making a state machine's behavior depend on something that is not in the log. It looks harmless: "if the file already exists, skip", "if this store already has the object, don't recreate it", "use time.Now() for the TTL". Each replica applies the same entries, but local state differs between replicas — because of timing, snapshots, restarts — so they compute different results and silently diverge. No single replica looks wrong; the error appears only when a different replica becomes leader and serves data the old leader never had. Every decision in apply must be a pure function of the replicated state and the entry; anything else (timestamps, random IDs, local facts) must be decided by the proposer and written into the entry.

Review questions¶

1. Why must the "is this key in my range" check happen at apply time rather than when the request arrives?
2. How does making the split a Raft command guarantee that no write is applied to the wrong half?
3. Why does a removed replica never learn of its removal, and what removes it in practice?
4. Why must a range created by a split start its log from a snapshot at index 1?
5. Why did 16 separate logs not commit faster than one on this machine, and how do CockroachDB and TiKV avoid the problem?