Feature/coordinated protocol updates execution (#790)

cli-tools/src/main/java/com/radixdlt/shell/RadixShell.java

@@ -99,13 +99,10 @@
 import com.radixdlt.p2p.transport.PeerServerBootstrap;
 import com.radixdlt.sbor.NodeSborCodecs;
 import com.radixdlt.serialization.DeserializeException;
-import com.radixdlt.serialization.Serialization;
-import com.radixdlt.sync.TransactionsAndProofReader;
 import com.radixdlt.utils.Compress;
 import com.radixdlt.utils.properties.RuntimeProperties;
 import io.reactivex.rxjava3.disposables.Disposable;
 import java.io.File;
-import java.io.IOException;
 import java.net.URI;
 import java.nio.file.Files;
 import java.util.*;
@@ -375,24 +372,6 @@ public void cleanMsgConsumers() {
       msgConsumers.clear();
     }
 
-    public void writeLedgerSyncToFile(String fileName) throws IOException {
-      final var start = System.currentTimeMillis();
-      LedgerFileSync.writeToFile(
-          fileName,
-          getInstance(Serialization.class),
-          getInstance(TransactionsAndProofReader.class));
-      final var time = System.currentTimeMillis() - start;
-      System.out.printf("Dump finished. Took %ss%n", time / 1000);
-    }
-
-    public void restoreLedgerFromFile(String fileName) throws IOException {
-      final var start = System.currentTimeMillis();
-      LedgerFileSync.restoreFromFile(
-          fileName, getInstance(Serialization.class), getInstance(new Key<>() {}));
-      final var time = System.currentTimeMillis() - start;
-      System.out.printf("Restore finished. Took %ss%n", time / 1000);
-    }
-
     @Override
     public String toString() {
       return self().toString();

common/src/main/java/com/radixdlt/lang/Option.java

@@ -93,6 +93,13 @@ default <U> Option<U> map(Func1<? super T, U> mapper) {
     return fold(t -> present(mapper.apply(t)), Option::empty);
   }
 
+  default void ifPresent(Consumer<? super T> consumer) {
+    switch (this) {
+      case Some<T> s -> consumer.accept(s.value);
+      case None<T> n -> {}
+    }
+  }
+
   /**
    * Replace current instance with the value returned by provided supplier.
    *
@@ -235,6 +242,29 @@ default Option<T> orElse(Option<T> replacement) {
     return fold(unused -> this, () -> replacement);
   }
 
+  default T orElse(T replacement) {
+    return fold(value -> value, () -> replacement);
+  }
+
+  default T orElseThrow() {
+    return orElseThrow(() -> new RuntimeException("Option value is missing"));
+  }
+
+  default T orElseThrow(Supplier<RuntimeException> exceptionSupplier) {
+    return fold(
+        value -> value,
+        () -> {
+          throw exceptionSupplier.get();
+        });
+  }
+
+  default void ifPresentOrElse(Consumer<T> consumer, Runnable emptyConsumer) {
+    switch (this) {
+      case Some<T> some -> consumer.accept(some.value);
+      case None<T> none -> emptyConsumer.run();
+    }
+  }
+
   /**
    * Return current instance if current instance is present. If current instance is empty then
    * retrieve replacement Option instance from given {@link Supplier}.

common/src/main/java/com/radixdlt/monitoring/Metrics.java

@@ -226,7 +226,11 @@ public record Pacemaker(
 
     public record SentVote(boolean isFallbackVertex, boolean isTimeout, boolean sentToAll) {}
 
-    public record Sync(Counter requestsSent, Counter requestsReceived, Counter requestTimeouts) {}
+    public record Sync(
+        Counter requestsSent,
+        Counter requestsReceived,
+        Counter requestTimeouts,
+        Counter invalidEpochInitialQcSyncStates) {}
 
     public record VertexStore(
         Gauge size, Counter forks, Counter rebuilds, Counter indirectParents) {}
@@ -245,7 +249,8 @@ public record Ledger(
       Counter syncTransactionsProcessed,
       Counter bftTransactionsProcessed,
       Timer commit,
-      Timer prepare) {}
+      Timer prepare,
+      Counter ignoredBftCommittedUpdates) {}
 
   public record LedgerSync(
       Counter validResponsesReceived,
@@ -254,7 +259,8 @@ public record LedgerSync(
       Gauge targetStateVersion, // UNTRUSTED: comes from a single peer Node and is not verified
       Gauge targetProposerTimestampEpochSecond, // UNTRUSTED: same as `targetStateVersion`
       LabelledCounter<UnexpectedSyncResponse> unexpectedResponsesReceived,
-      LabelledCounter<InvalidSyncResponse> invalidResponsesReceived) {
+      LabelledCounter<InvalidSyncResponse> invalidResponsesReceived,
+      LabelledCounter<UnfulfilledSyncRequest> unfulfilledSyncRequests) {
 
     public record UnexpectedSyncResponse(UnexpectedSyncResponseReason reason) {}
 
@@ -274,6 +280,14 @@ public enum InvalidSyncResponseReason {
       INSUFFICIENT_VALIDATOR_SET,
       MISMATCHED_SIGNATURES
     }
+
+    public record UnfulfilledSyncRequest(UnfulfilledSyncRequestReason reason) {}
+
+    public enum UnfulfilledSyncRequestReason {
+      NOTHING_TO_SERVE_AT_THE_GIVEN_STATE_VERSION,
+      REFUSED_TO_SERVE_GENESIS,
+      REFUSED_TO_SERVE_PROTOCOL_UPDATE
+    }
   }
 
   public record Mempool(Counter relaysSent) {}

core-rust-bridge/src/main/java/com/radixdlt/environment/StateManagerConfig.java

@@ -66,6 +66,7 @@
 
 import com.radixdlt.lang.Option;
 import com.radixdlt.mempool.RustMempoolConfig;
+import com.radixdlt.protocol.ProtocolConfig;
 import com.radixdlt.rev2.NetworkDefinition;
 import com.radixdlt.sbor.codec.CodecMap;
 import com.radixdlt.sbor.codec.StructCodec;
@@ -81,6 +82,7 @@ public record StateManagerConfig(
     StateHashTreeGcConfig stateHashTreeGcConfig,
     LedgerProofsGcConfig ledgerProofsGcConfig,
     LedgerSyncLimitsConfig ledgerSyncLimitsConfig,
+    ProtocolConfig protocolConfig,
     boolean noFees) {
   public static void registerCodec(CodecMap codecMap) {
     codecMap.register(