How would you design a distributed message queue?

// One replicated partition (3 brokers), simulated in-process: acks, ISR,
// high watermark, min.insync.replicas and unclean leader election.
const NAMES = ["b1", "b2", "b3"];
class Partition {
  constructor(minIsr, unclean) {
    this.logs = [[], [], []]; this.alive = [true, true, true]; this.isr = [0, 1, 2];
    this.leader = 0; this.minIsr = minIsr; this.unclean = unclean; this.pending = []; this.acked = [];
  }
  hw() { // high watermark: highest offset every ISR member has
    if (this.leader < 0) return 0;
    let h = this.logs[this.leader].length;
    for (const b of this.isr) h = Math.min(h, this.logs[b].length);
    return h;
  }
  produce(m, acks) {
    if (this.leader < 0) return m + ": partition OFFLINE";
    if (acks === "all" && this.isr.length < this.minIsr) return m + ": REJECTED NotEnoughReplicas (isr=" + this.isr.length + ")";
    this.logs[this.leader].push(m);
    if (acks === "1") { this.acked.push(m); return m + ": acked by leader only"; }
    this.pending.push(m); this.advance();
    return this.acked.indexOf(m) >= 0 ? m + ": acked, isr=" + this.isrNames() : m + ": waiting for ISR " + this.isrNames();
  }
  fetch(f) { // follower f copies the leader log
    if (this.alive[f] && this.leader >= 0) { this.logs[f] = this.logs[this.leader].slice(); this.advance(); }
  }
  advance() {
    const h = this.hw(), keep = [];
    for (const m of this.pending) { if (this.logs[this.leader].indexOf(m) < h) this.acked.push(m); else keep.push(m); }
    this.pending = keep;
  }
  shrink(b) { this.isr = this.isr.filter((x) => x !== b); }
  crash(b) {
    this.alive[b] = false; this.shrink(b);
    if (b !== this.leader) return NAMES[b] + " crashed, isr=" + this.isrNames();
    let cand = this.isr.filter((x) => this.alive[x]), how = "clean";
    if (cand.length === 0 && this.unclean) { cand = [0, 1, 2].filter((x) => this.alive[x]); how = "UNCLEAN"; }
    this.pending = []; // in-flight requests time out; the producer must retry
    if (cand.length === 0) { this.leader = -1; return NAMES[b] + " (leader) crashed, no in-sync replica left: partition offline"; }
    this.leader = cand[0];
    if (this.isr.indexOf(this.leader) < 0) this.isr = [this.leader];
    return NAMES[b] + " (leader) crashed, " + how + " election -> leader " + NAMES[this.leader] + " log=[" + this.logs[this.leader].join(",") + "]";
  }
  isrNames() { return "{" + this.isr.map((b) => NAMES[b]).join(",") + "}"; }
  report() {
    if (this.leader < 0) { console.log("  acked=" + this.acked.length + ", none readable: partition unavailable until b1, the last in-sync replica, returns"); return; }
    const log = this.logs[this.leader];
    const lost = this.acked.filter((m) => log.indexOf(m) < 0);
    console.log("  acked=" + this.acked.length + " lost acked=" + lost.length + (lost.length ? " [" + lost.join(",") + "]" : "") + " final log=[" + log.join(",") + "]");
  }
}

console.log("== 1. acks=1: leader acknowledges before followers copy ==");
let p = new Partition(1, false);
for (const m of ["m1", "m2", "m3"]) p.produce(m, "1");
p.fetch(1); p.fetch(2);
console.log("  " + p.produce("m4", "1")); console.log("  " + p.produce("m5", "1"));
console.log("  leader log end=" + p.logs[0].length + " high watermark=" + p.hw() + " (consumers can read 3)");
console.log("  " + p.crash(0)); p.report();

console.log("== 2. acks=all, min.insync.replicas=2: same crash ==");
p = new Partition(2, false);
for (const m of ["m1", "m2", "m3"]) p.produce(m, "all");
p.fetch(1); p.fetch(2);
console.log("  " + p.produce("m4", "all")); console.log("  " + p.produce("m5", "all"));
console.log("  " + p.crash(0));
console.log("  producer retries m4, m5 on the new leader:");
p.produce("m4", "all"); p.produce("m5", "all"); p.fetch(2); p.report();

for (const [minIsr, unclean] of [[2, false], [1, false], [1, true]]) {
  console.log("== 3. slow follower, then two crashes: min.insync.replicas=" + minIsr + " unclean.leader.election=" + unclean + " ==");
  p = new Partition(minIsr, unclean);
  p.produce("m1", "all"); p.fetch(1); p.fetch(2);
  p.shrink(2); // b3 stops fetching (slow disk) and drops out of the ISR
  for (const m of ["m2", "m3", "m4"]) { p.produce(m, "all"); p.fetch(1); }
  console.log("  b3 fell behind, isr=" + p.isrNames() + ", b3 log=[" + p.logs[2].join(",") + "]");
  console.log("  " + p.crash(1));
  console.log("  " + p.produce("m5", "all"));
  console.log("  " + p.crash(0)); p.report();
}

console.log("== 4. capacity estimate ==");
const msgs = 1000000, bytes = 1000, rf = 3, days = 7, brokerMBs = 200, partMBs = 10;
const ingress = (msgs * bytes) / 1000000, disk = ingress * rf;
const brokers = Math.ceil((disk * 10) / (brokerMBs * 6)); // keep each broker at 60% of its write budget
const storageGB = (ingress * 86400 * days * rf) / 1000;
console.log("  ingress " + ingress + " MB/s, disk writes with RF=" + rf + ": " + disk + " MB/s, replication traffic " + (disk - ingress) + " MB/s");
console.log("  brokers at 60% of " + brokerMBs + " MB/s: " + brokers + ", storage for " + days + " days: " + storageGB + " GB (" + Math.floor(storageGB / brokers) + " GB per broker)");
console.log("  partitions at " + partMBs + " MB/s each: at least " + Math.ceil(ingress / partMBs) + " (also the max useful consumers per group)");