app/eth2wrap: deep validator cache (#3114)

Implement a validator cache that caches the complete Validators endpoint request.

Short-cut the duty selection for the DV validator set to the known-active, cached list instead: this will reduce the load on the beacon node drastically, since before this change we were calling the BN once every slot.

Try to reduce at a minimum upstream BN requests when the VC calls for validator state.

All Validators() calls to the upstream BN, if greater than 200 validators, will have an exponential timeout calculated as `(50ms * validator amount)`.

category: feature
ticket: none

app/app.go

@@ -427,7 +427,7 @@ func wireCoreWorkflow(ctx context.Context, life *lifecycle.Manager, conf Config,
 			return nil
 		}
 		valCache.Trim()
-		_, err := valCache.Get(ctx)
+		_, _, err := valCache.Get(ctx)
 
 		return err
 	})

app/eth2wrap/httpwrap.go

@@ -15,6 +15,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/attestantio/go-eth2-client/api"
+	apiv1 "github.com/attestantio/go-eth2-client/api/v1"
 	eth2http "github.com/attestantio/go-eth2-client/http"
 	eth2p0 "github.com/attestantio/go-eth2-client/spec/phase0"
 
@@ -67,15 +69,15 @@ type httpAdapter struct {
 	address     string
 	timeout     time.Duration
 	valCacheMu  sync.RWMutex
-	valCache    func(context.Context) (ActiveValidators, error)
+	valCache    func(context.Context) (ActiveValidators, CompleteValidators, error)
 	forkVersion [4]byte
 }
 
 func (h *httpAdapter) SetForkVersion(forkVersion [4]byte) {
 	h.forkVersion = forkVersion
 }
 
-func (h *httpAdapter) SetValidatorCache(valCache func(context.Context) (ActiveValidators, error)) {
+func (h *httpAdapter) SetValidatorCache(valCache func(context.Context) (ActiveValidators, CompleteValidators, error)) {
 	h.valCacheMu.Lock()
 	h.valCache = valCache
 	h.valCacheMu.Unlock()
@@ -89,7 +91,48 @@ func (h *httpAdapter) ActiveValidators(ctx context.Context) (ActiveValidators, e
 		return nil, errors.New("no active validator cache")
 	}
 
-	return h.valCache(ctx)
+	active, _, err := h.valCache(ctx)
+
+	return active, err
+}
+
+func (h *httpAdapter) CompleteValidators(ctx context.Context) (CompleteValidators, error) {
+	h.valCacheMu.RLock()
+	defer h.valCacheMu.RUnlock()
+
+	if h.valCache == nil {
+		return nil, errors.New("no active validator cache")
+	}
+
+	_, complete, err := h.valCache(ctx)
+
+	return complete, err
+}
+
+// Validators returns the validators as requested in opts.
+// If the amount of validators requested is greater than 200, exponentially increase the timeout: on crowded testnets
+// this HTTP call takes a long time.
+func (h *httpAdapter) Validators(ctx context.Context, opts *api.ValidatorsOpts) (
+	*api.Response[map[eth2p0.ValidatorIndex]*apiv1.Validator],
+	error,
+) {
+	var cancel func()
+	reqCtx := ctx
+
+	maxValAmt := max(len(opts.PubKeys), len(opts.Indices))
+
+	if maxValAmt > 200 {
+		reqTimeout := time.Duration(50*maxValAmt) * time.Millisecond
+		reqCtx, cancel = context.WithTimeout(reqCtx, reqTimeout)
+	}
+
+	defer func() {
+		if cancel != nil {
+			cancel()
+		}
+	}()
+
+	return h.Service.Validators(reqCtx, opts)
 }
 
 // AggregateBeaconCommitteeSelections implements eth2exp.BeaconCommitteeSelectionAggregator.

app/eth2wrap/lazy.go

@@ -38,7 +38,7 @@ type lazy struct {
 
 	clientMu sync.RWMutex
 	client   Client
-	valCache func(context.Context) (ActiveValidators, error)
+	valCache func(context.Context) (ActiveValidators, CompleteValidators, error)
 }
 
 // getClient returns the client and true if it is available.
@@ -146,7 +146,16 @@ func (l *lazy) ActiveValidators(ctx context.Context) (ActiveValidators, error) {
 	return cl.ActiveValidators(ctx)
 }
 
-func (l *lazy) SetValidatorCache(valCache func(context.Context) (ActiveValidators, error)) {
+func (l *lazy) CompleteValidators(ctx context.Context) (CompleteValidators, error) {
+	cl, err := l.getOrCreateClient(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	return cl.CompleteValidators(ctx)
+}
+
+func (l *lazy) SetValidatorCache(valCache func(context.Context) (ActiveValidators, CompleteValidators, error)) {
 	l.clientMu.Lock()
 	l.valCache = valCache
 	l.clientMu.Unlock()

app/eth2wrap/lazy_test.go

@@ -135,8 +135,8 @@ func TestLazy_ActiveValidators(t *testing.T) {
 }
 
 func TestLazy_SetValidatorCache(t *testing.T) {
-	valCache := func(context.Context) (eth2wrap.ActiveValidators, error) {
-		return nil, nil
+	valCache := func(context.Context) (eth2wrap.ActiveValidators, eth2wrap.CompleteValidators, error) {
+		return nil, nil, nil
 	}
 
 	client := mocks.NewClient(t)

app/eth2wrap/multi.go

@@ -73,7 +73,7 @@ func (m multi) IsSynced() bool {
 	return false
 }
 
-func (m multi) SetValidatorCache(valCache func(context.Context) (ActiveValidators, error)) {
+func (m multi) SetValidatorCache(valCache func(context.Context) (ActiveValidators, CompleteValidators, error)) {
 	for _, cl := range m.clients {
 		cl.SetValidatorCache(valCache)
 	}
@@ -97,6 +97,24 @@ func (m multi) ActiveValidators(ctx context.Context) (ActiveValidators, error) {
 	return res0, err
 }
 
+func (m multi) CompleteValidators(ctx context.Context) (CompleteValidators, error) {
+	const label = "complete_validators"
+	// No latency since this is a cached endpoint.
+
+	res0, err := provide(ctx, m.clients,
+		func(ctx context.Context, cl Client) (CompleteValidators, error) {
+			return cl.CompleteValidators(ctx)
+		},
+		nil, nil,
+	)
+	if err != nil {
+		incError(label)
+		err = wrapError(ctx, err, label)
+	}
+
+	return res0, err
+}
+
 func (m multi) ProposerConfig(ctx context.Context) (*eth2exp.ProposerConfigResponse, error) {
 	const label = "proposer_config"
 	defer latency(label)()

app/eth2wrap/multi_test.go

@@ -174,8 +174,8 @@ func TestMulti_ActiveValidators(t *testing.T) {
 }
 
 func TestMulti_SetValidatorCache(t *testing.T) {
-	valCache := func(context.Context) (eth2wrap.ActiveValidators, error) {
-		return nil, nil
+	valCache := func(context.Context) (eth2wrap.ActiveValidators, eth2wrap.CompleteValidators, error) {
+		return nil, nil, nil
 	}
 
 	client := mocks.NewClient(t)

app/eth2wrap/synthproposer.go

@@ -36,7 +36,7 @@ const (
 )
 
 type synthProposerEth2Provider interface {
-	ActiveValidatorsProvider
+	CachedValidatorsProvider
 	eth2client.SlotsPerEpochProvider
 	eth2client.ProposerDutiesProvider
 }

app/eth2wrap/synthproposer_test.go

@@ -62,8 +62,8 @@ func TestSynthProposer(t *testing.T) {
 			},
 		}, nil
 	}
-	cached := bmock.ActiveValidatorsFunc
-	bmock.ActiveValidatorsFunc = func(ctx context.Context) (eth2wrap.ActiveValidators, error) {
+	cached := bmock.CachedValidatorsFunc
+	bmock.CachedValidatorsFunc = func(ctx context.Context) (eth2wrap.ActiveValidators, eth2wrap.CompleteValidators, error) {
 		activeVals++
 		return cached(ctx)
 	}
@@ -195,8 +195,8 @@ func TestSynthProposerBlockNotFound(t *testing.T) {
 			},
 		}, nil
 	}
-	cached := bmock.ActiveValidatorsFunc
-	bmock.ActiveValidatorsFunc = func(ctx context.Context) (eth2wrap.ActiveValidators, error) {
+	cached := bmock.CachedValidatorsFunc
+	bmock.CachedValidatorsFunc = func(ctx context.Context) (eth2wrap.ActiveValidators, eth2wrap.CompleteValidators, error) {
 		activeVals++
 		return cached(ctx)
 	}