make AddShareToCommittee atomic

operator/validator/controller.go

@@ -636,7 +636,7 @@ func (c *controller) UpdateValidatorsMetadata(data map[spectypes.ValidatorPK]*be
 				c.logger.Warn("could not start validator", zap.Error(err))
 			}
 
-			_, found = c.validatorsMap.AddShareToCommittee(share)
+			_, found = c.validatorsMap.AddShareToCommittee(share, nil)
 
 			if !found {
 				c.logger.Warn("committee not found", fields.PubKey(share.ValidatorPubKey[:]))
@@ -828,9 +828,7 @@ func (c *controller) onShareInit(share *ssvtypes.SSVShare) (*validator.Validator
 	}
 
 	// Start a committee validator.
-	vc, found := c.validatorsMap.AddShareToCommittee(share)
-
-	if !found {
+	vc, found := c.validatorsMap.AddShareToCommittee(share, func() {
 		// Share context with both the validator and the runners,
 		// so that when the validator is stopped, the runners are stopped as well.
 		ctx, cancel := context.WithCancel(c.ctx)
@@ -847,7 +845,7 @@ func (c *controller) onShareInit(share *ssvtypes.SSVShare) (*validator.Validator
 
 		committeeRunnerFunc := SetupCommitteeRunners(ctx, opts)
 
-		vc = validator.NewCommittee(
+		comittee := validator.NewCommittee(
 			ctx,
 			cancel,
 			logger,
@@ -857,9 +855,11 @@ func (c *controller) onShareInit(share *ssvtypes.SSVShare) (*validator.Validator
 			nil,
 			c.dutyGuard,
 		)
-		vc.AddShare(&share.Share)
-		c.validatorsMap.PutCommittee(operator.CommitteeID, vc)
+		comittee.AddShare(&share.Share)
+		c.validatorsMap.PutCommittee(operator.CommitteeID, comittee)
+	})
 
+	if !found {
 		c.printShare(share, "setup committee done")
 	} else {
 		c.printShare(share, "added share to committee")

operator/validators/validators_map.go

@@ -178,17 +178,17 @@ func (vm *ValidatorsMap) RemoveShareFromCommittee(share *types.SSVShare) (*valid
 }
 
 // AddShareToCommittee adds share to its committee
-func (vm *ValidatorsMap) AddShareToCommittee(share *types.SSVShare) (*validator.Committee, bool) {
+func (vm *ValidatorsMap) AddShareToCommittee(share *types.SSVShare, onMissingCommittee func()) (*validator.Committee, bool) {
 	vm.mlock.Lock()
 	defer vm.mlock.Unlock()
 
 	vc, found := vm.committees[share.CommitteeID()]
 	if !found || vc.Stopped() {
+		onMissingCommittee()
 		return nil, false
 	}
 
 	vc.AddShare(&share.Share)
-
 	return vc, true
 }
 

operator/validators/validators_map_test.go

@@ -26,7 +26,7 @@ func TestAddShareToCommittee(t *testing.T) {
 			},
 		}
 
-		vc, added := vm.AddShareToCommittee(share)
+		vc, added := vm.AddShareToCommittee(share, nil)
 
 		assert.Nil(t, vc)
 		assert.False(t, added)
@@ -55,7 +55,7 @@ func TestAddShareToCommittee(t *testing.T) {
 		vm.PutCommittee(cmtID, cmt)
 		assert.Equal(t, 1, vm.SizeCommittees())
 
-		vc, added := vm.AddShareToCommittee(share)
+		vc, added := vm.AddShareToCommittee(share, nil)
 
 		assert.Equal(t, cmt, vc)
 		assert.True(t, added)