Merge branch 'kishan/feat/network-bridge-rx' into kishan/feat/collato…

…r-protocol-handle-view
ChainSafe · Sep 30, 2024 · 9acee69 · 9acee69
2 parents a8e2010 + a1bc220
commit 9acee69
Show file tree

Hide file tree

Showing 4 changed files with 29 additions and 35 deletions.
diff --git a/dot/parachain/backing/integration_test.go b/dot/parachain/backing/integration_test.go
@@ -34,7 +34,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)

diff --git a/dot/parachain/collator-protocol/validator_side_test.go b/dot/parachain/collator-protocol/validator_side_test.go
@@ -6,7 +6,6 @@ package collatorprotocol
 import (
 	"sync"
 	"testing"
-	"time"
 
 	"github.com/ChainSafe/gossamer/dot/network"
 	"github.com/ChainSafe/gossamer/dot/parachain/backing"
@@ -347,7 +346,7 @@ func TestProcessBackedOverseerMessage(t *testing.T) {
 	testCases := []struct {
 		description                 string
 		msg                         any
-		expectedActions             []func(msg any) bool
+		canSecond                   bool
 		deletesBlockedAdvertisement bool
 		blockedAdvertisements       map[string][]blockedAdvertisement
 		errString                   string
@@ -358,17 +357,7 @@ func TestProcessBackedOverseerMessage(t *testing.T) {
 				ParaID:   parachaintypes.ParaID(6),
 				ParaHead: common.Hash{},
 			},
-			expectedActions: []func(msg any) bool{
-				func(msg any) bool {
-					canSecondMessage, ok := msg.(backing.CanSecondMessage)
-					if !ok {
-						return false
-					}
-					canSecondMessage.ResponseCh <- true
-
-					return true
-				},
-			},
+			canSecond:                   true,
 			deletesBlockedAdvertisement: true,
 			blockedAdvertisements: map[string][]blockedAdvertisement{
 				"para_id:_6,_para_head:_0x0000000000000000000000000000000000000000000000000000000000000000": {
@@ -396,17 +385,7 @@ func TestProcessBackedOverseerMessage(t *testing.T) {
 				ParaID:   parachaintypes.ParaID(6),
 				ParaHead: common.Hash{},
 			},
-			expectedActions: []func(msg any) bool{
-				func(msg any) bool {
-					canSecondMessage, ok := msg.(backing.CanSecondMessage)
-					if !ok {
-						return false
-					}
-					canSecondMessage.ResponseCh <- false
-
-					return true
-				},
-			},
+			canSecond: false,
 			blockedAdvertisements: map[string][]blockedAdvertisement{
 				"para_id:_6,_para_head:_0x0000000000000000000000000000000000000000000000000000000000000000": {
 					{
@@ -435,8 +414,18 @@ func TestProcessBackedOverseerMessage(t *testing.T) {
 
 			ctrl := gomock.NewController(t)
 			defer ctrl.Finish()
-			overseer := overseer.NewMockableOverseer(t)
-			overseer.ExpectActions(c.expectedActions...)
+			overseer := overseer.NewMockableOverseer(t, false)
+			overseer.ExpectActions([]func(msg any) bool{
+				func(msg any) bool {
+					canSecondMessage, ok := msg.(backing.CanSecondMessage)
+					if !ok {
+						return false
+					}
+					canSecondMessage.ResponseCh <- false
+
+					return true
+				},
+			}...)
 
 			collationProtocolID := "/6761727661676500000000000000000000000000000000000000000000000000/1/collations/1"
 
@@ -454,7 +443,6 @@ func TestProcessBackedOverseerMessage(t *testing.T) {
 
 			defer overseer.Stop()
 
-			time.Sleep(1 * time.Second)
 			lenBlackedAdvertisementsBefore := len(cpvs.BlockedAdvertisements)
 
 			err = cpvs.processMessage(c.msg)

diff --git a/dot/parachain/network-bridge/receiver.go b/dot/parachain/network-bridge/receiver.go
@@ -48,7 +48,6 @@ type NetworkBridgeReceiver struct {
 
 	finalizedNumber uint32
 
-	OverseerToSubSystem  <-chan any
 	SubsystemsToOverseer chan<- any
 
 	networkEventInfoChan chan *network.NetworkEventInfo
@@ -118,6 +117,11 @@ func (nbr *NetworkBridgeReceiver) Run(ctx context.Context, overseerToSubSystem <
 			}
 		case event := <-nbr.networkEventInfoChan:
 			nbr.handleNetworkEvents(*event)
+		case <-ctx.Done():
+			if err := ctx.Err(); err != nil {
+				logger.Errorf("ctx error: %s\n", err)
+			}
+			return
 		}
 	}
 }
@@ -144,8 +148,6 @@ func (nbr *NetworkBridgeReceiver) Name() parachaintypes.SubSystemName {
 func (nbr *NetworkBridgeReceiver) ProcessActiveLeavesUpdateSignal(
 	signal parachaintypes.ActiveLeavesUpdateSignal) error {
 
-	// TODO update cpvs.activeLeaves by adding new active leaves and removing deactivated ones
-
 	// TODO: get the value for majorSyncing for syncing package
 	// majorSyncing means you are 5 blocks behind the tip of the chain and thus more aggressively
 	// download blocks etc to reach the tip of the chain faster.

diff --git a/dot/parachain/overseer/mockable_overseer.go b/dot/parachain/overseer/mockable_overseer.go
@@ -18,6 +18,7 @@ type MockableOverseer struct {
 	SubsystemsToOverseer chan any
 	overseerToSubsystem  chan any
 	subSystem            parachaintypes.Subsystem
+	run                  bool
 
 	// actionsForExpectedMessages stores overseer messages we receive from the subsystem.
 	// need to return false if the message is unexpected
@@ -29,13 +30,14 @@ type MockableOverseer struct {
 	actionsForExpectedMessages []func(msg any) bool
 }
 
-func NewMockableOverseer(t *testing.T) *MockableOverseer {
+func NewMockableOverseer(t *testing.T, run bool) *MockableOverseer {
 	ctx, cancel := context.WithCancel(context.Background())
 
 	return &MockableOverseer{
 		t:                          t,
 		ctx:                        ctx,
 		cancel:                     cancel,
+		run:                        run,
 		SubsystemsToOverseer:       make(chan any),
 		actionsForExpectedMessages: []func(msg any) bool{},
 	}
@@ -53,9 +55,11 @@ func (m *MockableOverseer) RegisterSubsystem(subsystem parachaintypes.Subsystem)
 }
 
 func (m *MockableOverseer) Start() error {
-	go func(sub parachaintypes.Subsystem, overseerToSubSystem chan any) {
-		sub.Run(m.ctx, overseerToSubSystem)
-	}(m.subSystem, m.overseerToSubsystem)
+	if m.run {
+		go func(sub parachaintypes.Subsystem, overseerToSubSystem chan any) {
+			sub.Run(m.ctx, overseerToSubSystem)
+		}(m.subSystem, m.overseerToSubsystem)
+	}
 
 	go m.processMessages()
 	return nil