CommitteeDutyGuard: share between committees

operator/validator/controller.go

@@ -19,8 +19,6 @@ import (
 	genesisspectypes "github.com/ssvlabs/ssv-spec-pre-cc/types"
 	specqbft "github.com/ssvlabs/ssv-spec/qbft"
 	spectypes "github.com/ssvlabs/ssv-spec/types"
-	"go.uber.org/zap"
-
 	"github.com/ssvlabs/ssv/exporter/convert"
 	"github.com/ssvlabs/ssv/ibft/genesisstorage"
 	"github.com/ssvlabs/ssv/ibft/storage"
@@ -59,6 +57,7 @@ import (
 	ssvtypes "github.com/ssvlabs/ssv/protocol/v2/types"
 	registrystorage "github.com/ssvlabs/ssv/registry/storage"
 	"github.com/ssvlabs/ssv/storage/basedb"
+	"go.uber.org/zap"
 )
 
 //go:generate mockgen -package=mocks -destination=./mocks/controller.go -source=./controller.go
@@ -199,6 +198,7 @@ type controller struct {
 	validatorsMap           *validators.ValidatorsMap
 	validatorStartFunc      func(validator *validators.ValidatorContainer) (bool, error)
 	committeeValidatorSetup chan struct{}
+	dutyGuard               *validator.CommitteeDutyGuard
 
 	metadataUpdateInterval time.Duration
 
@@ -323,6 +323,7 @@ func NewController(logger *zap.Logger, options ControllerOptions) Controller {
 		indicesChange:           make(chan struct{}),
 		validatorExitCh:         make(chan duties.ExitDescriptor),
 		committeeValidatorSetup: make(chan struct{}, 1),
+		dutyGuard:               validator.NewCommitteeDutyGuard(),
 
 		messageValidator: options.MessageValidator,
 	}
@@ -964,7 +965,16 @@ func (c *controller) onShareInit(share *ssvtypes.SSVShare) (*validators.Validato
 
 		committeeRunnerFunc := SetupCommitteeRunners(ctx, opts)
 
-		vc = validator.NewCommittee(ctx, cancel, logger, c.beacon.GetBeaconNetwork(), operator, committeeRunnerFunc, nil)
+		vc = validator.NewCommittee(
+			ctx,
+			cancel,
+			logger,
+			c.beacon.GetBeaconNetwork(),
+			operator,
+			committeeRunnerFunc,
+			nil,
+			c.dutyGuard,
+		)
 		vc.AddShare(&share.Share)
 		c.validatorsMap.PutCommittee(operator.CommitteeID, vc)
 

protocol/v2/ssv/spectest/msg_processing_type.go

@@ -15,9 +15,6 @@ import (
 	spectypes "github.com/ssvlabs/ssv-spec/types"
 	spectestingutils "github.com/ssvlabs/ssv-spec/types/testingutils"
 	typescomparable "github.com/ssvlabs/ssv-spec/types/testingutils/comparable"
-	"github.com/stretchr/testify/require"
-	"go.uber.org/zap"
-
 	"github.com/ssvlabs/ssv/integration/qbft/tests"
 	"github.com/ssvlabs/ssv/logging"
 	"github.com/ssvlabs/ssv/networkconfig"
@@ -27,6 +24,8 @@ import (
 	ssvprotocoltesting "github.com/ssvlabs/ssv/protocol/v2/ssv/testing"
 	"github.com/ssvlabs/ssv/protocol/v2/ssv/validator"
 	protocoltesting "github.com/ssvlabs/ssv/protocol/v2/testing"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
 )
 
 type MsgProcessingSpecTest struct {
@@ -248,7 +247,7 @@ var baseCommitteeWithRunnerSample = func(
 	logger *zap.Logger,
 	keySetMap map[phase0.ValidatorIndex]*spectestingutils.TestKeySet,
 	runnerSample *runner.CommitteeRunner,
-	committeeDutyGuard runner.CommitteeDutyGuard,
+	committeeDutyGuard *validator.CommitteeDutyGuard,
 ) *validator.Committee {
 
 	var keySetSample *spectestingutils.TestKeySet
@@ -292,6 +291,7 @@ var baseCommitteeWithRunnerSample = func(
 		spectestingutils.TestingCommitteeMember(keySetSample),
 		createRunnerF,
 		shareMap,
+		committeeDutyGuard,
 	)
 
 	return c

protocol/v2/ssv/spectest/ssv_mapping_test.go

@@ -21,9 +21,6 @@ import (
 	spectypes "github.com/ssvlabs/ssv-spec/types"
 	"github.com/ssvlabs/ssv-spec/types/testingutils"
 	spectestingutils "github.com/ssvlabs/ssv-spec/types/testingutils"
-	"github.com/stretchr/testify/require"
-	"go.uber.org/zap"
-
 	"github.com/ssvlabs/ssv/exporter/convert"
 	tests2 "github.com/ssvlabs/ssv/integration/qbft/tests"
 	"github.com/ssvlabs/ssv/logging"
@@ -35,6 +32,8 @@ import (
 	ssvtesting "github.com/ssvlabs/ssv/protocol/v2/ssv/testing"
 	"github.com/ssvlabs/ssv/protocol/v2/ssv/validator"
 	protocoltesting "github.com/ssvlabs/ssv/protocol/v2/testing"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
 )
 
 func TestSSVMapping(t *testing.T) {
@@ -564,6 +563,7 @@ func fixCommitteeForRun(t *testing.T, ctx context.Context, logger *zap.Logger, c
 			return r.(*runner.CommitteeRunner), nil
 		},
 		specCommittee.Share,
+		validator.NewCommitteeDutyGuard(),
 	)
 	tmpSsvCommittee := &validator.Committee{}
 	require.NoError(t, json.Unmarshal(byts, tmpSsvCommittee))

protocol/v2/ssv/testing/runner.go

@@ -8,17 +8,15 @@ import (
 	specqbft "github.com/ssvlabs/ssv-spec/qbft"
 	spectypes "github.com/ssvlabs/ssv-spec/types"
 	spectestingutils "github.com/ssvlabs/ssv-spec/types/testingutils"
-	"go.uber.org/zap"
-
-	"github.com/ssvlabs/ssv/protocol/v2/qbft/controller"
-	"github.com/ssvlabs/ssv/protocol/v2/ssv"
-
 	"github.com/ssvlabs/ssv/exporter/convert"
 	"github.com/ssvlabs/ssv/integration/qbft/tests"
 	"github.com/ssvlabs/ssv/networkconfig"
+	"github.com/ssvlabs/ssv/protocol/v2/qbft/controller"
 	"github.com/ssvlabs/ssv/protocol/v2/qbft/testing"
+	"github.com/ssvlabs/ssv/protocol/v2/ssv"
 	"github.com/ssvlabs/ssv/protocol/v2/ssv/runner"
 	"github.com/ssvlabs/ssv/protocol/v2/ssv/validator"
+	"go.uber.org/zap"
 )
 
 var TestingHighestDecidedSlot = phase0.Slot(0)
@@ -117,6 +115,7 @@ var ConstructBaseRunner = func(
 
 	shareMap := make(map[phase0.ValidatorIndex]*spectypes.Share)
 	shareMap[share.ValidatorIndex] = share
+	dutyGuard := validator.NewCommitteeDutyGuard()
 
 	var r runner.Runner
 	var err error
@@ -132,7 +131,7 @@ var ConstructBaseRunner = func(
 			km,
 			opSigner,
 			valCheck,
-			validator.NewCommitteeDutyGuard(),
+			dutyGuard,
 		)
 	case spectypes.RoleAggregator:
 		r, err = runner.NewAggregatorRunner(
@@ -204,7 +203,7 @@ var ConstructBaseRunner = func(
 			km,
 			opSigner,
 			valCheck,
-			validator.NewCommitteeDutyGuard(),
+			dutyGuard,
 		)
 		r.(*runner.CommitteeRunner).BaseRunner.RunnerRoleType = spectestingutils.UnknownDutyType
 	default:
@@ -300,6 +299,7 @@ var ConstructBaseRunnerWithShareMap = func(
 	var contr *controller.Controller
 
 	km := spectestingutils.NewTestingKeyManager()
+	dutyGuard := validator.NewCommitteeDutyGuard()
 
 	if len(shareMap) > 0 {
 		var keySetInstance *spectestingutils.TestKeySet
@@ -382,7 +382,7 @@ var ConstructBaseRunnerWithShareMap = func(
 			km,
 			opSigner,
 			valCheck,
-			validator.NewCommitteeDutyGuard(),
+			dutyGuard,
 		)
 	case spectypes.RoleAggregator:
 		r, err = runner.NewAggregatorRunner(
@@ -454,7 +454,7 @@ var ConstructBaseRunnerWithShareMap = func(
 			km,
 			opSigner,
 			valCheck,
-			validator.NewCommitteeDutyGuard(),
+			dutyGuard,
 		)
 		if r != nil {
 			r.(*runner.CommitteeRunner).BaseRunner.RunnerRoleType = spectestingutils.UnknownDutyType

protocol/v2/ssv/validator/committee.go

@@ -12,14 +12,13 @@ import (
 	"github.com/pkg/errors"
 	"github.com/ssvlabs/ssv-spec/qbft"
 	spectypes "github.com/ssvlabs/ssv-spec/types"
-	"go.uber.org/zap"
-
 	"github.com/ssvlabs/ssv/ibft/storage"
 	"github.com/ssvlabs/ssv/logging/fields"
 	"github.com/ssvlabs/ssv/protocol/v2/message"
 	"github.com/ssvlabs/ssv/protocol/v2/ssv/queue"
 	"github.com/ssvlabs/ssv/protocol/v2/ssv/runner"
 	"github.com/ssvlabs/ssv/protocol/v2/types"
+	"go.uber.org/zap"
 )
 
 var (
@@ -57,6 +56,7 @@ func NewCommittee(
 	committeeMember *spectypes.CommitteeMember,
 	createRunnerFn CommitteeRunnerFunc,
 	shares map[phase0.ValidatorIndex]*spectypes.Share,
+	dutyGuard *CommitteeDutyGuard,
 ) *Committee {
 	if shares == nil {
 		shares = make(map[phase0.ValidatorIndex]*spectypes.Share)
@@ -71,7 +71,7 @@ func NewCommittee(
 		Shares:          shares,
 		CommitteeMember: committeeMember,
 		CreateRunnerFn:  createRunnerFn,
-		dutyGuard:       NewCommitteeDutyGuard(),
+		dutyGuard:       dutyGuard,
 	}
 }
 

protocol/v2/ssv/validator/committee_guard.go

@@ -33,6 +33,11 @@ func (a *CommitteeDutyGuard) StartDuty(role spectypes.BeaconRole, validator spec
 	if !ok {
 		return fmt.Errorf("unsupported role %d", role)
 	}
+	// If an older committee duty is still running for this validator we won't be interested in it
+	// anymore now that we have a fresher duty started. The older duty might or might not finish
+	// successfully, either outcome is fine but since it's always better to execute the freshest
+	// committee duty CommitteeDutyGuard will invalidate the older duty (potentially preventing it
+	// from execution so that we don't waste resources on it).
 	runningSlot, exists := duties[validator]
 	if exists && runningSlot >= slot {
 		return fmt.Errorf("duty already running at slot %d", runningSlot)