chore: merge dev into current branch

src/main/java/com/limechain/grandpa/GrandpaService.java

@@ -4,10 +4,13 @@
 import com.limechain.exception.storage.BlockStorageGenericException;
 import com.limechain.grandpa.state.GrandpaRound;
 import com.limechain.grandpa.state.GrandpaSetState;
+import com.limechain.network.PeerMessageCoordinator;
 import com.limechain.network.protocol.grandpa.messages.catchup.res.SignedVote;
+import com.limechain.network.protocol.grandpa.messages.commit.CommitMessage;
 import com.limechain.network.protocol.grandpa.messages.commit.Vote;
 import com.limechain.network.protocol.grandpa.messages.vote.Subround;
 import com.limechain.network.protocol.warp.dto.BlockHeader;
+import com.limechain.network.protocol.warp.dto.PreCommit;
 import com.limechain.storage.block.BlockState;
 import io.emeraldpay.polkaj.types.Hash256;
 import lombok.extern.java.Log;
@@ -25,10 +28,14 @@ public class GrandpaService {
 
     private final GrandpaSetState grandpaSetState;
     private final BlockState blockState;
+    private final PeerMessageCoordinator peerMessageCoordinator;
 
-    public GrandpaService(GrandpaSetState grandpaSetState) {
+
+    public GrandpaService(GrandpaSetState grandpaSetState,
+                          PeerMessageCoordinator peerMessageCoordinator) {
         this.grandpaSetState = grandpaSetState;
         this.blockState = BlockState.getInstance();
+        this.peerMessageCoordinator = peerMessageCoordinator;
     }
 
     private void attemptToFinalizeAt(GrandpaRound grandpaRound) {
@@ -470,4 +477,36 @@ private Vote getLastFinalizedBlockAsVote() {
                 lastFinalizedBlockHeader.getBlockNumber()
         );
     }
+
+    /**
+     * Broadcasts a commit message to network peers for the given round as part of the GRANDPA consensus process.
+     * <p>
+     * This method is used in two scenarios:
+     * 1. As the primary validator, broadcasting a commit message for the best candidate block of the previous round.
+     * 2. During attempt-to-finalize, broadcasting a commit message for the best candidate block of the current round.
+     */
+    public void broadcastCommitMessage(GrandpaRound grandpaRound) {
+        Vote bestCandidate = getBestFinalCandidate(grandpaRound);
+        PreCommit[] precommits = transformToCompactJustificationFormat(grandpaRound.getPreCommits());
+
+        CommitMessage commitMessage = new CommitMessage();
+        commitMessage.setSetId(grandpaSetState.getSetId());
+        commitMessage.setRoundNumber(grandpaRound.getRoundNumber());
+        commitMessage.setVote(bestCandidate);
+        commitMessage.setPreCommits(precommits);
+
+        peerMessageCoordinator.sendCommitMessageToPeers(commitMessage);
+    }
+
+    private PreCommit[] transformToCompactJustificationFormat(Map<Hash256, SignedVote> signedVotes) {
+        return signedVotes.values().stream()
+                .map(signedVote -> {
+                    PreCommit precommit = new PreCommit();
+                    precommit.setTargetHash(signedVote.getVote().getBlockHash());
+                    precommit.setTargetNumber(signedVote.getVote().getBlockNumber());
+                    precommit.setSignature(signedVote.getSignature());
+                    precommit.setAuthorityPublicKey(signedVote.getAuthorityPublicKey());
+                    return precommit;
+                }).toArray(PreCommit[]::new);
+    }
 }

src/main/java/com/limechain/network/PeerMessageCoordinator.java

@@ -4,6 +4,8 @@
 import com.limechain.network.protocol.blockannounce.NodeRole;
 import com.limechain.network.protocol.blockannounce.messages.BlockAnnounceMessage;
 import com.limechain.network.protocol.blockannounce.scale.BlockAnnounceMessageScaleWriter;
+import com.limechain.network.protocol.grandpa.messages.commit.CommitMessage;
+import com.limechain.network.protocol.grandpa.messages.commit.CommitMessageScaleWriter;
 import com.limechain.network.protocol.transaction.scale.TransactionWriter;
 import com.limechain.transaction.dto.Extrinsic;
 import com.limechain.transaction.dto.ExtrinsicArray;
@@ -99,4 +101,13 @@ public void handshakeBootNodes() {
     public void sendNeighbourMessageToPeer(PeerId peerId) {
         network.getGrandpaService().sendNeighbourMessage(network.getHost(), peerId);
     }
+
+    public void sendCommitMessageToPeers(CommitMessage commitMessage) {
+        byte[] scaleMessage = ScaleUtils.Encode.encode(CommitMessageScaleWriter.getInstance(), commitMessage);
+        sendMessageToActivePeers(peerId -> {
+            asyncExecutor.executeAndForget(() -> network.getGrandpaService().sendCommitMessage(
+                    network.getHost(), peerId, scaleMessage
+            ));
+        });
+    }
 }

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

@@ -26,4 +26,11 @@ public void sendHandshake() {
     public void sendNeighbourMessage() {
         engine.writeNeighbourMessage(stream, stream.remotePeerId());
     }
+
+    /**
+     * Sends a commit message over the controller stream.
+     */
+    public void sendCommitMessage(byte[] encodedCommitMessage) {
+        engine.writeCommitMessage(stream, encodedCommitMessage);
+    }
 }

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

@@ -1,6 +1,7 @@
 package com.limechain.network.protocol.grandpa;
 
 import com.limechain.exception.scale.ScaleEncodingException;
+import com.limechain.grandpa.GrandpaService;
 import com.limechain.grandpa.state.GrandpaSetState;
 import com.limechain.network.ConnectionManager;
 import com.limechain.network.protocol.blockannounce.messages.BlockAnnounceHandshakeBuilder;
@@ -201,4 +202,15 @@ public void writeNeighbourMessage(Stream stream, PeerId peerId) {
         log.log(Level.FINE, "Sending neighbour message to peer " + peerId);
         stream.writeAndFlush(buf.toByteArray());
     }
+
+    /**
+     * Send our GRANDPA commit message from {@link GrandpaService} on a given <b>responder</b> stream.
+     *
+     * @param stream               <b>responder</b> stream to write the message to
+     * @param encodedCommitMessage scale encoded CommitMessage object
+     */
+    public void writeCommitMessage(Stream stream, byte[] encodedCommitMessage) {
+        log.log(Level.FINE, "Sending commit message to peer " + stream.remotePeerId());
+        stream.writeAndFlush(encodedCommitMessage);
+    }
 }

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

@@ -37,6 +37,23 @@ public void sendNeighbourMessage(Host us, PeerId peerId) {
                 );
     }
 
+    /**
+     * Sends a commit message to a peer. If there is no initiator stream opened with the peer,
+     * sends a handshake instead.
+     *
+     * @param us                   our host object
+     * @param peerId               message receiver
+     * @param encodedCommitMessage scale encoded representation of the CommitMessage object
+     */
+    public void sendCommitMessage(Host us, PeerId peerId, byte[] encodedCommitMessage) {
+        Optional.ofNullable(connectionManager.getPeerInfo(peerId))
+                .map(p -> p.getGrandpaStreams().getInitiator())
+                .ifPresentOrElse(
+                        stream -> new GrandpaController(stream).sendCommitMessage(encodedCommitMessage),
+                        () -> sendHandshake(us, peerId)
+                );
+    }
+
     private void sendNeighbourMessage(Stream stream) {
         GrandpaController controller = new GrandpaController(stream);
         controller.sendNeighbourMessage();

src/test/java/com/limechain/grandpa/GrandpaServiceTest.java

@@ -3,7 +3,9 @@
 import com.limechain.exception.grandpa.GhostExecutionException;
 import com.limechain.grandpa.state.GrandpaRound;
 import com.limechain.grandpa.state.GrandpaSetState;
+import com.limechain.network.PeerMessageCoordinator;
 import com.limechain.network.protocol.grandpa.messages.catchup.res.SignedVote;
+import com.limechain.network.protocol.grandpa.messages.commit.CommitMessage;
 import com.limechain.network.protocol.grandpa.messages.commit.Vote;
 import com.limechain.network.protocol.grandpa.messages.vote.SignedMessage;
 import com.limechain.network.protocol.grandpa.messages.vote.Subround;
@@ -12,12 +14,14 @@
 import com.limechain.network.protocol.warp.dto.ConsensusEngine;
 import com.limechain.network.protocol.warp.dto.DigestType;
 import com.limechain.network.protocol.warp.dto.HeaderDigest;
+import com.limechain.network.protocol.warp.dto.PreCommit;
 import com.limechain.storage.block.BlockState;
 import io.emeraldpay.polkaj.types.Hash256;
 import io.emeraldpay.polkaj.types.Hash512;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
 import org.mockito.MockedStatic;
 
 import java.lang.reflect.Method;
@@ -34,6 +38,8 @@
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 class GrandpaServiceTest {
@@ -51,14 +57,16 @@ class GrandpaServiceTest {
     private MockedStatic<BlockState> mockedBlockState;
     private GrandpaService grandpaService;
     private GrandpaRound grandpaRound;
+    private PeerMessageCoordinator peerMessageCoordinator;
 
     @BeforeEach
     void setUp() {
         grandpaSetState = mock(GrandpaSetState.class);
         blockState = mock(BlockState.class);
         mockedBlockState = mockStatic(BlockState.class);
         mockedBlockState.when(BlockState::getInstance).thenReturn(blockState);
-        grandpaService = new GrandpaService(grandpaSetState);
+        peerMessageCoordinator = mock(PeerMessageCoordinator.class);
+        grandpaService = new GrandpaService(grandpaSetState, peerMessageCoordinator);
         grandpaRound = mock(GrandpaRound.class);
         when(grandpaRound.getPrevious()).thenReturn(new GrandpaRound());
     }
@@ -733,6 +741,47 @@ void testSelectBlockWithMostVotesWhereLastFinalizedBlockIsWithGreaterBlockNumber
         assertEquals(blockHeader.getBlockNumber(), result.getBlockNumber());
     }
 
+    @Test
+    void testBroadcastCommitMessageWhenPrimaryValidator() {
+        Hash256 authorityPublicKey = new Hash256(THREES_ARRAY);
+        Map<Hash256, SignedVote> signedVotes = new HashMap<>();
+        Vote vote = new Vote(new Hash256(ONES_ARRAY), BigInteger.valueOf(123L));
+        SignedVote signedVote = new SignedVote(vote, Hash512.empty(), authorityPublicKey);
+        signedVotes.put(authorityPublicKey, signedVote);
+
+        GrandpaRound previousRound = new GrandpaRound();
+        previousRound.setRoundNumber(BigInteger.ZERO);
+        previousRound.setPreCommits(signedVotes);
+        BlockHeader blockHeader = createBlockHeader();
+
+        when(grandpaSetState.getThreshold()).thenReturn(BigInteger.ONE);
+        when(blockState.getHighestFinalizedHeader()).thenReturn(blockHeader);
+        when(grandpaSetState.getSetId()).thenReturn(BigInteger.valueOf(42L));
+
+        grandpaService.broadcastCommitMessage(previousRound);
+
+        ArgumentCaptor<CommitMessage> commitMessageCaptor = ArgumentCaptor.forClass(CommitMessage.class);
+        verify(peerMessageCoordinator).sendCommitMessageToPeers(commitMessageCaptor.capture());
+        CommitMessage commitMessage = commitMessageCaptor.getValue();
+
+        assertEquals(BigInteger.valueOf(42L), commitMessage.getSetId());
+        assertEquals(BigInteger.valueOf(0L), commitMessage.getRoundNumber());
+        assertEquals(blockHeader.getBlockNumber(), commitMessage.getVote().getBlockNumber());
+        assertEquals(blockHeader.getHash(), commitMessage.getVote().getBlockHash());
+        assertEquals(1, commitMessage.getPreCommits().length);
+
+        PreCommit precommit = commitMessage.getPreCommits()[0];
+        assertEquals(vote.getBlockHash(), precommit.getTargetHash());
+        assertEquals(BigInteger.valueOf(123L), precommit.getTargetNumber());
+        assertEquals(Hash512.empty(), precommit.getSignature());
+        assertEquals(signedVotes.get(authorityPublicKey).getAuthorityPublicKey(),
+                precommit.getAuthorityPublicKey()
+        );
+
+        verify(peerMessageCoordinator, times(1))
+                .sendCommitMessageToPeers(any(CommitMessage.class));
+    }
+
     private BlockHeader createBlockHeader() {
         HeaderDigest headerDigest = new HeaderDigest();
         headerDigest.setType(DigestType.CONSENSUS_MESSAGE);

src/test/java/com/limechain/network/protocol/grandpa/GrandpaControllerTest.java

@@ -25,19 +25,27 @@ class GrandpaControllerTest {
 
     @BeforeEach
     void setup() {
-        when(stream.remotePeerId()).thenReturn(peerId);
         grandpaController.engine = engine;
     }
 
     @Test
     void sendHandshake() {
+        when(stream.remotePeerId()).thenReturn(peerId);
         grandpaController.sendHandshake();
         verify(engine).writeHandshakeToStream(stream, peerId);
     }
 
     @Test
     void sendNeighbourMessage() {
+        when(stream.remotePeerId()).thenReturn(peerId);
         grandpaController.sendNeighbourMessage();
         verify(engine).writeNeighbourMessage(stream, peerId);
     }
+
+    @Test
+    void sendCommitMessage() {
+        byte[] encodedCommitMessage = {1, 0, 0, 0, 2, 0, 1, 1, 1, 1, 0, 0, 0, 1, 2, 0};
+        grandpaController.sendCommitMessage(encodedCommitMessage);
+        verify(engine).writeCommitMessage(stream, encodedCommitMessage);
+    }
 }

src/test/java/com/limechain/network/protocol/grandpa/GrandpaEngineTest.java

@@ -33,7 +33,13 @@
 import java.math.BigInteger;
 
 import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.*;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
 
 @SuppressWarnings("unused")
 @ExtendWith(MockitoExtension.class)
@@ -58,6 +64,9 @@ class GrandpaEngineTest {
     private final byte[] encodedNeighbourMessage
             = new byte[]{2, 1, 1, 0, 0, 0, 0, 0, 0, 0, 2, 0, 0, 0, 0, 0, 0, 0, 10, 0, 0, 0};
 
+    private final byte[] encodedCommitMessage
+            = new byte[]{2, 1, 1, 0, 0, 0, 0, 0, 0, 0, 2, 0, 0, 0, 0, 0, 0, 0, 10, 0, 0, 0};
+
     @Test
     void receiveRequestWithUnknownGrandpaTypeShouldLogAndIgnore() {
         byte[] unknownTypeMessage = new byte[]{7, 1, 1, 0, 0, 0, 0, 0, 0, 0, 2, 0, 0, 0, 0, 0, 0, 0, 10, 0, 0, 0};
@@ -286,4 +295,10 @@ void writeNeighbourMessage() {
             verify(stream).writeAndFlush(encodedNeighbourMessage);
         }
     }
+
+    @Test
+    void writeCommitMessage() {
+        grandpaEngine.writeCommitMessage(stream, encodedCommitMessage);
+        verify(stream).writeAndFlush(encodedCommitMessage);
+    }
 }