Building a replicated key/value service
With the Raft consensus module complete, we can now build a realistic application on top of it: a replicated key/value database offering strong consistency. The full code for this part lives in part4kv/.
State machine and commands
The database is essentially a map from strings to strings, supporting three operations:
PUT(k,v): assign valuevto keykGET(k): retrieve the value for keykCAS(k, cmp, v): atomic compare-and-swap; reads current valuecurVfor keyk, and ifcurV==cmpsets it tov; either way, returnscurV
Given a Raft log containing these commands and an initially empty DB, the state machine transitions deterministically as commands are applied.
PUT("x","2") PUT("y","3") PUT("x","4") PUT("z","5") CAS("x","4","8") CAS("z","4","9")
x=8 y=3 z=5
System architecture
The setup includes a cluster of three replicas, each running a KV service. Every service embeds the Raft consensus module from Part 3 (using the RPC layer between replicas unchanged) plus a simple data store implemented in kvservice/datastore.go. A REST API fronts each service, and a client library encapsulates the HTTP interactions.
[1] A handful of replicas suffices; in production you'd likely run more.
Handling a PUT request
A successful PUT("k", "v") follows this path:
- The client sends the request via HTTP to a service, assumed to be the current Raft leader.
- The HTTP handler builds a
Commandof kindCommandPutand submits it to the consensus module. - The handler then must wait; it cannot acknowledge the client until the command is committed by the cluster. This occurs when the command shows up on the commit channel.
- Meanwhile, a background updater goroutine watches that commit channel and applies committed commands to the data store — both on the leader and on followers, keeping all replicas consistent.
Steps 2 (waiting) and 3 (applying) run concurrently: client requests are handled in separate goroutines by Go's HTTP server, while a single updater goroutine serializes state-machine application.
Service implementation
The core structure in kvservice/kvservice.go wires a Raft server, the datastore, the HTTP server, and the commit channel together.
type KVService struct {
sync.Mutex
// id is the service ID in a Raft cluster.
id int
// rs is the Raft server that contains a CM
rs *raft.Server
// commitChan is the commit channel passed to the Raft server; when commands
// are committed, they're sent on this channel.
commitChan chan raft.CommitEntry
// commitSubs are the commit subscriptions currently active in this service.
// See the createCommitSubscription method for more details.
commitSubs map[int]chan Command
// ds is the underlying data store implementing the KV DB.
ds *DataStore
// srv is the HTTP server exposed by the service to the external world.
srv *http.Server
}
// New creates a new KVService
//
// - id: this service's ID within its Raft cluster
// - peerIds: the IDs of the other Raft peers in the cluster
// - storage: a raft.Storage implementation the service can use for
// durable storage to persist its state.
// - readyChan: notification channel that has to be closed when the Raft
// cluster is ready (all peers are up and connected to each other).
func New(id int, peerIds []int, storage raft.Storage, readyChan <-chan any) *KVService {
gob.Register(Command{})
commitChan := make(chan Command)
// raft.Server handles the Raft RPCs in the cluster; after Serve is called,
// it's ready to accept RPC connections from peers.
rs := raft.NewServer(id, peerIds, storage, readyChan, commitChan)
rs.Serve()
kvs := &KVService{
id: id,
rs: rs,
commitChan: commitChan,
ds: NewDataStore(),
commitSubs: make(map[int]chan Command),
}
kvs.runUpdater()
return kvs
}
HTTP routing uses the standard Go server with handlers registered per operation:
// ServeHTTP starts serving the KV REST API on the given TCP port. This
// function does not block; it fires up the HTTP server and returns. To properly
// shut down the server, call the Shutdown method.
func (kvs *KVService) ServeHTTP(port int) {
if kvs.srv != nil {
panic("ServeHTTP called with existing server")
}
mux := http.NewServeMux()
mux.HandleFunc("POST /get/", kvs.handleGet)
mux.HandleFunc("POST /put/", kvs.handlePut)
mux.HandleFunc("POST /cas/", kvs.handleCAS)
kvs.srv = &http.Server{
Addr: fmt.Sprintf(":%d", port),
Handler: mux,
}
go func() {
kvs.kvlog("serving HTTP on %s", kvs.srv.Addr)
if err := kvs.srv.ListenAndServe(); err != http.ErrServerClosed {
log.Fatal(err)
}
kvs.srv = nil
}()
}
The updater goroutine
This goroutine reads committed entries from the consensus channel, applies them to the datastore, and notifies any waiters. For a handler waiting on its own submitted command, the flow is:
- The handler calls
Submiton the Raft module, receiving the log index for its entry — or-1if the current node isn't the leader, in which case it returns a special "not the leader" status to the client. - The handler registers a subscription keyed by that log index, using a channel for notification.
- It blocks receiving on that channel until either the subscription fires or the HTTP request is canceled.
The implementation of handlePut shows this exactly:
func (kvs *KVService) handlePut(w http.ResponseWriter, req *http.Request) {
pr := &api.PutRequest{}
if err := readRequestJSON(req, pr); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
kvs.kvlog("HTTP PUT %v", pr)
// Submit a command into the Raft server; this is the state change in the
// replicated state machine built on top of the Raft log.
cmd := Command{
Kind: CommandPut,
Key: pr.Key,
Value: pr.Value,
Id: kvs.id,
}
logIndex := kvs.rs.Submit(cmd)
// If we're not the Raft leader, send an appropriate status
if logIndex < 0 {
renderJSON(w, api.PutResponse{RespStatus: api.StatusNotLeader})
return
}
// Subscribe for a commit update for our log index. Then wait for it to
// be delivered.
sub := kvs.createCommitSubscription(logIndex)
// Wait on the sub channel: the updater will deliver a value when the Raft
// log has a commit at logIndex. To ensure clean shutdown of the service,
// also select on the request context - if the request is canceled, this
// handler aborts without sending data back to the client.
select {
case commitCmd := <-sub:
// If this is our command, all is good! If it's some other server's command,
// this means we lost leadership at some point and should return an error
// to the client.
if commitCmd.Id == kvs.id {
renderJSON(w, api.PutResponse{
RespStatus: api.StatusOK,
KeyFound: commitCmd.ResultFound,
PrevValue: commitCmd.ResultValue,
})
} else {
renderJSON(w, api.PutResponse{RespStatus: api.StatusFailedCommit})
}
case <-req.Context().Done():
return
}
}
A crucial safety check follows the notification: the handler must verify that the command committed at the expected index is actually its own command. This is why each Command includes an id field.
Consider a scenario where the leader A places a command at log index 42 but gets partitioned off before replicating it. When C later becomes leader and commits a different command at index 42, A will eventually see that commit once reconnected — but the ID won't match, so it correctly reports a "failed commit" status instead of falsely acknowledging the original request.
Consistency semantics
The service is linearizable: operations become visible only after being committed, and Raft consensus defines the serialization point. It is also serializable for compound operations like CAS, since the leader executes them atomically. Together these properties make the system strictly serializable — the strongest consistency guarantee available.
As a "CP"-type system, this consistency comes at the cost of availability during network partitions. It's not suited for high-throughput workloads, since every operation must reach consensus first. Instead, such services fit at the foundation of large distributed systems: coordinating locks, electing leaders (via CAS primitives), or storing small but critical configuration data.
Why GET Requests Must Join the Raft Log
Looking at the command implementations for PUT, GET, and CAS, it's clear they all follow the same flow through the Raft log. That raises a natural question: do read-only operations like GET really need to go through the log? They don't modify the state machine, after all.
They do need to. Skipping the log for reads introduces a linearizability violation. Imagine a key-value pair k=v in the database, with node A disconnected from the cluster while still believing it's the leader. Meanwhile, the cluster elects node C as the new leader. A client sends PUT(k,v2) to C, which replicates and commits the new value. Another client then reads GET(k) from C and correctly receives v2.
Now a third client sends GET(k) to A, which still considers itself leader. Since A would serve the read directly from its local datastore, it would respond with the stale value v. That sequence—one client reading v2 after a committed write, and another client later reading v—is impossible in any linearizable single-threaded history.
This is precisely the scenario outlined in Section 8 of the Raft paper. The standard fix, which this implementation follows, is to route every command—read-only or not—through the log. A service only responds to a client when the command is successfully committed. In the example above, node A would never reply while partitioned from the cluster; it would wait, discover it lost leadership, and the client would retry against the real leader. If A did regain leadership later, it would process the pending PUT(k,v2) before the GET(k), because the state machine applies entries in log order.
Client-Side Leader Discovery
The final component is the KV client library in kvclient/kvclient.go. While the service exposes a plain REST API that could be called with curl, the library encapsulates the tricky part: locating and remembering the current cluster leader.
The client type and its constructor look like this:
type KVClient struct {
addrs []string
// assumedLeader is the index (in addrs) of the service we assume is the
// current leader. It is zero-initialized by default, without loss of
// generality.
assumedLeader int
clientID int32
}
// New creates a new KVClient. serviceAddrs is the addresses (each a string
// with the format "host:port") of the services in the KVService cluster the
// client will contact.
func New(serviceAddrs []string) *KVClient {
return &KVClient{
addrs: serviceAddrs,
assumedLeader: 0,
clientID: clientCount.Add(1),
}
}
// clientCount is used internally for debugging
var clientCount atomic.Int32
Creating a client requires a list of addresses for the KV service nodes in the cluster; those services must be listening before the client makes its first request.
Each request follows the same flow, using Put as an example:
// Put the key=value pair into the store. Returns an error, or
// (prevValue, keyFound, false), where keyFound specifies whether the key was
// found in the store prior to this command, and prevValue is its previous
// value if it was found.
func (c *KVClient) Put(ctx context.Context, key string, value string) (string, bool, error) {
putReq := api.PutRequest{
Key: key,
Value: value,
}
var putResp api.PutResponse
err := c.send(ctx, "put", putReq, &putResp)
return putResp.PrevValue, putResp.KeyFound, err
}
Request and response types such as PutRequest and PutResponse live in api/api.go and are straightforward data structures. The core logic sits in the send method:
func (c *KVClient) send(ctx context.Context, route string, req any, resp api.Response) error {
// This loop rotates through the list of service addresses until we get
// a response that indicates we've found the leader of the cluster. It
// starts at c.assumedLeader
FindLeader:
for {
// There's a two-level context tree here: we have the user context - ctx,
// and we create our own context to impose a timeout on each request to
// the service. If our timeout expires, we move on to try the next service.
// In the meantime, we have to keep an eye on the user context - if that's
// canceled at any time (due to timeout, explicit cancellation, etc), we
// bail out.
retryCtx, retryCtxCancel := context.WithTimeout(ctx, 50*time.Millisecond)
path := fmt.Sprintf("http://%s/%s/", c.addrs[c.assumedLeader], route)
c.clientlog("sending %#v to %v", req, path)
if err := sendJSONRequest(retryCtx, path, req, resp); err != nil {
// Since the contexts are nested, the order of testing here matters.
// We have to check the parent context first - if it's done, it means
// we have to return.
if contextDone(ctx) {
c.clientlog("parent context done; bailing out")
retryCtxCancel()
return err
} else if contextDeadlineExceeded(retryCtx) {
// If the parent context is not done, but our retry context is done,
// it's time to retry a different service.
c.clientlog("timed out: will try next address")
c.assumedLeader = (c.assumedLeader + 1) % len(c.addrs)
retryCtxCancel()
continue FindLeader
}
retryCtxCancel()
return err
}
c.clientlog("received response %#v", resp)
// No context/timeout on this request - we've actually received a response.
switch resp.Status() {
case api.StatusNotLeader:
c.clientlog("not leader: will try next address")
c.assumedLeader = (c.assumedLeader + 1) % len(c.addrs)
retryCtxCancel()
continue FindLeader
case api.StatusOK:
retryCtxCancel()
return nil
case api.StatusFailedCommit:
retryCtxCancel()
return fmt.Errorf("commit failed; please retry")
default:
panic("unreachable")
}
}
}
The context handling carries some subtlety, which the code comments clarify.
The client remembers the last node that accepted a command as leader. When sending a new request, it starts with that node. If the request times out, or the node responds that it's no longer the leader, the client retries the next service in the provided list. During normal operations, leadership is stable: a client discovers the leader once and then addresses it directly. Cluster disruptions cause brief leader-search delays, which could be optimized further if needed.
If no leader can be found, the client keeps retrying indefinitely, but since Go's context idiom is used throughout, callers can enforce their own timeouts or cancelations.
Retry Semantics and Linearizability Gaps
As discussed, the service itself achieves strong consistency, but maintaining linearizability end-to-end across a client is notoriously difficult. The simple retry strategy is not immune to a subtle problem.
When a PUT request times out, the client simply retries against what it hopes is a different leader. That approach can be wrong. Suppose the original leader committed the command but crashed before replying. A retry could append a duplicate command to the log. Even though PUT is logically idempotent, the duplication breaks linearizability if another client wrote a different value for the same key between the two attempts.
This is also flagged in Section 8 of the Raft paper. It is a hard problem, and the next part of this series will explore it in detail—reviewing one potential solution and how production distributed key-value stores cope with it.
