Aggregator Committee - Fix committee runners management

ssv/committee.go

@@ -13,91 +13,96 @@ import (
 type CreateRunnerFn func(shareMap map[phase0.ValidatorIndex]*types.Share) Runner
 
 type Committee struct {
-	Runners         map[phase0.Slot]Runner
-	CommitteeMember types.CommitteeMember
-	CreateRunnerFn  CreateRunnerFn
-	Share           map[phase0.ValidatorIndex]*types.Share
+	CommitteeRunners                  map[phase0.Slot]Runner
+	AggregatorCommitteeRunners        map[phase0.Slot]Runner
+	CommitteeMember                   types.CommitteeMember
+	CreateCommitteeRunnerFn           CreateRunnerFn
+	CreateAggregatorCommitteeRunnerFn CreateRunnerFn
+	Share                             map[phase0.ValidatorIndex]*types.Share
 }
 
 // NewCommittee creates a new cluster
 func NewCommittee(
 	committeeMember types.CommitteeMember,
 	share map[phase0.ValidatorIndex]*types.Share,
 	createRunnerFn CreateRunnerFn,
+	createAggregatorCommitteeRunnerFn CreateRunnerFn,
 ) *Committee {
 	c := &Committee{
-		Runners:         make(map[phase0.Slot]Runner),
-		CommitteeMember: committeeMember,
-		CreateRunnerFn:  createRunnerFn,
-		Share:           share,
+		CommitteeRunners:                  make(map[phase0.Slot]Runner),
+		AggregatorCommitteeRunners:        make(map[phase0.Slot]Runner),
+		CommitteeMember:                   committeeMember,
+		CreateCommitteeRunnerFn:           createRunnerFn,
+		CreateAggregatorCommitteeRunnerFn: createAggregatorCommitteeRunnerFn,
+		Share:                             share,
 	}
 	return c
 }
 
 // StartDuty starts a new duty for the given slot
 func (c *Committee) StartDuty(duty types.Duty) error {
+
 	slot := duty.DutySlot()
-	if _, exists := c.Runners[slot]; exists {
-		return fmt.Errorf("Runner for slot %d already exists", slot)
-	}
 
-	// Handle different duty types
+	// Get objects according to duty type
+	var runnerMap *map[phase0.Slot]Runner
+	var createFn *CreateRunnerFn
+	var validatorDuties []*types.ValidatorDuty
 	switch d := duty.(type) {
 	case *types.CommitteeDuty:
-		if len(d.ValidatorDuties) == 0 {
-			return types.NewError(types.NoBeaconDutiesErrorCode, "no beacon duties")
-		}
-
-		// Filter duty and create share map according validators that belong to c.Share
-		dutyShares := make(map[phase0.ValidatorIndex]*types.Share)
-		filteredDuty := &types.CommitteeDuty{
-			Slot: d.Slot,
-		}
+		runnerMap = &c.CommitteeRunners
+		createFn = &c.CreateCommitteeRunnerFn
+		validatorDuties = d.ValidatorDuties
+	case *types.AggregatorCommitteeDuty:
+		runnerMap = &c.AggregatorCommitteeRunners
+		createFn = &c.CreateAggregatorCommitteeRunnerFn
+		validatorDuties = d.ValidatorDuties
+	default:
+		return errors.Errorf("unsupported duty type: %T", duty)
+	}
 
-		for _, bduty := range d.ValidatorDuties {
-			if _, exists := c.Share[bduty.ValidatorIndex]; !exists {
-				continue
-			}
-			dutyShares[bduty.ValidatorIndex] = c.Share[bduty.ValidatorIndex]
-			filteredDuty.ValidatorDuties = append(filteredDuty.ValidatorDuties, bduty)
-		}
+	if _, exists := (*runnerMap)[slot]; exists {
+		return fmt.Errorf("Runner for slot %d already exists", slot)
+	}
 
-		if len(dutyShares) == 0 {
-			return types.NewError(types.NoValidatorSharesErrorCode, "no shares for duty's validators")
-		}
+	if len(validatorDuties) == 0 {
+		return types.NewError(types.NoBeaconDutiesErrorCode, "no beacon duties")
+	}
 
-		c.Runners[slot] = c.CreateRunnerFn(dutyShares)
-		return c.Runners[slot].StartNewDuty(filteredDuty, c.CommitteeMember.GetQuorum())
+	// Filter duty and create share map according validators that belong to c.Share
+	dutyShares := make(map[phase0.ValidatorIndex]*types.Share)
+	filteredValidatorDuties := make([]*types.ValidatorDuty, 0)
 
-	case *types.AggregatorCommitteeDuty:
-		if len(d.ValidatorDuties) == 0 {
-			return types.NewError(types.NoBeaconDutiesErrorCode, "no beacon duties")
+	for _, bduty := range validatorDuties {
+		if _, exists := c.Share[bduty.ValidatorIndex]; !exists {
+			continue
 		}
+		dutyShares[bduty.ValidatorIndex] = c.Share[bduty.ValidatorIndex]
+		filteredValidatorDuties = append(filteredValidatorDuties, bduty)
+	}
 
-		// Filter duty and create share map according to validators that belong to c.Share
-		dutyShares := make(map[phase0.ValidatorIndex]*types.Share)
-		filteredDuty := &types.AggregatorCommitteeDuty{
-			Slot: d.Slot,
-		}
+	if len(dutyShares) == 0 {
+		return types.NewError(types.NoValidatorSharesErrorCode, "no shares for duty's validators")
+	}
 
-		for _, bduty := range d.ValidatorDuties {
-			if _, exists := c.Share[bduty.ValidatorIndex]; !exists {
-				continue
-			}
-			dutyShares[bduty.ValidatorIndex] = c.Share[bduty.ValidatorIndex]
-			filteredDuty.ValidatorDuties = append(filteredDuty.ValidatorDuties, bduty)
+	var filteredDuty types.Duty
+	switch duty.(type) {
+	case *types.CommitteeDuty:
+		filteredDuty = &types.CommitteeDuty{
+			Slot:            slot,
+			ValidatorDuties: filteredValidatorDuties,
 		}
-
-		if len(dutyShares) == 0 {
-			return types.NewError(types.NoValidatorSharesErrorCode, "no shares for duty's validators")
+	case *types.AggregatorCommitteeDuty:
+		filteredDuty = &types.AggregatorCommitteeDuty{
+			Slot:            slot,
+			ValidatorDuties: filteredValidatorDuties,
 		}
-
-		c.Runners[slot] = c.CreateRunnerFn(dutyShares)
-		return c.Runners[slot].StartNewDuty(filteredDuty, c.CommitteeMember.GetQuorum())
-
 	default:
 		return errors.Errorf("unsupported duty type: %T", duty)
 	}
+
+	(*runnerMap)[slot] = (*createFn)(dutyShares)
+	return (*runnerMap)[slot].StartNewDuty(filteredDuty, c.CommitteeMember.GetQuorum())
 }
 
 // ProcessMessage processes Network Message of all types
@@ -117,6 +122,18 @@ func (c *Committee) ProcessMessage(signedSSVMessage *types.SignedSSVMessage) err
 		return errors.Wrap(err, "Message invalid")
 	}
 
+	// Get runner map according to message role
+	var runnerMap *map[phase0.Slot]Runner
+	role := msg.MsgID.GetRoleType()
+	switch role {
+	case types.RoleCommittee:
+		runnerMap = &c.CommitteeRunners
+	case types.RoleAggregatorCommittee:
+		runnerMap = &c.AggregatorCommitteeRunners
+	default:
+		return types.NewError(types.CommitteeWrongRoleErrorCode, "msg role is invalid")
+	}
+
 	switch msg.GetType() {
 	case types.SSVConsensusMsgType:
 		qbftMsg := &qbft.Message{}
@@ -128,7 +145,7 @@ func (c *Committee) ProcessMessage(signedSSVMessage *types.SignedSSVMessage) err
 			return errors.Wrap(err, "invalid qbft Message")
 		}
 
-		runner, exists := c.Runners[phase0.Slot(qbftMsg.Height)]
+		runner, exists := (*runnerMap)[phase0.Slot(qbftMsg.Height)]
 		if !exists {
 			return types.NewError(types.NoRunnerForSlotErrorCode, "no runner found for message's slot")
 		}
@@ -148,7 +165,7 @@ func (c *Committee) ProcessMessage(signedSSVMessage *types.SignedSSVMessage) err
 			return errors.Wrap(err, "invalid PartialSignatureMessages")
 		}
 
-		runner, exists := c.Runners[pSigMessages.Slot]
+		runner, exists := (*runnerMap)[pSigMessages.Slot]
 		if !exists {
 			return types.NewError(types.NoRunnerForSlotErrorCode, "no runner found for message's slot")
 		}
@@ -157,6 +174,9 @@ func (c *Committee) ProcessMessage(signedSSVMessage *types.SignedSSVMessage) err
 		case types.PostConsensusPartialSig:
 			return runner.ProcessPostConsensus(pSigMessages)
 		case types.AggregatorCommitteePartialSig:
+			if role != types.RoleAggregatorCommittee {
+				return errors.Errorf("invalid aggregator partial sig msg for commmittee role")
+			}
 			return runner.ProcessPreConsensus(pSigMessages)
 		default:
 			return errors.Errorf("unknown partial signature message type: %v", pSigMessages.Type)
@@ -171,6 +191,11 @@ func (c *Committee) validateMessage(msg *types.SSVMessage) error {
 		return types.NewError(types.MessageIDCommitteeIDMismatchErrorCode, "msg ID doesn't match committee ID")
 	}
 
+	role := msg.GetID().GetRoleType()
+	if role != types.RoleCommittee && role != types.RoleAggregatorCommittee {
+		return types.NewError(types.CommitteeWrongRoleErrorCode, "msg role is invalid")
+	}
+
 	if len(msg.GetData()) == 0 {
 		return fmt.Errorf("msg data is invalid")
 	}

ssv/json_testutils.go

@@ -131,16 +131,18 @@ func (c *Committee) GetRoot() ([32]byte, error) {
 func (c *Committee) MarshalJSON() ([]byte, error) {
 
 	type CommitteeAlias struct {
-		Runners         map[phase0.Slot]Runner
-		CommitteeMember types.CommitteeMember
-		Share           map[phase0.ValidatorIndex]*types.Share
+		CommitteeRunners           map[phase0.Slot]Runner
+		AggregatorCommitteeRunners map[phase0.Slot]Runner
+		CommitteeMember            types.CommitteeMember
+		Share                      map[phase0.ValidatorIndex]*types.Share
 	}
 
 	// Create object and marshal
 	alias := &CommitteeAlias{
-		Runners:         c.Runners,
-		CommitteeMember: c.CommitteeMember,
-		Share:           c.Share,
+		CommitteeRunners:           c.CommitteeRunners,
+		AggregatorCommitteeRunners: c.AggregatorCommitteeRunners,
+		CommitteeMember:            c.CommitteeMember,
+		Share:                      c.Share,
 	}
 
 	byts, err := json.Marshal(alias)
@@ -151,9 +153,10 @@ func (c *Committee) MarshalJSON() ([]byte, error) {
 func (c *Committee) UnmarshalJSON(data []byte) error {
 	// First, unmarshal to get the raw JSON for runners
 	type CommitteeRaw struct {
-		Runners         map[phase0.Slot]json.RawMessage
-		CommitteeMember types.CommitteeMember
-		Share           map[phase0.ValidatorIndex]*types.Share
+		CommitteeRunners           map[phase0.Slot]json.RawMessage
+		AggregatorCommitteeRunners map[phase0.Slot]json.RawMessage
+		CommitteeMember            types.CommitteeMember
+		Share                      map[phase0.ValidatorIndex]*types.Share
 	}
 
 	raw := &CommitteeRaw{}
@@ -162,39 +165,24 @@ func (c *Committee) UnmarshalJSON(data []byte) error {
 	}
 
 	// Initialize the committee
-	c.Runners = make(map[phase0.Slot]Runner)
+	c.CommitteeRunners = make(map[phase0.Slot]Runner)
+	c.AggregatorCommitteeRunners = make(map[phase0.Slot]Runner)
 	c.CommitteeMember = raw.CommitteeMember
 	c.Share = raw.Share
 
-	// For each runner, detect its type and unmarshal accordingly
-	for slot, runnerData := range raw.Runners {
-		// Try to detect the runner type by looking for type-specific fields
-		var typeDetector struct {
-			BaseRunner struct {
-				RunnerRoleType types.RunnerRole `json:"RunnerRoleType"`
-			} `json:"BaseRunner"`
-		}
-
-		if err := json.Unmarshal(runnerData, &typeDetector); err != nil {
+	for slot, runnerData := range raw.CommitteeRunners {
+		var runner CommitteeRunner
+		if err := json.Unmarshal(runnerData, &runner); err != nil {
 			return err
 		}
-
-		switch typeDetector.BaseRunner.RunnerRoleType {
-		case types.RoleCommittee:
-			var runner CommitteeRunner
-			if err := json.Unmarshal(runnerData, &runner); err != nil {
-				return err
-			}
-			c.Runners[slot] = &runner
-		case types.RoleAggregatorCommittee:
-			var runner AggregatorCommitteeRunner
-			if err := json.Unmarshal(runnerData, &runner); err != nil {
-				return err
-			}
-			c.Runners[slot] = &runner
-		default:
-			return errors.Errorf("unknown runner type for slot %d: RunnerRoleType=%v", slot, typeDetector.BaseRunner.RunnerRoleType)
+		c.CommitteeRunners[slot] = &runner
+	}
+	for slot, runnerData := range raw.AggregatorCommitteeRunners {
+		var runner AggregatorCommitteeRunner
+		if err := json.Unmarshal(runnerData, &runner); err != nil {
+			return err
 		}
+		c.AggregatorCommitteeRunners[slot] = &runner
 	}
 
 	return nil

ssv/spectest/all_tests.go

@@ -3,6 +3,8 @@ package spectest
 import (
 	"github.com/ssvlabs/ssv-spec/ssv/spectest/tests"
 	aggregatorcommitteesingleduty "github.com/ssvlabs/ssv-spec/ssv/spectest/tests/aggregatorcommittee/singleduty"
+	"github.com/ssvlabs/ssv-spec/ssv/spectest/tests/committee"
+	committeemixedduties "github.com/ssvlabs/ssv-spec/ssv/spectest/tests/committee/mixedduties"
 	committeemultipleduty "github.com/ssvlabs/ssv-spec/ssv/spectest/tests/committee/multipleduty"
 	committeesingleduty "github.com/ssvlabs/ssv-spec/ssv/spectest/tests/committee/singleduty"
 	"github.com/ssvlabs/ssv-spec/ssv/spectest/tests/dutyexe"
@@ -79,6 +81,10 @@ var AllTests = []tests.TestF{
 	newduty.DuplicateDutyNotFinished,
 	newduty.FirstHeight,
 
+	committee.InvalidSig,
+	committee.CommitteeIDMismatch,
+	committee.WrongRole,
+
 	committeesingleduty.StartDuty,
 	committeesingleduty.StartNoDuty,
 	committeesingleduty.ValidBeaconVote,
@@ -104,6 +110,9 @@ var AllTests = []tests.TestF{
 	aggregatorcommitteesingleduty.DutyWithDifferentSlots,
 	aggregatorcommitteesingleduty.MaxValidators,
 
+	committeemixedduties.SameSlot,
+	committeemixedduties.MixedMultipleSlots,
+
 	consensus.FutureDecidedNoInstance,
 	consensus.FutureDecided,
 	consensus.InvalidDecidedValue,

ssv/spectest/generate/state_comparison/committee_CommitteeSpecTest/empty committee duty.json

@@ -1,5 +1,6 @@
 {
-  "Runners": {},
+  "CommitteeRunners": {},
+  "AggregatorCommitteeRunners": {},
   "CommitteeMember": {
     "OperatorID": 1,
     "CommitteeID": "cf97adeedb59e05bfd73a2b4c2a8885708c4f4f70c84c64b27120e72ab733b72",

ssv/spectest/generate/state_comparison/committee_CommitteeSpecTest/start with no shares for duty.json

@@ -1,5 +1,6 @@
 {
-  "Runners": {},
+  "CommitteeRunners": {},
+  "AggregatorCommitteeRunners": {},
   "CommitteeMember": {
     "OperatorID": 1,
     "CommitteeID": "cf97adeedb59e05bfd73a2b4c2a8885708c4f4f70c84c64b27120e72ab733b72",

ssv/spectest/generate/state_comparison/committee_MultiCommitteeSpecTest/aggregator_committee_runner_duty_with_different_slots/aggregator committee mixed (electra).json

@@ -1,5 +1,6 @@
 {
-  "Runners": {
+  "CommitteeRunners": {},
+  "AggregatorCommitteeRunners": {
     "7424012": {
       "BaseRunner": {
         "State": {

ssv/spectest/generate/state_comparison/committee_MultiCommitteeSpecTest/aggregator_committee_runner_duty_with_different_slots/aggregator committee mixed (phase0).json

@@ -1,5 +1,6 @@
 {
-  "Runners": {
+  "CommitteeRunners": {},
+  "AggregatorCommitteeRunners": {
     "12": {
       "BaseRunner": {
         "State": {

ssv/spectest/generate/state_comparison/committee_MultiCommitteeSpecTest/aggregator_committee_runner_happy_flow/1 aggregator (electra).json

@@ -1,5 +1,6 @@
 {
-  "Runners": {
+  "CommitteeRunners": {},
+  "AggregatorCommitteeRunners": {
     "7424012": {
       "BaseRunner": {
         "State": {

ssv/spectest/generate/state_comparison/committee_MultiCommitteeSpecTest/aggregator_committee_runner_happy_flow/1 aggregator (phase0).json

@@ -1,5 +1,6 @@
 {
-  "Runners": {
+  "CommitteeRunners": {},
+  "AggregatorCommitteeRunners": {
     "12": {
       "BaseRunner": {
         "State": {

ssv/spectest/generate/state_comparison/committee_MultiCommitteeSpecTest/aggregator_committee_runner_happy_flow/1 aggregator and sync committee contribution (electra).json

@@ -1,5 +1,6 @@
 {
-  "Runners": {
+  "CommitteeRunners": {},
+  "AggregatorCommitteeRunners": {
     "7424012": {
       "BaseRunner": {
         "State": {

ssv/spectest/generate/state_comparison/committee_MultiCommitteeSpecTest/aggregator_committee_runner_happy_flow/1 aggregator and sync committee contribution (phase0).json

@@ -1,5 +1,6 @@
 {
-  "Runners": {
+  "CommitteeRunners": {},
+  "AggregatorCommitteeRunners": {
     "12": {
       "BaseRunner": {
         "State": {