Hex check (#162)

* no need to pass prioritized executor.

Hides this. less chance for errors.

* clean up

* rename utils.proto

* HexBloom for CheckpointState crown

* Validate crown upon checkpoint assembly

* use HexBloom for checkpoint validation, no hash list.

Removed the list of segment hashes for the checkpoint.  Use the HexBloom crown for the validation of the segments.

* use HexBloom for checkpoint validation, no hash list.

Removed the list of segment hashes for the checkpoint.  Use the HexBloom crown for the validation of the segments.

* add compactWrapped() calculation to Accumulator

choam/src/main/java/com/salesforce/apollo/choam/CHOAM.java

@@ -8,15 +8,15 @@
 
 import com.chiralbehaviors.tron.Fsm;
 import com.google.common.base.Function;
-import com.google.protobuf.Any;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.google.protobuf.Message;
 import com.salesfoce.apollo.choam.proto.*;
 import com.salesfoce.apollo.choam.proto.SubmitResult.Result;
+import com.salesfoce.apollo.cryptography.proto.PubKey;
 import com.salesfoce.apollo.messaging.proto.AgedMessageOrBuilder;
-import com.salesfoce.apollo.utils.proto.PubKey;
 import com.salesforce.apollo.archipelago.RouterImpl.CommonCommunications;
+import com.salesforce.apollo.bloomFilters.BloomFilter;
 import com.salesforce.apollo.choam.comm.*;
 import com.salesforce.apollo.choam.fsm.Combine;
 import com.salesforce.apollo.choam.fsm.Combine.Merchantile;
@@ -25,16 +25,14 @@
 import com.salesforce.apollo.choam.support.HashedCertifiedBlock.NullBlock;
 import com.salesforce.apollo.crypto.*;
 import com.salesforce.apollo.crypto.Signer.SignerImpl;
-import com.salesforce.apollo.ethereal.Ethereal;
 import com.salesforce.apollo.membership.Context;
 import com.salesforce.apollo.membership.GroupIterator;
 import com.salesforce.apollo.membership.Member;
+import com.salesforce.apollo.membership.RoundScheduler;
 import com.salesforce.apollo.membership.messaging.rbc.ReliableBroadcaster;
 import com.salesforce.apollo.membership.messaging.rbc.ReliableBroadcaster.MessageAdapter;
 import com.salesforce.apollo.membership.messaging.rbc.ReliableBroadcaster.Msg;
-import com.salesforce.apollo.membership.RoundScheduler;
 import com.salesforce.apollo.utils.Utils;
-import com.salesforce.apollo.bloomFilters.BloomFilter;
 import io.grpc.StatusRuntimeException;
 import org.h2.mvstore.MVMap;
 import org.joou.ULong;
@@ -70,7 +68,6 @@ public class CHOAM {
     private final        AtomicReference<HashedCertifiedBlock>                 checkpoint            = new AtomicReference<>();
     private final        ReliableBroadcaster                                   combine;
     private final        CommonCommunications<Terminal, Concierge>             comm;
-    private final        ThreadPoolExecutor                                    consumer;
     private final        AtomicReference<Committee>                            current               = new AtomicReference<>();
     private final        ExecutorService                                       executions;
     private final        AtomicReference<CompletableFuture<SynchronizedState>> futureBootstrap       = new AtomicReference<>();
@@ -136,40 +133,44 @@ public CHOAM(Parameters params) {
                                             params.context().timeToLive());
         combine.register(i -> roundScheduler.tick());
         session = new Session(params, service());
-        consumer = Ethereal.consumer("CHOAM" + params.member().getId() + params.context().getId());
     }
 
-    public static Checkpoint checkpoint(DigestAlgorithm algo, File state, int segmentSize) {
-        Digest stateHash = algo.getOrigin();
+    public static Checkpoint checkpoint(DigestAlgorithm algo, File state, int segmentSize, Digest initial) {
+        assert segmentSize > 0 : "segment size must be > 0 : " + segmentSize;
         long length = 0;
         if (state != null) {
-            try (FileInputStream fis = new FileInputStream(state)) {
-                stateHash = algo.digest(fis);
-            } catch (IOException e) {
-                log.error("Invalid checkpoint!", e);
-                return null;
-            }
             length = state.length();
         }
+        int count = (int) (length / segmentSize);
+        if (length != 0 && count * segmentSize < length) {
+            count++;
+        }
+        var accumulator = new HexBloom.HexAccumulator(count, 2, initial);
         Checkpoint.Builder builder = Checkpoint.newBuilder()
+                                               .setCount(count)
                                                .setByteSize(length)
-                                               .setSegmentSize(segmentSize)
-                                               .setStateHash(stateHash.toDigeste());
+                                               .setSegmentSize(segmentSize);
+
         if (state != null) {
             byte[] buff = new byte[segmentSize];
             try (FileInputStream fis = new FileInputStream(state)) {
                 for (int read = fis.read(buff); read > 0; read = fis.read(buff)) {
                     ByteString segment = ByteString.copyFrom(buff, 0, read);
-                    builder.addSegments(algo.digest(segment).toDigeste());
+                    accumulator.add(algo.digest(segment));
                 }
             } catch (IOException e) {
                 log.error("Invalid checkpoint!", e);
                 return null;
             }
         }
-        log.info("Checkpoint length: {} segment size: {} count: {} stateHash: {}", length, segmentSize,
-                 builder.getSegmentsCount(), stateHash);
-        return builder.build();
+        var crown = accumulator.build();
+        log.info("Checkpoint length: {} segment size: {} count: {} crown: {} initial: {}", length, segmentSize,
+                 builder.getCount(), crown, initial);
+        var cp = builder.setCrown(crown.toHexBloome()).build();
+
+        var deserialized = HexBloom.from(cp.getCrown());
+        log.info("Deserialized checkpoint crown: {} initial: {}", deserialized, initial);
+        return cp;
     }
 
     public static Block genesis(Digest id, Map<Member, Join> joins, HashedBlock head, Context<Member> context,
@@ -384,14 +385,14 @@ private Block checkpoint() {
             transitions.fail();
             return null;
         }
-        Checkpoint cp = checkpoint(params.digestAlgorithm(), state, params.checkpointSegmentSize());
+        final HashedBlock c = checkpoint.get();
+        Checkpoint cp = checkpoint(params.digestAlgorithm(), state, params.checkpointSegmentSize(), c.hash);
         if (cp == null) {
             transitions.fail();
             return null;
         }
 
         final HashedCertifiedBlock v = view.get();
-        final HashedBlock c = checkpoint.get();
         final Block block = Block.newBuilder()
                                  .setHeader(
                                  buildHeader(params.digestAlgorithm(), cp, lb.hash, lb.height().add(1), c.height(),
@@ -771,7 +772,7 @@ private Function<SubmittedTransaction, SubmitResult> service() {
     private Digest signatureHash(ByteString any) {
         CertifiedBlock cb;
         try {
-            cb =  CertifiedBlock.parseFrom(any);
+            cb = CertifiedBlock.parseFrom(any);
         } catch (InvalidProtocolBufferException e) {
             throw new IllegalStateException(e);
         }
@@ -936,6 +937,10 @@ private void synchronizedProcess(CertifiedBlock certifiedBlock) {
         pending.add(hcb);
     }
 
+    private String getLabel() {
+        return "CHOAM" + params.member().getId() + params.context().getId();
+    }
+
     public interface BlockProducer {
         Block checkpoint();
 
@@ -1199,7 +1204,7 @@ private class Associate extends Administration {
                       params.member().getId());
             Signer signer = new SignerImpl(nextView.consensusKeyPair.getPrivate());
             viewContext = new ViewContext(context, params, signer, validators, constructBlock());
-            producer = new Producer(viewContext, head.get(), checkpoint.get(), comm, consumer);
+            producer = new Producer(viewContext, head.get(), checkpoint.get(), comm, getLabel());
             producer.start();
         }
 
@@ -1244,7 +1249,7 @@ private Formation() {
                           params.member().getId());
                 Signer signer = new SignerImpl(c.consensusKeyPair.getPrivate());
                 ViewContext vc = new GenesisContext(formation, params, signer, constructBlock());
-                assembly = new GenesisAssembly(vc, comm, next.get().member, consumer);
+                assembly = new GenesisAssembly(vc, comm, next.get().member, getLabel());
                 nextViewId.set(params.genesisViewId());
             } else {
                 log.trace("No formation on: {}", params.member().getId());

choam/src/main/java/com/salesforce/apollo/choam/GenesisAssembly.java

@@ -10,7 +10,7 @@
 import com.google.protobuf.ByteString;
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.salesfoce.apollo.choam.proto.*;
-import com.salesfoce.apollo.utils.proto.PubKey;
+import com.salesfoce.apollo.cryptography.proto.PubKey;
 import com.salesforce.apollo.archipelago.RouterImpl.CommonCommunications;
 import com.salesforce.apollo.choam.comm.Terminal;
 import com.salesforce.apollo.choam.fsm.Genesis;
@@ -66,7 +66,7 @@ public class GenesisAssembly implements Genesis {
     private volatile     HashedBlock           reconfiguration;
 
     public GenesisAssembly(ViewContext vc, CommonCommunications<Terminal, ?> comms, ViewMember genesisMember,
-                           ThreadPoolExecutor executor) {
+                           String label) {
         view = vc;
         ds = new OneShot();
         nextAssembly = Committee.viewMembersOf(view.context().getId(), params().context())
@@ -103,7 +103,7 @@ public GenesisAssembly(ViewContext vc, CommonCommunications<Terminal, ?> comms,
         config.setLabel("Genesis Assembly" + view.context().getId() + " on: " + params().member().getId());
         controller = new Ethereal(config.build(), params().producer().maxBatchByteSize(), dataSource(),
                                   (preblock, last) -> transitions.process(preblock, last),
-                                  epoch -> transitions.nextEpoch(epoch), executor);
+                                  epoch -> transitions.nextEpoch(epoch), label);
         coordinator = new ChRbcGossip(reContext, params().member(), controller.processor(), params().communications(),
                                       params().metrics() == null ? null : params().metrics().getGensisMetrics());
         log.debug("Genesis Assembly: {} recontext: {} next assembly: {} on: {}", view.context().getId(),

choam/src/main/java/com/salesforce/apollo/choam/Parameters.java

@@ -49,7 +49,7 @@
  */
 public record Parameters(Parameters.RuntimeParameters runtime, ReliableBroadcaster.Parameters combine,
                          Duration gossipDuration, int maxCheckpointSegments, Duration submitTimeout,
-                         Digest genesisViewId, int checkpointBlockDelta, DigestAlgorithm digestAlgorithm,
+                         Digest genesisViewId, int checkpointBlockDelta, int crowns, DigestAlgorithm digestAlgorithm,
                          SignatureAlgorithm viewSigAlgorithm, int synchronizationCycles, int regenerationCycles,
                          Parameters.BootstrapParameters bootstrap, Parameters.ProducerParameters producer,
                          Parameters.MvStoreBuilder mvBuilder, Parameters.LimiterBuilder txnLimiterBuilder,
@@ -677,12 +677,21 @@ public static class Builder implements Cloneable {
         private int                              synchronizationCycles = 10;
         private LimiterBuilder                   txnLimiterBuilder     = new LimiterBuilder();
         private SignatureAlgorithm               viewSigAlgorithm      = SignatureAlgorithm.DEFAULT;
+        private int                              crowns                = 2;
+
+        public int getCrowns() {
+            return crowns;
+        }
+
+        public void setCrowns(int crowns) {
+            this.crowns = crowns;
+        }
 
         public Parameters build(RuntimeParameters runtime) {
             return new Parameters(runtime, combine, gossipDuration, maxCheckpointSegments, submitTimeout, genesisViewId,
-                                  checkpointBlockDelta, digestAlgorithm, viewSigAlgorithm, synchronizationCycles,
-                                  regenerationCycles, bootstrap, producer, mvBuilder, txnLimiterBuilder, submitPolicy,
-                                  checkpointSegmentSize, drainPolicy);
+                                  checkpointBlockDelta, crowns, digestAlgorithm, viewSigAlgorithm,
+                                  synchronizationCycles, regenerationCycles, bootstrap, producer, mvBuilder,
+                                  txnLimiterBuilder, submitPolicy, checkpointSegmentSize, drainPolicy);
         }
 
         @Override

choam/src/main/java/com/salesforce/apollo/choam/Producer.java

@@ -184,7 +184,7 @@ public void startProduction() {
     private final ViewContext                       view;
 
     public Producer(  ViewContext view, HashedBlock lastBlock, HashedBlock checkpoint,
-                    CommonCommunications<Terminal, ?> comms, ThreadPoolExecutor consumer) {
+                    CommonCommunications<Terminal, ?> comms, String label) {
         assert view != null;
         this.view = view;
         this.previousBlock.set(lastBlock);
@@ -227,7 +227,7 @@ public Producer(  ViewContext view, HashedBlock lastBlock, HashedBlock checkpoin
         var producerMetrics = params().metrics() == null ? null : params().metrics().getProducerMetrics();
         controller = new Ethereal(config.build(), params().producer().maxBatchByteSize() + (8 * 1024), ds,
                                   (preblock, last) -> transitions.create(preblock, last), epoch -> newEpoch(epoch),
-                                  consumer);
+                                  label);
         coordinator = new ChRbcGossip(view.context(), params().member(), controller.processor(),
                                       params().communications(), producerMetrics);
         log.debug("Roster for: {} is: {} on: {}", getViewId(), view.roster(), params().member().getId());

choam/src/main/java/com/salesforce/apollo/choam/ViewAssembly.java

@@ -8,7 +8,7 @@
 
 import com.chiralbehaviors.tron.Fsm;
 import com.salesfoce.apollo.choam.proto.*;
-import com.salesfoce.apollo.utils.proto.PubKey;
+import com.salesfoce.apollo.cryptography.proto.PubKey;
 import com.salesforce.apollo.archipelago.RouterImpl.CommonCommunications;
 import com.salesforce.apollo.choam.comm.Terminal;
 import com.salesforce.apollo.choam.fsm.Reconfiguration;

choam/src/main/java/com/salesforce/apollo/choam/comm/TerminalServer.java

@@ -14,7 +14,7 @@
 import com.salesfoce.apollo.choam.proto.Synchronize;
 import com.salesfoce.apollo.choam.proto.TerminalGrpc.TerminalImplBase;
 import com.salesfoce.apollo.choam.proto.ViewMember;
-import com.salesfoce.apollo.utils.proto.Digeste;
+import com.salesfoce.apollo.cryptography.proto.Digeste;
 import com.salesforce.apollo.archipelago.RoutableService;
 import com.salesforce.apollo.choam.support.ChoamMetrics;
 import com.salesforce.apollo.crypto.Digest;

choam/src/main/java/com/salesforce/apollo/choam/support/Bootstrapper.java

@@ -10,18 +10,19 @@
 import com.google.common.collect.TreeMultiset;
 import com.salesfoce.apollo.choam.proto.*;
 import com.salesforce.apollo.archipelago.RouterImpl.CommonCommunications;
+import com.salesforce.apollo.bloomFilters.BloomFilter;
+import com.salesforce.apollo.bloomFilters.BloomFilter.ULongBloomFilter;
 import com.salesforce.apollo.choam.Parameters;
 import com.salesforce.apollo.choam.comm.Concierge;
 import com.salesforce.apollo.choam.comm.Terminal;
 import com.salesforce.apollo.crypto.Digest;
 import com.salesforce.apollo.crypto.DigestAlgorithm;
+import com.salesforce.apollo.crypto.HexBloom;
 import com.salesforce.apollo.membership.Member;
 import com.salesforce.apollo.ring.RingCommunications;
 import com.salesforce.apollo.ring.RingIterator;
 import com.salesforce.apollo.utils.Entropy;
 import com.salesforce.apollo.utils.Pair;
-import com.salesforce.apollo.bloomFilters.BloomFilter;
-import com.salesforce.apollo.bloomFilters.BloomFilter.ULongBloomFilter;
 import org.joou.ULong;
 import org.joou.Unsigned;
 import org.slf4j.Logger;
@@ -124,8 +125,9 @@ private void checkpointCompletion(int threshold, Initial mostRecent) {
         store.put(checkpointView);
         assert !checkpointView.height()
                               .equals(Unsigned.ulong(0)) : "Should not attempt when bootstrapping from genesis";
-        log.info("Assembling from checkpoint: {}:{} on: {}", checkpoint.height(), checkpoint.hash,
-                 params.member().getId());
+        var diadem = HexBloom.from(checkpoint.block.getCheckpoint().getCrown());
+        log.info("Assembling from checkpoint: {}:{} crown: {} last cp: {} on: {}", checkpoint.height(), checkpoint.hash,
+                 diadem, Digest.from(checkpoint.block.getHeader().getLastCheckpointHash()), params.member().getId());
 
         CheckpointAssembler assembler = new CheckpointAssembler(params.gossipDuration(), checkpoint.height(),
                                                                 checkpoint.block.getCheckpoint(), params.member(),
@@ -134,17 +136,18 @@ private void checkpointCompletion(int threshold, Initial mostRecent) {
 
         // assemble the checkpoint
         checkpointAssembled = assembler.assemble(scheduler, params.gossipDuration()).whenComplete((cps, t) -> {
-            log.info("Restored checkpoint: {} on: {}", checkpoint.height(), params.member().getId());
+            if (!cps.validate(diadem, Digest.from(checkpoint.block.getHeader().getLastCheckpointHash()))) {
+                throw new IllegalStateException("Cannot validate checkpoint: " + checkpoint.height());
+            }
+            log.info("Restored checkpoint: {} diadem: {} on: {}", checkpoint.height(), diadem, params.member().getId());
             checkpointState = cps;
         });
         // reconstruct chain to genesis
         mostRecent.getViewChainList()
                   .stream()
                   .filter(cb -> cb.getBlock().hasReconfigure())
                   .map(cb -> new HashedCertifiedBlock(params.digestAlgorithm(), cb))
-                  .forEach(reconfigure -> {
-                      store.put(reconfigure);
-                  });
+                  .forEach(reconfigure -> store.put(reconfigure));
         scheduleViewChainCompletion(new AtomicReference<>(checkpointView.height()), ULong.valueOf(0));
     }