feat(dot/parachain): added overseer signals

dot/parachain/availability-store/availability_store.go

@@ -9,6 +9,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"sync"
 
 	parachaintypes "github.com/ChainSafe/gossamer/dot/parachain/types"
 	"github.com/ChainSafe/gossamer/internal/database"
@@ -28,6 +29,10 @@ const (
 
 // AvailabilityStoreSubsystem is the struct that holds subsystem data for the availability store
 type AvailabilityStoreSubsystem struct {
+	ctx    context.Context
+	cancel context.CancelFunc
+	wg     sync.WaitGroup
+
 	SubSystemToOverseer chan<- any
 	OverseerToSubSystem <-chan any
 	availabilityStore   AvailabilityStore
@@ -170,7 +175,11 @@ func uint32ToBytes(value uint32) []byte {
 // Run runs the availability store subsystem
 func (av *AvailabilityStoreSubsystem) Run(ctx context.Context, OverseerToSubsystem chan any,
 	SubsystemToOverseer chan any) error {
-	av.processMessages()
+
+	av.wg.Add(2)
+	go av.processMessages()
+	go av.ProcessOverseerSignals()
+
 	return nil
 }
 
@@ -180,53 +189,70 @@ func (*AvailabilityStoreSubsystem) Name() parachaintypes.SubSystemName {
 }
 
 func (av *AvailabilityStoreSubsystem) processMessages() {
-	for msg := range av.OverseerToSubSystem {
-		logger.Debugf("received message %v", msg)
-		switch msg := msg.(type) {
-		case QueryAvailableData:
-			err := av.handleQueryAvailableData(msg)
-			if err != nil {
-				logger.Errorf("failed to handle available data: %w", err)
-			}
-		case QueryDataAvailability:
-			err := av.handleQueryDataAvailability(msg)
-			if err != nil {
-				logger.Errorf("failed to handle query data availability: %w", err)
-			}
-		case QueryChunk:
-			err := av.handleQueryChunk(msg)
-			if err != nil {
-				logger.Errorf("failed to handle query chunk: %w", err)
-			}
-		case QueryChunkSize:
-			err := av.handleQueryChunkSize(msg)
-			if err != nil {
-				logger.Errorf("failed to handle query chunk size: %w", err)
-			}
-		case QueryAllChunks:
-			err := av.handleQueryAllChunks(msg)
-			if err != nil {
-				logger.Errorf("failed to handle query all chunks: %w", err)
-			}
-		case QueryChunkAvailability:
-			err := av.handleQueryChunkAvailability(msg)
-			if err != nil {
-				logger.Errorf("failed to handle query chunk availability: %w", err)
-			}
-		case StoreChunk:
-			err := av.handleStoreChunk(msg)
-			if err != nil {
-				logger.Errorf("failed to handle store chunk: %w", err)
+	for {
+		select {
+		case msg := <-av.OverseerToSubSystem:
+			logger.Debugf("received message %v", msg)
+			switch msg := msg.(type) {
+			case QueryAvailableData:
+				err := av.handleQueryAvailableData(msg)
+				if err != nil {
+					logger.Errorf("failed to handle available data: %w", err)
+				}
+			case QueryDataAvailability:
+				err := av.handleQueryDataAvailability(msg)
+				if err != nil {
+					logger.Errorf("failed to handle query data availability: %w", err)
+				}
+			case QueryChunk:
+				err := av.handleQueryChunk(msg)
+				if err != nil {
+					logger.Errorf("failed to handle query chunk: %w", err)
+				}
+			case QueryChunkSize:
+				err := av.handleQueryChunkSize(msg)
+				if err != nil {
+					logger.Errorf("failed to handle query chunk size: %w", err)
+				}
+			case QueryAllChunks:
+				err := av.handleQueryAllChunks(msg)
+				if err != nil {
+					logger.Errorf("failed to handle query all chunks: %w", err)
+				}
+			case QueryChunkAvailability:
+				err := av.handleQueryChunkAvailability(msg)
+				if err != nil {
+					logger.Errorf("failed to handle query chunk availability: %w", err)
+				}
+			case StoreChunk:
+				err := av.handleStoreChunk(msg)
+				if err != nil {
+					logger.Errorf("failed to handle store chunk: %w", err)
+				}
+			case StoreAvailableData:
+				err := av.handleStoreAvailableData(msg)
+				if err != nil {
+					logger.Errorf("failed to handle store available data: %w", err)
+				}
+
 			}
-		case StoreAvailableData:
-			err := av.handleStoreAvailableData(msg)
-			if err != nil {
-				logger.Errorf("failed to handle store available data: %w", err)
+
+		case <-av.ctx.Done():
+			if err := av.ctx.Err(); err != nil {
+				logger.Errorf("ctx error: %v\n", err)
 			}
+			av.wg.Done()
+			return
 		}
+
 	}
 }
 
+func (av *AvailabilityStoreSubsystem) ProcessOverseerSignals() {
+	av.wg.Done()
+	// TODO: #3630
+}
+
 func (av *AvailabilityStoreSubsystem) handleQueryAvailableData(msg QueryAvailableData) error {
 	result, err := av.availabilityStore.loadAvailableData(msg.CandidateHash)
 	if err != nil {
@@ -333,3 +359,8 @@ func (av *AvailabilityStoreSubsystem) handleStoreAvailableData(msg StoreAvailabl
 	msg.Sender <- err // TODO: determine how to replicate Rust's Result type
 	return nil
 }
+
+func (av *AvailabilityStoreSubsystem) Stop() {
+	av.cancel()
+	av.wg.Wait()
+}

dot/parachain/availability-store/availability_store_test.go

@@ -1,3 +1,6 @@
+// Copyright 2023 ChainSafe Systems (ON)
+// SPDX-License-Identifier: LGPL-3.0-only
+
 package availabilitystore
 
 import (

dot/parachain/backing/candidate_backing.go

@@ -5,6 +5,7 @@ package backing
 
 import (
 	"context"
+	"sync"
 
 	parachaintypes "github.com/ChainSafe/gossamer/dot/parachain/types"
 	"github.com/ChainSafe/gossamer/internal/log"
@@ -14,6 +15,10 @@ import (
 var logger = log.NewFromGlobal(log.AddContext("pkg", "parachain-candidate-backing"))
 
 type CandidateBacking struct {
+	ctx    context.Context
+	cancel context.CancelFunc
+	wg     sync.WaitGroup
+
 	SubSystemToOverseer chan<- any
 	OverseerToSubSystem <-chan any
 }
@@ -72,31 +77,50 @@ func (cb *CandidateBacking) Run(ctx context.Context, overseerToSubSystem chan an
 	// other backing related overseer message.
 	// This would become more clear after we complete processMessages function. It would give us clarity
 	// if we need background_validation_rx or background_validation_tx, as done in rust.
-	cb.processMessages()
+
+	cb.wg.Add(2)
+	go cb.processMessages()
+	go cb.ProcessOverseerSignals()
+
 	return nil
 }
 
 func (*CandidateBacking) Name() parachaintypes.SubSystemName {
 	return parachaintypes.CandidateBacking
 }
 
+func (cb *CandidateBacking) ProcessOverseerSignals() {
+	cb.wg.Done()
+	// TODO #3644
+}
+
 func (cb *CandidateBacking) processMessages() {
-	for msg := range cb.OverseerToSubSystem {
-		// process these received messages by referencing
-		// https://github.com/paritytech/polkadot-sdk/blob/769bdd3ff33a291cbc70a800a3830638467e42a2/polkadot/node/core/backing/src/lib.rs#L741
-		switch msg.(type) {
-		case ActiveLeavesUpdate:
-			cb.handleActiveLeavesUpdate()
-		case GetBackedCandidates:
-			cb.handleGetBackedCandidates()
-		case CanSecond:
-			cb.handleCanSecond()
-		case Second:
-			cb.handleSecond()
-		case Statement:
-			cb.handleStatement()
-		default:
-			logger.Error("unknown message type")
+	for {
+		select {
+		case msg := <-cb.OverseerToSubSystem:
+			// process these received messages by referencing
+			// https://github.com/paritytech/polkadot-sdk/blob/769bdd3ff33a291cbc70a800a3830638467e42a2/polkadot/node/core/backing/src/lib.rs#L741
+			switch msg.(type) {
+			case ActiveLeavesUpdate:
+				cb.handleActiveLeavesUpdate()
+			case GetBackedCandidates:
+				cb.handleGetBackedCandidates()
+			case CanSecond:
+				cb.handleCanSecond()
+			case Second:
+				cb.handleSecond()
+			case Statement:
+				cb.handleStatement()
+			default:
+				logger.Error("unknown message type")
+			}
+
+		case <-cb.ctx.Done():
+			if err := cb.ctx.Err(); err != nil {
+				logger.Errorf("ctx error: %v\n", err)
+			}
+			cb.wg.Done()
+			return
 		}
 	}
 }
@@ -121,6 +145,11 @@ func (cb *CandidateBacking) handleStatement() {
 	// TODO: Implement this #3507
 }
 
+func (cb *CandidateBacking) Stop() {
+	cb.cancel()
+	cb.wg.Wait()
+}
+
 // SignedFullStatementWithPVD represents a signed full statement along with associated Persisted Validation Data (PVD).
 type SignedFullStatementWithPVD struct {
 	SignedFullStatement     parachaintypes.UncheckedSignedFullStatement

dot/parachain/collator-protocol/validator_side.go

@@ -83,6 +83,11 @@ func (cpvs CollatorProtocolValidatorSide) Run(
 				cpvs.fetchedCollations = append(cpvs.fetchedCollations, *collation)
 			}
 
+		case <-cpvs.ctx.Done():
+			if err := cpvs.ctx.Err(); err != nil {
+				logger.Errorf("ctx error: %v\n", err)
+			}
+			return nil
 		}
 	}
 }
@@ -91,6 +96,14 @@ func (CollatorProtocolValidatorSide) Name() parachaintypes.SubSystemName {
 	return parachaintypes.CollationProtocol
 }
 
+func (cpvs CollatorProtocolValidatorSide) ProcessOverseerSignals() {
+	// NOTE: nothing to do here
+}
+
+func (cpvs CollatorProtocolValidatorSide) Stop() {
+	cpvs.cancel()
+}
+
 // requestCollation requests a collation from the network.
 // This function will
 // - check for duplicate requests
@@ -291,6 +304,9 @@ type CollationEvent struct {
 }
 
 type CollatorProtocolValidatorSide struct {
+	ctx    context.Context
+	cancel context.CancelFunc
+
 	net Network
 
 	SubSystemToOverseer chan<- any

dot/parachain/overseer/mocks_generate_test.go

@@ -0,0 +1,6 @@
+// Copyright 2023 ChainSafe Systems (ON)
+// SPDX-License-Identifier: LGPL-3.0-only
+
+package overseer
+
+//go:generate mockgen -destination=mocks_test.go -package=$GOPACKAGE . BlockState