Merge branch 'main' into tendermint-main-fix-byzzfuzz-partitions

Author: Antoni Nowakowski

simulator/src/main/java/byzzbench/simulator/BaseScenario.java

@@ -166,6 +166,15 @@ public synchronized Node getNode(String nodeId) {
     @Override
     public final void setupScenario() {
         this.setup();
+
+        // sample f replicas to be faulty at start
+        List<String> replicaIds = new ArrayList<>(this.getReplicas().keySet());
+        Collections.shuffle(replicaIds);
+        int f = this.maxFaultyReplicas();
+        for (int i = 0; i < f; i++) {
+            this.markReplicaFaulty(replicaIds.get(i));
+        }
+
         this.getClients().values().forEach(Client::initialize);
         this.getNodes().values().forEach(Node::initialize);
         this.scheduler.initializeScenario(this);

simulator/src/main/java/byzzbench/simulator/Timekeeper.java

@@ -73,6 +73,11 @@ public void onEventDropped(Event event) {
         // nothing to do
     }
 
+    @Override
+    public void onEventRequeued(Event event) {
+        // nothing to do
+    }
+
     @Override
     public void onEventDelivered(Event event) {
         // check if it was a timeout

simulator/src/main/java/byzzbench/simulator/faults/factories/ByzzFuzzScenarioFaultFactory.java

@@ -37,6 +37,7 @@ public List<Fault> generateFaults(FaultContext input) {
         int d = scheduler.getNumRoundsWithNetworkFaults();
         int r = scheduler.getNumRoundsWithFaults();
         Set<String> replicaIds = scenario.getReplicas().keySet();
+        Set<String> faultyReplicaIds = scenario.getFaultyReplicaIds();
 
         // Create network faults
         for (int i = 1; i <= d; i++) {
@@ -49,7 +50,7 @@ public List<Fault> generateFaults(FaultContext input) {
         // Create process faults
         for (int i = 1; i <= c; i++) {
             int round = rand.nextInt(r) + 1;
-            String sender = replicaIds.stream().skip(rand.nextInt(replicaIds.size())).findFirst().orElseThrow();
+            String sender = input.getScenario().getFaultyReplicaIds().stream().skip(rand.nextInt(faultyReplicaIds.size())).findFirst().orElseThrow();
             Set<String> recipientIds = SetSubsets.getRandomNonEmptySubset(replicaIds);
 
             // generate process fault

simulator/src/main/java/byzzbench/simulator/scheduler/BaseScheduler.java

@@ -118,7 +118,7 @@ public List<ClientRequestEvent> getQueuedClientRequestEvents(Scenario scenario)
      * @param messageEvents The list of queued message events
      * @return The next message event
      */
-    public Event getNextMessageEvent(List<Event> messageEvents) {
+    public <T extends Event> T getNextMessageEvent(List<T> messageEvents) {
         switch (config.getScheduler().getExecutionMode()) {
             case SYNC -> {
                 return messageEvents.stream().min(Comparator.comparing(Event::getEventId)).orElseThrow();

simulator/src/main/java/byzzbench/simulator/scheduler/RandomScheduler.java

@@ -16,7 +16,7 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.Random;
-import java.util.stream.Collectors;
+import java.util.SortedSet;
 
 /**
  * A scheduler that randomly selects events to deliver, drop, mutate or timeout.
@@ -26,11 +26,14 @@
 public class RandomScheduler extends BaseScheduler {
     private final Random random = new Random();
 
-
     public RandomScheduler(ByzzBenchConfig config, MessageMutatorService messageMutatorService) {
         super(config, messageMutatorService);
     }
 
+    public <T> T getRandomElement(List<T> list) {
+        return list.get(random.nextInt(list.size()));
+    }
+
     @Override
     public String getId() {
         return "Random";
@@ -53,21 +56,24 @@ public synchronized Optional<EventDecision> scheduleNext(Scenario scenario) thro
         }
 
         List<TimeoutEvent> timeoutEvents = this.getQueuedTimeoutEvents(scenario);
-        List<Event> clientRequestEvents = availableEvents.stream().filter(ClientRequestEvent.class::isInstance).collect(Collectors.toList());
-        List<Event> messageEvents = availableEvents.stream().filter(MessageEvent.class::isInstance).collect(Collectors.toList());
+        List<Event> clientRequestEvents = availableEvents.stream().filter(ClientRequestEvent.class::isInstance).toList();
+        List<MessageEvent> messageEvents = availableEvents.stream().filter(MessageEvent.class::isInstance).map(MessageEvent.class::cast).toList();
+
+        SortedSet<String> faultyReplicaIds = scenario.getFaultyReplicaIds();
+        List<MessageEvent> mutateableMessageEvents = messageEvents.stream().filter(msg -> faultyReplicaIds.contains(msg.getSenderId())).toList();
 
         int timeoutWeight = timeoutEvents.size() * this.deliverTimeoutWeight();
         int deliverMessageWeight = messageEvents.size() * this.deliverMessageWeight();
         int deliverClientRequestWeight = clientRequestEvents.size() * this.deliverClientRequestWeight();
         int dropMessageWeight = (messageEvents.size() * this.dropMessageWeight(scenario));
-        int mutateMessageWeight = (messageEvents.size() * this.mutateMessageWeight(scenario));
+        int mutateMessageWeight = (mutateableMessageEvents.size() * this.mutateMessageWeight(scenario));
         int dieRoll = random.nextInt(timeoutWeight + deliverMessageWeight
                 + deliverClientRequestWeight + dropMessageWeight + mutateMessageWeight);
 
         // check if we should trigger a timeout
         dieRoll -= timeoutWeight;
         if (dieRoll < 0) {
-            Event timeout = timeoutEvents.get(random.nextInt(timeoutEvents.size()));
+            Event timeout = getRandomElement(timeoutEvents);
             scenario.getTransport().deliverEvent(timeout.getEventId());
             EventDecision decision = new EventDecision(EventDecision.DecisionType.DELIVERED, timeout.getEventId());
             return Optional.of(decision);
@@ -85,7 +91,7 @@ public synchronized Optional<EventDecision> scheduleNext(Scenario scenario) thro
         // check if we should target delivering a request from a client to a replica
         dieRoll -= deliverClientRequestWeight;
         if (dieRoll < 0) {
-            Event request = clientRequestEvents.get(random.nextInt(clientRequestEvents.size()));
+            Event request = getRandomElement(clientRequestEvents);
             scenario.getTransport().deliverEvent(request.getEventId());
             EventDecision decision = new EventDecision(EventDecision.DecisionType.DELIVERED, request.getEventId());
             return Optional.of(decision);
@@ -94,7 +100,7 @@ public synchronized Optional<EventDecision> scheduleNext(Scenario scenario) thro
         // check if we should drop a message sent between nodes
         dieRoll -= dropMessageWeight;
         if (dieRoll < 0) {
-            Event message = messageEvents.get(random.nextInt(messageEvents.size()));
+            Event message = getRandomElement(messageEvents);
             scenario.getTransport().dropEvent(message.getEventId());
             if(!(message instanceof GossipMessage)) {
                 EventDecision decision = new EventDecision(EventDecision.DecisionType.DROPPED, message.getEventId());
@@ -105,7 +111,7 @@ public synchronized Optional<EventDecision> scheduleNext(Scenario scenario) thro
         // check if we should mutate-and-deliver a message sent between nodes
         dieRoll -= mutateMessageWeight;
         if (dieRoll < 0) {
-            Event message = messageEvents.get(random.nextInt(messageEvents.size()));
+            Event message = getRandomElement(mutateableMessageEvents);
             List<MessageMutationFault> mutators = this.getMessageMutatorService().getMutatorsForEvent(message);
 
             if (mutators.isEmpty()) {
@@ -117,7 +123,7 @@ public synchronized Optional<EventDecision> scheduleNext(Scenario scenario) thro
             }
             scenario.getTransport().applyMutation(
                     message.getEventId(),
-                    mutators.get(random.nextInt(mutators.size())));
+                    getRandomElement(mutators));
             scenario.getTransport().deliverEvent(message.getEventId());
 
             EventDecision decision = new EventDecision(EventDecision.DecisionType.MUTATED_AND_DELIVERED, message.getEventId());

simulator/src/main/java/byzzbench/simulator/scheduler/twins/TwinsScheduler.java

@@ -47,7 +47,7 @@ public String getId() {
     @Override
     public void initializeScenario(Scenario scenario) {
         // Get the IDs of the replicas
-        List<String> replicaIds = scenario.getReplicas().keySet().stream().sorted().toList();
+        List<String> replicaIds = scenario.getFaultyReplicaIds().stream().toList();
 
         if (replicaIds.size() < numReplicas) {
             throw new IllegalArgumentException("Not enough replicas to create " + numReplicas + " twins");

simulator/src/main/java/byzzbench/simulator/transport/Transport.java

@@ -75,6 +75,8 @@ public class Transport {
      */
     @JsonIgnore
     private final List<TransportObserver> observers = new ArrayList<>();
+    @Getter
+    private boolean isGlobalStabilizationTime = false;
 
     /**
      * Adds an observer to the transport layer.
@@ -276,13 +278,19 @@ public synchronized void deliverEvent(long eventId) throws Exception {
      * @param eventId The ID of the message to drop.
      */
     public synchronized void dropEvent(long eventId) {
+        // Check if it is GST - no more dropping
+        if (this.isGlobalStabilizationTime) {
+            throw new IllegalStateException("Cannot drop events during GST");
+        }
+
         // check if event is a message
         Event e = events.get(eventId);
 
         if (e.getStatus() != Event.Status.QUEUED) {
             return;
 //            throw new IllegalArgumentException("Event not found or not in QUEUED state");
         }
+
         e.setStatus(Event.Status.DROPPED);
         this.observers.forEach(o -> o.onEventDropped(e));
         log.info("Dropped: " + e);
@@ -329,6 +337,13 @@ public synchronized void applyMutation(long eventId, Fault fault) {
                     "Event %d is not a message - cannot mutate it.", eventId));
         }
 
+        // check if sender is faulty
+        if (!this.scenario.isFaultyReplica(m.getSenderId())) {
+            throw new IllegalArgumentException(
+                    String.format("Cannot mutate message: sender %s is not marked as faulty", m.getSenderId())
+            );
+        }
+
         // create input for the fault
         FaultContext input = new FaultContext(this.scenario, e);
 
@@ -494,4 +509,32 @@ public synchronized List<Fault> getEnabledNetworkFaults() {
     public synchronized Fault getNetworkFault(String faultId) {
         return this.networkFaults.get(faultId);
     }
+
+    /**
+     * Simulates GST event, according to the partial-synchrony model:
+     * <ul>
+     *   <li>All dropped messages are re-queued</li>
+     *   <li>Prevents further dropping of messages</li>
+     *   <li>All network partitions are healed</li>
+     *   <li>Prevents further network partitions</li>
+     * </ul>
+     */
+    public void globalStabilizationTime() {
+        this.isGlobalStabilizationTime = true;
+
+        // re-queue all dropped messages
+        this.events.values().stream()
+                .filter(e -> e.getStatus() == Event.Status.DROPPED)
+                .forEach(e -> {
+                    e.setStatus(Event.Status.QUEUED);
+                    this.observers.forEach(o -> o.onEventRequeued(e));
+                });
+
+        // clear all network faults
+        // XXX: Is this the right thing to do?
+        this.networkFaults.clear();
+
+        // heal all partitions
+        this.router.resetPartitions();
+    }
 }

simulator/src/main/java/byzzbench/simulator/transport/TransportObserver.java

@@ -17,6 +17,13 @@ public interface TransportObserver {
      */
     void onEventDropped(Event event);
 
+    /**
+     * Called when the status of an event changes from {@link Event.Status#DROPPED} to {@link Event.Status#QUEUED}.
+     *
+     * @param event The event that was re-queued after being previously dropped.
+     */
+    void onEventRequeued(Event event);
+
     /**
      * Called when the status of an event changes to {@link Event.Status#DELIVERED}.
      *