Implement Stream Duties Functionality (#5685)

* add test for stream duties

* rem gomock

* context cancelation test

* use interface for epoch ticker in duties

* fix build

* compute duties on first call

* pass tests for streams

* gaz

* ss stream duties

* ensure only stream across epoch-wide reorgs

* regen mock for validator

* Update beacon-chain/blockchain/head.go

* Update beacon-chain/rpc/validator/server.go

* update workspac

* pregenesis check

* tests pass

* build fix

* Revert "tests pass"

This reverts commit 9029d63.

* ethereumapis

* gaz

* Update beacon-chain/rpc/validator/assignments.go

* test for reorg

* fix up required tests

* fmt

Co-authored-by: prylabs-bulldozer[bot] <58059840+prylabs-bulldozer[bot]@users.noreply.github.com>
Co-authored-by: Preston Van Loon <preston@prysmaticlabs.com>

WORKSPACE

@@ -1333,7 +1333,7 @@ go_repository(
 
 go_repository(
     name = "com_github_prysmaticlabs_ethereumapis",
-    commit = "ba9042096e9fc49606279513d3e24e5e8cdbd5a0",
+    commit = "df460bd3d84be4ff3df0658395c7dc9d2a7e7b3d",
     importpath = "github.com/prysmaticlabs/ethereumapis",
 )
 

beacon-chain/blockchain/head.go

@@ -2,14 +2,18 @@ package blockchain
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/pkg/errors"
 	ethpb "github.com/prysmaticlabs/ethereumapis/eth/v1alpha1"
+	"github.com/prysmaticlabs/prysm/beacon-chain/core/feed"
+	statefeed "github.com/prysmaticlabs/prysm/beacon-chain/core/feed/state"
 	"github.com/prysmaticlabs/prysm/beacon-chain/state"
 	stateTrie "github.com/prysmaticlabs/prysm/beacon-chain/state"
 	"github.com/prysmaticlabs/prysm/shared/bytesutil"
 	"github.com/prysmaticlabs/prysm/shared/featureconfig"
 	"github.com/prysmaticlabs/prysm/shared/params"
+	"github.com/sirupsen/logrus"
 	"go.opencensus.io/trace"
 )
 
@@ -100,6 +104,21 @@ func (s *Service) saveHead(ctx context.Context, headRoot [32]byte) error {
 		return errors.New("cannot save nil head state")
 	}
 
+	// A chain re-org occurred, so we fire an event notifying the rest of the services.
+	if bytesutil.ToBytes32(newHeadBlock.Block.ParentRoot) != s.headRoot() {
+		log.WithFields(logrus.Fields{
+			"newSlot": fmt.Sprintf("%d", newHeadBlock.Block.Slot),
+			"oldSlot": fmt.Sprintf("%d", s.headSlot()),
+		}).Debug("Chain reorg occurred")
+		s.stateNotifier.StateFeed().Send(&feed.Event{
+			Type: statefeed.Reorg,
+			Data: &statefeed.ReorgData{
+				NewSlot: newHeadBlock.Block.Slot,
+				OldSlot: s.headSlot(),
+			},
+		})
+	}
+
 	// Cache the new head info.
 	s.setHead(headRoot, newHeadBlock, newHeadState)
 

beacon-chain/blockchain/head_test.go

@@ -11,6 +11,7 @@ import (
 	"github.com/prysmaticlabs/prysm/beacon-chain/state/stateutil"
 	pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
 	"github.com/prysmaticlabs/prysm/shared/testutil"
+	logTest "github.com/sirupsen/logrus/hooks/test"
 )
 
 func TestSaveHead_Same(t *testing.T) {
@@ -82,3 +83,59 @@ func TestSaveHead_Different(t *testing.T) {
 		t.Error("Head did not change")
 	}
 }
+
+func TestSaveHead_Different_Reorg(t *testing.T) {
+	hook := logTest.NewGlobal()
+	db := testDB.SetupDB(t)
+	service := setupBeaconChain(t, db)
+
+	oldRoot := [32]byte{'A'}
+	service.head = &head{slot: 0, root: oldRoot}
+
+	reorgChainParent := [32]byte{'B'}
+	newHeadBlock := &ethpb.BeaconBlock{
+		Slot:       1,
+		ParentRoot: reorgChainParent[:],
+	}
+	newHeadSignedBlock := &ethpb.SignedBeaconBlock{Block: newHeadBlock}
+
+	if err := service.beaconDB.SaveBlock(context.Background(), newHeadSignedBlock); err != nil {
+		t.Fatal(err)
+	}
+	newRoot, err := stateutil.BlockRoot(newHeadBlock)
+	if err != nil {
+		t.Fatal(err)
+	}
+	headState := testutil.NewBeaconState()
+	if err := headState.SetSlot(1); err != nil {
+		t.Fatal(err)
+	}
+	if err := service.beaconDB.SaveStateSummary(context.Background(), &pb.StateSummary{Slot: 1, Root: newRoot[:]}); err != nil {
+		t.Fatal(err)
+	}
+	if err := service.beaconDB.SaveState(context.Background(), headState, newRoot); err != nil {
+		t.Fatal(err)
+	}
+	if err := service.saveHead(context.Background(), newRoot); err != nil {
+		t.Fatal(err)
+	}
+
+	if service.HeadSlot() != 1 {
+		t.Error("Head did not change")
+	}
+
+	cachedRoot, err := service.HeadRoot(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Equal(cachedRoot, newRoot[:]) {
+		t.Error("Head did not change")
+	}
+	if !reflect.DeepEqual(service.headBlock(), newHeadSignedBlock) {
+		t.Error("Head did not change")
+	}
+	if !reflect.DeepEqual(service.headState().CloneInnerState(), headState.CloneInnerState()) {
+		t.Error("Head did not change")
+	}
+	testutil.AssertLogsContain(t, hook, "Chain reorg occurred")
+}

beacon-chain/core/feed/state/events.go

@@ -14,13 +14,16 @@ const (
 	Initialized
 	// Synced is sent when the beacon node has completed syncing and is ready to participate in the network.
 	Synced
+	// Reorg is an event sent when the new head state's slot after a block
+	// transition is lower than its previous head state slot value.
+	Reorg
 )
 
 // BlockProcessedData is the data sent with BlockProcessed events.
 type BlockProcessedData struct {
 	// Slot is the slot of the processed block.
 	Slot uint64
-	// BlockRoot is the hash of the processed block.
+	// BlockRoot of the processed block.
 	BlockRoot [32]byte
 	// Verified is true if the block's BLS contents have been verified.
 	Verified bool
@@ -45,3 +48,11 @@ type InitializedData struct {
 	// GenesisValidatorsRoot represents ssz.HashTreeRoot(state.validators).
 	GenesisValidatorsRoot []byte
 }
+
+// ReorgData is the data alongside a reorg event.
+type ReorgData struct {
+	// NewSlot is the slot of new state after the reorg.
+	NewSlot uint64
+	// OldSlot is the slot of the head state before the reorg.
+	OldSlot uint64
+}

beacon-chain/rpc/testing/BUILD.bazel

@@ -11,6 +11,7 @@ go_library(
     importpath = "github.com/prysmaticlabs/prysm/beacon-chain/rpc/testing",
     visibility = ["//beacon-chain:__subpackages__"],
     deps = [
+        "@com_github_gogo_protobuf//types:go_default_library",
         "@com_github_golang_mock//gomock:go_default_library",
         "@com_github_prysmaticlabs_ethereumapis//eth/v1alpha1:go_default_library",
         "@org_golang_google_grpc//metadata:go_default_library",