feat(statement-distribution):Implement FanIn approach to statement distribution

dot/parachain/statement-distribution/statement_distribution.go

@@ -2,69 +2,84 @@ package statementdistribution
 
 import (
 	"context"
+	"time"
 
 	"github.com/ChainSafe/gossamer/internal/log"
 
 	statementedistributionmessages "github.com/ChainSafe/gossamer/dot/parachain/statement-distribution/messages"
 	parachaintypes "github.com/ChainSafe/gossamer/dot/parachain/types"
+	parachainutil "github.com/ChainSafe/gossamer/dot/parachain/util"
 )
 
 var logger = log.NewFromGlobal(log.AddContext("pkg", "statement-distribution"))
 
 type StatementDistribution struct {
+	reputationAggregator *parachainutil.ReputationAggregator
+	SubSystemToOverseer  chan<- parachainutil.NetworkBridgeTxMessage
 }
 
-func (s StatementDistribution) Run(ctx context.Context, overseerToSubSystem <-chan any) {
+func (s *StatementDistribution) Run(
+	ctx context.Context,
+	overseerToSubSystem <-chan any,
+	v2CommChannel <-chan any,
+	receiverRespCh <-chan any,
+	retryReqCh <-chan any,
+) {
+	// Timer for reputation aggregator trigger
+	reputationDelay := time.NewTicker(parachainutil.ReputationChangeInterval) // Adjust the duration as needed
+	defer reputationDelay.Stop()
+
 	for {
 		select {
-		case msg, ok := <-overseerToSubSystem:
-			if !ok {
-				return
-			}
+		case msg := <-overseerToSubSystem:
 			err := s.processMessage(msg)
 			if err != nil {
-				logger.Errorf("processing overseer message: %w", err)
+				logger.Errorf("error processing overseer message: %v", err)
 			}
+			// case _ = <-v2CommChannel:
+			// 	panic("Not Implemented")
+			// case _ = <-receiverRespCh:
+			// 	panic("Not Implemented")
+			// case _ = <-retryReqCh:
+			logger.Infof("received retry request, no action taken")
+		case <-reputationDelay.C:
+			// Trigger reputation aggregator logic
+			s.reputationAggregator.Send(s.SubSystemToOverseer)
 		case <-ctx.Done():
-			if err := ctx.Err(); err != nil {
-				logger.Errorf("ctx error: %v\n", err)
-			}
+			logger.Infof("shutting down: %v", ctx.Err())
+			return
 		}
 	}
 }
 
-func (s StatementDistribution) processMessage(msg any) error {
+func (s *StatementDistribution) processMessage(msg any) error {
 	switch msg := msg.(type) {
 	case statementedistributionmessages.Backed:
 		// TODO #4171
 	case statementedistributionmessages.Share:
 		// TODO #4170
-	// case statementedistributionmessages.NetworkBridgeUpdate
-	// TODO #4172 this above case would need to wait until network bridge receiver side is merged
 	case parachaintypes.ActiveLeavesUpdateSignal:
-		return s.ProcessActiveLeavesUpdateSignal(msg)
+		return s.processActiveLeavesUpdateSignal(msg)
 	case parachaintypes.BlockFinalizedSignal:
-		return s.ProcessBlockFinalizedSignal(msg)
-
+		return s.processBlockFinalizedSignal(msg)
 	default:
 		return parachaintypes.ErrUnknownOverseerMessage
 	}
-
 	return nil
 }
 
-func (s StatementDistribution) Name() parachaintypes.SubSystemName {
+func (s *StatementDistribution) Name() parachaintypes.SubSystemName {
 	return parachaintypes.StatementDistribution
 }
 
-func (s StatementDistribution) ProcessActiveLeavesUpdateSignal(signal parachaintypes.ActiveLeavesUpdateSignal) error {
+func (s *StatementDistribution) processActiveLeavesUpdateSignal(signal parachaintypes.ActiveLeavesUpdateSignal) error {
 	// TODO #4173
 	return nil
 }
 
-func (s StatementDistribution) ProcessBlockFinalizedSignal(signal parachaintypes.BlockFinalizedSignal) error {
+func (s *StatementDistribution) processBlockFinalizedSignal(signal parachaintypes.BlockFinalizedSignal) error {
 	// nothing to do here
 	return nil
 }
 
-func (s StatementDistribution) Stop() {}
+func (s *StatementDistribution) Stop() {}

dot/parachain/statement-distribution/statement_distribution_test.go

@@ -0,0 +1,22 @@
+package statementdistribution
+
+// import (
+// 	parachainutil "github.com/ChainSafe/gossamer/dot/parachain/util"
+// )
+
+// func createStatementDistribution() (*StatementDistribution, chan parachainutil.NetworkBridgeTxMessage) {
+// 	mockAggregator := parachainutil.NewReputationAggregator(func(rep parachainutil.UnifiedReputationChange) bool {
+// 		return rep.Type == parachainutil.Malicious
+// 	})
+
+// 	subSystemToOverseer := make(chan parachainutil.NetworkBridgeTxMessage, 10)
+
+// 	return &StatementDistribution{
+// 		reputationAggregator: mockAggregator,
+// 		SubSystemToOverseer:  subSystemToOverseer,
+// 	}, subSystemToOverseer
+// }
+
+// func createChannels() (chan any, chan any, chan any, chan any) {
+// 	return make(chan any, 10), make(chan any, 10), make(chan any, 10), make(chan any, 10)
+// }

dot/parachain/util/util.go

@@ -64,6 +64,8 @@ type UnifiedReputationChange struct {
 	Reason string
 }
 
+const ReputationChangeInterval = 30 * time.Second
+
 // CostOrBenefit returns the cost or benefit of the reputation change.
 func (u UnifiedReputationChange) CostOrBenefit() int32 {
 	switch u.Type {