From f0f12ecff3118a88ec686cd5081660a690ac2b33 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gianguido=20Sor=C3=A0?= Date: Tue, 14 May 2024 11:02:27 +0200 Subject: [PATCH] validator/client: process Sync Committee roles separately In a DV context, to be compatible with the proposed selection endpoint, the VC must push all partial selections to it instead of just one. Process sync committee role separately within the RolesAt method, so that partial selections can be pushed to the DV client appropriately, if configured. --- testing/util/state_test.go | 97 ------------------------------ validator/client/validator.go | 107 ++++++++++++++++++++++------------ 2 files changed, 69 insertions(+), 135 deletions(-) delete mode 100644 testing/util/state_test.go diff --git a/testing/util/state_test.go b/testing/util/state_test.go deleted file mode 100644 index 57362f7072c6..000000000000 --- a/testing/util/state_test.go +++ /dev/null @@ -1,97 +0,0 @@ -package util - -import ( - "context" - "testing" - - ethpb "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1" - "github.com/prysmaticlabs/prysm/v5/testing/assert" - "github.com/prysmaticlabs/prysm/v5/testing/require" -) - -func TestNewBeaconState(t *testing.T) { - st, err := NewBeaconState() - require.NoError(t, err) - b, err := st.MarshalSSZ() - require.NoError(t, err) - got := ðpb.BeaconState{} - require.NoError(t, got.UnmarshalSSZ(b)) - assert.DeepEqual(t, st.ToProtoUnsafe(), got) -} - -func TestNewBeaconStateAltair(t *testing.T) { - st, err := NewBeaconStateAltair() - require.NoError(t, err) - b, err := st.MarshalSSZ() - require.NoError(t, err) - got := ðpb.BeaconStateAltair{} - require.NoError(t, got.UnmarshalSSZ(b)) - assert.DeepEqual(t, st.ToProtoUnsafe(), got) -} - -func TestNewBeaconStateBellatrix(t *testing.T) { - st, err := NewBeaconStateBellatrix() - require.NoError(t, err) - b, err := st.MarshalSSZ() - require.NoError(t, err) - got := ðpb.BeaconStateBellatrix{} - require.NoError(t, got.UnmarshalSSZ(b)) - assert.DeepEqual(t, st.ToProtoUnsafe(), got) -} - -func TestNewBeaconStateCapella(t *testing.T) { - st, err := NewBeaconStateCapella() - require.NoError(t, err) - b, err := st.MarshalSSZ() - require.NoError(t, err) - got := ðpb.BeaconStateCapella{} - require.NoError(t, got.UnmarshalSSZ(b)) - assert.DeepEqual(t, st.ToProtoUnsafe(), got) -} - -func TestNewBeaconStateDeneb(t *testing.T) { - st, err := NewBeaconStateDeneb() - require.NoError(t, err) - b, err := st.MarshalSSZ() - require.NoError(t, err) - got := ðpb.BeaconStateDeneb{} - require.NoError(t, got.UnmarshalSSZ(b)) - assert.DeepEqual(t, st.ToProtoUnsafe(), got) -} - -func TestNewBeaconStateElectra(t *testing.T) { - st, err := NewBeaconStateElectra() - require.NoError(t, err) - b, err := st.MarshalSSZ() - require.NoError(t, err) - got := ðpb.BeaconStateElectra{} - require.NoError(t, got.UnmarshalSSZ(b)) - assert.DeepEqual(t, st.ToProtoUnsafe(), got) -} - -func TestNewBeaconState_HashTreeRoot(t *testing.T) { - st, err := NewBeaconState() - require.NoError(t, err) - _, err = st.HashTreeRoot(context.Background()) - require.NoError(t, err) - st, err = NewBeaconStateAltair() - require.NoError(t, err) - _, err = st.HashTreeRoot(context.Background()) - require.NoError(t, err) - st, err = NewBeaconStateBellatrix() - require.NoError(t, err) - _, err = st.HashTreeRoot(context.Background()) - require.NoError(t, err) - st, err = NewBeaconStateCapella() - require.NoError(t, err) - _, err = st.HashTreeRoot(context.Background()) - require.NoError(t, err) - st, err = NewBeaconStateDeneb() - require.NoError(t, err) - _, err = st.HashTreeRoot(context.Background()) - require.NoError(t, err) - st, err = NewBeaconStateElectra() - require.NoError(t, err) - _, err = st.HashTreeRoot(context.Background()) - require.NoError(t, err) -} diff --git a/validator/client/validator.go b/validator/client/validator.go index 83a0c431c155..c6032bec1cdc 100644 --- a/validator/client/validator.go +++ b/validator/client/validator.go @@ -682,11 +682,19 @@ func (v *validator) subscribeToSubnets(ctx context.Context, res *ethpb.DutiesRes // RolesAt slot returns the validator roles at the given slot. Returns nil if the // validator is known to not have a roles at the slot. Returns UNKNOWN if the -// validator assignments are unknown. Otherwise returns a valid ValidatorRole map. +// validator assignments are unknown. Otherwise, returns a valid ValidatorRole map. func (v *validator) RolesAt(ctx context.Context, slot primitives.Slot) (map[[fieldparams.BLSPubkeyLength]byte][]iface.ValidatorRole, error) { v.dutiesLock.RLock() defer v.dutiesLock.RUnlock() - rolesAt := make(map[[fieldparams.BLSPubkeyLength]byte][]iface.ValidatorRole) + + var ( + rolesAt = make(map[[fieldparams.BLSPubkeyLength]byte][]iface.ValidatorRole) + + // store sync committee duties pubkeys and share indices in slices for + // potential DV processing + syncCommitteeValidators = make(map[primitives.ValidatorIndex][fieldparams.BLSPubkeyLength]byte) + ) + for validator, duty := range v.duties.CurrentEpochDuties { var roles []iface.ValidatorRole @@ -701,6 +709,7 @@ func (v *validator) RolesAt(ctx context.Context, slot primitives.Slot) (map[[fie } } } + if duty.AttesterSlot == slot { roles = append(roles, iface.RoleAttester) @@ -726,19 +735,11 @@ func (v *validator) RolesAt(ctx context.Context, slot primitives.Slot) (map[[fie if duty.IsSyncCommittee { roles = append(roles, iface.RoleSyncCommittee) inSyncCommittee = true - } - } - if inSyncCommittee { - aggregator, err := v.isSyncCommitteeAggregator(ctx, slot, bytesutil.ToBytes48(duty.PublicKey), duty.ValidatorIndex) - if err != nil { - return nil, errors.Wrap(err, "could not check if a validator is a sync committee aggregator") - } - if aggregator { - roles = append(roles, iface.RoleSyncCommitteeAggregator) + syncCommitteeValidators[duty.ValidatorIndex] = bytesutil.ToBytes48(duty.PublicKey) } } - if len(roles) == 0 { + if len(roles) == 0 && !inSyncCommittee { roles = append(roles, iface.RoleUnknown) } @@ -746,6 +747,28 @@ func (v *validator) RolesAt(ctx context.Context, slot primitives.Slot) (map[[fie copy(pubKey[:], duty.PublicKey) rolesAt[pubKey] = roles } + + aggregator, err := v.isSyncCommitteeAggregator( + ctx, + slot, + syncCommitteeValidators, + ) + + if err != nil { + return nil, errors.Wrap(err, "could not check if validators are a sync committee aggregator") + } + + for valIdx, isAgg := range aggregator { + if isAgg { + valPubkey, ok := syncCommitteeValidators[valIdx] + if !ok { + return nil, errors.New("validator is marked as sync committee aggregator but cannot be found in sync committee validator list") + } + + rolesAt[bytesutil.ToBytes48(valPubkey[:])] = append(rolesAt[bytesutil.ToBytes48(valPubkey[:])], iface.RoleSyncCommitteeAggregator) + } + } + return rolesAt, nil } @@ -794,51 +817,59 @@ func (v *validator) isAggregator(ctx context.Context, committee []primitives.Val // // modulo = max(1, SYNC_COMMITTEE_SIZE // SYNC_COMMITTEE_SUBNET_COUNT // TARGET_AGGREGATORS_PER_SYNC_SUBCOMMITTEE) // return bytes_to_uint64(hash(signature)[0:8]) % modulo == 0 -func (v *validator) isSyncCommitteeAggregator(ctx context.Context, slot primitives.Slot, pubKey [fieldparams.BLSPubkeyLength]byte, validatorIndex primitives.ValidatorIndex) (bool, error) { - res, err := v.validatorClient.GetSyncSubcommitteeIndex(ctx, ðpb.SyncSubcommitteeIndexRequest{ - PublicKey: pubKey[:], - Slot: slot, - }) - if err != nil { - return false, err - } +func (v *validator) isSyncCommitteeAggregator(ctx context.Context, slot primitives.Slot, validators map[primitives.ValidatorIndex][fieldparams.BLSPubkeyLength]byte) (map[primitives.ValidatorIndex]bool, error) { + var ( + selections []iface.SyncCommitteeSelection + isAgg = make(map[primitives.ValidatorIndex]bool) + ) + + for valIdx, pubKey := range validators { + res, err := v.validatorClient.GetSyncSubcommitteeIndex(ctx, ðpb.SyncSubcommitteeIndexRequest{ + PublicKey: pubKey[:], + Slot: slot, + }) - var selections []iface.SyncCommitteeSelection - for _, index := range res.Indices { - subCommitteeSize := params.BeaconConfig().SyncCommitteeSize / params.BeaconConfig().SyncCommitteeSubnetCount - subnet := uint64(index) / subCommitteeSize - sig, err := v.signSyncSelectionData(ctx, pubKey, subnet, slot) if err != nil { - return false, err + return nil, errors.Wrap(err, "can't fetch sync subcommittee index") + } + + for _, index := range res.Indices { + subCommitteeSize := params.BeaconConfig().SyncCommitteeSize / params.BeaconConfig().SyncCommitteeSubnetCount + subnet := uint64(index) / subCommitteeSize + sig, err := v.signSyncSelectionData(ctx, pubKey, subnet, slot) + if err != nil { + return nil, errors.Wrap(err, "can't sign selection data") + } + + selections = append(selections, iface.SyncCommitteeSelection{ + SelectionProof: sig, + Slot: slot, + SubcommitteeIndex: primitives.CommitteeIndex(subnet), + ValidatorIndex: valIdx, + }) } - selections = append(selections, iface.SyncCommitteeSelection{ - SelectionProof: sig, - Slot: slot, - SubcommitteeIndex: primitives.CommitteeIndex(subnet), - ValidatorIndex: validatorIndex, - }) } // Override selections with aggregated ones if the node is part of a Distributed Validator. if v.distributed && len(selections) > 0 { + var err error selections, err = v.validatorClient.GetAggregatedSyncSelections(ctx, selections) if err != nil { - return false, errors.Wrap(err, "failed to get aggregated sync selections") + return nil, errors.Wrap(err, "failed to get aggregated sync selections") } } for _, s := range selections { isAggregator, err := altair.IsSyncCommitteeAggregator(s.SelectionProof) if err != nil { - return false, err - } - if isAggregator { - return true, nil + return nil, errors.Wrap(err, "can't detect sync committee aggregator") } + + isAgg[s.ValidatorIndex] = isAggregator } - return false, nil + return isAgg, nil } // UpdateDomainDataCaches by making calls for all of the possible domain data. These can change when