feat(dot/parachain): receiver side of network bridge (#3955)

- handled active leaves update message
- handled block finalized message
- relay network protocol messages
- handle view update message for receiver side of the network bridge #3864 
- decode messages to wire message (PR #4188), Fixes #4108
- handle our view change in collator protocol validator side (PR #4197), Issue #4156
- Handle network bridge messages (UpdateAuthorityIDs and NewGossipTopology) Fixes #3862 
- process network events for receiver side of network bridge Issue Fixes #3863 
-  process overseer signals for network bridge Fixes #3861

dot/parachain/backing/integration_test.go

@@ -35,7 +35,7 @@ func stopOverseerAndWaitForCompletion(overseer *overseer.MockableOverseer) {
 func initBackingAndOverseerMock(t *testing.T) (*backing.CandidateBacking, *overseer.MockableOverseer) {
 	t.Helper()
 
-	overseerMock := overseer.NewMockableOverseer(t)
+	overseerMock := overseer.NewMockableOverseer(t, true)
 
 	backing := backing.New(overseerMock.SubsystemsToOverseer)
 	overseerMock.RegisterSubsystem(backing)

dot/parachain/collator-protocol/collator_side.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/ChainSafe/gossamer/dot/network"
 	collatorprotocolmessages "github.com/ChainSafe/gossamer/dot/parachain/collator-protocol/messages"
+	networkbridgeevents "github.com/ChainSafe/gossamer/dot/parachain/network-bridge/events"
 	networkbridgemessages "github.com/ChainSafe/gossamer/dot/parachain/network-bridge/messages"
 	parachaintypes "github.com/ChainSafe/gossamer/dot/parachain/types"
 	"github.com/ChainSafe/gossamer/dot/peerset"
@@ -37,7 +38,13 @@ func (cpcs CollatorProtocolCollatorSide) processMessage(msg any) error { //nolin
 		// TODO: handle distribute collation message #3824
 	case collatorprotocolmessages.ReportCollator:
 		return fmt.Errorf("report collator %w", ErrNotExpectedOnCollatorSide)
-	case collatorprotocolmessages.NetworkBridgeUpdate:
+	case networkbridgeevents.PeerConnected,
+		networkbridgeevents.PeerDisconnected,
+		networkbridgeevents.PeerMessage[collatorprotocolmessages.CollationProtocol],
+		networkbridgeevents.NewGossipTopology,
+		networkbridgeevents.PeerViewChange,
+		networkbridgeevents.OurViewChange,
+		networkbridgeevents.UpdatedAuthorityIDs:
 		// TODO: handle network message #3824
 		// https://github.com/paritytech/polkadot-sdk/blob/db3fd687262c68b115ab6724dfaa6a71d4a48a59/polkadot/node/network/collator-protocol/src/validator_side/mod.rs#L1457 //nolint
 	case collatorprotocolmessages.Seconded:

dot/parachain/collator-protocol/message.go

@@ -21,8 +21,6 @@ import (
 	"github.com/libp2p/go-libp2p/core/peer"
 )
 
-const legacyCollationProtocolV1 = "/polkadot/collation/1"
-
 func decodeCollationMessage(in []byte) (network.NotificationsMessage, error) {
 	collationMessage := collatorprotocolmessages.CollationProtocol{}
 
@@ -304,44 +302,29 @@ func getDeclareSignaturePayload(peerID peer.ID) []byte {
 	return payload
 }
 
-func (cpvs CollatorProtocolValidatorSide) handleCollationMessage(
-	sender peer.ID, msg network.NotificationsMessage) (bool, error) {
-
-	// we don't propagate collation messages, so it will always be false
-	propagate := false
-
-	if msg.Type() != network.CollationMsgType {
-		return propagate, fmt.Errorf("%w, expected: %d, found:%d", ErrUnexpectedMessageOnCollationProtocol,
-			network.CollationMsgType, msg.Type())
-	}
-
-	collatorProtocol, ok := msg.(*collatorprotocolmessages.CollationProtocol)
-	if !ok {
-		return propagate, fmt.Errorf(
-			"failed to cast into collator protocol message, expected: *CollationProtocol, got: %T",
-			msg)
-	}
+func (cpvs CollatorProtocolValidatorSide) processCollatorProtocolMessage(sender peer.ID,
+	msg collatorprotocolmessages.CollationProtocol) error {
 
-	collatorProtocolV, err := collatorProtocol.Value()
+	collatorProtocolV, err := msg.Value()
 	if err != nil {
-		return propagate, fmt.Errorf("getting collator protocol value: %w", err)
+		return fmt.Errorf("getting collator protocol value: %w", err)
 	}
 	collatorProtocolMessage, ok := collatorProtocolV.(collatorprotocolmessages.CollatorProtocolMessage)
 	if !ok {
-		return propagate, errors.New("expected value to be collator protocol message")
+		return errors.New("expected value to be collator protocol message")
 	}
 
 	index, collatorProtocolMessageV, err := collatorProtocolMessage.IndexValue()
 	if err != nil {
-		return propagate, fmt.Errorf("getting collator protocol message value: %w", err)
+		return fmt.Errorf("getting collator protocol message value: %w", err)
 	}
 
 	switch index {
 	// TODO: Create an issue to cover v2 types. #3534
 	case 0: // Declare
 		declareMessage, ok := collatorProtocolMessageV.(collatorprotocolmessages.Declare)
 		if !ok {
-			return propagate, errors.New("expected message to be declare")
+			return errors.New("expected message to be declare")
 		}
 
 		// check if we already have the collator id declared in this message. If so, punish the
@@ -356,7 +339,7 @@ func (cpvs CollatorProtocolValidatorSide) handleCollationMessage(
 				},
 			}
 
-			return propagate, nil
+			return nil
 		}
 
 		// NOTE: peerData for sender will be filled when it gets connected to us
@@ -370,7 +353,7 @@ func (cpvs CollatorProtocolValidatorSide) handleCollationMessage(
 				},
 			}
 
-			return propagate, fmt.Errorf("%w: %s", ErrUnknownPeer, sender)
+			return fmt.Errorf("%w: %s", ErrUnknownPeer, sender)
 		}
 
 		if peerData.state.PeerState == Collating {
@@ -383,7 +366,7 @@ func (cpvs CollatorProtocolValidatorSide) handleCollationMessage(
 				},
 			}
 
-			return propagate, nil
+			return nil
 		}
 
 		// check signature declareMessage.CollatorSignature
@@ -398,10 +381,10 @@ func (cpvs CollatorProtocolValidatorSide) handleCollationMessage(
 				},
 			}
 
-			return propagate, fmt.Errorf("invalid signature: %w", err)
+			return fmt.Errorf("invalid signature: %w", err)
 		}
 		if err != nil {
-			return propagate, fmt.Errorf("verifying signature: %w", err)
+			return fmt.Errorf("verifying signature: %w", err)
 		}
 
 		// NOTE: assignments are setting when we handle view changes
@@ -433,12 +416,12 @@ func (cpvs CollatorProtocolValidatorSide) handleCollationMessage(
 	case 1: // AdvertiseCollation
 		advertiseCollationMessage, ok := collatorProtocolMessageV.(collatorprotocolmessages.AdvertiseCollation)
 		if !ok {
-			return propagate, errors.New("expected message to be advertise collation")
+			return errors.New("expected message to be advertise collation")
 		}
 
 		err := cpvs.handleAdvertisement(common.Hash(advertiseCollationMessage), sender, nil)
 		if err != nil {
-			return propagate, fmt.Errorf("handling v1 advertisement: %w", err)
+			return fmt.Errorf("handling v1 advertisement: %w", err)
 		}
 		// TODO:
 		// - tracks advertisements received and the source (peer id) of the advertisement
@@ -455,7 +438,7 @@ func (cpvs CollatorProtocolValidatorSide) handleCollationMessage(
 		}
 	}
 
-	return propagate, nil
+	return nil
 }
 
 func getCollatorHandshake() (network.Handshake, error) {

dot/parachain/collator-protocol/message_test.go

@@ -16,7 +16,6 @@ import (
 	"github.com/stretchr/testify/require"
 	"gopkg.in/yaml.v3"
 
-	"github.com/ChainSafe/gossamer/dot/network"
 	collatorprotocolmessages "github.com/ChainSafe/gossamer/dot/parachain/collator-protocol/messages"
 	networkbridgemessages "github.com/ChainSafe/gossamer/dot/parachain/network-bridge/messages"
 	parachaintypes "github.com/ChainSafe/gossamer/dot/parachain/types"
@@ -184,23 +183,9 @@ func TestHandleCollationMessageCommon(t *testing.T) {
 
 	peerID := peer.ID("testPeerID")
 
-	// fail with wrong message type
-	msg1 := &network.BlockAnnounceMessage{}
-	propagate, err := cpvs.handleCollationMessage(peerID, msg1)
-	require.False(t, propagate)
-	require.ErrorIs(t, err, ErrUnexpectedMessageOnCollationProtocol)
-
-	// fail if we can't cast the message to type `*CollationProtocol`
-	msg2 := collatorprotocolmessages.NewCollationProtocol()
-	propagate, err = cpvs.handleCollationMessage(peerID, msg2)
-	require.False(t, propagate)
-	require.ErrorContains(t, err, "failed to cast into collator protocol message, "+
-		"expected: *CollationProtocol, got: messages.CollationProtocol")
-
 	// fail if no value set in the collator protocol message
-	msg3 := collatorprotocolmessages.NewCollationProtocol()
-	propagate, err = cpvs.handleCollationMessage(peerID, &msg3)
-	require.False(t, propagate)
+	msg := collatorprotocolmessages.NewCollationProtocol()
+	err := cpvs.processCollatorProtocolMessage(peerID, msg)
 	require.ErrorContains(t, err, "getting collator protocol value: unsupported varying data type value")
 }
 
@@ -400,8 +385,7 @@ func TestHandleCollationMessageDeclare(t *testing.T) {
 			err = msg.SetValue(vdtChild)
 			require.NoError(t, err)
 
-			propagate, err := cpvs.handleCollationMessage(peerID, &msg)
-			require.False(t, propagate)
+			err := cpvs.processCollatorProtocolMessage(peerID, msg)
 			if c.errString == "" {
 				require.NoError(t, err)
 			} else {
@@ -579,11 +563,10 @@ func TestHandleCollationMessageAdvertiseCollation(t *testing.T) {
 
 			subsystemToOverseer := make(chan any)
 			cpvs := CollatorProtocolValidatorSide{
-				SubSystemToOverseer: subsystemToOverseer,
-				net:                 c.net,
 				perRelayParent:      c.perRelayParent,
 				peerData:            c.peerData,
 				activeLeaves:        c.activeLeaves,
+				SubSystemToOverseer: subsystemToOverseer,
 			}
 
 			// ensure that the expected messages are sent to the overseer
@@ -602,8 +585,7 @@ func TestHandleCollationMessageAdvertiseCollation(t *testing.T) {
 			err = msg.SetValue(vdtChild)
 			require.NoError(t, err)
 
-			propagate, err := cpvs.handleCollationMessage(peerID, &msg)
-			require.False(t, propagate)
+			err = cpvs.processCollatorProtocolMessage(peerID, msg)
 			if c.errString == "" {
 				require.NoError(t, err)
 			} else {

dot/parachain/collator-protocol/messages/overseer_messages.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 
 	parachaintypes "github.com/ChainSafe/gossamer/dot/parachain/types"
+
 	"github.com/ChainSafe/gossamer/lib/common"
 )
 
@@ -19,10 +20,6 @@ type DistributeCollation struct {
 
 type ReportCollator parachaintypes.CollatorID
 
-type NetworkBridgeUpdate struct {
-	// TODO: not quite sure if we would need this or something similar to this
-}
-
 // Seconded represents that the candidate we recommended to be seconded was validated
 // successfully.
 type Seconded struct {