Design a Distributed Key-Value Store: Consensus, Partition Tolerance, and Production Survival

// io.thecodeforge — System Design tutorial

import java.security.MessageDigest;
import java.util.*;

public class ConsistentHashRing {
    private final TreeMap<Integer, String> ring = new TreeMap<>();
    private final int virtualNodes;
    private final MessageDigest md;

    public ConsistentHashRing(int virtualNodes) throws Exception {
        this.virtualNodes = virtualNodes;
        this.md = MessageDigest.getInstance("MD5");
    }

    public void addNode(String nodeId) {
        for (int i = 0; i < virtualNodes; i++) {
            int hash = hash(nodeId + ":" + i);
            ring.put(hash, nodeId);
        }
    }

    public void removeNode(String nodeId) {
        for (int i = 0; i < virtualNodes; i++) {
            int hash = hash(nodeId + ":" + i);
            ring.remove(hash);
        }
    }

    public String getNode(String key) {
        if (ring.isEmpty()) return null;
        int hash = hash(key);
        Map.Entry<Integer, String> entry = ring.ceilingEntry(hash);
        if (entry == null) entry = ring.firstEntry();
        return entry.getValue();
    }

    private int hash(String key) {
        byte[] digest = md.digest(key.getBytes());
        return ((digest[0] & 0xFF) << 24) | ((digest[1] & 0xFF) << 16) |
               ((digest[2] & 0xFF) << 8) | (digest[3] & 0xFF);
    }

    public static void main(String[] args) throws Exception {
        ConsistentHashRing ring = new ConsistentHashRing(100);
        ring.addNode("node1");
        ring.addNode("node2");
        ring.addNode("node3");
        System.out.println("Key 'user123' -> " + ring.getNode("user123"));
        System.out.println("Key 'session456' -> " + ring.getNode("session456"));
        ring.removeNode("node2");
        System.out.println("After removing node2:");
        System.out.println("Key 'user123' -> " + ring.getNode("user123"));
        System.out.println("Key 'session456' -> " + ring.getNode("session456"));
    }
}

// io.thecodeforge — System Design tutorial

// Configuration for a distributed KV store with N=3, W=2, R=2
public class ReplicationConfig {
    public static final int N = 3;          // replication factor
    public static final int W = 2;          // write quorum
    public static final int R = 2;          // read quorum
    public static final int TIMEOUT_MS = 5000;

    // Ensure strong consistency: W + R > N
    static {
        assert W + R > N : "W + R must be > N for strong consistency";
    }
}

// io.thecodeforge — System Design tutorial

// Simplified Raft node state machine
public enum RaftState {
    FOLLOWER,
    CANDIDATE,
    LEADER
}

public class RaftNode {
    private RaftState state = RaftState.FOLLOWER;
    private int currentTerm = 0;
    private String votedFor = null;
    private int electionTimeoutMs = 1500; // 1.5 seconds

    public void startElection() {
        state = RaftState.CANDIDATE;
        currentTerm++;
        votedFor = selfId;
        // Send RequestVote RPCs to all peers
        // If majority votes, become leader
    }

    public void onHeartbeatTimeout() {
        if (state == RaftState.FOLLOWER) {
            startElection();
        }
    }
}

// io.thecodeforge — System Design tutorial

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class HintedHandoff {
    private final ConcurrentHashMap<String, String> hints = new ConcurrentHashMap<>();
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

    public void storeHint(String targetNode, String key, String value) {
        hints.put(targetNode + ":" + key, value);
    }

    public void replayHints(String targetNode) {
        hints.entrySet().removeIf(entry -> {
            if (entry.getKey().startsWith(targetNode + ":")) {
                // Send write to targetNode
                System.out.println("Replaying hint: " + entry.getKey() + " -> " + entry.getValue());
                return true;
            }
            return false;
        });
    }

    public void startPeriodicCleanup(long intervalMs) {
        scheduler.scheduleAtFixedRate(() -> {
            // Remove hints older than 3 hours
            // (implementation omitted for brevity)
        }, intervalMs, intervalMs, TimeUnit.MILLISECONDS);
    }
}

// io.thecodeforge — System Design tutorial

import java.time.Instant;
import java.util.UUID;

public class VersionedValue {
    private final String value;
    private final long timestamp; // hybrid logical clock in practice
    private final UUID nodeId;

    public VersionedValue(String value, long timestamp, UUID nodeId) {
        this.value = value;
        this.timestamp = timestamp;
        this.nodeId = nodeId;
    }

    // Last-write-wins: higher timestamp wins, tie-break by nodeId
    public boolean isNewerThan(VersionedValue other) {
        if (this.timestamp != other.timestamp) {
            return this.timestamp > other.timestamp;
        }
        return this.nodeId.compareTo(other.nodeId) > 0;
    }

    public String getValue() { return value; }
}

// io.thecodeforge — System Design tutorial

import java.util.LinkedList;
import java.util.Queue;

public class PhiAccrualFailureDetector {
    private final Queue<Long> heartbeats = new LinkedList<>();
    private final int windowSize = 1000; // keep last 1000 intervals
    private double mean = 0;
    private double variance = 0;

    public void recordHeartbeat(long timestamp) {
        if (!heartbeats.isEmpty()) {
            long interval = timestamp - heartbeats.peek();
            heartbeats.add(interval);
            if (heartbeats.size() > windowSize) {
                heartbeats.poll();
            }
            recalcStats();
        }
        heartbeats.add(timestamp);
    }

    private void recalcStats() {
        // Calculate mean and variance of intervals
        // (omitted for brevity)
    }

    public double computePhi(long now) {
        long lastHeartbeat = heartbeats.peek();
        long elapsed = now - lastHeartbeat;
        double probability = 1 - normalCdf(elapsed, mean, Math.sqrt(variance));
        return -Math.log10(probability);
    }

    private double normalCdf(double x, double mean, double stddev) {
        // Approximation of cumulative distribution function
        return 0.5 * (1 + erf((x - mean) / (stddev * Math.sqrt(2))));
    }

    private double erf(double z) {
        // Error function approximation
        double t = 1.0 / (1.0 + 0.5 * Math.abs(z));
        double ans = 1 - t * Math.exp(-z * z - 1.26551223 +
                            t * (1.00002368 +
                            t * (0.37409196 +
                            t * (0.09678418 +
                            t * (-0.18628806 +
                            t * (0.27886807 +
                            t * (-1.13520398 +
                            t * (1.48851587 +
                            t * (-0.82215223 +
                            t * (0.17087277)))))))));
        return z >= 0 ? ans : -ans;
    }

    public boolean isSuspect(double phiThreshold) {
        return computePhi(System.currentTimeMillis()) > phiThreshold;
    }
}