Consensus, Leader Election and Distributed Locks

package main

import (
	"fmt"
	"strings"
)

// 1. Clocks. Server B's clock runs 2 seconds fast. B writes price=90 first; A reads it and
// then writes price=100. The later write is A's.
func clocks() {
	type write struct {
		server  string
		price   int
		wall    int // what the server's own clock said, in ms
		lamport int
	}
	// Real times: B at 500 ms, A at 1000 ms. B's clock adds 2000 ms.
	b := write{"B", 90, 500 + 2000, 1}
	// A received B's write (carrying Lamport time 1) before writing, so A's counter becomes
	// max(own 0, seen 1) + 1 = 2.
	a := write{"A", 100, 1000, max(0, b.lamport) + 1}

	pick := func(by func(w write) int) write {
		if by(a) > by(b) {
			return a
		}
		return b
	}
	byWall := pick(func(w write) int { return w.wall })
	byLamport := pick(func(w write) int { return w.lamport })
	fmt.Println("clocks: B writes price=90, A sees it and writes price=100; B's clock is 2 s fast")
	fmt.Printf("  last write wins by wall clock: price=%d (B's stamp %d ms beats A's %d ms)\n", byWall.price, b.wall, a.wall)
	fmt.Printf("  last write wins by Lamport clock: price=%d (A's counter %d beats B's %d)\n", byLamport.price, a.lamport, b.lamport)
}

// 2. Leader election, Raft style. Each node votes at most once per term, for the first
// candidate whose request reaches it. A candidate needs a majority of all 5 nodes.
func election(term int, candidates []string, arrival map[string][]string) (string, bool) {
	nodes := []string{"N1", "N2", "N3", "N4", "N5"}
	votes := map[string]int{}
	votedFor := map[string]string{}
	for _, c := range candidates { // a candidate votes for itself
		votedFor[c] = c
		votes[c]++
	}
	for _, n := range nodes {
		if n == "N1" || votedFor[n] != "" { // N1 is down
			continue
		}
		votedFor[n] = arrival[n][0] // the first request to arrive wins this node's vote
		votes[arrival[n][0]]++
	}
	var tally []string
	winner := ""
	for _, c := range candidates {
		tally = append(tally, fmt.Sprintf("%s %d", c, votes[c]))
		if votes[c] >= 3 {
			winner = c
		}
	}
	fmt.Printf("  term %d: candidates %s, votes %s -> ", term, strings.Join(candidates, ", "), strings.Join(tally, ", "))
	if winner == "" {
		fmt.Println("no majority, wait a random time and try again")
		return "", false
	}
	fmt.Printf("%s leads with %d of 5\n", winner, votes[winner])
	return winner, true
}

// 3. A lock with a 5 s lease. Client A gets it, then freezes for 10 s (a long GC pause).
func lockDemo(fencing bool) string {
	token := 32
	highest := 0 // the storage remembers the highest token it has accepted
	file := ""
	write := func(who string, tok int) {
		if fencing && tok < highest {
			return // a stale lock holder: reject
		}
		highest = max(highest, tok)
		file = who
	}
	token++
	aToken := token // t=0: A gets the lock
	token++
	bToken := token // t=6 s: the lease has expired, B gets the lock
	write("B", bToken)
	write("A", aToken) // t=10 s: A wakes up, still believing it holds the lock
	return file
}

func main() {
	clocks()

	fmt.Println("election: leader N1 (term 1) dies; 4 nodes left, a majority is 3 of 5")
	if _, ok := election(2, []string{"N3", "N5"}, map[string][]string{"N2": {"N3", "N5"}, "N4": {"N5", "N3"}}); !ok {
		election(3, []string{"N5"}, map[string][]string{"N2": {"N5"}, "N3": {"N5"}, "N4": {"N5"}})
	}

	fmt.Println("lock with a 5 s lease: A gets token 33, pauses 10 s; B gets token 34 at 6 s and writes; A wakes and writes")
	fmt.Printf("  without fencing: the file says %s (B's write is lost)\n", lockDemo(false))
	fmt.Printf("  with fencing:    the file says %s (token 33 rejected, 34 already seen)\n", lockDemo(true))
}

from typing import Callable, NamedTuple


class Write(NamedTuple):
    server: str
    price: int
    wall: int  # what the server's own clock said, in ms
    lamport: int


def clocks() -> None:
    """1. Clocks. Server B's clock runs 2 seconds fast. B writes price=90 first; A reads it and
    then writes price=100. The later write is A's."""
    # Real times: B at 500 ms, A at 1000 ms. B's clock adds 2000 ms.
    b = Write("B", 90, 500 + 2000, 1)
    # A received B's write (carrying Lamport time 1) before writing, so A's counter becomes
    # max(own 0, seen 1) + 1 = 2.
    a = Write("A", 100, 1000, max(0, b.lamport) + 1)

    def pick(by: Callable[[Write], int]) -> Write:
        return a if by(a) > by(b) else b

    by_wall = pick(lambda w: w.wall)
    by_lamport = pick(lambda w: w.lamport)
    print("clocks: B writes price=90, A sees it and writes price=100; B's clock is 2 s fast")
    print(f"  last write wins by wall clock: price={by_wall.price} (B's stamp {b.wall} ms beats A's {a.wall} ms)")
    print(f"  last write wins by Lamport clock: price={by_lamport.price} (A's counter {a.lamport} beats B's {b.lamport})")


def election(term: int, candidates: list[str], arrival: dict[str, list[str]]) -> bool:
    """2. Leader election, Raft style. Each node votes at most once per term, for the first
    candidate whose request reaches it. A candidate needs a majority of all 5 nodes."""
    votes: dict[str, int] = {}
    voted_for: dict[str, str] = {}
    for c in candidates:  # a candidate votes for itself
        voted_for[c] = c
        votes[c] = votes.get(c, 0) + 1
    for n in ["N1", "N2", "N3", "N4", "N5"]:
        if n == "N1" or n in voted_for:  # N1 is down
            continue
        voted_for[n] = arrival[n][0]  # the first request to arrive wins this node's vote
        votes[arrival[n][0]] = votes.get(arrival[n][0], 0) + 1
    tally = ", ".join(f"{c} {votes[c]}" for c in candidates)
    winner = next((c for c in reversed(candidates) if votes[c] >= 3), "")
    line = f"  term {term}: candidates {', '.join(candidates)}, votes {tally} -> "
    if not winner:
        print(line + "no majority, wait a random time and try again")
        return False
    print(line + f"{winner} leads with {votes[winner]} of 5")
    return True


def lock_demo(fencing: bool) -> str:
    """3. A lock with a 5 s lease. Client A gets it, then freezes for 10 s (a long GC pause)."""
    state = {"highest": 0, "file": ""}  # the storage remembers the highest token it has accepted

    def write(who: str, tok: int) -> None:
        if fencing and tok < state["highest"]:
            return  # a stale lock holder: reject
        state["highest"] = max(state["highest"], tok)
        state["file"] = who

    token = 32
    token += 1
    a_token = token  # t=0: A gets the lock
    token += 1
    b_token = token  # t=6 s: the lease has expired, B gets the lock
    write("B", b_token)
    write("A", a_token)  # t=10 s: A wakes up, still believing it holds the lock
    return str(state["file"])


clocks()

print("election: leader N1 (term 1) dies; 4 nodes left, a majority is 3 of 5")
if not election(2, ["N3", "N5"], {"N2": ["N3", "N5"], "N4": ["N5", "N3"]}):
    election(3, ["N5"], {"N2": ["N5"], "N3": ["N5"], "N4": ["N5"]})

print("lock with a 5 s lease: A gets token 33, pauses 10 s; B gets token 34 at 6 s and writes; A wakes and writes")
print(f"  without fencing: the file says {lock_demo(False)} (B's write is lost)")
print(f"  with fencing:    the file says {lock_demo(True)} (token 33 rejected, 34 already seen)")

type Write = { server: string; price: number; wall: number; lamport: number }; // wall: the server's own clock, in ms

/** 1. Clocks. Server B's clock runs 2 seconds fast. B writes price=90 first; A reads it and
 * then writes price=100. The later write is A's. */
function clocks() {
  // Real times: B at 500 ms, A at 1000 ms. B's clock adds 2000 ms.
  const b: Write = { server: "B", price: 90, wall: 500 + 2000, lamport: 1 };
  // A received B's write (carrying Lamport time 1) before writing, so A's counter becomes
  // max(own 0, seen 1) + 1 = 2.
  const a: Write = { server: "A", price: 100, wall: 1000, lamport: Math.max(0, b.lamport) + 1 };

  const pick = (by: (w: Write) => number) => (by(a) > by(b) ? a : b);
  const byWall = pick((w) => w.wall);
  const byLamport = pick((w) => w.lamport);
  console.log("clocks: B writes price=90, A sees it and writes price=100; B's clock is 2 s fast");
  console.log(`  last write wins by wall clock: price=${byWall.price} (B's stamp ${b.wall} ms beats A's ${a.wall} ms)`);
  console.log(`  last write wins by Lamport clock: price=${byLamport.price} (A's counter ${a.lamport} beats B's ${b.lamport})`);
}

/** 2. Leader election, Raft style. Each node votes at most once per term, for the first
 * candidate whose request reaches it. A candidate needs a majority of all 5 nodes. */
function election(term: number, candidates: string[], arrival: Record<string, string[]>): boolean {
  const votes = new Map<string, number>();
  const votedFor = new Map<string, string>();
  for (const c of candidates) {
    // a candidate votes for itself
    votedFor.set(c, c);
    votes.set(c, (votes.get(c) ?? 0) + 1);
  }
  for (const n of ["N1", "N2", "N3", "N4", "N5"]) {
    if (n === "N1" || votedFor.has(n)) continue; // N1 is down
    const first = arrival[n]![0]!; // the first request to arrive wins this node's vote
    votedFor.set(n, first);
    votes.set(first, (votes.get(first) ?? 0) + 1);
  }
  const tally = candidates.map((c) => `${c} ${votes.get(c)}`).join(", ");
  const winner = [...candidates].reverse().find((c) => votes.get(c)! >= 3) ?? "";
  const line = `  term ${term}: candidates ${candidates.join(", ")}, votes ${tally} -> `;
  if (winner === "") {
    console.log(line + "no majority, wait a random time and try again");
    return false;
  }
  console.log(line + `${winner} leads with ${votes.get(winner)} of 5`);
  return true;
}

/** 3. A lock with a 5 s lease. Client A gets it, then freezes for 10 s (a long GC pause). */
function lockDemo(fencing: boolean): string {
  let highest = 0; // the storage remembers the highest token it has accepted
  let file = "";
  const write = (who: string, tok: number) => {
    if (fencing && tok < highest) return; // a stale lock holder: reject
    highest = Math.max(highest, tok);
    file = who;
  };
  let token = 32;
  const aToken = ++token; // t=0: A gets the lock
  const bToken = ++token; // t=6 s: the lease has expired, B gets the lock
  write("B", bToken);
  write("A", aToken); // t=10 s: A wakes up, still believing it holds the lock
  return file;
}

clocks();

console.log("election: leader N1 (term 1) dies; 4 nodes left, a majority is 3 of 5");
if (!election(2, ["N3", "N5"], { N2: ["N3", "N5"], N4: ["N5", "N3"] })) {
  election(3, ["N5"], { N2: ["N5"], N3: ["N5"], N4: ["N5"] });
}

console.log("lock with a 5 s lease: A gets token 33, pauses 10 s; B gets token 34 at 6 s and writes; A wakes and writes");
console.log(`  without fencing: the file says ${lockDemo(false)} (B's write is lost)`);
console.log(`  with fencing:    the file says ${lockDemo(true)} (token 33 rejected, 34 already seen)`);

use std::collections::HashMap;

struct Write {
    price: i32,
    wall: i32, // what the server's own clock said, in ms
    lamport: i32,
}

/// 1. Clocks. Server B's clock runs 2 seconds fast. B writes price=90 first; A reads it and
/// then writes price=100. The later write is A's.
fn clocks() {
    // Real times: B at 500 ms, A at 1000 ms. B's clock adds 2000 ms.
    let b = Write { price: 90, wall: 500 + 2000, lamport: 1 };
    // A received B's write (carrying Lamport time 1) before writing, so A's counter becomes
    // max(own 0, seen 1) + 1 = 2.
    let a = Write { price: 100, wall: 1000, lamport: 0.max(b.lamport) + 1 };

    let pick = |by: fn(&Write) -> i32| if by(&a) > by(&b) { &a } else { &b };
    let by_wall = pick(|w| w.wall);
    let by_lamport = pick(|w| w.lamport);
    println!("clocks: B writes price=90, A sees it and writes price=100; B's clock is 2 s fast");
    println!("  last write wins by wall clock: price={} (B's stamp {} ms beats A's {} ms)", by_wall.price, b.wall, a.wall);
    println!("  last write wins by Lamport clock: price={} (A's counter {} beats B's {})", by_lamport.price, a.lamport, b.lamport);
}

/// 2. Leader election, Raft style. Each node votes at most once per term, for the first
/// candidate whose request reaches it. A candidate needs a majority of all 5 nodes.
fn election(term: i32, candidates: &[&str], arrival: &HashMap<&str, Vec<&str>>) -> bool {
    let mut votes: HashMap<&str, i32> = HashMap::new();
    let mut voted_for: HashMap<&str, &str> = HashMap::new();
    for &c in candidates {
        // a candidate votes for itself
        voted_for.insert(c, c);
        *votes.entry(c).or_default() += 1;
    }
    for n in ["N1", "N2", "N3", "N4", "N5"] {
        if n == "N1" || voted_for.contains_key(n) {
            continue; // N1 is down
        }
        let first = arrival[n][0]; // the first request to arrive wins this node's vote
        voted_for.insert(n, first);
        *votes.entry(first).or_default() += 1;
    }
    let tally: Vec<String> = candidates.iter().map(|c| format!("{c} {}", votes[c])).collect();
    let winner = candidates.iter().rev().find(|c| votes[*c] >= 3);
    let line = format!("  term {term}: candidates {}, votes {} -> ", candidates.join(", "), tally.join(", "));
    match winner {
        None => {
            println!("{line}no majority, wait a random time and try again");
            false
        }
        Some(w) => {
            println!("{line}{w} leads with {} of 5", votes[w]);
            true
        }
    }
}

/// 3. A lock with a 5 s lease. Client A gets it, then freezes for 10 s (a long GC pause).
fn lock_demo(fencing: bool) -> String {
    let mut highest = 0; // the storage remembers the highest token it has accepted
    let mut file = String::new();
    let mut write = |who: &str, tok: i32| {
        if fencing && tok < highest {
            return; // a stale lock holder: reject
        }
        highest = highest.max(tok);
        file = who.to_string();
    };
    let mut token = 32;
    token += 1;
    let a_token = token; // t=0: A gets the lock
    token += 1;
    let b_token = token; // t=6 s: the lease has expired, B gets the lock
    write("B", b_token);
    write("A", a_token); // t=10 s: A wakes up, still believing it holds the lock
    file
}

fn main() {
    clocks();

    println!("election: leader N1 (term 1) dies; 4 nodes left, a majority is 3 of 5");
    let first = HashMap::from([("N2", vec!["N3", "N5"]), ("N4", vec!["N5", "N3"])]);
    if !election(2, &["N3", "N5"], &first) {
        let second = HashMap::from([("N2", vec!["N5"]), ("N3", vec!["N5"]), ("N4", vec!["N5"])]);
        election(3, &["N5"], &second);
    }

    println!("lock with a 5 s lease: A gets token 33, pauses 10 s; B gets token 34 at 6 s and writes; A wakes and writes");
    println!("  without fencing: the file says {} (B's write is lost)", lock_demo(false));
    println!("  with fencing:    the file says {} (token 33 rejected, 34 already seen)", lock_demo(true));
}

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.ToIntFunction;

public class Main {
    record Write(String server, int price, int wall, int lamport) {} // wall: the server's own clock, in ms

    /** 1. Clocks. Server B's clock runs 2 seconds fast. B writes price=90 first; A reads it and
     *  then writes price=100. The later write is A's. */
    static void clocks() {
        // Real times: B at 500 ms, A at 1000 ms. B's clock adds 2000 ms.
        Write b = new Write("B", 90, 500 + 2000, 1);
        // A received B's write (carrying Lamport time 1) before writing, so A's counter becomes
        // max(own 0, seen 1) + 1 = 2.
        Write a = new Write("A", 100, 1000, Math.max(0, b.lamport()) + 1);

        ToIntFunction<Write> wall = Write::wall, lamport = Write::lamport;
        Write byWall = wall.applyAsInt(a) > wall.applyAsInt(b) ? a : b;
        Write byLamport = lamport.applyAsInt(a) > lamport.applyAsInt(b) ? a : b;
        System.out.println("clocks: B writes price=90, A sees it and writes price=100; B's clock is 2 s fast");
        System.out.printf("  last write wins by wall clock: price=%d (B's stamp %d ms beats A's %d ms)%n", byWall.price(), b.wall(), a.wall());
        System.out.printf("  last write wins by Lamport clock: price=%d (A's counter %d beats B's %d)%n", byLamport.price(), a.lamport(), b.lamport());
    }

    /** 2. Leader election, Raft style. Each node votes at most once per term, for the first
     *  candidate whose request reaches it. A candidate needs a majority of all 5 nodes. */
    static boolean election(int term, List<String> candidates, Map<String, List<String>> arrival) {
        Map<String, Integer> votes = new HashMap<>();
        Map<String, String> votedFor = new HashMap<>();
        for (String c : candidates) { // a candidate votes for itself
            votedFor.put(c, c);
            votes.merge(c, 1, Integer::sum);
        }
        for (String n : List.of("N1", "N2", "N3", "N4", "N5")) {
            if (n.equals("N1") || votedFor.containsKey(n)) continue; // N1 is down
            String first = arrival.get(n).get(0); // the first request to arrive wins this node's vote
            votedFor.put(n, first);
            votes.merge(first, 1, Integer::sum);
        }
        List<String> tally = new ArrayList<>();
        String winner = "";
        for (String c : candidates) {
            tally.add(c + " " + votes.get(c));
            if (votes.get(c) >= 3) winner = c;
        }
        String line = "  term " + term + ": candidates " + String.join(", ", candidates) + ", votes " + String.join(", ", tally) + " -> ";
        if (winner.isEmpty()) {
            System.out.println(line + "no majority, wait a random time and try again");
            return false;
        }
        System.out.println(line + winner + " leads with " + votes.get(winner) + " of 5");
        return true;
    }

    /** 3. A lock with a 5 s lease. Client A gets it, then freezes for 10 s (a long GC pause). */
    static String lockDemo(boolean fencing) {
        int[] highest = {0}; // the storage remembers the highest token it has accepted
        String[] file = {""};
        interface Storage { void write(String who, int tok); }
        Storage storage = (who, tok) -> {
            if (fencing && tok < highest[0]) return; // a stale lock holder: reject
            highest[0] = Math.max(highest[0], tok);
            file[0] = who;
        };
        int token = 32;
        int aToken = ++token; // t=0: A gets the lock
        int bToken = ++token; // t=6 s: the lease has expired, B gets the lock
        storage.write("B", bToken);
        storage.write("A", aToken); // t=10 s: A wakes up, still believing it holds the lock
        return file[0];
    }

    public static void main(String[] args) {
        clocks();

        System.out.println("election: leader N1 (term 1) dies; 4 nodes left, a majority is 3 of 5");
        if (!election(2, List.of("N3", "N5"), Map.of("N2", List.of("N3", "N5"), "N4", List.of("N5", "N3")))) {
            election(3, List.of("N5"), Map.of("N2", List.of("N5"), "N3", List.of("N5"), "N4", List.of("N5")));
        }

        System.out.println("lock with a 5 s lease: A gets token 33, pauses 10 s; B gets token 34 at 6 s and writes; A wakes and writes");
        System.out.printf("  without fencing: the file says %s (B's write is lost)%n", lockDemo(false));
        System.out.printf("  with fencing:    the file says %s (token 33 rejected, 34 already seen)%n", lockDemo(true));
    }
}

#include <algorithm>
#include <cstdio>
#include <functional>
#include <map>
#include <string>
#include <vector>

struct Write {
    const char* server;
    int price;
    int wall;  // what the server's own clock said, in ms
    int lamport;
};

std::string join(const std::vector<std::string>& xs, const std::string& sep) {
    std::string out;
    for (size_t i = 0; i < xs.size(); i++) out += (i ? sep : "") + xs[i];
    return out;
}

// 1. Clocks. Server B's clock runs 2 seconds fast. B writes price=90 first; A reads it and
// then writes price=100. The later write is A's.
void clocks() {
    // Real times: B at 500 ms, A at 1000 ms. B's clock adds 2000 ms.
    Write b{"B", 90, 500 + 2000, 1};
    // A received B's write (carrying Lamport time 1) before writing, so A's counter becomes
    // max(own 0, seen 1) + 1 = 2.
    Write a{"A", 100, 1000, std::max(0, b.lamport) + 1};

    auto pick = [&](const std::function<int(const Write&)>& by) { return by(a) > by(b) ? a : b; };
    Write byWall = pick([](const Write& w) { return w.wall; });
    Write byLamport = pick([](const Write& w) { return w.lamport; });
    std::printf("clocks: B writes price=90, A sees it and writes price=100; B's clock is 2 s fast\n");
    std::printf("  last write wins by wall clock: price=%d (B's stamp %d ms beats A's %d ms)\n", byWall.price, b.wall, a.wall);
    std::printf("  last write wins by Lamport clock: price=%d (A's counter %d beats B's %d)\n", byLamport.price, a.lamport, b.lamport);
}

// 2. Leader election, Raft style. Each node votes at most once per term, for the first
// candidate whose request reaches it. A candidate needs a majority of all 5 nodes.
bool election(int term, const std::vector<std::string>& candidates, std::map<std::string, std::vector<std::string>> arrival) {
    std::map<std::string, int> votes;
    std::map<std::string, std::string> votedFor;
    for (const auto& c : candidates) {  // a candidate votes for itself
        votedFor[c] = c;
        votes[c]++;
    }
    for (const std::string n : {"N1", "N2", "N3", "N4", "N5"}) {
        if (n == "N1" || votedFor.count(n)) continue;  // N1 is down
        std::string first = arrival[n][0];          // the first request to arrive wins this node's vote
        votedFor[n] = first;
        votes[first]++;
    }
    std::vector<std::string> tally;
    std::string winner;
    for (const auto& c : candidates) {
        tally.push_back(c + " " + std::to_string(votes[c]));
        if (votes[c] >= 3) winner = c;
    }
    std::string line = "  term " + std::to_string(term) + ": candidates " + join(candidates, ", ") + ", votes " + join(tally, ", ") + " -> ";
    if (winner.empty()) {
        std::printf("%sno majority, wait a random time and try again\n", line.c_str());
        return false;
    }
    std::printf("%s%s leads with %d of 5\n", line.c_str(), winner.c_str(), votes[winner]);
    return true;
}

// 3. A lock with a 5 s lease. Client A gets it, then freezes for 10 s (a long GC pause).
std::string lockDemo(bool fencing) {
    int highest = 0;  // the storage remembers the highest token it has accepted
    std::string file;
    auto write = [&](const std::string& who, int tok) {
        if (fencing && tok < highest) return;  // a stale lock holder: reject
        highest = std::max(highest, tok);
        file = who;
    };
    int token = 32;
    int aToken = ++token;  // t=0: A gets the lock
    int bToken = ++token;  // t=6 s: the lease has expired, B gets the lock
    write("B", bToken);
    write("A", aToken);  // t=10 s: A wakes up, still believing it holds the lock
    return file;
}

int main() {
    clocks();

    std::printf("election: leader N1 (term 1) dies; 4 nodes left, a majority is 3 of 5\n");
    if (!election(2, {"N3", "N5"}, {{"N2", {"N3", "N5"}}, {"N4", {"N5", "N3"}}})) {
        election(3, {"N5"}, {{"N2", {"N5"}}, {"N3", {"N5"}}, {"N4", {"N5"}}});
    }

    std::printf("lock with a 5 s lease: A gets token 33, pauses 10 s; B gets token 34 at 6 s and writes; A wakes and writes\n");
    std::printf("  without fencing: the file says %s (B's write is lost)\n", lockDemo(false).c_str());
    std::printf("  with fencing:    the file says %s (token 33 rejected, 34 already seen)\n", lockDemo(true).c_str());
}