github/prysm
1
1
Fork 0
mirror of https://github.com/OffchainLabs/prysm.git synced 2026-01-09 15:37:56 -05:00
Code Issues Packages Projects Releases Wiki Activity

Substantial VC cleanup (#13593)

Browse Source 
* Cleanup part 1

* Cleanup part 2

* Cleanup part 3

* remove lock field init

* doc for SignerConfig

* remove vars

* use full Keymanager word in function

* revert interface rename

* linter

* fix build issues

* review
This commit is contained in:
Radosław Kapka
2024-05-22 01:39:00 +09:00
committed by GitHub
parent 9befb6bd06
commit 30cc23c5de
49 changed files with 673 additions and 802 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
testing/validator-mock/BUILD.bazel
View File

@@ -5,9 +5,9 @@ package(default_testonly = True)
go_library(
name = "go_default_library",
srcs = [
"beacon_chain_client_mock.go",
"chain_client_mock.go",
"node_client_mock.go",
"prysm_beacon_chain_client_mock.go",
"prysm_chain_client_mock.go",
"validator_client_mock.go",
],
importpath = "github.com/prysmaticlabs/prysm/v5/testing/validator-mock",

58
testing/validator-mock/beacon_chain_client_mock.go → testing/validator-mock/chain_client_mock.go generated
View File

@@ -18,31 +18,31 @@ import (
emptypb "google.golang.org/protobuf/types/known/emptypb"
)
// MockBeaconChainClient is a mock of BeaconChainClient interface.
type MockBeaconChainClient struct {
// MockChainClient is a mock of ChainClient interface.
type MockChainClient struct {
ctrl *gomock.Controller
recorder *MockBeaconChainClientMockRecorder
recorder *MockChainClientMockRecorder
}
// MockBeaconChainClientMockRecorder is the mock recorder for MockBeaconChainClient.
type MockBeaconChainClientMockRecorder struct {
mock *MockBeaconChainClient
// MockChainClientMockRecorder is the mock recorder for MockChainClient.
type MockChainClientMockRecorder struct {
mock *MockChainClient
}
// NewMockBeaconChainClient creates a new mock instance.
func NewMockBeaconChainClient(ctrl *gomock.Controller) *MockBeaconChainClient {
mock := &MockBeaconChainClient{ctrl: ctrl}
mock.recorder = &MockBeaconChainClientMockRecorder{mock}
// NewMockChainClient creates a new mock instance.
func NewMockChainClient(ctrl *gomock.Controller) *MockChainClient {
mock := &MockChainClient{ctrl: ctrl}
mock.recorder = &MockChainClientMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
func (m *MockBeaconChainClient) EXPECT() *MockBeaconChainClientMockRecorder {
func (m *MockChainClient) EXPECT() *MockChainClientMockRecorder {
return m.recorder
}
// GetChainHead mocks base method.
func (m *MockBeaconChainClient) GetChainHead(arg0 context.Context, arg1 *emptypb.Empty) (*eth.ChainHead, error) {
func (m *MockChainClient) GetChainHead(arg0 context.Context, arg1 *emptypb.Empty) (*eth.ChainHead, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetChainHead", arg0, arg1)
ret0, _ := ret[0].(*eth.ChainHead)
@@ -51,13 +51,13 @@ func (m *MockBeaconChainClient) GetChainHead(arg0 context.Context, arg1 *emptypb
}
// GetChainHead indicates an expected call of GetChainHead.
func (mr *MockBeaconChainClientMockRecorder) GetChainHead(arg0, arg1 any) *gomock.Call {
func (mr *MockChainClientMockRecorder) GetChainHead(arg0, arg1 any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetChainHead", reflect.TypeOf((*MockBeaconChainClient)(nil).GetChainHead), arg0, arg1)
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetChainHead", reflect.TypeOf((*MockChainClient)(nil).GetChainHead), arg0, arg1)
}
// GetValidatorParticipation mocks base method.
func (m *MockBeaconChainClient) GetValidatorParticipation(arg0 context.Context, arg1 *eth.GetValidatorParticipationRequest) (*eth.ValidatorParticipationResponse, error) {
func (m *MockChainClient) GetValidatorParticipation(arg0 context.Context, arg1 *eth.GetValidatorParticipationRequest) (*eth.ValidatorParticipationResponse, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetValidatorParticipation", arg0, arg1)
ret0, _ := ret[0].(*eth.ValidatorParticipationResponse)
@@ -66,13 +66,13 @@ func (m *MockBeaconChainClient) GetValidatorParticipation(arg0 context.Context,
}
// GetValidatorParticipation indicates an expected call of GetValidatorParticipation.
func (mr *MockBeaconChainClientMockRecorder) GetValidatorParticipation(arg0, arg1 any) *gomock.Call {
func (mr *MockChainClientMockRecorder) GetValidatorParticipation(arg0, arg1 any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetValidatorParticipation", reflect.TypeOf((*MockBeaconChainClient)(nil).GetValidatorParticipation), arg0, arg1)
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetValidatorParticipation", reflect.TypeOf((*MockChainClient)(nil).GetValidatorParticipation), arg0, arg1)
}
// GetValidatorPerformance mocks base method.
func (m *MockBeaconChainClient) GetValidatorPerformance(arg0 context.Context, arg1 *eth.ValidatorPerformanceRequest) (*eth.ValidatorPerformanceResponse, error) {
func (m *MockChainClient) GetValidatorPerformance(arg0 context.Context, arg1 *eth.ValidatorPerformanceRequest) (*eth.ValidatorPerformanceResponse, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetValidatorPerformance", arg0, arg1)
ret0, _ := ret[0].(*eth.ValidatorPerformanceResponse)
@@ -81,13 +81,13 @@ func (m *MockBeaconChainClient) GetValidatorPerformance(arg0 context.Context, ar
}
// GetValidatorPerformance indicates an expected call of GetValidatorPerformance.
func (mr *MockBeaconChainClientMockRecorder) GetValidatorPerformance(arg0, arg1 any) *gomock.Call {
func (mr *MockChainClientMockRecorder) GetValidatorPerformance(arg0, arg1 any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetValidatorPerformance", reflect.TypeOf((*MockBeaconChainClient)(nil).GetValidatorPerformance), arg0, arg1)
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetValidatorPerformance", reflect.TypeOf((*MockChainClient)(nil).GetValidatorPerformance), arg0, arg1)
}
// GetValidatorQueue mocks base method.
func (m *MockBeaconChainClient) GetValidatorQueue(arg0 context.Context, arg1 *emptypb.Empty) (*eth.ValidatorQueue, error) {
func (m *MockChainClient) GetValidatorQueue(arg0 context.Context, arg1 *emptypb.Empty) (*eth.ValidatorQueue, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetValidatorQueue", arg0, arg1)
ret0, _ := ret[0].(*eth.ValidatorQueue)
@@ -96,13 +96,13 @@ func (m *MockBeaconChainClient) GetValidatorQueue(arg0 context.Context, arg1 *em
}
// GetValidatorQueue indicates an expected call of GetValidatorQueue.
func (mr *MockBeaconChainClientMockRecorder) GetValidatorQueue(arg0, arg1 any) *gomock.Call {
func (mr *MockChainClientMockRecorder) GetValidatorQueue(arg0, arg1 any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetValidatorQueue", reflect.TypeOf((*MockBeaconChainClient)(nil).GetValidatorQueue), arg0, arg1)
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetValidatorQueue", reflect.TypeOf((*MockChainClient)(nil).GetValidatorQueue), arg0, arg1)
}
// ListValidatorBalances mocks base method.
func (m *MockBeaconChainClient) ListValidatorBalances(arg0 context.Context, arg1 *eth.ListValidatorBalancesRequest) (*eth.ValidatorBalances, error) {
func (m *MockChainClient) ListValidatorBalances(arg0 context.Context, arg1 *eth.ListValidatorBalancesRequest) (*eth.ValidatorBalances, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "ListValidatorBalances", arg0, arg1)
ret0, _ := ret[0].(*eth.ValidatorBalances)
@@ -111,13 +111,13 @@ func (m *MockBeaconChainClient) ListValidatorBalances(arg0 context.Context, arg1
}
// ListValidatorBalances indicates an expected call of ListValidatorBalances.
func (mr *MockBeaconChainClientMockRecorder) ListValidatorBalances(arg0, arg1 any) *gomock.Call {
func (mr *MockChainClientMockRecorder) ListValidatorBalances(arg0, arg1 any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListValidatorBalances", reflect.TypeOf((*MockBeaconChainClient)(nil).ListValidatorBalances), arg0, arg1)
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListValidatorBalances", reflect.TypeOf((*MockChainClient)(nil).ListValidatorBalances), arg0, arg1)
}
// ListValidators mocks base method.
func (m *MockBeaconChainClient) ListValidators(arg0 context.Context, arg1 *eth.ListValidatorsRequest) (*eth.Validators, error) {
func (m *MockChainClient) ListValidators(arg0 context.Context, arg1 *eth.ListValidatorsRequest) (*eth.Validators, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "ListValidators", arg0, arg1)
ret0, _ := ret[0].(*eth.Validators)
@@ -126,7 +126,7 @@ func (m *MockBeaconChainClient) ListValidators(arg0 context.Context, arg1 *eth.L
}
// ListValidators indicates an expected call of ListValidators.
func (mr *MockBeaconChainClientMockRecorder) ListValidators(arg0, arg1 any) *gomock.Call {
func (mr *MockChainClientMockRecorder) ListValidators(arg0, arg1 any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListValidators", reflect.TypeOf((*MockBeaconChainClient)(nil).ListValidators), arg0, arg1)
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListValidators", reflect.TypeOf((*MockChainClient)(nil).ListValidators), arg0, arg1)
}

57
testing/validator-mock/prysm_beacon_chain_client_mock.go generated
View File

@@ -1,57 +0,0 @@
// Code generated by MockGen. DO NOT EDIT.
// Source: github.com/prysmaticlabs/prysm/v5/validator/client/iface (interfaces: PrysmBeaconChainClient)
//
// Generated by this command:
//
// mockgen -package=validator_mock -destination=testing/validator-mock/prysm_beacon_chain_client_mock.go github.com/prysmaticlabs/prysm/v5/validator/client/iface PrysmBeaconChainClient
//
// Package validator_mock is a generated GoMock package.
package validator_mock
import (
context "context"
reflect "reflect"
validator "github.com/prysmaticlabs/prysm/v5/consensus-types/validator"
iface "github.com/prysmaticlabs/prysm/v5/validator/client/iface"
gomock "go.uber.org/mock/gomock"
)
// MockPrysmBeaconChainClient is a mock of PrysmBeaconChainClient interface.
type MockPrysmBeaconChainClient struct {
ctrl *gomock.Controller
recorder *MockPrysmBeaconChainClientMockRecorder
}
// MockPrysmBeaconChainClientMockRecorder is the mock recorder for MockPrysmBeaconChainClient.
type MockPrysmBeaconChainClientMockRecorder struct {
mock *MockPrysmBeaconChainClient
}
// NewMockPrysmBeaconChainClient creates a new mock instance.
func NewMockPrysmBeaconChainClient(ctrl *gomock.Controller) *MockPrysmBeaconChainClient {
mock := &MockPrysmBeaconChainClient{ctrl: ctrl}
mock.recorder = &MockPrysmBeaconChainClientMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
func (m *MockPrysmBeaconChainClient) EXPECT() *MockPrysmBeaconChainClientMockRecorder {
return m.recorder
}
// GetValidatorCount mocks base method.
func (m *MockPrysmBeaconChainClient) GetValidatorCount(arg0 context.Context, arg1 string, arg2 []validator.Status) ([]iface.ValidatorCount, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetValidatorCount", arg0, arg1, arg2)
ret0, _ := ret[0].([]iface.ValidatorCount)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// GetValidatorCount indicates an expected call of GetValidatorCount.
func (mr *MockPrysmBeaconChainClientMockRecorder) GetValidatorCount(arg0, arg1, arg2 any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetValidatorCount", reflect.TypeOf((*MockPrysmBeaconChainClient)(nil).GetValidatorCount), arg0, arg1, arg2)
}

52
testing/validator-mock/prysm_chain_client_mock.go generated Normal file
View File

@@ -0,0 +1,52 @@
// Code generated by MockGen. DO NOT EDIT.
// Source: github.com/prysmaticlabs/prysm/v4/validator/client/iface (interfaces: PrysmChainClient)
// Package validator_mock is a generated GoMock package.
package validator_mock
import (
context "context"
reflect "reflect"
validator "github.com/prysmaticlabs/prysm/v5/consensus-types/validator"
iface "github.com/prysmaticlabs/prysm/v5/validator/client/iface"
gomock "go.uber.org/mock/gomock"
)
// MockPrysmChainClient is a mock of PrysmChainClient interface.
type MockPrysmChainClient struct {
ctrl *gomock.Controller
recorder *MockPrysmChainClientMockRecorder
}
// MockPrysmChainClientMockRecorder is the mock recorder for MockPrysmChainClient.
type MockPrysmChainClientMockRecorder struct {
mock *MockPrysmChainClient
}
// NewMockPrysmChainClient creates a new mock instance.
func NewMockPrysmChainClient(ctrl *gomock.Controller) *MockPrysmChainClient {
mock := &MockPrysmChainClient{ctrl: ctrl}
mock.recorder = &MockPrysmChainClientMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
func (m *MockPrysmChainClient) EXPECT() *MockPrysmChainClientMockRecorder {
return m.recorder
}
// GetValidatorCount mocks base method.
func (m *MockPrysmChainClient) GetValidatorCount(arg0 context.Context, arg1 string, arg2 []validator.Status) ([]iface.ValidatorCount, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetValidatorCount", arg0, arg1, arg2)
ret0, _ := ret[0].([]iface.ValidatorCount)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// GetValidatorCount indicates an expected call of GetValidatorCount.
func (mr *MockPrysmChainClientMockRecorder) GetValidatorCount(arg0, arg1, arg2 interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetValidatorCount", reflect.TypeOf((*MockPrysmChainClient)(nil).GetValidatorCount), arg0, arg1, arg2)
}

6
validator/client/aggregate.go
View File

@@ -44,7 +44,7 @@ func (v *validator) SubmitAggregateAndProof(ctx context.Context, slot primitives
}
// Avoid sending beacon node duplicated aggregation requests.
k := validatorSubscribeKey(slot, duty.CommitteeIndex)
k := validatorSubnetSubscriptionKey(slot, duty.CommitteeIndex)
v.aggregatedSlotCommitteeIDCacheLock.Lock()
if v.aggregatedSlotCommitteeIDCache.Contains(k) {
v.aggregatedSlotCommitteeIDCacheLock.Unlock()
@@ -149,7 +149,7 @@ func (v *validator) signSlotWithSelectionProof(ctx context.Context, pubKey [fiel
if err != nil {
return nil, err
}
sig, err = v.keyManager.Sign(ctx, &validatorpb.SignRequest{
sig, err = v.km.Sign(ctx, &validatorpb.SignRequest{
PublicKey: pubKey[:],
SigningRoot: root[:],
SignatureDomain: domain.SignatureDomain,
@@ -203,7 +203,7 @@ func (v *validator) aggregateAndProofSig(ctx context.Context, pubKey [fieldparam
if err != nil {
return nil, err
}
sig, err = v.keyManager.Sign(ctx, &validatorpb.SignRequest{
sig, err = v.km.Sign(ctx, &validatorpb.SignRequest{
PublicKey: pubKey[:],
SigningRoot: root[:],
SignatureDomain: d.SignatureDomain,

2
validator/client/attest.go
View File

@@ -203,7 +203,7 @@ func (v *validator) signAtt(ctx context.Context, pubKey [fieldparams.BLSPubkeyLe
if err != nil {
return nil, [32]byte{}, err
}
sig, err := v.keyManager.Sign(ctx, &validatorpb.SignRequest{
sig, err := v.km.Sign(ctx, &validatorpb.SignRequest{
PublicKey: pubKey[:],
SigningRoot: root[:],
SignatureDomain: domain.SignatureDomain,

4
validator/client/attest_test.go
View File

@@ -167,7 +167,7 @@ func TestAttestToBlockHead_AttestsCorrectly(t *testing.T) {
root, err := signing.ComputeSigningRoot(expectedAttestation.Data, make([]byte, 32))
require.NoError(t, err)
sig, err := validator.keyManager.Sign(context.Background(), &validatorpb.SignRequest{
sig, err := validator.km.Sign(context.Background(), &validatorpb.SignRequest{
PublicKey: validatorKey.PublicKey().Marshal(),
SigningRoot: root[:],
})
@@ -504,7 +504,7 @@ func TestSignAttestation(t *testing.T) {
att.Data.BeaconBlockRoot = bytesutil.PadTo([]byte("blockRoot"), 32)
pk := testKeyFromBytes(t, []byte{1})
validator.keyManager = newMockKeymanager(t, pk)
validator.km = newMockKeymanager(t, pk)
sig, sr, err := validator.signAtt(ctx, pk.pub, att.Data, att.Data.Slot)
require.NoError(t, err, "%x,%x,%v", sig, sr, err)
require.Equal(t, "b6a60f8497bd328908be83634d045"+

28
validator/client/beacon-api/beacon_api_beacon_chain_client.go
View File

@@ -17,15 +17,15 @@ import (
"github.com/prysmaticlabs/prysm/v5/validator/client/iface"
)
type beaconApiBeaconChainClient struct {
fallbackClient iface.BeaconChainClient
type beaconApiChainClient struct {
fallbackClient iface.ChainClient
jsonRestHandler JsonRestHandler
stateValidatorsProvider StateValidatorsProvider
}
const getValidatorPerformanceEndpoint = "/prysm/validators/performance"
func (c beaconApiBeaconChainClient) getHeadBlockHeaders(ctx context.Context) (*structs.GetBlockHeaderResponse, error) {
func (c beaconApiChainClient) getHeadBlockHeaders(ctx context.Context) (*structs.GetBlockHeaderResponse, error) {
blockHeader := structs.GetBlockHeaderResponse{}
err := c.jsonRestHandler.Get(ctx, "/eth/v1/beacon/headers/head", &blockHeader)
if err != nil {
@@ -43,7 +43,7 @@ func (c beaconApiBeaconChainClient) getHeadBlockHeaders(ctx context.Context) (*s
return &blockHeader, nil
}
func (c beaconApiBeaconChainClient) GetChainHead(ctx context.Context, _ *empty.Empty) (*ethpb.ChainHead, error) {
func (c beaconApiChainClient) GetChainHead(ctx context.Context, _ *empty.Empty) (*ethpb.ChainHead, error) {
const endpoint = "/eth/v1/beacon/states/head/finality_checkpoints"
finalityCheckpoints := structs.GetFinalityCheckpointsResponse{}
@@ -146,16 +146,16 @@ func (c beaconApiBeaconChainClient) GetChainHead(ctx context.Context, _ *empty.E
}, nil
}
func (c beaconApiBeaconChainClient) ListValidatorBalances(ctx context.Context, in *ethpb.ListValidatorBalancesRequest) (*ethpb.ValidatorBalances, error) {
func (c beaconApiChainClient) ListValidatorBalances(ctx context.Context, in *ethpb.ListValidatorBalancesRequest) (*ethpb.ValidatorBalances, error) {
if c.fallbackClient != nil {
return c.fallbackClient.ListValidatorBalances(ctx, in)
}
// TODO: Implement me
panic("beaconApiBeaconChainClient.ListValidatorBalances is not implemented. To use a fallback client, pass a fallback client as the last argument of NewBeaconApiBeaconChainClientWithFallback.")
panic("beaconApiChainClient.ListValidatorBalances is not implemented. To use a fallback client, pass a fallback client as the last argument of NewBeaconApiChainClientWithFallback.")
}
func (c beaconApiBeaconChainClient) ListValidators(ctx context.Context, in *ethpb.ListValidatorsRequest) (*ethpb.Validators, error) {
func (c beaconApiChainClient) ListValidators(ctx context.Context, in *ethpb.ListValidatorsRequest) (*ethpb.Validators, error) {
pageSize := in.PageSize
// We follow the gRPC behavior here, which returns a maximum of 250 results when pageSize == 0
@@ -310,16 +310,16 @@ func (c beaconApiBeaconChainClient) ListValidators(ctx context.Context, in *ethp
}, nil
}
func (c beaconApiBeaconChainClient) GetValidatorQueue(ctx context.Context, in *empty.Empty) (*ethpb.ValidatorQueue, error) {
func (c beaconApiChainClient) GetValidatorQueue(ctx context.Context, in *empty.Empty) (*ethpb.ValidatorQueue, error) {
if c.fallbackClient != nil {
return c.fallbackClient.GetValidatorQueue(ctx, in)
}
// TODO: Implement me
panic("beaconApiBeaconChainClient.GetValidatorQueue is not implemented. To use a fallback client, pass a fallback client as the last argument of NewBeaconApiBeaconChainClientWithFallback.")
panic("beaconApiChainClient.GetValidatorQueue is not implemented. To use a fallback client, pass a fallback client as the last argument of NewBeaconApiChainClientWithFallback.")
}
func (c beaconApiBeaconChainClient) GetValidatorPerformance(ctx context.Context, in *ethpb.ValidatorPerformanceRequest) (*ethpb.ValidatorPerformanceResponse, error) {
func (c beaconApiChainClient) GetValidatorPerformance(ctx context.Context, in *ethpb.ValidatorPerformanceRequest) (*ethpb.ValidatorPerformanceResponse, error) {
request, err := json.Marshal(structs.GetValidatorPerformanceRequest{
PublicKeys: in.PublicKeys,
Indices: in.Indices,
@@ -345,17 +345,17 @@ func (c beaconApiBeaconChainClient) GetValidatorPerformance(ctx context.Context,
}, nil
}
func (c beaconApiBeaconChainClient) GetValidatorParticipation(ctx context.Context, in *ethpb.GetValidatorParticipationRequest) (*ethpb.ValidatorParticipationResponse, error) {
func (c beaconApiChainClient) GetValidatorParticipation(ctx context.Context, in *ethpb.GetValidatorParticipationRequest) (*ethpb.ValidatorParticipationResponse, error) {
if c.fallbackClient != nil {
return c.fallbackClient.GetValidatorParticipation(ctx, in)
}
// TODO: Implement me
panic("beaconApiBeaconChainClient.GetValidatorParticipation is not implemented. To use a fallback client, pass a fallback client as the last argument of NewBeaconApiBeaconChainClientWithFallback.")
panic("beaconApiChainClient.GetValidatorParticipation is not implemented. To use a fallback client, pass a fallback client as the last argument of NewBeaconApiChainClientWithFallback.")
}
func NewBeaconApiBeaconChainClientWithFallback(jsonRestHandler JsonRestHandler, fallbackClient iface.BeaconChainClient) iface.BeaconChainClient {
return &beaconApiBeaconChainClient{
func NewBeaconApiChainClientWithFallback(jsonRestHandler JsonRestHandler, fallbackClient iface.ChainClient) iface.ChainClient {
return &beaconApiChainClient{
jsonRestHandler: jsonRestHandler,
fallbackClient: fallbackClient,
stateValidatorsProvider: beaconApiStateValidatorsProvider{jsonRestHandler: jsonRestHandler},

26
validator/client/beacon-api/beacon_api_beacon_chain_client_test.go
View File

@@ -31,7 +31,7 @@ func TestListValidators(t *testing.T) {
defer ctrl.Finish()
ctx := context.Background()
beaconChainClient := beaconApiBeaconChainClient{}
beaconChainClient := beaconApiChainClient{}
_, err := beaconChainClient.ListValidators(ctx, &ethpb.ListValidatorsRequest{
PageToken: "foo",
})
@@ -43,7 +43,7 @@ func TestListValidators(t *testing.T) {
defer ctrl.Finish()
ctx := context.Background()
beaconChainClient := beaconApiBeaconChainClient{}
beaconChainClient := beaconApiChainClient{}
_, err := beaconChainClient.ListValidators(ctx, &ethpb.ListValidatorsRequest{
QueryFilter: &ethpb.ListValidatorsRequest_Epoch{
Epoch: math.MaxUint64,
@@ -63,7 +63,7 @@ func TestListValidators(t *testing.T) {
errors.New("foo error"),
)
beaconChainClient := beaconApiBeaconChainClient{stateValidatorsProvider: stateValidatorsProvider}
beaconChainClient := beaconApiChainClient{stateValidatorsProvider: stateValidatorsProvider}
_, err := beaconChainClient.ListValidators(ctx, &ethpb.ListValidatorsRequest{
QueryFilter: &ethpb.ListValidatorsRequest_Epoch{
Epoch: 0,
@@ -83,7 +83,7 @@ func TestListValidators(t *testing.T) {
errors.New("bar error"),
)
beaconChainClient := beaconApiBeaconChainClient{stateValidatorsProvider: stateValidatorsProvider}
beaconChainClient := beaconApiChainClient{stateValidatorsProvider: stateValidatorsProvider}
_, err := beaconChainClient.ListValidators(ctx, &ethpb.ListValidatorsRequest{
QueryFilter: &ethpb.ListValidatorsRequest_Genesis{},
})
@@ -101,7 +101,7 @@ func TestListValidators(t *testing.T) {
errors.New("foo error"),
)
beaconChainClient := beaconApiBeaconChainClient{stateValidatorsProvider: stateValidatorsProvider}
beaconChainClient := beaconApiChainClient{stateValidatorsProvider: stateValidatorsProvider}
_, err := beaconChainClient.ListValidators(ctx, &ethpb.ListValidatorsRequest{
QueryFilter: nil,
})
@@ -122,7 +122,7 @@ func TestListValidators(t *testing.T) {
jsonRestHandler := mock.NewMockJsonRestHandler(ctrl)
jsonRestHandler.EXPECT().Get(ctx, blockHeaderEndpoint, gomock.Any()).Return(errors.New("bar error"))
beaconChainClient := beaconApiBeaconChainClient{
beaconChainClient := beaconApiChainClient{
stateValidatorsProvider: stateValidatorsProvider,
jsonRestHandler: jsonRestHandler,
}
@@ -200,7 +200,7 @@ func TestListValidators(t *testing.T) {
testCase.blockHeaderResponse,
)
beaconChainClient := beaconApiBeaconChainClient{
beaconChainClient := beaconApiChainClient{
stateValidatorsProvider: stateValidatorsProvider,
jsonRestHandler: jsonRestHandler,
}
@@ -333,7 +333,7 @@ func TestListValidators(t *testing.T) {
nil,
)
beaconChainClient := beaconApiBeaconChainClient{stateValidatorsProvider: stateValidatorsProvider}
beaconChainClient := beaconApiChainClient{stateValidatorsProvider: stateValidatorsProvider}
_, err := beaconChainClient.ListValidators(ctx, &ethpb.ListValidatorsRequest{
QueryFilter: &ethpb.ListValidatorsRequest_Genesis{},
})
@@ -561,7 +561,7 @@ func TestListValidators(t *testing.T) {
nil,
)
beaconChainClient := beaconApiBeaconChainClient{stateValidatorsProvider: stateValidatorsProvider}
beaconChainClient := beaconApiChainClient{stateValidatorsProvider: stateValidatorsProvider}
validators, err := beaconChainClient.ListValidators(ctx, &ethpb.ListValidatorsRequest{
QueryFilter: &ethpb.ListValidatorsRequest_Genesis{},
PublicKeys: [][]byte{},
@@ -752,7 +752,7 @@ func TestGetChainHead(t *testing.T) {
testCase.generateFinalityCheckpointsResponse(),
)
beaconChainClient := beaconApiBeaconChainClient{jsonRestHandler: jsonRestHandler}
beaconChainClient := beaconApiChainClient{jsonRestHandler: jsonRestHandler}
_, err := beaconChainClient.GetChainHead(ctx, &emptypb.Empty{})
assert.ErrorContains(t, testCase.expectedError, err)
})
@@ -859,7 +859,7 @@ func TestGetChainHead(t *testing.T) {
testCase.generateHeadBlockHeadersResponse(),
)
beaconChainClient := beaconApiBeaconChainClient{jsonRestHandler: jsonRestHandler}
beaconChainClient := beaconApiChainClient{jsonRestHandler: jsonRestHandler}
_, err := beaconChainClient.GetChainHead(ctx, &emptypb.Empty{})
assert.ErrorContains(t, testCase.expectedError, err)
})
@@ -913,7 +913,7 @@ func TestGetChainHead(t *testing.T) {
HeadEpoch: slots.ToEpoch(8),
}
beaconChainClient := beaconApiBeaconChainClient{jsonRestHandler: jsonRestHandler}
beaconChainClient := beaconApiChainClient{jsonRestHandler: jsonRestHandler}
chainHead, err := beaconChainClient.GetChainHead(ctx, &emptypb.Empty{})
require.NoError(t, err)
assert.DeepEqual(t, expectedChainHead, chainHead)
@@ -949,7 +949,7 @@ func Test_beaconApiBeaconChainClient_GetValidatorPerformance(t *testing.T) {
nil,
)
c := beaconApiBeaconChainClient{
c := beaconApiChainClient{
jsonRestHandler: jsonRestHandler,
}

4
validator/client/beacon-api/beacon_api_validator_client.go
View File

@@ -23,7 +23,7 @@ type beaconApiValidatorClient struct {
stateValidatorsProvider StateValidatorsProvider
jsonRestHandler JsonRestHandler
beaconBlockConverter BeaconBlockConverter
prysmBeaconChainCLient iface.PrysmBeaconChainClient
prysmChainClient iface.PrysmChainClient
isEventStreamRunning bool
}
@@ -34,7 +34,7 @@ func NewBeaconApiValidatorClient(jsonRestHandler JsonRestHandler, opts ...Valida
stateValidatorsProvider: beaconApiStateValidatorsProvider{jsonRestHandler: jsonRestHandler},
jsonRestHandler: jsonRestHandler,
beaconBlockConverter: beaconApiBeaconBlockConverter{},
prysmBeaconChainCLient: prysmBeaconChainClient{
prysmChainClient: prysmChainClient{
nodeClient: &beaconApiNodeClient{jsonRestHandler: jsonRestHandler},
jsonRestHandler: jsonRestHandler,
},

10
validator/client/beacon-api/prysm_beacon_chain_client.go
View File

@@ -13,20 +13,20 @@ import (
"github.com/prysmaticlabs/prysm/v5/validator/client/iface"
)
// NewPrysmBeaconChainClient returns implementation of iface.PrysmBeaconChainClient.
func NewPrysmBeaconChainClient(jsonRestHandler JsonRestHandler, nodeClient iface.NodeClient) iface.PrysmBeaconChainClient {
return prysmBeaconChainClient{
// NewPrysmChainClient returns implementation of iface.PrysmChainClient.
func NewPrysmChainClient(jsonRestHandler JsonRestHandler, nodeClient iface.NodeClient) iface.PrysmChainClient {
return prysmChainClient{
jsonRestHandler: jsonRestHandler,
nodeClient: nodeClient,
}
}
type prysmBeaconChainClient struct {
type prysmChainClient struct {
jsonRestHandler JsonRestHandler
nodeClient iface.NodeClient
}
func (c prysmBeaconChainClient) GetValidatorCount(ctx context.Context, stateID string, statuses []validator2.Status) ([]iface.ValidatorCount, error) {
func (c prysmChainClient) GetValidatorCount(ctx context.Context, stateID string, statuses []validator2.Status) ([]iface.ValidatorCount, error) {
// Check node version for prysm beacon node as it is a custom endpoint for prysm beacon node.
nodeVersion, err := c.nodeClient.GetVersion(ctx, nil)
if err != nil {

2
validator/client/beacon-api/status.go
View File

@@ -79,7 +79,7 @@ func (c *beaconApiValidatorClient) getValidatorsStatusResponse(ctx context.Conte
return nil, nil, nil, errors.Wrap(err, "failed to get state validators")
}
validatorsCountResponse, err := c.prysmBeaconChainCLient.GetValidatorCount(ctx, "head", nil)
validatorsCountResponse, err := c.prysmChainClient.GetValidatorCount(ctx, "head", nil)
if err != nil && !errors.Is(err, iface.ErrNotSupported) {
return nil, nil, nil, errors.Wrap(err, "failed to get total validator count")
}

10
validator/client/beacon-api/status_test.go
View File

@@ -55,7 +55,7 @@ func TestValidatorStatus_Nominal(t *testing.T) {
jsonRestHandler := mock.NewMockJsonRestHandler(ctrl)
validatorClient := beaconApiValidatorClient{
stateValidatorsProvider: stateValidatorsProvider,
prysmBeaconChainCLient: prysmBeaconChainClient{
prysmChainClient: prysmChainClient{
nodeClient: &beaconApiNodeClient{
jsonRestHandler: jsonRestHandler,
},
@@ -181,7 +181,7 @@ func TestMultipleValidatorStatus_Nominal(t *testing.T) {
validatorClient := beaconApiValidatorClient{
stateValidatorsProvider: stateValidatorsProvider,
prysmBeaconChainCLient: prysmBeaconChainClient{
prysmChainClient: prysmChainClient{
nodeClient: &beaconApiNodeClient{
jsonRestHandler: jsonRestHandler,
},
@@ -429,7 +429,7 @@ func TestGetValidatorsStatusResponse_Nominal_SomeActiveValidators(t *testing.T)
validatorClient := beaconApiValidatorClient{
stateValidatorsProvider: stateValidatorsProvider,
prysmBeaconChainCLient: prysmBeaconChainClient{
prysmChainClient: prysmChainClient{
nodeClient: &beaconApiNodeClient{
jsonRestHandler: jsonRestHandler,
},
@@ -499,7 +499,7 @@ func TestGetValidatorsStatusResponse_Nominal_NoActiveValidators(t *testing.T) {
validatorClient := beaconApiValidatorClient{
stateValidatorsProvider: stateValidatorsProvider,
prysmBeaconChainCLient: prysmBeaconChainClient{
prysmChainClient: prysmChainClient{
nodeClient: &beaconApiNodeClient{
jsonRestHandler: jsonRestHandler,
},
@@ -729,7 +729,7 @@ func TestValidatorStatusResponse_InvalidData(t *testing.T) {
validatorClient := beaconApiValidatorClient{
stateValidatorsProvider: stateValidatorsProvider,
prysmBeaconChainCLient: prysmBeaconChainClient{
prysmChainClient: prysmChainClient{
nodeClient: &beaconApiNodeClient{
jsonRestHandler: jsonRestHandler,
},

2
validator/client/beacon-api/validator_count_test.go
View File

@@ -141,7 +141,7 @@ func TestGetValidatorCount(t *testing.T) {
).Times(test.validatorCountCalled)
// Type assertion.
var client iface.PrysmBeaconChainClient = &prysmBeaconChainClient{
var client iface.PrysmChainClient = &prysmChainClient{
nodeClient: &beaconApiNodeClient{jsonRestHandler: jsonRestHandler},
jsonRestHandler: jsonRestHandler,
}

12
validator/client/beacon-chain-client-factory/beacon_chain_client_factory.go
View File

@@ -9,19 +9,19 @@ import (
validatorHelpers "github.com/prysmaticlabs/prysm/v5/validator/helpers"
)
func NewBeaconChainClient(validatorConn validatorHelpers.NodeConnection, jsonRestHandler beaconApi.JsonRestHandler) iface.BeaconChainClient {
grpcClient := grpcApi.NewGrpcBeaconChainClient(validatorConn.GetGrpcClientConn())
func NewChainClient(validatorConn validatorHelpers.NodeConnection, jsonRestHandler beaconApi.JsonRestHandler) iface.ChainClient {
grpcClient := grpcApi.NewGrpcChainClient(validatorConn.GetGrpcClientConn())
if features.Get().EnableBeaconRESTApi {
return beaconApi.NewBeaconApiBeaconChainClientWithFallback(jsonRestHandler, grpcClient)
return beaconApi.NewBeaconApiChainClientWithFallback(jsonRestHandler, grpcClient)
} else {
return grpcClient
}
}
func NewPrysmBeaconClient(validatorConn validatorHelpers.NodeConnection, jsonRestHandler beaconApi.JsonRestHandler) iface.PrysmBeaconChainClient {
func NewPrysmChainClient(validatorConn validatorHelpers.NodeConnection, jsonRestHandler beaconApi.JsonRestHandler) iface.PrysmChainClient {
if features.Get().EnableBeaconRESTApi {
return beaconApi.NewPrysmBeaconChainClient(jsonRestHandler, nodeClientFactory.NewNodeClient(validatorConn, jsonRestHandler))
return beaconApi.NewPrysmChainClient(jsonRestHandler, nodeClientFactory.NewNodeClient(validatorConn, jsonRestHandler))
} else {
return grpcApi.NewGrpcPrysmBeaconChainClient(validatorConn.GetGrpcClientConn())
return grpcApi.NewGrpcPrysmChainClient(validatorConn.GetGrpcClientConn())
}
}

18
validator/client/grpc-api/grpc_beacon_chain_client.go
View File

@@ -9,34 +9,34 @@ import (
"google.golang.org/grpc"
)
type grpcBeaconChainClient struct {
type grpcChainClient struct {
beaconChainClient ethpb.BeaconChainClient
}
func (c *grpcBeaconChainClient) GetChainHead(ctx context.Context, in *empty.Empty) (*ethpb.ChainHead, error) {
func (c *grpcChainClient) GetChainHead(ctx context.Context, in *empty.Empty) (*ethpb.ChainHead, error) {
return c.beaconChainClient.GetChainHead(ctx, in)
}
func (c *grpcBeaconChainClient) ListValidatorBalances(ctx context.Context, in *ethpb.ListValidatorBalancesRequest) (*ethpb.ValidatorBalances, error) {
func (c *grpcChainClient) ListValidatorBalances(ctx context.Context, in *ethpb.ListValidatorBalancesRequest) (*ethpb.ValidatorBalances, error) {
return c.beaconChainClient.ListValidatorBalances(ctx, in)
}
func (c *grpcBeaconChainClient) ListValidators(ctx context.Context, in *ethpb.ListValidatorsRequest) (*ethpb.Validators, error) {
func (c *grpcChainClient) ListValidators(ctx context.Context, in *ethpb.ListValidatorsRequest) (*ethpb.Validators, error) {
return c.beaconChainClient.ListValidators(ctx, in)
}
func (c *grpcBeaconChainClient) GetValidatorQueue(ctx context.Context, in *empty.Empty) (*ethpb.ValidatorQueue, error) {
func (c *grpcChainClient) GetValidatorQueue(ctx context.Context, in *empty.Empty) (*ethpb.ValidatorQueue, error) {
return c.beaconChainClient.GetValidatorQueue(ctx, in)
}
func (c *grpcBeaconChainClient) GetValidatorPerformance(ctx context.Context, in *ethpb.ValidatorPerformanceRequest) (*ethpb.ValidatorPerformanceResponse, error) {
func (c *grpcChainClient) GetValidatorPerformance(ctx context.Context, in *ethpb.ValidatorPerformanceRequest) (*ethpb.ValidatorPerformanceResponse, error) {
return c.beaconChainClient.GetValidatorPerformance(ctx, in)
}
func (c *grpcBeaconChainClient) GetValidatorParticipation(ctx context.Context, in *ethpb.GetValidatorParticipationRequest) (*ethpb.ValidatorParticipationResponse, error) {
func (c *grpcChainClient) GetValidatorParticipation(ctx context.Context, in *ethpb.GetValidatorParticipationRequest) (*ethpb.ValidatorParticipationResponse, error) {
return c.beaconChainClient.GetValidatorParticipation(ctx, in)
}
func NewGrpcBeaconChainClient(cc grpc.ClientConnInterface) iface.BeaconChainClient {
return &grpcBeaconChainClient{ethpb.NewBeaconChainClient(cc)}
func NewGrpcChainClient(cc grpc.ClientConnInterface) iface.ChainClient {
return &grpcChainClient{ethpb.NewBeaconChainClient(cc)}
}

14
validator/client/grpc-api/grpc_prysm_beacon_chain_client.go
View File

@@ -18,12 +18,12 @@ import (
"google.golang.org/grpc"
)
type grpcPrysmBeaconChainClient struct {
beaconChainClient iface.BeaconChainClient
type grpcPrysmChainClient struct {
chainClient iface.ChainClient
}
func (g grpcPrysmBeaconChainClient) GetValidatorCount(ctx context.Context, _ string, statuses []validator.Status) ([]iface.ValidatorCount, error) {
resp, err := g.beaconChainClient.ListValidators(ctx, &ethpb.ListValidatorsRequest{PageSize: 0})
func (g grpcPrysmChainClient) GetValidatorCount(ctx context.Context, _ string, statuses []validator.Status) ([]iface.ValidatorCount, error) {
resp, err := g.chainClient.ListValidators(ctx, &ethpb.ListValidatorsRequest{PageSize: 0})
if err != nil {
return nil, errors.Wrap(err, "list validators failed")
}
@@ -33,7 +33,7 @@ func (g grpcPrysmBeaconChainClient) GetValidatorCount(ctx context.Context, _ str
vals = append(vals, val.Validator)
}
head, err := g.beaconChainClient.GetChainHead(ctx, &empty.Empty{})
head, err := g.chainClient.GetChainHead(ctx, &empty.Empty{})
if err != nil {
return nil, errors.Wrap(err, "get chain head")
}
@@ -92,6 +92,6 @@ func validatorCountByStatus(validators []*ethpb.Validator, statuses []validator.
return resp, nil
}
func NewGrpcPrysmBeaconChainClient(cc grpc.ClientConnInterface) iface.PrysmBeaconChainClient {
return &grpcPrysmBeaconChainClient{beaconChainClient: &grpcBeaconChainClient{ethpb.NewBeaconChainClient(cc)}}
func NewGrpcPrysmChainClient(cc grpc.ClientConnInterface) iface.PrysmChainClient {
return &grpcPrysmChainClient{chainClient: &grpcChainClient{ethpb.NewBeaconChainClient(cc)}}
}

10
validator/client/grpc-api/grpc_prysm_beacon_chain_client_test.go
View File

@@ -291,8 +291,8 @@ func TestGetValidatorCount(t *testing.T) {
})
}
beaconChainClient := mock.NewMockBeaconChainClient(ctrl)
beaconChainClient.EXPECT().ListValidators(
chainClient := mock.NewMockChainClient(ctrl)
chainClient.EXPECT().ListValidators(
gomock.Any(),
gomock.Any(),
).Return(
@@ -300,7 +300,7 @@ func TestGetValidatorCount(t *testing.T) {
nil,
)
beaconChainClient.EXPECT().GetChainHead(
chainClient.EXPECT().GetChainHead(
gomock.Any(),
gomock.Any(),
).Return(
@@ -308,8 +308,8 @@ func TestGetValidatorCount(t *testing.T) {
nil,
)
prysmBeaconChainClient := &grpcPrysmBeaconChainClient{
beaconChainClient: beaconChainClient,
prysmBeaconChainClient := &grpcPrysmChainClient{
chainClient: chainClient,
}
var statuses []validator.Status

2
validator/client/iface/beacon_chain_client.go
View File

@@ -7,7 +7,7 @@ import (
ethpb "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
)
type BeaconChainClient interface {
type ChainClient interface {
GetChainHead(ctx context.Context, in *empty.Empty) (*ethpb.ChainHead, error)
ListValidatorBalances(ctx context.Context, in *ethpb.ListValidatorBalancesRequest) (*ethpb.ValidatorBalances, error)
ListValidators(ctx context.Context, in *ethpb.ListValidatorsRequest) (*ethpb.Validators, error)

4
validator/client/iface/prysm_beacon_chain_client.go
View File

@@ -14,7 +14,7 @@ type ValidatorCount struct {
Count uint64
}
// PrysmBeaconChainClient defines an interface required to implement all the prysm specific custom endpoints.
type PrysmBeaconChainClient interface {
// PrysmChainClient defines an interface required to implement all the prysm specific custom endpoints.
type PrysmChainClient interface {
GetValidatorCount(context.Context, string, []validator.Status) ([]ValidatorCount, error)
}

2
validator/client/key_reload.go
View File

@@ -39,7 +39,7 @@ func (v *validator) HandleKeyReload(ctx context.Context, currentKeys [][fieldpar
// "-1" indicates that validator count endpoint is not supported by the beacon node.
var valCount int64 = -1
valCounts, err := v.prysmBeaconClient.GetValidatorCount(ctx, "head", []validator2.Status{validator2.Active})
valCounts, err := v.prysmChainClient.GetValidatorCount(ctx, "head", []validator2.Status{validator2.Active})
if err != nil && !errors.Is(err, iface.ErrNotSupported) {
return false, errors.Wrap(err, "could not get active validator count")
}

34
validator/client/key_reload_test.go
View File

@@ -29,14 +29,14 @@ func TestValidator_HandleKeyReload(t *testing.T) {
active := randKeypair(t)
client := validatormock.NewMockValidatorClient(ctrl)
beaconClient := validatormock.NewMockBeaconChainClient(ctrl)
prysmBeaconClient := validatormock.NewMockPrysmBeaconChainClient(ctrl)
chainClient := validatormock.NewMockChainClient(ctrl)
prysmChainClient := validatormock.NewMockPrysmChainClient(ctrl)
v := validator{
validatorClient: client,
keyManager: newMockKeymanager(t, inactive),
genesisTime: 1,
beaconClient: beaconClient,
prysmBeaconClient: prysmBeaconClient,
validatorClient: client,
km: newMockKeymanager(t, inactive),
genesisTime: 1,
chainClient: chainClient,
prysmChainClient: prysmChainClient,
}
resp := testutil.GenerateMultipleValidatorStatusResponse([][]byte{inactive.pub[:], active.pub[:]})
@@ -48,7 +48,7 @@ func TestValidator_HandleKeyReload(t *testing.T) {
PublicKeys: [][]byte{inactive.pub[:], active.pub[:]},
},
).Return(resp, nil)
prysmBeaconClient.EXPECT().GetValidatorCount(
prysmChainClient.EXPECT().GetValidatorCount(
gomock.Any(),
"head",
[]validator2.Status{validator2.Active},
@@ -65,15 +65,15 @@ func TestValidator_HandleKeyReload(t *testing.T) {
hook := logTest.NewGlobal()
client := validatormock.NewMockValidatorClient(ctrl)
beaconClient := validatormock.NewMockBeaconChainClient(ctrl)
prysmBeaconClient := validatormock.NewMockPrysmBeaconChainClient(ctrl)
chainClient := validatormock.NewMockChainClient(ctrl)
prysmChainClient := validatormock.NewMockPrysmChainClient(ctrl)
kp := randKeypair(t)
v := validator{
validatorClient: client,
keyManager: newMockKeymanager(t, kp),
genesisTime: 1,
beaconClient: beaconClient,
prysmBeaconClient: prysmBeaconClient,
validatorClient: client,
km: newMockKeymanager(t, kp),
genesisTime: 1,
chainClient: chainClient,
prysmChainClient: prysmChainClient,
}
resp := testutil.GenerateMultipleValidatorStatusResponse([][]byte{kp.pub[:]})
@@ -84,7 +84,7 @@ func TestValidator_HandleKeyReload(t *testing.T) {
PublicKeys: [][]byte{kp.pub[:]},
},
).Return(resp, nil)
prysmBeaconClient.EXPECT().GetValidatorCount(
prysmChainClient.EXPECT().GetValidatorCount(
gomock.Any(),
"head",
[]validator2.Status{validator2.Active},
@@ -102,7 +102,7 @@ func TestValidator_HandleKeyReload(t *testing.T) {
client := validatormock.NewMockValidatorClient(ctrl)
v := validator{
validatorClient: client,
keyManager: newMockKeymanager(t, kp),
km: newMockKeymanager(t, kp),
genesisTime: 1,
}

22
validator/client/metrics.go
View File

@@ -234,13 +234,13 @@ func (v *validator) LogValidatorGainsAndLosses(ctx context.Context, slot primiti
// Do nothing unless we are at the end of the epoch, and not in the first epoch.
return nil
}
if !v.logValidatorBalances {
if !v.logValidatorPerformance {
return nil
}
var pks [][fieldparams.BLSPubkeyLength]byte
var err error
pks, err = v.keyManager.FetchValidatingPublicKeys(ctx)
pks, err = v.km.FetchValidatingPublicKeys(ctx)
if err != nil {
return err
}
@@ -249,7 +249,7 @@ func (v *validator) LogValidatorGainsAndLosses(ctx context.Context, slot primiti
req := &ethpb.ValidatorPerformanceRequest{
PublicKeys: pubKeys,
}
resp, err := v.beaconClient.GetValidatorPerformance(ctx, req)
resp, err := v.chainClient.GetValidatorPerformance(ctx, req)
if err != nil {
return err
}
@@ -270,11 +270,11 @@ func (v *validator) LogValidatorGainsAndLosses(ctx context.Context, slot primiti
v.voteStats.startEpoch = prevEpoch
}
}
v.prevBalanceLock.Lock()
v.prevEpochBalancesLock.Lock()
for i, pubKey := range resp.PublicKeys {
v.logForEachValidator(i, pubKey, resp, slot, prevEpoch)
}
v.prevBalanceLock.Unlock()
v.prevEpochBalancesLock.Unlock()
v.UpdateLogAggregateStats(resp, slot)
return nil
@@ -284,7 +284,7 @@ func (v *validator) logForEachValidator(index int, pubKey []byte, resp *ethpb.Va
truncatedKey := fmt.Sprintf("%#x", bytesutil.Trunc(pubKey))
pubKeyBytes := bytesutil.ToBytes48(pubKey)
if slot < params.BeaconConfig().SlotsPerEpoch {
v.prevBalance[pubKeyBytes] = params.BeaconConfig().MaxEffectiveBalance
v.prevEpochBalances[pubKeyBytes] = params.BeaconConfig().MaxEffectiveBalance
}
// Safely load data from response with slice out of bounds checks. The server should return
@@ -325,7 +325,7 @@ func (v *validator) logForEachValidator(index int, pubKey []byte, resp *ethpb.Va
fmtKey := fmt.Sprintf("%#x", pubKey)
gweiPerEth := float64(params.BeaconConfig().GweiPerEth)
if v.prevBalance[pubKeyBytes] > 0 {
if v.prevEpochBalances[pubKeyBytes] > 0 {
newBalance := float64(balAfterEpoch) / gweiPerEth
prevBalance := float64(balBeforeEpoch) / gweiPerEth
startBalance := float64(v.startBalances[pubKeyBytes]) / gweiPerEth
@@ -380,7 +380,7 @@ func (v *validator) logForEachValidator(index int, pubKey []byte, resp *ethpb.Va
}
}
}
v.prevBalance[pubKeyBytes] = balBeforeEpoch
v.prevEpochBalances[pubKeyBytes] = balBeforeEpoch
}
// UpdateLogAggregateStats updates and logs the voteStats struct of a validator using the RPC response obtained from LogValidatorGainsAndLosses.
@@ -438,12 +438,12 @@ func (v *validator) UpdateLogAggregateStats(resp *ethpb.ValidatorPerformanceResp
log.WithFields(epochSummaryFields).Info("Previous epoch aggregated voting summary")
var totalStartBal, totalPrevBal uint64
v.prevBalanceLock.RLock()
v.prevEpochBalancesLock.RLock()
for i, val := range v.startBalances {
totalStartBal += val
totalPrevBal += v.prevBalance[i]
totalPrevBal += v.prevEpochBalances[i]
}
v.prevBalanceLock.RUnlock()
v.prevEpochBalancesLock.RUnlock()
if totalStartBal == 0 || summary.totalAttestedCount == 0 {
log.Error("Failed to print launch summary: one or more divisors is 0")

24
validator/client/metrics_test.go
View File

@@ -15,9 +15,9 @@ import (
func TestUpdateLogAggregateStats(t *testing.T) {
v := &validator{
logValidatorBalances: true,
startBalances: make(map[[fieldparams.BLSPubkeyLength]byte]uint64),
prevBalance: make(map[[fieldparams.BLSPubkeyLength]byte]uint64),
logValidatorPerformance: true,
startBalances: make(map[[fieldparams.BLSPubkeyLength]byte]uint64),
prevEpochBalances: make(map[[fieldparams.BLSPubkeyLength]byte]uint64),
voteStats: voteStats{
startEpoch: 0, // this would otherwise have been previously set in LogValidatorGainsAndLosses()
},
@@ -66,9 +66,9 @@ func TestUpdateLogAggregateStats(t *testing.T) {
},
}
v.prevBalance[pubKeyBytes[0]] = uint64(33200000000)
v.prevBalance[pubKeyBytes[1]] = uint64(33300000000)
v.prevBalance[pubKeyBytes[2]] = uint64(31000000000)
v.prevEpochBalances[pubKeyBytes[0]] = uint64(33200000000)
v.prevEpochBalances[pubKeyBytes[1]] = uint64(33300000000)
v.prevEpochBalances[pubKeyBytes[2]] = uint64(31000000000)
var hook *logTest.Hook
@@ -89,9 +89,9 @@ func TestUpdateLogAggregateStats(t *testing.T) {
func TestUpdateLogAltairAggregateStats(t *testing.T) {
v := &validator{
logValidatorBalances: true,
startBalances: make(map[[fieldparams.BLSPubkeyLength]byte]uint64),
prevBalance: make(map[[fieldparams.BLSPubkeyLength]byte]uint64),
logValidatorPerformance: true,
startBalances: make(map[[fieldparams.BLSPubkeyLength]byte]uint64),
prevEpochBalances: make(map[[fieldparams.BLSPubkeyLength]byte]uint64),
voteStats: voteStats{
startEpoch: params.BeaconConfig().AltairForkEpoch, // this would otherwise have been previously set in LogValidatorGainsAndLosses()
},
@@ -141,9 +141,9 @@ func TestUpdateLogAltairAggregateStats(t *testing.T) {
},
}
v.prevBalance[pubKeyBytes[0]] = uint64(33200000000)
v.prevBalance[pubKeyBytes[1]] = uint64(33300000000)
v.prevBalance[pubKeyBytes[2]] = uint64(31000000000)
v.prevEpochBalances[pubKeyBytes[0]] = uint64(33200000000)
v.prevEpochBalances[pubKeyBytes[1]] = uint64(33300000000)
v.prevEpochBalances[pubKeyBytes[2]] = uint64(31000000000)
var hook *logTest.Hook

4
validator/client/propose.go
View File

@@ -341,7 +341,7 @@ func (v *validator) signRandaoReveal(ctx context.Context, pubKey [fieldparams.BL
if err != nil {
return nil, err
}
randaoReveal, err = v.keyManager.Sign(ctx, &validatorpb.SignRequest{
randaoReveal, err = v.km.Sign(ctx, &validatorpb.SignRequest{
PublicKey: pubKey[:],
SigningRoot: root[:],
SignatureDomain: domain.SignatureDomain,
@@ -373,7 +373,7 @@ func (v *validator) signBlock(ctx context.Context, pubKey [fieldparams.BLSPubkey
if err != nil {
return nil, [32]byte{}, err
}
sig, err := v.keyManager.Sign(ctx, &validatorpb.SignRequest{
sig, err := v.km.Sign(ctx, &validatorpb.SignRequest{
PublicKey: pubKey[:],
SigningRoot: blockRoot[:],
SignatureDomain: domain.SignatureDomain,

8
validator/client/propose_test.go
View File

@@ -94,7 +94,7 @@ func setupWithKey(t *testing.T, validatorKey bls.SecretKey, isSlashingProtection
validator := &validator{
db: valDB,
keyManager: newMockKeymanager(t, keypair{pub: pubKey, pri: validatorKey}),
km: newMockKeymanager(t, keypair{pub: pubKey, pri: validatorKey}),
validatorClient: m.validatorClient,
graffiti: []byte{},
submittedAtts: make(map[submittedAttKey]*submittedAtt),
@@ -879,7 +879,7 @@ func TestSignBlock(t *testing.T) {
kp := testKeyFromBytes(t, []byte{1})
validator.keyManager = newMockKeymanager(t, kp)
validator.km = newMockKeymanager(t, kp)
b, err := blocks.NewBeaconBlock(blk.Block)
require.NoError(t, err)
sig, blockRoot, err := validator.signBlock(ctx, kp.pub, 0, 0, b)
@@ -915,7 +915,7 @@ func TestSignAltairBlock(t *testing.T) {
blk := util.NewBeaconBlockAltair()
blk.Block.Slot = 1
blk.Block.ProposerIndex = 100
validator.keyManager = newMockKeymanager(t, kp)
validator.km = newMockKeymanager(t, kp)
wb, err := blocks.NewBeaconBlock(blk.Block)
require.NoError(t, err)
sig, blockRoot, err := validator.signBlock(ctx, kp.pub, 0, 0, wb)
@@ -948,7 +948,7 @@ func TestSignBellatrixBlock(t *testing.T) {
blk.Block.ProposerIndex = 100
kp := randKeypair(t)
validator.keyManager = newMockKeymanager(t, kp)
validator.km = newMockKeymanager(t, kp)
wb, err := blocks.NewBeaconBlock(blk.Block)
require.NoError(t, err)
sig, blockRoot, err := validator.signBlock(ctx, kp.pub, 0, 0, wb)

228
validator/client/service.go
View File

@@ -3,7 +3,6 @@ package client
import (
"context"
"net/http"
"strings"
"time"
"github.com/dgraph-io/ristretto"
@@ -35,75 +34,54 @@ import (
"go.opencensus.io/plugin/ocgrpc"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"google.golang.org/protobuf/types/known/emptypb"
)
// SyncChecker is able to determine if a beacon node is currently
// going through chain synchronization.
type SyncChecker interface {
Syncing(ctx context.Context) (bool, error)
}
// GenesisFetcher can retrieve genesis information such as
// the genesis time and the validator deposit contract address.
type GenesisFetcher interface {
GenesisInfo(ctx context.Context) (*ethpb.Genesis, error)
}
// ValidatorService represents a service to manage the validator client
// routine.
type ValidatorService struct {
useWeb bool
emitAccountMetrics bool
logValidatorBalances bool
distributed bool
interopKeysConfig *local.InteropKeymanagerConfig
conn validatorHelpers.NodeConnection
grpcRetryDelay time.Duration
grpcRetries uint
maxCallRecvMsgSize int
cancel context.CancelFunc
walletInitializedFeed *event.Feed
wallet *wallet.Wallet
graffitiStruct *graffiti.Graffiti
dataDir string
withCert string
endpoint string
ctx context.Context
validator iface.Validator
db db.Database
grpcHeaders []string
graffiti []byte
Web3SignerConfig *remoteweb3signer.SetupConfig
proposerSettings *proposer.Settings
validatorsRegBatchSize int
ctx context.Context
cancel context.CancelFunc
validator iface.Validator
db db.Database
conn validatorHelpers.NodeConnection
wallet *wallet.Wallet
walletInitializedFeed *event.Feed
graffiti []byte
graffitiStruct *graffiti.Graffiti
interopKeysConfig *local.InteropKeymanagerConfig
web3SignerConfig *remoteweb3signer.SetupConfig
proposerSettings *proposer.Settings
validatorsRegBatchSize int
useWeb bool
emitAccountMetrics bool
logValidatorPerformance bool
distributed bool
}
// Config for the validator service.
type Config struct {
UseWeb bool
LogValidatorBalances bool
EmitAccountMetrics bool
Distributed bool
InteropKeysConfig *local.InteropKeymanagerConfig
Wallet *wallet.Wallet
WalletInitializedFeed *event.Feed
GrpcRetriesFlag uint
GrpcMaxCallRecvMsgSizeFlag int
GrpcRetryDelay time.Duration
GraffitiStruct *graffiti.Graffiti
Validator iface.Validator
ValDB db.Database
CertFlag string
DataDir string
GrpcHeadersFlag string
GraffitiFlag string
Endpoint string
Web3SignerConfig *remoteweb3signer.SetupConfig
ProposerSettings *proposer.Settings
BeaconApiEndpoint string
BeaconApiTimeout time.Duration
ValidatorsRegBatchSize int
Validator iface.Validator
DB db.Database
Wallet *wallet.Wallet
WalletInitializedFeed *event.Feed
GRPCMaxCallRecvMsgSize int
GRPCRetries uint
GRPCRetryDelay time.Duration
GRPCHeaders []string
BeaconNodeGRPCEndpoint string
BeaconNodeCert string
BeaconApiEndpoint string
BeaconApiTimeout time.Duration
Graffiti string
GraffitiStruct *graffiti.Graffiti
InteropKmConfig *local.InteropKeymanagerConfig
Web3SignerConfig *remoteweb3signer.SetupConfig
ProposerSettings *proposer.Settings
ValidatorsRegBatchSize int
UseWeb bool
LogValidatorPerformance bool
EmitAccountMetrics bool
Distributed bool
}
// NewValidatorService creates a new validator service for the service
@@ -111,48 +89,41 @@ type Config struct {
func NewValidatorService(ctx context.Context, cfg *Config) (*ValidatorService, error) {
ctx, cancel := context.WithCancel(ctx)
s := &ValidatorService{
ctx: ctx,
cancel: cancel,
endpoint: cfg.Endpoint,
withCert: cfg.CertFlag,
dataDir: cfg.DataDir,
graffiti: []byte(cfg.GraffitiFlag),
logValidatorBalances: cfg.LogValidatorBalances,
emitAccountMetrics: cfg.EmitAccountMetrics,
maxCallRecvMsgSize: cfg.GrpcMaxCallRecvMsgSizeFlag,
grpcRetries: cfg.GrpcRetriesFlag,
grpcRetryDelay: cfg.GrpcRetryDelay,
grpcHeaders: strings.Split(cfg.GrpcHeadersFlag, ","),
validator: cfg.Validator,
db: cfg.ValDB,
wallet: cfg.Wallet,
walletInitializedFeed: cfg.WalletInitializedFeed,
useWeb: cfg.UseWeb,
interopKeysConfig: cfg.InteropKeysConfig,
graffitiStruct: cfg.GraffitiStruct,
Web3SignerConfig: cfg.Web3SignerConfig,
proposerSettings: cfg.ProposerSettings,
validatorsRegBatchSize: cfg.ValidatorsRegBatchSize,
distributed: cfg.Distributed,
ctx: ctx,
cancel: cancel,
validator: cfg.Validator,
db: cfg.DB,
wallet: cfg.Wallet,
walletInitializedFeed: cfg.WalletInitializedFeed,
graffiti: []byte(cfg.Graffiti),
graffitiStruct: cfg.GraffitiStruct,
interopKeysConfig: cfg.InteropKmConfig,
web3SignerConfig: cfg.Web3SignerConfig,
proposerSettings: cfg.ProposerSettings,
validatorsRegBatchSize: cfg.ValidatorsRegBatchSize,
useWeb: cfg.UseWeb,
emitAccountMetrics: cfg.EmitAccountMetrics,
logValidatorPerformance: cfg.LogValidatorPerformance,
distributed: cfg.Distributed,
}
dialOpts := ConstructDialOptions(
s.maxCallRecvMsgSize,
s.withCert,
s.grpcRetries,
s.grpcRetryDelay,
cfg.GRPCMaxCallRecvMsgSize,
cfg.BeaconNodeCert,
cfg.GRPCRetries,
cfg.GRPCRetryDelay,
)
if dialOpts == nil {
return s, nil
}
s.ctx = grpcutil.AppendHeaders(ctx, s.grpcHeaders)
s.ctx = grpcutil.AppendHeaders(ctx, cfg.GRPCHeaders)
grpcConn, err := grpc.DialContext(ctx, s.endpoint, dialOpts...)
grpcConn, err := grpc.DialContext(ctx, cfg.BeaconNodeGRPCEndpoint, dialOpts...)
if err != nil {
return s, err
}
if s.withCert != "" {
if cfg.BeaconNodeCert != "" {
log.Info("Established secure gRPC connection")
}
s.conn = validatorHelpers.NewNodeConnection(
@@ -202,38 +173,39 @@ func (v *ValidatorService) Start() {
validatorClient := validatorClientFactory.NewValidatorClient(v.conn, restHandler)
valStruct := &validator{
validatorClient: validatorClient,
beaconClient: beaconChainClientFactory.NewBeaconChainClient(v.conn, restHandler),
nodeClient: nodeClientFactory.NewNodeClient(v.conn, restHandler),
prysmBeaconClient: beaconChainClientFactory.NewPrysmBeaconClient(v.conn, restHandler),
db: v.db,
graffiti: v.graffiti,
logValidatorBalances: v.logValidatorBalances,
emitAccountMetrics: v.emitAccountMetrics,
startBalances: make(map[[fieldparams.BLSPubkeyLength]byte]uint64),
prevBalance: make(map[[fieldparams.BLSPubkeyLength]byte]uint64),
pubkeyToValidatorIndex: make(map[[fieldparams.BLSPubkeyLength]byte]primitives.ValidatorIndex),
signedValidatorRegistrations: make(map[[fieldparams.BLSPubkeyLength]byte]*ethpb.SignedValidatorRegistrationV1),
submittedAtts: make(map[submittedAttKey]*submittedAtt),
submittedAggregates: make(map[submittedAttKey]*submittedAtt),
domainDataCache: cache,
aggregatedSlotCommitteeIDCache: aggregatedSlotCommitteeIDCache,
voteStats: voteStats{startEpoch: primitives.Epoch(^uint64(0))},
syncCommitteeStats: syncCommitteeStats{},
useWeb: v.useWeb,
interopKeysConfig: v.interopKeysConfig,
wallet: v.wallet,
walletInitializedFeed: v.walletInitializedFeed,
slotFeed: new(event.Feed),
startBalances: make(map[[fieldparams.BLSPubkeyLength]byte]uint64),
prevEpochBalances: make(map[[fieldparams.BLSPubkeyLength]byte]uint64),
blacklistedPubkeys: slashablePublicKeys,
pubkeyToValidatorIndex: make(map[[fieldparams.BLSPubkeyLength]byte]primitives.ValidatorIndex),
wallet: v.wallet,
walletInitializedChan: make(chan *wallet.Wallet, 1),
walletInitializedFeed: v.walletInitializedFeed,
graffiti: v.graffiti,
graffitiStruct: v.graffitiStruct,
graffitiOrderedIndex: graffitiOrderedIndex,
eipImportBlacklistedPublicKeys: slashablePublicKeys,
Web3SignerConfig: v.Web3SignerConfig,
validatorClient: validatorClient,
chainClient: beaconChainClientFactory.NewChainClient(v.conn, restHandler),
nodeClient: nodeClientFactory.NewNodeClient(v.conn, restHandler),
prysmChainClient: beaconChainClientFactory.NewPrysmChainClient(v.conn, restHandler),
db: v.db,
km: nil,
web3SignerConfig: v.web3SignerConfig,
proposerSettings: v.proposerSettings,
walletInitializedChannel: make(chan *wallet.Wallet, 1),
signedValidatorRegistrations: make(map[[fieldparams.BLSPubkeyLength]byte]*ethpb.SignedValidatorRegistrationV1),
validatorsRegBatchSize: v.validatorsRegBatchSize,
distributed: v.distributed,
interopKeysConfig: v.interopKeysConfig,
attSelections: make(map[attSelectionKey]iface.BeaconCommitteeSelection),
aggregatedSlotCommitteeIDCache: aggregatedSlotCommitteeIDCache,
domainDataCache: cache,
voteStats: voteStats{startEpoch: primitives.Epoch(^uint64(0))},
syncCommitteeStats: syncCommitteeStats{},
submittedAtts: make(map[submittedAttKey]*submittedAtt),
submittedAggregates: make(map[submittedAttKey]*submittedAtt),
logValidatorPerformance: v.logValidatorPerformance,
emitAccountMetrics: v.emitAccountMetrics,
useWeb: v.useWeb,
distributed: v.distributed,
}
v.validator = valStruct
@@ -268,6 +240,11 @@ func (v *ValidatorService) Keymanager() (keymanager.IKeymanager, error) {
return v.validator.Keymanager()
}
// RemoteSignerConfig returns the web3signer configuration
func (v *ValidatorService) RemoteSignerConfig() *remoteweb3signer.SetupConfig {
return v.web3SignerConfig
}
// ProposerSettings returns a deep copy of the underlying proposer settings in the validator
func (v *ValidatorService) ProposerSettings() *proposer.Settings {
settings := v.validator.ProposerSettings()
@@ -342,23 +319,6 @@ func ConstructDialOptions(
return dialOpts
}
// Syncing returns whether or not the beacon node is currently synchronizing the chain.
func (v *ValidatorService) Syncing(ctx context.Context) (bool, error) {
nc := ethpb.NewNodeClient(v.conn.GetGrpcClientConn())
resp, err := nc.GetSyncStatus(ctx, &emptypb.Empty{})
if err != nil {
return false, err
}
return resp.Syncing, nil
}
// GenesisInfo queries the beacon node for the chain genesis info containing
// the genesis time along with the validator deposit contract address.
func (v *ValidatorService) GenesisInfo(ctx context.Context) (*ethpb.Genesis, error) {
nc := ethpb.NewNodeClient(v.conn.GetGrpcClientConn())
return nc.GetGenesis(ctx, &emptypb.Empty{})
}
func (v *ValidatorService) GetGraffiti(ctx context.Context, pubKey [fieldparams.BLSPubkeyLength]byte) ([]byte, error) {
if v.validator == nil {
return nil, errors.New("validator is unavailable")

5
validator/client/service_test.go
View File

@@ -2,6 +2,7 @@ package client
import (
"context"
"strings"
"testing"
"time"
@@ -13,8 +14,6 @@ import (
)
var _ runtime.Service = (*ValidatorService)(nil)
var _ GenesisFetcher = (*ValidatorService)(nil)
var _ SyncChecker = (*ValidatorService)(nil)
func TestStop_CancelsContext(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
@@ -59,7 +58,7 @@ func TestStart_GrpcHeaders(t *testing.T) {
"Authorization", "this is a valid value",
},
} {
cfg := &Config{GrpcHeadersFlag: input}
cfg := &Config{GRPCHeaders: strings.Split(input, ",")}
validatorService, err := NewValidatorService(ctx, cfg)
require.NoError(t, err)
md, _ := metadata.FromOutgoingContext(validatorService.ctx)

6
validator/client/sync_committee.go
View File

@@ -58,7 +58,7 @@ func (v *validator) SubmitSyncCommitteeMessage(ctx context.Context, slot primiti
return
}
sig, err := v.keyManager.Sign(ctx, &validatorpb.SignRequest{
sig, err := v.km.Sign(ctx, &validatorpb.SignRequest{
PublicKey: pubKey[:],
SigningRoot: r[:],
SignatureDomain: d.SignatureDomain,
@@ -243,7 +243,7 @@ func (v *validator) signSyncSelectionData(ctx context.Context, pubKey [fieldpara
if err != nil {
return nil, err
}
sig, err := v.keyManager.Sign(ctx, &validatorpb.SignRequest{
sig, err := v.km.Sign(ctx, &validatorpb.SignRequest{
PublicKey: pubKey[:],
SigningRoot: root[:],
SignatureDomain: domain.SignatureDomain,
@@ -266,7 +266,7 @@ func (v *validator) signContributionAndProof(ctx context.Context, pubKey [fieldp
if err != nil {
return nil, err
}
sig, err := v.keyManager.Sign(ctx, &validatorpb.SignRequest{
sig, err := v.km.Sign(ctx, &validatorpb.SignRequest{
PublicKey: pubKey[:],
SigningRoot: root[:],
SignatureDomain: d.SignatureDomain,

173
validator/client/validator.go
View File

@@ -41,7 +41,7 @@ import (
"github.com/prysmaticlabs/prysm/v5/validator/accounts/wallet"
beacon_api "github.com/prysmaticlabs/prysm/v5/validator/client/beacon-api"
"github.com/prysmaticlabs/prysm/v5/validator/client/iface"
vdb "github.com/prysmaticlabs/prysm/v5/validator/db"
"github.com/prysmaticlabs/prysm/v5/validator/db"
dbCommon "github.com/prysmaticlabs/prysm/v5/validator/db/common"
"github.com/prysmaticlabs/prysm/v5/validator/graffiti"
"github.com/prysmaticlabs/prysm/v5/validator/keymanager"
@@ -69,51 +69,51 @@ var (
)
type validator struct {
logValidatorBalances bool
useWeb bool
duties *ethpb.DutiesResponse
ticker slots.Ticker
genesisTime uint64
highestValidSlot primitives.Slot
slotFeed *event.Feed
startBalances map[[fieldparams.BLSPubkeyLength]byte]uint64
prevEpochBalances map[[fieldparams.BLSPubkeyLength]byte]uint64
blacklistedPubkeys map[[fieldparams.BLSPubkeyLength]byte]bool
pubkeyToValidatorIndex map[[fieldparams.BLSPubkeyLength]byte]primitives.ValidatorIndex
wallet *wallet.Wallet
walletInitializedChan chan *wallet.Wallet
walletInitializedFeed *event.Feed
graffiti []byte
graffitiStruct *graffiti.Graffiti
graffitiOrderedIndex uint64
validatorClient iface.ValidatorClient
chainClient iface.ChainClient
nodeClient iface.NodeClient
prysmChainClient iface.PrysmChainClient
db db.Database
km keymanager.IKeymanager
web3SignerConfig *remoteweb3signer.SetupConfig
proposerSettings *proposer.Settings
signedValidatorRegistrations map[[fieldparams.BLSPubkeyLength]byte]*ethpb.SignedValidatorRegistrationV1
validatorsRegBatchSize int
interopKeysConfig *local.InteropKeymanagerConfig
attSelections map[attSelectionKey]iface.BeaconCommitteeSelection
aggregatedSlotCommitteeIDCache *lru.Cache
domainDataCache *ristretto.Cache
voteStats voteStats
syncCommitteeStats syncCommitteeStats
submittedAtts map[submittedAttKey]*submittedAtt
submittedAggregates map[submittedAttKey]*submittedAtt
logValidatorPerformance bool
emitAccountMetrics bool
useWeb bool
distributed bool
domainDataLock sync.RWMutex
attLogsLock sync.Mutex
aggregatedSlotCommitteeIDCacheLock sync.Mutex
highestValidSlotLock sync.Mutex
prevBalanceLock sync.RWMutex
slashableKeysLock sync.RWMutex
prevEpochBalancesLock sync.RWMutex
blacklistedPubkeysLock sync.RWMutex
attSelectionLock sync.Mutex
eipImportBlacklistedPublicKeys map[[fieldparams.BLSPubkeyLength]byte]bool
walletInitializedFeed *event.Feed
submittedAtts map[submittedAttKey]*submittedAtt
submittedAggregates map[submittedAttKey]*submittedAtt
startBalances map[[fieldparams.BLSPubkeyLength]byte]uint64
dutiesLock sync.RWMutex
duties *ethpb.DutiesResponse
prevBalance map[[fieldparams.BLSPubkeyLength]byte]uint64
pubkeyToValidatorIndex map[[fieldparams.BLSPubkeyLength]byte]primitives.ValidatorIndex
signedValidatorRegistrations map[[fieldparams.BLSPubkeyLength]byte]*ethpb.SignedValidatorRegistrationV1
attSelections map[attSelectionKey]iface.BeaconCommitteeSelection
graffitiOrderedIndex uint64
aggregatedSlotCommitteeIDCache *lru.Cache
domainDataCache *ristretto.Cache
highestValidSlot primitives.Slot
genesisTime uint64
slotFeed *event.Feed
interopKeysConfig *local.InteropKeymanagerConfig
wallet *wallet.Wallet
graffitiStruct *graffiti.Graffiti
beaconClient iface.BeaconChainClient
nodeClient iface.NodeClient
validatorClient iface.ValidatorClient
prysmBeaconClient iface.PrysmBeaconChainClient
db vdb.Database
keyManager keymanager.IKeymanager
ticker slots.Ticker
graffiti []byte
voteStats voteStats
syncCommitteeStats syncCommitteeStats
Web3SignerConfig *remoteweb3signer.SetupConfig
proposerSettings *proposer.Settings
walletInitializedChannel chan *wallet.Wallet
validatorsRegBatchSize int
}
type validatorStatus struct {
@@ -132,7 +132,7 @@ func (v *validator) Done() {
v.ticker.Done()
}
// WaitForKeymanagerInitialization checks if the validator needs to wait for
// WaitForKmInitialization checks if the validator needs to wait for keymanager initialization.
func (v *validator) WaitForKeymanagerInitialization(ctx context.Context) error {
genesisRoot, err := v.db.GenesisValidatorsRoot(ctx)
if err != nil {
@@ -142,32 +142,32 @@ func (v *validator) WaitForKeymanagerInitialization(ctx context.Context) error {
if v.useWeb && v.wallet == nil {
log.Info("Waiting for keymanager to initialize validator client with web UI")
// if wallet is not set, wait for it to be set through the UI
km, err := waitForWebWalletInitialization(ctx, v.walletInitializedFeed, v.walletInitializedChannel)
km, err := waitForWebWalletInitialization(ctx, v.walletInitializedFeed, v.walletInitializedChan)
if err != nil {
return err
}
v.keyManager = km
v.km = km
} else {
if v.interopKeysConfig != nil {
keyManager, err := local.NewInteropKeymanager(ctx, v.interopKeysConfig.Offset, v.interopKeysConfig.NumValidatorKeys)
if err != nil {
return errors.Wrap(err, "could not generate interop keys for key manager")
}
v.keyManager = keyManager
v.km = keyManager
} else if v.wallet == nil {
return errors.New("wallet not set")
} else {
if v.Web3SignerConfig != nil {
v.Web3SignerConfig.GenesisValidatorsRoot = genesisRoot
if v.web3SignerConfig != nil {
v.web3SignerConfig.GenesisValidatorsRoot = genesisRoot
}
keyManager, err := v.wallet.InitializeKeymanager(ctx, accountsiface.InitKeymanagerConfig{ListenForChanges: true, Web3SignerConfig: v.Web3SignerConfig})
keyManager, err := v.wallet.InitializeKeymanager(ctx, accountsiface.InitKeymanagerConfig{ListenForChanges: true, Web3SignerConfig: v.web3SignerConfig})
if err != nil {
return errors.Wrap(err, "could not initialize key manager")
}
v.keyManager = keyManager
v.km = keyManager
}
}
recheckKeys(ctx, v.db, v.keyManager)
recheckKeys(ctx, v.db, v.km)
return nil
}
@@ -197,23 +197,22 @@ func waitForWebWalletInitialization(
}
// recheckKeys checks if the validator has any keys that need to be rechecked.
// the keymanager implements a subscription to push these updates to the validator.
func recheckKeys(ctx context.Context, valDB vdb.Database, keyManager keymanager.IKeymanager) {
// The keymanager implements a subscription to push these updates to the validator.
func recheckKeys(ctx context.Context, valDB db.Database, km keymanager.IKeymanager) {
var validatingKeys [][fieldparams.BLSPubkeyLength]byte
var err error
validatingKeys, err = keyManager.FetchValidatingPublicKeys(ctx)
validatingKeys, err = km.FetchValidatingPublicKeys(ctx)
if err != nil {
log.WithError(err).Debug("Could not fetch validating keys")
}
if err := valDB.UpdatePublicKeysBuckets(validatingKeys); err != nil {
log.WithError(err).Debug("Could not update public keys buckets")
go recheckValidatingKeysBucket(ctx, valDB, km)
}
go recheckValidatingKeysBucket(ctx, valDB, keyManager)
}
// to accounts changes in the keymanager, then updates those keys'
// buckets in bolt DB if a bucket for a key does not exist.
func recheckValidatingKeysBucket(ctx context.Context, valDB vdb.Database, km keymanager.IKeymanager) {
func recheckValidatingKeysBucket(ctx context.Context, valDB db.Database, km keymanager.IKeymanager) {
importedKeymanager, ok := km.(*local.Keymanager)
if !ok {
return
@@ -403,7 +402,7 @@ func (v *validator) checkAndLogValidatorStatus(statuses []*validatorStatus, acti
func (v *validator) CanonicalHeadSlot(ctx context.Context) (primitives.Slot, error) {
ctx, span := trace.StartSpan(ctx, "validator.CanonicalHeadSlot")
defer span.End()
head, err := v.beaconClient.GetChainHead(ctx, &emptypb.Empty{})
head, err := v.chainClient.GetChainHead(ctx, &emptypb.Empty{})
if err != nil {
return 0, errors.Wrap(client.ErrConnectionIssue, err.Error())
}
@@ -427,7 +426,7 @@ func (v *validator) CheckDoppelGanger(ctx context.Context) error {
if !features.Get().EnableDoppelGanger {
return nil
}
pubkeys, err := v.keyManager.FetchValidatingPublicKeys(ctx)
pubkeys, err := v.km.FetchValidatingPublicKeys(ctx)
if err != nil {
return err
}
@@ -533,16 +532,16 @@ func (v *validator) UpdateDuties(ctx context.Context, slot primitives.Slot) erro
ctx, span := trace.StartSpan(ctx, "validator.UpdateAssignments")
defer span.End()
validatingKeys, err := v.keyManager.FetchValidatingPublicKeys(ctx)
validatingKeys, err := v.km.FetchValidatingPublicKeys(ctx)
if err != nil {
return err
}
// Filter out the slashable public keys from the duties request.
filteredKeys := make([][fieldparams.BLSPubkeyLength]byte, 0, len(validatingKeys))
v.slashableKeysLock.RLock()
v.blacklistedPubkeysLock.RLock()
for _, pubKey := range validatingKeys {
if ok := v.eipImportBlacklistedPublicKeys[pubKey]; !ok {
if ok := v.blacklistedPubkeys[pubKey]; !ok {
filteredKeys = append(filteredKeys, pubKey)
} else {
log.WithField(
@@ -551,7 +550,7 @@ func (v *validator) UpdateDuties(ctx context.Context, slot primitives.Slot) erro
"in request to update validator duties")
}
}
v.slashableKeysLock.RUnlock()
v.blacklistedPubkeysLock.RUnlock()
req := &ethpb.DutiesRequest{
Epoch: primitives.Epoch(slot / params.BeaconConfig().SlotsPerEpoch),
@@ -601,28 +600,28 @@ func (v *validator) UpdateDuties(ctx context.Context, slot primitives.Slot) erro
// subscribeToSubnets iterates through each validator duty, signs each slot, and asks beacon node
// to eagerly subscribe to subnets so that the aggregator has attestations to aggregate.
func (v *validator) subscribeToSubnets(ctx context.Context, res *ethpb.DutiesResponse) error {
subscribeSlots := make([]primitives.Slot, 0, len(res.CurrentEpochDuties)+len(res.NextEpochDuties))
subscribeCommitteeIndices := make([]primitives.CommitteeIndex, 0, len(res.CurrentEpochDuties)+len(res.NextEpochDuties))
subscribeIsAggregator := make([]bool, 0, len(res.CurrentEpochDuties)+len(res.NextEpochDuties))
activeDuties := make([]*ethpb.DutiesResponse_Duty, 0, len(res.CurrentEpochDuties)+len(res.NextEpochDuties))
func (v *validator) subscribeToSubnets(ctx context.Context, duties *ethpb.DutiesResponse) error {
subscribeSlots := make([]primitives.Slot, 0, len(duties.CurrentEpochDuties)+len(duties.NextEpochDuties))
subscribeCommitteeIndices := make([]primitives.CommitteeIndex, 0, len(duties.CurrentEpochDuties)+len(duties.NextEpochDuties))
subscribeIsAggregator := make([]bool, 0, len(duties.CurrentEpochDuties)+len(duties.NextEpochDuties))
activeDuties := make([]*ethpb.DutiesResponse_Duty, 0, len(duties.CurrentEpochDuties)+len(duties.NextEpochDuties))
alreadySubscribed := make(map[[64]byte]bool)
if v.distributed {
// Get aggregated selection proofs to calculate isAggregator.
if err := v.getAggregatedSelectionProofs(ctx, res); err != nil {
if err := v.getAggregatedSelectionProofs(ctx, duties); err != nil {
return errors.Wrap(err, "could not get aggregated selection proofs")
}
}
for _, duty := range res.CurrentEpochDuties {
for _, duty := range duties.CurrentEpochDuties {
pk := bytesutil.ToBytes48(duty.PublicKey)
if duty.Status == ethpb.ValidatorStatus_ACTIVE || duty.Status == ethpb.ValidatorStatus_EXITING {
attesterSlot := duty.AttesterSlot
committeeIndex := duty.CommitteeIndex
validatorIndex := duty.ValidatorIndex
alreadySubscribedKey := validatorSubscribeKey(attesterSlot, committeeIndex)
alreadySubscribedKey := validatorSubnetSubscriptionKey(attesterSlot, committeeIndex)
if _, ok := alreadySubscribed[alreadySubscribedKey]; ok {
continue
}
@@ -642,13 +641,13 @@ func (v *validator) subscribeToSubnets(ctx context.Context, res *ethpb.DutiesRes
}
}
for _, duty := range res.NextEpochDuties {
for _, duty := range duties.NextEpochDuties {
if duty.Status == ethpb.ValidatorStatus_ACTIVE || duty.Status == ethpb.ValidatorStatus_EXITING {
attesterSlot := duty.AttesterSlot
committeeIndex := duty.CommitteeIndex
validatorIndex := duty.ValidatorIndex
alreadySubscribedKey := validatorSubscribeKey(attesterSlot, committeeIndex)
alreadySubscribedKey := validatorSubnetSubscriptionKey(attesterSlot, committeeIndex)
if _, ok := alreadySubscribed[alreadySubscribedKey]; ok {
continue
}
@@ -751,18 +750,24 @@ func (v *validator) RolesAt(ctx context.Context, slot primitives.Slot) (map[[fie
// Keymanager returns the underlying validator's keymanager.
func (v *validator) Keymanager() (keymanager.IKeymanager, error) {
if v.keyManager == nil {
if v.km == nil {
return nil, errors.New("keymanager is not initialized")
}
return v.keyManager, nil
return v.km, nil
}
// isAggregator checks if a validator is an aggregator of a given slot and committee,
// it uses a modulo calculated by validator count in committee and samples randomness around it.
func (v *validator) isAggregator(ctx context.Context, committee []primitives.ValidatorIndex, slot primitives.Slot, pubKey [fieldparams.BLSPubkeyLength]byte, validatorIndex primitives.ValidatorIndex) (bool, error) {
func (v *validator) isAggregator(
ctx context.Context,
committeeIndex []primitives.ValidatorIndex,
slot primitives.Slot,
pubKey [fieldparams.BLSPubkeyLength]byte,
validatorIndex primitives.ValidatorIndex,
) (bool, error) {
modulo := uint64(1)
if len(committee)/int(params.BeaconConfig().TargetAggregatorsPerCommittee) > 1 {
modulo = uint64(len(committee)) / params.BeaconConfig().TargetAggregatorsPerCommittee
if len(committeeIndex)/int(params.BeaconConfig().TargetAggregatorsPerCommittee) > 1 {
modulo = uint64(len(committeeIndex)) / params.BeaconConfig().TargetAggregatorsPerCommittee
}
var (
@@ -1041,7 +1046,7 @@ func (v *validator) PushProposerSettings(ctx context.Context, km keymanager.IKey
if err != nil {
return err
}
proposerReqs, err := v.buildPrepProposerReqs(ctx, filteredKeys)
proposerReqs, err := v.buildPrepProposerReqs(filteredKeys)
if err != nil {
return err
}
@@ -1152,9 +1157,9 @@ func (v *validator) filterAndCacheActiveKeys(ctx context.Context, pubkeys [][fie
return filteredKeys, nil
}
func (v *validator) buildPrepProposerReqs(ctx context.Context, pubkeys [][fieldparams.BLSPubkeyLength]byte /* only active pubkeys */) ([]*ethpb.PrepareBeaconProposerRequest_FeeRecipientContainer, error) {
func (v *validator) buildPrepProposerReqs(activePubkeys [][fieldparams.BLSPubkeyLength]byte) ([]*ethpb.PrepareBeaconProposerRequest_FeeRecipientContainer, error) {
var prepareProposerReqs []*ethpb.PrepareBeaconProposerRequest_FeeRecipientContainer
for _, k := range pubkeys {
for _, k := range activePubkeys {
// Default case: Define fee recipient to burn address
var feeRecipient common.Address
@@ -1185,7 +1190,11 @@ func (v *validator) buildPrepProposerReqs(ctx context.Context, pubkeys [][fieldp
return prepareProposerReqs, nil
}
func (v *validator) buildSignedRegReqs(ctx context.Context, pubkeys [][fieldparams.BLSPubkeyLength]byte /* only active pubkeys */, signer iface.SigningFunc) []*ethpb.SignedValidatorRegistrationV1 {
func (v *validator) buildSignedRegReqs(
ctx context.Context,
activePubkeys [][fieldparams.BLSPubkeyLength]byte,
signer iface.SigningFunc,
) []*ethpb.SignedValidatorRegistrationV1 {
var signedValRegRegs []*ethpb.SignedValidatorRegistrationV1
if v.ProposerSettings() == nil {
return signedValRegRegs
@@ -1194,7 +1203,7 @@ func (v *validator) buildSignedRegReqs(ctx context.Context, pubkeys [][fieldpara
if v.genesisTime > uint64(time.Now().UTC().Unix()) {
return signedValRegRegs
}
for i, k := range pubkeys {
for i, k := range activePubkeys {
feeRecipient := common.HexToAddress(params.BeaconConfig().EthBurnAddressHex)
gasLimit := params.BeaconConfig().DefaultBuilderGasLimit
enabled := false
@@ -1244,7 +1253,7 @@ func (v *validator) buildSignedRegReqs(ctx context.Context, pubkeys [][fieldpara
FeeRecipient: feeRecipient[:],
GasLimit: gasLimit,
Timestamp: uint64(time.Now().UTC().Unix()),
Pubkey: pubkeys[i][:],
Pubkey: activePubkeys[i][:],
}
signedReq, err := v.SignValidatorRegistrationRequest(ctx, signer, req)
@@ -1372,8 +1381,8 @@ func (v *validator) getAttSelection(key attSelectionKey) ([]byte, error) {
// This constructs a validator subscribed key, it's used to track
// which subnet has already been pending requested.
func validatorSubscribeKey(slot primitives.Slot, committeeID primitives.CommitteeIndex) [64]byte {
return bytesutil.ToBytes64(append(bytesutil.Bytes32(uint64(slot)), bytesutil.Bytes32(uint64(committeeID))...))
func validatorSubnetSubscriptionKey(slot primitives.Slot, committeeIndex primitives.CommitteeIndex) [64]byte {
return bytesutil.ToBytes64(append(bytesutil.Bytes32(uint64(slot)), bytesutil.Bytes32(uint64(committeeIndex))...))
}
// This tracks all validators' voting status.

68
validator/client/validator_test.go
View File

@@ -305,10 +305,10 @@ func TestWaitForChainStart_ReceiveErrorFromStream(t *testing.T) {
func TestCanonicalHeadSlot_FailedRPC(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
client := validatormock.NewMockBeaconChainClient(ctrl)
client := validatormock.NewMockChainClient(ctrl)
v := validator{
beaconClient: client,
genesisTime: 1,
chainClient: client,
genesisTime: 1,
}
client.EXPECT().GetChainHead(
gomock.Any(),
@@ -321,9 +321,9 @@ func TestCanonicalHeadSlot_FailedRPC(t *testing.T) {
func TestCanonicalHeadSlot_OK(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
client := validatormock.NewMockBeaconChainClient(ctrl)
client := validatormock.NewMockChainClient(ctrl)
v := validator{
beaconClient: client,
chainClient: client,
}
client.EXPECT().GetChainHead(
gomock.Any(),
@@ -339,22 +339,22 @@ func TestWaitMultipleActivation_LogsActivationEpochOK(t *testing.T) {
hook := logTest.NewGlobal()
ctrl := gomock.NewController(t)
defer ctrl.Finish()
validatorClient := validatormock.NewMockValidatorClient(ctrl)
beaconClient := validatormock.NewMockBeaconChainClient(ctrl)
prysmBeaconClient := validatormock.NewMockPrysmBeaconChainClient(ctrl)
client := validatormock.NewMockValidatorClient(ctrl)
chainClient := validatormock.NewMockChainClient(ctrl)
prysmChainClient := validatormock.NewMockPrysmChainClient(ctrl)
kp := randKeypair(t)
v := validator{
validatorClient: validatorClient,
keyManager: newMockKeymanager(t, kp),
beaconClient: beaconClient,
prysmBeaconClient: prysmBeaconClient,
validatorClient: client,
km: newMockKeymanager(t, kp),
chainClient: chainClient,
prysmChainClient: prysmChainClient,
}
resp := generateMockStatusResponse([][]byte{kp.pub[:]})
resp.Statuses[0].Status.Status = ethpb.ValidatorStatus_ACTIVE
clientStream := mock2.NewMockBeaconNodeValidator_WaitForActivationClient(ctrl)
validatorClient.EXPECT().WaitForActivation(
client.EXPECT().WaitForActivation(
gomock.Any(),
&ethpb.ValidatorActivationRequest{
PublicKeys: [][]byte{kp.pub[:]},
@@ -364,7 +364,7 @@ func TestWaitMultipleActivation_LogsActivationEpochOK(t *testing.T) {
resp,
nil,
)
prysmBeaconClient.EXPECT().GetValidatorCount(
prysmChainClient.EXPECT().GetValidatorCount(
gomock.Any(),
"head",
[]validatorType.Status{validatorType.Active},
@@ -465,7 +465,7 @@ func TestUpdateDuties_ReturnsError(t *testing.T) {
v := validator{
validatorClient: client,
keyManager: newMockKeymanager(t, randKeypair(t)),
km: newMockKeymanager(t, randKeypair(t)),
duties: &ethpb.DutiesResponse{
CurrentEpochDuties: []*ethpb.DutiesResponse_Duty{
{
@@ -505,7 +505,7 @@ func TestUpdateDuties_OK(t *testing.T) {
},
}
v := validator{
keyManager: newMockKeymanager(t, randKeypair(t)),
km: newMockKeymanager(t, randKeypair(t)),
validatorClient: client,
}
client.EXPECT().GetDuties(
@@ -549,9 +549,9 @@ func TestUpdateDuties_OK_FilterBlacklistedPublicKeys(t *testing.T) {
blacklistedPublicKeys[k] = true
}
v := validator{
keyManager: km,
validatorClient: client,
eipImportBlacklistedPublicKeys: blacklistedPublicKeys,
km: km,
validatorClient: client,
blacklistedPubkeys: blacklistedPublicKeys,
}
resp := &ethpb.DutiesResponse{
@@ -611,7 +611,7 @@ func TestUpdateDuties_AllValidatorsExited(t *testing.T) {
},
}
v := validator{
keyManager: newMockKeymanager(t, randKeypair(t)),
km: newMockKeymanager(t, randKeypair(t)),
validatorClient: client,
}
client.EXPECT().GetDuties(
@@ -654,7 +654,7 @@ func TestUpdateDuties_Distributed(t *testing.T) {
}
v := validator{
keyManager: newMockKeymanager(t, keys),
km: newMockKeymanager(t, keys),
validatorClient: client,
distributed: true,
}
@@ -997,7 +997,7 @@ func TestValidator_CheckDoppelGanger(t *testing.T) {
}
v := &validator{
validatorClient: client,
keyManager: km,
km: km,
db: db,
}
client.EXPECT().CheckDoppelGanger(
@@ -1038,7 +1038,7 @@ func TestValidator_CheckDoppelGanger(t *testing.T) {
}
v := &validator{
validatorClient: client,
keyManager: km,
km: km,
db: db,
}
client.EXPECT().CheckDoppelGanger(
@@ -1077,7 +1077,7 @@ func TestValidator_CheckDoppelGanger(t *testing.T) {
}
v := &validator{
validatorClient: client,
keyManager: km,
km: km,
db: db,
}
client.EXPECT().CheckDoppelGanger(
@@ -1122,7 +1122,7 @@ func TestValidator_CheckDoppelGanger(t *testing.T) {
}
v := &validator{
validatorClient: client,
keyManager: km,
km: km,
db: db,
}
client.EXPECT().CheckDoppelGanger(
@@ -1150,7 +1150,7 @@ func TestValidator_CheckDoppelGanger(t *testing.T) {
}
v := &validator{
validatorClient: client,
keyManager: km,
km: km,
db: db,
}
client.EXPECT().CheckDoppelGanger(
@@ -1366,7 +1366,7 @@ func TestValidator_WaitForKeymanagerInitialization_web3Signer(t *testing.T) {
db: db,
useWeb: false,
wallet: w,
Web3SignerConfig: &remoteweb3signer.SetupConfig{
web3SignerConfig: &remoteweb3signer.SetupConfig{
BaseEndpoint: "http://localhost:8545",
ProvidedPublicKeys: keys,
},
@@ -1391,10 +1391,10 @@ func TestValidator_WaitForKeymanagerInitialization_Web(t *testing.T) {
require.NoError(t, err)
walletChan := make(chan *wallet.Wallet, 1)
v := validator{
db: db,
useWeb: true,
walletInitializedFeed: &event.Feed{},
walletInitializedChannel: walletChan,
db: db,
useWeb: true,
walletInitializedFeed: &event.Feed{},
walletInitializedChan: walletChan,
}
wait := make(chan struct{})
go func() {
@@ -1992,7 +1992,7 @@ func TestValidator_PushSettings(t *testing.T) {
pubkeys, err := km.FetchValidatingPublicKeys(ctx)
require.NoError(t, err)
if tt.feeRecipientMap != nil {
feeRecipients, err := v.buildPrepProposerReqs(ctx, pubkeys)
feeRecipients, err := v.buildPrepProposerReqs(pubkeys)
require.NoError(t, err)
signedRegisterValidatorRequests := v.buildSignedRegReqs(ctx, pubkeys, km.Sign)
for _, recipient := range feeRecipients {
@@ -2148,7 +2148,7 @@ func TestValidator_buildPrepProposerReqs_WithoutDefaultConfig(t *testing.T) {
}
filteredKeys, err := v.filterAndCacheActiveKeys(ctx, pubkeys, 0)
require.NoError(t, err)
actual, err := v.buildPrepProposerReqs(ctx, filteredKeys)
actual, err := v.buildPrepProposerReqs(filteredKeys)
require.NoError(t, err)
assert.DeepEqual(t, expected, actual)
}
@@ -2316,7 +2316,7 @@ func TestValidator_buildPrepProposerReqs_WithDefaultConfig(t *testing.T) {
}
filteredKeys, err := v.filterAndCacheActiveKeys(ctx, pubkeys, 641)
require.NoError(t, err)
actual, err := v.buildPrepProposerReqs(ctx, filteredKeys)
actual, err := v.buildPrepProposerReqs(filteredKeys)
require.NoError(t, err)
assert.DeepEqual(t, expected, actual)
}

4
validator/client/wait_for_activation.go
View File

@@ -50,7 +50,7 @@ func (v *validator) WaitForActivation(ctx context.Context, accountsChangedChan c
func (v *validator) internalWaitForActivation(ctx context.Context, accountsChangedChan <-chan [][fieldparams.BLSPubkeyLength]byte) error {
ctx, span := trace.StartSpan(ctx, "validator.WaitForActivation")
defer span.End()
validatingKeys, err := v.keyManager.FetchValidatingPublicKeys(ctx)
validatingKeys, err := v.km.FetchValidatingPublicKeys(ctx)
if err != nil {
return errors.Wrap(err, msgCouldNotFetchKeys)
}
@@ -120,7 +120,7 @@ func (v *validator) internalWaitForActivation(ctx context.Context, accountsChang
// "-1" indicates that validator count endpoint is not supported by the beacon node.
var valCount int64 = -1
valCounts, err := v.prysmBeaconClient.GetValidatorCount(ctx, "head", []validator2.Status{validator2.Active})
valCounts, err := v.prysmChainClient.GetValidatorCount(ctx, "head", []validator2.Status{validator2.Active})
if err != nil && !errors.Is(err, iface.ErrNotSupported) {
return errors.Wrap(err, "could not get active validator count")
}

122
validator/client/wait_for_activation_test.go
View File

@@ -32,12 +32,12 @@ func TestWaitActivation_ContextCanceled(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
validatorClient := validatormock.NewMockValidatorClient(ctrl)
beaconClient := validatormock.NewMockBeaconChainClient(ctrl)
chainClient := validatormock.NewMockChainClient(ctrl)
kp := randKeypair(t)
v := validator{
validatorClient: validatorClient,
keyManager: newMockKeymanager(t, kp),
beaconClient: beaconClient,
km: newMockKeymanager(t, kp),
chainClient: chainClient,
}
clientStream := mock.NewMockBeaconNodeValidator_WaitForActivationClient(ctrl)
ctx, cancel := context.WithCancel(context.Background())
@@ -58,14 +58,14 @@ func TestWaitActivation_StreamSetupFails_AttemptsToReconnect(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
validatorClient := validatormock.NewMockValidatorClient(ctrl)
beaconClient := validatormock.NewMockBeaconChainClient(ctrl)
prysmBeaconClient := validatormock.NewMockPrysmBeaconChainClient(ctrl)
chainClient := validatormock.NewMockChainClient(ctrl)
prysmChainClient := validatormock.NewMockPrysmChainClient(ctrl)
kp := randKeypair(t)
v := validator{
validatorClient: validatorClient,
keyManager: newMockKeymanager(t, kp),
beaconClient: beaconClient,
prysmBeaconClient: prysmBeaconClient,
validatorClient: validatorClient,
km: newMockKeymanager(t, kp),
chainClient: chainClient,
prysmChainClient: prysmChainClient,
}
clientStream := mock.NewMockBeaconNodeValidator_WaitForActivationClient(ctrl)
validatorClient.EXPECT().WaitForActivation(
@@ -74,7 +74,7 @@ func TestWaitActivation_StreamSetupFails_AttemptsToReconnect(t *testing.T) {
PublicKeys: [][]byte{kp.pub[:]},
},
).Return(clientStream, errors.New("failed stream")).Return(clientStream, nil)
prysmBeaconClient.EXPECT().GetValidatorCount(
prysmChainClient.EXPECT().GetValidatorCount(
gomock.Any(),
"head",
[]validatorType.Status{validatorType.Active},
@@ -89,14 +89,14 @@ func TestWaitForActivation_ReceiveErrorFromStream_AttemptsReconnection(t *testin
ctrl := gomock.NewController(t)
defer ctrl.Finish()
validatorClient := validatormock.NewMockValidatorClient(ctrl)
beaconClient := validatormock.NewMockBeaconChainClient(ctrl)
prysmBeaconClient := validatormock.NewMockPrysmBeaconChainClient(ctrl)
chainClient := validatormock.NewMockChainClient(ctrl)
prysmChainClient := validatormock.NewMockPrysmChainClient(ctrl)
kp := randKeypair(t)
v := validator{
validatorClient: validatorClient,
keyManager: newMockKeymanager(t, kp),
beaconClient: beaconClient,
prysmBeaconClient: prysmBeaconClient,
validatorClient: validatorClient,
km: newMockKeymanager(t, kp),
chainClient: chainClient,
prysmChainClient: prysmChainClient,
}
clientStream := mock.NewMockBeaconNodeValidator_WaitForActivationClient(ctrl)
validatorClient.EXPECT().WaitForActivation(
@@ -105,7 +105,7 @@ func TestWaitForActivation_ReceiveErrorFromStream_AttemptsReconnection(t *testin
PublicKeys: [][]byte{kp.pub[:]},
},
).Return(clientStream, nil)
prysmBeaconClient.EXPECT().GetValidatorCount(
prysmChainClient.EXPECT().GetValidatorCount(
gomock.Any(),
"head",
[]validatorType.Status{validatorType.Active},
@@ -125,15 +125,15 @@ func TestWaitActivation_LogsActivationEpochOK(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
validatorClient := validatormock.NewMockValidatorClient(ctrl)
beaconClient := validatormock.NewMockBeaconChainClient(ctrl)
prysmBeaconClient := validatormock.NewMockPrysmBeaconChainClient(ctrl)
chainClient := validatormock.NewMockChainClient(ctrl)
prysmChainClient := validatormock.NewMockPrysmChainClient(ctrl)
kp := randKeypair(t)
v := validator{
validatorClient: validatorClient,
keyManager: newMockKeymanager(t, kp),
genesisTime: 1,
beaconClient: beaconClient,
prysmBeaconClient: prysmBeaconClient,
validatorClient: validatorClient,
km: newMockKeymanager(t, kp),
genesisTime: 1,
chainClient: chainClient,
prysmChainClient: prysmChainClient,
}
resp := generateMockStatusResponse([][]byte{kp.pub[:]})
resp.Statuses[0].Status.Status = ethpb.ValidatorStatus_ACTIVE
@@ -144,7 +144,7 @@ func TestWaitActivation_LogsActivationEpochOK(t *testing.T) {
PublicKeys: [][]byte{kp.pub[:]},
},
).Return(clientStream, nil)
prysmBeaconClient.EXPECT().GetValidatorCount(
prysmChainClient.EXPECT().GetValidatorCount(
gomock.Any(),
"head",
[]validatorType.Status{validatorType.Active},
@@ -161,14 +161,14 @@ func TestWaitForActivation_Exiting(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
validatorClient := validatormock.NewMockValidatorClient(ctrl)
beaconClient := validatormock.NewMockBeaconChainClient(ctrl)
prysmBeaconClient := validatormock.NewMockPrysmBeaconChainClient(ctrl)
chainClient := validatormock.NewMockChainClient(ctrl)
prysmChainClient := validatormock.NewMockPrysmChainClient(ctrl)
kp := randKeypair(t)
v := validator{
validatorClient: validatorClient,
keyManager: newMockKeymanager(t, kp),
beaconClient: beaconClient,
prysmBeaconClient: prysmBeaconClient,
validatorClient: validatorClient,
km: newMockKeymanager(t, kp),
chainClient: chainClient,
prysmChainClient: prysmChainClient,
}
resp := generateMockStatusResponse([][]byte{kp.pub[:]})
resp.Statuses[0].Status.Status = ethpb.ValidatorStatus_EXITING
@@ -179,7 +179,7 @@ func TestWaitForActivation_Exiting(t *testing.T) {
PublicKeys: [][]byte{kp.pub[:]},
},
).Return(clientStream, nil)
prysmBeaconClient.EXPECT().GetValidatorCount(
prysmChainClient.EXPECT().GetValidatorCount(
gomock.Any(),
"head",
[]validatorType.Status{validatorType.Active},
@@ -201,17 +201,17 @@ func TestWaitForActivation_RefetchKeys(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
validatorClient := validatormock.NewMockValidatorClient(ctrl)
beaconClient := validatormock.NewMockBeaconChainClient(ctrl)
prysmBeaconClient := validatormock.NewMockPrysmBeaconChainClient(ctrl)
chainClient := validatormock.NewMockChainClient(ctrl)
prysmChainClient := validatormock.NewMockPrysmChainClient(ctrl)
kp := randKeypair(t)
km := newMockKeymanager(t)
v := validator{
validatorClient: validatorClient,
keyManager: km,
beaconClient: beaconClient,
prysmBeaconClient: prysmBeaconClient,
validatorClient: validatorClient,
km: km,
chainClient: chainClient,
prysmChainClient: prysmChainClient,
}
resp := generateMockStatusResponse([][]byte{kp.pub[:]})
resp.Statuses[0].Status.Status = ethpb.ValidatorStatus_ACTIVE
@@ -222,7 +222,7 @@ func TestWaitForActivation_RefetchKeys(t *testing.T) {
PublicKeys: [][]byte{kp.pub[:]},
},
).Return(clientStream, nil)
prysmBeaconClient.EXPECT().GetValidatorCount(
prysmChainClient.EXPECT().GetValidatorCount(
gomock.Any(),
"head",
[]validatorType.Status{validatorType.Active},
@@ -258,13 +258,13 @@ func TestWaitForActivation_AccountsChanged(t *testing.T) {
active := randKeypair(t)
km := newMockKeymanager(t, inactive)
validatorClient := validatormock.NewMockValidatorClient(ctrl)
beaconClient := validatormock.NewMockBeaconChainClient(ctrl)
prysmBeaconClient := validatormock.NewMockPrysmBeaconChainClient(ctrl)
chainClient := validatormock.NewMockChainClient(ctrl)
prysmChainClient := validatormock.NewMockPrysmChainClient(ctrl)
v := validator{
validatorClient: validatorClient,
keyManager: km,
beaconClient: beaconClient,
prysmBeaconClient: prysmBeaconClient,
validatorClient: validatorClient,
km: km,
chainClient: chainClient,
prysmChainClient: prysmChainClient,
}
inactiveResp := generateMockStatusResponse([][]byte{inactive.pub[:]})
inactiveResp.Statuses[0].Status.Status = ethpb.ValidatorStatus_UNKNOWN_STATUS
@@ -279,7 +279,7 @@ func TestWaitForActivation_AccountsChanged(t *testing.T) {
time.Sleep(time.Second * 2)
return inactiveClientStream, nil
})
prysmBeaconClient.EXPECT().GetValidatorCount(
prysmChainClient.EXPECT().GetValidatorCount(
gomock.Any(),
"head",
[]validatorType.Status{validatorType.Active},
@@ -348,14 +348,14 @@ func TestWaitForActivation_AccountsChanged(t *testing.T) {
err = km.RecoverAccountsFromMnemonic(ctx, constant.TestMnemonic, derived.DefaultMnemonicLanguage, "", 1)
require.NoError(t, err)
validatorClient := validatormock.NewMockValidatorClient(ctrl)
beaconClient := validatormock.NewMockBeaconChainClient(ctrl)
prysmBeaconClient := validatormock.NewMockPrysmBeaconChainClient(ctrl)
chainClient := validatormock.NewMockChainClient(ctrl)
prysmChainClient := validatormock.NewMockPrysmChainClient(ctrl)
v := validator{
validatorClient: validatorClient,
keyManager: km,
genesisTime: 1,
beaconClient: beaconClient,
prysmBeaconClient: prysmBeaconClient,
validatorClient: validatorClient,
km: km,
genesisTime: 1,
chainClient: chainClient,
prysmChainClient: prysmChainClient,
}
inactiveResp := generateMockStatusResponse([][]byte{inactivePubKey[:]})
@@ -371,7 +371,7 @@ func TestWaitForActivation_AccountsChanged(t *testing.T) {
time.Sleep(time.Second * 2)
return inactiveClientStream, nil
})
prysmBeaconClient.EXPECT().GetValidatorCount(
prysmChainClient.EXPECT().GetValidatorCount(
gomock.Any(),
"head",
[]validatorType.Status{validatorType.Active},
@@ -415,15 +415,15 @@ func TestWaitActivation_NotAllValidatorsActivatedOK(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
validatorClient := validatormock.NewMockValidatorClient(ctrl)
beaconClient := validatormock.NewMockBeaconChainClient(ctrl)
prysmBeaconClient := validatormock.NewMockPrysmBeaconChainClient(ctrl)
chainClient := validatormock.NewMockChainClient(ctrl)
prysmChainClient := validatormock.NewMockPrysmChainClient(ctrl)
kp := randKeypair(t)
v := validator{
validatorClient: validatorClient,
keyManager: newMockKeymanager(t, kp),
beaconClient: beaconClient,
prysmBeaconClient: prysmBeaconClient,
validatorClient: validatorClient,
km: newMockKeymanager(t, kp),
chainClient: chainClient,
prysmChainClient: prysmChainClient,
}
resp := generateMockStatusResponse([][]byte{kp.pub[:]})
resp.Statuses[0].Status.Status = ethpb.ValidatorStatus_ACTIVE
@@ -432,7 +432,7 @@ func TestWaitActivation_NotAllValidatorsActivatedOK(t *testing.T) {
gomock.Any(),
gomock.Any(),
).Return(clientStream, nil)
prysmBeaconClient.EXPECT().GetValidatorCount(
prysmChainClient.EXPECT().GetValidatorCount(
gomock.Any(),
"head",
[]validatorType.Status{validatorType.Active},

160
validator/node/node.go
View File

@@ -63,15 +63,15 @@ import (
// ValidatorClient defines an instance of an Ethereum validator that manages
// the entire lifecycle of services attached to it participating in proof of stake.
type ValidatorClient struct {
cliCtx *cli.Context
ctx context.Context
cancel context.CancelFunc
db iface.ValidatorDB
services *runtime.ServiceRegistry // Lifecycle and service store.
lock sync.RWMutex
wallet *wallet.Wallet
walletInitialized *event.Feed
stop chan struct{} // Channel to wait for termination notifications.
cliCtx *cli.Context
ctx context.Context
cancel context.CancelFunc
db iface.ValidatorDB
services *runtime.ServiceRegistry // Lifecycle and service store.
lock sync.RWMutex
wallet *wallet.Wallet
walletInitializedFeed *event.Feed
stop chan struct{} // Channel to wait for termination notifications.
}
// NewValidatorClient creates a new instance of the Prysm validator client.
@@ -100,12 +100,12 @@ func NewValidatorClient(cliCtx *cli.Context) (*ValidatorClient, error) {
registry := runtime.NewServiceRegistry()
ctx, cancel := context.WithCancel(cliCtx.Context)
validatorClient := &ValidatorClient{
cliCtx: cliCtx,
ctx: ctx,
cancel: cancel,
services: registry,
walletInitialized: new(event.Feed),
stop: make(chan struct{}),
cliCtx: cliCtx,
ctx: ctx,
cancel: cancel,
services: registry,
walletInitializedFeed: new(event.Feed),
stop: make(chan struct{}),
}
if err := features.ConfigureValidator(cliCtx); err != nil {
@@ -485,23 +485,13 @@ func (c *ValidatorClient) registerPrometheusService(cliCtx *cli.Context) error {
func (c *ValidatorClient) registerValidatorService(cliCtx *cli.Context) error {
var (
endpoint string = c.cliCtx.String(flags.BeaconRPCProviderFlag.Name)
dataDir string = c.cliCtx.String(cmd.DataDirFlag.Name)
logValidatorBalances bool = !c.cliCtx.Bool(flags.DisablePenaltyRewardLogFlag.Name)
emitAccountMetrics bool = !c.cliCtx.Bool(flags.DisableAccountMetricsFlag.Name)
cert string = c.cliCtx.String(flags.CertFlag.Name)
graffiti string = c.cliCtx.String(flags.GraffitiFlag.Name)
maxCallRecvMsgSize int = c.cliCtx.Int(cmd.GrpcMaxCallRecvMsgSizeFlag.Name)
grpcRetries uint = c.cliCtx.Uint(flags.GrpcRetriesFlag.Name)
grpcRetryDelay time.Duration = c.cliCtx.Duration(flags.GrpcRetryDelayFlag.Name)
interopKeysConfig *local.InteropKeymanagerConfig
err error
interopKmConfig *local.InteropKeymanagerConfig
err error
)
// Configure interop.
if c.cliCtx.IsSet(flags.InteropNumValidators.Name) {
interopKeysConfig = &local.InteropKeymanagerConfig{
interopKmConfig = &local.InteropKeymanagerConfig{
Offset: cliCtx.Uint64(flags.InteropStartIndex.Name),
NumValidatorKeys: cliCtx.Uint64(flags.InteropNumValidators.Name),
}
@@ -529,28 +519,27 @@ func (c *ValidatorClient) registerValidatorService(cliCtx *cli.Context) error {
}
validatorService, err := client.NewValidatorService(c.cliCtx.Context, &client.Config{
Endpoint: endpoint,
DataDir: dataDir,
LogValidatorBalances: logValidatorBalances,
EmitAccountMetrics: emitAccountMetrics,
CertFlag: cert,
GraffitiFlag: g.ParseHexGraffiti(graffiti),
GrpcMaxCallRecvMsgSizeFlag: maxCallRecvMsgSize,
GrpcRetriesFlag: grpcRetries,
GrpcRetryDelay: grpcRetryDelay,
GrpcHeadersFlag: c.cliCtx.String(flags.GrpcHeadersFlag.Name),
ValDB: c.db,
UseWeb: c.cliCtx.Bool(flags.EnableWebFlag.Name),
InteropKeysConfig: interopKeysConfig,
Wallet: c.wallet,
WalletInitializedFeed: c.walletInitialized,
GraffitiStruct: graffitiStruct,
Web3SignerConfig: web3signerConfig,
ProposerSettings: ps,
BeaconApiTimeout: time.Second * 30,
BeaconApiEndpoint: c.cliCtx.String(flags.BeaconRESTApiProviderFlag.Name),
ValidatorsRegBatchSize: c.cliCtx.Int(flags.ValidatorsRegistrationBatchSizeFlag.Name),
Distributed: c.cliCtx.Bool(flags.EnableDistributed.Name),
DB: c.db,
Wallet: c.wallet,
WalletInitializedFeed: c.walletInitializedFeed,
GRPCMaxCallRecvMsgSize: c.cliCtx.Int(cmd.GrpcMaxCallRecvMsgSizeFlag.Name),
GRPCRetries: c.cliCtx.Uint(flags.GrpcRetriesFlag.Name),
GRPCRetryDelay: c.cliCtx.Duration(flags.GrpcRetryDelayFlag.Name),
GRPCHeaders: strings.Split(c.cliCtx.String(flags.GrpcHeadersFlag.Name), ","),
BeaconNodeGRPCEndpoint: c.cliCtx.String(flags.BeaconRPCProviderFlag.Name),
BeaconNodeCert: c.cliCtx.String(flags.CertFlag.Name),
BeaconApiEndpoint: c.cliCtx.String(flags.BeaconRESTApiProviderFlag.Name),
BeaconApiTimeout: time.Second * 30,
Graffiti: g.ParseHexGraffiti(c.cliCtx.String(flags.GraffitiFlag.Name)),
GraffitiStruct: graffitiStruct,
InteropKmConfig: interopKmConfig,
Web3SignerConfig: web3signerConfig,
ProposerSettings: ps,
ValidatorsRegBatchSize: c.cliCtx.Int(flags.ValidatorsRegistrationBatchSizeFlag.Name),
UseWeb: c.cliCtx.Bool(flags.EnableWebFlag.Name),
LogValidatorPerformance: !c.cliCtx.Bool(flags.DisablePenaltyRewardLogFlag.Name),
EmitAccountMetrics: !c.cliCtx.Bool(flags.DisableAccountMetricsFlag.Name),
Distributed: c.cliCtx.Bool(flags.EnableDistributed.Name),
})
if err != nil {
return errors.Wrap(err, "could not initialize validator service")
@@ -625,58 +614,27 @@ func (c *ValidatorClient) registerRPCService(router *mux.Router) error {
if err := c.services.FetchService(&vs); err != nil {
return err
}
validatorGatewayHost := c.cliCtx.String(flags.GRPCGatewayHost.Name)
validatorGatewayPort := c.cliCtx.Int(flags.GRPCGatewayPort.Name)
validatorMonitoringHost := c.cliCtx.String(cmd.MonitoringHostFlag.Name)
validatorMonitoringPort := c.cliCtx.Int(flags.MonitoringPortFlag.Name)
rpcHost := c.cliCtx.String(flags.RPCHost.Name)
rpcPort := c.cliCtx.Int(flags.RPCPort.Name)
nodeGatewayEndpoint := c.cliCtx.String(flags.BeaconRPCGatewayProviderFlag.Name)
beaconClientEndpoint := c.cliCtx.String(flags.BeaconRPCProviderFlag.Name)
maxCallRecvMsgSize := c.cliCtx.Int(cmd.GrpcMaxCallRecvMsgSizeFlag.Name)
grpcRetries := c.cliCtx.Uint(flags.GrpcRetriesFlag.Name)
grpcRetryDelay := c.cliCtx.Duration(flags.GrpcRetryDelayFlag.Name)
walletDir := c.cliCtx.String(flags.WalletDirFlag.Name)
grpcHeaders := c.cliCtx.String(flags.GrpcHeadersFlag.Name)
clientCert := c.cliCtx.String(flags.CertFlag.Name)
authTokenPath := c.cliCtx.String(flags.AuthTokenPathFlag.Name)
// if no auth token path flag was passed try to set a default value
if authTokenPath == "" {
authTokenPath = flags.AuthTokenPathFlag.Value
// if a wallet dir is passed without an auth token then override the default with the wallet dir
if walletDir != "" {
authTokenPath = filepath.Join(walletDir, api.AuthTokenFileName)
}
}
server := rpc.NewServer(c.cliCtx.Context, &rpc.Config{
ValDB: c.db,
Host: rpcHost,
Port: fmt.Sprintf("%d", rpcPort),
WalletInitializedFeed: c.walletInitialized,
ValidatorService: vs,
SyncChecker: vs,
GenesisFetcher: vs,
NodeGatewayEndpoint: nodeGatewayEndpoint,
AuthTokenPath: authTokenPath,
WalletDir: walletDir,
Wallet: c.wallet,
ValidatorGatewayHost: validatorGatewayHost,
ValidatorGatewayPort: validatorGatewayPort,
ValidatorMonitoringHost: validatorMonitoringHost,
ValidatorMonitoringPort: validatorMonitoringPort,
BeaconClientEndpoint: beaconClientEndpoint,
ClientMaxCallRecvMsgSize: maxCallRecvMsgSize,
ClientGrpcRetries: grpcRetries,
ClientGrpcRetryDelay: grpcRetryDelay,
ClientGrpcHeaders: strings.Split(grpcHeaders, ","),
ClientWithCert: clientCert,
BeaconApiTimeout: time.Second * 30,
BeaconApiEndpoint: c.cliCtx.String(flags.BeaconRESTApiProviderFlag.Name),
Router: router,
s := rpc.NewServer(c.cliCtx.Context, &rpc.Config{
Host: c.cliCtx.String(flags.RPCHost.Name),
Port: fmt.Sprintf("%d", c.cliCtx.Int(flags.RPCPort.Name)),
GRPCGatewayHost: c.cliCtx.String(flags.GRPCGatewayHost.Name),
GRPCGatewayPort: c.cliCtx.Int(flags.GRPCGatewayPort.Name),
GRPCMaxCallRecvMsgSize: c.cliCtx.Int(cmd.GrpcMaxCallRecvMsgSizeFlag.Name),
GRPCRetries: c.cliCtx.Uint(flags.GrpcRetriesFlag.Name),
GRPCRetryDelay: c.cliCtx.Duration(flags.GrpcRetryDelayFlag.Name),
GRPCHeaders: strings.Split(c.cliCtx.String(flags.GrpcHeadersFlag.Name), ","),
BeaconNodeGRPCEndpoint: c.cliCtx.String(flags.BeaconRPCProviderFlag.Name),
BeaconApiEndpoint: c.cliCtx.String(flags.BeaconRPCGatewayProviderFlag.Name),
BeaconApiTimeout: time.Second * 30,
BeaconNodeCert: c.cliCtx.String(flags.CertFlag.Name),
DB: c.db,
Wallet: c.wallet,
WalletDir: c.cliCtx.String(flags.WalletDirFlag.Name),
WalletInitializedFeed: c.walletInitializedFeed,
ValidatorService: vs,
Router: router,
})
return c.services.RegisterService(server)
return c.services.RegisterService(s)
}
func (c *ValidatorClient) registerRPCGatewayService(router *mux.Router) error {

1
validator/rpc/BUILD.bazel
View File

@@ -81,7 +81,6 @@ go_library(
"@io_opencensus_go//trace:go_default_library",
"@org_golang_google_grpc//:go_default_library",
"@org_golang_google_grpc//codes:go_default_library",
"@org_golang_google_grpc//credentials:go_default_library",
"@org_golang_google_grpc//metadata:go_default_library",
"@org_golang_google_grpc//reflection:go_default_library",
"@org_golang_google_grpc//status:go_default_library",

2
validator/rpc/auth_token.go
View File

@@ -104,7 +104,7 @@ func (s *Server) refreshAuthTokenFromFileChanges(ctx context.Context, authTokenP
log.WithError(err).Errorf("Could not watch for file changes for: %s", authTokenPath)
continue
}
validatorWebAddr := fmt.Sprintf("%s:%d", s.validatorGatewayHost, s.validatorGatewayPort)
validatorWebAddr := fmt.Sprintf("%s:%d", s.grpcGatewayHost, s.grpcGatewayPort)
logValidatorWebAuth(validatorWebAddr, s.authToken, authTokenPath)
case err := <-watcher.Errors:
log.WithError(err).Errorf("Could not watch for file changes for: %s", authTokenPath)

22
validator/rpc/beacon.go
View File

@@ -27,26 +27,26 @@ func (s *Server) registerBeaconClient() error {
grpcretry.StreamClientInterceptor(),
))
dialOpts := client.ConstructDialOptions(
s.clientMaxCallRecvMsgSize,
s.clientWithCert,
s.clientGrpcRetries,
s.clientGrpcRetryDelay,
s.grpcMaxCallRecvMsgSize,
s.beaconNodeCert,
s.grpcRetries,
s.grpcRetryDelay,
streamInterceptor,
)
if dialOpts == nil {
return errors.New("no dial options for beacon chain gRPC client")
}
s.ctx = grpcutil.AppendHeaders(s.ctx, s.clientGrpcHeaders)
s.ctx = grpcutil.AppendHeaders(s.ctx, s.grpcHeaders)
grpcConn, err := grpc.DialContext(s.ctx, s.beaconClientEndpoint, dialOpts...)
grpcConn, err := grpc.DialContext(s.ctx, s.beaconNodeEndpoint, dialOpts...)
if err != nil {
return errors.Wrapf(err, "could not dial endpoint: %s", s.beaconClientEndpoint)
return errors.Wrapf(err, "could not dial endpoint: %s", s.beaconNodeEndpoint)
}
if s.clientWithCert != "" {
if s.beaconNodeCert != "" {
log.Info("Established secure gRPC connection")
}
s.beaconNodeHealthClient = ethpb.NewHealthClient(grpcConn)
s.healthClient = ethpb.NewHealthClient(grpcConn)
conn := validatorHelpers.NewNodeConnection(
grpcConn,
@@ -56,8 +56,8 @@ func (s *Server) registerBeaconClient() error {
restHandler := beaconApi.NewBeaconApiJsonRestHandler(http.Client{Timeout: s.beaconApiTimeout}, s.beaconApiEndpoint)
s.beaconChainClient = beaconChainClientFactory.NewBeaconChainClient(conn, restHandler)
s.beaconNodeClient = nodeClientFactory.NewNodeClient(conn, restHandler)
s.chainClient = beaconChainClientFactory.NewChainClient(conn, restHandler)
s.nodeClient = nodeClientFactory.NewNodeClient(conn, restHandler)
s.beaconNodeValidatorClient = validatorClientFactory.NewValidatorClient(conn, restHandler)
return nil

4
validator/rpc/beacon_test.go
View File

@@ -11,8 +11,8 @@ import (
func TestGrpcHeaders(t *testing.T) {
s := &Server{
ctx: context.Background(),
clientGrpcHeaders: []string{"first=value1", "second=value2"},
ctx: context.Background(),
grpcHeaders: []string{"first=value1", "second=value2"},
}
err := s.registerBeaconClient()
require.NoError(t, err)

2
validator/rpc/handlers_accounts.go
View File

@@ -258,7 +258,7 @@ func (s *Server) VoluntaryExit(w http.ResponseWriter, r *http.Request) {
}
cfg := accounts.PerformExitCfg{
ValidatorClient: s.beaconNodeValidatorClient,
NodeClient: s.beaconNodeClient,
NodeClient: s.nodeClient,
Keymanager: km,
RawPubKeys: pubKeys,
FormattedPubKeys: req.PublicKeys,

2
validator/rpc/handlers_accounts_test.go
View File

@@ -292,7 +292,7 @@ func TestServer_VoluntaryExit(t *testing.T) {
s := &Server{
walletInitialized: true,
wallet: w,
beaconNodeClient: mockNodeClient,
nodeClient: mockNodeClient,
beaconNodeValidatorClient: mockValidatorClient,
validatorService: vs,
}

18
validator/rpc/handlers_beacon.go
View File

@@ -25,30 +25,30 @@ import (
func (s *Server) GetBeaconStatus(w http.ResponseWriter, r *http.Request) {
ctx, span := trace.StartSpan(r.Context(), "validator.web.beacon.GetBeaconStatus")
defer span.End()
syncStatus, err := s.beaconNodeClient.GetSyncStatus(ctx, &emptypb.Empty{})
syncStatus, err := s.nodeClient.GetSyncStatus(ctx, &emptypb.Empty{})
if err != nil {
log.WithError(err).Error("beacon node call to get sync status failed")
httputil.WriteJson(w, &BeaconStatusResponse{
BeaconNodeEndpoint: s.nodeGatewayEndpoint,
BeaconNodeEndpoint: s.beaconNodeEndpoint,
Connected: false,
Syncing: false,
})
return
}
genesis, err := s.beaconNodeClient.GetGenesis(ctx, &emptypb.Empty{})
genesis, err := s.nodeClient.GetGenesis(ctx, &emptypb.Empty{})
if err != nil {
httputil.HandleError(w, errors.Wrap(err, "GetGenesis call failed").Error(), http.StatusInternalServerError)
return
}
genesisTime := uint64(time.Unix(genesis.GenesisTime.Seconds, 0).Unix())
address := genesis.DepositContractAddress
chainHead, err := s.beaconChainClient.GetChainHead(ctx, &emptypb.Empty{})
chainHead, err := s.chainClient.GetChainHead(ctx, &emptypb.Empty{})
if err != nil {
httputil.HandleError(w, errors.Wrap(err, "GetChainHead").Error(), http.StatusInternalServerError)
return
}
httputil.WriteJson(w, &BeaconStatusResponse{
BeaconNodeEndpoint: s.beaconClientEndpoint,
BeaconNodeEndpoint: s.beaconNodeEndpoint,
Connected: true,
Syncing: syncStatus.Syncing,
GenesisTime: fmt.Sprintf("%d", genesisTime),
@@ -85,7 +85,7 @@ func (s *Server) GetValidatorPerformance(w http.ResponseWriter, r *http.Request)
req := &ethpb.ValidatorPerformanceRequest{
PublicKeys: pubkeys,
}
validatorPerformance, err := s.beaconChainClient.GetValidatorPerformance(ctx, req)
validatorPerformance, err := s.chainClient.GetValidatorPerformance(ctx, req)
if err != nil {
httputil.HandleError(w, errors.Wrap(err, "GetValidatorPerformance call failed").Error(), http.StatusInternalServerError)
return
@@ -133,7 +133,7 @@ func (s *Server) GetValidatorBalances(w http.ResponseWriter, r *http.Request) {
PageSize: int32(ps),
PageToken: pageToken,
}
listValidatorBalances, err := s.beaconChainClient.ListValidatorBalances(ctx, req)
listValidatorBalances, err := s.chainClient.ListValidatorBalances(ctx, req)
if err != nil {
httputil.HandleError(w, errors.Wrap(err, "ListValidatorBalances call failed").Error(), http.StatusInternalServerError)
return
@@ -187,7 +187,7 @@ func (s *Server) GetValidators(w http.ResponseWriter, r *http.Request) {
PageSize: int32(ps),
PageToken: pageToken,
}
validators, err := s.beaconChainClient.ListValidators(ctx, req)
validators, err := s.chainClient.ListValidators(ctx, req)
if err != nil {
httputil.HandleError(w, errors.Wrap(err, "ListValidators call failed").Error(), http.StatusInternalServerError)
return
@@ -204,7 +204,7 @@ func (s *Server) GetValidators(w http.ResponseWriter, r *http.Request) {
func (s *Server) GetPeers(w http.ResponseWriter, r *http.Request) {
ctx, span := trace.StartSpan(r.Context(), "validator.web.beacon.GetPeers")
defer span.End()
peers, err := s.beaconNodeClient.ListPeers(ctx, &emptypb.Empty{})
peers, err := s.nodeClient.ListPeers(ctx, &emptypb.Empty{})
if err != nil {
httputil.HandleError(w, errors.Wrap(err, "ListPeers call failed").Error(), http.StatusInternalServerError)
return

14
validator/rpc/handlers_beacon_test.go
View File

@@ -27,7 +27,7 @@ func TestGetBeaconStatus_NotConnected(t *testing.T) {
gomock.Any(),
).Return(nil /*response*/, errors.New("uh oh"))
srv := &Server{
beaconNodeClient: nodeClient,
nodeClient: nodeClient,
}
req := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/v2/validator/beacon/status"), nil)
wr := httptest.NewRecorder()
@@ -47,7 +47,7 @@ func TestGetBeaconStatus_NotConnected(t *testing.T) {
func TestGetBeaconStatus_OK(t *testing.T) {
ctrl := gomock.NewController(t)
nodeClient := validatormock.NewMockNodeClient(ctrl)
beaconChainClient := validatormock.NewMockBeaconChainClient(ctrl)
chainClient := validatormock.NewMockChainClient(ctrl)
nodeClient.EXPECT().GetSyncStatus(
gomock.Any(), // ctx
gomock.Any(),
@@ -60,15 +60,15 @@ func TestGetBeaconStatus_OK(t *testing.T) {
GenesisTime: timeStamp,
DepositContractAddress: []byte("hello"),
}, nil)
beaconChainClient.EXPECT().GetChainHead(
chainClient.EXPECT().GetChainHead(
gomock.Any(), // ctx
gomock.Any(),
).Return(&ethpb.ChainHead{
HeadEpoch: 1,
}, nil)
srv := &Server{
beaconNodeClient: nodeClient,
beaconChainClient: beaconChainClient,
nodeClient: nodeClient,
chainClient: chainClient,
}
req := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/v2/validator/beacon/status"), nil)
@@ -228,7 +228,7 @@ func TestServer_GetValidators(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
ctrl := gomock.NewController(t)
beaconChainClient := validatormock.NewMockBeaconChainClient(ctrl)
beaconChainClient := validatormock.NewMockChainClient(ctrl)
if tt.wantErr == "" {
beaconChainClient.EXPECT().ListValidators(
gomock.Any(), // ctx
@@ -236,7 +236,7 @@ func TestServer_GetValidators(t *testing.T) {
).Return(tt.chainResp, nil)
}
s := &Server{
beaconChainClient: beaconChainClient,
chainClient: beaconChainClient,
}
req := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/v2/validator/beacon/validators?%s", tt.query), http.NoBody)
wr := httptest.NewRecorder()

10
validator/rpc/handlers_health.go
View File

@@ -18,7 +18,7 @@ func (s *Server) GetVersion(w http.ResponseWriter, r *http.Request) {
ctx, span := trace.StartSpan(r.Context(), "validator.web.health.GetVersion")
defer span.End()
beacon, err := s.beaconNodeClient.GetVersion(ctx, &emptypb.Empty{})
beacon, err := s.nodeClient.GetVersion(ctx, &emptypb.Empty{})
if err != nil {
httputil.HandleError(w, err.Error(), http.StatusInternalServerError)
return
@@ -51,7 +51,7 @@ func (s *Server) StreamBeaconLogs(w http.ResponseWriter, r *http.Request) {
return
}
// TODO: StreamBeaconLogs grpc will need to be replaced in the future
client, err := s.beaconNodeHealthClient.StreamBeaconLogs(ctx, &emptypb.Empty{})
client, err := s.healthClient.StreamBeaconLogs(ctx, &emptypb.Empty{})
if err != nil {
httputil.HandleError(w, err.Error(), http.StatusInternalServerError)
return
@@ -102,8 +102,8 @@ func (s *Server) StreamValidatorLogs(w http.ResponseWriter, r *http.Request) {
return
}
ch := make(chan []byte, s.streamLogsBufferSize)
sub := s.logsStreamer.LogsFeed().Subscribe(ch)
ch := make(chan []byte, s.logStreamerBufferSize)
sub := s.logStreamer.LogsFeed().Subscribe(ch)
defer func() {
sub.Unsubscribe()
close(ch)
@@ -113,7 +113,7 @@ func (s *Server) StreamValidatorLogs(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Connection", api.KeepAlive)
recentLogs := s.logsStreamer.GetLastFewLogs()
recentLogs := s.logStreamer.GetLastFewLogs()
logStrings := make([]string, len(recentLogs))
for i, l := range recentLogs {
logStrings[i] = string(l)

14
validator/rpc/handlers_health_test.go
View File

@@ -73,8 +73,8 @@ func TestStreamBeaconLogs(t *testing.T) {
// Setting up the mock in the server struct
s := Server{
ctx: context.Background(),
beaconNodeHealthClient: mockClient,
ctx: context.Background(),
healthClient: mockClient,
}
// Create a mock ResponseWriter and Request
@@ -119,9 +119,9 @@ func TestStreamValidatorLogs(t *testing.T) {
logStreamer := mock.NewMockStreamer(mockLogs)
// Setting up the mock in the server struct
s := Server{
ctx: ctx,
logsStreamer: logStreamer,
streamLogsBufferSize: 100,
ctx: ctx,
logStreamer: logStreamer,
logStreamerBufferSize: 100,
}
w := &flushableResponseRecorder{
@@ -170,8 +170,8 @@ func TestServer_GetVersion(t *testing.T) {
ctx := context.Background()
mockNodeClient := validatormock.NewMockNodeClient(ctrl)
s := Server{
ctx: ctx,
beaconNodeClient: mockNodeClient,
ctx: ctx,
nodeClient: mockNodeClient,
}
mockNodeClient.EXPECT().GetVersion(gomock.Any(), gomock.Any()).Return(&eth.Version{
Version: "4.10.1",

12
validator/rpc/handlers_keymanager.go
View File

@@ -133,7 +133,7 @@ func (s *Server) ImportKeystores(w http.ResponseWriter, r *http.Request) {
keystores[i] = k
}
if req.SlashingProtection != "" {
if s.valDB == nil || s.valDB.ImportStandardProtectionJSON(ctx, bytes.NewBufferString(req.SlashingProtection)) != nil {
if s.db == nil || s.db.ImportStandardProtectionJSON(ctx, bytes.NewBufferString(req.SlashingProtection)) != nil {
statuses := make([]*keymanager.KeyStatus, len(req.Keystores))
for i := 0; i < len(req.Keystores); i++ {
statuses[i] = &keymanager.KeyStatus{
@@ -285,11 +285,11 @@ func (s *Server) transformDeletedKeysStatuses(
// Gets a map of all public keys in the database, useful for O(1) lookups.
func (s *Server) publicKeysInDB(ctx context.Context) (map[[fieldparams.BLSPubkeyLength]byte]bool, error) {
pubKeysInDB := make(map[[fieldparams.BLSPubkeyLength]byte]bool)
attestedPublicKeys, err := s.valDB.AttestedPublicKeys(ctx)
attestedPublicKeys, err := s.db.AttestedPublicKeys(ctx)
if err != nil {
return nil, fmt.Errorf("could not get attested public keys from DB: %v", err)
}
proposedPublicKeys, err := s.valDB.ProposedPublicKeys(ctx)
proposedPublicKeys, err := s.db.ProposedPublicKeys(ctx)
if err != nil {
return nil, fmt.Errorf("could not get proposed public keys from DB: %v", err)
}
@@ -313,7 +313,7 @@ func (s *Server) slashingProtectionHistoryForDeletedKeys(
filteredKeys = append(filteredKeys, pk)
}
}
return slashingprotection.ExportStandardProtectionJSON(ctx, s.valDB, filteredKeys...)
return slashingprotection.ExportStandardProtectionJSON(ctx, s.db, filteredKeys...)
}
// SetVoluntaryExit creates a signed voluntary exit message and returns a VoluntaryExit object.
@@ -347,7 +347,7 @@ func (s *Server) SetVoluntaryExit(w http.ResponseWriter, r *http.Request) {
epoch := primitives.Epoch(e)
if rawEpoch == "" {
genesisResponse, err := s.beaconNodeClient.GetGenesis(ctx, &emptypb.Empty{})
genesisResponse, err := s.nodeClient.GetGenesis(ctx, &emptypb.Empty{})
if err != nil {
httputil.HandleError(w, errors.Wrap(err, "Failed to get genesis time").Error(), http.StatusInternalServerError)
return
@@ -414,7 +414,7 @@ func (s *Server) ListRemoteKeys(w http.ResponseWriter, r *http.Request) {
for i := 0; i < len(pubKeys); i++ {
keystoreResponse[i] = &RemoteKey{
Pubkey: hexutil.Encode(pubKeys[i][:]),
Url: s.validatorService.Web3SignerConfig.BaseEndpoint,
Url: s.validatorService.RemoteSignerConfig().BaseEndpoint,
Readonly: true,
}
}

26
validator/rpc/handlers_keymanager_test.go
View File

@@ -289,7 +289,7 @@ func TestServer_ImportKeystores(t *testing.T) {
})
}
require.NoError(t, err)
s.valDB = validatorDB
s.db = validatorDB
// Have to close it after import is done otherwise it complains db is not open.
defer func() {
@@ -413,7 +413,7 @@ func TestServer_DeleteKeystores(t *testing.T) {
})
}
require.NoError(t, err)
srv.valDB = validatorDB
srv.db = validatorDB
// Have to close it after import is done otherwise it complains db is not open.
defer func() {
@@ -589,7 +589,7 @@ func TestServer_DeleteKeystores_FailedSlashingProtectionExport(t *testing.T) {
require.NoError(t, err)
err = validatorDB.SaveGenesisValidatorsRoot(ctx, make([]byte, fieldparams.RootLength))
require.NoError(t, err)
srv.valDB = validatorDB
srv.db = validatorDB
// Have to close it after import is done otherwise it complains db is not open.
defer func() {
@@ -746,7 +746,7 @@ func TestServer_SetVoluntaryExit(t *testing.T) {
validatorService: vs,
beaconNodeValidatorClient: beaconClient,
wallet: w,
beaconNodeClient: mockNodeClient,
nodeClient: mockNodeClient,
walletInitialized: w != nil,
}
@@ -841,7 +841,7 @@ func TestServer_SetVoluntaryExit(t *testing.T) {
resp := &SetVoluntaryExitResponse{}
require.NoError(t, json.Unmarshal(w.Body.Bytes(), resp))
if tt.w.epoch == 0 {
genesisResponse, err := s.beaconNodeClient.GetGenesis(ctx, &emptypb.Empty{})
genesisResponse, err := s.nodeClient.GetGenesis(ctx, &emptypb.Empty{})
require.NoError(t, err)
tt.w.epoch, err = client.CurrentEpoch(genesisResponse.GenesisTime)
require.NoError(t, err)
@@ -1091,14 +1091,14 @@ func TestServer_SetGasLimit(t *testing.T) {
validatorDB := dbtest.SetupDB(t, [][fieldparams.BLSPubkeyLength]byte{}, isSlashingProtectionMinimal)
vs, err := client.NewValidatorService(ctx, &client.Config{
Validator: m,
ValDB: validatorDB,
DB: validatorDB,
})
require.NoError(t, err)
s := &Server{
validatorService: vs,
beaconNodeValidatorClient: beaconClient,
valDB: validatorDB,
db: validatorDB,
}
if tt.beaconReturn != nil {
@@ -1280,12 +1280,12 @@ func TestServer_DeleteGasLimit(t *testing.T) {
validatorDB := dbtest.SetupDB(t, [][fieldparams.BLSPubkeyLength]byte{}, isSlashingProtectionMinimal)
vs, err := client.NewValidatorService(ctx, &client.Config{
Validator: m,
ValDB: validatorDB,
DB: validatorDB,
})
require.NoError(t, err)
s := &Server{
validatorService: vs,
valDB: validatorDB,
db: validatorDB,
}
// Set up global default value for builder gas limit.
params.BeaconConfig().DefaultBuilderGasLimit = uint64(globalDefaultGasLimit)
@@ -1744,13 +1744,13 @@ func TestServer_FeeRecipientByPubkey(t *testing.T) {
// save a default here
vs, err := client.NewValidatorService(ctx, &client.Config{
Validator: m,
ValDB: validatorDB,
DB: validatorDB,
})
require.NoError(t, err)
s := &Server{
validatorService: vs,
beaconNodeValidatorClient: beaconClient,
valDB: validatorDB,
db: validatorDB,
}
request := &SetFeeRecipientByPubkeyRequest{
Ethaddress: tt.args,
@@ -1854,12 +1854,12 @@ func TestServer_DeleteFeeRecipientByPubkey(t *testing.T) {
validatorDB := dbtest.SetupDB(t, [][fieldparams.BLSPubkeyLength]byte{}, isSlashingProtectionMinimal)
vs, err := client.NewValidatorService(ctx, &client.Config{
Validator: m,
ValDB: validatorDB,
DB: validatorDB,
})
require.NoError(t, err)
s := &Server{
validatorService: vs,
valDB: validatorDB,
db: validatorDB,
}
req := httptest.NewRequest(http.MethodDelete, fmt.Sprintf("/eth/v1/validator/{pubkey}/feerecipient"), nil)
req = mux.SetURLVars(req, map[string]string{"pubkey": pubkey})

8
validator/rpc/handlers_slashing.go
View File

@@ -24,12 +24,12 @@ func (s *Server) ExportSlashingProtection(w http.ResponseWriter, r *http.Request
ctx, span := trace.StartSpan(r.Context(), "validator.ExportSlashingProtection")
defer span.End()
if s.valDB == nil {
if s.db == nil {
httputil.HandleError(w, "could not find validator database", http.StatusInternalServerError)
return
}
eipJSON, err := slashing.ExportStandardProtectionJSON(ctx, s.valDB)
eipJSON, err := slashing.ExportStandardProtectionJSON(ctx, s.db)
if err != nil {
httputil.HandleError(w, errors.Wrap(err, "could not export slashing protection history").Error(), http.StatusInternalServerError)
return
@@ -54,7 +54,7 @@ func (s *Server) ImportSlashingProtection(w http.ResponseWriter, r *http.Request
ctx, span := trace.StartSpan(r.Context(), "validator.ImportSlashingProtection")
defer span.End()
if s.valDB == nil {
if s.db == nil {
httputil.HandleError(w, "could not find validator database", http.StatusInternalServerError)
return
}
@@ -76,7 +76,7 @@ func (s *Server) ImportSlashingProtection(w http.ResponseWriter, r *http.Request
}
enc := []byte(req.SlashingProtectionJson)
buf := bytes.NewBuffer(enc)
if err := s.valDB.ImportStandardProtectionJSON(ctx, buf); err != nil {
if err := s.db.ImportStandardProtectionJSON(ctx, buf); err != nil {
httputil.HandleError(w, errors.Wrap(err, "could not import slashing protection history").Error(), http.StatusInternalServerError)
return
}

6
validator/rpc/handlers_slashing_test.go
View File

@@ -77,7 +77,7 @@ func TestImportSlashingProtection_Preconditions(t *testing.T) {
})
}
require.NoError(t, err)
s.valDB = validatorDB
s.db = validatorDB
// Have to close it after import is done otherwise it complains db is not open.
defer func() {
@@ -151,7 +151,7 @@ func TestExportSlashingProtection_Preconditions(t *testing.T) {
})
}
require.NoError(t, err)
s.valDB = validatorDB
s.db = validatorDB
// Have to close it after export is done otherwise it complains db is not open.
defer func() {
@@ -189,7 +189,7 @@ func TestImportExportSlashingProtection_RoundTrip(t *testing.T) {
PubKeys: pubKeys,
})
require.NoError(t, err)
s.valDB = validatorDB
s.db = validatorDB
// Have to close it after import is done otherwise it complains db is not open.
defer func() {

149
validator/rpc/server.go
View File

@@ -23,122 +23,83 @@ import (
"github.com/prysmaticlabs/prysm/v5/validator/client"
iface "github.com/prysmaticlabs/prysm/v5/validator/client/iface"
"github.com/prysmaticlabs/prysm/v5/validator/db"
"github.com/sirupsen/logrus"
"go.opencensus.io/plugin/ocgrpc"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/reflection"
)
// Config options for the gRPC server.
type Config struct {
ValidatorGatewayHost string
ValidatorGatewayPort int
ValidatorMonitoringHost string
ValidatorMonitoringPort int
BeaconClientEndpoint string
ClientMaxCallRecvMsgSize int
ClientGrpcRetries uint
ClientGrpcRetryDelay time.Duration
ClientGrpcHeaders []string
ClientWithCert string
Host string
Port string
CertFlag string
KeyFlag string
ValDB db.Database
AuthTokenPath string
WalletDir string
ValidatorService *client.ValidatorService
SyncChecker client.SyncChecker
GenesisFetcher client.GenesisFetcher
WalletInitializedFeed *event.Feed
NodeGatewayEndpoint string
BeaconApiEndpoint string
BeaconApiTimeout time.Duration
Router *mux.Router
Wallet *wallet.Wallet
Host string
Port string
GRPCGatewayHost string
GRPCGatewayPort int
GRPCMaxCallRecvMsgSize int
GRPCRetries uint
GRPCRetryDelay time.Duration
GRPCHeaders []string
BeaconNodeGRPCEndpoint string
BeaconApiEndpoint string
BeaconApiTimeout time.Duration
BeaconNodeCert string
DB db.Database
Wallet *wallet.Wallet
WalletDir string
WalletInitializedFeed *event.Feed
ValidatorService *client.ValidatorService
Router *mux.Router
}
// Server defining a gRPC server for the remote signer API.
type Server struct {
logsStreamer logs.Streamer
streamLogsBufferSize int
beaconChainClient iface.BeaconChainClient
beaconNodeClient iface.NodeClient
beaconNodeValidatorClient iface.ValidatorClient
beaconNodeHealthClient ethpb.HealthClient
valDB db.Database
ctx context.Context
cancel context.CancelFunc
beaconClientEndpoint string
clientMaxCallRecvMsgSize int
clientGrpcRetries uint
clientGrpcRetryDelay time.Duration
clientGrpcHeaders []string
clientWithCert string
host string
port string
grpcGatewayHost string
grpcGatewayPort int
listener net.Listener
withCert string
withKey string
credentialError error
grpcMaxCallRecvMsgSize int
grpcRetries uint
grpcRetryDelay time.Duration
grpcHeaders []string
grpcServer *grpc.Server
beaconNodeValidatorClient iface.ValidatorClient
chainClient iface.ChainClient
nodeClient iface.NodeClient
healthClient ethpb.HealthClient
beaconNodeEndpoint string
beaconApiEndpoint string
beaconApiTimeout time.Duration
beaconNodeCert string
jwtSecret []byte
validatorService *client.ValidatorService
syncChecker client.SyncChecker
genesisFetcher client.GenesisFetcher
authTokenPath string
authToken string
db db.Database
walletDir string
wallet *wallet.Wallet
walletInitializedFeed *event.Feed
walletInitialized bool
nodeGatewayEndpoint string
validatorMonitoringHost string
validatorMonitoringPort int
validatorGatewayHost string
validatorGatewayPort int
beaconApiEndpoint string
beaconApiTimeout time.Duration
validatorService *client.ValidatorService
router *mux.Router
logStreamer logs.Streamer
logStreamerBufferSize int
}
// NewServer instantiates a new gRPC server.
func NewServer(ctx context.Context, cfg *Config) *Server {
ctx, cancel := context.WithCancel(ctx)
server := &Server{
ctx: ctx,
cancel: cancel,
logsStreamer: logs.NewStreamServer(),
streamLogsBufferSize: 1000, // Enough to handle most bursts of logs in the validator client.
host: cfg.Host,
port: cfg.Port,
withCert: cfg.CertFlag,
withKey: cfg.KeyFlag,
beaconClientEndpoint: cfg.BeaconClientEndpoint,
clientMaxCallRecvMsgSize: cfg.ClientMaxCallRecvMsgSize,
clientGrpcRetries: cfg.ClientGrpcRetries,
clientGrpcRetryDelay: cfg.ClientGrpcRetryDelay,
clientGrpcHeaders: cfg.ClientGrpcHeaders,
clientWithCert: cfg.ClientWithCert,
valDB: cfg.ValDB,
validatorService: cfg.ValidatorService,
syncChecker: cfg.SyncChecker,
genesisFetcher: cfg.GenesisFetcher,
authTokenPath: cfg.AuthTokenPath,
walletDir: cfg.WalletDir,
walletInitializedFeed: cfg.WalletInitializedFeed,
walletInitialized: cfg.Wallet != nil,
wallet: cfg.Wallet,
nodeGatewayEndpoint: cfg.NodeGatewayEndpoint,
validatorMonitoringHost: cfg.ValidatorMonitoringHost,
validatorMonitoringPort: cfg.ValidatorMonitoringPort,
validatorGatewayHost: cfg.ValidatorGatewayHost,
validatorGatewayPort: cfg.ValidatorGatewayPort,
beaconApiTimeout: cfg.BeaconApiTimeout,
beaconApiEndpoint: cfg.BeaconApiEndpoint,
router: cfg.Router,
ctx: ctx,
cancel: cancel,
host: cfg.Host,
port: cfg.Port,
grpcGatewayHost: cfg.GRPCGatewayHost,
grpcGatewayPort: cfg.GRPCGatewayPort,
grpcMaxCallRecvMsgSize: cfg.GRPCMaxCallRecvMsgSize,
grpcRetries: cfg.GRPCRetries,
grpcRetryDelay: cfg.GRPCRetryDelay,
grpcHeaders: cfg.GRPCHeaders,
}
if server.authTokenPath == "" && server.walletDir != "" {
@@ -149,7 +110,7 @@ func NewServer(ctx context.Context, cfg *Config) *Server {
if err := server.initializeAuthToken(); err != nil {
log.WithError(err).Error("Could not initialize web auth token")
}
validatorWebAddr := fmt.Sprintf("%s:%d", server.validatorGatewayHost, server.validatorGatewayPort)
validatorWebAddr := fmt.Sprintf("%s:%d", server.grpcGatewayHost, server.grpcGatewayPort)
logValidatorWebAuth(validatorWebAddr, server.authToken, server.authTokenPath)
go server.refreshAuthTokenFromFileChanges(server.ctx, server.authTokenPath)
}
@@ -184,17 +145,7 @@ func (s *Server) Start() {
)),
}
grpcprometheus.EnableHandlingTimeHistogram()
if s.withCert != "" && s.withKey != "" {
creds, err := credentials.NewServerTLSFromFile(s.withCert, s.withKey)
if err != nil {
log.WithError(err).Fatal("Could not load TLS keys")
}
opts = append(opts, grpc.Creds(creds))
log.WithFields(logrus.Fields{
"certPath": s.withCert,
"keyPath": s.withKey,
}).Info("Loaded TLS certificates")
}
s.grpcServer = grpc.NewServer(opts...)
// Register a gRPC client to the beacon node.
@@ -282,7 +233,7 @@ func (s *Server) Stop() error {
return nil
}
// Status returns nil or credentialError.
// Status returns an error if the service is unhealthy.
func (s *Server) Status() error {
return s.credentialError
return nil
}