Build a Three-Region Raft Simulator

Introduction

30 Second Summary

A service spread across regions can look reliable until one region loses contact. Every accepted update then needs proof that enough of the system agreed to make it real.

In this project, you will build a terminal-driven three-region consensus simulator in Java with Apache Pekko Typed. The simulation proves how majority agreement protects acknowledged writes during a network partition.

What You'll Build

Run one command to watch a network partition trigger leader failover before a benchmark report proves which writes committed.

By the end of this project, you'll have:

• A three-region terminal demo where us-east, eu-west, and ap-south own independent state behind visible regional delays.
• Clear majority-commit evidence that ties every successful acknowledgement to a quorum. The isolated former leader visibly retains uncommitted entries.
• A reproducible benchmark report covering election duration, throughput, average and p95 commit latency, rejected writes, outstanding work, and per-node state. A guarantee matrix separates the demonstrated behavior from full Raft safety.
• Secret Mission: Compare request windows 4, 16, and 32 to choose the smallest window near the best throughput while defending its latency and outstanding-work trade-offs.

Are there any prerequisites?

You need a Mac with a JDK and code editor plus experience with Java syntax and clustered services. The first step pins Java 17 before setting up JBang and Visual Studio Code.

Before We Start

Before you build the simulator, lock in the evidence rule that guides every write and partition test. This commitment gives you a clear standard for judging the finished result.

Prepare the Local Java Workspace

Your evidence rule is clear. The simulator can only acknowledge writes that reach majority commitment.

The next challenge is creating a reproducible workspace without build-tool scaffolding. JBang keeps dependency pins inside the Java source files. It also gives the project access to Java 17 for the later Apache Pekko Typed 1.7.1 simulation.

Why JBang for this project?

A traditional Java build would require separate project configuration before you could model consensus behavior. JBang resolves dependencies from directives in the source file.

That keeps your attention on actor protocols and quorum behavior.

In this step, get ready to:

• Create the regional-raft-simulator folder in macOS Terminal.
• Prepare JBang 0.142.0 with a managed Java 17 runtime.
• Open the empty project folder in Visual Studio Code.

Create the local project folder

A fixed project location makes every later command reproducible. You will keep the simulator on your Desktop so it is easy to find.

• Click Spotlight in the macOS menu bar.
• Enter Terminal in the search field.
• Select Terminal from the results.
• Move to your Desktop by running this command:

cd ~/Desktop

What does this command do?

The cd command changes Terminal's current location. The ~ character represents your home folder.

• Create the project folder as your current Terminal location by running these commands:

mkdir regional-raft-simulator
cd regional-raft-simulator
pwd

What do these commands do?

• The first command creates the regional-raft-simulator directory.
• The second command moves Terminal into that directory.
• The final command prints the current path so you can confirm your location.

You should see a path ending in Desktop/regional-raft-simulator. Your empty project folder is now ready.

Folder command did not work?

An existing-folder message means the directory is already on your Desktop. Use the second command to switch into it.

A different printed path means Terminal is in another directory. Run the Desktop command again before repeating the folder commands.

Still stuck? Help me get Terminal into my regional-raft-simulator folder.

Install and verify JBang

JBang needs to be available on your shell's PATH before it can run the simulator. The version check reveals whether you can continue or need the installer.

• Check your current JBang state by running this command:

jbang --version

What does this command do?

The version command checks whether Terminal can find JBang on PATH. A version result also tells you whether the installed release matches this project.

✔️ I see version 0.142.0

That confirms the pinned JBang release is available on PATH. Your shell is ready for the Java runtime check.

ⓧ I see an older version

Your PATH currently points to an earlier JBang release. Installing the current release aligns your environment with the project.

• Install the pinned JBang release by running this command:

curl -Ls https://sh.jbang.dev | bash -s - app setup

What does this command do?

The command downloads JBang's setup script over HTTPS. The setup process installs JBang and adds its application commands to PATH.

• Restart Terminal if the installer requests it.
• Return to the project folder by running this command:

cd ~/Desktop/regional-raft-simulator

Why return to this path?

A restarted Terminal begins in your home folder. This command restores regional-raft-simulator as the current location.

• Verify the updated JBang release by running this command:

jbang --version

What should this confirm?

The command should now report JBang 0.142.0. That result confirms the upgrade is active in your restarted shell.

ⓧ Command not found

Terminal cannot find a JBang executable on PATH yet. The official setup script installs it for your shell.

• Install JBang by running this command:

curl -Ls https://sh.jbang.dev | bash -s - app setup

What does this command do?

The command downloads JBang's setup script over HTTPS. The setup process installs JBang and configures its application commands on PATH.

• Restart Terminal if the installer requests it.
• Return to the project folder by running this command:

cd ~/Desktop/regional-raft-simulator

Why return to this path?

Restarting Terminal resets its current location. This command brings you back to the folder used throughout the project.

• Verify the JBang installation by running this command:

jbang --version

What should this confirm?

The command should report JBang 0.142.0. That result proves Terminal can now find JBang on PATH.

Installer or PATH issue?

If the download does not finish, check that your Mac has an internet connection before repeating the installer.

If Terminal still cannot find JBang after installation, restart Terminal so it reloads the updated PATH.

Need help? Help me troubleshoot my JBang installation on macOS.

Java 17 is the runtime pinned by the simulator's entry point. JBang can manage that runtime independently of another JDK already installed on your Mac.

• Make Java 17 available to JBang by running this command:

jbang jdk install 17

What does this command do?

JBang checks its managed JDK collection for Java 17. It downloads the runtime into its cache when that version is missing.

• List the JDKs managed by JBang by running this command:

jbang jdk list

What should the list show?

The list should contain an installed Java 17 runtime. This is the runtime JBang can select when a script contains the project's exact Java directive.

Good progress. JBang can now supply the Java runtime required by the simulator.

Java download did not finish?

A failed download usually points to a network restriction or an interrupted transfer. Reconnect before running the installation command again.

Need a hand? Help me diagnose why JBang cannot install Java 17.

Open the folder in Visual Studio Code

The command-line launcher connects your current Terminal location to the editor. This gives every source file a clear home when you begin modeling the regional actors.

Before you check, picture the exact JBang version Terminal should print.

• Run the final JBang version check with this command:

jbang --version

What does this final check prove?

A 0.142.0 result proves the pinned JBang executable is active in this Terminal session.

You should see 0.142.0 in the output. The JBang part of your workspace is verified.

Before you launch the editor, picture the folder name you expect to see in the Explorer sidebar.

• Open the current project folder in Visual Studio Code by running this command:

code .

What does this command do?

The code command launches Visual Studio Code. The dot means the current Terminal folder.

✔️ Visual Studio Code opens the folder

You should see regional-raft-simulator as the folder name in the Explorer sidebar. The folder should be empty because no simulator source files exist yet.

ⓧ Terminal cannot find code

The Visual Studio Code launcher still needs to be added to PATH. You can install it from the editor's Command Palette.

• Click Spotlight in the macOS menu bar.
• Enter Visual Studio Code in the search field.
• Select Visual Studio Code from the results.
• Press Cmd+Shift+P (macOS) or Ctrl+Shift+P (Windows) to open the Command Palette.
• Enter shell command in the Command Palette.

What might Visual Studio Code ask?

The PATH action may offer to install GitHub Copilot CLI. That extra command is optional for this project.

• Select Shell Command: Install 'code' and optionally 'copilot' commands in PATH.
• Select No if the optional GitHub Copilot CLI prompt appears.
• Restart Terminal.
• Return to the project folder by running this command:

cd ~/Desktop/regional-raft-simulator

Why return to the folder?

The new Terminal session starts from your home folder. This command restores the project directory as the current location.

• Launch Visual Studio Code from the project folder by running this command:

code .

What should happen now?

Visual Studio Code should open regional-raft-simulator as the current folder. The successful launch confirms the command is now on PATH.

Your local workspace is ready. Terminal and Visual Studio Code now point to the same empty project folder.

Next, you will turn this empty workspace into three regional actors connected through a latency-aware simulated network.

Model Three Regional Actors

Your local Java workspace is ready. Visual Studio Code points at regional-raft-simulator while JBang can run Java 17 scripts.

Shared objects would hide the concurrency this simulator must expose. Apache Pekko Typed actors give each region its own state behind a typed protocol.

In this step, get ready to:

• Declare the Java 17 entry point with Apache Pekko Typed 1.7.1 as its pinned dependency.
• Model three regional actors behind a latency-aware simulated network.
• Run the simulator to observe three ready nodes plus an initial leader election.

Create the JBang entry point

The entry point tells JBang which Java version to use. It also lists every source file that belongs to the simulation.

• Select the Explorer view in the Visual Studio Code Activity Bar.
• Select the New File... button in the Explorer view.
• Enter RegionalRaftSimulation.java.
• Press Enter.
• Select the full-code tab below.
• Copy the complete entry point into RegionalRaftSimulation.java.
• Save the file with Cmd+S.

✔️ Awesome, I've got everything!

Your entry point should declare Java 17. It should also reference the three supporting source files.

ⓧ I'd like to double check the full code

///usr/bin/env jbang "$0" "$@" ; exit $?
//JAVA 17
//DEPS org.apache.pekko:pekko-actor-typed_2.13:1.7.1
//SOURCES SimulatedNetwork.java RaftNode.java Benchmark.java

import java.util.ArrayList;
import java.util.List;

import org.apache.pekko.actor.typed.ActorRef;
import org.apache.pekko.actor.typed.ActorSystem;
import org.apache.pekko.actor.typed.Behavior;
import org.apache.pekko.actor.typed.javadsl.Behaviors;

public class RegionalRaftSimulation {
    private record NodeSpec(String region, long electionTimeoutMs) {}

    public static void main(String[] args) {
        ActorSystem<Void> system = ActorSystem.create(guardian(), "regional-raft-simulation");
        system.getWhenTerminated().toCompletableFuture().join();
    }

    private static Behavior<Void> guardian() {
        return Behaviors.setup(context -> {
            ActorRef<SimulatedNetwork.Command> network =
                    context.spawn(SimulatedNetwork.create(), "simulated-network");
            ActorRef<Benchmark.Command> benchmark =
                    context.spawn(Benchmark.create(network), "benchmark");
            List<NodeSpec> specs = List.of(
                    new NodeSpec("us-east", 500),
                    new NodeSpec("eu-west", 800),
                    new NodeSpec("ap-south", 1100));
            List<ActorRef<RaftNode.Command>> nodeRefs = new ArrayList<>();
            for (NodeSpec spec : specs) {
                nodeRefs.add(context.spawn(
                        RaftNode.create(spec.region(), spec.electionTimeoutMs()),
                        "node-" + spec.region()));
            }
            List<RaftNode.Peer> cluster = new ArrayList<>();
            for (int i = 0; i < specs.size(); i++) {
                cluster.add(new RaftNode.Peer(specs.get(i).region(), nodeRefs.get(i)));
            }
            benchmark.tell(new Benchmark.Configure(cluster));
            for (ActorRef<RaftNode.Command> nodeRef : nodeRefs) {
                nodeRef.tell(new RaftNode.Initialize(cluster, network, benchmark));
            }
            System.out.println("[BOOT] Three regional actors started in one ActorSystem");
            return Behaviors.empty();
        });
    }
}

What does this file do?

• The JBang directives select Java 17. They also resolve the pinned Apache Pekko dependency.
• The guardian creates one ActorSystem named regional-raft-simulation.
• The guardian spawns the network actor plus the benchmark actor.
• The three NodeSpec records give each region a deterministic election timeout.

• Confirm the Explorer view lists RegionalRaftSimulation.java inside regional-raft-simulator.

The entry point now defines the actor topology. The supporting actor classes come next.

Give each region actor-owned state

An actor processes one message at a time. That boundary lets each regional node own mutable term, role, replica, commit, vote, leader, and pending-write state without shared-state guards.

Why actors instead of shared objects?

A shared object would let several execution paths touch the same consensus state. Actor-owned state changes only while that actor handles a typed message.

The protocol becomes the concurrency boundary. Every vote, heartbeat, append, commit, and status request crosses that boundary as a message.

The simulated network sits between regional actors. It delays delivery according to the sender and receiver regions.

• Select the New File... button in the Explorer view.
• Enter SimulatedNetwork.java.
• Press Enter.
• Select the full-code tab below.
• Copy the complete network actor into SimulatedNetwork.java.
• Save the file with Cmd+S.

✔️ Awesome, I've got everything!

Your network actor should delay messages between connected regions. It should drop traffic whenever either endpoint is isolated.

ⓧ I'd like to double check the full code

import java.time.Duration;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import org.apache.pekko.actor.typed.ActorRef;
import org.apache.pekko.actor.typed.Behavior;
import org.apache.pekko.actor.typed.javadsl.AbstractBehavior;
import org.apache.pekko.actor.typed.javadsl.ActorContext;
import org.apache.pekko.actor.typed.javadsl.Behaviors;
import org.apache.pekko.actor.typed.javadsl.Receive;
import org.apache.pekko.actor.typed.javadsl.TimerScheduler;

public final class SimulatedNetwork extends AbstractBehavior<SimulatedNetwork.Command> {
    public interface Command {}

    public record Send(String from, String to, ActorRef<RaftNode.Command> target,
                       RaftNode.Command payload) implements Command {}
    public record SetPartition(String region, boolean isolated,
                               ActorRef<Benchmark.Command> replyTo) implements Command {}
    private record Deliver(ActorRef<RaftNode.Command> target,
                           RaftNode.Command payload) implements Command {}

    private static final Map<String, Long> REGION_DELAY_MS = Map.of(
            "us-east", 20L, "eu-west", 60L, "ap-south", 100L);
    private final TimerScheduler<Command> timers;
    private final Set<String> isolatedRegions = new HashSet<>();
    private long sequence;
    private long droppedMessages;

    public static Behavior<Command> create() {
        return Behaviors.setup(context ->
                Behaviors.withTimers(timers -> new SimulatedNetwork(context, timers)));
    }

    private SimulatedNetwork(ActorContext<Command> context, TimerScheduler<Command> timers) {
        super(context);
        this.timers = timers;
    }

    @Override
    public Receive<Command> createReceive() {
        return newReceiveBuilder()
                .onMessage(Send.class, this::onSend)
                .onMessage(SetPartition.class, this::onSetPartition)
                .onMessage(Deliver.class, this::onDeliver)
                .build();
    }

    private Behavior<Command> onSend(Send message) {
        if (isolatedRegions.contains(message.from()) || isolatedRegions.contains(message.to())) {
            droppedMessages++;
            return this;
        }
        long delay = Math.max(5L,
                (REGION_DELAY_MS.getOrDefault(message.from(), 50L)
                        + REGION_DELAY_MS.getOrDefault(message.to(), 50L)) / 2L);
        timers.startSingleTimer(sequence++, new Deliver(message.target(), message.payload()),
                Duration.ofMillis(delay));
        return this;
    }

    private Behavior<Command> onSetPartition(SetPartition message) {
        if (message.isolated()) isolatedRegions.add(message.region());
        else isolatedRegions.remove(message.region());
        System.out.printf("[NETWORK] region=%s isolated=%s droppedSoFar=%d%n",
                message.region(), message.isolated(), droppedMessages);
        message.replyTo().tell(new Benchmark.PartitionApplied(
                message.region(), message.isolated()));
        return this;
    }

    private Behavior<Command> onDeliver(Deliver message) {
        message.target().tell(message.payload());
        return this;
    }
}

How does the network behave?

• The REGION_DELAY_MS map assigns delays of 20 ms to us-east, 60 ms to eu-west, and 100 ms to ap-south.
• The Send command schedules delivery after a calculated regional delay.
• The SetPartition command adds or removes a region from the isolated set.
• The network reports PartitionApplied before the benchmark continues.

• Confirm the Explorer view now lists SimulatedNetwork.java.

The transport boundary is now visible as its own actor. The regional node actor can use that boundary for every peer message.

• Select the New File... button in the Explorer view.
• Enter RaftNode.java.
• Press Enter.
• Select the full-code tab below.
• Copy the complete regional node actor into RaftNode.java.
• Save the file with Cmd+S.

✔️ Awesome, I've got everything!

Your regional node should own its election state plus separate replica and committed values. Its typed protocol should cover initialization, voting, heartbeats, replication, commitment, client writes, and status requests.

ⓧ I'd like to double check the full code

import java.time.Duration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.pekko.actor.typed.ActorRef;
import org.apache.pekko.actor.typed.Behavior;
import org.apache.pekko.actor.typed.javadsl.AbstractBehavior;
import org.apache.pekko.actor.typed.javadsl.ActorContext;
import org.apache.pekko.actor.typed.javadsl.Behaviors;
import org.apache.pekko.actor.typed.javadsl.Receive;
import org.apache.pekko.actor.typed.javadsl.TimerScheduler;

public final class RaftNode extends AbstractBehavior<RaftNode.Command> {
    public interface Command {}
    public record Peer(String region, ActorRef<Command> ref) {}
    public record Initialize(List<Peer> cluster, ActorRef<SimulatedNetwork.Command> network,
                             ActorRef<Benchmark.Command> observer) implements Command {}
    public record ClientWrite(long requestId, int delta, ActorRef<Benchmark.Command> replyTo,
                              int phaseToken) implements Command {}
    public record GetStatus(ActorRef<Benchmark.Command> replyTo) implements Command {}
    public record NodeReady(String region) implements Benchmark.Command {}
    public record LeaderElected(String region, int term,
                                long observedAtNanos) implements Benchmark.Command {}
    public record WriteResult(long requestId, int phaseToken, boolean committed,
                              String node, String reason) implements Benchmark.Command {}
    public record Status(String region, String role, int term, long commitIndex,
                         int committedValue, long replicaIndex, int replicaValue,
                         int pendingWrites) implements Benchmark.Command {}

    private record RequestVote(int term, String candidateRegion,
                               ActorRef<Command> candidateRef) implements Command {}
    private record VoteResponse(int term, String voterRegion,
                                boolean granted) implements Command {}
    private record Heartbeat(int term, String leaderRegion,
                             ActorRef<Command> leaderRef) implements Command {}
    private record AppendEntry(int term, String leaderRegion, ActorRef<Command> leaderRef,
                               long index, int value) implements Command {}
    private record AppendAck(int term, long index, String followerRegion,
                             boolean accepted) implements Command {}
    private record CommitEntry(int term, String leaderRegion, ActorRef<Command> leaderRef,
                               long index, int value) implements Command {}
    private enum ElectionTimeout implements Command { INSTANCE }
    private enum HeartbeatTick implements Command { INSTANCE }
    private enum Role { FOLLOWER, CANDIDATE, LEADER }
    private record LogEntry(int term, int value) {}

    private static final class PendingWrite {
        final int value;
        final long requestId;
        final int phaseToken;
        final ActorRef<Benchmark.Command> replyTo;
        final Set<String> acknowledgements = new HashSet<>();

        PendingWrite(int value, long requestId, int phaseToken,
                     ActorRef<Benchmark.Command> replyTo, String leaderRegion) {
            this.value = value;
            this.requestId = requestId;
            this.phaseToken = phaseToken;
            this.replyTo = replyTo;
            acknowledgements.add(leaderRegion);
        }
    }

    private static final String ELECTION_TIMER = "election-timer";
    private static final String HEARTBEAT_TIMER = "heartbeat-timer";
    private static final long HEARTBEAT_MS = 150L;
    private final String region;
    private final long electionTimeoutMs;
    private final TimerScheduler<Command> timers;
    private List<Peer> cluster = List.of();
    private ActorRef<SimulatedNetwork.Command> network;
    private ActorRef<Benchmark.Command> observer;
    private Role role = Role.FOLLOWER;
    private int currentTerm;
    private String votedFor;
    private String leaderRegion;
    private ActorRef<Command> leaderRef;
    private final Set<String> votes = new HashSet<>();
    private long lastIndex;
    private long commitIndex;
    private int replicaValue;
    private int committedValue;
    private final Map<Long, LogEntry> replicaLog = new HashMap<>();
    private final Map<Long, PendingWrite> pendingWrites = new HashMap<>();

    public static Behavior<Command> create(String region, long electionTimeoutMs) {
        return Behaviors.setup(context -> Behaviors.withTimers(timers ->
                new RaftNode(context, timers, region, electionTimeoutMs)));
    }

    private RaftNode(ActorContext<Command> context, TimerScheduler<Command> timers,
                     String region, long electionTimeoutMs) {
        super(context);
        this.timers = timers;
        this.region = region;
        this.electionTimeoutMs = electionTimeoutMs;
    }

    @Override
    public Receive<Command> createReceive() {
        return newReceiveBuilder()
                .onMessage(Initialize.class, this::onInitialize)
                .onMessage(ClientWrite.class, this::onClientWrite)
                .onMessage(GetStatus.class, this::onGetStatus)
                .onMessage(RequestVote.class, this::onRequestVote)
                .onMessage(VoteResponse.class, this::onVoteResponse)
                .onMessage(Heartbeat.class, this::onHeartbeat)
                .onMessage(AppendEntry.class, this::onAppendEntry)
                .onMessage(AppendAck.class, this::onAppendAck)
                .onMessage(CommitEntry.class, this::onCommitEntry)
                .onMessage(ElectionTimeout.class, ignored -> onElectionTimeout())
                .onMessage(HeartbeatTick.class, ignored -> onHeartbeatTick())
                .build();
    }

    private Behavior<Command> onInitialize(Initialize m) {
        cluster = List.copyOf(m.cluster()); network = m.network(); observer = m.observer();
        resetElectionTimer(); observer.tell(new NodeReady(region));
        System.out.printf("[NODE] region=%s ready electionTimeoutMs=%d%n", region, electionTimeoutMs);
        return this;
    }

    private Behavior<Command> onElectionTimeout() {
        if (role == Role.LEADER) return this;
        currentTerm++; role = Role.CANDIDATE; votedFor = region;
        leaderRegion = null; leaderRef = null; votes.clear(); votes.add(region);
        resetElectionTimer();
        for (Peer peer : otherPeers())
            send(peer, new RequestVote(currentTerm, region, getContext().getSelf()));
        return this;
    }

    private Behavior<Command> onRequestVote(RequestVote m) {
        if (m.term() > currentTerm) stepDown(m.term());
        boolean grant = m.term() == currentTerm
                && (votedFor == null || votedFor.equals(m.candidateRegion()));
        if (grant) {
            role = Role.FOLLOWER; votedFor = m.candidateRegion();
            leaderRegion = null; leaderRef = null; resetElectionTimer();
        }
        send(new Peer(m.candidateRegion(), m.candidateRef()),
                new VoteResponse(currentTerm, region, grant));
        return this;
    }

    private Behavior<Command> onVoteResponse(VoteResponse m) {
        if (m.term() > currentTerm) { stepDown(m.term()); return this; }
        if (role != Role.CANDIDATE || m.term() != currentTerm) return this;
        if (m.granted() && votes.add(m.voterRegion()) && votes.size() >= majority()) becomeLeader();
        return this;
    }

    private void becomeLeader() {
        role = Role.LEADER; leaderRegion = region; leaderRef = getContext().getSelf();
        timers.cancel(ELECTION_TIMER); votes.clear();
        observer.tell(new LeaderElected(region, currentTerm, System.nanoTime()));
        System.out.printf("[ELECTION] leader=%s term=%d%n", region, currentTerm);
        sendHeartbeats();
        timers.startTimerWithFixedDelay(HEARTBEAT_TIMER, HeartbeatTick.INSTANCE,
                Duration.ofMillis(HEARTBEAT_MS));
    }

    private Behavior<Command> onHeartbeatTick() {
        if (role == Role.LEADER) sendHeartbeats();
        return this;
    }

    private void sendHeartbeats() {
        for (Peer peer : otherPeers())
            send(peer, new Heartbeat(currentTerm, region, getContext().getSelf()));
    }

    private Behavior<Command> onHeartbeat(Heartbeat m) {
        if (m.term() < currentTerm) return this;
        if (m.term() > currentTerm || role != Role.FOLLOWER) stepDown(m.term());
        leaderRegion = m.leaderRegion(); leaderRef = m.leaderRef(); resetElectionTimer();
        return this;
    }

    private Behavior<Command> onClientWrite(ClientWrite m) {
        if (role != Role.LEADER) {
            m.replyTo().tell(new WriteResult(m.requestId(), m.phaseToken(), false,
                    region, "NOT_LEADER"));
            return this;
        }
        long index = ++lastIndex;
        replicaValue += m.delta();
        replicaLog.put(index, new LogEntry(currentTerm, replicaValue));
        pendingWrites.put(index, new PendingWrite(replicaValue, m.requestId(),
                m.phaseToken(), m.replyTo(), region));
        for (Peer peer : otherPeers())
            send(peer, new AppendEntry(currentTerm, region, getContext().getSelf(),
                    index, replicaValue));
        return this;
    }

    private Behavior<Command> onAppendEntry(AppendEntry m) {
        if (m.term() < currentTerm) {
            send(new Peer(m.leaderRegion(), m.leaderRef()),
                    new AppendAck(currentTerm, m.index(), region, false));
            return this;
        }
        if (m.term() > currentTerm || role != Role.FOLLOWER) stepDown(m.term());
        leaderRegion = m.leaderRegion(); leaderRef = m.leaderRef(); resetElectionTimer();
        replicaLog.put(m.index(), new LogEntry(m.term(), m.value()));
        if (m.index() >= lastIndex) { lastIndex = m.index(); replicaValue = m.value(); }
        send(new Peer(m.leaderRegion(), m.leaderRef()),
                new AppendAck(currentTerm, m.index(), region, true));
        return this;
    }

    private Behavior<Command> onAppendAck(AppendAck m) {
        if (m.term() > currentTerm) { stepDown(m.term()); return this; }
        if (role != Role.LEADER || m.term() != currentTerm || !m.accepted()) return this;
        PendingWrite pending = pendingWrites.get(m.index());
        if (pending != null) { pending.acknowledgements.add(m.followerRegion()); advanceCommitIndex(); }
        return this;
    }

    private void advanceCommitIndex() {
        while (true) {
            long next = commitIndex + 1;
            PendingWrite pending = pendingWrites.get(next);
            if (pending == null || pending.acknowledgements.size() < majority()) return;
            commitIndex = next; committedValue = pending.value; pendingWrites.remove(next);
            for (Peer peer : otherPeers())
                send(peer, new CommitEntry(currentTerm, region, getContext().getSelf(),
                        commitIndex, committedValue));
            pending.replyTo.tell(new WriteResult(pending.requestId, pending.phaseToken,
                    true, region, "MAJORITY_COMMITTED"));
        }
    }

    private Behavior<Command> onCommitEntry(CommitEntry m) {
        if (m.term() < currentTerm) return this;
        if (m.term() > currentTerm || role != Role.FOLLOWER) stepDown(m.term());
        leaderRegion = m.leaderRegion(); leaderRef = m.leaderRef(); resetElectionTimer();
        if (m.index() > commitIndex) {
            commitIndex = m.index(); committedValue = m.value();
            if (m.index() >= lastIndex) { lastIndex = m.index(); replicaValue = m.value(); }
        }
        return this;
    }

    private Behavior<Command> onGetStatus(GetStatus m) {
        m.replyTo().tell(new Status(region, role.name(), currentTerm, commitIndex,
                committedValue, lastIndex, replicaValue, pendingWrites.size()));
        return this;
    }

    private void stepDown(int newTerm) {
        if (newTerm > currentTerm) { currentTerm = newTerm; votedFor = null; }
        role = Role.FOLLOWER; leaderRegion = null; leaderRef = null; votes.clear();
        timers.cancel(HEARTBEAT_TIMER); resetElectionTimer();
    }

    private void resetElectionTimer() {
        timers.startSingleTimer(ELECTION_TIMER, ElectionTimeout.INSTANCE,
                Duration.ofMillis(electionTimeoutMs));
    }

    private int majority() { return cluster.size() / 2 + 1; }
    private List<Peer> otherPeers() {
        return cluster.stream().filter(peer -> !peer.region().equals(region)).toList();
    }
    private void send(Peer peer, Command payload) {
        network.tell(new SimulatedNetwork.Send(region, peer.region(), peer.ref(), payload));
    }
}

How does each node own its state?

• The public command records form the typed boundary used by clients plus the benchmark actor.
• The private command records carry elections, heartbeats, replication acknowledgements, and commit notices between nodes.
• The actor stores its current term plus its role and vote as private fields.
• The actor keeps replicaValue separate from committedValue so later partition evidence can expose unresolved work.

• Confirm the Explorer view now lists RaftNode.java.

Two actor-owned boundaries are now in place. Regional state lives in RaftNode while delayed delivery lives in SimulatedNetwork.

Add the benchmark and run the actors

The benchmark actor is the control plane for the demonstration. It receives readiness plus election events directly while node-to-node traffic still crosses the simulated network.

• Select the New File... button in the Explorer view.
• Enter Benchmark.java.
• Press Enter.
• Select the full-code tab below.
• Copy the complete benchmark actor into Benchmark.java.
• Save the file with Cmd+S.

✔️ Awesome, I've got everything!

Your benchmark actor should configure the three-node cluster. It should also observe node readiness plus leader-election events.

ⓧ I'd like to double check the full code

import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;

import org.apache.pekko.actor.typed.ActorRef;
import org.apache.pekko.actor.typed.Behavior;
import org.apache.pekko.actor.typed.javadsl.AbstractBehavior;
import org.apache.pekko.actor.typed.javadsl.ActorContext;
import org.apache.pekko.actor.typed.javadsl.Behaviors;
import org.apache.pekko.actor.typed.javadsl.Receive;

public final class Benchmark extends AbstractBehavior<Benchmark.Command> {
    public interface Command {}
    public record Configure(List<RaftNode.Peer> cluster) implements Command {}
    public record PartitionApplied(String region, boolean isolated) implements Command {}
    private enum StartBaseline implements Command { INSTANCE }
    private enum StartRecovery implements Command { INSTANCE }
    private enum ElectionProbe implements Command { INSTANCE }
    private record LoadTick(int token) implements Command {}
    private record EndPhase(int token) implements Command {}
    private enum Phase { BOOT, BASELINE, PARTITION, RECOVERY, REPORT }

    private static final int WINDOW = Math.max(1, Integer.getInteger("sim.window", 16));
    private static final Duration LOAD_DURATION = Duration.ofSeconds(2);
    private static final Duration LOAD_TICK = Duration.ofMillis(50);

    private static final class Stats {
        long sent;
        long committed;
        long rejected;
        int maxInFlight;
        long windowFullSamples;
        double durationSeconds;
        final List<Double> commitLatencyMs = new ArrayList<>();

        double averageLatencyMs() {
            return commitLatencyMs.stream().mapToDouble(Double::doubleValue).average().orElse(0.0);
        }

        double p95LatencyMs() {
            if (commitLatencyMs.isEmpty()) return 0.0;
            List<Double> sorted = new ArrayList<>(commitLatencyMs);
            sorted.sort(Double::compareTo);
            int index = Math.max(0, (int) Math.ceil(sorted.size() * 0.95) - 1);
            return sorted.get(index);
        }

        Stats copy() {
            Stats copy = new Stats();
            copy.sent = sent; copy.committed = committed; copy.rejected = rejected;
            copy.maxInFlight = maxInFlight; copy.windowFullSamples = windowFullSamples;
            copy.durationSeconds = durationSeconds;
            copy.commitLatencyMs.addAll(commitLatencyMs);
            return copy;
        }
    }

    private final ActorRef<SimulatedNetwork.Command> network;
    private final long bootStartedNanos = System.nanoTime();
    private List<RaftNode.Peer> cluster = List.of();
    private final Map<String, ActorRef<RaftNode.Command>> nodes = new HashMap<>();
    private final Map<String, RaftNode.Status> statuses = new HashMap<>();
    private final Map<Long, Long> requestStartedNanos = new HashMap<>();
    private Phase phase = Phase.BOOT;
    private int readyNodes;
    private boolean baselineScheduled;
    private boolean recoveryScheduled;
    private boolean loading;
    private String currentLeaderRegion;
    private ActorRef<RaftNode.Command> currentLeaderRef;
    private int currentLeaderTerm;
    private String oldLeaderRegion;
    private ActorRef<RaftNode.Command> oldLeaderRef;
    private int oldLeaderTerm;
    private String recoveredLeaderRegion;
    private ActorRef<RaftNode.Command> recoveredLeaderRef;
    private int recoveredLeaderTerm;
    private long nextRequestId = 1;
    private int phaseToken;
    private int isolatedToken = -1;
    private int probeToken = -1;
    private int inFlight;
    private long phaseStartedNanos;
    private long electionStartedNanos;
    private double initialElectionMs;
    private double failoverElectionMs;
    private Stats currentStats;
    private Stats baselineStats;
    private Stats recoveryStats;
    private long isolatedSent;
    private long isolatedAcknowledged;
    private long isolatedRejected;
    private long failoverProbes;
    private long failoverRejected;
    private long failoverProbeCommitted;

    public static Behavior<Command> create(ActorRef<SimulatedNetwork.Command> network) {
        return Behaviors.setup(context -> new Benchmark(context, network));
    }

    private Benchmark(ActorContext<Command> context,
                      ActorRef<SimulatedNetwork.Command> network) {
        super(context); this.network = network;
    }

    @Override
    public Receive<Command> createReceive() {
        return newReceiveBuilder()
                .onMessage(Configure.class, this::onConfigure)
                .onMessage(PartitionApplied.class, this::onPartitionApplied)
                .onMessage(RaftNode.NodeReady.class, this::onNodeReady)
                .onMessage(RaftNode.LeaderElected.class, this::onLeaderElected)
                .onMessage(RaftNode.WriteResult.class, this::onWriteResult)
                .onMessage(RaftNode.Status.class, this::onStatus)
                .onMessage(StartBaseline.class, ignored -> onStartBaseline())
                .onMessage(StartRecovery.class, ignored -> onStartRecovery())
                .onMessage(ElectionProbe.class, ignored -> onElectionProbe())
                .onMessage(LoadTick.class, this::onLoadTick)
                .onMessage(EndPhase.class, this::onEndPhase)
                .build();
    }

    private Behavior<Command> onConfigure(Configure m) {
        cluster = List.copyOf(m.cluster());
        for (RaftNode.Peer peer : cluster) nodes.put(peer.region(), peer.ref());
        System.out.printf("[BENCH] requestWindow=%d%n", WINDOW);
        return this;
    }

    private Behavior<Command> onNodeReady(RaftNode.NodeReady m) {
        if (++readyNodes == 3) System.out.println("[BENCH] all regional nodes are ready");
        return this;
    }

    private Behavior<Command> onLeaderElected(RaftNode.LeaderElected m) {
        if (oldLeaderRegion == null) {
            currentLeaderRegion = m.region(); currentLeaderRef = nodes.get(m.region());
            currentLeaderTerm = m.term();
            initialElectionMs = nanosToMillis(m.observedAtNanos() - bootStartedNanos);
            if (!baselineScheduled) {
                baselineScheduled = true;
                getContext().scheduleOnce(Duration.ofMillis(300), getContext().getSelf(),
                        StartBaseline.INSTANCE);
            }
        } else if (m.term() > oldLeaderTerm && !m.region().equals(oldLeaderRegion)
                && recoveredLeaderRef == null) {
            recoveredLeaderRegion = m.region(); recoveredLeaderRef = nodes.get(m.region());
            recoveredLeaderTerm = m.term();
            failoverElectionMs = nanosToMillis(m.observedAtNanos() - electionStartedNanos);
            System.out.printf(Locale.ROOT,
                    "[FAILOVER] newLeader=%s term=%d electionMs=%.1f%n",
                    recoveredLeaderRegion, recoveredLeaderTerm, failoverElectionMs);
            if (!recoveryScheduled) {
                recoveryScheduled = true;
                getContext().scheduleOnce(Duration.ofMillis(200), getContext().getSelf(),
                        StartRecovery.INSTANCE);
            }
        }
        return this;
    }

    private Behavior<Command> onStartBaseline() {
        if (currentLeaderRef == null || cluster.size() != 3) {
            getContext().scheduleOnce(Duration.ofMillis(100), getContext().getSelf(),
                    StartBaseline.INSTANCE);
            return this;
        }
        beginLoadPhase(Phase.BASELINE, currentLeaderRef);
        for (RaftNode.Peer peer : cluster)
            if (!peer.region().equals(currentLeaderRegion)) sendTracked(peer.ref());
        fillWindow();
        return this;
    }

    private void beginLoadPhase(Phase newPhase, ActorRef<RaftNode.Command> leaderRef) {
        phase = newPhase; currentLeaderRef = leaderRef; loading = true;
        phaseToken++; inFlight = 0; currentStats = new Stats();
        phaseStartedNanos = System.nanoTime(); scheduleLoadTick();
        getContext().scheduleOnce(LOAD_DURATION, getContext().getSelf(),
                new EndPhase(phaseToken));
    }

    private Behavior<Command> onLoadTick(LoadTick m) {
        if (!loading || m.token() != phaseToken) return this;
        if (inFlight >= WINDOW) currentStats.windowFullSamples++;
        fillWindow(); scheduleLoadTick(); return this;
    }

    private void scheduleLoadTick() {
        getContext().scheduleOnce(LOAD_TICK, getContext().getSelf(), new LoadTick(phaseToken));
    }

    private Behavior<Command> onEndPhase(EndPhase m) {
        if (!loading || m.token() != phaseToken) return this;
        loading = false;
        if (inFlight == 0) finishLoadPhase();
        return this;
    }

    private Behavior<Command> onWriteResult(RaftNode.WriteResult m) {
        if ((phase == Phase.BASELINE || phase == Phase.RECOVERY)
                && m.phaseToken() == phaseToken) {
            inFlight = Math.max(0, inFlight - 1);
            Long started = requestStartedNanos.remove(m.requestId());
            if (m.committed()) {
                currentStats.committed++;
                if (started != null)
                    currentStats.commitLatencyMs.add(nanosToMillis(System.nanoTime() - started));
            } else currentStats.rejected++;
            if (loading) fillWindow();
            else if (inFlight == 0) finishLoadPhase();
        } else if (m.phaseToken() == isolatedToken) {
            if (m.committed()) isolatedAcknowledged++; else isolatedRejected++;
        } else if (m.phaseToken() == probeToken) {
            if (m.committed()) failoverProbeCommitted++; else failoverRejected++;
        }
        return this;
    }

    private void fillWindow() {
        while (loading && inFlight < WINDOW) sendTracked(currentLeaderRef);
    }

    private void sendTracked(ActorRef<RaftNode.Command> target) {
        long requestId = nextRequestId++;
        requestStartedNanos.put(requestId, System.nanoTime());
        target.tell(new RaftNode.ClientWrite(requestId, 1, getContext().getSelf(), phaseToken));
        inFlight++; currentStats.sent++;
        currentStats.maxInFlight = Math.max(currentStats.maxInFlight, inFlight);
    }

    private void finishLoadPhase() {
        currentStats.durationSeconds =
                (System.nanoTime() - phaseStartedNanos) / 1_000_000_000.0;
        if (phase == Phase.BASELINE) {
            baselineStats = currentStats.copy(); printStats("before-failover", baselineStats);
            beginPartition();
        } else if (phase == Phase.RECOVERY) {
            recoveryStats = currentStats.copy(); printStats("after-failover", recoveryStats);
            requestStatuses();
        }
    }

    private void beginPartition() {
        phase = Phase.PARTITION; oldLeaderRegion = currentLeaderRegion;
        oldLeaderRef = currentLeaderRef; oldLeaderTerm = currentLeaderTerm;
        currentLeaderRegion = null; currentLeaderRef = null;
        network.tell(new SimulatedNetwork.SetPartition(oldLeaderRegion, true,
                getContext().getSelf()));
    }

    private Behavior<Command> onPartitionApplied(PartitionApplied m) {
        if (phase != Phase.PARTITION || !m.isolated()
                || !m.region().equals(oldLeaderRegion)) return this;
        electionStartedNanos = System.nanoTime();
        isolatedToken = ++phaseToken; probeToken = ++phaseToken;
        for (int i = 0; i < WINDOW; i++) {
            oldLeaderRef.tell(new RaftNode.ClientWrite(nextRequestId++, 1,
                    getContext().getSelf(), isolatedToken));
            isolatedSent++;
        }
        getContext().scheduleOnce(Duration.ofMillis(100), getContext().getSelf(),
                ElectionProbe.INSTANCE);
        return this;
    }

    private Behavior<Command> onElectionProbe() {
        if (phase != Phase.PARTITION || recoveredLeaderRef != null) return this;
        RaftNode.Peer target = cluster.stream()
                .filter(peer -> !peer.region().equals(oldLeaderRegion)).findFirst().orElseThrow();
        target.ref().tell(new RaftNode.ClientWrite(nextRequestId++, 1,
                getContext().getSelf(), probeToken));
        failoverProbes++;
        getContext().scheduleOnce(Duration.ofMillis(100), getContext().getSelf(),
                ElectionProbe.INSTANCE);
        return this;
    }

    private Behavior<Command> onStartRecovery() {
        if (phase != Phase.PARTITION || recoveredLeaderRef == null) return this;
        System.out.printf(
                "[PARTITION] formerLeader=%s acceptedLocally=%d acknowledged=%d rejected=%d unacknowledged=%d%n",
                oldLeaderRegion, isolatedSent, isolatedAcknowledged, isolatedRejected,
                isolatedOutstanding());
        currentLeaderRegion = recoveredLeaderRegion; currentLeaderRef = recoveredLeaderRef;
        currentLeaderTerm = recoveredLeaderTerm;
        beginLoadPhase(Phase.RECOVERY, recoveredLeaderRef); fillWindow();
        return this;
    }

    private long isolatedOutstanding() {
        return isolatedSent - isolatedAcknowledged - isolatedRejected;
    }

    private void requestStatuses() {
        phase = Phase.REPORT; statuses.clear();
        for (RaftNode.Peer peer : cluster)
            peer.ref().tell(new RaftNode.GetStatus(getContext().getSelf()));
    }

    private Behavior<Command> onStatus(RaftNode.Status m) {
        statuses.put(m.region(), m);
        if (statuses.size() == cluster.size()) {
            printFinalReport(); getContext().getSystem().terminate();
        }
        return this;
    }

    private void printStats(String label, Stats s) {
        System.out.printf(Locale.ROOT,
                "[METRIC] phase=%s acknowledgedCommitted=%d rejected=%d throughputOpsPerSec=%.1f avgCommitLatencyMs=%.1f p95CommitLatencyMs=%.1f maxInFlight=%d fullWindowSamples=%d durationSec=%.2f%n",
                label, s.committed, s.rejected, s.committed / s.durationSeconds,
                s.averageLatencyMs(), s.p95LatencyMs(), s.maxInFlight,
                s.windowFullSamples, s.durationSeconds);
    }

    private void printFinalReport() {
        long totalRejected = baselineStats.rejected + recoveryStats.rejected + failoverRejected;
        System.out.println("\n========== REPRODUCIBLE BENCHMARK REPORT ==========");
        System.out.printf(Locale.ROOT, "requestWindow=%d%n", WINDOW);
        System.out.printf(Locale.ROOT, "initialElectionMs=%.1f%n", initialElectionMs);
        System.out.printf(Locale.ROOT, "failoverElectionMs=%.1f%n", failoverElectionMs);
        System.out.printf(Locale.ROOT,
                "before=acknowledgedCommitted:%d,throughput:%.1f,avgLatencyMs:%.1f,p95LatencyMs:%.1f%n",
                baselineStats.committed, baselineStats.committed / baselineStats.durationSeconds,
                baselineStats.averageLatencyMs(), baselineStats.p95LatencyMs());
        System.out.printf(Locale.ROOT,
                "after=acknowledgedCommitted:%d,throughput:%.1f,avgLatencyMs:%.1f,p95LatencyMs:%.1f%n",
                recoveryStats.committed, recoveryStats.committed / recoveryStats.durationSeconds,
                recoveryStats.averageLatencyMs(), recoveryStats.p95LatencyMs());
        System.out.printf("explicitlyRejected=%d%n", totalRejected);
        System.out.printf("unacknowledgedUncommitted=%d%n", isolatedOutstanding());
        System.out.println("acknowledgedButUncommitted=0");
        System.out.printf("outstandingBefore=maxInFlight:%d,fullWindowSamples:%d%n",
                baselineStats.maxInFlight, baselineStats.windowFullSamples);
        System.out.printf("outstandingAfter=maxInFlight:%d,fullWindowSamples:%d%n",
                recoveryStats.maxInFlight, recoveryStats.windowFullSamples);
        System.out.printf("failoverProbes=%d failoverProbeCommitted=%d%n",
                failoverProbes, failoverProbeCommitted);
        System.out.println("\nregion      role       term  commitIndex  committed  replicaIndex  replica  pending");
        statuses.values().stream().sorted((a, b) -> a.region().compareTo(b.region()))
                .forEach(s -> System.out.printf(Locale.ROOT,
                        "%-11s %-10s %-5d %-12d %-10d %-13d %-8d %d%n",
                        s.region(), s.role(), s.term(), s.commitIndex(), s.committedValue(),
                        s.replicaIndex(), s.replicaValue(), s.pendingWrites()));
        System.out.println("\nDemonstrated: terms, one vote per term, majority election, heartbeats,");
        System.out.println("acknowledgement only after majority commitment, minority blocking, and bounded work.");
        System.out.println("Not claimed: full Log Matching, Leader Completeness, arbitrary-history State Machine Safety,");
        System.out.println("stable storage, retries, snapshots, membership changes, deduplication, linearizable reads,");
        System.out.println("Byzantine tolerance, or real multi-process networking.");
        System.out.println("===================================================");
    }

    private static double nanosToMillis(long nanos) { return nanos / 1_000_000.0; }
}

What does the benchmark control?

• The Configure command stores references to the three regional actors.
• The benchmark counts NodeReady events until all three nodes have initialized.
• The benchmark records each LeaderElected event with its region plus term.
• The phase fields prepare the control plane for the partition and measurement work in later steps.

• Confirm the Explorer view lists Benchmark.java.
• Confirm the Explorer view also lists the other three Java files.

Before you run the simulation, which region do you expect to time out first and become the initial leader?

• Run the four-file simulation from the existing macOS Terminal by running this command:

jbang RegionalRaftSimulation.java

What happens on the first run?

JBang reads the source directives before compiling the four Java files together. It also resolves the pinned Apache Pekko dependency plus its transitive dependencies.

The first run may pause during dependency resolution. The terminal is still working during that pause.

You will see one ready line for us-east, eu-west, and ap-south. You will then see [ELECTION] leader=us-east term=1 after the shortest election timeout wins.

Later in the same run, [NETWORK] region=us-east isolated=true confirms that the network applied the requested partition before the benchmark continued.

Compilation or startup failed?

Check that all four Java files sit directly inside regional-raft-simulator. A source file in another folder cannot satisfy the entry point's source directive.

Compare every filename with the //SOURCES line. Java filenames are case-sensitive.

Still stuck? Help me diagnose my four-file JBang and Apache Pekko compilation failure.

That is the concurrency boundary working: three independent regional actors are ready inside one ActorSystem. Their peer traffic now passes through a network actor that can delay or drop messages.

Your three-region actor system can now start, exchange delayed messages, and elect an initial leader. Next, you will make successful write acknowledgement depend on majority commitment.

Commit Writes Through a Majority

Your three regional actors now own separate state inside one Apache Pekko Typed system. They can exchange delayed messages through the simulated network.

A write stored by one actor is only a local proposal. This step makes successful acknowledgement depend on a two-node quorum. You will then isolate the leader to expose the gap between replica state and committed state.

In this step, get ready to:

• Store client writes as pending replica entries.
• Advance committed state only after majority replication.
• Run a bounded partition batch against the isolated former leader.

Track local writes as pending

Each accepted write needs a client identity plus the value proposed for replication. A PendingWrite keeps that request unresolved until another region acknowledges the same entry.

• In RaftNode.java, locate the PendingWrite class.
• Replace that class with this implementation:

private static final class PendingWrite {
    final int value;
    final long requestId;
    final int phaseToken;
    final ActorRef<Benchmark.Command> replyTo;
    final Set<String> acknowledgements = new HashSet<>();

    PendingWrite(int value, long requestId, int phaseToken,
                 ActorRef<Benchmark.Command> replyTo, String leaderRegion) {
        this.value = value;
        this.requestId = requestId;
        this.phaseToken = phaseToken;
        this.replyTo = replyTo;
        acknowledgements.add(leaderRegion);
    }
}

What does this state track?

• The proposed value stays attached to the client request that produced it.
• The phase token lets the benchmark classify the eventual result.
• The acknowledgement set starts with the leader's region because the leader already stores the entry locally.

The leader can now preserve local acceptance without claiming that the write is committed. The client request remains inside pendingWrites while replication is in progress.

• Locate the onClientWrite method in RaftNode.java.
• Replace the method with this version:

private Behavior<Command> onClientWrite(ClientWrite m) {
    if (role != Role.LEADER) {
        m.replyTo().tell(new WriteResult(m.requestId(), m.phaseToken(), false,
                region, "NOT_LEADER"));
        return this;
    }
    long index = ++lastIndex;
    replicaValue += m.delta();
    replicaLog.put(index, new LogEntry(currentTerm, replicaValue));
    pendingWrites.put(index, new PendingWrite(replicaValue, m.requestId(),
            m.phaseToken(), m.replyTo(), region));
    for (Peer peer : otherPeers())
        send(peer, new AppendEntry(currentTerm, region, getContext().getSelf(),
                index, replicaValue));
    return this;
}

How does local acceptance work?

• A non-leader returns NOT_LEADER without creating a local entry.
• The leader increments lastIndex before storing the proposed value in replicaLog.
• The new PendingWrite receives no successful client reply at this point.
• An AppendEntry message asks every other region to replicate the proposal.

• Save RaftNode.java.
• Confirm the updated actor still compiles by running this command in the Terminal panel from earlier:

jbang RegionalRaftSimulation.java

What should I see?

You should see ready messages for us-east, eu-west, and ap-south. You should also see an initial leader election.

Seeing a Java compilation error?

Check that the PendingWrite constructor includes the leader region as its final parameter.

Check that onClientWrite passes region into that constructor.

Still stuck? Help me compare my PendingWrite and onClientWrite implementations.

Advance commitment through a quorum

A follower first copies the proposed entry into its own replica state. It then sends an acknowledgement back through the simulated network.

• Locate the onAppendEntry method in RaftNode.java.
• Replace the method with this implementation:

private Behavior<Command> onAppendEntry(AppendEntry m) {
    if (m.term() < currentTerm) {
        send(new Peer(m.leaderRegion(), m.leaderRef()),
                new AppendAck(currentTerm, m.index(), region, false));
        return this;
    }
    if (m.term() > currentTerm || role != Role.FOLLOWER) stepDown(m.term());
    leaderRegion = m.leaderRegion(); leaderRef = m.leaderRef(); resetElectionTimer();
    replicaLog.put(m.index(), new LogEntry(m.term(), m.value()));
    if (m.index() >= lastIndex) { lastIndex = m.index(); replicaValue = m.value(); }
    send(new Peer(m.leaderRegion(), m.leaderRef()),
            new AppendAck(currentTerm, m.index(), region, true));
    return this;
}

What happens on the follower?

• A stale-term proposal receives a rejected acknowledgement.
• A current proposal updates the follower's replica log.
• The follower sends AppendAck only after storing the entry.

The leader counts acknowledgements by region. Duplicate messages from one follower cannot inflate the vote count because the regions live in a Set.

• Locate the onAppendAck method.
• Replace the method with this version:

private Behavior<Command> onAppendAck(AppendAck m) {
    if (m.term() > currentTerm) { stepDown(m.term()); return this; }
    if (role != Role.LEADER || m.term() != currentTerm || !m.accepted()) return this;
    PendingWrite pending = pendingWrites.get(m.index());
    if (pending != null) { pending.acknowledgements.add(m.followerRegion()); advanceCommitIndex(); }
    return this;
}

Why filter acknowledgements?

Only the current leader can use an accepted acknowledgement from its current term. A higher term forces the node to step down before it can commit more work.

• Save RaftNode.java.
• Check the follower replication path by running the simulator again:

jbang RegionalRaftSimulation.java

What does this run prove?

You should see the three ready messages followed by a leader election. A clean run proves that the leader and follower message types still align.

Replication becomes commitment when the next contiguous entry has acknowledgements from two distinct regions. That condition keeps commitIndex from skipping unresolved entries.

• Locate the advanceCommitIndex method.
• Replace the method with this implementation:

private void advanceCommitIndex() {
    while (true) {
        long next = commitIndex + 1;
        PendingWrite pending = pendingWrites.get(next);
        if (pending == null || pending.acknowledgements.size() < majority()) return;
        commitIndex = next; committedValue = pending.value; pendingWrites.remove(next);
        for (Peer peer : otherPeers())
            send(peer, new CommitEntry(currentTerm, region, getContext().getSelf(),
                    commitIndex, committedValue));
        pending.replyTo.tell(new WriteResult(pending.requestId, pending.phaseToken,
                true, region, "MAJORITY_COMMITTED"));
    }
}

Where does commitment happen?

• The method inspects the entry immediately after the current commitIndex.
• A two-node majority advances both commitIndex and committedValue.
• The leader removes the resolved request from pendingWrites.
• The client receives MAJORITY_COMMITTED only after the state transition succeeds.

• Save RaftNode.java.
• Verify the majority-commit path compiles by running the simulator:

jbang RegionalRaftSimulation.java

What should I see now?

The simulator should start all three nodes without a compilation error. Baseline writes can now receive successful results only through MAJORITY_COMMITTED.

Writes never become committed?

Check that PendingWrite adds the leader region when it is created.

Check that onAppendAck adds the follower region before calling advanceCommitIndex().

Need another pair of eyes? Help me trace why my acknowledgements never reach a majority.

Expose uncommitted work during a partition

A load generator can overwhelm the simulation if it sends requests without a limit. Bounded in-flight work caps unresolved requests at sim.window. The default window is 16.

What changes in the benchmark?

• The WINDOW constant reads sim.window and enforces a minimum of one request.
• Each tracked write stores its start time before the benchmark sends it.
• The load loop refills available window slots as results return.
• The partition phase sends one full window directly to the isolated former leader.
• Separate counters preserve acknowledged results, explicit rejections, and unresolved writes.

• Switch back to Benchmark.java from earlier.
• Update the benchmark constants so WINDOW defaults to 16.
• Update the tracking fields to preserve request start times.
• Update beginLoadPhase to reset bounded phase state.
• Update fillWindow to stop sending when inFlight reaches WINDOW.
• Update sendTracked to timestamp each request before delivery.
• Update onWriteResult to classify committed, rejected, and unresolved outcomes.

Where are you in the flow?

The benchmark can now drive a bounded baseline without mixing completed work into the next phase. The remaining changes create the controlled partition batch.

• Update beginPartition to preserve the current leader as the former leader.
• Update onPartitionApplied to send exactly one bounded window to that isolated node.
• Update onStartRecovery to print the partition outcome before recovery traffic starts.
• Update isolatedOutstanding to subtract acknowledged and rejected requests from the isolated batch.
• Save Benchmark.java.

Use the full-file references

Compare both edited files with the reference tabs below. Focus on the named classes and methods so the surrounding election code from the previous step stays intact.

✔️ Awesome, I've got everything!

Your RaftNode.java file now separates replica acceptance from majority commitment.

ⓧ I'd like to double check RaftNode.java

import java.time.Duration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.pekko.actor.typed.ActorRef;
import org.apache.pekko.actor.typed.Behavior;
import org.apache.pekko.actor.typed.javadsl.AbstractBehavior;
import org.apache.pekko.actor.typed.javadsl.ActorContext;
import org.apache.pekko.actor.typed.javadsl.Behaviors;
import org.apache.pekko.actor.typed.javadsl.Receive;
import org.apache.pekko.actor.typed.javadsl.TimerScheduler;

public final class RaftNode extends AbstractBehavior<RaftNode.Command> {
    public interface Command {}
    public record Peer(String region, ActorRef<Command> ref) {}
    public record Initialize(List<Peer> cluster, ActorRef<SimulatedNetwork.Command> network,
                             ActorRef<Benchmark.Command> observer) implements Command {}
    public record ClientWrite(long requestId, int delta, ActorRef<Benchmark.Command> replyTo,
                              int phaseToken) implements Command {}
    public record GetStatus(ActorRef<Benchmark.Command> replyTo) implements Command {}
    public record NodeReady(String region) implements Benchmark.Command {}
    public record LeaderElected(String region, int term,
                                long observedAtNanos) implements Benchmark.Command {}
    public record WriteResult(long requestId, int phaseToken, boolean committed,
                              String node, String reason) implements Benchmark.Command {}
    public record Status(String region, String role, int term, long commitIndex,
                         int committedValue, long replicaIndex, int replicaValue,
                         int pendingWrites) implements Benchmark.Command {}

    private record RequestVote(int term, String candidateRegion,
                               ActorRef<Command> candidateRef) implements Command {}
    private record VoteResponse(int term, String voterRegion,
                                boolean granted) implements Command {}
    private record Heartbeat(int term, String leaderRegion,
                             ActorRef<Command> leaderRef) implements Command {}
    private record AppendEntry(int term, String leaderRegion, ActorRef<Command> leaderRef,
                               long index, int value) implements Command {}
    private record AppendAck(int term, long index, String followerRegion,
                             boolean accepted) implements Command {}
    private record CommitEntry(int term, String leaderRegion, ActorRef<Command> leaderRef,
                               long index, int value) implements Command {}
    private enum ElectionTimeout implements Command { INSTANCE }
    private enum HeartbeatTick implements Command { INSTANCE }
    private enum Role { FOLLOWER, CANDIDATE, LEADER }
    private record LogEntry(int term, int value) {}

    private static final class PendingWrite {
        final int value;
        final long requestId;
        final int phaseToken;
        final ActorRef<Benchmark.Command> replyTo;
        final Set<String> acknowledgements = new HashSet<>();

        PendingWrite(int value, long requestId, int phaseToken,
                     ActorRef<Benchmark.Command> replyTo, String leaderRegion) {
            this.value = value;
            this.requestId = requestId;
            this.phaseToken = phaseToken;
            this.replyTo = replyTo;
            acknowledgements.add(leaderRegion);
        }
    }

    private static final String ELECTION_TIMER = "election-timer";
    private static final String HEARTBEAT_TIMER = "heartbeat-timer";
    private static final long HEARTBEAT_MS = 150L;
    private final String region;
    private final long electionTimeoutMs;
    private final TimerScheduler<Command> timers;
    private List<Peer> cluster = List.of();
    private ActorRef<SimulatedNetwork.Command> network;
    private ActorRef<Benchmark.Command> observer;
    private Role role = Role.FOLLOWER;
    private int currentTerm;
    private String votedFor;
    private String leaderRegion;
    private ActorRef<Command> leaderRef;
    private final Set<String> votes = new HashSet<>();
    private long lastIndex;
    private long commitIndex;
    private int replicaValue;
    private int committedValue;
    private final Map<Long, LogEntry> replicaLog = new HashMap<>();
    private final Map<Long, PendingWrite> pendingWrites = new HashMap<>();

    public static Behavior<Command> create(String region, long electionTimeoutMs) {
        return Behaviors.setup(context -> Behaviors.withTimers(timers ->
                new RaftNode(context, timers, region, electionTimeoutMs)));
    }

    private RaftNode(ActorContext<Command> context, TimerScheduler<Command> timers,
                     String region, long electionTimeoutMs) {
        super(context);
        this.timers = timers;
        this.region = region;
        this.electionTimeoutMs = electionTimeoutMs;
    }

    @Override
    public Receive<Command> createReceive() {
        return newReceiveBuilder()
                .onMessage(Initialize.class, this::onInitialize)
                .onMessage(ClientWrite.class, this::onClientWrite)
                .onMessage(GetStatus.class, this::onGetStatus)
                .onMessage(RequestVote.class, this::onRequestVote)
                .onMessage(VoteResponse.class, this::onVoteResponse)
                .onMessage(Heartbeat.class, this::onHeartbeat)
                .onMessage(AppendEntry.class, this::onAppendEntry)
                .onMessage(AppendAck.class, this::onAppendAck)
                .onMessage(CommitEntry.class, this::onCommitEntry)
                .onMessage(ElectionTimeout.class, ignored -> onElectionTimeout())
                .onMessage(HeartbeatTick.class, ignored -> onHeartbeatTick())
                .build();
    }

    private Behavior<Command> onInitialize(Initialize m) {
        cluster = List.copyOf(m.cluster()); network = m.network(); observer = m.observer();
        resetElectionTimer(); observer.tell(new NodeReady(region));
        System.out.printf("[NODE] region=%s ready electionTimeoutMs=%d%n", region, electionTimeoutMs);
        return this;
    }

    private Behavior<Command> onElectionTimeout() {
        if (role == Role.LEADER) return this;
        currentTerm++; role = Role.CANDIDATE; votedFor = region;
        leaderRegion = null; leaderRef = null; votes.clear(); votes.add(region);
        resetElectionTimer();
        for (Peer peer : otherPeers())
            send(peer, new RequestVote(currentTerm, region, getContext().getSelf()));
        return this;
    }

    private Behavior<Command> onRequestVote(RequestVote m) {
        if (m.term() > currentTerm) stepDown(m.term());
        boolean grant = m.term() == currentTerm
                && (votedFor == null || votedFor.equals(m.candidateRegion()));
        if (grant) {
            role = Role.FOLLOWER; votedFor = m.candidateRegion();
            leaderRegion = null; leaderRef = null; resetElectionTimer();
        }
        send(new Peer(m.candidateRegion(), m.candidateRef()),
                new VoteResponse(currentTerm, region, grant));
        return this;
    }

    private Behavior<Command> onVoteResponse(VoteResponse m) {
        if (m.term() > currentTerm) { stepDown(m.term()); return this; }
        if (role != Role.CANDIDATE || m.term() != currentTerm) return this;
        if (m.granted() && votes.add(m.voterRegion()) && votes.size() >= majority()) becomeLeader();
        return this;
    }

    private void becomeLeader() {
        role = Role.LEADER; leaderRegion = region; leaderRef = getContext().getSelf();
        timers.cancel(ELECTION_TIMER); votes.clear();
        observer.tell(new LeaderElected(region, currentTerm, System.nanoTime()));
        System.out.printf("[ELECTION] leader=%s term=%d%n", region, currentTerm);
        sendHeartbeats();
        timers.startTimerWithFixedDelay(HEARTBEAT_TIMER, HeartbeatTick.INSTANCE,
                Duration.ofMillis(HEARTBEAT_MS));
    }

    private Behavior<Command> onHeartbeatTick() {
        if (role == Role.LEADER) sendHeartbeats();
        return this;
    }

    private void sendHeartbeats() {
        for (Peer peer : otherPeers())
            send(peer, new Heartbeat(currentTerm, region, getContext().getSelf()));
    }

    private Behavior<Command> onHeartbeat(Heartbeat m) {
        if (m.term() < currentTerm) return this;
        if (m.term() > currentTerm || role != Role.FOLLOWER) stepDown(m.term());
        leaderRegion = m.leaderRegion(); leaderRef = m.leaderRef(); resetElectionTimer();
        return this;
    }

    private Behavior<Command> onClientWrite(ClientWrite m) {
        if (role != Role.LEADER) {
            m.replyTo().tell(new WriteResult(m.requestId(), m.phaseToken(), false,
                    region, "NOT_LEADER"));
            return this;
        }
        long index = ++lastIndex;
        replicaValue += m.delta();
        replicaLog.put(index, new LogEntry(currentTerm, replicaValue));
        pendingWrites.put(index, new PendingWrite(replicaValue, m.requestId(),
                m.phaseToken(), m.replyTo(), region));
        for (Peer peer : otherPeers())
            send(peer, new AppendEntry(currentTerm, region, getContext().getSelf(),
                    index, replicaValue));
        return this;
    }

    private Behavior<Command> onAppendEntry(AppendEntry m) {
        if (m.term() < currentTerm) {
            send(new Peer(m.leaderRegion(), m.leaderRef()),
                    new AppendAck(currentTerm, m.index(), region, false));
            return this;
        }
        if (m.term() > currentTerm || role != Role.FOLLOWER) stepDown(m.term());
        leaderRegion = m.leaderRegion(); leaderRef = m.leaderRef(); resetElectionTimer();
        replicaLog.put(m.index(), new LogEntry(m.term(), m.value()));
        if (m.index() >= lastIndex) { lastIndex = m.index(); replicaValue = m.value(); }
        send(new Peer(m.leaderRegion(), m.leaderRef()),
                new AppendAck(currentTerm, m.index(), region, true));
        return this;
    }

    private Behavior<Command> onAppendAck(AppendAck m) {
        if (m.term() > currentTerm) { stepDown(m.term()); return this; }
        if (role != Role.LEADER || m.term() != currentTerm || !m.accepted()) return this;
        PendingWrite pending = pendingWrites.get(m.index());
        if (pending != null) { pending.acknowledgements.add(m.followerRegion()); advanceCommitIndex(); }
        return this;
    }

    private void advanceCommitIndex() {
        while (true) {
            long next = commitIndex + 1;
            PendingWrite pending = pendingWrites.get(next);
            if (pending == null || pending.acknowledgements.size() < majority()) return;
            commitIndex = next; committedValue = pending.value; pendingWrites.remove(next);
            for (Peer peer : otherPeers())
                send(peer, new CommitEntry(currentTerm, region, getContext().getSelf(),
                        commitIndex, committedValue));
            pending.replyTo.tell(new WriteResult(pending.requestId, pending.phaseToken,
                    true, region, "MAJORITY_COMMITTED"));
        }
    }

    private Behavior<Command> onCommitEntry(CommitEntry m) {
        if (m.term() < currentTerm) return this;
        if (m.term() > currentTerm || role != Role.FOLLOWER) stepDown(m.term());
        leaderRegion = m.leaderRegion(); leaderRef = m.leaderRef(); resetElectionTimer();
        if (m.index() > commitIndex) {
            commitIndex = m.index(); committedValue = m.value();
            if (m.index() >= lastIndex) { lastIndex = m.index(); replicaValue = m.value(); }
        }
        return this;
    }

    private Behavior<Command> onGetStatus(GetStatus m) {
        m.replyTo().tell(new Status(region, role.name(), currentTerm, commitIndex,
                committedValue, lastIndex, replicaValue, pendingWrites.size()));
        return this;
    }

    private void stepDown(int newTerm) {
        if (newTerm > currentTerm) { currentTerm = newTerm; votedFor = null; }
        role = Role.FOLLOWER; leaderRegion = null; leaderRef = null; votes.clear();
        timers.cancel(HEARTBEAT_TIMER); resetElectionTimer();
    }

    private void resetElectionTimer() {
        timers.startSingleTimer(ELECTION_TIMER, ElectionTimeout.INSTANCE,
                Duration.ofMillis(electionTimeoutMs));
    }

    private int majority() { return cluster.size() / 2 + 1; }
    private List<Peer> otherPeers() {
        return cluster.stream().filter(peer -> !peer.region().equals(region)).toList();
    }
    private void send(Peer peer, Command payload) {
        network.tell(new SimulatedNetwork.Send(region, peer.region(), peer.ref(), payload));
    }
}

✔️ Awesome, I've got everything!

Your Benchmark.java file now bounds in-flight work and records the isolated batch separately.

ⓧ I'd like to double check Benchmark.java

import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;

import org.apache.pekko.actor.typed.ActorRef;
import org.apache.pekko.actor.typed.Behavior;
import org.apache.pekko.actor.typed.javadsl.AbstractBehavior;
import org.apache.pekko.actor.typed.javadsl.ActorContext;
import org.apache.pekko.actor.typed.javadsl.Behaviors;
import org.apache.pekko.actor.typed.javadsl.Receive;

public final class Benchmark extends AbstractBehavior<Benchmark.Command> {
    public interface Command {}
    public record Configure(List<RaftNode.Peer> cluster) implements Command {}
    public record PartitionApplied(String region, boolean isolated) implements Command {}
    private enum StartBaseline implements Command { INSTANCE }
    private enum StartRecovery implements Command { INSTANCE }
    private enum ElectionProbe implements Command { INSTANCE }
    private record LoadTick(int token) implements Command {}
    private record EndPhase(int token) implements Command {}
    private enum Phase { BOOT, BASELINE, PARTITION, RECOVERY, REPORT }

    private static final int WINDOW = Math.max(1, Integer.getInteger("sim.window", 16));
    private static final Duration LOAD_DURATION = Duration.ofSeconds(2);
    private static final Duration LOAD_TICK = Duration.ofMillis(50);

    private static final class Stats {
        long sent;
        long committed;
        long rejected;
        int maxInFlight;
        long windowFullSamples;
        double durationSeconds;
        final List<Double> commitLatencyMs = new ArrayList<>();

        double averageLatencyMs() {
            return commitLatencyMs.stream().mapToDouble(Double::doubleValue).average().orElse(0.0);
        }

        double p95LatencyMs() {
            if (commitLatencyMs.isEmpty()) return 0.0;
            List<Double> sorted = new ArrayList<>(commitLatencyMs);
            sorted.sort(Double::compareTo);
            int index = Math.max(0, (int) Math.ceil(sorted.size() * 0.95) - 1);
            return sorted.get(index);
        }

        Stats copy() {
            Stats copy = new Stats();
            copy.sent = sent; copy.committed = committed; copy.rejected = rejected;
            copy.maxInFlight = maxInFlight; copy.windowFullSamples = windowFullSamples;
            copy.durationSeconds = durationSeconds;
            copy.commitLatencyMs.addAll(commitLatencyMs);
            return copy;
        }
    }

    private final ActorRef<SimulatedNetwork.Command> network;
    private final long bootStartedNanos = System.nanoTime();
    private List<RaftNode.Peer> cluster = List.of();
    private final Map<String, ActorRef<RaftNode.Command>> nodes = new HashMap<>();
    private final Map<String, RaftNode.Status> statuses = new HashMap<>();
    private final Map<Long, Long> requestStartedNanos = new HashMap<>();
    private Phase phase = Phase.BOOT;
    private int readyNodes;
    private boolean baselineScheduled;
    private boolean recoveryScheduled;
    private boolean loading;
    private String currentLeaderRegion;
    private ActorRef<RaftNode.Command> currentLeaderRef;
    private int currentLeaderTerm;
    private String oldLeaderRegion;
    private ActorRef<RaftNode.Command> oldLeaderRef;
    private int oldLeaderTerm;
    private String recoveredLeaderRegion;
    private ActorRef<RaftNode.Command> recoveredLeaderRef;
    private int recoveredLeaderTerm;
    private long nextRequestId = 1;
    private int phaseToken;
    private int isolatedToken = -1;
    private int probeToken = -1;
    private int inFlight;
    private long phaseStartedNanos;
    private long electionStartedNanos;
    private double initialElectionMs;
    private double failoverElectionMs;
    private Stats currentStats;
    private Stats baselineStats;
    private Stats recoveryStats;
    private long isolatedSent;
    private long isolatedAcknowledged;
    private long isolatedRejected;
    private long failoverProbes;
    private long failoverRejected;
    private long failoverProbeCommitted;

    public static Behavior<Command> create(ActorRef<SimulatedNetwork.Command> network) {
        return Behaviors.setup(context -> new Benchmark(context, network));
    }

    private Benchmark(ActorContext<Command> context,
                      ActorRef<SimulatedNetwork.Command> network) {
        super(context); this.network = network;
    }

    @Override
    public Receive<Command> createReceive() {
        return newReceiveBuilder()
                .onMessage(Configure.class, this::onConfigure)
                .onMessage(PartitionApplied.class, this::onPartitionApplied)
                .onMessage(RaftNode.NodeReady.class, this::onNodeReady)
                .onMessage(RaftNode.LeaderElected.class, this::onLeaderElected)
                .onMessage(RaftNode.WriteResult.class, this::onWriteResult)
                .onMessage(RaftNode.Status.class, this::onStatus)
                .onMessage(StartBaseline.class, ignored -> onStartBaseline())
                .onMessage(StartRecovery.class, ignored -> onStartRecovery())
                .onMessage(ElectionProbe.class, ignored -> onElectionProbe())
                .onMessage(LoadTick.class, this::onLoadTick)
                .onMessage(EndPhase.class, this::onEndPhase)
                .build();
    }

    private Behavior<Command> onConfigure(Configure m) {
        cluster = List.copyOf(m.cluster());
        for (RaftNode.Peer peer : cluster) nodes.put(peer.region(), peer.ref());
        System.out.printf("[BENCH] requestWindow=%d%n", WINDOW);
        return this;
    }

    private Behavior<Command> onNodeReady(RaftNode.NodeReady m) {
        if (++readyNodes == 3) System.out.println("[BENCH] all regional nodes are ready");
        return this;
    }

    private Behavior<Command> onLeaderElected(RaftNode.LeaderElected m) {
        if (oldLeaderRegion == null) {
            currentLeaderRegion = m.region(); currentLeaderRef = nodes.get(m.region());
            currentLeaderTerm = m.term();
            initialElectionMs = nanosToMillis(m.observedAtNanos() - bootStartedNanos);
            if (!baselineScheduled) {
                baselineScheduled = true;
                getContext().scheduleOnce(Duration.ofMillis(300), getContext().getSelf(),
                        StartBaseline.INSTANCE);
            }
        } else if (m.term() > oldLeaderTerm && !m.region().equals(oldLeaderRegion)
                && recoveredLeaderRef == null) {
            recoveredLeaderRegion = m.region(); recoveredLeaderRef = nodes.get(m.region());
            recoveredLeaderTerm = m.term();
            failoverElectionMs = nanosToMillis(m.observedAtNanos() - electionStartedNanos);
            System.out.printf(Locale.ROOT,
                    "[FAILOVER] newLeader=%s term=%d electionMs=%.1f%n",
                    recoveredLeaderRegion, recoveredLeaderTerm, failoverElectionMs);
            if (!recoveryScheduled) {
                recoveryScheduled = true;
                getContext().scheduleOnce(Duration.ofMillis(200), getContext().getSelf(),
                        StartRecovery.INSTANCE);
            }
        }
        return this;
    }

    private Behavior<Command> onStartBaseline() {
        if (currentLeaderRef == null || cluster.size() != 3) {
            getContext().scheduleOnce(Duration.ofMillis(100), getContext().getSelf(),
                    StartBaseline.INSTANCE);
            return this;
        }
        beginLoadPhase(Phase.BASELINE, currentLeaderRef);
        for (RaftNode.Peer peer : cluster)
            if (!peer.region().equals(currentLeaderRegion)) sendTracked(peer.ref());
        fillWindow();
        return this;
    }

    private void beginLoadPhase(Phase newPhase, ActorRef<RaftNode.Command> leaderRef) {
        phase = newPhase; currentLeaderRef = leaderRef; loading = true;
        phaseToken++; inFlight = 0; currentStats = new Stats();
        phaseStartedNanos = System.nanoTime(); scheduleLoadTick();
        getContext().scheduleOnce(LOAD_DURATION, getContext().getSelf(),
                new EndPhase(phaseToken));
    }

    private Behavior<Command> onLoadTick(LoadTick m) {
        if (!loading || m.token() != phaseToken) return this;
        if (inFlight >= WINDOW) currentStats.windowFullSamples++;
        fillWindow(); scheduleLoadTick(); return this;
    }

    private void scheduleLoadTick() {
        getContext().scheduleOnce(LOAD_TICK, getContext().getSelf(), new LoadTick(phaseToken));
    }

    private Behavior<Command> onEndPhase(EndPhase m) {
        if (!loading || m.token() != phaseToken) return this;
        loading = false;
        if (inFlight == 0) finishLoadPhase();
        return this;
    }

    private Behavior<Command> onWriteResult(RaftNode.WriteResult m) {
        if ((phase == Phase.BASELINE || phase == Phase.RECOVERY)
                && m.phaseToken() == phaseToken) {
            inFlight = Math.max(0, inFlight - 1);
            Long started = requestStartedNanos.remove(m.requestId());
            if (m.committed()) {
                currentStats.committed++;
                if (started != null)
                    currentStats.commitLatencyMs.add(nanosToMillis(System.nanoTime() - started));
            } else currentStats.rejected++;
            if (loading) fillWindow();
            else if (inFlight == 0) finishLoadPhase();
        } else if (m.phaseToken() == isolatedToken) {
            if (m.committed()) isolatedAcknowledged++; else isolatedRejected++;
        } else if (m.phaseToken() == probeToken) {
            if (m.committed()) failoverProbeCommitted++; else failoverRejected++;
        }
        return this;
    }

    private void fillWindow() {
        while (loading && inFlight < WINDOW) sendTracked(currentLeaderRef);
    }

    private void sendTracked(ActorRef<RaftNode.Command> target) {
        long requestId = nextRequestId++;
        requestStartedNanos.put(requestId, System.nanoTime());
        target.tell(new RaftNode.ClientWrite(requestId, 1, getContext().getSelf(), phaseToken));
        inFlight++; currentStats.sent++;
        currentStats.maxInFlight = Math.max(currentStats.maxInFlight, inFlight);
    }

    private void finishLoadPhase() {
        currentStats.durationSeconds =
                (System.nanoTime() - phaseStartedNanos) / 1_000_000_000.0;
        if (phase == Phase.BASELINE) {
            baselineStats = currentStats.copy(); printStats("before-failover", baselineStats);
            beginPartition();
        } else if (phase == Phase.RECOVERY) {
            recoveryStats = currentStats.copy(); printStats("after-failover", recoveryStats);
            requestStatuses();
        }
    }

    private void beginPartition() {
        phase = Phase.PARTITION; oldLeaderRegion = currentLeaderRegion;
        oldLeaderRef = currentLeaderRef; oldLeaderTerm = currentLeaderTerm;
        currentLeaderRegion = null; currentLeaderRef = null;
        network.tell(new SimulatedNetwork.SetPartition(oldLeaderRegion, true,
                getContext().getSelf()));
    }

    private Behavior<Command> onPartitionApplied(PartitionApplied m) {
        if (phase != Phase.PARTITION || !m.isolated()
                || !m.region().equals(oldLeaderRegion)) return this;
        electionStartedNanos = System.nanoTime();
        isolatedToken = ++phaseToken; probeToken = ++phaseToken;
        for (int i = 0; i < WINDOW; i++) {
            oldLeaderRef.tell(new RaftNode.ClientWrite(nextRequestId++, 1,
                    getContext().getSelf(), isolatedToken));
            isolatedSent++;
        }
        getContext().scheduleOnce(Duration.ofMillis(100), getContext().getSelf(),
                ElectionProbe.INSTANCE);
        return this;
    }

    private Behavior<Command> onElectionProbe() {
        if (phase != Phase.PARTITION || recoveredLeaderRef != null) return this;
        RaftNode.Peer target = cluster.stream()
                .filter(peer -> !peer.region().equals(oldLeaderRegion)).findFirst().orElseThrow();
        target.ref().tell(new RaftNode.ClientWrite(nextRequestId++, 1,
                getContext().getSelf(), probeToken));
        failoverProbes++;
        getContext().scheduleOnce(Duration.ofMillis(100), getContext().getSelf(),
                ElectionProbe.INSTANCE);
        return this;
    }

    private Behavior<Command> onStartRecovery() {
        if (phase != Phase.PARTITION || recoveredLeaderRef == null) return this;
        System.out.printf(
                "[PARTITION] formerLeader=%s acceptedLocally=%d acknowledged=%d rejected=%d unacknowledged=%d%n",
                oldLeaderRegion, isolatedSent, isolatedAcknowledged, isolatedRejected,
                isolatedOutstanding());
        currentLeaderRegion = recoveredLeaderRegion; currentLeaderRef = recoveredLeaderRef;
        currentLeaderTerm = recoveredLeaderTerm;
        beginLoadPhase(Phase.RECOVERY, recoveredLeaderRef); fillWindow();
        return this;
    }

    private long isolatedOutstanding() {
        return isolatedSent - isolatedAcknowledged - isolatedRejected;
    }

    private void requestStatuses() {
        phase = Phase.REPORT; statuses.clear();
        for (RaftNode.Peer peer : cluster)
            peer.ref().tell(new RaftNode.GetStatus(getContext().getSelf()));
    }

    private Behavior<Command> onStatus(RaftNode.Status m) {
        statuses.put(m.region(), m);
        if (statuses.size() == cluster.size()) {
            printFinalReport(); getContext().getSystem().terminate();
        }
        return this;
    }

    private void printStats(String label, Stats s) {
        System.out.printf(Locale.ROOT,
                "[METRIC] phase=%s acknowledgedCommitted=%d rejected=%d throughputOpsPerSec=%.1f avgCommitLatencyMs=%.1f p95CommitLatencyMs=%.1f maxInFlight=%d fullWindowSamples=%d durationSec=%.2f%n",
                label, s.committed, s.rejected, s.committed / s.durationSeconds,
                s.averageLatencyMs(), s.p95LatencyMs(), s.maxInFlight,
                s.windowFullSamples, s.durationSeconds);
    }

    private void printFinalReport() {
        long totalRejected = baselineStats.rejected + recoveryStats.rejected + failoverRejected;
        System.out.println("\n========== REPRODUCIBLE BENCHMARK REPORT ==========");
        System.out.printf(Locale.ROOT, "requestWindow=%d%n", WINDOW);
        System.out.printf(Locale.ROOT, "initialElectionMs=%.1f%n", initialElectionMs);
        System.out.printf(Locale.ROOT, "failoverElectionMs=%.1f%n", failoverElectionMs);
        System.out.printf(Locale.ROOT,
                "before=acknowledgedCommitted:%d,throughput:%.1f,avgLatencyMs:%.1f,p95LatencyMs:%.1f%n",
                baselineStats.committed, baselineStats.committed / baselineStats.durationSeconds,
                baselineStats.averageLatencyMs(), baselineStats.p95LatencyMs());
        System.out.printf(Locale.ROOT,
                "after=acknowledgedCommitted:%d,throughput:%.1f,avgLatencyMs:%.1f,p95LatencyMs:%.1f%n",
                recoveryStats.committed, recoveryStats.committed / recoveryStats.durationSeconds,
                recoveryStats.averageLatencyMs(), recoveryStats.p95LatencyMs());
        System.out.printf("explicitlyRejected=%d%n", totalRejected);
        System.out.printf("unacknowledgedUncommitted=%d%n", isolatedOutstanding());
        System.out.println("acknowledgedButUncommitted=0");
        System.out.printf("outstandingBefore=maxInFlight:%d,fullWindowSamples:%d%n",
                baselineStats.maxInFlight, baselineStats.windowFullSamples);
        System.out.printf("outstandingAfter=maxInFlight:%d,fullWindowSamples:%d%n",
                recoveryStats.maxInFlight, recoveryStats.windowFullSamples);
        System.out.printf("failoverProbes=%d failoverProbeCommitted=%d%n",
                failoverProbes, failoverProbeCommitted);
        System.out.println("\nregion      role       term  commitIndex  committed  replicaIndex  replica  pending");
        statuses.values().stream().sorted((a, b) -> a.region().compareTo(b.region()))
                .forEach(s -> System.out.printf(Locale.ROOT,
                        "%-11s %-10s %-5d %-12d %-10d %-13d %-8d %d%n",
                        s.region(), s.role(), s.term(), s.commitIndex(), s.committedValue(),
                        s.replicaIndex(), s.replicaValue(), s.pendingWrites()));
        System.out.println("\nDemonstrated: terms, one vote per term, majority election, heartbeats,");
        System.out.println("acknowledgement only after majority commitment, minority blocking, and bounded work.");
        System.out.println("Not claimed: full Log Matching, Leader Completeness, arbitrary-history State Machine Safety,");
        System.out.println("stable storage, retries, snapshots, membership changes, deduplication, linearizable reads,");
        System.out.println("Byzantine tolerance, or real multi-process networking.");
        System.out.println("===================================================");
    }

    private static double nanosToMillis(long nanos) { return nanos / 1_000_000.0; }
}

Before you run the final check, predict whether the isolated former leader can produce a successful acknowledgement without either follower.

• Run the completed majority-commit experiment from the Terminal panel:

jbang RegionalRaftSimulation.java

What should I see?

The partition summary should report acknowledged=0 and unacknowledged=16 for the default window.

In the per-region state table, the isolated former leader should have a replica value above its committed value. Its pending count should match the unacknowledged partition batch.

That is the intended shortfall. The former leader can accept entries into local replica state, but the partition prevents it from crossing the majority boundary.

Partition writes were acknowledged?

Check that onClientWrite does not send a successful WriteResult.

Check that only advanceCommitIndex() creates a committed result with the reason MAJORITY_COMMITTED.

Still seeing the wrong outcome? Help me find where my isolated leader acknowledges writes too early.

You now have a visible safety boundary between local replica state and majority commitment. Next, the two connected regions can restore progress by electing a higher-term leader.

Elect a Replacement Leader

Your partition test now proves that the isolated former leader can collect pending replica entries without committing them. Those writes remain unacknowledged because the node cannot reach a majority.

The two connected nodes still form a quorum. However, they need to elect a higher-term leader before the simulation can resume majority commits.

In this step, get ready to:

• Add follower, candidate, and leader roles with term-based voting.
• Send heartbeats that keep connected followers from starting unnecessary elections.
• Measure the election gap before recovery traffic begins.

Expose the leaderless majority

The current simulator blocks the isolated leader correctly. It still lacks a way for the connected majority to restore progress.

Before you run the current version, do you think the two surviving nodes can resume committed writes on their own?

• Switch back to the macOS Terminal from the previous step.
• Run the current simulation with this command:

jbang RegionalRaftSimulation.java

What does this run prove?

The partition leaves the former leader unable to commit. The connected nodes also lack an election path, so recovery traffic cannot begin.

You will see the isolated batch remain unacknowledged. You will not see a higher-term replacement leader restore majority commits.

• Press Ctrl+C to stop the leaderless simulation.

Why the majority needs an election

A Raft candidate increments its term before requesting votes. A candidate becomes leader after receiving votes from a majority.

Each node records its vote for the current term. That rule prevents one node from supporting multiple candidates in the same term.

• Switch back to RaftNode.java in Visual Studio Code.
• Add the election messages and node roles beside the existing private protocol records:

    private record RequestVote(int term, String candidateRegion,
                               ActorRef<Command> candidateRef) implements Command {}
    private record VoteResponse(int term, String voterRegion,
                                boolean granted) implements Command {}
    private record Heartbeat(int term, String leaderRegion,
                             ActorRef<Command> leaderRef) implements Command {}
    private enum ElectionTimeout implements Command { INSTANCE }
    private enum HeartbeatTick implements Command { INSTANCE }
    private enum Role { FOLLOWER, CANDIDATE, LEADER }

What do these protocol types do?

• RequestVote carries the candidate's term plus the reference that receives the response.
• VoteResponse records which region granted or rejected the vote.
• Heartbeat announces the current leader before a follower's election timer expires.
• Role keeps each actor's leadership state inside that actor.

• Replace the election-related field section with this state:

    private static final String ELECTION_TIMER = "election-timer";
    private static final String HEARTBEAT_TIMER = "heartbeat-timer";
    private static final long HEARTBEAT_MS = 150L;
    private final String region;
    private final long electionTimeoutMs;
    private final TimerScheduler<Command> timers;
    private List<Peer> cluster = List.of();
    private ActorRef<SimulatedNetwork.Command> network;
    private ActorRef<Benchmark.Command> observer;
    private Role role = Role.FOLLOWER;
    private int currentTerm;
    private String votedFor;
    private String leaderRegion;
    private ActorRef<Command> leaderRef;
    private final Set<String> votes = new HashSet<>();

What does this state track?

• currentTerm increases whenever the node starts a new election.
• votedFor enforces one vote per term.
• votes stores the distinct regions that support the current candidate.
• leaderRef gives followers a typed reference to the elected leader.

• Replace the actor factory and constructor with the timer-aware versions below:

    public static Behavior<Command> create(String region, long electionTimeoutMs) {
        return Behaviors.setup(context -> Behaviors.withTimers(timers ->
                new RaftNode(context, timers, region, electionTimeoutMs)));
    }

    private RaftNode(ActorContext<Command> context, TimerScheduler<Command> timers,
                     String region, long electionTimeoutMs) {
        super(context);
        this.timers = timers;
        this.region = region;
        this.electionTimeoutMs = electionTimeoutMs;
    }

Why give each actor a timer?

Each regional actor owns its election timer. The configured regional timeout decides which connected follower becomes a candidate first.

• Add the election-timeout handler below onInitialize() in RaftNode.java.

    private Behavior<Command> onElectionTimeout() {
        if (role == Role.LEADER) return this;
        currentTerm++; role = Role.CANDIDATE; votedFor = region;
        leaderRegion = null; leaderRef = null; votes.clear(); votes.add(region);
        resetElectionTimer();
        for (Peer peer : otherPeers())
            send(peer, new RequestVote(currentTerm, region, getContext().getSelf()));
        return this;
    }

What happens when a timer expires?

The node increments its term before becoming a candidate. It votes for itself before requesting one vote from each peer.

The candidate starts a fresh timer in case the first election does not produce a majority.

• Add the vote-request handler below onElectionTimeout().

    private Behavior<Command> onRequestVote(RequestVote m) {
        if (m.term() > currentTerm) stepDown(m.term());
        boolean grant = m.term() == currentTerm
                && (votedFor == null || votedFor.equals(m.candidateRegion()));
        if (grant) {
            role = Role.FOLLOWER; votedFor = m.candidateRegion();
            leaderRegion = null; leaderRef = null; resetElectionTimer();
        }
        send(new Peer(m.candidateRegion(), m.candidateRef()),
                new VoteResponse(currentTerm, region, grant));
        return this;
    }

How is one vote per term enforced?

A node grants the request only when the incoming term matches its current term. The node must also have no recorded vote or have already voted for that candidate.

A request with a higher term first moves the node back to the follower role.

• Add the vote-response handler and leader transition below onRequestVote().

    private Behavior<Command> onVoteResponse(VoteResponse m) {
        if (m.term() > currentTerm) { stepDown(m.term()); return this; }
        if (role != Role.CANDIDATE || m.term() != currentTerm) return this;
        if (m.granted() && votes.add(m.voterRegion()) && votes.size() >= majority()) becomeLeader();
        return this;
    }

    private void becomeLeader() {
        role = Role.LEADER; leaderRegion = region; leaderRef = getContext().getSelf();
        timers.cancel(ELECTION_TIMER); votes.clear();
        observer.tell(new LeaderElected(region, currentTerm, System.nanoTime()));
        System.out.printf("[ELECTION] leader=%s term=%d%n", region, currentTerm);
        sendHeartbeats();
        timers.startTimerWithFixedDelay(HEARTBEAT_TIMER, HeartbeatTick.INSTANCE,
                Duration.ofMillis(HEARTBEAT_MS));
    }

When does a candidate become leader?

The candidate counts each voter region once. Two votes satisfy the majority for this three-node simulation.

The new leader cancels its election timer before announcing the result to Benchmark. It then starts the heartbeat cycle.

Keep followers behind the active leader

A connected follower should keep its current leader while heartbeats continue to arrive. An isolated follower receives no heartbeats, so its election timer eventually expires.

• Add the heartbeat handlers below becomeLeader() in RaftNode.java.

    private Behavior<Command> onHeartbeatTick() {
        if (role == Role.LEADER) sendHeartbeats();
        return this;
    }

    private void sendHeartbeats() {
        for (Peer peer : otherPeers())
            send(peer, new Heartbeat(currentTerm, region, getContext().getSelf()));
    }

    private Behavior<Command> onHeartbeat(Heartbeat m) {
        if (m.term() < currentTerm) return this;
        if (m.term() > currentTerm || role != Role.FOLLOWER) stepDown(m.term());
        leaderRegion = m.leaderRegion(); leaderRef = m.leaderRef(); resetElectionTimer();
        return this;
    }

How do heartbeats preserve leadership?

The leader sends a heartbeat every 150 milliseconds. Each accepted heartbeat resets the follower's election timer.

A stale heartbeat cannot replace a node's newer term. A heartbeat with the current or a newer term records the active leader.

• Add the step-down and timer helpers near the bottom of RaftNode.java.

    private void stepDown(int newTerm) {
        if (newTerm > currentTerm) { currentTerm = newTerm; votedFor = null; }
        role = Role.FOLLOWER; leaderRegion = null; leaderRef = null; votes.clear();
        timers.cancel(HEARTBEAT_TIMER); resetElectionTimer();
    }

    private void resetElectionTimer() {
        timers.startSingleTimer(ELECTION_TIMER, ElectionTimeout.INSTANCE,
                Duration.ofMillis(electionTimeoutMs));
    }

Why clear leadership state?

A node that observes a higher term returns to the follower role. It clears candidate votes plus its remembered leader before starting a fresh election timer.

Cancelling the heartbeat timer prevents a former leader from continuing its leader duties after stepping down.

• Replace createReceive() with this complete handler map:

    @Override
    public Receive<Command> createReceive() {
        return newReceiveBuilder()
                .onMessage(Initialize.class, this::onInitialize)
                .onMessage(ClientWrite.class, this::onClientWrite)
                .onMessage(GetStatus.class, this::onGetStatus)
                .onMessage(RequestVote.class, this::onRequestVote)
                .onMessage(VoteResponse.class, this::onVoteResponse)
                .onMessage(Heartbeat.class, this::onHeartbeat)
                .onMessage(AppendEntry.class, this::onAppendEntry)
                .onMessage(AppendAck.class, this::onAppendAck)
                .onMessage(CommitEntry.class, this::onCommitEntry)
                .onMessage(ElectionTimeout.class, ignored -> onElectionTimeout())
                .onMessage(HeartbeatTick.class, ignored -> onHeartbeatTick())
                .build();
    }

What does the handler map connect?

The receive map routes election requests, vote responses, heartbeats, and timer events to their handlers. Existing replication messages continue through their original handlers.

• Find this line inside onInitialize().

observer.tell(new NodeReady(region));

What does the current line do?

The current initialization path only announces that the node is ready. It does not start the node's election timer.

• Replace the current line with this timer-aware statement:

        resetElectionTimer(); observer.tell(new NodeReady(region));

What changes during initialization?

Every node now starts its own deterministic election timer before reporting readiness. The 500 millisecond regional timeout makes us-east the first candidate during the initial election.

• Find the leader check at the start of onClientWrite().
• Replace that check with this role-based gate:

    private Behavior<Command> onClientWrite(ClientWrite m) {
        if (role != Role.LEADER) {
            m.replyTo().tell(new WriteResult(m.requestId(), m.phaseToken(), false,
                    region, "NOT_LEADER"));
            return this;
        }

How are writes gated now?

Only the actor currently in the leader role can append a client write to replica state. Every other role returns an explicit NOT_LEADER result without accepting the entry.

• Save RaftNode.java.

Measure the failover gap

Node elections restore a leader inside the connected majority. Benchmark still needs to recognize that replacement, count rejected probes during the gap, and delay recovery load until the new leader stabilizes.

• Switch back to Benchmark.java.
• Add the election probe command beside the existing phase commands:

    private enum ElectionProbe implements Command { INSTANCE }

What is an election probe?

An election probe is a client write sent to a surviving node while no replacement leader is stable. Its NOT_LEADER response makes the availability gap measurable.

• Replace the leader and failover field section with this state:

    private String currentLeaderRegion;
    private ActorRef<RaftNode.Command> currentLeaderRef;
    private int currentLeaderTerm;
    private String oldLeaderRegion;
    private ActorRef<RaftNode.Command> oldLeaderRef;
    private int oldLeaderTerm;
    private String recoveredLeaderRegion;
    private ActorRef<RaftNode.Command> recoveredLeaderRef;
    private int recoveredLeaderTerm;
    private long nextRequestId = 1;
    private int phaseToken;
    private int isolatedToken = -1;
    private int probeToken = -1;
    private int inFlight;
    private long phaseStartedNanos;
    private long electionStartedNanos;
    private double initialElectionMs;
    private double failoverElectionMs;

What does Benchmark remember?

Benchmark keeps the original leader separate from the recovered leader. It also assigns different phase tokens to isolated writes and election probes.

The election timestamps measure the disruption from partition confirmation to replacement-leader observation.

• Replace onLeaderElected() with this initial-election and failover logic:

    private Behavior<Command> onLeaderElected(RaftNode.LeaderElected m) {
        if (oldLeaderRegion == null) {
            currentLeaderRegion = m.region(); currentLeaderRef = nodes.get(m.region());
            currentLeaderTerm = m.term();
            initialElectionMs = nanosToMillis(m.observedAtNanos() - bootStartedNanos);
            if (!baselineScheduled) {
                baselineScheduled = true;
                getContext().scheduleOnce(Duration.ofMillis(300), getContext().getSelf(),
                        StartBaseline.INSTANCE);
            }
        } else if (m.term() > oldLeaderTerm && !m.region().equals(oldLeaderRegion)
                && recoveredLeaderRef == null) {
            recoveredLeaderRegion = m.region(); recoveredLeaderRef = nodes.get(m.region());
            recoveredLeaderTerm = m.term();
            failoverElectionMs = nanosToMillis(m.observedAtNanos() - electionStartedNanos);
            System.out.printf(Locale.ROOT,
                    "[FAILOVER] newLeader=%s term=%d electionMs=%.1f%n",
                    recoveredLeaderRegion, recoveredLeaderTerm, failoverElectionMs);
            if (!recoveryScheduled) {
                recoveryScheduled = true;
                getContext().scheduleOnce(Duration.ofMillis(200), getContext().getSelf(),
                        StartRecovery.INSTANCE);
            }
        }
        return this;
    }

How is the replacement recognized?

The replacement must have a higher term than the isolated leader. It must also be a different region.

Benchmark waits 200 milliseconds after observing that leader. This brief stabilization period lets its first heartbeats reach the connected follower before recovery traffic begins.

• Replace the partition-start method with this election-aware version:

    private void beginPartition() {
        phase = Phase.PARTITION; oldLeaderRegion = currentLeaderRegion;
        oldLeaderRef = currentLeaderRef; oldLeaderTerm = currentLeaderTerm;
        currentLeaderRegion = null; currentLeaderRef = null;
        network.tell(new SimulatedNetwork.SetPartition(oldLeaderRegion, true,
                getContext().getSelf()));
    }

Why clear the current leader?

Benchmark preserves the original leader in oldLeaderRef before clearing its active-leader fields. Recovery cannot start until a different higher-term leader fills those fields.

• Replace onPartitionApplied() with this bounded-batch and probe setup:

    private Behavior<Command> onPartitionApplied(PartitionApplied m) {
        if (phase != Phase.PARTITION || !m.isolated()
                || !m.region().equals(oldLeaderRegion)) return this;
        electionStartedNanos = System.nanoTime();
        isolatedToken = ++phaseToken; probeToken = ++phaseToken;
        for (int i = 0; i < WINDOW; i++) {
            oldLeaderRef.tell(new RaftNode.ClientWrite(nextRequestId++, 1,
                    getContext().getSelf(), isolatedToken));
            isolatedSent++;
        }
        getContext().scheduleOnce(Duration.ofMillis(100), getContext().getSelf(),
                ElectionProbe.INSTANCE);
        return this;
    }

What starts when the partition is confirmed?

Benchmark starts the failover clock before sending the bounded batch to the isolated leader. It also schedules the first probe against the surviving side.

Separate phase tokens keep unacknowledged isolated writes distinct from explicitly rejected probes.

• Add the probe loop below onPartitionApplied().

    private Behavior<Command> onElectionProbe() {
        if (phase != Phase.PARTITION || recoveredLeaderRef != null) return this;
        RaftNode.Peer target = cluster.stream()
                .filter(peer -> !peer.region().equals(oldLeaderRegion)).findFirst().orElseThrow();
        target.ref().tell(new RaftNode.ClientWrite(nextRequestId++, 1,
                getContext().getSelf(), probeToken));
        failoverProbes++;
        getContext().scheduleOnce(Duration.ofMillis(100), getContext().getSelf(),
                ElectionProbe.INSTANCE);
        return this;
    }

How does the probe loop measure disruption?

Benchmark sends one probe every 100 milliseconds until it observes a replacement leader. A surviving follower or candidate rejects each probe with NOT_LEADER.

The loop ends as soon as recoveredLeaderRef is set.

• Add the probe result branch inside onWriteResult() after the isolated-write branch:

        } else if (m.phaseToken() == probeToken) {
            if (m.committed()) failoverProbeCommitted++; else failoverRejected++;
        }

What outcomes are counted?

A rejected probe contributes to the measured election gap. A committed probe would reveal that traffic reached a leader before Benchmark recognized the recovery event.

• Replace onStartRecovery() with this guarded recovery transition:

    private Behavior<Command> onStartRecovery() {
        if (phase != Phase.PARTITION || recoveredLeaderRef == null) return this;
        System.out.printf(
                "[PARTITION] formerLeader=%s acceptedLocally=%d acknowledged=%d rejected=%d unacknowledged=%d%n",
                oldLeaderRegion, isolatedSent, isolatedAcknowledged, isolatedRejected,
                isolatedOutstanding());
        currentLeaderRegion = recoveredLeaderRegion; currentLeaderRef = recoveredLeaderRef;
        currentLeaderTerm = recoveredLeaderTerm;
        beginLoadPhase(Phase.RECOVERY, recoveredLeaderRef); fillWindow();
        return this;
    }

Why wait before recovery traffic?

Recovery begins only when Benchmark has a typed reference to the higher-term replacement leader. The isolated former leader remains separated, so its pending replica entries still cannot reach a majority.

• Replace createReceive() in Benchmark.java with this handler map:

    @Override
    public Receive<Command> createReceive() {
        return newReceiveBuilder()
                .onMessage(Configure.class, this::onConfigure)
                .onMessage(PartitionApplied.class, this::onPartitionApplied)
                .onMessage(RaftNode.NodeReady.class, this::onNodeReady)
                .onMessage(RaftNode.LeaderElected.class, this::onLeaderElected)
                .onMessage(RaftNode.WriteResult.class, this::onWriteResult)
                .onMessage(RaftNode.Status.class, this::onStatus)
                .onMessage(StartBaseline.class, ignored -> onStartBaseline())
                .onMessage(StartRecovery.class, ignored -> onStartRecovery())
                .onMessage(ElectionProbe.class, ignored -> onElectionProbe())
                .onMessage(LoadTick.class, this::onLoadTick)
                .onMessage(EndPhase.class, this::onEndPhase)
                .build();
    }

What completes the Benchmark wiring?

The receive map now routes election events from RaftNode. It also routes each scheduled ElectionProbe back into the probe loop.

• Save Benchmark.java.

Before you run the updated simulation, do you expect the surviving majority to elect the same region or a different region?

• Return to the macOS Terminal from earlier.
• Run the completed election flow with this command:

jbang RegionalRaftSimulation.java

What does the final run test?

This run exercises the full transition from initial leader to partitioned former leader. It then tests higher-term election plus stabilized recovery traffic.

You will see a different leader elected in a higher term. The FAILOVER line includes the measured election duration.

You will also see rejected probes from the election gap. Recovery commits resume through the connected two-node majority.

The former leader still reports pending state with replicaValue above committedValue. Its isolated batch remains unacknowledged.

No replacement leader appears?

Check that every node starts resetElectionTimer() inside onInitialize(). Confirm that onHeartbeat() resets the timer only after accepting a current or newer term.

Check that RequestVote and VoteResponse are both registered inside createReceive(). Confirm that the candidate counts its own vote before sending requests.

Still stuck? Help me debug why my connected RaftNode actors do not elect a replacement leader.

✔️ Awesome, I've got everything!

Your connected majority now elects a higher-term replacement leader. Save both Java files before moving on.

ⓧ I'd like to double check the election code

Compare your election protocol, timers, voting handlers, heartbeat handlers, and role transitions against the snippets above.

• Confirm RaftNode.java defines FOLLOWER, CANDIDATE, and LEADER roles.
• Confirm each election increments currentTerm before requesting votes.
• Confirm a candidate needs two distinct votes before calling becomeLeader().

Compare Benchmark.java against the failover flow above.

• Confirm the partition records oldLeaderTerm before clearing the active leader.
• Confirm ElectionProbe repeats until recoveredLeaderRef is available.
• Confirm recovery begins only after the higher-term replacement leader stabilizes.

Your surviving majority now restores progress without help from the isolated former leader. Next, you will turn this behavior into a reproducible benchmark report.

Produce the Benchmark Report

Your connected majority now elects a replacement leader and resumes committed writes. The isolated former leader still holds pending entries without a successful acknowledgement.

A failover log shows that something happened. This step turns the run into measurable Raft evidence that separates committed work, rejected work, and unresolved work.

In this step, get ready to:

• Measure throughput and commit latency from timestamped writes.
• Report each write outcome alongside per-node replica and committed state.
• Document the demonstrated guarantees and omitted safety mechanisms.

Measure completed writes

Commit latency starts when the benchmark sends a tracked write. A latency sample only finishes when the majority-commit acknowledgement returns.

• In Benchmark.java, locate the existing Stats inner class.
• Replace that class with this version:

    private static final class Stats {
        long sent;
        long committed;
        long rejected;
        int maxInFlight;
        long windowFullSamples;
        double durationSeconds;
        final List<Double> commitLatencyMs = new ArrayList<>();

        double averageLatencyMs() {
            return commitLatencyMs.stream().mapToDouble(Double::doubleValue).average().orElse(0.0);
        }

        double p95LatencyMs() {
            if (commitLatencyMs.isEmpty()) return 0.0;
            List<Double> sorted = new ArrayList<>(commitLatencyMs);
            sorted.sort(Double::compareTo);
            int index = Math.max(0, (int) Math.ceil(sorted.size() * 0.95) - 1);
            return sorted.get(index);
        }

        Stats copy() {
            Stats copy = new Stats();
            copy.sent = sent; copy.committed = committed; copy.rejected = rejected;
            copy.maxInFlight = maxInFlight; copy.windowFullSamples = windowFullSamples;
            copy.durationSeconds = durationSeconds;
            copy.commitLatencyMs.addAll(commitLatencyMs);
            return copy;
        }
    }

What does this code measure?

• The counters separate sent writes from committed writes and explicit rejections.
• The latency list stores samples only for writes that return a successful majority-commit acknowledgement.
• The copied snapshot prevents later benchmark phases from changing an earlier phase's results.

• Save Benchmark.java.
• Confirm the editor shows no red syntax markers around the Stats class.

Seeing errors around Stats?

Check that the replacement remains inside the Benchmark class. Confirm that the imports for ArrayList and List remain at the top of the file.

Still stuck? Help me fix the Stats class in Benchmark.java.

The benchmark must also remove completed requests from its in-flight count. This keeps the bounded window moving while preserving unresolved requests as outstanding work.

• In Benchmark.java, locate the onWriteResult method.
• Replace that method with this version:

    private Behavior<Command> onWriteResult(RaftNode.WriteResult m) {
        if ((phase == Phase.BASELINE || phase == Phase.RECOVERY)
                && m.phaseToken() == phaseToken) {
            inFlight = Math.max(0, inFlight - 1);
            Long started = requestStartedNanos.remove(m.requestId());
            if (m.committed()) {
                currentStats.committed++;
                if (started != null)
                    currentStats.commitLatencyMs.add(nanosToMillis(System.nanoTime() - started));
            } else currentStats.rejected++;
            if (loading) fillWindow();
            else if (inFlight == 0) finishLoadPhase();
        } else if (m.phaseToken() == isolatedToken) {
            if (m.committed()) isolatedAcknowledged++; else isolatedRejected++;
        } else if (m.phaseToken() == probeToken) {
            if (m.committed()) failoverProbeCommitted++; else failoverRejected++;
        }
        return this;
    }

How are outcomes separated?

• Baseline and recovery replies update phase metrics.
• Isolated-leader replies update the partition counters.
• Election-gap replies update the probe counters.
• A drained phase finishes only after its final in-flight reply returns.

• Save Benchmark.java.
• Confirm the method references finishLoadPhase() after the in-flight count reaches zero.

Seeing unresolved method names?

Keep the method names exactly as shown in the complete file reference below. A spelling change prevents the benchmark actor from compiling.

Need another pair of eyes? Help me trace the unresolved Benchmark method.

The complete benchmark connects phase timing, bounded load generation, partition accounting, status collection, and actor-system termination. Use the reference to catch any missing reporting lines before the first run.

• Compare your Benchmark.java file with the complete reference in the second tab.
• Apply any missing lines from the reference to Benchmark.java.

✔️ Awesome, I've got everything!

Great. Save Benchmark.java before running the simulation.

ⓧ I'd like to double check the full code

import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;

import org.apache.pekko.actor.typed.ActorRef;
import org.apache.pekko.actor.typed.Behavior;
import org.apache.pekko.actor.typed.javadsl.AbstractBehavior;
import org.apache.pekko.actor.typed.javadsl.ActorContext;
import org.apache.pekko.actor.typed.javadsl.Behaviors;
import org.apache.pekko.actor.typed.javadsl.Receive;

public final class Benchmark extends AbstractBehavior<Benchmark.Command> {
    public interface Command {}
    public record Configure(List<RaftNode.Peer> cluster) implements Command {}
    public record PartitionApplied(String region, boolean isolated) implements Command {}
    private enum StartBaseline implements Command { INSTANCE }
    private enum StartRecovery implements Command { INSTANCE }
    private enum ElectionProbe implements Command { INSTANCE }
    private record LoadTick(int token) implements Command {}
    private record EndPhase(int token) implements Command {}
    private enum Phase { BOOT, BASELINE, PARTITION, RECOVERY, REPORT }

    private static final int WINDOW = Math.max(1, Integer.getInteger("sim.window", 16));
    private static final Duration LOAD_DURATION = Duration.ofSeconds(2);
    private static final Duration LOAD_TICK = Duration.ofMillis(50);

    private static final class Stats {
        long sent;
        long committed;
        long rejected;
        int maxInFlight;
        long windowFullSamples;
        double durationSeconds;
        final List<Double> commitLatencyMs = new ArrayList<>();

        double averageLatencyMs() {
            return commitLatencyMs.stream().mapToDouble(Double::doubleValue).average().orElse(0.0);
        }

        double p95LatencyMs() {
            if (commitLatencyMs.isEmpty()) return 0.0;
            List<Double> sorted = new ArrayList<>(commitLatencyMs);
            sorted.sort(Double::compareTo);
            int index = Math.max(0, (int) Math.ceil(sorted.size() * 0.95) - 1);
            return sorted.get(index);
        }

        Stats copy() {
            Stats copy = new Stats();
            copy.sent = sent; copy.committed = committed; copy.rejected = rejected;
            copy.maxInFlight = maxInFlight; copy.windowFullSamples = windowFullSamples;
            copy.durationSeconds = durationSeconds;
            copy.commitLatencyMs.addAll(commitLatencyMs);
            return copy;
        }
    }

    private final ActorRef<SimulatedNetwork.Command> network;
    private final long bootStartedNanos = System.nanoTime();
    private List<RaftNode.Peer> cluster = List.of();
    private final Map<String, ActorRef<RaftNode.Command>> nodes = new HashMap<>();
    private final Map<String, RaftNode.Status> statuses = new HashMap<>();
    private final Map<Long, Long> requestStartedNanos = new HashMap<>();
    private Phase phase = Phase.BOOT;
    private int readyNodes;
    private boolean baselineScheduled;
    private boolean recoveryScheduled;
    private boolean loading;
    private String currentLeaderRegion;
    private ActorRef<RaftNode.Command> currentLeaderRef;
    private int currentLeaderTerm;
    private String oldLeaderRegion;
    private ActorRef<RaftNode.Command> oldLeaderRef;
    private int oldLeaderTerm;
    private String recoveredLeaderRegion;
    private ActorRef<RaftNode.Command> recoveredLeaderRef;
    private int recoveredLeaderTerm;
    private long nextRequestId = 1;
    private int phaseToken;
    private int isolatedToken = -1;
    private int probeToken = -1;
    private int inFlight;
    private long phaseStartedNanos;
    private long electionStartedNanos;
    private double initialElectionMs;
    private double failoverElectionMs;
    private Stats currentStats;
    private Stats baselineStats;
    private Stats recoveryStats;
    private long isolatedSent;
    private long isolatedAcknowledged;
    private long isolatedRejected;
    private long failoverProbes;
    private long failoverRejected;
    private long failoverProbeCommitted;

    public static Behavior<Command> create(ActorRef<SimulatedNetwork.Command> network) {
        return Behaviors.setup(context -> new Benchmark(context, network));
    }

    private Benchmark(ActorContext<Command> context,
                      ActorRef<SimulatedNetwork.Command> network) {
        super(context); this.network = network;
    }

    @Override
    public Receive<Command> createReceive() {
        return newReceiveBuilder()
                .onMessage(Configure.class, this::onConfigure)
                .onMessage(PartitionApplied.class, this::onPartitionApplied)
                .onMessage(RaftNode.NodeReady.class, this::onNodeReady)
                .onMessage(RaftNode.LeaderElected.class, this::onLeaderElected)
                .onMessage(RaftNode.WriteResult.class, this::onWriteResult)
                .onMessage(RaftNode.Status.class, this::onStatus)
                .onMessage(StartBaseline.class, ignored -> onStartBaseline())
                .onMessage(StartRecovery.class, ignored -> onStartRecovery())
                .onMessage(ElectionProbe.class, ignored -> onElectionProbe())
                .onMessage(LoadTick.class, this::onLoadTick)
                .onMessage(EndPhase.class, this::onEndPhase)
                .build();
    }

    private Behavior<Command> onConfigure(Configure m) {
        cluster = List.copyOf(m.cluster());
        for (RaftNode.Peer peer : cluster) nodes.put(peer.region(), peer.ref());
        System.out.printf("[BENCH] requestWindow=%d%n", WINDOW);
        return this;
    }

    private Behavior<Command> onNodeReady(RaftNode.NodeReady m) {
        if (++readyNodes == 3) System.out.println("[BENCH] all regional nodes are ready");
        return this;
    }

    private Behavior<Command> onLeaderElected(RaftNode.LeaderElected m) {
        if (oldLeaderRegion == null) {
            currentLeaderRegion = m.region(); currentLeaderRef = nodes.get(m.region());
            currentLeaderTerm = m.term();
            initialElectionMs = nanosToMillis(m.observedAtNanos() - bootStartedNanos);
            if (!baselineScheduled) {
                baselineScheduled = true;
                getContext().scheduleOnce(Duration.ofMillis(300), getContext().getSelf(),
                        StartBaseline.INSTANCE);
            }
        } else if (m.term() > oldLeaderTerm && !m.region().equals(oldLeaderRegion)
                && recoveredLeaderRef == null) {
            recoveredLeaderRegion = m.region(); recoveredLeaderRef = nodes.get(m.region());
            recoveredLeaderTerm = m.term();
            failoverElectionMs = nanosToMillis(m.observedAtNanos() - electionStartedNanos);
            System.out.printf(Locale.ROOT,
                    "[FAILOVER] newLeader=%s term=%d electionMs=%.1f%n",
                    recoveredLeaderRegion, recoveredLeaderTerm, failoverElectionMs);
            if (!recoveryScheduled) {
                recoveryScheduled = true;
                getContext().scheduleOnce(Duration.ofMillis(200), getContext().getSelf(),
                        StartRecovery.INSTANCE);
            }
        }
        return this;
    }

    private Behavior<Command> onStartBaseline() {
        if (currentLeaderRef == null || cluster.size() != 3) {
            getContext().scheduleOnce(Duration.ofMillis(100), getContext().getSelf(),
                    StartBaseline.INSTANCE);
            return this;
        }
        beginLoadPhase(Phase.BASELINE, currentLeaderRef);
        for (RaftNode.Peer peer : cluster)
            if (!peer.region().equals(currentLeaderRegion)) sendTracked(peer.ref());
        fillWindow();
        return this;
    }

    private void beginLoadPhase(Phase newPhase, ActorRef<RaftNode.Command> leaderRef) {
        phase = newPhase; currentLeaderRef = leaderRef; loading = true;
        phaseToken++; inFlight = 0; currentStats = new Stats();
        phaseStartedNanos = System.nanoTime(); scheduleLoadTick();
        getContext().scheduleOnce(LOAD_DURATION, getContext().getSelf(),
                new EndPhase(phaseToken));
    }

    private Behavior<Command> onLoadTick(LoadTick m) {
        if (!loading || m.token() != phaseToken) return this;
        if (inFlight >= WINDOW) currentStats.windowFullSamples++;
        fillWindow(); scheduleLoadTick(); return this;
    }

    private void scheduleLoadTick() {
        getContext().scheduleOnce(LOAD_TICK, getContext().getSelf(), new LoadTick(phaseToken));
    }

    private Behavior<Command> onEndPhase(EndPhase m) {
        if (!loading || m.token() != phaseToken) return this;
        loading = false;
        if (inFlight == 0) finishLoadPhase();
        return this;
    }

    private Behavior<Command> onWriteResult(RaftNode.WriteResult m) {
        if ((phase == Phase.BASELINE || phase == Phase.RECOVERY)
                && m.phaseToken() == phaseToken) {
            inFlight = Math.max(0, inFlight - 1);
            Long started = requestStartedNanos.remove(m.requestId());
            if (m.committed()) {
                currentStats.committed++;
                if (started != null)
                    currentStats.commitLatencyMs.add(nanosToMillis(System.nanoTime() - started));
            } else currentStats.rejected++;
            if (loading) fillWindow();
            else if (inFlight == 0) finishLoadPhase();
        } else if (m.phaseToken() == isolatedToken) {
            if (m.committed()) isolatedAcknowledged++; else isolatedRejected++;
        } else if (m.phaseToken() == probeToken) {
            if (m.committed()) failoverProbeCommitted++; else failoverRejected++;
        }
        return this;
    }

    private void fillWindow() {
        while (loading && inFlight < WINDOW) sendTracked(currentLeaderRef);
    }

    private void sendTracked(ActorRef<RaftNode.Command> target) {
        long requestId = nextRequestId++;
        requestStartedNanos.put(requestId, System.nanoTime());
        target.tell(new RaftNode.ClientWrite(requestId, 1, getContext().getSelf(), phaseToken));
        inFlight++; currentStats.sent++;
        currentStats.maxInFlight = Math.max(currentStats.maxInFlight, inFlight);
    }

    private void finishLoadPhase() {
        currentStats.durationSeconds =
                (System.nanoTime() - phaseStartedNanos) / 1_000_000_000.0;
        if (phase == Phase.BASELINE) {
            baselineStats = currentStats.copy(); printStats("before-failover", baselineStats);
            beginPartition();
        } else if (phase == Phase.RECOVERY) {
            recoveryStats = currentStats.copy(); printStats("after-failover", recoveryStats);
            requestStatuses();
        }
    }

    private void beginPartition() {
        phase = Phase.PARTITION; oldLeaderRegion = currentLeaderRegion;
        oldLeaderRef = currentLeaderRef; oldLeaderTerm = currentLeaderTerm;
        currentLeaderRegion = null; currentLeaderRef = null;
        network.tell(new SimulatedNetwork.SetPartition(oldLeaderRegion, true,
                getContext().getSelf()));
    }

    private Behavior<Command> onPartitionApplied(PartitionApplied m) {
        if (phase != Phase.PARTITION || !m.isolated()
                || !m.region().equals(oldLeaderRegion)) return this;
        electionStartedNanos = System.nanoTime();
        isolatedToken = ++phaseToken; probeToken = ++phaseToken;
        for (int i = 0; i < WINDOW; i++) {
            oldLeaderRef.tell(new RaftNode.ClientWrite(nextRequestId++, 1,
                    getContext().getSelf(), isolatedToken));
            isolatedSent++;
        }
        getContext().scheduleOnce(Duration.ofMillis(100), getContext().getSelf(),
                ElectionProbe.INSTANCE);
        return this;
    }

    private Behavior<Command> onElectionProbe() {
        if (phase != Phase.PARTITION || recoveredLeaderRef != null) return this;
        RaftNode.Peer target = cluster.stream()
                .filter(peer -> !peer.region().equals(oldLeaderRegion)).findFirst().orElseThrow();
        target.ref().tell(new RaftNode.ClientWrite(nextRequestId++, 1,
                getContext().getSelf(), probeToken));
        failoverProbes++;
        getContext().scheduleOnce(Duration.ofMillis(100), getContext().getSelf(),
                ElectionProbe.INSTANCE);
        return this;
    }

    private Behavior<Command> onStartRecovery() {
        if (phase != Phase.PARTITION || recoveredLeaderRef == null) return this;
        System.out.printf(
                "[PARTITION] formerLeader=%s acceptedLocally=%d acknowledged=%d rejected=%d unacknowledged=%d%n",
                oldLeaderRegion, isolatedSent, isolatedAcknowledged, isolatedRejected,
                isolatedOutstanding());
        currentLeaderRegion = recoveredLeaderRegion; currentLeaderRef = recoveredLeaderRef;
        currentLeaderTerm = recoveredLeaderTerm;
        beginLoadPhase(Phase.RECOVERY, recoveredLeaderRef); fillWindow();
        return this;
    }

    private long isolatedOutstanding() {
        return isolatedSent - isolatedAcknowledged - isolatedRejected;
    }

    private void requestStatuses() {
        phase = Phase.REPORT; statuses.clear();
        for (RaftNode.Peer peer : cluster)
            peer.ref().tell(new RaftNode.GetStatus(getContext().getSelf()));
    }

    private Behavior<Command> onStatus(RaftNode.Status m) {
        statuses.put(m.region(), m);
        if (statuses.size() == cluster.size()) {
            printFinalReport(); getContext().getSystem().terminate();
        }
        return this;
    }

    private void printStats(String label, Stats s) {
        System.out.printf(Locale.ROOT,
                "[METRIC] phase=%s acknowledgedCommitted=%d rejected=%d throughputOpsPerSec=%.1f avgCommitLatencyMs=%.1f p95CommitLatencyMs=%.1f maxInFlight=%d fullWindowSamples=%d durationSec=%.2f%n",
                label, s.committed, s.rejected, s.committed / s.durationSeconds,
                s.averageLatencyMs(), s.p95LatencyMs(), s.maxInFlight,
                s.windowFullSamples, s.durationSeconds);
    }

    private void printFinalReport() {
        long totalRejected = baselineStats.rejected + recoveryStats.rejected + failoverRejected;
        System.out.println("\n========== REPRODUCIBLE BENCHMARK REPORT ==========");
        System.out.printf(Locale.ROOT, "requestWindow=%d%n", WINDOW);
        System.out.printf(Locale.ROOT, "initialElectionMs=%.1f%n", initialElectionMs);
        System.out.printf(Locale.ROOT, "failoverElectionMs=%.1f%n", failoverElectionMs);
        System.out.printf(Locale.ROOT,
                "before=acknowledgedCommitted:%d,throughput:%.1f,avgLatencyMs:%.1f,p95LatencyMs:%.1f%n",
                baselineStats.committed, baselineStats.committed / baselineStats.durationSeconds,
                baselineStats.averageLatencyMs(), baselineStats.p95LatencyMs());
        System.out.printf(Locale.ROOT,
                "after=acknowledgedCommitted:%d,throughput:%.1f,avgLatencyMs:%.1f,p95LatencyMs:%.1f%n",
                recoveryStats.committed, recoveryStats.committed / recoveryStats.durationSeconds,
                recoveryStats.averageLatencyMs(), recoveryStats.p95LatencyMs());
        System.out.printf("explicitlyRejected=%d%n", totalRejected);
        System.out.printf("unacknowledgedUncommitted=%d%n", isolatedOutstanding());
        System.out.println("acknowledgedButUncommitted=0");
        System.out.printf("outstandingBefore=maxInFlight:%d,fullWindowSamples:%d%n",
                baselineStats.maxInFlight, baselineStats.windowFullSamples);
        System.out.printf("outstandingAfter=maxInFlight:%d,fullWindowSamples:%d%n",
                recoveryStats.maxInFlight, recoveryStats.windowFullSamples);
        System.out.printf("failoverProbes=%d failoverProbeCommitted=%d%n",
                failoverProbes, failoverProbeCommitted);
        System.out.println("\nregion      role       term  commitIndex  committed  replicaIndex  replica  pending");
        statuses.values().stream().sorted((a, b) -> a.region().compareTo(b.region()))
                .forEach(s -> System.out.printf(Locale.ROOT,
                        "%-11s %-10s %-5d %-12d %-10d %-13d %-8d %d%n",
                        s.region(), s.role(), s.term(), s.commitIndex(), s.committedValue(),
                        s.replicaIndex(), s.replicaValue(), s.pendingWrites()));
        System.out.println("\nDemonstrated: terms, one vote per term, majority election, heartbeats,");
        System.out.println("acknowledgement only after majority commitment, minority blocking, and bounded work.");
        System.out.println("Not claimed: full Log Matching, Leader Completeness, arbitrary-history State Machine Safety,");
        System.out.println("stable storage, retries, snapshots, membership changes, deduplication, linearizable reads,");
        System.out.println("Byzantine tolerance, or real multi-process networking.");
        System.out.println("===================================================");
    }

    private static double nanosToMillis(long nanos) { return nanos / 1_000_000.0; }
}

This reference is the cumulative Benchmark.java file for the project. Its final status handler prints the report before terminating the actor system.

Before you run the simulation, which values do you expect to differ between the original leader's replica state and committed state?

• Run the complete benchmark from the terminal in the regional-raft-simulator folder by running:

jbang RegionalRaftSimulation.java

What does this check prove?

The command runs the initial load, partition, failover, recovery load, and final status collection. The process terminates after all three node statuses reach the benchmark actor.

You should see metric lines for before-failover and after-failover. The final report should follow them before the terminal prompt returns.

Does the benchmark keep running?

Check that onStatus calls printFinalReport() after all three statuses arrive. Confirm that the same block calls the actor system's termination method.

Need help tracing the final phase? Help me debug why the benchmark does not terminate.

Interpret the three write outcomes

A trustworthy report keeps success, rejection, and unresolved work separate. Combining them would hide whether a request committed or merely entered one replica.

• Find acknowledgedCommitted in each metric phase to confirm that successful replies represent majority commits.
• Find explicitlyRejected to confirm that election-gap probes received definite non-leader responses.
• Find unacknowledgedUncommitted to confirm that the isolated batch remains unresolved.
• Compare the original leader's replica value with its committed value to confirm that local acceptance moved ahead of commitment.
• Confirm that acknowledgedButUncommitted=0 appears in the report.

Why drain each phase?

The benchmark stops generating work when a phase duration ends. It waits for that phase's in-flight replies before calculating throughput and latency.

This boundary keeps baseline work out of recovery metrics. Each phase measures its own completed workload.

Document the evidence and safety boundary

A benchmark becomes reusable when another person can run it and interpret the result. The README.md file records the outcome definitions before documenting the safety boundary.

• Create README.md inside the regional-raft-simulator folder using the new-file control in the Visual Studio Code sidebar.
• Add the project overview and outcome definitions by pasting:

# Three-Region Raft and Actor-State Simulator

A local Java 17 simulation built with Apache Pekko Typed 1.7.1 and JBang. Three actors represent `us-east`, `eu-west`, and `ap-south` behind a latency and partition simulator.

## Run

```bash
jbang RegionalRaftSimulation.java
```

## Write outcomes

- Acknowledged and committed: the leader has its own replica plus one follower acknowledgement, advances `commitIndex`, and only then returns `MAJORITY_COMMITTED`.
- Explicitly rejected: a non-leader returns `NOT_LEADER`; no local entry is accepted.
- Unacknowledged and uncommitted: the isolated former leader accepts the entry into replica state, but cannot get a follower acknowledgement, cannot advance `commitIndex`, and sends no successful reply.
- Acknowledged but uncommitted: forbidden by this simulator's main-path rule and reported as zero.

Commit latency starts when the benchmark actor sends a tracked write and ends when its majority-commit acknowledgement returns. Average and p95 values therefore include the simulated replication round trip plus local actor scheduling. Writes that never receive an acknowledgement have no finite commit-latency sample and are reported as outstanding work instead.

## Evidence

- The former leader's `replicaValue` and `replicaIndex` finish ahead of its committed fields.
- Its pending count matches the bounded unacknowledged batch.
- A new higher-term leader commits through the two connected nodes.
- The report contains election time, throughput, average and p95 latency, rejections, and outstanding work.

What does this section establish?

The outcome list defines exactly what each counter means. The evidence list tells a reviewer which report lines prove partition behavior and recovery.

• Save README.md.
• Confirm the file shows the Write outcomes and Evidence headings.

Does the Markdown look broken?

Check that the command has one opening code fence and one closing code fence. Confirm that every outcome starts with a single list marker.

Still seeing formatting problems? Help me fix the README Markdown structure.

A production consensus claim requires more than the behavior demonstrated here. The guarantee matrix names the tested subset and records each missing mechanism.

• Append the control-plane note and guarantee matrix to README.md by pasting:

The benchmark observer is a control plane outside the simulated node-to-node partition. Fixed demonstration timeouts improve repeatability but do not test randomized split-vote recovery. Compare runs on the same Mac because JVM warm-up and scheduling affect absolute results.

## Guarantee matrix

| Property | Status | Boundary |
|---|---|---|
| Terms and stale-term rejection | Demonstrated | In memory only. |
| One vote per term | Demonstrated | No durable vote storage. |
| Election Safety in this fixed run | Demonstrated | Fixed membership and execution model only. |
| Majority acknowledgement | Demonstrated | Success is returned only after the two-node quorum commits. |
| Minority cannot commit | Demonstrated | Isolated entries stay pending and unacknowledged. |
| Heartbeat failover | Demonstrated | Deterministic rather than randomized timeouts. |
| Log Matching | Not implemented | No `prevLogIndex`, `prevLogTerm`, or conflict truncation. |
| Leader Completeness | Not implemented | Voting omits candidate-log freshness. |
| State Machine Safety for arbitrary histories | Not claimed | Depends on omitted safety mechanisms. |
| Stable recovery, retries, snapshots, membership changes | Not implemented | One short in-memory run. |
| Deduplication and linearizable reads | Not implemented | Write benchmark only. |
| Real distributed transport | Not implemented | One JVM with simulated regions. |

Why limit the safety claim?

The simulator demonstrates majority acknowledgement and minority blocking for this fixed run. It omits the log comparison and conflict-repair mechanisms required for broader state-machine safety.

• Save README.md.
• Confirm the guarantee matrix contains both Demonstrated and Not implemented status values.

Is the table misaligned?

Check that every table row has three cells. Keep one separator row directly below the header row.

Need help with the table? Help me repair the guarantee matrix.

The final section prepares a controlled backpressure experiment. Its blank rows preserve the measured results for each window size without claiming a universal winner.

• Append the blank experiment matrix and its run commands to README.md by pasting:

## Backpressure experiment

| Window | Before throughput | After throughput | Before p95 latency | After p95 latency | Max in flight | Uncommitted outstanding | Interpretation |
|---:|---:|---:|---:|---:|---:|---:|---|
| 4 | | | | | | | |
| 16 | | | | | | | |
| 32 | | | | | | | |

```bash
jbang -Dsim.window=4 RegionalRaftSimulation.java
jbang -Dsim.window=16 RegionalRaftSimulation.java
jbang -Dsim.window=32 RegionalRaftSimulation.java
```

Choose the smallest window close to the best throughput. Explain whether larger windows increase p95 latency or uncommitted outstanding work without a proportionate throughput gain.

How will this matrix help?

Each row holds measurements from one request window. Comparing the same fields across all three runs reveals whether extra outstanding work produces a useful throughput gain.

• Save README.md.
• Confirm the experiment matrix contains blank rows for windows 4, 16, and 32.

Missing an experiment row?

Check that the table has one row for each required window. Keep the cells blank so a later controlled run supplies the values.

Need help checking the matrix? Help me compare my backpressure table with the required structure.

Your Markdown report should now match the simulator's observable behavior. Use the complete reference to catch a missing line or table cell.

✔️ Awesome, I've got everything!

Your report now defines the outcomes, records the safety boundary, and prepares the three-window experiment.

ⓧ I'd like to double check the full code

# Three-Region Raft and Actor-State Simulator

A local Java 17 simulation built with Apache Pekko Typed 1.7.1 and JBang. Three actors represent `us-east`, `eu-west`, and `ap-south` behind a latency and partition simulator.

## Run

```bash
jbang RegionalRaftSimulation.java
```

## Write outcomes

- Acknowledged and committed: the leader has its own replica plus one follower acknowledgement, advances `commitIndex`, and only then returns `MAJORITY_COMMITTED`.
- Explicitly rejected: a non-leader returns `NOT_LEADER`; no local entry is accepted.
- Unacknowledged and uncommitted: the isolated former leader accepts the entry into replica state, but cannot get a follower acknowledgement, cannot advance `commitIndex`, and sends no successful reply.
- Acknowledged but uncommitted: forbidden by this simulator's main-path rule and reported as zero.

Commit latency starts when the benchmark actor sends a tracked write and ends when its majority-commit acknowledgement returns. Average and p95 values therefore include the simulated replication round trip plus local actor scheduling. Writes that never receive an acknowledgement have no finite commit-latency sample and are reported as outstanding work instead.

## Evidence

- The former leader's `replicaValue` and `replicaIndex` finish ahead of its committed fields.
- Its pending count matches the bounded unacknowledged batch.
- A new higher-term leader commits through the two connected nodes.
- The report contains election time, throughput, average and p95 latency, rejections, and outstanding work.

The benchmark observer is a control plane outside the simulated node-to-node partition. Fixed demonstration timeouts improve repeatability but do not test randomized split-vote recovery. Compare runs on the same Mac because JVM warm-up and scheduling affect absolute results.

## Guarantee matrix

| Property | Status | Boundary |
|---|---|---|
| Terms and stale-term rejection | Demonstrated | In memory only. |
| One vote per term | Demonstrated | No durable vote storage. |
| Election Safety in this fixed run | Demonstrated | Fixed membership and execution model only. |
| Majority acknowledgement | Demonstrated | Success is returned only after the two-node quorum commits. |
| Minority cannot commit | Demonstrated | Isolated entries stay pending and unacknowledged. |
| Heartbeat failover | Demonstrated | Deterministic rather than randomized timeouts. |
| Log Matching | Not implemented | No `prevLogIndex`, `prevLogTerm`, or conflict truncation. |
| Leader Completeness | Not implemented | Voting omits candidate-log freshness. |
| State Machine Safety for arbitrary histories | Not claimed | Depends on omitted safety mechanisms. |
| Stable recovery, retries, snapshots, membership changes | Not implemented | One short in-memory run. |
| Deduplication and linearizable reads | Not implemented | Write benchmark only. |
| Real distributed transport | Not implemented | One JVM with simulated regions. |

## Backpressure experiment

| Window | Before throughput | After throughput | Before p95 latency | After p95 latency | Max in flight | Uncommitted outstanding | Interpretation |
|---:|---:|---:|---:|---:|---:|---:|---|
| 4 | | | | | | | |
| 16 | | | | | | | |
| 32 | | | | | | | |

```bash
jbang -Dsim.window=4 RegionalRaftSimulation.java
jbang -Dsim.window=16 RegionalRaftSimulation.java
jbang -Dsim.window=32 RegionalRaftSimulation.java
```

Choose the smallest window close to the best throughput. Explain whether larger windows increase p95 latency or uncommitted outstanding work without a proportionate throughput gain.

This reference is the complete README.md file. Compare its headings, tables, and command block with your saved file.

Before the final run, predict which node will finish with replica state ahead of committed state. Consider which node was isolated after the baseline phase.

• Run the documented command from the terminal in the regional-raft-simulator folder:

jbang RegionalRaftSimulation.java

What should the final run contain?

• The report includes initial and failover election durations.
• The before and after phases include acknowledged commits, throughput, average latency, and p95 latency.
• The outcome totals include explicit rejections and unacknowledged uncommitted work.
• The node table includes role, term, commit index, committed value, replica index, replica value, and pending count.

You should see acknowledgedButUncommitted=0 in the final report. The isolated former leader should finish with newer replica state and a pending count equal to the bounded window.

Does the final report look incomplete?

Compare Benchmark.java with the complete reference above. Check the printFinalReport method first if the run ends without the summary or node table.

If a metric is missing, help me trace it from collection to output.

Secret mission

Find the Best Backpressure Window

Turn your simulator into a controlled performance experiment. Compare request windows 4, 16, and 32 on the same Mac, then defend the smallest window that stays close to the best throughput without carrying unnecessary latency or uncommitted work.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely on your Mac with no ongoing costs. Decide whether to keep the files, pause your work, or delete the local resources.

Resources you used:

• The regional-raft-simulator folder with its five project files.
• Local JBang cache entries created by the project runs.

Keep everything running

No action needed. Choose this if you want to keep developing the simulator or preserve your benchmark evidence.

• Keep the regional-raft-simulator folder on your Mac.
• Retain the completed benchmark matrix in README.md.
• Add the regional-raft-simulator folder to a Git repository if you want version history.
• Leave the JBang cache in place to speed up future runs.

Pause - I'll come back to this later

Stop using the workspace for now. The actor system already terminates after each report, so no Java process remains running.

• Close the Visual Studio Code window for regional-raft-simulator.
• Close the macOS Terminal window used for the benchmark runs.
• Leave the regional-raft-simulator folder in place.
• Keep the JBang cache entries for your next session.

Delete - I don't want to use this again

Remove all project resources to start fresh. The completed benchmark matrix disappears with the project folder.

• Copy README.md to another location if you want to keep the completed benchmark matrix.
• Close the Visual Studio Code window for regional-raft-simulator.
• Switch back to the macOS Terminal window from your benchmark runs.

• Move to the folder above regional-raft-simulator by running this command:

cd ..

This moves Terminal outside the project folder. The folder can now be removed safely.

Before You Remove the Folder

Deleting this folder is permanent. Any results stored only in README.md disappear with it.

This removal only affects your local simulator files. The project created no cloud resources.

• Remove the regional-raft-simulator folder by running this command:

rm -rf regional-raft-simulator

This deletes RegionalRaftSimulation.java, SimulatedNetwork.java, RaftNode.java, Benchmark.java, and README.md with their containing folder.

• Check the parent folder in Finder.

You should no longer see the regional-raft-simulator folder. That is the local project cleared from your Mac.

JBang keeps downloaded dependencies in its cache. Clearing the cache frees local disk space.

• Clear the JBang cache by running this command:

jbang cache clear

This removes the cached JBang artifacts. Your JBang installation remains available.

Nice Work!

Nice Work!

Outstanding work! You've built a three-region simulation with Apache Pekko Typed actors that keep independent replica state behind a latency-aware network. Your benchmark report now demonstrates majority commitment, partition behavior, higher-term failover, bounded work, and explicit safety limits.

You've learned how to:

• Model three regions as independently stateful typed actors behind simulated latency. Elect leaders through terms, votes, heartbeats, and deterministic timeouts.
• Make successful acknowledgements depend on majority commitment. Show isolated former-leader writes as pending, uncommitted, and unacknowledged.
• Produce a reproducible benchmark report covering throughput, commit latency, rejected writes, outstanding work, election duration, and per-node state. Document the missing mechanisms that prevent a full Raft safety claim.
• Secret Mission: Compared request windows 4, 16, and 32 using measured throughput, p95 commit latency, maximum in-flight work, and uncommitted outstanding work. Selected the smallest window close to the best throughput from the evidence.

Ready to quiz yourself?