Infostream (#4760)

* Add validators stream

* Ignore unknown keys rather than error on them

* Reduce accesses to common structures

* Ensure correct information returned for deposited validators

* Short-term cache for remote deposit data

* Name epoch duration for clarity

* Break out duplicated logic in to a single function

* Add capacities for slices and maps where appropriate

* Break out functions; add tests

* Allow stream errors not related to context

Co-authored-by: Nishant Das <nish1993@hotmail.com>
Co-authored-by: Raul Jordan <raul@prysmaticlabs.com>

beacon-chain/core/helpers/validators.go

@@ -254,7 +254,7 @@ func Domain(fork *pb.Fork, epoch uint64, domainType []byte) uint64 {
 }
 
 // IsEligibleForActivationQueue checks if the validator is eligible to
-// be places into the activation queue.
+// be placed into the activation queue.
 //
 // Spec pseudocode definition:
 //  def is_eligible_for_activation_queue(validator: Validator) -> bool:
@@ -266,8 +266,19 @@ func Domain(fork *pb.Fork, epoch uint64, domainType []byte) uint64 {
 //        and validator.effective_balance == MAX_EFFECTIVE_BALANCE
 //    )
 func IsEligibleForActivationQueue(validator *ethpb.Validator) bool {
-	return validator.ActivationEligibilityEpoch == params.BeaconConfig().FarFutureEpoch &&
-		validator.EffectiveBalance == params.BeaconConfig().MaxEffectiveBalance
+	return isEligibileForActivationQueue(validator.ActivationEligibilityEpoch, validator.EffectiveBalance)
+}
+
+// IsEligibleForActivationQueueUsingTrie checks if the read-only validator is eligible to
+// be placed into the activation queue.
+func IsEligibleForActivationQueueUsingTrie(validator *stateTrie.ReadOnlyValidator) bool {
+	return isEligibileForActivationQueue(validator.ActivationEligibilityEpoch(), validator.EffectiveBalance())
+}
+
+// isEligibleForActivationQueue carries out the logic for IsEligibleForActivationQueue*
+func isEligibileForActivationQueue(activationEligibilityEpoch uint64, effectiveBalance uint64) bool {
+	return activationEligibilityEpoch == params.BeaconConfig().FarFutureEpoch &&
+		effectiveBalance == params.BeaconConfig().MaxEffectiveBalance
 }
 
 // IsEligibleForActivation checks if the validator is eligible for activation.
@@ -285,6 +296,20 @@ func IsEligibleForActivationQueue(validator *ethpb.Validator) bool {
 //    )
 func IsEligibleForActivation(state *stateTrie.BeaconState, validator *ethpb.Validator) bool {
 	finalizedEpoch := state.FinalizedCheckpointEpoch()
-	return validator.ActivationEligibilityEpoch <= finalizedEpoch &&
-		validator.ActivationEpoch == params.BeaconConfig().FarFutureEpoch
+	return isEligibileForActivation(validator.ActivationEligibilityEpoch, validator.ActivationEpoch, finalizedEpoch)
+}
+
+// IsEligibleForActivationUsingTrie checks if the validator is eligible for activation.
+func IsEligibleForActivationUsingTrie(state *stateTrie.BeaconState, validator *stateTrie.ReadOnlyValidator) bool {
+	cpt := state.FinalizedCheckpoint()
+	if cpt == nil {
+		return false
+	}
+	return isEligibileForActivation(validator.ActivationEligibilityEpoch(), validator.ActivationEpoch(), cpt.Epoch)
+}
+
+// isEligibleForActivation carries out the logic for IsEligibleForActivation*
+func isEligibileForActivation(activationEligibilityEpoch uint64, activationEpoch uint64, finalizedEpoch uint64) bool {
+	return activationEligibilityEpoch <= finalizedEpoch &&
+		activationEpoch == params.BeaconConfig().FarFutureEpoch
 }

beacon-chain/rpc/beacon/BUILD.bazel

@@ -11,11 +11,13 @@ go_library(
         "server.go",
         "slashings.go",
         "validators.go",
+        "validators_stream.go",
     ],
     importpath = "github.com/prysmaticlabs/prysm/beacon-chain/rpc/beacon",
     visibility = ["//beacon-chain:__subpackages__"],
     deps = [
         "//beacon-chain/blockchain:go_default_library",
+        "//beacon-chain/cache/depositcache:go_default_library",
         "//beacon-chain/core/epoch/precompute:go_default_library",
         "//beacon-chain/core/feed:go_default_library",
         "//beacon-chain/core/feed/block:go_default_library",
@@ -29,18 +31,22 @@ go_library(
         "//beacon-chain/operations/attestations:go_default_library",
         "//beacon-chain/operations/slashings:go_default_library",
         "//beacon-chain/powchain:go_default_library",
+        "//beacon-chain/state:go_default_library",
         "//proto/beacon/p2p/v1:go_default_library",
         "//shared/attestationutil:go_default_library",
         "//shared/bytesutil:go_default_library",
+        "//shared/event:go_default_library",
         "//shared/hashutil:go_default_library",
         "//shared/pagination:go_default_library",
         "//shared/params:go_default_library",
         "//shared/sliceutil:go_default_library",
         "//shared/slotutil:go_default_library",
         "@com_github_gogo_protobuf//types:go_default_library",
+        "@com_github_patrickmn_go_cache//:go_default_library",
         "@com_github_pkg_errors//:go_default_library",
         "@com_github_prysmaticlabs_ethereumapis//eth/v1alpha1:go_default_library",
         "@com_github_prysmaticlabs_go_ssz//:go_default_library",
+        "@com_github_sirupsen_logrus//:go_default_library",
         "@org_golang_google_grpc//codes:go_default_library",
         "@org_golang_google_grpc//status:go_default_library",
     ],
@@ -55,6 +61,7 @@ go_test(
         "committees_test.go",
         "config_test.go",
         "slashings_test.go",
+        "validators_stream_test.go",
         "validators_test.go",
     ],
     embed = [":go_default_library"],

beacon-chain/rpc/beacon/server.go

@@ -6,6 +6,7 @@ import (
 
 	ethpb "github.com/prysmaticlabs/ethereumapis/eth/v1alpha1"
 	"github.com/prysmaticlabs/prysm/beacon-chain/blockchain"
+	"github.com/prysmaticlabs/prysm/beacon-chain/cache/depositcache"
 	blockfeed "github.com/prysmaticlabs/prysm/beacon-chain/core/feed/block"
 	statefeed "github.com/prysmaticlabs/prysm/beacon-chain/core/feed/state"
 	"github.com/prysmaticlabs/prysm/beacon-chain/db"
@@ -26,6 +27,8 @@ type Server struct {
 	HeadFetcher          blockchain.HeadFetcher
 	FinalizationFetcher  blockchain.FinalizationFetcher
 	ParticipationFetcher blockchain.ParticipationFetcher
+	DepositFetcher       depositcache.DepositFetcher
+	BlockFetcher         powchain.POWBlockFetcher
 	GenesisTimeFetcher   blockchain.TimeFetcher
 	StateNotifier        statefeed.Notifier
 	BlockNotifier        blockfeed.Notifier

beacon-chain/rpc/beacon/validators.go

@@ -629,12 +629,6 @@ func (bs *Server) GetValidatorPerformance(
 	}, nil
 }
 
-// StreamValidatorsInfo streams out important validator information and metadata
-// each epoch.
-func (bs *Server) StreamValidatorsInfo(stream ethpb.BeaconChain_StreamValidatorsInfoServer) error {
-	return status.Error(codes.Unimplemented, "Unimplemented")
-}
-
 // Determines whether a validator has already exited.
 func validatorHasExited(validator *ethpb.Validator, currentEpoch uint64) bool {
 	farFutureEpoch := params.BeaconConfig().FarFutureEpoch