Handle Active Leaves update in collator protocol

dot/network/notifications.go

@@ -270,6 +270,7 @@ func (s *Service) sendData(peer peer.ID, hs Handshake, info *notificationsProtoc
 		return
 	}
 
+	info.peersData.setMutex(peer)
 	stream, err := s.sendHandshake(peer, hs, info)
 	if err != nil {
 		logger.Debugf("failed to send handshake to peer %s on protocol %s: %s", peer, info.protocolID, err)

dot/node.go

@@ -64,7 +64,8 @@ type nodeBuilderIface interface {
 	) (*core.Service, error)
 	createGRANDPAService(config *cfg.Config, st *state.Service, ks KeyStore,
 		net *network.Service, telemetryMailer Telemetry) (*grandpa.Service, error)
-	createParachainHostService(net *network.Service, forkID string, st *state.Service) (*parachain.Service, error)
+	createParachainHostService(net *network.Service, forkID string, st *state.Service, ks keystore.Keystore,
+	) (*parachain.Service, error)
 	newSyncService(config *cfg.Config, st *state.Service, finalityGadget BlockJustificationVerifier,
 		verifier *babe.VerificationManager, cs *core.Service, net *network.Service,
 		telemetryMailer Telemetry) (*dotsync.Service, error)
@@ -381,7 +382,7 @@ func newNode(config *cfg.Config,
 	}
 	nodeSrvcs = append(nodeSrvcs, fg)
 
-	phs, err := builder.createParachainHostService(networkSrvc, gd.ForkID, stateSrvc)
+	phs, err := builder.createParachainHostService(networkSrvc, gd.ForkID, stateSrvc, ks.Para)
 	if err != nil {
 		return nil, err
 	}

dot/node_integration_test.go

@@ -143,7 +143,7 @@ func TestNewNode(t *testing.T) {
 	mockStateService := &state.Service{
 		Block: mockBlockState,
 	}
-	phs, err := parachain.NewService(testNetworkService, "random_fork_id", mockStateService)
+	phs, err := parachain.NewService(testNetworkService, "random_fork_id", mockStateService, ks.Para)
 	require.NoError(t, err)
 
 	m.EXPECT().createRuntimeStorage(gomock.AssignableToTypeOf(&state.Service{})).Return(&runtime.
@@ -180,6 +180,7 @@ func TestNewNode(t *testing.T) {
 		gomock.AssignableToTypeOf(&network.Service{}),
 		gomock.AssignableToTypeOf("random_fork_id"),
 		gomock.AssignableToTypeOf(mockStateService),
+		gomock.AssignableToTypeOf(ks.Para),
 	).Return(phs, nil)
 
 	got, err := newNode(initConfig, ks, m, mockServiceRegistry)

dot/parachain/availability-store/availability_store.go

@@ -456,10 +456,10 @@ func (av *AvailabilityStoreSubsystem) processMessages() {
 				}
 
 			case parachaintypes.ActiveLeavesUpdateSignal:
-				av.ProcessActiveLeavesUpdateSignal()
+				av.ProcessActiveLeavesUpdateSignal(msg)
 
 			case parachaintypes.BlockFinalizedSignal:
-				av.ProcessBlockFinalizedSignal()
+				av.ProcessBlockFinalizedSignal(msg)
 
 			default:
 				logger.Error(parachaintypes.ErrUnknownOverseerMessage.Error())
@@ -476,11 +476,11 @@ func (av *AvailabilityStoreSubsystem) processMessages() {
 	}
 }
 
-func (av *AvailabilityStoreSubsystem) ProcessActiveLeavesUpdateSignal() {
+func (av *AvailabilityStoreSubsystem) ProcessActiveLeavesUpdateSignal(signal parachaintypes.ActiveLeavesUpdateSignal) {
 	// TODO: #3630
 }
 
-func (av *AvailabilityStoreSubsystem) ProcessBlockFinalizedSignal() {
+func (av *AvailabilityStoreSubsystem) ProcessBlockFinalizedSignal(signal parachaintypes.BlockFinalizedSignal) {
 	// TODO: #3630
 }
 

dot/parachain/backing/candidate_backing.go

@@ -247,20 +247,20 @@ func (cb *CandidateBacking) processMessage(msg any, chRelayParentAndCommand chan
 	case StatementMessage:
 		return cb.handleStatementMessage(msg.RelayParent, msg.SignedFullStatement, chRelayParentAndCommand)
 	case parachaintypes.ActiveLeavesUpdateSignal:
-		cb.ProcessActiveLeavesUpdateSignal()
+		cb.ProcessActiveLeavesUpdateSignal(msg)
 	case parachaintypes.BlockFinalizedSignal:
-		cb.ProcessBlockFinalizedSignal()
+		cb.ProcessBlockFinalizedSignal(msg)
 	default:
 		return fmt.Errorf("%w: %T", parachaintypes.ErrUnknownOverseerMessage, msg)
 	}
 	return nil
 }
 
-func (cb *CandidateBacking) ProcessActiveLeavesUpdateSignal() {
+func (cb *CandidateBacking) ProcessActiveLeavesUpdateSignal(signal parachaintypes.ActiveLeavesUpdateSignal) {
 	// TODO #3503
 }
 
-func (cb *CandidateBacking) ProcessBlockFinalizedSignal() {
+func (cb *CandidateBacking) ProcessBlockFinalizedSignal(signal parachaintypes.BlockFinalizedSignal) {
 	// TODO #3644
 }
 

dot/parachain/collator-protocol/collator_side.go

@@ -22,10 +22,10 @@ var ErrNotExpectedOnCollatorSide = errors.New("message is not expected on the co
 
 type CollatorProtocolCollatorSide struct {
 	net         Network
-	collatingOn parachaintypes.ParaID //nolint:unused
+	collatingOn parachaintypes.ParaID //nolint
 }
 
-func (cpcs CollatorProtocolCollatorSide) processMessage(msg any) error { //nolint:unused
+func (cpcs CollatorProtocolCollatorSide) processMessage(msg any) error { //nolint
 	// run this function as a goroutine, ideally
 
 	switch msg := msg.(type) {