Stream Duties Client Implementation

validator/client/BUILD.bazel

@@ -10,6 +10,7 @@ go_library(
         "validator.go",
         "validator_aggregate.go",
         "validator_attest.go",
+        "validator_duties.go",
         "validator_log.go",
         "validator_metrics.go",
         "validator_propose.go",
@@ -63,6 +64,7 @@ go_test(
         "service_test.go",
         "validator_aggregate_test.go",
         "validator_attest_test.go",
+        "validator_duties_test.go",
         "validator_propose_test.go",
         "validator_test.go",
     ],
@@ -93,4 +95,4 @@ go_test(
         "@com_github_sirupsen_logrus//hooks/test:go_default_library",
         "@in_gopkg_d4l3k_messagediff_v1//:go_default_library",
     ],
-)
+)

validator/client/fake_validator_test.go

@@ -15,7 +15,7 @@ type fakeValidator struct {
 	WaitForSyncedCalled              bool
 	NextSlotCalled                   bool
 	CanonicalHeadSlotCalled          bool
-	UpdateDutiesCalled               bool
+	StreamDutiesCalled               bool
 	UpdateProtectionsCalled          bool
 	RoleAtCalled                     bool
 	AttestToBlockHeadCalled          bool
@@ -26,10 +26,9 @@ type fakeValidator struct {
 	ProposeBlockArg1                 uint64
 	AttestToBlockHeadArg1            uint64
 	RoleAtArg1                       uint64
-	UpdateDutiesArg1                 uint64
 	NextSlotRet                      <-chan uint64
 	PublicKey                        string
-	UpdateDutiesRet                  error
+	StreamDutiesRet                  error
 	RolesAtRet                       []validatorRole
 }
 
@@ -72,10 +71,9 @@ func (fv *fakeValidator) NextSlot() <-chan uint64 {
 	return fv.NextSlotRet
 }
 
-func (fv *fakeValidator) UpdateDuties(_ context.Context, slot uint64) error {
-	fv.UpdateDutiesCalled = true
-	fv.UpdateDutiesArg1 = slot
-	return fv.UpdateDutiesRet
+func (fv *fakeValidator) StreamDuties(_ context.Context) error {
+	fv.StreamDutiesCalled = true
+	return fv.StreamDutiesRet
 }
 
 func (fv *fakeValidator) UpdateProtections(_ context.Context, slot uint64) error {

validator/client/runner.go

@@ -26,7 +26,7 @@ type Validator interface {
 	NextSlot() <-chan uint64
 	SlotDeadline(slot uint64) time.Time
 	LogValidatorGainsAndLosses(ctx context.Context, slot uint64) error
-	UpdateDuties(ctx context.Context, slot uint64) error
+	StreamDuties(ctx context.Context) error
 	UpdateProtections(ctx context.Context, slot uint64) error
 	RolesAt(ctx context.Context, slot uint64) (map[[48]byte][]validatorRole, error) // validator pubKey -> roles
 	SubmitAttestation(ctx context.Context, slot uint64, pubKey [48]byte)
@@ -43,8 +43,8 @@ type Validator interface {
 // Order of operations:
 // 1 - Initialize validator data
 // 2 - Wait for validator activation
-// 3 - Wait for the next slot start
-// 4 - Update assignments
+// 3 - Listen to a server-side stream of validator duties
+// 4 - Wait for the next slot start
 // 5 - Determine role at current slot
 // 6 - Perform assigned role, if any
 func run(ctx context.Context, v Validator) {
@@ -64,13 +64,20 @@ func run(ctx context.Context, v Validator) {
 	if err := v.WaitForActivation(ctx); err != nil {
 		log.Fatalf("Could not wait for validator activation: %v", err)
 	}
+	log.Info("Got activation")
 	headSlot, err := v.CanonicalHeadSlot(ctx)
 	if err != nil {
 		log.Fatalf("Could not get current canonical head slot: %v", err)
 	}
-	if err := v.UpdateDuties(ctx, headSlot); err != nil {
-		handleAssignmentError(err, headSlot)
-	}
+	log.Info("Got canonical head slot")
+	// We listen to a server-side stream of validator duties in the
+	// background of the validator client.
+	go func() {
+		log.Info("Requesting duties stream")
+		if err := v.StreamDuties(ctx); err != nil {
+			handleAssignmentError(err, headSlot)
+		}
+	}()
 	for {
 		ctx, span := trace.StartSpan(ctx, "validator.processSlot")
 
@@ -81,23 +88,14 @@ func run(ctx context.Context, v Validator) {
 		case slot := <-v.NextSlot():
 			span.AddAttributes(trace.Int64Attribute("slot", int64(slot)))
 			deadline := v.SlotDeadline(slot)
-			slotCtx, cancel := context.WithDeadline(ctx, deadline)
+			slotCtx, _ := context.WithDeadline(ctx, deadline)
 			// Report this validator client's rewards and penalties throughout its lifecycle.
 			log := log.WithField("slot", slot)
 			log.WithField("deadline", deadline).Debug("Set deadline for proposals and attestations")
 			if err := v.LogValidatorGainsAndLosses(slotCtx, slot); err != nil {
 				log.WithError(err).Error("Could not report validator's rewards/penalties")
 			}
 
-			// Keep trying to update assignments if they are nil or if we are past an
-			// epoch transition in the beacon node's state.
-			if err := v.UpdateDuties(ctx, slot); err != nil {
-				handleAssignmentError(err, slot)
-				cancel()
-				span.End()
-				continue
-			}
-
 			if featureconfig.Get().ProtectAttester {
 				if err := v.UpdateProtections(ctx, slot); err != nil {
 					log.WithError(err).Error("Could not update validator protection")

validator/client/runner_test.go

@@ -2,13 +2,10 @@ package client
 
 import (
 	"context"
-	"errors"
 	"testing"
 	"time"
 
 	"github.com/prysmaticlabs/prysm/shared/featureconfig"
-	"github.com/prysmaticlabs/prysm/shared/testutil"
-	logTest "github.com/sirupsen/logrus/hooks/test"
 )
 
 func cancelledContext() context.Context {
@@ -54,49 +51,6 @@ func TestCancelledContext_WaitsForActivation(t *testing.T) {
 	}
 }
 
-func TestUpdateDuties_NextSlot(t *testing.T) {
-	v := &fakeValidator{}
-	ctx, cancel := context.WithCancel(context.Background())
-
-	slot := uint64(55)
-	ticker := make(chan uint64)
-	v.NextSlotRet = ticker
-	go func() {
-		ticker <- slot
-
-		cancel()
-	}()
-
-	run(ctx, v)
-
-	if !v.UpdateDutiesCalled {
-		t.Fatalf("Expected UpdateAssignments(%d) to be called", slot)
-	}
-	if v.UpdateDutiesArg1 != slot {
-		t.Errorf("UpdateAssignments was called with wrong argument. Want=%d, got=%d", slot, v.UpdateDutiesArg1)
-	}
-}
-
-func TestUpdateDuties_HandlesError(t *testing.T) {
-	hook := logTest.NewGlobal()
-	v := &fakeValidator{}
-	ctx, cancel := context.WithCancel(context.Background())
-
-	slot := uint64(55)
-	ticker := make(chan uint64)
-	v.NextSlotRet = ticker
-	go func() {
-		ticker <- slot
-
-		cancel()
-	}()
-	v.UpdateDutiesRet = errors.New("bad")
-
-	run(ctx, v)
-
-	testutil.AssertLogsContain(t, hook, "Failed to update assignments")
-}
-
 func TestRoleAt_NextSlot(t *testing.T) {
 	v := &fakeValidator{}
 	ctx, cancel := context.WithCancel(context.Background())

validator/client/service.go

@@ -175,6 +175,7 @@ func (v *ValidatorService) Start() {
 	}
 	v.validator = &validator{
 		db:                             valDB,
+		dutiesByEpoch:                  make(map[uint64][]*ethpb.DutiesResponse_Duty, 2), // 2 epochs worth of duties.
 		validatorClient:                ethpb.NewBeaconNodeValidatorClient(v.conn),
 		beaconClient:                   ethpb.NewBeaconChainClient(v.conn),
 		node:                           ethpb.NewNodeClient(v.conn),
@@ -200,9 +201,7 @@ func (v *ValidatorService) Stop() error {
 	return nil
 }
 
-// Status ...
-//
-// WIP - not done.
+// Status of the validator service's health.
 func (v *ValidatorService) Status() error {
 	if v.conn == nil {
 		return errors.New("no connection to beacon RPC")