BEEFY consensus message scale classes

src/main/java/com/limechain/network/protocol/beefy/messages/consensus/BeefyConsensusMessage.java

@@ -0,0 +1,16 @@
+package com.limechain.network.protocol.beefy.messages.consensus;
+
+import lombok.Data;
+
+import java.math.BigInteger;
+import java.util.List;
+
+@Data
+public class BeefyConsensusMessage {
+    private BeefyConsensusMessageFormat format;
+    private List<byte[]> authorityPublicKeys;
+    private BigInteger authoritySetId;
+    private BigInteger disabledAuthority;
+    // The 32-byte Merkle Mountain Range (MMR) root payload hash.
+    private byte[] mmrRootHash;
+}

src/main/java/com/limechain/network/protocol/beefy/messages/consensus/BeefyConsensusMessageFormat.java

@@ -0,0 +1,23 @@
+package com.limechain.network.protocol.beefy.messages.consensus;
+
+import lombok.Getter;
+
+@Getter
+public enum BeefyConsensusMessageFormat {
+    BEEFY_CHANGED_AUTHORITIES(1), BEEFY_ON_DISABLED(2), BEEFY_MMR_ROOT(3);
+
+    private final int format;
+
+    BeefyConsensusMessageFormat(int format) {
+        this.format = format;
+    }
+
+    public static BeefyConsensusMessageFormat fromFormat(byte format) {
+        for (BeefyConsensusMessageFormat messageFormat : values()) {
+            if (messageFormat.getFormat() == format) {
+                return messageFormat;
+            }
+        }
+        throw new IllegalArgumentException("Unknown beefy consensus message format: " + format);
+    }
+}

src/main/java/com/limechain/network/protocol/beefy/messages/consensus/BeefyConsensusMessageReader.java

@@ -0,0 +1,42 @@
+package com.limechain.network.protocol.beefy.messages.consensus;
+
+import io.emeraldpay.polkaj.scale.ScaleCodecReader;
+import io.emeraldpay.polkaj.scale.ScaleReader;
+import io.emeraldpay.polkaj.scale.reader.ListReader;
+import io.emeraldpay.polkaj.scale.reader.UInt64Reader;
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import java.util.List;
+
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class BeefyConsensusMessageReader implements ScaleReader<BeefyConsensusMessage> {
+
+    private static final BeefyConsensusMessageReader INSTANCE = new BeefyConsensusMessageReader();
+    public static final int ECDSA_PUBLIC_KEY_LENGTH = 33;
+
+    public static BeefyConsensusMessageReader getInstance() {
+        return INSTANCE;
+    }
+
+    @Override
+    public BeefyConsensusMessage read(ScaleCodecReader reader) {
+
+        BeefyConsensusMessage beefyConsensusMessage = new BeefyConsensusMessage();
+        BeefyConsensusMessageFormat format = BeefyConsensusMessageFormat.fromFormat(reader.readByte());
+        beefyConsensusMessage.setFormat(format);
+
+        switch (format) {
+            case BEEFY_CHANGED_AUTHORITIES -> {
+                List<byte[]> authorityPublicKeys = new ListReader<>(
+                        rdr -> rdr.readByteArray(ECDSA_PUBLIC_KEY_LENGTH)).read(reader);
+                beefyConsensusMessage.setAuthorityPublicKeys(authorityPublicKeys);
+                beefyConsensusMessage.setAuthoritySetId(new UInt64Reader().read(reader));
+            }
+            case BEEFY_ON_DISABLED -> beefyConsensusMessage.setDisabledAuthority(new UInt64Reader().read(reader));
+            case BEEFY_MMR_ROOT -> beefyConsensusMessage.setMmrRootHash(reader.readUint256());
+        }
+
+        return beefyConsensusMessage;
+    }
+}

src/main/java/com/limechain/network/protocol/grandpa/GrandpaMessageHandler.java

@@ -494,9 +494,10 @@ private void processNeighbourUpdates(SyncMessage.BlockData block) {
 
         stateManager.getSyncState().finalizeHeader(header);
 
-        DigestHelper.getGrandpaConsensusMessage(header.getDigest())
-                .ifPresent(cm -> stateManager.getGrandpaSetState()
-                        .handleGrandpaConsensusMessage(cm, header.getBlockNumber()));
+        DigestHelper.getGrandpaConsensusMessages(header.getDigest())
+                .forEach(cm -> stateManager.getGrandpaSetState().handleGrandpaConsensusMessage(
+                        cm, header.getBlockNumber())
+                );
 
         // Executes scheduled or forced authority changes for the last finalized block.
         boolean changeInAuthoritySet = stateManager.getGrandpaSetState().handleAuthoritySetChange(

src/main/java/com/limechain/network/protocol/warp/DigestHelper.java

@@ -4,6 +4,8 @@
 import com.limechain.babe.consensus.scale.BabeConsensusMessageReader;
 import com.limechain.babe.predigest.BabePreDigest;
 import com.limechain.babe.predigest.scale.PreDigestReader;
+import com.limechain.network.protocol.beefy.messages.consensus.BeefyConsensusMessage;
+import com.limechain.network.protocol.beefy.messages.consensus.BeefyConsensusMessageReader;
 import com.limechain.network.protocol.grandpa.messages.consensus.GrandpaConsensusMessage;
 import com.limechain.network.protocol.grandpa.messages.consensus.GrandpaConsensusMessageReader;
 import com.limechain.network.protocol.warp.dto.BlockHeader;
@@ -17,30 +19,41 @@
 import lombok.NoArgsConstructor;
 
 import java.util.Arrays;
+import java.util.List;
 import java.util.Optional;
+import java.util.stream.Collectors;
 
 /**
  * Helper class for processing different types of header digests
  */
 @NoArgsConstructor(access = AccessLevel.PRIVATE)
 public class DigestHelper {
 
-    public static Optional<BabeConsensusMessage> getBabeConsensusMessage(HeaderDigest[] headerDigests) {
+    public static List<BabeConsensusMessage> getBabeConsensusMessages(HeaderDigest[] headerDigests) {
         return Arrays.stream(headerDigests)
                 .filter(headerDigest -> DigestType.CONSENSUS_MESSAGE.equals(headerDigest.getType()) &&
                         ConsensusEngine.BABE.equals(headerDigest.getId()))
-                .findFirst()
                 .map(HeaderDigest::getMessage)
-                .map(message -> ScaleUtils.Decode.decode(message, BabeConsensusMessageReader.getInstance()));
+                .map(message -> ScaleUtils.Decode.decode(message, BabeConsensusMessageReader.getInstance()))
+                .collect(Collectors.toList());
     }
 
-    public static Optional<GrandpaConsensusMessage> getGrandpaConsensusMessage(HeaderDigest[] headerDigests) {
+    public static List<GrandpaConsensusMessage> getGrandpaConsensusMessages(HeaderDigest[] headerDigests) {
         return Arrays.stream(headerDigests)
                 .filter(headerDigest -> DigestType.CONSENSUS_MESSAGE.equals(headerDigest.getType()) &&
                         ConsensusEngine.GRANDPA.equals(headerDigest.getId()))
-                .findFirst()
                 .map(HeaderDigest::getMessage)
-                .map(message -> ScaleUtils.Decode.decode(message, GrandpaConsensusMessageReader.getInstance()));
+                .map(message -> ScaleUtils.Decode.decode(message, GrandpaConsensusMessageReader.getInstance()))
+                .collect(Collectors.toList());
+    }
+
+    public static List<BeefyConsensusMessage> getBeefyConsensusMessages(HeaderDigest[] headerDigests) {
+        return Arrays.stream(headerDigests)
+                .filter(headerDigest -> DigestType.CONSENSUS_MESSAGE.equals(headerDigest.getType()) &&
+                        ConsensusEngine.BEEFY.equals(headerDigest.getId()))
+                .map(HeaderDigest::getMessage)
+                .map(message -> ScaleUtils.Decode.decode(message, BeefyConsensusMessageReader.getInstance()))
+                .collect(Collectors.toList());
     }
 
     public static Optional<BabePreDigest> getBabePreRuntimeDigest(HeaderDigest[] headerDigests) {

src/main/java/com/limechain/storage/block/BlockHandler.java

@@ -154,21 +154,27 @@ public void addBlockToTree(Block block, Instant arrivalTime) {
 
         EpochState epochState = stateManager.getEpochState();
         if (epochState.isInitialized()) {
-            asyncExecutor.executeAndForget(() -> DigestHelper.getBabeConsensusMessage(header.getDigest())
-                    .ifPresent(cm -> {
+            DigestHelper.getBabeConsensusMessages(header.getDigest())
+                    .forEach(cm -> {
                         stateManager.getEpochState().updateNextEpochConfig(cm);
                         log.fine(String.format("Updated epoch block config: %s", cm.getFormat().toString()));
-                    }));
+                    });
         }
 
         GrandpaSetState grandpaSetState = stateManager.getGrandpaSetState();
         if (grandpaSetState.isInitialized()) {
-            asyncExecutor.executeAndForget(() -> DigestHelper.getGrandpaConsensusMessage(header.getDigest())
-                    .ifPresent(cm ->
-                            grandpaSetState.handleGrandpaConsensusMessage(cm, header.getBlockNumber())
-                    ));
+            DigestHelper.getGrandpaConsensusMessages(header.getDigest())
+                    .forEach(cm -> grandpaSetState.handleGrandpaConsensusMessage(
+                            cm, header.getBlockNumber())
+                    );
 
             grandpaSetState.handleAuthoritySetChange(header.getBlockNumber());
+
+            DigestHelper.getBeefyConsensusMessages(header.getDigest())
+                    .forEach(cm -> {
+                                //Todo: handleBeefyConsensusMessage
+                            }
+                    );
         }
     }
 

src/main/java/com/limechain/sync/warpsync/action/VerifyJustificationAction.java

@@ -61,7 +61,7 @@ public void handle(WarpSyncMachine sync) {
 
             WarpSyncFragment fragment = sync.getFragmentsQueue().poll();
             log.log(Level.INFO, "Verifying justification...");
-            
+
             if (fragment == null) {
                 throw new JustificationVerificationException("No such fragment");
             }
@@ -83,9 +83,10 @@ public void handle(WarpSyncMachine sync) {
     private void handleAuthorityChanges(WarpSyncFragment fragment) {
         BlockHeader header = fragment.getHeader();
 
-        DigestHelper.getGrandpaConsensusMessage(header.getDigest())
-                .ifPresent(cm -> stateManager.getGrandpaSetState()
-                        .handleGrandpaConsensusMessage(cm, header.getBlockNumber()));
+        DigestHelper.getGrandpaConsensusMessages(header.getDigest())
+                .forEach(cm -> stateManager.getGrandpaSetState().handleGrandpaConsensusMessage(
+                        cm, header.getBlockNumber())
+                );
 
         SyncState syncState = stateManager.getSyncState();
         log.log(Level.INFO, "Verified justification. Block hash is now at #"

src/test/java/com/limechain/network/protocol/warp/DigestHelperTest.java

@@ -1,7 +1,9 @@
 package com.limechain.network.protocol.warp;
 
+import com.limechain.babe.consensus.BabeConsensusMessage;
 import com.limechain.babe.consensus.BabeConsensusMessageFormat;
 import com.limechain.babe.predigest.PreDigestType;
+import com.limechain.network.protocol.grandpa.messages.consensus.GrandpaConsensusMessage;
 import com.limechain.network.protocol.grandpa.messages.consensus.GrandpaConsensusMessageFormat;
 import com.limechain.network.protocol.warp.dto.BlockHeader;
 import com.limechain.network.protocol.warp.dto.ConsensusEngine;
@@ -25,7 +27,7 @@
 class DigestHelperTest {
 
     @Test
-    void getBabeConsensusMessageTest() {
+    void getBabeConsensusMessagesTest() {
         HeaderDigest consensusDigest = new HeaderDigest();
         consensusDigest.setId(ConsensusEngine.BABE);
         consensusDigest.setType(DigestType.CONSENSUS_MESSAGE);
@@ -35,21 +37,21 @@ void getBabeConsensusMessageTest() {
         message[0] = 2;
         consensusDigest.setMessage(message);
 
-        HeaderDigest[] headerDigests = new HeaderDigest[] {consensusDigest};
-        var optResult = DigestHelper.getBabeConsensusMessage(headerDigests);
+        HeaderDigest[] headerDigests = new HeaderDigest[]{consensusDigest};
+        var result = DigestHelper.getBabeConsensusMessages(headerDigests);
 
-        assertTrue(optResult.isPresent());
+        assertEquals(1, result.size());
 
-        var result = optResult.get();
-        assertEquals(BabeConsensusMessageFormat.DISABLED_AUTHORITY, result.getFormat());
-        assertEquals(BigInteger.ZERO, result.getDisabledAuthority());
-        assertNull(result.getNextEpochData());
-        assertNull(result.getNextEpochDescriptor());
+        BabeConsensusMessage firstResult = result.getFirst();
+        assertEquals(BabeConsensusMessageFormat.DISABLED_AUTHORITY, firstResult.getFormat());
+        assertEquals(BigInteger.ZERO, firstResult.getDisabledAuthority());
+        assertNull(firstResult.getNextEpochData());
+        assertNull(firstResult.getNextEpochDescriptor());
     }
 
     @Test
-    void getBabeConsensusMessageWithoutSuchDigestInHeadersTest() {
-        var optResult = DigestHelper.getBabeConsensusMessage(new HeaderDigest[0]);
+    void getBabeConsensusMessagesWithoutSuchDigestInHeadersTest() {
+        var optResult = DigestHelper.getBabeConsensusMessages(new HeaderDigest[0]);
         assertTrue(optResult.isEmpty());
     }
 
@@ -64,20 +66,20 @@ void getGrandpaConsensusMessageTest() {
         message[0] = 3;
         consensusDigest.setMessage(message);
 
-        HeaderDigest[] headerDigests = new HeaderDigest[] {consensusDigest};
-        var optResult = DigestHelper.getGrandpaConsensusMessage(headerDigests);
+        HeaderDigest[] headerDigests = new HeaderDigest[]{consensusDigest};
+        var result = DigestHelper.getGrandpaConsensusMessages(headerDigests);
 
-        assertTrue(optResult.isPresent());
+        assertEquals(1, result.size());
 
-        var result = optResult.get();
-        assertEquals(GrandpaConsensusMessageFormat.GRANDPA_ON_DISABLED, result.getFormat());
-        assertEquals(BigInteger.ZERO, result.getDisabledAuthority());
-        assertNull(result.getAuthorities());
+        GrandpaConsensusMessage firstResult = result.getFirst();
+        assertEquals(GrandpaConsensusMessageFormat.GRANDPA_ON_DISABLED, firstResult.getFormat());
+        assertEquals(BigInteger.ZERO, firstResult.getDisabledAuthority());
+        assertNull(firstResult.getAuthorities());
     }
 
     @Test
     void getGrandpaConsensusMessageWithoutSuchDigestInHeadersTest() {
-        var optResult = DigestHelper.getGrandpaConsensusMessage(new HeaderDigest[0]);
+        var optResult = DigestHelper.getGrandpaConsensusMessages(new HeaderDigest[0]);
         assertTrue(optResult.isEmpty());
     }
 
@@ -97,7 +99,7 @@ void getBabePreRuntimeDigestForPrimarySlotTest() {
         message[0] = 1;
         consensusDigest.setMessage(message);
 
-        HeaderDigest[] headerDigests = new HeaderDigest[] {consensusDigest};
+        HeaderDigest[] headerDigests = new HeaderDigest[]{consensusDigest};
         var optResult = DigestHelper.getBabePreRuntimeDigest(headerDigests);
 
         assertTrue(optResult.isPresent());
@@ -124,7 +126,7 @@ void getBabePreRuntimeDigestForSecondaryPlainSlotTest() {
         message[0] = 2;
         consensusDigest.setMessage(message);
 
-        HeaderDigest[] headerDigests = new HeaderDigest[] {consensusDigest};
+        HeaderDigest[] headerDigests = new HeaderDigest[]{consensusDigest};
         var optResult = DigestHelper.getBabePreRuntimeDigest(headerDigests);
 
         assertTrue(optResult.isPresent());