feat(parachain/collator): handle backed overseer message

dot/node_integration_test.go

@@ -120,7 +120,26 @@ func TestNewNode(t *testing.T) {
 		return stateSrvc, nil
 	})
 
-	mockBlockState := &state.BlockState{}
+	tries := state.NewTries()
+	tries.SetEmptyTrie()
+
+	db := state.NewInMemoryDB(t)
+	prefix := []byte{98, 108, 111, 99, 107}
+
+	err = db.Put(append(prefix, []byte{104, 115, 104, 0, 0, 0, 0, 0, 0, 0, 0}...),
+		[]byte{98, 108, 111, 99, 107, 104, 115, 104})
+	require.NoError(t, err)
+	err = db.Put(append(prefix, []byte{104, 114, 115}...),
+		[]byte{98, 108, 111, 99, 107, 104, 115, 104, 98, 108, 111, 99, 107, 104, 115, 104})
+	require.NoError(t, err)
+	err = db.Put(append(prefix, []byte{102, 105, 110, 97, 108, 105, 115, 101, 100, 95, 104, 101, 97, 100, 98, 108, 111, 99, 107, 104, 115, 104, 98, 108, 111, 99, 107, 104, 115, 104}...), //nolint
+		[]byte{98, 108, 111, 99, 107, 104, 115, 104, 98, 108, 111, 99, 107, 104, 115, 104})
+	require.NoError(t, err)
+	err = db.Put(append(prefix, []byte{104, 100, 114, 98, 108, 111, 99, 107, 104, 115, 104, 98, 108, 111, 99, 107, 104, 115, 104, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}...), //nolint
+		[]byte{59, 69, 201, 194, 45, 206, 206, 117, 163, 10, 204, 156, 41, 104, 203, 49, 30, 107, 5, 87, 53, 15, 131, 180, 48, 244, 117, 89, 219, 120, 105, 117, 0, 9, 249, 202, 40, 223, 5, 96, 194, 41, 26, 161, 107, 86, 225, 94, 7, 209, 225, 146, 112, 136, 245, 19, 86, 213, 34, 114, 42, 169, 12, 167, 203, 218, 38, 220, 140, 20, 85, 248, 248, 28, 174, 18, 228, 252, 89, 226, 60, 233, 97, 178, 200, 55, 246, 211, 246, 100, 40, 58, 249, 6, 211, 68, 224, 0})
+
+	mockBlockState, err := state.NewBlockState(db, tries, nil)
+	require.NoError(t, err)
 	mockStateService := &state.Service{
 		Block: mockBlockState,
 	}

dot/parachain/collator-protocol/message.go

@@ -6,9 +6,11 @@ package collatorprotocol
 import (
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/ChainSafe/gossamer/dot/network"
 	"github.com/ChainSafe/gossamer/dot/parachain/backing"
+	collatorprotocolmessages "github.com/ChainSafe/gossamer/dot/parachain/collator-protocol/messages"
 	parachaintypes "github.com/ChainSafe/gossamer/dot/parachain/types"
 	"github.com/ChainSafe/gossamer/dot/peerset"
 	"github.com/ChainSafe/gossamer/lib/common"
@@ -178,10 +180,10 @@ const (
 	Seconded
 )
 
-// BlockedAdvertisement is vstaging advertisement that was rejected by the backing
+// blockedAdvertisement is vstaging advertisement that was rejected by the backing
 // subsystem. Validator may fetch it later if its fragment
 // membership gets recognised before relay parent goes out of view.
-type BlockedAdvertisement struct {
+type blockedAdvertisement struct {
 	// peer that advertised the collation
 	peerID               peer.ID
 	collatorID           parachaintypes.CollatorID
@@ -194,26 +196,23 @@ func (cpvs CollatorProtocolValidatorSide) canSecond(
 	candidateRelayParent common.Hash,
 	candidateHash parachaintypes.CandidateHash,
 	parentHeadDataHash common.Hash,
-) bool {
+) (bool, error) {
+
 	canSecondRequest := backing.CanSecondMessage{
 		CandidateParaID:      candidateParaID,
 		CandidateRelayParent: candidateRelayParent,
 		CandidateHash:        candidateHash,
 		ParentHeadDataHash:   parentHeadDataHash,
+		ResponseCh:           make(chan bool),
 	}
 
-	responseChan := make(chan bool)
-
-	cpvs.SubSystemToOverseer <- struct {
-		responseChan     chan bool
-		canSecondRequest backing.CanSecondMessage
-	}{
-		responseChan:     responseChan,
-		canSecondRequest: canSecondRequest,
+	cpvs.SubSystemToOverseer <- canSecondRequest
+	select {
+	case canSecondResponse := <-canSecondRequest.ResponseCh:
+		return canSecondResponse, nil
+	case <-time.After(parachaintypes.SubsystemRequestTimeout):
+		return false, parachaintypes.ErrSubsystemRequestTimeout
 	}
-
-	// TODO: Add timeout
-	return <-responseChan
 }
 
 // Enqueue collation for fetching. The advertisement is expected to be
@@ -297,10 +296,6 @@ func (cpvs *CollatorProtocolValidatorSide) fetchCollation(pendingCollation Pendi
 
 func (cpvs *CollatorProtocolValidatorSide) handleAdvertisement(relayParent common.Hash, sender peer.ID,
 	prospectiveCandidate *ProspectiveCandidate) error {
-	// TODO:
-	// - tracks advertisements received and the source (peer id) of the advertisement
-	// - accept one advertisement per collator per source per relay-parent
-
 	perRelayParent, ok := cpvs.perRelayParent[relayParent]
 	if !ok {
 		cpvs.net.ReportPeer(peerset.ReputationChange{
@@ -366,26 +361,38 @@ func (cpvs *CollatorProtocolValidatorSide) handleAdvertisement(relayParent commo
 	}
 
 	/*NOTE:---------------------------------------Matters only in V2----------------------------------------------*/
-	isSecondingAllowed := !perRelayParent.prospectiveParachainMode.IsEnabled || cpvs.canSecond(
-		collatorParaID,
-		relayParent,
-		prospectiveCandidate.CandidateHash,
-		prospectiveCandidate.ParentHeadDataHash,
-	)
+	var isSecondingAllowed bool
+	if !perRelayParent.prospectiveParachainMode.IsEnabled {
+		isSecondingAllowed = true
+	} else {
+		isSecondingAllowed, err = cpvs.canSecond(
+			collatorParaID,
+			relayParent,
+			prospectiveCandidate.CandidateHash,
+			prospectiveCandidate.ParentHeadDataHash,
+		)
+		if err != nil {
+			return fmt.Errorf("checking if seconding is allowed: %w", err)
+		}
+	}
 
 	if !isSecondingAllowed {
 		logger.Infof("Seconding is not allowed by backing, queueing advertisement,"+
 			" relay parent: %s, para id: %d, candidate hash: %s",
 			relayParent, collatorParaID, prospectiveCandidate.CandidateHash)
 
-		blockedAdvertisements := append(cpvs.BlockedAdvertisements, BlockedAdvertisement{
+		backed := collatorprotocolmessages.Backed{
+			ParaID:   collatorParaID,
+			ParaHead: prospectiveCandidate.ParentHeadDataHash,
+		}
+
+		blockedAd := blockedAdvertisement{
 			peerID:               sender,
 			collatorID:           peerData.state.CollatingPeerState.CollatorID,
 			candidateRelayParent: relayParent,
 			candidateHash:        prospectiveCandidate.CandidateHash,
-		})
-
-		cpvs.BlockedAdvertisements = blockedAdvertisements
+		}
+		cpvs.BlockedAdvertisements[backed.String()] = []blockedAdvertisement{blockedAd}
 		return nil
 	}
 	/*--------------------------------------------END----------------------------------------------------------*/

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

@@ -4,6 +4,8 @@
 package messages
 
 import (
+	"fmt"
+
 	parachaintypes "github.com/ChainSafe/gossamer/dot/parachain/types"
 	"github.com/ChainSafe/gossamer/lib/common"
 )
@@ -35,6 +37,10 @@ type Backed struct {
 	ParaHead common.Hash
 }
 
+func (b Backed) String() string {
+	return fmt.Sprintf("para_id:_%d,_para_head:_%s", b.ParaID, b.ParaHead.String())
+}
+
 // Invalid represents an invalid candidata.
 // We recommended a particular candidate to be seconded, but it was invalid; penalise the collator.
 type Invalid struct {

dot/parachain/collator-protocol/mocks_generate_test.go

@@ -4,3 +4,5 @@
 package collatorprotocol
 
 //go:generate mockgen -destination=mocks_test.go -package=$GOPACKAGE . Network
+//go:generate mockgen -destination=mock_blockstate_test.go -package=$GOPACKAGE github.com/ChainSafe/gossamer/dot/parachain/overseer BlockState
+//go:generate mockgen -destination=mock_subsystem_test.go -package=$GOPACKAGE github.com/ChainSafe/gossamer/dot/parachain/overseer Subsystem