updated duties using split endpoints

changelog/james-prysm_new-update-duties.md

@@ -0,0 +1,3 @@
+### Changed
+
+- validator client begins to call new separate endpoints for duties post gloas instead of get duties v2.

validator/client/aggregator_selector.go

@@ -253,7 +253,7 @@ func (p *distributedSelector) RefreshSelectionProofs(ctx context.Context) error
 
 func (p *distributedSelector) fetchSelectionProofs(ctx context.Context) (map[attSelectionKey]iface.BeaconCommitteeSelection, error) {
 	var req []iface.BeaconCommitteeSelection
-	for pk, duty := range p.v.duties.CurrentEpochDuties() {
+	for pk, duty := range p.v.duties.snapshot().currentDuties() {
 		if duty.Status != ethpb.ValidatorStatus_ACTIVE && duty.Status != ethpb.ValidatorStatus_EXITING {
 			continue
 		}

validator/client/aggregator_selector_test.go

@@ -168,14 +168,18 @@ func TestDistributedSelector_EpochGuard(t *testing.T) {
 	ds := v.aggSelector.(*distributedSelector)
 
 	slot := primitives.Slot(2) * params.BeaconConfig().SlotsPerEpoch
-	ds.v.duties.SetFromCombinedDutiesResponse(&ethpb.ValidatorDutiesContainer{
-		CurrentEpochDuties: []*ethpb.ValidatorDuty{{
-			AttesterSlot:   slot,
-			ValidatorIndex: 200,
-			PublicKey:      keys.pub[:],
-			Status:         ethpb.ValidatorStatus_ACTIVE,
-		}},
-	})
+	{
+		var data dutyStoreData
+		data.setFromContainer(&ethpb.ValidatorDutiesContainer{
+			CurrentEpochDuties: []*ethpb.ValidatorDuty{{
+				AttesterSlot:   slot,
+				ValidatorIndex: 200,
+				PublicKey:      keys.pub[:],
+				Status:         ethpb.ValidatorStatus_ACTIVE,
+			}},
+		})
+		ds.v.duties.write(data)
+	}
 
 	sigDomain := make([]byte, 32)
 	client.EXPECT().DomainData(gomock.Any(), gomock.Any()).
@@ -197,14 +201,18 @@ func TestDistributedSelector_ReadyCh_BlocksUntilRefresh(t *testing.T) {
 	ds := v.aggSelector.(*distributedSelector)
 
 	slot := primitives.Slot(3) * params.BeaconConfig().SlotsPerEpoch
-	ds.v.duties.SetFromCombinedDutiesResponse(&ethpb.ValidatorDutiesContainer{
-		CurrentEpochDuties: []*ethpb.ValidatorDuty{{
-			AttesterSlot:   slot,
-			ValidatorIndex: 200,
-			PublicKey:      keys.pub[:],
-			Status:         ethpb.ValidatorStatus_ACTIVE,
-		}},
-	})
+	{
+		var data dutyStoreData
+		data.setFromContainer(&ethpb.ValidatorDutiesContainer{
+			CurrentEpochDuties: []*ethpb.ValidatorDuty{{
+				AttesterSlot:   slot,
+				ValidatorIndex: 200,
+				PublicKey:      keys.pub[:],
+				Status:         ethpb.ValidatorStatus_ACTIVE,
+			}},
+		})
+		ds.v.duties.write(data)
+	}
 
 	proof := make([]byte, 96)
 	proof[0] = 0xAB
@@ -254,14 +262,18 @@ func TestDistributedSelector_ErrorIsStickyWithinEpoch(t *testing.T) {
 	ds := v.aggSelector.(*distributedSelector)
 
 	slot := primitives.Slot(4) * params.BeaconConfig().SlotsPerEpoch
-	ds.v.duties.SetFromCombinedDutiesResponse(&ethpb.ValidatorDutiesContainer{
-		CurrentEpochDuties: []*ethpb.ValidatorDuty{{
-			AttesterSlot:   slot,
-			ValidatorIndex: 200,
-			PublicKey:      keys.pub[:],
-			Status:         ethpb.ValidatorStatus_ACTIVE,
-		}},
-	})
+	{
+		var data dutyStoreData
+		data.setFromContainer(&ethpb.ValidatorDutiesContainer{
+			CurrentEpochDuties: []*ethpb.ValidatorDuty{{
+				AttesterSlot:   slot,
+				ValidatorIndex: 200,
+				PublicKey:      keys.pub[:],
+				Status:         ethpb.ValidatorStatus_ACTIVE,
+			}},
+		})
+		ds.v.duties.write(data)
+	}
 
 	sigDomain := make([]byte, 32)
 	client.EXPECT().DomainData(gomock.Any(), gomock.Any()).

validator/client/attest.go

@@ -191,12 +191,11 @@ func (v *validator) SubmitAttestation(ctx context.Context, slot primitives.Slot,
 
 // Given the validator public key, this gets the validator assignment.
 func (v *validator) duty(pubKey [fieldparams.BLSPubkeyLength]byte) (*ethpb.ValidatorDuty, error) {
-	v.dutiesLock.RLock()
-	defer v.dutiesLock.RUnlock()
-	if !v.duties.IsInitialized() {
+	snap := v.duties.snapshot()
+	if !snap.isInitialized() {
 		return nil, errors.New("no duties for validators")
 	}
-	d, ok := v.duties.CurrentDuty(pubKey)
+	d, ok := snap.currentDuty(pubKey)
 	if !ok {
 		return nil, fmt.Errorf("pubkey %#x not in duties", bytesutil.Trunc(pubKey[:]))
 	}

validator/client/beacon-api/beacon_api_validator_client.go

@@ -77,6 +77,38 @@ func (c *beaconApiValidatorClient) Duties(ctx context.Context, in *ethpb.DutiesR
 	})
 }
 
+func (c *beaconApiValidatorClient) AttesterDuties(ctx context.Context, epoch primitives.Epoch, validatorIndices []primitives.ValidatorIndex) (*ethpb.AttesterDutiesResponse, error) {
+	ctx, span := trace.StartSpan(ctx, "beacon-api.AttesterDuties")
+	defer span.End()
+	return wrapInMetrics[*ethpb.AttesterDutiesResponse]("AttesterDuties", func() (*ethpb.AttesterDutiesResponse, error) {
+		return c.attesterDuties(ctx, epoch, validatorIndices)
+	})
+}
+
+func (c *beaconApiValidatorClient) ProposerDuties(ctx context.Context, epoch primitives.Epoch) (*ethpb.ProposerDutiesResponse, error) {
+	ctx, span := trace.StartSpan(ctx, "beacon-api.ProposerDuties")
+	defer span.End()
+	return wrapInMetrics[*ethpb.ProposerDutiesResponse]("ProposerDuties", func() (*ethpb.ProposerDutiesResponse, error) {
+		return c.proposerDuties(ctx, epoch)
+	})
+}
+
+func (c *beaconApiValidatorClient) SyncCommitteeDuties(ctx context.Context, epoch primitives.Epoch, validatorIndices []primitives.ValidatorIndex) (*ethpb.SyncCommitteeDutiesResponse, error) {
+	ctx, span := trace.StartSpan(ctx, "beacon-api.SyncCommitteeDuties")
+	defer span.End()
+	return wrapInMetrics[*ethpb.SyncCommitteeDutiesResponse]("SyncCommitteeDuties", func() (*ethpb.SyncCommitteeDutiesResponse, error) {
+		return c.syncCommitteeDuties(ctx, epoch, validatorIndices)
+	})
+}
+
+func (c *beaconApiValidatorClient) PTCDuties(ctx context.Context, epoch primitives.Epoch, validatorIndices []primitives.ValidatorIndex) (*ethpb.PTCDutiesResponse, error) {
+	ctx, span := trace.StartSpan(ctx, "beacon-api.PTCDuties")
+	defer span.End()
+	return wrapInMetrics[*ethpb.PTCDutiesResponse]("PTCDuties", func() (*ethpb.PTCDutiesResponse, error) {
+		return c.ptcDuties(ctx, epoch, validatorIndices)
+	})
+}
+
 func (c *beaconApiValidatorClient) CheckDoppelGanger(ctx context.Context, in *ethpb.DoppelGangerRequest) (*ethpb.DoppelGangerResponse, error) {
 	ctx, span := trace.StartSpan(ctx, "beacon-api.CheckDoppelGanger")
 	defer span.End()

validator/client/beacon-api/duties.go

@@ -228,7 +228,7 @@ func (c *beaconApiValidatorClient) dutiesForEpoch(
 	return nil
 }
 
-func (c *beaconApiValidatorClient) AttesterDuties(ctx context.Context, epoch primitives.Epoch, validatorIndices []primitives.ValidatorIndex) (*ethpb.AttesterDutiesResponse, error) {
+func (c *beaconApiValidatorClient) attesterDuties(ctx context.Context, epoch primitives.Epoch, validatorIndices []primitives.ValidatorIndex) (*ethpb.AttesterDutiesResponse, error) {
 	resp, err := c.dutiesProvider.AttesterDuties(ctx, epoch, validatorIndices)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to get attester duties")
@@ -284,7 +284,7 @@ func (c *beaconApiValidatorClient) AttesterDuties(ctx context.Context, epoch pri
 	}, nil
 }
 
-func (c *beaconApiValidatorClient) ProposerDuties(ctx context.Context, epoch primitives.Epoch) (*ethpb.ProposerDutiesResponse, error) {
+func (c *beaconApiValidatorClient) proposerDuties(ctx context.Context, epoch primitives.Epoch) (*ethpb.ProposerDutiesResponse, error) {
 	resp, err := c.dutiesProvider.ProposerDuties(ctx, epoch)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to get proposer duties")
@@ -320,7 +320,7 @@ func (c *beaconApiValidatorClient) ProposerDuties(ctx context.Context, epoch pri
 	}, nil
 }
 
-func (c *beaconApiValidatorClient) SyncCommitteeDuties(ctx context.Context, epoch primitives.Epoch, validatorIndices []primitives.ValidatorIndex) (*ethpb.SyncCommitteeDutiesResponse, error) {
+func (c *beaconApiValidatorClient) syncCommitteeDuties(ctx context.Context, epoch primitives.Epoch, validatorIndices []primitives.ValidatorIndex) (*ethpb.SyncCommitteeDutiesResponse, error) {
 	syncDuties, err := c.dutiesProvider.SyncDuties(ctx, epoch, validatorIndices)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to get sync committee duties")
@@ -535,7 +535,7 @@ func (c beaconApiDutiesProvider) PTCDuties(ctx context.Context, epoch primitives
 	return &ptcDuties, nil
 }
 
-func (c *beaconApiValidatorClient) PTCDuties(ctx context.Context, epoch primitives.Epoch, validatorIndices []primitives.ValidatorIndex) (*ethpb.PTCDutiesResponse, error) {
+func (c *beaconApiValidatorClient) ptcDuties(ctx context.Context, epoch primitives.Epoch, validatorIndices []primitives.ValidatorIndex) (*ethpb.PTCDutiesResponse, error) {
 	resp, err := c.dutiesProvider.PTCDuties(ctx, epoch, validatorIndices)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to get PTC duties")