Implement Stream Duties Functionality

WORKSPACE

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

beacon-chain/blockchain/head.go

@@ -5,6 +5,8 @@ import (
 
 	"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"
@@ -100,6 +102,8 @@ func (s *Service) saveHead(ctx context.Context, headRoot [32]byte) error {
 		return errors.New("cannot save nil head state")
 	}
 
+	currentHeadSlot := s.headSlot()
+
 	// Cache the new head info.
 	s.setHead(headRoot, newHeadBlock, newHeadState)
 
@@ -108,6 +112,17 @@ func (s *Service) saveHead(ctx context.Context, headRoot [32]byte) error {
 		return errors.Wrap(err, "could not save head root in DB")
 	}
 
+	// A chain re-org occurred, so we fire an event notifying the rest of the services.
+	if newHeadState.Slot() < currentHeadSlot {
+		s.stateNotifier.StateFeed().Send(&feed.Event{
+			Type: statefeed.Reorg,
+			Data: &statefeed.ReorgData{
+				NewSlot: newHeadState.Slot(),
+				OldSlot: currentHeadSlot,
+			},
+		})
+	}
+
 	return nil
 }
 

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",