diff --git a/changelog/james-prysm_new-update-duties.md b/changelog/james-prysm_new-update-duties.md new file mode 100644 index 000000000000..3deaf3fa2021 --- /dev/null +++ b/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. \ No newline at end of file diff --git a/validator/client/aggregator_selector.go b/validator/client/aggregator_selector.go index ebc23bf6caa0..a6d308ae84d3 100644 --- a/validator/client/aggregator_selector.go +++ b/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 } diff --git a/validator/client/aggregator_selector_test.go b/validator/client/aggregator_selector_test.go index 5118ebf6d127..2426272bc702 100644 --- a/validator/client/aggregator_selector_test.go +++ b/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(ðpb.ValidatorDutiesContainer{ - CurrentEpochDuties: []*ethpb.ValidatorDuty{{ - AttesterSlot: slot, - ValidatorIndex: 200, - PublicKey: keys.pub[:], - Status: ethpb.ValidatorStatus_ACTIVE, - }}, - }) + { + var data dutyStoreData + data.setFromContainer(ðpb.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(ðpb.ValidatorDutiesContainer{ - CurrentEpochDuties: []*ethpb.ValidatorDuty{{ - AttesterSlot: slot, - ValidatorIndex: 200, - PublicKey: keys.pub[:], - Status: ethpb.ValidatorStatus_ACTIVE, - }}, - }) + { + var data dutyStoreData + data.setFromContainer(ðpb.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(ðpb.ValidatorDutiesContainer{ - CurrentEpochDuties: []*ethpb.ValidatorDuty{{ - AttesterSlot: slot, - ValidatorIndex: 200, - PublicKey: keys.pub[:], - Status: ethpb.ValidatorStatus_ACTIVE, - }}, - }) + { + var data dutyStoreData + data.setFromContainer(ðpb.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()). diff --git a/validator/client/attest.go b/validator/client/attest.go index b5a14767bb98..d69f7e224d7a 100644 --- a/validator/client/attest.go +++ b/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[:])) } diff --git a/validator/client/beacon-api/beacon_api_validator_client.go b/validator/client/beacon-api/beacon_api_validator_client.go index 80f7224e61e6..5a6668a2e5d2 100644 --- a/validator/client/beacon-api/beacon_api_validator_client.go +++ b/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() diff --git a/validator/client/beacon-api/duties.go b/validator/client/beacon-api/duties.go index 15a915aa5eee..75cb57c56bd6 100644 --- a/validator/client/beacon-api/duties.go +++ b/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") diff --git a/validator/client/duties.go b/validator/client/duties.go index 55294dbda2fd..d5c8226413ee 100644 --- a/validator/client/duties.go +++ b/validator/client/duties.go @@ -4,6 +4,8 @@ import ( "bytes" "context" "fmt" + "slices" + "sync" "time" "github.com/OffchainLabs/prysm/v7/api/server/structs" @@ -19,8 +21,9 @@ import ( "google.golang.org/grpc/metadata" ) -// filterBlacklistedKeys returns validating keys with slashable keys removed. -func (v *validator) filterBlacklistedKeys(ctx context.Context) ([][fieldparams.BLSPubkeyLength]byte, error) { +// nonBlacklistedKeys returns the keymanager's validating keys with +// slashing-protection-blacklisted keys removed. +func (v *validator) nonBlacklistedKeys(ctx context.Context) ([][fieldparams.BLSPubkeyLength]byte, error) { validatingKeys, err := v.km.FetchValidatingPublicKeys(ctx) if err != nil { return nil, err @@ -41,6 +44,44 @@ func (v *validator) filterBlacklistedKeys(ctx context.Context) ([][fieldparams.B return filtered, nil } +// isActiveForDuties reports whether a validator status entry indicates the +// validator currently has duties to perform — i.e. it is in (or about to be +// in) the beacon-state active set. Shared by filteredKeysAndIndices and +// filterAndCacheActiveKeys so both call sites agree on the same predicate. +func isActiveForDuties(s *ethpb.ValidatorStatusResponse, currEpoch primitives.Epoch) bool { + if s == nil { + return false + } + switch s.Status { + case ethpb.ValidatorStatus_ACTIVE, ethpb.ValidatorStatus_EXITING: + return true + case ethpb.ValidatorStatus_PENDING: + // Cache may be stale: include validators whose activation epoch has + // already arrived but whose status hasn't been refreshed yet. + return currEpoch >= s.ActivationEpoch + } + return false +} + +// filteredKeysAndIndices returns the subset of keys with duties to fetch for +// the given epoch (see isActiveForDuties), and the corresponding sorted +// validator indices. Sorted indices let callers compare against a previously +// stored set to detect drift. +func (v *validator) filteredKeysAndIndices(keys [][fieldparams.BLSPubkeyLength]byte, epoch primitives.Epoch) ([][fieldparams.BLSPubkeyLength]byte, []primitives.ValidatorIndex) { + outKeys := make([][fieldparams.BLSPubkeyLength]byte, 0, len(keys)) + indices := make([]primitives.ValidatorIndex, 0, len(keys)) + for _, pk := range keys { + st, ok := v.pubkeyToStatus[pk] + if !ok || !isActiveForDuties(st.status, epoch) { + continue + } + outKeys = append(outKeys, pk) + indices = append(indices, st.index) + } + slices.Sort(indices) + return outKeys, indices +} + // UpdateDuties checks the slot number to determine if the validator's // list of upcoming assignments needs to be updated. For example, at the // beginning of a new epoch. @@ -48,53 +89,511 @@ func (v *validator) UpdateDuties(ctx context.Context) error { ctx, span := trace.StartSpan(ctx, "validator.UpdateDuties") defer span.End() - filteredKeys, err := v.filterBlacklistedKeys(ctx) + keys, err := v.nonBlacklistedKeys(ctx) if err != nil { - return err + return errors.Wrap(err, "could not filter blacklisted keys") } epoch := slots.ToEpoch(slots.CurrentSlot(v.genesisTime) + 1) + + filteredKeys, filteredIndices := v.filteredKeysAndIndices(keys, epoch) + if epoch >= params.BeaconConfig().GloasForkEpoch { + err = v.updateDutiesSplit(ctx, epoch, filteredIndices) + } else { + err = v.updateDutiesCombined(ctx, epoch, filteredKeys) + } + if err != nil { + return errors.Wrap(err, "could not fetch duties") + } + + if !v.duties.isInitialized() { + return nil + } + + ss, err := slots.EpochStart(epoch) + if err != nil { + return errors.Wrap(err, "could not compute epoch start slot") + } + v.logDuties(ss) + + return v.onDutiesUpdated(ctx) +} + +// updateDutiesCombined uses the combined Duties() endpoint (pre-GLOAS). +func (v *validator) updateDutiesCombined(ctx context.Context, epoch primitives.Epoch, filteredKeys [][fieldparams.BLSPubkeyLength]byte) error { req := ðpb.DutiesRequest{ Epoch: epoch, PublicKeys: bytesutil.FromBytes48Array(filteredKeys), } resp, err := v.validatorClient.Duties(ctx, req) - if err != nil || resp == nil { - v.dutiesLock.Lock() - v.duties.Reset() - v.dutiesLock.Unlock() - log.WithError(err).Error("Error getting validator duties") - return err + if err != nil { + return errors.Wrap(err, "could not get validator duties") + } + if resp == nil { + return errors.New("nil duties response from beacon node") } - ss, err := slots.EpochStart(epoch) - if err != nil { - return err + var data dutyStoreData + data.setFromContainer(resp) + data.missingNext = missingNextPtc + v.duties.write(data) + + if allCurrentDutiesExited(resp.CurrentEpochDuties) { + return ErrValidatorsAllExited + } + return nil +} + +// depRootsDiverged reports whether the freshly fetched next-epoch attester and +// proposer dependent roots disagree. +func depRootsDiverged(epoch primitives.Epoch, res dutiesFetchResult) bool { + if epoch < params.BeaconConfig().FuluForkEpoch { + // Pre-fulu, attester dep root for epoch+1 is get_block_root_at_slot( + // compute_start_slot_at_epoch(epoch) - 1) + // proposer dep root is get_block_root_at_slot(compute_start_slot_at_epoch(epoch+1) - 1) + // they differ by design + return false + } + if res.propNext == nil || res.attNext == nil { + return false + } + return !bytes.Equal(res.propNext.DependentRoot, res.attNext.DependentRoot) +} + +// allCurrentDutiesExited reports whether there is at least one duty and all are EXITED. +func allCurrentDutiesExited(duties []*ethpb.ValidatorDuty) bool { + if len(duties) == 0 { + return false + } + for _, d := range duties { + if d.Status != ethpb.ValidatorStatus_EXITED { + return false + } + } + return true +} + +// dutiesFetchResult holds the successful results from fetching or +// promoting current-epoch duties plus raw next-epoch API responses. +type dutiesFetchResult struct { + currentDuties []*ethpb.ValidatorDuty + prevDepRoot []byte + currDepRoot []byte + attNext *ethpb.AttesterDutiesResponse + propNext *ethpb.ProposerDutiesResponse + syncNext *ethpb.SyncCommitteeDutiesResponse + ptcNext *ethpb.PTCDutiesResponse + missingNext missingNextDuties +} + +// missingNextDuties is a bitmask of next-epoch duty types that were expected +// but missing after a fetch (soft failures). Tracked so the next promotion +// can fall back to a full fresh fetch instead of propagating incomplete data. +type missingNextDuties uint8 + +const ( + missingNextProposer missingNextDuties = 1 << iota + missingNextSync + missingNextPtc +) + +// updateDutiesSplit fetches duties from the split V3 endpoints and +// populates the duty store. When the epoch has advanced by exactly one +// and duties are already initialized, it promotes the cached next-epoch +// duties to current and only fetches the new next-epoch. indices must be +// sorted (see filteredKeysAndIndices). +func (v *validator) updateDutiesSplit(ctx context.Context, epoch primitives.Epoch, indices []primitives.ValidatorIndex) error { + if len(indices) == 0 { + // No active keys for this client; drop any previously cached duties so + // stale entries don't keep appearing in RolesAt etc. + v.duties.reset() + return nil + } + + canPromote := v.duties.canPromote(epoch, indices) + + var ( + res dutiesFetchResult + err error + ) + // On fetch failure, leave existing duties intact so the validator can + // continue serving the current epoch from cache while we retry next tick. + if canPromote { + log.WithField("epoch", epoch).Debug("Promoting cached next-epoch duties to current") + res, err = v.promoteDuties(ctx, epoch, indices) + if err != nil { + return errors.Wrap(err, "promote duties") + } + } else { + res, err = v.fetchAllDuties(ctx, epoch, indices) + if err != nil { + return errors.Wrap(err, "fetch all duties") + } } - v.dutiesLock.Lock() - v.duties.SetFromCombinedDutiesResponse(resp) - v.logDuties(ss) - v.dutiesLock.Unlock() - allExitedCounter := 0 - for _, d := range resp.CurrentEpochDuties { - if d.Status == ethpb.ValidatorStatus_EXITED { - allExitedCounter++ + if depRootsDiverged(epoch, res) { + if canPromote { + log.Warn("Proposer and attester dependent roots diverged on promotion, refetching all duties") + res, err = v.fetchAllDuties(ctx, epoch, indices) + if err != nil { + return errors.Wrap(err, "refetch all duties after promotion divergence") + } + } else { + log.Warn("Proposer and attester dependent roots diverged on fresh fetch") } } - if allExitedCounter != 0 && allExitedCounter == len(resp.CurrentEpochDuties) { + + nextDuties := v.buildNextDuties(res) + + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{ + PrevDependentRoot: res.prevDepRoot, + CurrDependentRoot: res.currDepRoot, + CurrentEpochDuties: res.currentDuties, + NextEpochDuties: nextDuties, + }) + data.epoch = epoch + data.missingNext = res.missingNext + data.indices = indices + v.duties.write(data) + + if allCurrentDutiesExited(res.currentDuties) { return ErrValidatorsAllExited } + return nil +} + +// promoteDuties promotes cached next-epoch duties to current and fetches the +// new next-epoch duties. Cached duties already carry PtcSlots from the prior +// fetch, so no current-epoch refetch is needed. +func (v *validator) promoteDuties(ctx context.Context, epoch primitives.Epoch, indices []primitives.ValidatorIndex) (dutiesFetchResult, error) { + snap := v.duties.snapshot() + currentDuties := make([]*ethpb.ValidatorDuty, 0, snap.nextDutyCount()) + for _, d := range snap.nextDuties() { + if d == nil { + continue + } + // nextDuties yields read-only aliases into the live store, so clone + // before refreshing the status to avoid mutating cached state in place. + promoted := cloneValidatorDuty(d) + promoted.Status = v.statusForPubkey(promoted.PublicKey) + currentDuties = append(currentDuties, promoted) + } + res := dutiesFetchResult{ + currentDuties: currentDuties, + // On promotion, last cycle's currDependentRoot (which covered next-epoch + // duties) becomes this cycle's prevDepRoot (covering current-epoch + // duties). + prevDepRoot: snap.currDependentRoot(), + } + + var ( + attErr, propErr error + syncErr, ptcErr error + wg sync.WaitGroup + ) + wg.Go(func() { + res.attNext, attErr = v.validatorClient.AttesterDuties(ctx, epoch.Add(1), indices) + }) + wg.Go(func() { + res.propNext, propErr = v.validatorClient.ProposerDuties(ctx, epoch.Add(1)) + }) + wg.Go(func() { + if epoch.Add(1) < params.BeaconConfig().AltairForkEpoch { + return + } + res.syncNext, syncErr = v.validatorClient.SyncCommitteeDuties(ctx, epoch.Add(1), indices) + }) + wg.Go(func() { + if epoch.Add(1) < params.BeaconConfig().GloasForkEpoch { + return + } + res.ptcNext, ptcErr = v.validatorClient.PTCDuties(ctx, epoch.Add(1), indices) + }) + wg.Wait() + + if attErr != nil { + return res, attErr + } + if propErr != nil { + log.WithError(propErr).Debug("Could not get next epoch proposer duties") + } + if syncErr != nil { + log.WithError(syncErr).Debug("Could not get next epoch sync committee duties") + } + if ptcErr != nil { + log.WithError(ptcErr).Debug("Could not get next epoch PTC duties") + } + + res.missingNext = missingNextMask(epoch.Add(1), res.propNext, res.syncNext, res.ptcNext) + + // currDepRoot comes from the newly fetched next-epoch attester root, + // which matches the head event's CurrentDutyDependentRoot. + if res.attNext != nil { + res.currDepRoot = res.attNext.DependentRoot + } + return res, nil +} + +// missingNextMask reports which next-epoch duty types are missing post-fetch. +// Only types that were expected at nextEpoch (per fork gating) are flagged. +func missingNextMask(nextEpoch primitives.Epoch, prop *ethpb.ProposerDutiesResponse, sync *ethpb.SyncCommitteeDutiesResponse, ptc *ethpb.PTCDutiesResponse) missingNextDuties { + var m missingNextDuties + if prop == nil && nextEpoch >= params.BeaconConfig().FuluForkEpoch { + m |= missingNextProposer + } + if sync == nil && nextEpoch >= params.BeaconConfig().AltairForkEpoch { + m |= missingNextSync + } + if ptc == nil && nextEpoch >= params.BeaconConfig().GloasForkEpoch { + m |= missingNextPtc + } + return m +} + +// fetchAllDuties fetches both current and next epoch duties from all endpoints. +func (v *validator) fetchAllDuties(ctx context.Context, epoch primitives.Epoch, indices []primitives.ValidatorIndex) (dutiesFetchResult, error) { + var ( + res dutiesFetchResult + attCurr *ethpb.AttesterDutiesResponse + propCurr *ethpb.ProposerDutiesResponse + syncCurr *ethpb.SyncCommitteeDutiesResponse + ptcCurr *ethpb.PTCDutiesResponse + attErr, propErr error + syncErr, ptcErr error + wg sync.WaitGroup + ) + wg.Go(func() { + attCurr, res.attNext, attErr = v.fetchAttesterDuties(ctx, epoch, indices) + }) + wg.Go(func() { + propCurr, res.propNext, propErr = v.fetchProposerDuties(ctx, epoch) + }) + wg.Go(func() { + syncCurr, res.syncNext, syncErr = v.fetchSyncDuties(ctx, epoch, indices) + }) + wg.Go(func() { + ptcCurr, res.ptcNext, ptcErr = v.fetchPtcDuties(ctx, epoch, indices) + }) + wg.Wait() + + if attErr != nil { + return res, attErr + } + if propErr != nil { + return res, propErr + } + if syncErr != nil { + log.WithError(syncErr).Warn("Error getting sync committee duties") + } + if ptcErr != nil { + log.WithError(ptcErr).Warn("Error getting PTC duties") + } + + res.missingNext = missingNextMask(epoch.Add(1), res.propNext, res.syncNext, res.ptcNext) + + if attCurr != nil { + res.prevDepRoot = attCurr.DependentRoot + } + // Use the next-epoch attester dependent root as currDepRoot. + // The head event's CurrentDutyDependentRoot = DependentRoot(epoch), + // and attester duties for epoch+1 have DependentRoot(epoch), so they match. + if res.attNext != nil { + res.currDepRoot = res.attNext.DependentRoot + } + res.currentDuties = v.assembleDuties(attCurr, propCurr, syncCurr, ptcCurr) + return res, nil +} + +// buildNextDuties constructs next-epoch ValidatorDuty entries from +// the raw API responses in the fetch result. +func (v *validator) buildNextDuties(res dutiesFetchResult) []*ethpb.ValidatorDuty { + return v.assembleDuties(res.attNext, res.propNext, res.syncNext, res.ptcNext) +} + +// assembleDuties stitches together the four per-duty-type API responses for +// a single epoch into a slice of ValidatorDuty entries, one per attester +// assignment. Used by fetchAllDuties (current epoch) and buildNextDuties +// (next epoch). +func (v *validator) assembleDuties( + att *ethpb.AttesterDutiesResponse, + prop *ethpb.ProposerDutiesResponse, + sync *ethpb.SyncCommitteeDutiesResponse, + ptc *ethpb.PTCDutiesResponse, +) []*ethpb.ValidatorDuty { + proposerSlots := make(map[primitives.ValidatorIndex][]primitives.Slot) + if prop != nil { + for _, d := range prop.Duties { + proposerSlots[d.ValidatorIndex] = append(proposerSlots[d.ValidatorIndex], d.Slot) + } + } + ptcSlots := make(map[primitives.ValidatorIndex][]primitives.Slot) + if ptc != nil { + for _, d := range ptc.Duties { + ptcSlots[d.ValidatorIndex] = append(ptcSlots[d.ValidatorIndex], d.Slot) + } + } + syncSet := make(map[primitives.ValidatorIndex]bool) + if sync != nil { + for _, d := range sync.Duties { + syncSet[d.ValidatorIndex] = true + } + } + if att == nil { + return nil + } + duties := make([]*ethpb.ValidatorDuty, 0, len(att.Duties)) + for _, d := range att.Duties { + duties = append(duties, ðpb.ValidatorDuty{ + PublicKey: d.Pubkey, + ValidatorIndex: d.ValidatorIndex, + CommitteeIndex: d.CommitteeIndex, + CommitteeLength: d.CommitteeLength, + CommitteesAtSlot: d.CommitteesAtSlot, + ValidatorCommitteeIndex: d.ValidatorCommitteeIndex, + AttesterSlot: d.Slot, + ProposerSlots: proposerSlots[d.ValidatorIndex], + IsSyncCommittee: syncSet[d.ValidatorIndex], + PtcSlots: ptcSlots[d.ValidatorIndex], + Status: v.statusForPubkey(d.Pubkey), + }) + } + return duties +} + +// statusForPubkey returns the cached validator status for a pubkey. +func (v *validator) statusForPubkey(pk []byte) ethpb.ValidatorStatus { + if v.pubkeyToStatus == nil { + return ethpb.ValidatorStatus_UNKNOWN_STATUS + } + st, ok := v.pubkeyToStatus[bytesutil.ToBytes48(pk)] + if !ok || st.status == nil { + return ethpb.ValidatorStatus_UNKNOWN_STATUS + } + return st.status.Status +} + +// fetchAttesterDuties fetches attester duties for current and next epoch in parallel. +func (v *validator) fetchAttesterDuties( + ctx context.Context, epoch primitives.Epoch, indices []primitives.ValidatorIndex, +) (current, next *ethpb.AttesterDutiesResponse, err error) { + var ( + currErr, nextErr error + wg sync.WaitGroup + ) + wg.Go(func() { + current, currErr = v.validatorClient.AttesterDuties(ctx, epoch, indices) + }) + wg.Go(func() { + next, nextErr = v.validatorClient.AttesterDuties(ctx, epoch.Add(1), indices) + }) + wg.Wait() + + if currErr != nil { + return nil, nil, currErr + } + if nextErr != nil { + return nil, nil, nextErr + } + return current, next, nil +} + +// fetchProposerDuties fetches proposer duties for the current epoch. +// Post-fulu, also fetches next-epoch duties (deterministic via proposer_lookahead). +// Pre-fulu, next-epoch proposer duties are not deterministic and not fetched. +func (v *validator) fetchProposerDuties( + ctx context.Context, epoch primitives.Epoch, +) (current, next *ethpb.ProposerDutiesResponse, err error) { + var ( + currErr, nextErr error + wg sync.WaitGroup + ) + wg.Go(func() { + current, currErr = v.validatorClient.ProposerDuties(ctx, epoch) + }) + if epoch >= params.BeaconConfig().FuluForkEpoch { + wg.Go(func() { + next, nextErr = v.validatorClient.ProposerDuties(ctx, epoch.Add(1)) + }) + } + wg.Wait() + + if currErr != nil { + return nil, nil, currErr + } + if nextErr != nil { + log.WithError(nextErr).Debug("Could not get next epoch proposer duties") + } + return current, next, nil +} + +// fetchSyncDuties fetches sync committee duties for current and next epoch. +func (v *validator) fetchSyncDuties( + ctx context.Context, epoch primitives.Epoch, indices []primitives.ValidatorIndex, +) (current, next *ethpb.SyncCommitteeDutiesResponse, err error) { + if epoch < params.BeaconConfig().AltairForkEpoch { + return nil, nil, nil + } + + var ( + currErr, nextErr error + wg sync.WaitGroup + ) + wg.Go(func() { + current, currErr = v.validatorClient.SyncCommitteeDuties(ctx, epoch, indices) + }) + wg.Go(func() { + next, nextErr = v.validatorClient.SyncCommitteeDuties(ctx, epoch.Add(1), indices) + }) + wg.Wait() + + if currErr != nil { + return nil, nil, currErr + } + if nextErr != nil { + log.WithError(nextErr).Debug("Could not get next epoch sync committee duties") + } + return current, next, nil +} + +// fetchPtcDuties fetches PTC duties for the current and next epoch in parallel. +func (v *validator) fetchPtcDuties( + ctx context.Context, epoch primitives.Epoch, indices []primitives.ValidatorIndex, +) (current, next *ethpb.PTCDutiesResponse, err error) { + if epoch < params.BeaconConfig().GloasForkEpoch { + return nil, nil, nil + } + var ( + currErr, nextErr error + wg sync.WaitGroup + ) + wg.Go(func() { + current, currErr = v.validatorClient.PTCDuties(ctx, epoch, indices) + }) + wg.Go(func() { + next, nextErr = v.validatorClient.PTCDuties(ctx, epoch.Add(1), indices) + }) + wg.Wait() + if currErr != nil { + return nil, nil, currErr + } + if nextErr != nil { + log.WithError(nextErr).Debug("Could not get next epoch PTC duties") + } + return current, next, nil +} - // Non-blocking call for beacon node to start subscriptions for aggregators. +// onDutiesUpdated kicks off subnet subscriptions for the current duty set. +func (v *validator) onDutiesUpdated(ctx context.Context) error { md, exists := metadata.FromOutgoingContext(ctx) ctx = context.Background() if exists { ctx = metadata.NewOutgoingContext(ctx, md) } + container := v.duties.toContainer() go func() { - if err := v.subscribeToSubnets(ctx, resp); err != nil { + if err := v.subscribeToSubnets(ctx, container); err != nil { log.WithError(err).Error("Failed to subscribe to subnets") } }() @@ -103,24 +602,22 @@ func (v *validator) UpdateDuties(ctx context.Context) error { } func (v *validator) logDuties(slot primitives.Slot) { + snap := v.duties.snapshot() + if !snap.isInitialized() { + return + } + epochStartSlot, err := slots.EpochStart(slots.ToEpoch(slot)) if err != nil { log.WithError(err).Error("Could not calculate epoch start. Ignoring logging duties.") return } attesterKeys := make([][]string, params.BeaconConfig().SlotsPerEpoch) - for i := range attesterKeys { - attesterKeys[i] = make([]string, 0) - } proposerKeys := make([]string, params.BeaconConfig().SlotsPerEpoch) ptcKeys := make([][]string, params.BeaconConfig().SlotsPerEpoch) - for i := range attesterKeys { - attesterKeys[i] = make([]string, 0) - ptcKeys[i] = make([]string, 0) - } var totalProposingKeys, totalAttestingKeys, totalPTCKeys uint64 - for _, duty := range v.duties.CurrentEpochDuties() { + for _, duty := range snap.currentDuties() { pk := fmt.Sprintf("%#x", duty.PublicKey) if v.emitAccountMetrics { ValidatorStatusesGaugeVec.WithLabelValues(pk, fmt.Sprintf("%#x", duty.ValidatorIndex)).Set(float64(duty.Status)) @@ -171,7 +668,7 @@ func (v *validator) logDuties(slot primitives.Slot) { } } } - for _, duty := range v.duties.NextEpochDuties() { + for _, duty := range snap.nextDuties() { pk := fmt.Sprintf("%#x", duty.PublicKey) if duty.Status != ethpb.ValidatorStatus_ACTIVE && duty.Status != ethpb.ValidatorStatus_EXITING { continue @@ -246,10 +743,8 @@ func (v *validator) checkDependentRoots(ctx context.Context, head *structs.HeadE dutiesCtx, cancel := context.WithDeadline(ctx, v.SlotDeadline(ss-1)) defer cancel() - v.dutiesLock.RLock() - storedPrev, _ := v.duties.DependentRoots() + storedPrev := v.duties.prevDependentRoot() needsPrevUpdate := storedPrev == nil || !bytes.Equal(prevDependentRoot, storedPrev) - v.dutiesLock.RUnlock() if needsPrevUpdate { if err := v.UpdateDuties(dutiesCtx); err != nil { @@ -267,9 +762,7 @@ func (v *validator) checkDependentRoots(ctx context.Context, head *structs.HeadE if bytes.Equal(currDependentRoot, params.BeaconConfig().ZeroHash[:]) { return nil } - v.dutiesLock.RLock() - _, storedCurr := v.duties.DependentRoots() - v.dutiesLock.RUnlock() + storedCurr := v.duties.currDependentRoot() needsCurrUpdate := storedCurr == nil || !bytes.Equal(currDependentRoot, storedCurr) if !needsCurrUpdate { return nil diff --git a/validator/client/duties_test.go b/validator/client/duties_test.go index 4867425df30a..0f6e1da9188c 100644 --- a/validator/client/duties_test.go +++ b/validator/client/duties_test.go @@ -23,28 +23,6 @@ import ( "google.golang.org/protobuf/types/known/emptypb" ) -func TestUpdateDuties_DoesNothingWhenNotEpochStart_AlreadyExistingAssignments(t *testing.T) { - ctrl := gomock.NewController(t) - defer ctrl.Finish() - client := validatormock.NewMockValidatorClient(ctrl) - - v := validator{ - km: newMockKeymanager(t, randKeypair(t)), - validatorClient: client, - duties: func() *dutyStore { - ds := testDutyStore(ðpb.ValidatorDuty{AttesterSlot: 10, CommitteeIndex: 20}) - ds.nextDuties[pubkey{}] = ðpb.ValidatorDuty{AttesterSlot: 10, CommitteeIndex: 20} - return ds - }(), - } - client.EXPECT().Duties( - gomock.Any(), - gomock.Any(), - ).Times(1) - - assert.NoError(t, v.UpdateDuties(t.Context()), "Could not update assignments") -} - func TestUpdateDuties_ReturnsError(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() @@ -64,7 +42,7 @@ func TestUpdateDuties_ReturnsError(t *testing.T) { ).Return(nil, expected) assert.ErrorContains(t, expected.Error(), v.UpdateDuties(t.Context())) - assert.Equal(t, false, v.duties.IsInitialized(), "Assignments should have been cleared on failure") + assert.Equal(t, true, v.duties.isInitialized(), "Existing assignments should be preserved across transient errors") } func TestUpdateDuties_OK(t *testing.T) { @@ -111,10 +89,10 @@ func TestUpdateDuties_OK(t *testing.T) { util.WaitTimeout(&wg, 2*time.Second) - duties := v.duties.CurrentEpochDuties() - require.Equal(t, 1, len(duties), "Expected one duty") + snap := v.duties.snapshot() + require.Equal(t, 1, snap.currentDutyCount(), "Expected one duty") var gotDuty *ethpb.ValidatorDuty - for _, d := range duties { + for _, d := range snap.currentDuties() { gotDuty = d } assert.Equal(t, params.BeaconConfig().SlotsPerEpoch+1, gotDuty.ProposerSlots[0], "Unexpected validator assignments") @@ -252,7 +230,11 @@ func TestUpdateDuties_Distributed(t *testing.T) { duties: &dutyStore{}, genesisTime: genesis, pubkeyToStatus: map[[fieldparams.BLSPubkeyLength]byte]*validatorStatus{ - keys.pub: {publicKey: keys.pub[:], index: 200}, + keys.pub: { + publicKey: keys.pub[:], + status: ðpb.ValidatorStatusResponse{Status: ethpb.ValidatorStatus_ACTIVE}, + index: 200, + }, }, } v.aggSelector = newDistributedSelector(&v) @@ -327,7 +309,11 @@ func TestValidator_CheckDependentRoots(t *testing.T) { CurrDependentRoot: bytesutil.PadTo([]byte{0x04, 0x05, 0x06}, fieldparams.RootLength), } ds := &dutyStore{} - ds.SetFromCombinedDutiesResponse(dutiesContainer) + { + var data dutyStoreData + data.setFromContainer(dutiesContainer) + ds.write(data) + } v := &validator{ km: newMockKeymanager(t, randKeypair(t)), validatorClient: client, @@ -407,8 +393,581 @@ func TestValidator_CheckDependentRoots(t *testing.T) { } curr, err := bytesutil.DecodeHexWithLength(head.CurrentDutyDependentRoot, fieldparams.RootLength) require.NoError(t, err) - _, storedCurr := v.duties.DependentRoots() - require.DeepEqual(t, curr, storedCurr) + require.DeepEqual(t, curr, v.duties.currDependentRoot()) require.NoError(t, v.checkDependentRoots(ctx, head)) }) } + +// TestValidator_CheckDependentRoots_NoEmptyWindowDuringRefetch asserts that +// concurrent readers of the duty store never observe an empty store while +// checkDependentRoots is refetching. A previous implementation called +// clearDuties() before UpdateDuties(), leaving a window in which other +// goroutines would fail with "no duties for validators". +func TestValidator_CheckDependentRoots_NoEmptyWindowDuringRefetch(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + ctx := t.Context() + client := validatormock.NewMockValidatorClient(ctrl) + + oldContainer := ðpb.ValidatorDutiesContainer{ + CurrentEpochDuties: []*ethpb.ValidatorDuty{{ + AttesterSlot: params.BeaconConfig().SlotsPerEpoch, + ValidatorIndex: 200, + CommitteeIndex: 100, + CommitteeLength: 4, + PublicKey: []byte("testPubKey_1"), + }}, + PrevDependentRoot: bytesutil.PadTo([]byte{0x01, 0x02, 0x03}, fieldparams.RootLength), + CurrDependentRoot: bytesutil.PadTo([]byte{0x04, 0x05, 0x06}, fieldparams.RootLength), + } + newContainer := ðpb.ValidatorDutiesContainer{ + CurrentEpochDuties: oldContainer.CurrentEpochDuties, + PrevDependentRoot: bytesutil.PadTo([]byte{0xaa, 0xbb, 0xcc}, fieldparams.RootLength), + CurrDependentRoot: bytesutil.PadTo([]byte{0xdd, 0xee, 0xff}, fieldparams.RootLength), + } + ds := &dutyStore{} + { + var data dutyStoreData + data.setFromContainer(oldContainer) + ds.write(data) + } + v := &validator{ + km: newMockKeymanager(t, randKeypair(t)), + validatorClient: client, + duties: ds, + } + v.aggSelector = testLocalSelector(t, v) + + // Block the RPC inside UpdateDuties until we release it, and signal when + // the goroutine is actually inside the call so we can probe store state. + entered := make(chan struct{}) + release := make(chan struct{}) + client.EXPECT().Duties(gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, _ *ethpb.DutiesRequest) (*ethpb.ValidatorDutiesContainer, error) { + close(entered) + <-release + return newContainer, nil + }, + ) + client.EXPECT().SubscribeCommitteeSubnets( + gomock.Any(), gomock.Any(), gomock.Any(), + ).Return(&emptypb.Empty{}, nil).AnyTimes() + + // Head event with a prev root that differs from stored — triggers + // needsPrevUpdate. + head := &structs.HeadEvent{ + Slot: "1", + PreviousDutyDependentRoot: "0xe3f7a1b2c489d56f03a6b8d9c7e1fa2456bb09f3de42a67c8910fc3e7a5d4b12", + CurrentDutyDependentRoot: "0xe3f7a1b2c489d56f03a6b8d9c7e1fa2456bb09f3de42a67c8910fc3e7a5d4b12", + } + + done := make(chan error, 1) + go func() { done <- v.checkDependentRoots(ctx, head) }() + + <-entered // refetch is in flight + + // The bug: with clearDuties() before UpdateDuties(), the dependent roots + // would be (nil, nil) here. The fix keeps the OLD values visible until + // the atomic swap at the end of updateDuties. + prev := v.duties.prevDependentRoot() + curr := v.duties.currDependentRoot() + require.NotNil(t, prev, "duty store was cleared mid-refetch (prev)") + require.NotNil(t, curr, "duty store was cleared mid-refetch (curr)") + require.DeepEqual(t, oldContainer.PrevDependentRoot, prev) + require.DeepEqual(t, oldContainer.CurrDependentRoot, curr) + require.Equal(t, true, v.duties.isInitialized()) + + close(release) + require.NoError(t, <-done) + + // After completion, the new roots must be in place. + require.DeepEqual(t, newContainer.PrevDependentRoot, v.duties.prevDependentRoot()) + require.DeepEqual(t, newContainer.CurrDependentRoot, v.duties.currDependentRoot()) +} + +func TestUpdateDutiesSplit(t *testing.T) { + epoch := primitives.Epoch(5) + + setup := func(t *testing.T) (*validator, *validatormock.MockValidatorClient, keypair) { + params.SetupTestConfigCleanup(t) + cfg := params.BeaconConfig().Copy() + cfg.AltairForkEpoch = 0 + cfg.FuluForkEpoch = 0 + cfg.GloasForkEpoch = 0 + params.OverrideBeaconConfig(cfg) + + ctrl := gomock.NewController(t) + client := validatormock.NewMockValidatorClient(ctrl) + keys := randKeypair(t) + v := &validator{ + validatorClient: client, + duties: &dutyStore{}, + pubkeyToStatus: map[pubkey]*validatorStatus{ + keys.pub: { + publicKey: keys.pub[:], + status: ðpb.ValidatorStatusResponse{Status: ethpb.ValidatorStatus_ACTIVE}, + index: 42, + }, + }, + } + return v, client, keys + } + + t.Run("OK", func(t *testing.T) { + v, client, keys := setup(t) + spe := params.BeaconConfig().SlotsPerEpoch + + client.EXPECT().AttesterDuties(gomock.Any(), epoch, gomock.Any()).Return(ðpb.AttesterDutiesResponse{ + DependentRoot: make([]byte, 32), + Duties: []*ethpb.AttesterDuty{{ + Pubkey: keys.pub[:], ValidatorIndex: 42, + Slot: primitives.Slot(epoch)*spe + 3, CommitteeIndex: 1, CommitteeLength: 64, CommitteesAtSlot: 4, + }}, + }, nil) + client.EXPECT().AttesterDuties(gomock.Any(), epoch+1, gomock.Any()).Return(ðpb.AttesterDutiesResponse{ + Duties: []*ethpb.AttesterDuty{{ + Pubkey: keys.pub[:], ValidatorIndex: 42, + Slot: primitives.Slot(epoch+1)*spe + 7, CommitteeIndex: 2, CommitteeLength: 64, CommitteesAtSlot: 4, + }}, + }, nil) + client.EXPECT().ProposerDuties(gomock.Any(), epoch).Return(ðpb.ProposerDutiesResponse{ + DependentRoot: make([]byte, 32), + Duties: []*ethpb.ProposerDutyV2{{Pubkey: keys.pub[:], ValidatorIndex: 42, Slot: primitives.Slot(epoch)*spe + 1}}, + }, nil) + client.EXPECT().ProposerDuties(gomock.Any(), epoch+1).Return(ðpb.ProposerDutiesResponse{}, nil) + client.EXPECT().SyncCommitteeDuties(gomock.Any(), epoch, gomock.Any()).Return(ðpb.SyncCommitteeDutiesResponse{ + Duties: []*ethpb.SyncCommitteeDuty{{Pubkey: keys.pub[:], ValidatorIndex: 42}}, + }, nil) + client.EXPECT().SyncCommitteeDuties(gomock.Any(), epoch+1, gomock.Any()).Return(ðpb.SyncCommitteeDutiesResponse{}, nil) + client.EXPECT().PTCDuties(gomock.Any(), epoch, gomock.Any()).Return(ðpb.PTCDutiesResponse{ + Duties: []*ethpb.PTCDuty{{Pubkey: keys.pub[:], ValidatorIndex: 42, Slot: primitives.Slot(epoch)*spe + 5}}, + }, nil) + client.EXPECT().PTCDuties(gomock.Any(), epoch+1, gomock.Any()).Return(ðpb.PTCDutiesResponse{ + Duties: []*ethpb.PTCDuty{{Pubkey: keys.pub[:], ValidatorIndex: 42, Slot: primitives.Slot(epoch+1)*spe + 2}}, + }, nil) + + require.NoError(t, v.updateDutiesSplit(t.Context(), epoch, []primitives.ValidatorIndex{42})) + + snap := v.duties.snapshot() + // Current epoch: attester + proposer + sync + PTC. + require.Equal(t, 1, snap.currentDutyCount()) + for _, d := range snap.currentDuties() { + assert.Equal(t, primitives.Slot(epoch)*spe+3, d.AttesterSlot) + require.Equal(t, 1, len(d.ProposerSlots)) + assert.Equal(t, primitives.Slot(epoch)*spe+1, d.ProposerSlots[0]) + assert.Equal(t, true, d.IsSyncCommittee) + require.Equal(t, 1, len(d.PtcSlots)) + assert.Equal(t, primitives.Slot(epoch)*spe+5, d.PtcSlots[0]) + } + + // Next epoch: attester + PTC look-ahead. + require.Equal(t, 1, snap.nextDutyCount()) + for _, d := range snap.nextDuties() { + assert.Equal(t, primitives.Slot(epoch+1)*spe+7, d.AttesterSlot) + require.Equal(t, 1, len(d.PtcSlots)) + assert.Equal(t, primitives.Slot(epoch+1)*spe+2, d.PtcSlots[0]) + assert.Equal(t, false, d.IsSyncCommittee) + } + + // Duty store accessors. + assert.DeepEqual(t, []primitives.Slot{primitives.Slot(epoch)*spe + 1}, v.duties.proposerSlots(42)) + assert.DeepEqual(t, []primitives.Slot{primitives.Slot(epoch)*spe + 5}, v.duties.ptcSlots(42)) + assert.Equal(t, true, v.duties.isSyncCommittee(42)) + assert.Equal(t, false, v.duties.isNextSyncCommittee(42)) + }) + + t.Run("attester error preserves existing duties", func(t *testing.T) { + v, client, keys := setup(t) + spe := params.BeaconConfig().SlotsPerEpoch + seedDuty := ðpb.ValidatorDuty{ + PublicKey: keys.pub[:], ValidatorIndex: 42, + AttesterSlot: primitives.Slot(epoch)*spe + 3, Status: ethpb.ValidatorStatus_ACTIVE, + } + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{ + CurrentEpochDuties: []*ethpb.ValidatorDuty{seedDuty}, + }) + v.duties.write(data) + } + + client.EXPECT().AttesterDuties(gomock.Any(), epoch, gomock.Any()).Return(nil, errors.New("attester fail")) + client.EXPECT().AttesterDuties(gomock.Any(), epoch+1, gomock.Any()).Return(nil, nil).AnyTimes() + client.EXPECT().ProposerDuties(gomock.Any(), gomock.Any()).Return(ðpb.ProposerDutiesResponse{}, nil).AnyTimes() + client.EXPECT().SyncCommitteeDuties(gomock.Any(), gomock.Any(), gomock.Any()).Return(ðpb.SyncCommitteeDutiesResponse{}, nil).AnyTimes() + client.EXPECT().PTCDuties(gomock.Any(), gomock.Any(), gomock.Any()).Return(ðpb.PTCDutiesResponse{}, nil).AnyTimes() + + err := v.updateDutiesSplit(t.Context(), epoch, []primitives.ValidatorIndex{42}) + require.ErrorContains(t, "attester fail", err) + assert.Equal(t, true, v.duties.isInitialized()) + assert.Equal(t, 1, v.duties.snapshot().currentDutyCount()) + }) + + t.Run("proposer error preserves existing duties", func(t *testing.T) { + v, client, keys := setup(t) + spe := params.BeaconConfig().SlotsPerEpoch + seedDuty := ðpb.ValidatorDuty{ + PublicKey: keys.pub[:], ValidatorIndex: 42, + AttesterSlot: primitives.Slot(epoch)*spe + 3, Status: ethpb.ValidatorStatus_ACTIVE, + } + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{ + CurrentEpochDuties: []*ethpb.ValidatorDuty{seedDuty}, + }) + v.duties.write(data) + } + + client.EXPECT().AttesterDuties(gomock.Any(), gomock.Any(), gomock.Any()).Return(ðpb.AttesterDutiesResponse{}, nil).AnyTimes() + client.EXPECT().ProposerDuties(gomock.Any(), epoch).Return(nil, errors.New("proposer fail")) + client.EXPECT().ProposerDuties(gomock.Any(), epoch+1).Return(nil, nil).AnyTimes() + client.EXPECT().SyncCommitteeDuties(gomock.Any(), gomock.Any(), gomock.Any()).Return(ðpb.SyncCommitteeDutiesResponse{}, nil).AnyTimes() + client.EXPECT().PTCDuties(gomock.Any(), gomock.Any(), gomock.Any()).Return(ðpb.PTCDutiesResponse{}, nil).AnyTimes() + + err := v.updateDutiesSplit(t.Context(), epoch, []primitives.ValidatorIndex{42}) + require.ErrorContains(t, "proposer fail", err) + assert.Equal(t, true, v.duties.isInitialized()) + assert.Equal(t, 1, v.duties.snapshot().currentDutyCount()) + }) + + t.Run("PTC error is non-fatal", func(t *testing.T) { + v, client, keys := setup(t) + spe := params.BeaconConfig().SlotsPerEpoch + + client.EXPECT().AttesterDuties(gomock.Any(), epoch, gomock.Any()).Return(ðpb.AttesterDutiesResponse{ + DependentRoot: make([]byte, 32), + Duties: []*ethpb.AttesterDuty{{ + Pubkey: keys.pub[:], ValidatorIndex: 42, + Slot: primitives.Slot(epoch)*spe + 3, CommitteeIndex: 1, CommitteeLength: 64, CommitteesAtSlot: 4, + }}, + }, nil) + client.EXPECT().AttesterDuties(gomock.Any(), epoch+1, gomock.Any()).Return(ðpb.AttesterDutiesResponse{}, nil) + client.EXPECT().ProposerDuties(gomock.Any(), epoch).Return(ðpb.ProposerDutiesResponse{DependentRoot: make([]byte, 32)}, nil) + client.EXPECT().ProposerDuties(gomock.Any(), epoch+1).Return(ðpb.ProposerDutiesResponse{}, nil) + client.EXPECT().SyncCommitteeDuties(gomock.Any(), gomock.Any(), gomock.Any()).Return(ðpb.SyncCommitteeDutiesResponse{}, nil).AnyTimes() + client.EXPECT().PTCDuties(gomock.Any(), epoch, gomock.Any()).Return(nil, errors.New("ptc fail")) + client.EXPECT().PTCDuties(gomock.Any(), epoch+1, gomock.Any()).Return(ðpb.PTCDutiesResponse{}, nil) + + require.NoError(t, v.updateDutiesSplit(t.Context(), epoch, []primitives.ValidatorIndex{42})) + assert.Equal(t, true, v.duties.isInitialized()) + assert.Equal(t, 0, len(v.duties.ptcSlots(42))) + }) + + t.Run("no known indices clears existing duties", func(t *testing.T) { + v, _, keys := setup(t) + v.pubkeyToStatus = map[pubkey]*validatorStatus{} + + // Seed the store with prior duties so the test verifies they're cleared + // (rather than passing tautologically against an empty store). + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{ + CurrentEpochDuties: []*ethpb.ValidatorDuty{{ + PublicKey: keys.pub[:], ValidatorIndex: 42, + Status: ethpb.ValidatorStatus_ACTIVE, + }}, + }) + v.duties.write(data) + require.Equal(t, true, v.duties.isInitialized()) + } + + require.NoError(t, v.updateDutiesSplit(t.Context(), epoch, nil)) + assert.Equal(t, false, v.duties.isInitialized()) + }) + + t.Run("promote-path dependent root divergence falls back to full refetch", func(t *testing.T) { + hook := logTest.NewGlobal() + v, client, keys := setup(t) + spe := params.BeaconConfig().SlotsPerEpoch + + // Seed the store so canPromote is true (epoch-1 cached, next-epoch + // duties present, init flag set). + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{ + NextEpochDuties: []*ethpb.ValidatorDuty{{ + PublicKey: keys.pub[:], ValidatorIndex: 42, + AttesterSlot: primitives.Slot(epoch)*spe + 3, + Status: ethpb.ValidatorStatus_ACTIVE, + }}, + }) + v.duties.write(data) + } + v.duties.data.epoch = epoch - 1 + v.duties.data.currDependentRoot = bytesutil.PadTo([]byte{0xaa}, 32) + v.duties.data.indices = []primitives.ValidatorIndex{42} + + rootA := bytesutil.PadTo([]byte{0x01}, 32) + rootB := bytesutil.PadTo([]byte{0x02}, 32) + rootC := bytesutil.PadTo([]byte{0x03}, 32) + + // Promote path: mismatched roots. + client.EXPECT().AttesterDuties(gomock.Any(), epoch+1, gomock.Any()).Return(ðpb.AttesterDutiesResponse{ + DependentRoot: rootA, + Duties: []*ethpb.AttesterDuty{{ + Pubkey: keys.pub[:], ValidatorIndex: 42, + Slot: primitives.Slot(epoch+1)*spe + 7, CommitteeIndex: 2, CommitteeLength: 64, CommitteesAtSlot: 4, + }}, + }, nil) + client.EXPECT().ProposerDuties(gomock.Any(), epoch+1).Return(ðpb.ProposerDutiesResponse{DependentRoot: rootB}, nil) + client.EXPECT().SyncCommitteeDuties(gomock.Any(), epoch+1, gomock.Any()).Return(ðpb.SyncCommitteeDutiesResponse{}, nil) + client.EXPECT().PTCDuties(gomock.Any(), epoch+1, gomock.Any()).Return(ðpb.PTCDutiesResponse{}, nil) + + // Refetch path: aligned roots, full set of RPCs. + client.EXPECT().AttesterDuties(gomock.Any(), epoch, gomock.Any()).Return(ðpb.AttesterDutiesResponse{ + DependentRoot: bytesutil.PadTo([]byte{0x10}, 32), + Duties: []*ethpb.AttesterDuty{{ + Pubkey: keys.pub[:], ValidatorIndex: 42, + Slot: primitives.Slot(epoch)*spe + 3, CommitteeIndex: 1, CommitteeLength: 64, CommitteesAtSlot: 4, + }}, + }, nil) + client.EXPECT().AttesterDuties(gomock.Any(), epoch+1, gomock.Any()).Return(ðpb.AttesterDutiesResponse{ + DependentRoot: rootC, + Duties: []*ethpb.AttesterDuty{{ + Pubkey: keys.pub[:], ValidatorIndex: 42, + Slot: primitives.Slot(epoch+1)*spe + 7, CommitteeIndex: 2, CommitteeLength: 64, CommitteesAtSlot: 4, + }}, + }, nil) + client.EXPECT().ProposerDuties(gomock.Any(), epoch).Return(ðpb.ProposerDutiesResponse{DependentRoot: bytesutil.PadTo([]byte{0x11}, 32)}, nil) + client.EXPECT().ProposerDuties(gomock.Any(), epoch+1).Return(ðpb.ProposerDutiesResponse{DependentRoot: rootC}, nil) + client.EXPECT().SyncCommitteeDuties(gomock.Any(), epoch, gomock.Any()).Return(ðpb.SyncCommitteeDutiesResponse{}, nil) + client.EXPECT().SyncCommitteeDuties(gomock.Any(), epoch+1, gomock.Any()).Return(ðpb.SyncCommitteeDutiesResponse{}, nil) + client.EXPECT().PTCDuties(gomock.Any(), epoch, gomock.Any()).Return(ðpb.PTCDutiesResponse{}, nil) + client.EXPECT().PTCDuties(gomock.Any(), epoch+1, gomock.Any()).Return(ðpb.PTCDutiesResponse{}, nil) + + require.NoError(t, v.updateDutiesSplit(t.Context(), epoch, []primitives.ValidatorIndex{42})) + assert.LogsContain(t, hook, "diverged on promotion") + + // Refetch's currDepRoot is the next-epoch attester root. + require.DeepEqual(t, rootC, v.duties.currDependentRoot()) + assert.Equal(t, epoch, v.duties.data.epoch) + }) + + t.Run("incomplete cache forces full refetch instead of promote", func(t *testing.T) { + v, client, keys := setup(t) + spe := params.BeaconConfig().SlotsPerEpoch + + // First iteration at epoch: next-epoch proposer soft-fails. All other RPCs succeed. + // fetchProposerDuties logs nextErr at Debug and returns next=nil, so propErr is nil. + client.EXPECT().AttesterDuties(gomock.Any(), epoch, gomock.Any()).Return(ðpb.AttesterDutiesResponse{ + DependentRoot: make([]byte, 32), + Duties: []*ethpb.AttesterDuty{{ + Pubkey: keys.pub[:], ValidatorIndex: 42, + Slot: primitives.Slot(epoch) * spe, CommitteeIndex: 1, CommitteeLength: 64, CommitteesAtSlot: 4, + }}, + }, nil) + client.EXPECT().AttesterDuties(gomock.Any(), epoch+1, gomock.Any()).Return(ðpb.AttesterDutiesResponse{ + Duties: []*ethpb.AttesterDuty{{ + Pubkey: keys.pub[:], ValidatorIndex: 42, + Slot: primitives.Slot(epoch+1) * spe, CommitteeIndex: 2, CommitteeLength: 64, CommitteesAtSlot: 4, + }}, + }, nil) + client.EXPECT().ProposerDuties(gomock.Any(), epoch).Return(ðpb.ProposerDutiesResponse{}, nil) + client.EXPECT().ProposerDuties(gomock.Any(), epoch+1).Return(nil, errors.New("next proposer fail")) + client.EXPECT().SyncCommitteeDuties(gomock.Any(), gomock.Any(), gomock.Any()).Return(ðpb.SyncCommitteeDutiesResponse{}, nil).Times(2) + client.EXPECT().PTCDuties(gomock.Any(), gomock.Any(), gomock.Any()).Return(ðpb.PTCDutiesResponse{}, nil).Times(2) + + require.NoError(t, v.updateDutiesSplit(t.Context(), epoch, []primitives.ValidatorIndex{42})) + require.Equal(t, missingNextProposer, v.duties.data.missingNext&missingNextProposer) + + // Second iteration at epoch+1. v.duties.epoch+1 == epoch+1 would normally trigger + // the promote path (only 4 next-epoch RPCs). The dirty mask must force a full fetch, + // so we expect all 8 RPCs (current+next for each duty type). + nextEpoch := epoch + 1 + client.EXPECT().AttesterDuties(gomock.Any(), nextEpoch, gomock.Any()).Return(ðpb.AttesterDutiesResponse{ + DependentRoot: make([]byte, 32), + Duties: []*ethpb.AttesterDuty{{ + Pubkey: keys.pub[:], ValidatorIndex: 42, + Slot: primitives.Slot(nextEpoch) * spe, CommitteeIndex: 1, CommitteeLength: 64, CommitteesAtSlot: 4, + }}, + }, nil) + client.EXPECT().AttesterDuties(gomock.Any(), nextEpoch+1, gomock.Any()).Return(ðpb.AttesterDutiesResponse{ + Duties: []*ethpb.AttesterDuty{{ + Pubkey: keys.pub[:], ValidatorIndex: 42, + Slot: primitives.Slot(nextEpoch+1) * spe, CommitteeIndex: 2, CommitteeLength: 64, CommitteesAtSlot: 4, + }}, + }, nil) + client.EXPECT().ProposerDuties(gomock.Any(), nextEpoch).Return(ðpb.ProposerDutiesResponse{}, nil) + client.EXPECT().ProposerDuties(gomock.Any(), nextEpoch+1).Return(ðpb.ProposerDutiesResponse{}, nil) + client.EXPECT().SyncCommitteeDuties(gomock.Any(), gomock.Any(), gomock.Any()).Return(ðpb.SyncCommitteeDutiesResponse{}, nil).Times(2) + client.EXPECT().PTCDuties(gomock.Any(), gomock.Any(), gomock.Any()).Return(ðpb.PTCDutiesResponse{}, nil).Times(2) + + require.NoError(t, v.updateDutiesSplit(t.Context(), nextEpoch, []primitives.ValidatorIndex{42})) + require.Equal(t, missingNextDuties(0), v.duties.data.missingNext) + }) + + t.Run("validator set drift forces full refetch instead of promote", func(t *testing.T) { + v, client, keys := setup(t) + spe := params.BeaconConfig().SlotsPerEpoch + + // Seed the store with indices=[42] and a complete next-epoch cache so + // that, ignoring drift, canPromote would otherwise return true. + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{ + NextEpochDuties: []*ethpb.ValidatorDuty{{ + PublicKey: keys.pub[:], ValidatorIndex: 42, + Status: ethpb.ValidatorStatus_ACTIVE, + }}, + }) + data.epoch = epoch - 1 + data.indices = []primitives.ValidatorIndex{42} + v.duties.write(data) + } + + // Caller now presents a different (larger) index set; canPromote must + // reject promotion and fall through to fetchAllDuties. + client.EXPECT().AttesterDuties(gomock.Any(), epoch, gomock.Any()).Return(ðpb.AttesterDutiesResponse{ + DependentRoot: make([]byte, 32), + Duties: []*ethpb.AttesterDuty{{ + Pubkey: keys.pub[:], ValidatorIndex: 42, + Slot: primitives.Slot(epoch) * spe, CommitteeIndex: 1, CommitteeLength: 64, CommitteesAtSlot: 4, + }}, + }, nil) + client.EXPECT().AttesterDuties(gomock.Any(), epoch+1, gomock.Any()).Return(ðpb.AttesterDutiesResponse{}, nil) + client.EXPECT().ProposerDuties(gomock.Any(), epoch).Return(ðpb.ProposerDutiesResponse{}, nil) + client.EXPECT().ProposerDuties(gomock.Any(), epoch+1).Return(ðpb.ProposerDutiesResponse{}, nil) + client.EXPECT().SyncCommitteeDuties(gomock.Any(), gomock.Any(), gomock.Any()).Return(ðpb.SyncCommitteeDutiesResponse{}, nil).Times(2) + client.EXPECT().PTCDuties(gomock.Any(), gomock.Any(), gomock.Any()).Return(ðpb.PTCDutiesResponse{}, nil).Times(2) + + require.NoError(t, v.updateDutiesSplit(t.Context(), epoch, []primitives.ValidatorIndex{42, 99})) + require.DeepEqual(t, []primitives.ValidatorIndex{42, 99}, v.duties.data.indices) + }) + + t.Run("combined-endpoint cache cannot promote into split", func(t *testing.T) { + v, client, keys := setup(t) + spe := params.BeaconConfig().SlotsPerEpoch + + // Simulate what updateDutiesCombined leaves behind: a populated next- + // epoch cache, missingNext=missingNextPtc, and indices empty (combined + // path doesn't track them). The first split call must refetch. + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{ + NextEpochDuties: []*ethpb.ValidatorDuty{{ + PublicKey: keys.pub[:], ValidatorIndex: 42, + Status: ethpb.ValidatorStatus_ACTIVE, + }}, + }) + data.missingNext = missingNextPtc + v.duties.write(data) + } + + // Expect full-fetch RPC pattern (8 endpoints), not promote (4). + client.EXPECT().AttesterDuties(gomock.Any(), epoch, gomock.Any()).Return(ðpb.AttesterDutiesResponse{ + DependentRoot: make([]byte, 32), + Duties: []*ethpb.AttesterDuty{{ + Pubkey: keys.pub[:], ValidatorIndex: 42, + Slot: primitives.Slot(epoch) * spe, CommitteeIndex: 1, CommitteeLength: 64, CommitteesAtSlot: 4, + }}, + }, nil) + client.EXPECT().AttesterDuties(gomock.Any(), epoch+1, gomock.Any()).Return(ðpb.AttesterDutiesResponse{}, nil) + client.EXPECT().ProposerDuties(gomock.Any(), epoch).Return(ðpb.ProposerDutiesResponse{}, nil) + client.EXPECT().ProposerDuties(gomock.Any(), epoch+1).Return(ðpb.ProposerDutiesResponse{}, nil) + client.EXPECT().SyncCommitteeDuties(gomock.Any(), gomock.Any(), gomock.Any()).Return(ðpb.SyncCommitteeDutiesResponse{}, nil).Times(2) + client.EXPECT().PTCDuties(gomock.Any(), gomock.Any(), gomock.Any()).Return(ðpb.PTCDutiesResponse{}, nil).Times(2) + + require.NoError(t, v.updateDutiesSplit(t.Context(), epoch, []primitives.ValidatorIndex{42})) + // After a full fetch, missingNext is reset. + require.Equal(t, missingNextDuties(0), v.duties.data.missingNext) + }) + + t.Run("promote refreshes Status from pubkeyToStatus", func(t *testing.T) { + v, client, keys := setup(t) + spe := params.BeaconConfig().SlotsPerEpoch + + // Seed the store as if the prior fetch saw the validator as PENDING + // (activation epoch reached, so it was admitted into the duty set). + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{ + NextEpochDuties: []*ethpb.ValidatorDuty{{ + PublicKey: keys.pub[:], ValidatorIndex: 42, + AttesterSlot: primitives.Slot(epoch)*spe + 3, + Status: ethpb.ValidatorStatus_PENDING, + }}, + CurrDependentRoot: bytesutil.PadTo([]byte{0xaa}, 32), + }) + data.epoch = epoch - 1 + data.indices = []primitives.ValidatorIndex{42} + v.duties.write(data) + } + + root := bytesutil.PadTo([]byte{0x01}, 32) + client.EXPECT().AttesterDuties(gomock.Any(), epoch+1, gomock.Any()).Return(ðpb.AttesterDutiesResponse{ + DependentRoot: root, + Duties: []*ethpb.AttesterDuty{{ + Pubkey: keys.pub[:], ValidatorIndex: 42, + Slot: primitives.Slot(epoch+1)*spe + 7, CommitteeIndex: 2, CommitteeLength: 64, CommitteesAtSlot: 4, + }}, + }, nil) + client.EXPECT().ProposerDuties(gomock.Any(), epoch+1).Return(ðpb.ProposerDutiesResponse{DependentRoot: root}, nil) + client.EXPECT().SyncCommitteeDuties(gomock.Any(), epoch+1, gomock.Any()).Return(ðpb.SyncCommitteeDutiesResponse{}, nil) + client.EXPECT().PTCDuties(gomock.Any(), epoch+1, gomock.Any()).Return(ðpb.PTCDutiesResponse{}, nil) + + require.NoError(t, v.updateDutiesSplit(t.Context(), epoch, []primitives.ValidatorIndex{42})) + + snap := v.duties.snapshot() + require.Equal(t, 1, snap.currentDutyCount()) + for _, d := range snap.currentDuties() { + assert.Equal(t, ethpb.ValidatorStatus_ACTIVE, d.Status) + } + }) +} + +func TestIsActiveForDuties(t *testing.T) { + tests := []struct { + name string + status *ethpb.ValidatorStatusResponse + epoch primitives.Epoch + expected bool + }{ + {"nil", nil, 5, false}, + {"unknown", ðpb.ValidatorStatusResponse{Status: ethpb.ValidatorStatus_UNKNOWN_STATUS}, 5, false}, + {"deposited", ðpb.ValidatorStatusResponse{Status: ethpb.ValidatorStatus_DEPOSITED}, 5, false}, + {"pending before activation", ðpb.ValidatorStatusResponse{Status: ethpb.ValidatorStatus_PENDING, ActivationEpoch: 10}, 5, false}, + {"pending at activation", ðpb.ValidatorStatusResponse{Status: ethpb.ValidatorStatus_PENDING, ActivationEpoch: 5}, 5, true}, + {"pending after activation", ðpb.ValidatorStatusResponse{Status: ethpb.ValidatorStatus_PENDING, ActivationEpoch: 3}, 5, true}, + {"active", ðpb.ValidatorStatusResponse{Status: ethpb.ValidatorStatus_ACTIVE}, 5, true}, + {"exiting", ðpb.ValidatorStatusResponse{Status: ethpb.ValidatorStatus_EXITING}, 5, true}, + {"slashing", ðpb.ValidatorStatusResponse{Status: ethpb.ValidatorStatus_SLASHING}, 5, false}, + {"exited", ðpb.ValidatorStatusResponse{Status: ethpb.ValidatorStatus_EXITED}, 5, false}, + {"invalid", ðpb.ValidatorStatusResponse{Status: ethpb.ValidatorStatus_INVALID}, 5, false}, + {"partially deposited", ðpb.ValidatorStatusResponse{Status: ethpb.ValidatorStatus_PARTIALLY_DEPOSITED}, 5, false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.expected, isActiveForDuties(tt.status, tt.epoch)) + }) + } +} + +func TestFilteredKeysAndIndices(t *testing.T) { + pkActive := bytesutil.ToBytes48([]byte{1}) + pkPending := bytesutil.ToBytes48([]byte{2}) + pkExited := bytesutil.ToBytes48([]byte{3}) + pkUnknown := bytesutil.ToBytes48([]byte{4}) // not in pubkeyToStatus + pkActive2 := bytesutil.ToBytes48([]byte{5}) + + v := &validator{ + pubkeyToStatus: map[pubkey]*validatorStatus{ + pkActive: {status: ðpb.ValidatorStatusResponse{Status: ethpb.ValidatorStatus_ACTIVE}, index: 99}, + pkPending: {status: ðpb.ValidatorStatusResponse{Status: ethpb.ValidatorStatus_PENDING, ActivationEpoch: 10}, index: 50}, + pkExited: {status: ðpb.ValidatorStatusResponse{Status: ethpb.ValidatorStatus_EXITED}, index: 7}, + // pkActive2 has a smaller index than pkActive to verify sorting. + pkActive2: {status: ðpb.ValidatorStatusResponse{Status: ethpb.ValidatorStatus_ACTIVE}, index: 3}, + }, + } + + // At epoch 5, pkPending's activation epoch (10) hasn't been reached. + keys, idx := v.filteredKeysAndIndices([][fieldparams.BLSPubkeyLength]byte{pkActive, pkPending, pkExited, pkUnknown, pkActive2}, 5) + + // Indices are sorted; pkActive2 (3) precedes pkActive (99). + require.DeepEqual(t, []primitives.ValidatorIndex{3, 99}, idx) + require.Equal(t, 2, len(keys)) + + // At epoch 10, pkPending qualifies (activation epoch reached). + keys, idx = v.filteredKeysAndIndices([][fieldparams.BLSPubkeyLength]byte{pkActive, pkPending, pkExited, pkUnknown, pkActive2}, 10) + require.DeepEqual(t, []primitives.ValidatorIndex{3, 50, 99}, idx) + require.Equal(t, 3, len(keys)) +} diff --git a/validator/client/duty_store.go b/validator/client/duty_store.go index 5b9038f88364..e68b4f5e3a14 100644 --- a/validator/client/duty_store.go +++ b/validator/client/duty_store.go @@ -1,153 +1,388 @@ package client import ( + "bytes" + "iter" + "slices" + "sync" + fieldparams "github.com/OffchainLabs/prysm/v7/config/fieldparams" "github.com/OffchainLabs/prysm/v7/consensus-types/primitives" "github.com/OffchainLabs/prysm/v7/encoding/bytesutil" ethpb "github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1" ) -type pubkey = [fieldparams.BLSPubkeyLength]byte +// cloneValidatorDuty returns a deep copy: scalar fields are copied by value +// and slice fields are independently allocated, so the returned duty shares +// no memory with d. +func cloneValidatorDuty(d *ethpb.ValidatorDuty) *ethpb.ValidatorDuty { + if d == nil { + return nil + } + return ðpb.ValidatorDuty{ + CommitteeLength: d.CommitteeLength, + CommitteeIndex: d.CommitteeIndex, + CommitteesAtSlot: d.CommitteesAtSlot, + ValidatorCommitteeIndex: d.ValidatorCommitteeIndex, + AttesterSlot: d.AttesterSlot, + ProposerSlots: slices.Clone(d.ProposerSlots), + PublicKey: bytes.Clone(d.PublicKey), + Status: d.Status, + ValidatorIndex: d.ValidatorIndex, + IsSyncCommittee: d.IsSyncCommittee, + PtcSlots: slices.Clone(d.PtcSlots), + } +} -// dutyStore stores validator duties from the beacon node. -// Both the legacy combined endpoint and the split per-duty endpoints -// populate the same internal maps, so accessor methods have a single code path. -type dutyStore struct { - currentDuties map[pubkey]*ethpb.ValidatorDuty - nextDuties map[pubkey]*ethpb.ValidatorDuty +type pubkey = [fieldparams.BLSPubkeyLength]byte - prevDependentRoot []byte +// dutyStoreData holds duty state with no synchronization. Methods on this +// type never lock; the surrounding dutyStore is responsible for serializing +// access. Maps and slices are aliased by snapshots rather than deep copied, +// which is safe because writers replace them wholesale via setFromContainer. +type dutyStoreData struct { + missingNext missingNextDuties + initialized bool + syncNextMap map[primitives.ValidatorIndex]bool + syncCurrentMap map[primitives.ValidatorIndex]bool + ptcSlots map[primitives.ValidatorIndex][]primitives.Slot + proposerSlots map[primitives.ValidatorIndex][]primitives.Slot + nextDuties map[pubkey]*ethpb.ValidatorDuty + currentDuties map[pubkey]*ethpb.ValidatorDuty + epoch primitives.Epoch currDependentRoot []byte + prevDependentRoot []byte + // indices is the sorted set of validator indices the last fetch was built + // from. canPromote requires this to match the current request's indices. + indices []primitives.ValidatorIndex +} - proposerSlots map[primitives.ValidatorIndex][]primitives.Slot - ptcSlots map[primitives.ValidatorIndex][]primitives.Slot - syncCurrentMap map[primitives.ValidatorIndex]bool - syncNextMap map[primitives.ValidatorIndex]bool +func (d *dutyStoreData) isInitialized() bool { return d.initialized } - initialized bool +func (d *dutyStoreData) currentDuty(pk pubkey) (*ethpb.ValidatorDuty, bool) { + if !d.initialized { + return nil, false + } + v, ok := d.currentDuties[pk] + if !ok { + return nil, false + } + return cloneValidatorDuty(v), true } -// Reset clears all duty data, marking the store as uninitialized. -func (ds *dutyStore) Reset() { - *ds = dutyStore{} +func (d *dutyStoreData) isSyncCommittee(idx primitives.ValidatorIndex) bool { + if !d.initialized { + return false + } + return d.syncCurrentMap[idx] } -// IsInitialized returns true if any duty data has been populated. -func (ds *dutyStore) IsInitialized() bool { - if ds == nil { +func (d *dutyStoreData) isNextSyncCommittee(idx primitives.ValidatorIndex) bool { + if !d.initialized { return false } - return ds.initialized + return d.syncNextMap[idx] } -// DependentRoots returns the previous and current dependent roots. -func (ds *dutyStore) DependentRoots() (prev, curr []byte) { - if !ds.IsInitialized() { - return nil, nil +func (d *dutyStoreData) canPromote(nextEpoch primitives.Epoch, indices []primitives.ValidatorIndex) bool { + if !d.initialized || d.epoch+1 != nextEpoch || d.missingNext != 0 { + return false + } + // Both slices are kept sorted; differing length or any element mismatch + // signals a validator-set drift (activation, exit, keymanager change) and + // invalidates the cached duties for promotion. + if len(d.indices) != len(indices) { + return false } - return ds.prevDependentRoot, ds.currDependentRoot + for i, idx := range d.indices { + if idx != indices[i] { + return false + } + } + return true } -// CurrentEpochDuties returns the current epoch duties. -func (ds *dutyStore) CurrentEpochDuties() map[pubkey]*ethpb.ValidatorDuty { - if !ds.IsInitialized() { - return nil +func (d *dutyStoreData) toContainer() *ethpb.ValidatorDutiesContainer { + if !d.initialized { + return ðpb.ValidatorDutiesContainer{} + } + current := make([]*ethpb.ValidatorDuty, 0, len(d.currentDuties)) + for _, duty := range d.currentDuties { + current = append(current, duty) + } + next := make([]*ethpb.ValidatorDuty, 0, len(d.nextDuties)) + for _, duty := range d.nextDuties { + next = append(next, duty) } - return ds.currentDuties + return ðpb.ValidatorDutiesContainer{ + PrevDependentRoot: d.prevDependentRoot, + CurrDependentRoot: d.currDependentRoot, + CurrentEpochDuties: current, + NextEpochDuties: next, + } +} + +// reset returns d to its zero value. Clearing every field is what keeps stale +// state (notably indices, which canPromote keys on) from surviving a rebuild. +func (d *dutyStoreData) reset() { + *d = dutyStoreData{} } -// NextEpochDuties returns the next epoch duties. -func (ds *dutyStore) NextEpochDuties() map[pubkey]*ethpb.ValidatorDuty { - if !ds.IsInitialized() { +func (d *dutyStoreData) setFromContainer(container *ethpb.ValidatorDutiesContainer) { + // Rebuild from scratch so no field can leak from a prior fetch, even if this + // is ever called on an already-populated struct. + d.reset() + if container == nil { + return + } + + d.proposerSlots = make(map[primitives.ValidatorIndex][]primitives.Slot) + d.ptcSlots = make(map[primitives.ValidatorIndex][]primitives.Slot) + d.syncCurrentMap = make(map[primitives.ValidatorIndex]bool) + d.syncNextMap = make(map[primitives.ValidatorIndex]bool) + + d.currentDuties = make(map[pubkey]*ethpb.ValidatorDuty, len(container.CurrentEpochDuties)) + for _, duty := range container.CurrentEpochDuties { + if duty == nil { + continue + } + d.currentDuties[bytesutil.ToBytes48(duty.PublicKey)] = duty + if len(duty.ProposerSlots) > 0 { + d.proposerSlots[duty.ValidatorIndex] = duty.ProposerSlots + } + if duty.IsSyncCommittee { + d.syncCurrentMap[duty.ValidatorIndex] = true + } + if len(duty.PtcSlots) > 0 { + d.ptcSlots[duty.ValidatorIndex] = duty.PtcSlots + } + } + + d.nextDuties = make(map[pubkey]*ethpb.ValidatorDuty, len(container.NextEpochDuties)) + for _, duty := range container.NextEpochDuties { + if duty == nil { + continue + } + d.nextDuties[bytesutil.ToBytes48(duty.PublicKey)] = duty + if duty.IsSyncCommittee { + d.syncNextMap[duty.ValidatorIndex] = true + } + } + + d.prevDependentRoot = container.PrevDependentRoot + d.currDependentRoot = container.CurrDependentRoot + d.initialized = true +} + +// dutyStore is the concurrency-safe wrapper around dutyStoreData. All methods +// acquire mu internally. Compound reads should use snapshot to get a coherent +// view without holding the lock across long operations. +type dutyStore struct { + mu sync.RWMutex + data dutyStoreData +} + +// roDutySnapshot is a read-only view of dutyStore. Getters return copies; the +// duty iterators yield aliases that callers must not mutate. +type roDutySnapshot struct { + d dutyStoreData +} + +func (s roDutySnapshot) isInitialized() bool { return s.d.isInitialized() } + +func (s roDutySnapshot) prevDependentRoot() []byte { + if !s.d.initialized { return nil } - return ds.nextDuties + return bytes.Clone(s.d.prevDependentRoot) } -// CurrentDuty returns the current epoch duty for a given pubkey. -func (ds *dutyStore) CurrentDuty(pk pubkey) (*ethpb.ValidatorDuty, bool) { - if !ds.IsInitialized() { - return nil, false +func (s roDutySnapshot) currDependentRoot() []byte { + if !s.d.initialized { + return nil } - d, ok := ds.currentDuties[pk] - return d, ok + return bytes.Clone(s.d.currDependentRoot) +} + +func (s roDutySnapshot) currentDuty(pk pubkey) (*ethpb.ValidatorDuty, bool) { + return s.d.currentDuty(pk) } -// ProposerSlots returns the proposer slots for a given validator index. -func (ds *dutyStore) ProposerSlots(idx primitives.ValidatorIndex) []primitives.Slot { - if !ds.IsInitialized() { +func (s roDutySnapshot) proposerSlots(idx primitives.ValidatorIndex) []primitives.Slot { + if !s.d.initialized { return nil } - return ds.proposerSlots[idx] + return slices.Clone(s.d.proposerSlots[idx]) } -// PtcSlots returns the PTC slots for a given validator index. -func (ds *dutyStore) PtcSlots(idx primitives.ValidatorIndex) []primitives.Slot { - if !ds.IsInitialized() { +func (s roDutySnapshot) ptcSlots(idx primitives.ValidatorIndex) []primitives.Slot { + if !s.d.initialized { return nil } - return ds.ptcSlots[idx] + return slices.Clone(s.d.ptcSlots[idx]) } -// IsSyncCommittee returns whether a validator is in the current sync committee. -func (ds *dutyStore) IsSyncCommittee(idx primitives.ValidatorIndex) bool { - if !ds.IsInitialized() { +func (s roDutySnapshot) isSyncCommittee(idx primitives.ValidatorIndex) bool { + return s.d.isSyncCommittee(idx) +} + +func (s roDutySnapshot) isNextSyncCommittee(idx primitives.ValidatorIndex) bool { + return s.d.isNextSyncCommittee(idx) +} + +// currentDuties yields read-only current-epoch duty aliases. Re-rangeable. +func (s roDutySnapshot) currentDuties() iter.Seq2[pubkey, *ethpb.ValidatorDuty] { + return func(yield func(pubkey, *ethpb.ValidatorDuty) bool) { + if !s.d.initialized { + return + } + for pk, duty := range s.d.currentDuties { + if !yield(pk, duty) { + return + } + } + } +} + +// nextDuties yields read-only next-epoch duty aliases. Re-rangeable. +func (s roDutySnapshot) nextDuties() iter.Seq2[pubkey, *ethpb.ValidatorDuty] { + return func(yield func(pubkey, *ethpb.ValidatorDuty) bool) { + if !s.d.initialized { + return + } + for pk, duty := range s.d.nextDuties { + if !yield(pk, duty) { + return + } + } + } +} + +func (s roDutySnapshot) currentDutyCount() int { + if !s.d.initialized { + return 0 + } + return len(s.d.currentDuties) +} + +func (s roDutySnapshot) nextDutyCount() int { + if !s.d.initialized { + return 0 + } + return len(s.d.nextDuties) +} + +// snapshot returns a coherent read-only view of the store. The returned value +// can be inspected without holding any lock; maps and slices alias internal +// state but are never mutated in place (setFromContainer replaces them). +func (ds *dutyStore) snapshot() roDutySnapshot { + if ds == nil { + return roDutySnapshot{} + } + ds.mu.RLock() + defer ds.mu.RUnlock() + return roDutySnapshot{d: ds.data} +} + +func (ds *dutyStore) reset() { + ds.mu.Lock() + defer ds.mu.Unlock() + ds.data.reset() +} + +func (ds *dutyStore) isInitialized() bool { + if ds == nil { return false } - return ds.syncCurrentMap[idx] + ds.mu.RLock() + defer ds.mu.RUnlock() + return ds.data.isInitialized() } -// IsNextSyncCommittee returns whether a validator is in the next epoch's sync committee. -func (ds *dutyStore) IsNextSyncCommittee(idx primitives.ValidatorIndex) bool { - if !ds.IsInitialized() { +func (ds *dutyStore) canPromote(nextEpoch primitives.Epoch, indices []primitives.ValidatorIndex) bool { + if ds == nil { return false } - return ds.syncNextMap[idx] + ds.mu.RLock() + defer ds.mu.RUnlock() + return ds.data.canPromote(nextEpoch, indices) } -// SetFromCombinedDutiesResponse stores a combined duties response by decomposing it into -// duty maps, proposer slots, and sync committee maps. -// DEPRECATED: GetDutiesV2, use the split GetAttesterDuties, GetProposerDutiesV2, GetSyncCommitteeDuties, GetPTCduties endpoints. -func (ds *dutyStore) SetFromCombinedDutiesResponse(container *ethpb.ValidatorDutiesContainer) { - if container == nil { - ds.Reset() - return +func (ds *dutyStore) prevDependentRoot() []byte { + ds.mu.RLock() + defer ds.mu.RUnlock() + if !ds.data.initialized { + return nil } + return bytes.Clone(ds.data.prevDependentRoot) +} - ds.proposerSlots = make(map[primitives.ValidatorIndex][]primitives.Slot) - ds.ptcSlots = make(map[primitives.ValidatorIndex][]primitives.Slot) - ds.syncCurrentMap = make(map[primitives.ValidatorIndex]bool) - ds.syncNextMap = make(map[primitives.ValidatorIndex]bool) +func (ds *dutyStore) currDependentRoot() []byte { + ds.mu.RLock() + defer ds.mu.RUnlock() + if !ds.data.initialized { + return nil + } + return bytes.Clone(ds.data.currDependentRoot) +} - ds.currentDuties = make(map[pubkey]*ethpb.ValidatorDuty, len(container.CurrentEpochDuties)) - for _, d := range container.CurrentEpochDuties { - if d == nil { - continue - } - ds.currentDuties[bytesutil.ToBytes48(d.PublicKey)] = d - if len(d.ProposerSlots) > 0 { - ds.proposerSlots[d.ValidatorIndex] = d.ProposerSlots - } - if d.IsSyncCommittee { - ds.syncCurrentMap[d.ValidatorIndex] = true - } - if len(d.PtcSlots) > 0 { - ds.ptcSlots[d.ValidatorIndex] = d.PtcSlots - } +// dependentRoots returns both dependent roots. Retained for compatibility +// with callers that want them in a single call; see prevDependentRoot and +// currDependentRoot for naming semantics. +func (ds *dutyStore) dependentRoots() (prev, curr []byte) { + ds.mu.RLock() + defer ds.mu.RUnlock() + if !ds.data.initialized { + return nil, nil } + return bytes.Clone(ds.data.prevDependentRoot), bytes.Clone(ds.data.currDependentRoot) +} - ds.nextDuties = make(map[pubkey]*ethpb.ValidatorDuty, len(container.NextEpochDuties)) - for _, d := range container.NextEpochDuties { - if d == nil { - continue - } - ds.nextDuties[bytesutil.ToBytes48(d.PublicKey)] = d - if d.IsSyncCommittee { - ds.syncNextMap[d.ValidatorIndex] = true - } +func (ds *dutyStore) currentDuty(pk pubkey) (*ethpb.ValidatorDuty, bool) { + ds.mu.RLock() + defer ds.mu.RUnlock() + return ds.data.currentDuty(pk) +} + +func (ds *dutyStore) proposerSlots(idx primitives.ValidatorIndex) []primitives.Slot { + ds.mu.RLock() + defer ds.mu.RUnlock() + if !ds.data.initialized { + return nil + } + return slices.Clone(ds.data.proposerSlots[idx]) +} + +func (ds *dutyStore) ptcSlots(idx primitives.ValidatorIndex) []primitives.Slot { + ds.mu.RLock() + defer ds.mu.RUnlock() + if !ds.data.initialized { + return nil } + return slices.Clone(ds.data.ptcSlots[idx]) +} + +func (ds *dutyStore) isSyncCommittee(idx primitives.ValidatorIndex) bool { + ds.mu.RLock() + defer ds.mu.RUnlock() + return ds.data.isSyncCommittee(idx) +} + +func (ds *dutyStore) isNextSyncCommittee(idx primitives.ValidatorIndex) bool { + ds.mu.RLock() + defer ds.mu.RUnlock() + return ds.data.isNextSyncCommittee(idx) +} + +func (ds *dutyStore) toContainer() *ethpb.ValidatorDutiesContainer { + ds.mu.RLock() + defer ds.mu.RUnlock() + return ds.data.toContainer() +} - ds.prevDependentRoot = container.PrevDependentRoot - ds.currDependentRoot = container.CurrDependentRoot - ds.initialized = true +// write atomically replaces the store's state with the given data. +func (ds *dutyStore) write(data dutyStoreData) { + ds.mu.Lock() + defer ds.mu.Unlock() + ds.data = data } diff --git a/validator/client/duty_store_test.go b/validator/client/duty_store_test.go index c9b58efde548..9d2f383ac616 100644 --- a/validator/client/duty_store_test.go +++ b/validator/client/duty_store_test.go @@ -1,6 +1,7 @@ package client import ( + "reflect" "testing" "github.com/OffchainLabs/prysm/v7/consensus-types/primitives" @@ -10,7 +11,8 @@ import ( ) func testDutyStore(current ...*ethpb.ValidatorDuty) *dutyStore { - ds := &dutyStore{ + ds := &dutyStore{} + ds.data = dutyStoreData{ currentDuties: make(map[pubkey]*ethpb.ValidatorDuty), nextDuties: make(map[pubkey]*ethpb.ValidatorDuty), proposerSlots: make(map[primitives.ValidatorIndex][]primitives.Slot), @@ -20,15 +22,15 @@ func testDutyStore(current ...*ethpb.ValidatorDuty) *dutyStore { initialized: true, } for _, d := range current { - ds.currentDuties[bytesutil.ToBytes48(d.PublicKey)] = d + ds.data.currentDuties[bytesutil.ToBytes48(d.PublicKey)] = d if len(d.ProposerSlots) > 0 { - ds.proposerSlots[d.ValidatorIndex] = d.ProposerSlots + ds.data.proposerSlots[d.ValidatorIndex] = d.ProposerSlots } if d.IsSyncCommittee { - ds.syncCurrentMap[d.ValidatorIndex] = true + ds.data.syncCurrentMap[d.ValidatorIndex] = true } if len(d.PtcSlots) > 0 { - ds.ptcSlots[d.ValidatorIndex] = d.PtcSlots + ds.data.ptcSlots[d.ValidatorIndex] = d.PtcSlots } } return ds @@ -36,30 +38,30 @@ func testDutyStore(current ...*ethpb.ValidatorDuty) *dutyStore { func TestDutyStore_Uninitialized(t *testing.T) { ds := &dutyStore{} - assert.Equal(t, false, ds.IsInitialized()) - assert.Equal(t, true, ds.CurrentEpochDuties() == nil) - assert.Equal(t, true, ds.NextEpochDuties() == nil) + assert.Equal(t, false, ds.isInitialized()) + snap := ds.snapshot() + assert.Equal(t, 0, snap.currentDutyCount()) + assert.Equal(t, 0, snap.nextDutyCount()) - prev, curr := ds.DependentRoots() - assert.Equal(t, true, prev == nil) - assert.Equal(t, true, curr == nil) + assert.Equal(t, true, ds.prevDependentRoot() == nil) + assert.Equal(t, true, ds.currDependentRoot() == nil) - d, ok := ds.CurrentDuty(pubkey{}) + d, ok := ds.currentDuty(pubkey{}) assert.Equal(t, false, ok) assert.Equal(t, (*ethpb.ValidatorDuty)(nil), d) - assert.Equal(t, true, ds.ProposerSlots(0) == nil) - assert.Equal(t, true, ds.PtcSlots(0) == nil) - assert.Equal(t, false, ds.IsSyncCommittee(0)) - assert.Equal(t, false, ds.IsNextSyncCommittee(0)) + assert.Equal(t, true, ds.proposerSlots(0) == nil) + assert.Equal(t, true, ds.ptcSlots(0) == nil) + assert.Equal(t, false, ds.isSyncCommittee(0)) + assert.Equal(t, false, ds.isNextSyncCommittee(0)) } func TestDutyStore_ZeroValueIsNotInitialized(t *testing.T) { ds := &dutyStore{} - assert.Equal(t, false, ds.IsInitialized()) + assert.Equal(t, false, ds.isInitialized()) } -func TestDutyStore_SetFromCombinedDutiesResponse(t *testing.T) { +func TestDutyStore_Write(t *testing.T) { pk1 := bytesutil.ToBytes48([]byte{1}) pk2 := bytesutil.ToBytes48([]byte{2}) @@ -87,68 +89,174 @@ func TestDutyStore_SetFromCombinedDutiesResponse(t *testing.T) { } ds := &dutyStore{} - ds.SetFromCombinedDutiesResponse(container) + { + var data dutyStoreData + data.setFromContainer(container) + ds.write(data) + } - assert.Equal(t, true, ds.IsInitialized()) + assert.Equal(t, true, ds.isInitialized()) // Current duties. - d, ok := ds.CurrentDuty(pk1) + d, ok := ds.currentDuty(pk1) assert.Equal(t, true, ok) assert.Equal(t, primitives.ValidatorIndex(10), d.ValidatorIndex) - _, ok = ds.CurrentDuty(pk2) + _, ok = ds.currentDuty(pk2) assert.Equal(t, false, ok) // Next duties. - next := ds.NextEpochDuties() - assert.Equal(t, 1, len(next)) - assert.Equal(t, primitives.ValidatorIndex(20), next[pk2].ValidatorIndex) + snap := ds.snapshot() + assert.Equal(t, 1, snap.nextDutyCount()) + for pk, duty := range snap.nextDuties() { + assert.Equal(t, pk2, pk) + assert.Equal(t, primitives.ValidatorIndex(20), duty.ValidatorIndex) + } // Dependent roots. - prev, curr := ds.DependentRoots() - assert.DeepEqual(t, []byte("prev"), prev) - assert.DeepEqual(t, []byte("curr"), curr) + assert.DeepEqual(t, []byte("prev"), ds.prevDependentRoot()) + assert.DeepEqual(t, []byte("curr"), ds.currDependentRoot()) // Proposer slots. - assert.DeepEqual(t, []primitives.Slot{3, 7}, ds.ProposerSlots(10)) - assert.Equal(t, true, ds.ProposerSlots(20) == nil) + assert.DeepEqual(t, []primitives.Slot{3, 7}, ds.proposerSlots(10)) + assert.Equal(t, true, ds.proposerSlots(20) == nil) // PTC slots. - assert.DeepEqual(t, []primitives.Slot{4, 6}, ds.PtcSlots(10)) - assert.Equal(t, true, ds.PtcSlots(20) == nil) + assert.DeepEqual(t, []primitives.Slot{4, 6}, ds.ptcSlots(10)) + assert.Equal(t, true, ds.ptcSlots(20) == nil) // Sync committee. - assert.Equal(t, true, ds.IsSyncCommittee(10)) - assert.Equal(t, false, ds.IsSyncCommittee(20)) - assert.Equal(t, false, ds.IsNextSyncCommittee(10)) - assert.Equal(t, true, ds.IsNextSyncCommittee(20)) + assert.Equal(t, true, ds.isSyncCommittee(10)) + assert.Equal(t, false, ds.isSyncCommittee(20)) + assert.Equal(t, false, ds.isNextSyncCommittee(10)) + assert.Equal(t, true, ds.isNextSyncCommittee(20)) } func TestDutyStore_Reset(t *testing.T) { ds := testDutyStore(ðpb.ValidatorDuty{PublicKey: make([]byte, 48)}) - ds.prevDependentRoot = []byte("prev") - ds.currDependentRoot = []byte("curr") - assert.Equal(t, true, ds.IsInitialized()) + ds.data.prevDependentRoot = []byte("prev") + ds.data.currDependentRoot = []byte("curr") + assert.Equal(t, true, ds.isInitialized()) - ds.Reset() - assert.Equal(t, false, ds.IsInitialized()) - assert.Equal(t, true, ds.CurrentEpochDuties() == nil) + ds.reset() + assert.Equal(t, false, ds.isInitialized()) + assert.Equal(t, 0, ds.snapshot().currentDutyCount()) } -func TestDutyStore_SetFromCombinedDutiesResponseNilResets(t *testing.T) { +func TestDutyStoreData_Reset(t *testing.T) { + populated := func() dutyStoreData { + return dutyStoreData{ + initialized: true, + epoch: 9, + missingNext: missingNextPtc, + indices: []primitives.ValidatorIndex{1, 5, 7}, + currentDuties: map[pubkey]*ethpb.ValidatorDuty{{}: {}}, + prevDependentRoot: []byte("prev"), + } + } + + t.Run("reset zeroes every field", func(t *testing.T) { + d := populated() + d.reset() + // Covers every field, including any added later: IsZero reports whether + // the whole struct equals its zero value. + assert.Equal(t, true, reflect.ValueOf(d).IsZero()) + }) + + t.Run("setFromContainer drops stale indices on a populated struct", func(t *testing.T) { + d := populated() + d.setFromContainer(ðpb.ValidatorDutiesContainer{ + CurrentEpochDuties: []*ethpb.ValidatorDuty{{PublicKey: make([]byte, 48), ValidatorIndex: 2}}, + }) + assert.Equal(t, true, d.indices == nil) + // With indices cleared, a stale validator set can't satisfy canPromote + // even when the epoch lines up. + d.epoch = 9 + assert.Equal(t, false, d.canPromote(10, []primitives.ValidatorIndex{1, 5, 7})) + }) +} + +func TestDutyStore_WriteNilResets(t *testing.T) { ds := testDutyStore(ðpb.ValidatorDuty{PublicKey: make([]byte, 48)}) - assert.Equal(t, true, ds.IsInitialized()) + assert.Equal(t, true, ds.isInitialized()) - ds.SetFromCombinedDutiesResponse(nil) - assert.Equal(t, false, ds.IsInitialized()) + { + var data dutyStoreData + data.setFromContainer(nil) + ds.write(data) + } + assert.Equal(t, false, ds.isInitialized()) } -func TestDutyStore_SetFromCombinedDutiesResponseSkipsNilDuties(t *testing.T) { +func TestDutyStore_WriteSkipsNilDuties(t *testing.T) { ds := &dutyStore{} - ds.SetFromCombinedDutiesResponse(ðpb.ValidatorDutiesContainer{ - CurrentEpochDuties: []*ethpb.ValidatorDuty{nil, {PublicKey: make([]byte, 48), ValidatorIndex: 1}}, - NextEpochDuties: []*ethpb.ValidatorDuty{nil}, + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{ + CurrentEpochDuties: []*ethpb.ValidatorDuty{nil, {PublicKey: make([]byte, 48), ValidatorIndex: 1}}, + NextEpochDuties: []*ethpb.ValidatorDuty{nil}, + }) + ds.write(data) + } + snap := ds.snapshot() + assert.Equal(t, 1, snap.currentDutyCount()) + assert.Equal(t, 0, snap.nextDutyCount()) +} + +func TestDutyStoreData_CanPromote(t *testing.T) { + base := func() dutyStoreData { + return dutyStoreData{ + initialized: true, + epoch: 9, + indices: []primitives.ValidatorIndex{1, 5, 7}, + } + } + + t.Run("happy path: matching epoch + indices + zero missing", func(t *testing.T) { + d := base() + assert.Equal(t, true, d.canPromote(10, []primitives.ValidatorIndex{1, 5, 7})) + }) + + t.Run("uninitialized cannot promote", func(t *testing.T) { + d := base() + d.initialized = false + assert.Equal(t, false, d.canPromote(10, []primitives.ValidatorIndex{1, 5, 7})) + }) + + t.Run("non-adjacent epoch cannot promote", func(t *testing.T) { + d := base() + assert.Equal(t, false, d.canPromote(11, []primitives.ValidatorIndex{1, 5, 7})) + assert.Equal(t, false, d.canPromote(9, []primitives.ValidatorIndex{1, 5, 7})) + }) + + t.Run("non-zero missingNext blocks promote", func(t *testing.T) { + d := base() + d.missingNext = missingNextPtc + assert.Equal(t, false, d.canPromote(10, []primitives.ValidatorIndex{1, 5, 7})) + }) + + t.Run("drift: added index blocks promote", func(t *testing.T) { + d := base() + assert.Equal(t, false, d.canPromote(10, []primitives.ValidatorIndex{1, 5, 7, 9})) + }) + + t.Run("drift: removed index blocks promote", func(t *testing.T) { + d := base() + assert.Equal(t, false, d.canPromote(10, []primitives.ValidatorIndex{1, 7})) + }) + + t.Run("drift: substituted index blocks promote", func(t *testing.T) { + d := base() + assert.Equal(t, false, d.canPromote(10, []primitives.ValidatorIndex{1, 5, 8})) + }) + + t.Run("nil current indices treated as empty: blocks promote when stored is non-empty", func(t *testing.T) { + d := base() + assert.Equal(t, false, d.canPromote(10, nil)) + }) + + t.Run("both empty index sets is a no-op promote", func(t *testing.T) { + d := dutyStoreData{initialized: true, epoch: 9} + assert.Equal(t, true, d.canPromote(10, nil)) }) - assert.Equal(t, 1, len(ds.CurrentEpochDuties())) - assert.Equal(t, 0, len(ds.NextEpochDuties())) } diff --git a/validator/client/payload_attestation_test.go b/validator/client/payload_attestation_test.go index 33523a9176aa..873a3ae90aaf 100644 --- a/validator/client/payload_attestation_test.go +++ b/validator/client/payload_attestation_test.go @@ -83,7 +83,11 @@ func TestSubmitPayloadAttestation_ValidatorDutiesRequestFailure(t *testing.T) { hook := logTest.NewGlobal() validator, m, validatorKey, finish := setup(t, isSlashingProtectionMinimal) validator.duties = &dutyStore{} - validator.duties.SetFromCombinedDutiesResponse(ðpb.ValidatorDutiesContainer{CurrentEpochDuties: []*ethpb.ValidatorDuty{}}) + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{CurrentEpochDuties: []*ethpb.ValidatorDuty{}}) + validator.duties.write(data) + } defer finish() m.validatorClient.EXPECT(). @@ -119,12 +123,16 @@ func TestSubmitPayloadAttestation_BadDomainData(t *testing.T) { defer finish() validatorIndex := primitives.ValidatorIndex(7) validator.duties = &dutyStore{} - validator.duties.SetFromCombinedDutiesResponse(ðpb.ValidatorDutiesContainer{CurrentEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: validatorKey.PublicKey().Marshal(), - ValidatorIndex: validatorIndex, - }, - }}) + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{CurrentEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: validatorKey.PublicKey().Marshal(), + ValidatorIndex: validatorIndex, + }, + }}) + validator.duties.write(data) + } m.validatorClient.EXPECT(). PayloadAttestationData(gomock.Any(), gomock.Any()). @@ -159,12 +167,16 @@ func TestSubmitPayloadAttestation_CouldNotSubmit(t *testing.T) { defer finish() validatorIndex := primitives.ValidatorIndex(7) validator.duties = &dutyStore{} - validator.duties.SetFromCombinedDutiesResponse(ðpb.ValidatorDutiesContainer{CurrentEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: validatorKey.PublicKey().Marshal(), - ValidatorIndex: validatorIndex, - }, - }}) + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{CurrentEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: validatorKey.PublicKey().Marshal(), + ValidatorIndex: validatorIndex, + }, + }}) + validator.duties.write(data) + } m.validatorClient.EXPECT(). PayloadAttestationData(gomock.Any(), gomock.Any()). @@ -203,12 +215,16 @@ func TestSubmitPayloadAttestation_OK(t *testing.T) { defer finish() validatorIndex := primitives.ValidatorIndex(7) validator.duties = &dutyStore{} - validator.duties.SetFromCombinedDutiesResponse(ðpb.ValidatorDutiesContainer{CurrentEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: validatorKey.PublicKey().Marshal(), - ValidatorIndex: validatorIndex, - }, - }}) + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{CurrentEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: validatorKey.PublicKey().Marshal(), + ValidatorIndex: validatorIndex, + }, + }}) + validator.duties.write(data) + } blockRoot := bytesutil.PadTo([]byte{'b'}, 32) m.validatorClient.EXPECT(). diff --git a/validator/client/validator.go b/validator/client/validator.go index 97af8a2e5b5c..0afbe451cd6e 100644 --- a/validator/client/validator.go +++ b/validator/client/validator.go @@ -10,6 +10,7 @@ import ( "encoding/json" "fmt" "io" + "iter" "slices" "strconv" "strings" @@ -64,56 +65,56 @@ var ( ) type validator struct { - logValidatorPerformance bool distributed bool enableAPI bool disableDutiesPolling bool emitAccountMetrics bool + logValidatorPerformance bool attLogsLock sync.Mutex highestValidSlotLock sync.Mutex - domainDataLock sync.RWMutex blacklistedPubkeysLock sync.RWMutex prevEpochBalancesLock sync.RWMutex cachedAttestationDataLock sync.RWMutex - dutiesLock sync.RWMutex - aggSelector aggregatorSelector + submittedPrefSlotsLock sync.RWMutex + domainDataLock sync.RWMutex cachedAttestationData *ethpb.AttestationData + graffitiOrderedIndex uint64 + walletInitializedFeed *event.Feed + walletInitializedChan chan *wallet.Wallet + wallet *wallet.Wallet accountsChangedChannel chan [][fieldparams.BLSPubkeyLength]byte - eventsChannel chan *eventClient.Event - payloadAvailability *payloadAvailability - highestValidSlot primitives.Slot - submittedAggregates map[submittedAttKey]*submittedAtt - graffitiStruct *graffiti.Graffiti - syncCommitteeStats syncCommitteeStats - slotFeed *event.Feed - domainDataCache *ristretto.Cache[string, proto.Message] - interopKeysConfig *local.InteropKeymanagerConfig - duties *dutyStore - signedValidatorRegistrations map[[fieldparams.BLSPubkeyLength]byte]*ethpb.SignedValidatorRegistrationV1 - submittedPrefSlots map[primitives.Slot]bool - proposerSettings *proposer.Settings - web3SignerConfig *remoteweb3signer.SetupConfig - startBalances map[[fieldparams.BLSPubkeyLength]byte]uint64 - prevEpochBalances map[[fieldparams.BLSPubkeyLength]byte]uint64 blacklistedPubkeys map[[fieldparams.BLSPubkeyLength]byte]bool - pubkeyToStatus map[[fieldparams.BLSPubkeyLength]byte]*validatorStatus - wallet *wallet.Wallet - walletInitializedChan chan *wallet.Wallet - walletInitializedFeed *event.Feed - graffitiOrderedIndex uint64 - conn validatorHelpers.NodeConnection + prevEpochBalances map[[fieldparams.BLSPubkeyLength]byte]uint64 + startBalances map[[fieldparams.BLSPubkeyLength]byte]uint64 + web3SignerConfig *remoteweb3signer.SetupConfig + proposerSettings *proposer.Settings + submittedPrefSlots map[primitives.Slot]bool submittedAtts map[submittedAttKey]*submittedAtt validatorsRegBatchSize int + duties *dutyStore + interopKeysConfig *local.InteropKeymanagerConfig + domainDataCache *ristretto.Cache[string, proto.Message] + slotFeed *event.Feed + syncCommitteeStats syncCommitteeStats + graffitiStruct *graffiti.Graffiti + submittedAggregates map[submittedAttKey]*submittedAtt + highestValidSlot primitives.Slot + eventsChannel chan *eventClient.Event + payloadAvailability *payloadAvailability + pubkeyToStatus map[[fieldparams.BLSPubkeyLength]byte]*validatorStatus + signedValidatorRegistrations map[[fieldparams.BLSPubkeyLength]byte]*ethpb.SignedValidatorRegistrationV1 + aggSelector aggregatorSelector validatorClient iface.ValidatorClient chainClient iface.ChainClient nodeClient iface.NodeClient prysmChainClient iface.PrysmChainClient db db.Database - km keymanager.IKeymanager + conn validatorHelpers.NodeConnection accountChangedSub event.Subscription ticker slots.Ticker - genesisTime time.Time + km keymanager.IKeymanager graffiti []byte + genesisTime time.Time voteStats voteStats } @@ -542,10 +543,8 @@ func (v *validator) RolesAt(ctx context.Context, slot primitives.Slot) (map[[fie ctx, span := trace.StartSpan(ctx, "validator.RolesAt") defer span.End() - v.dutiesLock.RLock() - defer v.dutiesLock.RUnlock() - - if !v.duties.IsInitialized() { + snap := v.duties.snapshot() + if !snap.isInitialized() { return nil, errors.New("validator duties are not initialized") } @@ -554,7 +553,7 @@ func (v *validator) RolesAt(ctx context.Context, slot primitives.Slot) (map[[fie syncCommitteePubkeys [][fieldparams.BLSPubkeyLength]byte ) - for pk, duty := range v.duties.CurrentEpochDuties() { + for pk, duty := range snap.currentDuties() { var roles []iface.ValidatorRole if duty == nil { @@ -587,7 +586,7 @@ func (v *validator) RolesAt(ctx context.Context, slot primitives.Slot) (map[[fie // the validator checks whether it's in the sync committee of following epoch. inSyncCommittee := false if slots.IsEpochEnd(slot) { - if v.duties.IsNextSyncCommittee(duty.ValidatorIndex) { + if snap.isNextSyncCommittee(duty.ValidatorIndex) { roles = append(roles, iface.RoleSyncCommittee) inSyncCommittee = true } @@ -602,7 +601,7 @@ func (v *validator) RolesAt(ctx context.Context, slot primitives.Slot) (map[[fie syncCommitteePubkeys = append(syncCommitteePubkeys, pk) } - if slices.Contains(v.duties.PtcSlots(duty.ValidatorIndex), slot) { + if slices.Contains(snap.ptcSlots(duty.ValidatorIndex), slot) { roles = append(roles, iface.RolePTCMember) } @@ -950,14 +949,9 @@ func (v *validator) filterAndCacheActiveKeys(ctx context.Context, pubkeys [][fie return nil, errors.Wrap(err, "failed to update validator status cache") } } + currEpoch := slots.ToEpoch(slot) for k, s := range v.pubkeyToStatus { - currEpoch := primitives.Epoch(slot / params.BeaconConfig().SlotsPerEpoch) - currActivating := s.status.Status == ethpb.ValidatorStatus_PENDING && currEpoch >= s.status.ActivationEpoch - - active := s.status.Status == ethpb.ValidatorStatus_ACTIVE - exiting := s.status.Status == ethpb.ValidatorStatus_EXITING - - if currActivating || active || exiting { + if isActiveForDuties(s.status, currEpoch) { filteredKeys = append(filteredKeys, k) } else { log.WithFields(logrus.Fields{ @@ -1069,6 +1063,7 @@ func (v *validator) buildProposerPreferences( } midEpoch := epochStart + params.BeaconConfig().SlotsPerEpoch/2 + v.submittedPrefSlotsLock.Lock() if force { v.submittedPrefSlots = make(map[primitives.Slot]bool) } else { @@ -1078,14 +1073,13 @@ func (v *validator) buildProposerPreferences( } } } - v.dutiesLock.RLock() - defer v.dutiesLock.RUnlock() + v.submittedPrefSlotsLock.Unlock() - if !v.duties.IsInitialized() { + snap := v.duties.snapshot() + if !snap.isInitialized() { return nil } - ps := v.ProposerSettings() var signedPrefs []*ethpb.SignedProposerPreferences var sigFailCount int @@ -1093,76 +1087,10 @@ func (v *validator) buildProposerPreferences( // dependent root the beacon node uses to compute proposer duties for E: // - proposal in current epoch → previous_duty_dependent_root // - proposal in next epoch → current_duty_dependent_root - prevDepRoot, currDepRoot := v.duties.DependentRoots() - - processDuties := func(duties map[pubkey]*ethpb.ValidatorDuty, isNextEpoch bool) { - dependentRoot := prevDepRoot - if isNextEpoch { - dependentRoot = currDepRoot - } - if len(dependentRoot) != fieldparams.RootLength { - return - } - for pk, duty := range duties { - if len(duty.ProposerSlots) == 0 { - continue - } - if duty.Status != ethpb.ValidatorStatus_ACTIVE && duty.Status != ethpb.ValidatorStatus_EXITING { - continue - } - - feeRecipient := common.HexToAddress(params.BeaconConfig().EthBurnAddressHex) - gasLimit := params.BeaconConfig().DefaultBuilderGasLimit - if ps != nil && ps.DefaultConfig != nil { - if ps.DefaultConfig.FeeRecipientConfig != nil { - feeRecipient = ps.DefaultConfig.FeeRecipientConfig.FeeRecipient - } - if ps.DefaultConfig.BuilderConfig != nil && ps.DefaultConfig.BuilderConfig.Enabled { - gasLimit = uint64(ps.DefaultConfig.BuilderConfig.GasLimit) - } - } - if ps != nil && ps.ProposeConfig != nil { - if config, ok := ps.ProposeConfig[pk]; ok && config != nil { - if config.FeeRecipientConfig != nil { - feeRecipient = config.FeeRecipientConfig.FeeRecipient - } - if config.BuilderConfig != nil && config.BuilderConfig.Enabled { - gasLimit = uint64(config.BuilderConfig.GasLimit) - } - } - } - - for _, proposalSlot := range duty.ProposerSlots { - if v.submittedPrefSlots[proposalSlot] { - continue - } - // Skip slots that have passed or are too close. Preferences are - // submitted at mid-slot, so the proposer needs to be at least 1 - // full slot away for the beacon node to receive them in time. - if !isNextEpoch && proposalSlot <= slot+1 { - continue - } - - pref := ðpb.ProposerPreferences{ - DependentRoot: dependentRoot, - ProposalSlot: proposalSlot, - ValidatorIndex: duty.ValidatorIndex, - FeeRecipient: feeRecipient[:], - TargetGasLimit: gasLimit, - } - signedPref, err := v.signProposerPreferences(ctx, km, pk, pref) - if err != nil { - sigFailCount++ - continue - } - signedPrefs = append(signedPrefs, signedPref) - v.submittedPrefSlots[proposalSlot] = true - } - } - } + prevDepRoot, currDepRoot := v.duties.dependentRoots() - currentDuties := v.duties.CurrentEpochDuties() - nextDuties := v.duties.NextEpochDuties() + currentDuties := snap.currentDuties() + nextDuties := snap.nextDuties() var currentProposerCount, nextProposerCount int for _, d := range currentDuties { @@ -1175,13 +1103,17 @@ func (v *validator) buildProposerPreferences( // Current-epoch: submit after first slot of epoch to avoid stale state. // force bypasses the timing gate for reorg resubmission. if currentEpoch >= gloasEpoch && (force || slot > epochStart) { - processDuties(currentDuties, false) + signed, fails := v.processProposerDuties(ctx, km, currentDuties, slot, prevDepRoot, false) + signedPrefs = append(signedPrefs, signed...) + sigFailCount += fails } // Next-epoch: submit at or after mid-epoch. The gate is not bypassed // by force because the beacon node may not have the next-epoch state ready. if slot >= midEpoch { - processDuties(nextDuties, true) + signed, fails := v.processProposerDuties(ctx, km, nextDuties, slot, currDepRoot, true) + signedPrefs = append(signedPrefs, signed...) + sigFailCount += fails } if sigFailCount > 0 { @@ -1195,11 +1127,119 @@ func (v *validator) buildProposerPreferences( "currentProposerSlots": currentProposerCount, "nextProposerSlots": nextProposerCount, "prefsBuilt": len(signedPrefs), - "alreadySubmitted": len(v.submittedPrefSlots), + "alreadySubmitted": v.submittedPrefSlotsCount(), }).Debug("Build proposer preferences result") return signedPrefs } +// processProposerDuties signs proposer preferences for the given duties and +// records the slots submitted, returning the signed preferences and the number +// of signing failures. +func (v *validator) processProposerDuties( + ctx context.Context, + km keymanager.IKeymanager, + duties iter.Seq2[pubkey, *ethpb.ValidatorDuty], + slot primitives.Slot, + dependentRoot []byte, + isNextEpoch bool, +) (signedPrefs []*ethpb.SignedProposerPreferences, sigFailCount int) { + if len(dependentRoot) != fieldparams.RootLength { + return nil, 0 + } + + for pk, duty := range duties { + if len(duty.ProposerSlots) == 0 { + continue + } + if duty.Status != ethpb.ValidatorStatus_ACTIVE && duty.Status != ethpb.ValidatorStatus_EXITING { + continue + } + + feeRecipient, gasLimit := v.proposerConfigForKey(pk) + for _, proposalSlot := range duty.ProposerSlots { + // Skip slots that have passed or are too close. Preferences are + // submitted at mid-slot, so the proposer needs to be at least 1 + // full slot away for the beacon node to receive them in time. + if !isNextEpoch && proposalSlot <= slot+1 { + continue + } + if !v.reservePrefSlot(proposalSlot) { + continue + } + + pref := ðpb.ProposerPreferences{ + DependentRoot: dependentRoot, + ProposalSlot: proposalSlot, + ValidatorIndex: duty.ValidatorIndex, + FeeRecipient: feeRecipient[:], + TargetGasLimit: gasLimit, + } + signedPref, err := v.signProposerPreferences(ctx, km, pk, pref) + if err != nil { + sigFailCount++ + v.releasePrefSlot(proposalSlot) + continue + } + signedPrefs = append(signedPrefs, signedPref) + } + } + return signedPrefs, sigFailCount +} + +// reservePrefSlot marks proposalSlot as submitted, returning false if another +// pass already claimed it. +func (v *validator) reservePrefSlot(proposalSlot primitives.Slot) bool { + v.submittedPrefSlotsLock.Lock() + defer v.submittedPrefSlotsLock.Unlock() + if v.submittedPrefSlots[proposalSlot] { + return false + } + v.submittedPrefSlots[proposalSlot] = true + return true +} + +func (v *validator) releasePrefSlot(proposalSlot primitives.Slot) { + v.submittedPrefSlotsLock.Lock() + defer v.submittedPrefSlotsLock.Unlock() + delete(v.submittedPrefSlots, proposalSlot) +} + +// proposerConfigForKey returns the fee recipient and gas limit for pk, using the +// per-key proposer config when present and otherwise the defaults. +func (v *validator) proposerConfigForKey(pk pubkey) (common.Address, uint64) { + feeRecipient := common.HexToAddress(params.BeaconConfig().EthBurnAddressHex) + gasLimit := params.BeaconConfig().DefaultBuilderGasLimit + ps := v.ProposerSettings() + if ps == nil { + return feeRecipient, gasLimit + } + if ps.DefaultConfig != nil { + if ps.DefaultConfig.FeeRecipientConfig != nil { + feeRecipient = ps.DefaultConfig.FeeRecipientConfig.FeeRecipient + } + if ps.DefaultConfig.BuilderConfig != nil && ps.DefaultConfig.BuilderConfig.Enabled { + gasLimit = uint64(ps.DefaultConfig.BuilderConfig.GasLimit) + } + } + if ps.ProposeConfig != nil { + if config, ok := ps.ProposeConfig[pk]; ok && config != nil { + if config.FeeRecipientConfig != nil { + feeRecipient = config.FeeRecipientConfig.FeeRecipient + } + if config.BuilderConfig != nil && config.BuilderConfig.Enabled { + gasLimit = uint64(config.BuilderConfig.GasLimit) + } + } + } + return feeRecipient, gasLimit +} + +func (v *validator) submittedPrefSlotsCount() int { + v.submittedPrefSlotsLock.RLock() + defer v.submittedPrefSlotsLock.RUnlock() + return len(v.submittedPrefSlots) +} + // submitProposerPreferences builds and submits proposer preferences for the // current slot, bypassing the mid-epoch gate. Called when duties change due to // a reorg so that the new proposer's preferences reach the network promptly. diff --git a/validator/client/validator_test.go b/validator/client/validator_test.go index de857b18c259..f31a35fbca2c 100644 --- a/validator/client/validator_test.go +++ b/validator/client/validator_test.go @@ -370,13 +370,13 @@ func TestRolesAt_OK(t *testing.T) { PtcSlots: []primitives.Slot{1}, }) nextPk := bytesutil.ToBytes48(validatorKey.PublicKey().Marshal()) - v.duties.nextDuties[nextPk] = ðpb.ValidatorDuty{ + v.duties.data.nextDuties[nextPk] = ðpb.ValidatorDuty{ CommitteeIndex: 1, AttesterSlot: 1, PublicKey: validatorKey.PublicKey().Marshal(), IsSyncCommittee: true, } - v.duties.syncNextMap[v.duties.nextDuties[nextPk].ValidatorIndex] = true + v.duties.data.syncNextMap[v.duties.data.nextDuties[nextPk].ValidatorIndex] = true m.validatorClient.EXPECT().DomainData( gomock.Any(), // ctx @@ -407,13 +407,13 @@ func TestRolesAt_OK(t *testing.T) { PublicKey: validatorKey.PublicKey().Marshal(), IsSyncCommittee: false, }) - v.duties.nextDuties[nextPk] = ðpb.ValidatorDuty{ + v.duties.data.nextDuties[nextPk] = ðpb.ValidatorDuty{ CommitteeIndex: 1, AttesterSlot: 1, PublicKey: validatorKey.PublicKey().Marshal(), IsSyncCommittee: true, } - v.duties.syncNextMap[v.duties.nextDuties[nextPk].ValidatorIndex] = true + v.duties.data.syncNextMap[v.duties.data.nextDuties[nextPk].ValidatorIndex] = true m.validatorClient.EXPECT().SyncSubcommitteeIndex( gomock.Any(), // ctx @@ -2189,24 +2189,28 @@ func TestValidator_buildProposerPreferences(t *testing.T) { v.duties = &dutyStore{} v.submittedPrefSlots = make(map[primitives.Slot]bool) - v.duties.SetFromCombinedDutiesResponse(ðpb.ValidatorDutiesContainer{ - CurrentEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{ + CurrentEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + }, }, - }, - NextEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, + NextEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + }, }, - }, - PrevDependentRoot: testProposerPrefDependentRoot, - CurrDependentRoot: testProposerPrefDependentRoot, - }) + PrevDependentRoot: testProposerPrefDependentRoot, + CurrDependentRoot: testProposerPrefDependentRoot, + }) + v.duties.write(data) + } prefs := v.buildProposerPreferences(t.Context(), km, midEpochSlot, false) require.Equal(t, 0, len(prefs)) @@ -2219,25 +2223,29 @@ func TestValidator_buildProposerPreferences(t *testing.T) { v.duties = &dutyStore{} v.submittedPrefSlots = make(map[primitives.Slot]bool) - v.duties.SetFromCombinedDutiesResponse(ðpb.ValidatorDutiesContainer{ - CurrentEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{ + CurrentEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + }, }, - }, - NextEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, - ProposerSlots: []primitives.Slot{nextEpochProposerSlot}, + NextEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + ProposerSlots: []primitives.Slot{nextEpochProposerSlot}, + }, }, - }, - PrevDependentRoot: testProposerPrefDependentRoot, - CurrDependentRoot: testProposerPrefDependentRoot, - }) + PrevDependentRoot: testProposerPrefDependentRoot, + CurrDependentRoot: testProposerPrefDependentRoot, + }) + v.duties.write(data) + } // DomainData is cached after the first call, so subsequent subtests // using the same epoch will hit the cache. Use AnyTimes() here. @@ -2262,25 +2270,29 @@ func TestValidator_buildProposerPreferences(t *testing.T) { v.duties = &dutyStore{} v.submittedPrefSlots = make(map[primitives.Slot]bool) - v.duties.SetFromCombinedDutiesResponse(ðpb.ValidatorDutiesContainer{ - CurrentEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{ + CurrentEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + }, }, - }, - NextEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, - ProposerSlots: []primitives.Slot{nextEpochProposerSlot}, + NextEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + ProposerSlots: []primitives.Slot{nextEpochProposerSlot}, + }, }, - }, - PrevDependentRoot: testProposerPrefDependentRoot, - CurrDependentRoot: testProposerPrefDependentRoot, - }) + PrevDependentRoot: testProposerPrefDependentRoot, + CurrDependentRoot: testProposerPrefDependentRoot, + }) + v.duties.write(data) + } // Slot 0 is start of epoch 0 (before mid-epoch), should not build yet. prefs := v.buildProposerPreferences(t.Context(), km, 0, false) @@ -2294,25 +2306,29 @@ func TestValidator_buildProposerPreferences(t *testing.T) { v.duties = &dutyStore{} v.submittedPrefSlots = make(map[primitives.Slot]bool) - v.duties.SetFromCombinedDutiesResponse(ðpb.ValidatorDutiesContainer{ - CurrentEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{ + CurrentEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + }, }, - }, - NextEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, - ProposerSlots: []primitives.Slot{nextEpochProposerSlot}, + NextEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + ProposerSlots: []primitives.Slot{nextEpochProposerSlot}, + }, }, - }, - PrevDependentRoot: testProposerPrefDependentRoot, - CurrDependentRoot: testProposerPrefDependentRoot, - }) + PrevDependentRoot: testProposerPrefDependentRoot, + CurrDependentRoot: testProposerPrefDependentRoot, + }) + v.duties.write(data) + } midSlot := params.BeaconConfig().SlotsPerEpoch / 2 prefs := v.buildProposerPreferences(t.Context(), km, midSlot, false) @@ -2330,25 +2346,29 @@ func TestValidator_buildProposerPreferences(t *testing.T) { v.duties = &dutyStore{} v.submittedPrefSlots = make(map[primitives.Slot]bool) - v.duties.SetFromCombinedDutiesResponse(ðpb.ValidatorDutiesContainer{ - CurrentEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{ + CurrentEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + }, }, - }, - NextEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, - ProposerSlots: []primitives.Slot{slot1, slot2}, + NextEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + ProposerSlots: []primitives.Slot{slot1, slot2}, + }, }, - }, - PrevDependentRoot: testProposerPrefDependentRoot, - CurrDependentRoot: testProposerPrefDependentRoot, - }) + PrevDependentRoot: testProposerPrefDependentRoot, + CurrDependentRoot: testProposerPrefDependentRoot, + }) + v.duties.write(data) + } // DomainData calls served from cache (populated in prior subtest). prefs := v.buildProposerPreferences(t.Context(), km, midEpochSlot, false) @@ -2367,25 +2387,29 @@ func TestValidator_buildProposerPreferences(t *testing.T) { v.duties = &dutyStore{} v.submittedPrefSlots = make(map[primitives.Slot]bool) - v.duties.SetFromCombinedDutiesResponse(ðpb.ValidatorDutiesContainer{ - CurrentEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_EXITED, + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{ + CurrentEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_EXITED, + }, }, - }, - NextEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_EXITED, - ProposerSlots: []primitives.Slot{nextEpochProposerSlot}, + NextEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_EXITED, + ProposerSlots: []primitives.Slot{nextEpochProposerSlot}, + }, }, - }, - PrevDependentRoot: testProposerPrefDependentRoot, - CurrDependentRoot: testProposerPrefDependentRoot, - }) + PrevDependentRoot: testProposerPrefDependentRoot, + CurrDependentRoot: testProposerPrefDependentRoot, + }) + v.duties.write(data) + } prefs := v.buildProposerPreferences(t.Context(), km, midEpochSlot, false) require.Equal(t, 0, len(prefs)) @@ -2422,25 +2446,29 @@ func TestValidator_buildProposerPreferences(t *testing.T) { v.duties = &dutyStore{} v.submittedPrefSlots = make(map[primitives.Slot]bool) - v.duties.SetFromCombinedDutiesResponse(ðpb.ValidatorDutiesContainer{ - CurrentEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{ + CurrentEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + }, }, - }, - NextEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, - ProposerSlots: []primitives.Slot{nextEpochProposerSlot}, + NextEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + ProposerSlots: []primitives.Slot{nextEpochProposerSlot}, + }, }, - }, - PrevDependentRoot: testProposerPrefDependentRoot, - CurrDependentRoot: testProposerPrefDependentRoot, - }) + PrevDependentRoot: testProposerPrefDependentRoot, + CurrDependentRoot: testProposerPrefDependentRoot, + }) + v.duties.write(data) + } // DomainData calls served from cache (populated in prior subtest). prefs := v.buildProposerPreferences(t.Context(), km, midEpochSlot, false) @@ -2471,25 +2499,29 @@ func TestValidator_buildProposerPreferences(t *testing.T) { v.duties = &dutyStore{} v.submittedPrefSlots = make(map[primitives.Slot]bool) - v.duties.SetFromCombinedDutiesResponse(ðpb.ValidatorDutiesContainer{ - CurrentEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, - ProposerSlots: []primitives.Slot{currentEpochSlot}, + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{ + CurrentEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + ProposerSlots: []primitives.Slot{currentEpochSlot}, + }, }, - }, - NextEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, + NextEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + }, }, - }, - PrevDependentRoot: testProposerPrefDependentRoot, - CurrDependentRoot: testProposerPrefDependentRoot, - }) + PrevDependentRoot: testProposerPrefDependentRoot, + CurrDependentRoot: testProposerPrefDependentRoot, + }) + v.duties.write(data) + } // Slot 1 (past epoch start) allows current-epoch preferences. prefs := v.buildProposerPreferences(t.Context(), km, 1, false) @@ -2508,26 +2540,30 @@ func TestValidator_buildProposerPreferences(t *testing.T) { v.duties = &dutyStore{} v.submittedPrefSlots = make(map[primitives.Slot]bool) - v.duties.SetFromCombinedDutiesResponse(ðpb.ValidatorDutiesContainer{ - CurrentEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, - ProposerSlots: []primitives.Slot{currentEpochSlot}, + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{ + CurrentEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + ProposerSlots: []primitives.Slot{currentEpochSlot}, + }, }, - }, - NextEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, - ProposerSlots: []primitives.Slot{nextEpochSlot}, + NextEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + ProposerSlots: []primitives.Slot{nextEpochSlot}, + }, }, - }, - PrevDependentRoot: testProposerPrefDependentRoot, - CurrDependentRoot: testProposerPrefDependentRoot, - }) + PrevDependentRoot: testProposerPrefDependentRoot, + CurrDependentRoot: testProposerPrefDependentRoot, + }) + v.duties.write(data) + } // At mid-epoch, both current and next epoch preferences are eligible. prefs := v.buildProposerPreferences(t.Context(), km, midEpochSlot, false) @@ -2546,25 +2582,29 @@ func TestValidator_buildProposerPreferences(t *testing.T) { v.duties = &dutyStore{} v.submittedPrefSlots = make(map[primitives.Slot]bool) - v.duties.SetFromCombinedDutiesResponse(ðpb.ValidatorDutiesContainer{ - CurrentEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, - ProposerSlots: []primitives.Slot{3}, + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{ + CurrentEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + ProposerSlots: []primitives.Slot{3}, + }, }, - }, - NextEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, + NextEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + }, }, - }, - PrevDependentRoot: testProposerPrefDependentRoot, - CurrDependentRoot: testProposerPrefDependentRoot, - }) + PrevDependentRoot: testProposerPrefDependentRoot, + CurrDependentRoot: testProposerPrefDependentRoot, + }) + v.duties.write(data) + } // Slot 0 (epoch start) skips current-epoch preferences. prefs := v.buildProposerPreferences(t.Context(), km, 0, false) @@ -2578,25 +2618,29 @@ func TestValidator_buildProposerPreferences(t *testing.T) { v.duties = &dutyStore{} v.submittedPrefSlots = make(map[primitives.Slot]bool) - v.duties.SetFromCombinedDutiesResponse(ðpb.ValidatorDutiesContainer{ - CurrentEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, - ProposerSlots: []primitives.Slot{5}, + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{ + CurrentEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + ProposerSlots: []primitives.Slot{5}, + }, }, - }, - NextEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, + NextEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + }, }, - }, - PrevDependentRoot: testProposerPrefDependentRoot, - CurrDependentRoot: testProposerPrefDependentRoot, - }) + PrevDependentRoot: testProposerPrefDependentRoot, + CurrDependentRoot: testProposerPrefDependentRoot, + }) + v.duties.write(data) + } prefs := v.buildProposerPreferences(t.Context(), km, 1, false) require.Equal(t, 1, len(prefs)) @@ -2616,25 +2660,29 @@ func TestValidator_buildProposerPreferences(t *testing.T) { v.duties = &dutyStore{} v.submittedPrefSlots = make(map[primitives.Slot]bool) - v.duties.SetFromCombinedDutiesResponse(ðpb.ValidatorDutiesContainer{ - CurrentEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, - ProposerSlots: []primitives.Slot{5}, + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{ + CurrentEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + ProposerSlots: []primitives.Slot{5}, + }, }, - }, - NextEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, + NextEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + }, }, - }, - PrevDependentRoot: testProposerPrefDependentRoot, - CurrDependentRoot: testProposerPrefDependentRoot, - }) + PrevDependentRoot: testProposerPrefDependentRoot, + CurrDependentRoot: testProposerPrefDependentRoot, + }) + v.duties.write(data) + } prefs := v.buildProposerPreferences(t.Context(), km, 1, false) require.Equal(t, 1, len(prefs)) @@ -2646,25 +2694,29 @@ func TestValidator_buildProposerPreferences(t *testing.T) { index: 2, } v.duties = &dutyStore{} - v.duties.SetFromCombinedDutiesResponse(ðpb.ValidatorDutiesContainer{ - CurrentEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, - ProposerSlots: []primitives.Slot{5}, - }, - { - PublicKey: kp2.pub[:], - ValidatorIndex: 2, - Status: ethpb.ValidatorStatus_ACTIVE, - ProposerSlots: []primitives.Slot{7}, + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{ + CurrentEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + ProposerSlots: []primitives.Slot{5}, + }, + { + PublicKey: kp2.pub[:], + ValidatorIndex: 2, + Status: ethpb.ValidatorStatus_ACTIVE, + ProposerSlots: []primitives.Slot{7}, + }, }, - }, - NextEpochDuties: []*ethpb.ValidatorDuty{}, - PrevDependentRoot: testProposerPrefDependentRoot, - CurrDependentRoot: testProposerPrefDependentRoot, - }) + NextEpochDuties: []*ethpb.ValidatorDuty{}, + PrevDependentRoot: testProposerPrefDependentRoot, + CurrDependentRoot: testProposerPrefDependentRoot, + }) + v.duties.write(data) + } // Only the new validator's slot is submitted. prefs = v.buildProposerPreferences(t.Context(), km, 2, false) @@ -2682,25 +2734,29 @@ func TestValidator_buildProposerPreferences(t *testing.T) { v.duties = &dutyStore{} v.submittedPrefSlots = make(map[primitives.Slot]bool) - v.duties.SetFromCombinedDutiesResponse(ðpb.ValidatorDutiesContainer{ - CurrentEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{ + CurrentEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + }, }, - }, - NextEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, - ProposerSlots: []primitives.Slot{nextEpochProposerSlot}, + NextEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + ProposerSlots: []primitives.Slot{nextEpochProposerSlot}, + }, }, - }, - PrevDependentRoot: testProposerPrefDependentRoot, - CurrDependentRoot: testProposerPrefDependentRoot, - }) + PrevDependentRoot: testProposerPrefDependentRoot, + CurrDependentRoot: testProposerPrefDependentRoot, + }) + v.duties.write(data) + } // Slot 1 is before mid-epoch, next-epoch prefs should not be sent. prefs := v.buildProposerPreferences(t.Context(), km, 1, false) @@ -2719,19 +2775,23 @@ func TestValidator_buildProposerPreferences(t *testing.T) { v.duties = &dutyStore{} v.submittedPrefSlots = make(map[primitives.Slot]bool) - v.duties.SetFromCombinedDutiesResponse(ðpb.ValidatorDutiesContainer{ - CurrentEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, - ProposerSlots: []primitives.Slot{midEpochSlot + 2}, + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{ + CurrentEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + ProposerSlots: []primitives.Slot{midEpochSlot + 2}, + }, }, - }, - NextEpochDuties: []*ethpb.ValidatorDuty{}, - PrevDependentRoot: testProposerPrefDependentRoot, - CurrDependentRoot: testProposerPrefDependentRoot, - }) + NextEpochDuties: []*ethpb.ValidatorDuty{}, + PrevDependentRoot: testProposerPrefDependentRoot, + CurrDependentRoot: testProposerPrefDependentRoot, + }) + v.duties.write(data) + } // Normal submission at mid-epoch so the slot is in the future. prefs := v.buildProposerPreferences(t.Context(), km, midEpochSlot, false) @@ -2759,19 +2819,23 @@ func TestValidator_buildProposerPreferences(t *testing.T) { v.duties = &dutyStore{} v.submittedPrefSlots = make(map[primitives.Slot]bool) - v.duties.SetFromCombinedDutiesResponse(ðpb.ValidatorDutiesContainer{ - CurrentEpochDuties: []*ethpb.ValidatorDuty{ - { - PublicKey: kp.pub[:], - ValidatorIndex: 1, - Status: ethpb.ValidatorStatus_ACTIVE, - ProposerSlots: []primitives.Slot{5}, + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{ + CurrentEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + ProposerSlots: []primitives.Slot{5}, + }, }, - }, - NextEpochDuties: []*ethpb.ValidatorDuty{}, - PrevDependentRoot: testProposerPrefDependentRoot, - CurrDependentRoot: testProposerPrefDependentRoot, - }) + NextEpochDuties: []*ethpb.ValidatorDuty{}, + PrevDependentRoot: testProposerPrefDependentRoot, + CurrDependentRoot: testProposerPrefDependentRoot, + }) + v.duties.write(data) + } // slot == epochStart (0) with force=false — gate blocks current-epoch duties. prefs := v.buildProposerPreferences(t.Context(), km, 0, false) @@ -2782,6 +2846,66 @@ func TestValidator_buildProposerPreferences(t *testing.T) { require.Equal(t, 1, len(prefs)) require.Equal(t, primitives.Slot(5), prefs[0].Message.ProposalSlot) }) + + t.Run("concurrent builds never double-submit a slot", func(t *testing.T) { + cfg := params.BeaconConfig().Copy() + cfg.GloasForkEpoch = 0 + params.OverrideBeaconConfig(cfg) + + client.EXPECT(). + DomainData(gomock.Any(), gomock.Any()). + Return(ðpb.DomainResponse{SignatureDomain: make([]byte, 32)}, nil). + AnyTimes() + + proposalSlots := []primitives.Slot{ + midEpochSlot + 2, midEpochSlot + 3, midEpochSlot + 4, + midEpochSlot + 5, midEpochSlot + 6, midEpochSlot + 7, + } + v.duties = &dutyStore{} + v.submittedPrefSlots = make(map[primitives.Slot]bool) + { + var data dutyStoreData + data.setFromContainer(ðpb.ValidatorDutiesContainer{ + CurrentEpochDuties: []*ethpb.ValidatorDuty{ + { + PublicKey: kp.pub[:], + ValidatorIndex: 1, + Status: ethpb.ValidatorStatus_ACTIVE, + ProposerSlots: proposalSlots, + }, + }, + NextEpochDuties: []*ethpb.ValidatorDuty{}, + PrevDependentRoot: testProposerPrefDependentRoot, + CurrDependentRoot: testProposerPrefDependentRoot, + }) + v.duties.write(data) + } + + ctx := t.Context() + const builders = 8 + var wg sync.WaitGroup + results := make([][]*ethpb.SignedProposerPreferences, builders) + for i := range builders { + wg.Add(1) + go func(i int) { + defer wg.Done() + results[i] = v.buildProposerPreferences(ctx, km, midEpochSlot, false) + }(i) + } + wg.Wait() + + submitted := make(map[primitives.Slot]int) + for _, prefs := range results { + for _, p := range prefs { + submitted[p.Message.ProposalSlot]++ + } + } + for _, s := range proposalSlots { + require.Equal(t, 1, submitted[s], "slot must be submitted exactly once") + } + require.Equal(t, len(proposalSlots), len(submitted)) + require.Equal(t, len(proposalSlots), v.submittedPrefSlotsCount()) + }) } func TestValidator_buildSignedRegReqs_DefaultConfigDisabled(t *testing.T) {