How would you design a distributed key-value store?

// A Dynamo-style key-value store in miniature: a hash ring with virtual nodes, N=3 replicas,
// W=2 / R=2 quorums, sloppy quorum with hinted handoff, read repair and vector clocks.
const NODES = ["A", "B", "C", "D", "E"], VNODES = 4, N = 3, W = 2, R = 2;
const fnv = (s) => { let h = 0x811c9dc5; for (let i = 0; i < s.length; i++) { h ^= s.charCodeAt(i); h = Math.imul(h, 0x01000193) >>> 0; } return h; };
const ring = [];
for (const n of NODES) for (let v = 0; v < VNODES; v++) ring.push([fnv(n + "#" + v), n]);
ring.sort((a, b) => a[0] - b[0]);
function prefList(key) { // every physical node in clockwise order from the key; first N are home replicas
  const h = fnv(key);
  let i = ring.findIndex((t) => t[0] >= h);
  if (i < 0) i = 0;
  const out = [];
  for (let k = 0; out.length < NODES.length; k++) { const n = ring[(i + k) % ring.length][1]; if (!out.includes(n)) out.push(n); }
  return out;
}
const store = {}, up = {}, hints = [];
for (const n of NODES) { store[n] = new Map(); up[n] = true; }
const clockStr = (c) => "{" + c.map((x, i) => (x ? NODES[i] + ":" + x : "")).filter(Boolean).join(",") + "}";
function cmp(a, b) { // "before", "after", "equal" or "concurrent"
  let lt = false, gt = false;
  for (let i = 0; i < a.length; i++) { if (a[i] < b[i]) lt = true; if (a[i] > b[i]) gt = true; }
  return lt && gt ? "concurrent" : lt ? "before" : gt ? "after" : "equal";
}
function keep(versions, v) { // add v unless an existing version dominates it; drop versions v dominates
  if (versions.some((x) => ["after", "equal"].includes(cmp(x.clock, v.clock)))) return versions;
  return versions.filter((x) => cmp(x.clock, v.clock) !== "before").concat([v]).sort((x, y) => (x.value < y.value ? -1 : 1));
}
function write(node, key, v) { store[node].set(key, keep(store[node].get(key) || [], v)); }
function put(key, value, ctx, coord) {
  const pref = prefList(key);
  coord = coord || pref.find((n) => up[n]);
  const clock = ctx.slice();
  clock[NODES.indexOf(coord)]++;
  const v = { value, clock };
  const acks = [], spare = pref.slice(N).filter((n) => up[n]);
  for (const home of pref.slice(0, N)) {
    if (up[home]) { write(home, key, v); acks.push(home); }
    else if (spare.length) { const sub = spare.shift(); hints.push({ sub, home, key, v }); acks.push(sub + "(hint for " + home + ")"); }
  }
  return "put " + key + "=" + value + " " + clockStr(clock) + " via " + coord + " -> " + acks.join(", ") + (acks.length >= W ? " OK" : " FAILED");
}
function get(key, r, from) {
  const replicas = from || prefList(key).slice(0, N).filter((n) => up[n]).slice(0, r);
  let merged = [];
  for (const n of replicas) for (const v of store[n].get(key) || []) merged = keep(merged, v);
  const repaired = [];
  for (const n of replicas) {
    const mine = (store[n].get(key) || []).map((v) => clockStr(v.clock)).join();
    if (mine !== merged.map((v) => clockStr(v.clock)).join()) { for (const v of merged) write(n, key, v); repaired.push(n); }
  }
  return { merged, text: "get " + key + " R=" + r + " from " + replicas.join(",") + ": " + merged.map((v) => v.value + " " + clockStr(v.clock)).join(" | ") + (repaired.length ? "  (read repair -> " + repaired.join(",") + ")" : "") };
}

console.log("== 1. Ring placement (5 nodes x 4 virtual nodes, N=3) ==");
for (const k of ["cart:7", "user:42", "session:9"]) console.log("  " + k + " -> home " + prefList(k).slice(0, N).join(",") + "  spares " + prefList(k).slice(N).join(","));
const KEY = "cart:7", [p0, p1, p2] = prefList(KEY);
const zero = NODES.map(() => 0);

console.log("\n== 2. Normal write, then one home replica fails ==");
console.log("  " + put(KEY, "book", zero));
up[p1] = false;
console.log("  node " + p1 + " is down");
let r = get(KEY, R);
console.log("  " + r.text);
console.log("  " + put(KEY, "book+pen", r.merged[0].clock));
up[p2] = false;
console.log("  node " + p2 + " is down too: " + put(KEY, "book+pen+cup", get(KEY, 1).merged[0].clock));
console.log("  strict quorum would need " + W + " of the home replicas, only 1 is up; sloppy quorum used hints");

console.log("\n== 3. Both nodes return: a stale read, read repair, hinted handoff ==");
up[p1] = true; up[p2] = true;
console.log("  " + get(KEY, 1, [p1]).text + "   <- stale");
console.log("  " + get(KEY, R, [p1, p0]).text);
for (const h of hints.splice(0)) {
  const before = (store[h.home].get(h.key) || []).length && cmp(store[h.home].get(h.key)[0].clock, h.v.clock) !== "before";
  write(h.home, h.key, h.v);
  console.log("  hint " + h.sub + " -> " + h.home + " " + clockStr(h.v.clock) + (before ? " already current" : " delivered"));
}
console.log("  " + get(KEY, N).text);

console.log("\n== 4. Two clients write concurrently from the same version ==");
const ctx = get(KEY, R).merged[0].clock;
console.log("  " + put(KEY, "book+pen+cup+hat", ctx, p0));
const later = "book+pen+cup+scarf";
console.log("  " + put(KEY, later, ctx, p2));
r = get(KEY, R);
console.log("  " + r.text);
const union = [...new Set(r.merged.flatMap((v) => v.value.split("+")))].sort().join("+");
const mergedCtx = r.merged.reduce((c, v) => c.map((x, i) => Math.max(x, v.clock[i])), zero);
console.log("  client merges siblings: " + put(KEY, union, mergedCtx));
console.log("  " + get(KEY, R).text);
console.log("  last-write-wins would keep only the later write (" + later + ") and silently lose the hat");

// One storage node as an LSM tree: a write-ahead log and memtable, flushes to immutable sorted
// SSTables with Bloom filters, tombstones for deletes, and a compaction that merges the files.
const fnv = (s) => { let h = 0x811c9dc5; for (let i = 0; i < s.length; i++) { h ^= s.charCodeAt(i); h = Math.imul(h, 0x01000193) >>> 0; } return h; };
const BITS = 64, K = 3, MEMTABLE_LIMIT = 4, TOMB = "<tombstone>";
const mix = (h) => { h = Math.imul(h ^ (h >>> 16), 0x85ebca6b) >>> 0; h = Math.imul(h ^ (h >>> 13), 0xc2b2ae35) >>> 0; return (h ^ (h >>> 16)) >>> 0; }; // murmur3 finalizer
const bitsFor = (key) => { const h1 = mix(fnv(key)), h2 = (mix(fnv(key + "#")) | 1) >>> 0; const out = []; for (let i = 0; i < K; i++) out.push((h1 + i * h2) % BITS); return out; };
let memtable = new Map(), sstables = [], walWrites = 0, flushWrites = 0, compactWrites = 0, userWrites = 0, deletes = 0;

function flush() {
  const rows = [...memtable.entries()].sort((a, b) => (a[0] < b[0] ? -1 : 1));
  const bloom = new Array(BITS).fill(false);
  for (const [k] of rows) for (const b of bitsFor(k)) bloom[b] = true;
  sstables.unshift({ id: "sst-" + (sstables.length + 1), rows, bloom }); // newest first
  flushWrites += rows.length;
  memtable = new Map();
}
function put(k, v) { walWrites++; userWrites++; if (v === TOMB) deletes++; memtable.set(k, v); if (memtable.size >= MEMTABLE_LIMIT) flush(); }
function get(k, useBloom) {
  const trace = [];
  if (memtable.has(k)) return [memtable.get(k), "memtable", 0];
  let reads = 0;
  for (const t of sstables) {
    if (useBloom && !bitsFor(k).every((b) => t.bloom[b])) { trace.push(t.id + ":skip"); continue; }
    reads++;
    let lo = 0, hi = t.rows.length - 1; // binary search inside the sorted file
    while (lo <= hi) {
      const mid = (lo + hi) >> 1;
      if (t.rows[mid][0] === k) { trace.push(t.id + ":hit"); return [t.rows[mid][1], trace.join(" "), reads]; }
      if (t.rows[mid][0] < k) lo = mid + 1; else hi = mid - 1;
    }
    trace.push(t.id + ":miss");
  }
  return [null, trace.join(" "), reads];
}
const show = (k) => { const [v, trace, reads] = get(k, true); return k + " = " + (v === null ? "(not found)" : v === TOMB ? "(deleted)" : v) + "   [" + trace + "] file reads " + reads; };

for (let i = 1; i <= 10; i++) put("user:" + String(i).padStart(2, "0"), "v1");
put("user:03", "v2"); put("user:07", "v2"); put("user:03", "v3"); put("user:11", "v1");
put("user:05", TOMB); put("user:12", "v1"); put("user:09", TOMB); put("user:13", "v1");
console.log("== 1. Writes: " + (userWrites - deletes) + " puts and " + deletes + " deletes, memtable flushes every " + MEMTABLE_LIMIT + " keys ==");
for (const t of sstables) console.log("  " + t.id + ": " + t.rows.map((r) => r[0].slice(5) + "=" + (r[1] === TOMB ? "X" : r[1])).join(" "));
console.log("  memtable: " + [...memtable.entries()].map((r) => r[0].slice(5) + "=" + (r[1] === TOMB ? "X" : r[1])).join(" "));

console.log("\n== 2. Reads go newest to oldest; Bloom filters skip files ==");
for (const k of ["user:09", "user:03", "user:07", "user:01", "user:05"]) console.log("  " + show(k));
let withBloom = 0, without = 0;
for (let i = 100; i < 140; i++) {
  const k = "user:" + i;
  withBloom += get(k, true)[2];
  without += get(k, false)[2];
}
console.log("  40 lookups of absent keys: " + without + " file reads without Bloom filters, " + withBloom + " with them (each one a false positive)");

console.log("\n== 3. Compaction merges every SSTable, newest version wins, tombstones dropped ==");
const before = sstables.reduce((n, t) => n + t.rows.length, 0);
const merged = new Map();
for (const t of [...sstables].reverse()) for (const [k, v] of t.rows) merged.set(k, v); // oldest first, newer overwrite
const live = [...merged.entries()].filter((r) => r[1] !== TOMB).sort((a, b) => (a[0] < b[0] ? -1 : 1));
compactWrites += live.length;
const filesBefore = sstables.length;
const bloom = new Array(BITS).fill(false);
for (const [k] of live) for (const b of bitsFor(k)) bloom[b] = true;
sstables = [{ id: "sst-c1", rows: live, bloom }];
console.log("  " + filesBefore + " files with " + before + " rows -> 1 file with " + live.length + " live rows (space amplification was " + (before / live.length).toFixed(2) + "x)");
console.log("  " + show("user:03"));
console.log("  " + show("user:05"));
console.log("  write amplification: " + userWrites + " user writes -> " + walWrites + " WAL + " + flushWrites + " flush + " + compactWrites + " compaction = " + ((walWrites + flushWrites + compactWrites) / userWrites).toFixed(2) + "x");