Merge pull request #303 from maticnetwork/merge-subs2

Merge subs2

core/blockchain.go

@@ -217,6 +217,7 @@ type BlockChain struct {
 	borReceiptsCache *lru.Cache             // Cache for the most recent bor receipt receipts per block
 	stateSyncData    []*types.StateSyncData // State sync data
 	stateSyncFeed    event.Feed             // State sync feed
+	chain2HeadFeed   event.Feed             // Reorg/NewHead/Fork data feed
 }
 
 // NewBlockChain returns a fully initialised block chain using information
@@ -1640,10 +1641,21 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types.
 			for _, data := range bc.stateSyncData {
 				bc.stateSyncFeed.Send(StateSyncEvent{Data: data})
 			}
+
+			bc.chain2HeadFeed.Send(Chain2HeadEvent{
+				Type:     Chain2HeadCanonicalEvent,
+				NewChain: []*types.Block{block},
+			})
+
 			// BOR
 		}
 	} else {
 		bc.chainSideFeed.Send(ChainSideEvent{Block: block})
+
+		bc.chain2HeadFeed.Send(Chain2HeadEvent{
+			Type:     Chain2HeadForkEvent,
+			NewChain: []*types.Block{block},
+		})
 	}
 	return status, nil
 }
@@ -1737,6 +1749,11 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, er
 	defer func() {
 		if lastCanon != nil && bc.CurrentBlock().Hash() == lastCanon.Hash() {
 			bc.chainHeadFeed.Send(ChainHeadEvent{lastCanon})
+
+			bc.chain2HeadFeed.Send(Chain2HeadEvent{
+				Type:     Chain2HeadCanonicalEvent,
+				NewChain: []*types.Block{lastCanon},
+			})
 		}
 	}()
 	// Start the parallel header verifier
@@ -2262,6 +2279,13 @@ func (bc *BlockChain) reorg(oldBlock, newBlock *types.Block) error {
 	}
 	// Ensure the user sees large reorgs
 	if len(oldChain) > 0 && len(newChain) > 0 {
+
+		bc.chain2HeadFeed.Send(Chain2HeadEvent{
+			Type:     Chain2HeadReorgEvent,
+			NewChain: newChain,
+			OldChain: oldChain,
+		})
+
 		logFn := log.Info
 		msg := "Chain reorg detected"
 		if len(oldChain) > 63 {
@@ -2570,6 +2594,11 @@ func (bc *BlockChain) SubscribeChainHeadEvent(ch chan<- ChainHeadEvent) event.Su
 	return bc.scope.Track(bc.chainHeadFeed.Subscribe(ch))
 }
 
+// SubscribeChain2HeadEvent registers a subscription of ChainHeadEvent. ()
+func (bc *BlockChain) SubscribeChain2HeadEvent(ch chan<- Chain2HeadEvent) event.Subscription {
+	return bc.scope.Track(bc.chain2HeadFeed.Subscribe(ch))
+}
+
 // SubscribeChainSideEvent registers a subscription of ChainSideEvent.
 func (bc *BlockChain) SubscribeChainSideEvent(ch chan<- ChainSideEvent) event.Subscription {
 	return bc.scope.Track(bc.chainSideFeed.Subscribe(ch))

core/blockchain_bor_test.go

@@ -0,0 +1,134 @@
+package core
+
+import (
+	"math/big"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/consensus/ethash"
+	"github.com/ethereum/go-ethereum/core/rawdb"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/core/vm"
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/params"
+)
+
+func TestChain2HeadEvent(t *testing.T) {
+	var (
+		db      = rawdb.NewMemoryDatabase()
+		key1, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291")
+		addr1   = crypto.PubkeyToAddress(key1.PublicKey)
+		gspec   = &Genesis{
+			Config: params.TestChainConfig,
+			Alloc:  GenesisAlloc{addr1: {Balance: big.NewInt(10000000000000000)}},
+		}
+		genesis = gspec.MustCommit(db)
+		signer  = types.LatestSigner(gspec.Config)
+	)
+
+	blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil)
+	defer blockchain.Stop()
+
+	chain2HeadCh := make(chan Chain2HeadEvent, 64)
+	blockchain.SubscribeChain2HeadEvent(chain2HeadCh)
+
+	chain, _ := GenerateChain(gspec.Config, genesis, ethash.NewFaker(), db, 3, func(i int, gen *BlockGen) {})
+	if _, err := blockchain.InsertChain(chain); err != nil {
+		t.Fatalf("failed to insert chain: %v", err)
+	}
+
+	replacementBlocks, _ := GenerateChain(gspec.Config, genesis, ethash.NewFaker(), db, 4, func(i int, gen *BlockGen) {
+		tx, err := types.SignTx(types.NewContractCreation(gen.TxNonce(addr1), new(big.Int), 1000000, gen.header.BaseFee, nil), signer, key1)
+		if i == 2 {
+			gen.OffsetTime(-9)
+		}
+		if err != nil {
+			t.Fatalf("failed to create tx: %v", err)
+		}
+		gen.AddTx(tx)
+	})
+
+	if _, err := blockchain.InsertChain(replacementBlocks); err != nil {
+		t.Fatalf("failed to insert chain: %v", err)
+	}
+
+	type eventTest struct {
+		Type    string
+		Added   []common.Hash
+		Removed []common.Hash
+	}
+
+	readEvent := func(expect *eventTest) {
+		select {
+		case ev := <-chain2HeadCh:
+			if ev.Type != expect.Type {
+				t.Fatal("Type mismatch")
+			}
+
+			if len(ev.NewChain) != len(expect.Added) {
+				t.Fatal("Newchain and Added Array Size don't match")
+			}
+			if len(ev.OldChain) != len(expect.Removed) {
+				t.Fatal("Oldchain and Removed Array Size don't match")
+			}
+
+			for j := 0; j < len(ev.OldChain); j++ {
+				if ev.OldChain[j].Hash() != expect.Removed[j] {
+					t.Fatal("Oldchain hashes Do Not Match")
+				}
+			}
+			for j := 0; j < len(ev.NewChain); j++ {
+				if ev.NewChain[j].Hash() != expect.Added[j] {
+					t.Fatal("Newchain hashes Do Not Match")
+				}
+			}
+		case <-time.After(2 * time.Second):
+			t.Fatal("timeout")
+		}
+	}
+
+	// head event
+	readEvent(&eventTest{
+		Type: Chain2HeadCanonicalEvent,
+		Added: []common.Hash{
+			chain[2].Hash(),
+		}})
+
+	// fork event
+	readEvent(&eventTest{
+		Type: Chain2HeadForkEvent,
+		Added: []common.Hash{
+			replacementBlocks[0].Hash(),
+		}})
+
+	// fork event
+	readEvent(&eventTest{
+		Type: Chain2HeadForkEvent,
+		Added: []common.Hash{
+			replacementBlocks[1].Hash(),
+		}})
+
+	// reorg event
+	//In this event the channel recieves an array of Blocks in NewChain and OldChain
+	readEvent(&eventTest{
+		Type: Chain2HeadReorgEvent,
+		Added: []common.Hash{
+			replacementBlocks[2].Hash(),
+			replacementBlocks[1].Hash(),
+			replacementBlocks[0].Hash(),
+		},
+		Removed: []common.Hash{
+			chain[2].Hash(),
+			chain[1].Hash(),
+			chain[0].Hash(),
+		},
+	})
+
+	// head event
+	readEvent(&eventTest{
+		Type: Chain2HeadCanonicalEvent,
+		Added: []common.Hash{
+			replacementBlocks[3].Hash(),
+		}})
+}

core/bor_events.go

@@ -8,3 +8,16 @@ import (
 type StateSyncEvent struct {
 	Data *types.StateSyncData
 }
+
+var (
+	Chain2HeadReorgEvent     = "reorg"
+	Chain2HeadCanonicalEvent = "head"
+	Chain2HeadForkEvent      = "fork"
+)
+
+// For tracking reorgs related information
+type Chain2HeadEvent struct {
+	NewChain []*types.Block
+	OldChain []*types.Block
+	Type     string
+}

docs/cli/README.md

@@ -31,4 +31,6 @@
 
 - [```status```](./status.md)
 
+- [```chain watch```](./chain_watch.md)
+
 - [```version```](./version.md)

docs/cli/chain_watch.md

@@ -0,0 +1,3 @@
+# Chain watch
+
+The ```chain watch``` command is used to view the chainHead, reorg and fork events in real-time.

eth/bor_api_backend.go

@@ -67,3 +67,8 @@ func (b *EthAPIBackend) GetBorBlockTransactionWithBlockHash(ctx context.Context,
 func (b *EthAPIBackend) SubscribeStateSyncEvent(ch chan<- core.StateSyncEvent) event.Subscription {
 	return b.eth.BlockChain().SubscribeStateSyncEvent(ch)
 }
+
+// SubscribeChain2HeadEvent subscribes to reorg/head/fork event
+func (b *EthAPIBackend) SubscribeChain2HeadEvent(ch chan<- core.Chain2HeadEvent) event.Subscription {
+	return b.eth.BlockChain().SubscribeChain2HeadEvent(ch)
+}

ethstats/ethstats.go

@@ -57,6 +57,9 @@ const (
 	txChanSize = 4096
 	// chainHeadChanSize is the size of channel listening to ChainHeadEvent.
 	chainHeadChanSize = 10
+
+	// chain2HeadChanSize is the size of channel listening to Chain2HeadEvent.
+	chain2HeadChanSize = 10
 )
 
 // backend encompasses the bare-minimum functionality needed for ethstats reporting
@@ -68,6 +71,9 @@ type backend interface {
 	GetTd(ctx context.Context, hash common.Hash) *big.Int
 	Stats() (pending int, queued int)
 	SyncProgress() ethereum.SyncProgress
+
+	// Bor
+	SubscribeChain2HeadEvent(ch chan<- core.Chain2HeadEvent) event.Subscription
 }
 
 // fullNodeBackend encompasses the functionality necessary for a full node
@@ -96,6 +102,9 @@ type Service struct {
 
 	headSub event.Subscription
 	txSub   event.Subscription
+
+	//bor related sub
+	chain2headSub event.Subscription
 }
 
 // connWrapper is a wrapper to prevent concurrent-write or concurrent-read on the
@@ -195,7 +204,9 @@ func (s *Service) Start() error {
 	s.headSub = s.backend.SubscribeChainHeadEvent(chainHeadCh)
 	txEventCh := make(chan core.NewTxsEvent, txChanSize)
 	s.txSub = s.backend.SubscribeNewTxsEvent(txEventCh)
-	go s.loop(chainHeadCh, txEventCh)
+	chain2HeadCh := make(chan core.Chain2HeadEvent, chain2HeadChanSize)
+	s.chain2headSub = s.backend.SubscribeChain2HeadEvent(chain2HeadCh)
+	go s.loop(chainHeadCh, chain2HeadCh, txEventCh)
 
 	log.Info("Stats daemon started")
 	return nil
@@ -211,12 +222,13 @@ func (s *Service) Stop() error {
 
 // loop keeps trying to connect to the netstats server, reporting chain events
 // until termination.
-func (s *Service) loop(chainHeadCh chan core.ChainHeadEvent, txEventCh chan core.NewTxsEvent) {
+func (s *Service) loop(chainHeadCh chan core.ChainHeadEvent, chain2HeadCh chan core.Chain2HeadEvent, txEventCh chan core.NewTxsEvent) {
 	// Start a goroutine that exhausts the subscriptions to avoid events piling up
 	var (
-		quitCh = make(chan struct{})
+		quitCh  = make(chan struct{})
-		headCh = make(chan *types.Block, 1)
+		headCh  = make(chan *types.Block, 1)
-		txCh   = make(chan struct{}, 1)
+		txCh    = make(chan struct{}, 1)
+		head2Ch = make(chan core.Chain2HeadEvent, 100)
 	)
 	go func() {
 		var lastTx mclock.AbsTime
@@ -231,6 +243,13 @@ func (s *Service) loop(chainHeadCh chan core.ChainHeadEvent, txEventCh chan core
 				default:
 				}
 
+			// Notify of chain2head events, but drop if too frequent
+			case chain2head := <-chain2HeadCh:
+				select {
+				case head2Ch <- chain2head:
+				default:
+				}
+
 			// Notify of new transaction events, but drop if too frequent
 			case <-txEventCh:
 				if time.Duration(mclock.Now()-lastTx) < time.Second {
@@ -333,6 +352,12 @@ func (s *Service) loop(chainHeadCh chan core.ChainHeadEvent, txEventCh chan core
 					if err = s.reportPending(conn); err != nil {
 						log.Warn("Post-block transaction stats report failed", "err", err)
 					}
+
+				case chain2head := <-head2Ch:
+					if err = s.reportChain2Head(conn, &chain2head); err != nil {
+						log.Warn("Reorg stats report failed", "err", err)
+					}
+
 				case <-txCh:
 					if err = s.reportPending(conn); err != nil {
 						log.Warn("Transaction stats report failed", "err", err)
@@ -750,6 +775,49 @@ func (s *Service) reportPending(conn *connWrapper) error {
 	return conn.WriteJSON(report)
 }
 
+type blockStub struct {
+	Hash       string `json:"hash"`
+	Number     uint64 `json:"number"`
+	ParentHash string `json:"parent_hash"`
+}
+
+func createStub(b *types.Block) *blockStub {
+	s := &blockStub{
+		Hash:       b.Hash().String(),
+		ParentHash: b.ParentHash().String(),
+		Number:     b.NumberU64(),
+	}
+	return s
+}
+
+type ChainHeadEvent struct {
+	NewChain []*blockStub `json:"added"`
+	OldChain []*blockStub `json:"removed"`
+	Type     string       `json:"type"`
+}
+
+// reportChain2Head checks for reorg and sends current head to stats server.
+func (s *Service) reportChain2Head(conn *connWrapper, chain2HeadData *core.Chain2HeadEvent) error {
+	chainHeadEvent := ChainHeadEvent{
+		Type: chain2HeadData.Type,
+	}
+	for _, block := range chain2HeadData.NewChain {
+		chainHeadEvent.NewChain = append(chainHeadEvent.NewChain, createStub(block))
+	}
+	for _, block := range chain2HeadData.OldChain {
+		chainHeadEvent.OldChain = append(chainHeadEvent.OldChain, createStub(block))
+	}
+
+	stats := map[string]interface{}{
+		"id":    s.node,
+		"event": chainHeadEvent,
+	}
+	report := map[string][]interface{}{
+		"emit": {"headEvent", stats},
+	}
+	return conn.WriteJSON(report)
+}
+
 // nodeStats is the information to report about the local node.
 type nodeStats struct {
 	Active   bool `json:"active"`

go.sum

@@ -33,7 +33,6 @@ github.com/Azure/go-autorest/autorest/date v0.2.0 h1:yW+Zlqf26583pE43KhfnhFcdmSW
 github.com/Azure/go-autorest/autorest/date v0.2.0/go.mod h1:vcORJHLJEh643/Ioh9+vPmf1Ij9AEBM5FuBIXLmIy0g=
 github.com/Azure/go-autorest/autorest/mocks v0.1.0/go.mod h1:OTyCOPRA2IgIlWxVYxBee2F5Gr4kF2zd2J5cFRaIDN0=
 github.com/Azure/go-autorest/autorest/mocks v0.2.0/go.mod h1:OTyCOPRA2IgIlWxVYxBee2F5Gr4kF2zd2J5cFRaIDN0=
-github.com/Azure/go-autorest/autorest/mocks v0.3.0 h1:qJumjCaCudz+OcqE9/XtEPfvtOjOmKaui4EOpFI6zZc=
 github.com/Azure/go-autorest/autorest/mocks v0.3.0/go.mod h1:a8FDP3DYzQ4RYfVAxAN3SVSiiO77gL2j2ronKKP0syM=
 github.com/Azure/go-autorest/logger v0.1.0 h1:ruG4BSDXONFRrZZJ2GUXDiUyVpayPmb1GnWeHDdaNKY=
 github.com/Azure/go-autorest/logger v0.1.0/go.mod h1:oExouG+K6PryycPJfVSxi/koC6LSNgds39diKLz7Vrc=

internal/cli/chain_watch.go

@@ -0,0 +1,89 @@
+package cli
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"github.com/ethereum/go-ethereum/core"
+	"github.com/ethereum/go-ethereum/internal/cli/flagset"
+	"github.com/ethereum/go-ethereum/internal/cli/server/proto"
+)
+
+// ChainWatchCommand is the command to group the peers commands
+type ChainWatchCommand struct {
+	*Meta2
+}
+
+// Help implements the cli.Command interface
+func (c *ChainWatchCommand) Help() string {
+	return `Usage: bor chain watch
+
+  This command is used to view the chainHead, reorg and fork events in real-time`
+}
+
+func (c *ChainWatchCommand) Flags() *flagset.Flagset {
+	flags := c.NewFlagSet("chain watch")
+
+	return flags
+}
+
+// Synopsis implements the cli.Command interface
+func (c *ChainWatchCommand) Synopsis() string {
+	return "Watch the chainHead, reorg and fork events in real-time"
+}
+
+// Run implements the cli.Command interface
+func (c *ChainWatchCommand) Run(args []string) int {
+	flags := c.Flags()
+	if err := flags.Parse(args); err != nil {
+		c.UI.Error(err.Error())
+		return 1
+	}
+
+	borClt, err := c.BorConn()
+	if err != nil {
+		c.UI.Error(err.Error())
+		return 1
+	}
+
+	sub, err := borClt.ChainWatch(context.Background(), &proto.ChainWatchRequest{})
+	if err != nil {
+		c.UI.Error(err.Error())
+		return 1
+	}
+
+	signalCh := make(chan os.Signal, 1)
+	signal.Notify(signalCh, os.Interrupt, syscall.SIGTERM)
+
+	go func() {
+		<-signalCh
+		sub.CloseSend()
+	}()
+
+	for {
+		msg, err := sub.Recv()
+		if err != nil {
+			// if err == EOF if finished on the other side
+			c.UI.Output(err.Error())
+			break
+		}
+		c.UI.Output(formatHeadEvent(msg))
+	}
+
+	return 0
+}
+
+func formatHeadEvent(msg *proto.ChainWatchResponse) string {
+	var out string
+	if msg.Type == core.Chain2HeadCanonicalEvent {
+		out = fmt.Sprintf("Block Added : %v", msg.Newchain)
+	} else if msg.Type == core.Chain2HeadForkEvent {
+		out = fmt.Sprintf("New Fork Block : %v", msg.Newchain)
+	} else if msg.Type == core.Chain2HeadReorgEvent {
+		out = fmt.Sprintf("Reorg Detected \nAdded : %v \nRemoved : %v", msg.Newchain, msg.Oldchain)
+	}
+	return out
+}

internal/cli/command.go

@@ -65,6 +65,11 @@ func commands() map[string]cli.CommandFactory {
 				UI: ui,
 			}, nil
 		},
+		"chain watch": func() (cli.Command, error) {
+			return &ChainWatchCommand{
+				Meta2: meta2,
+			}, nil
+		},
 		"chain sethead": func() (cli.Command, error) {
 			return &ChainSetHeadCommand{
 				Meta2: meta2,

internal/cli/server/proto/server.proto

@@ -20,8 +20,24 @@ service Bor {
     rpc ChainSetHead(ChainSetHeadRequest) returns (ChainSetHeadResponse);
 
     rpc Status(google.protobuf.Empty) returns (StatusResponse);
+    
+    rpc ChainWatch(ChainWatchRequest) returns (stream ChainWatchResponse);
 }
 
+message ChainWatchRequest {
+
+}
+
+message ChainWatchResponse {
+    repeated BlockStub oldchain = 1;
+    repeated BlockStub newchain = 2;
+    string type = 3;
+}
+
+message BlockStub {
+    string hash = 1;
+    uint64 number = 2;
+}
 
 message PeersAddRequest {
     string enode = 1;

internal/cli/server/proto/server_grpc.pb.go

@@ -1,13 +1,17 @@
 // Code generated by protoc-gen-go-grpc. DO NOT EDIT.
+// versions:
+// - protoc-gen-go-grpc v1.2.0
+// - protoc             v3.17.3
+// source: command/server/proto/server.proto
 
 package proto
 
 import (
 	context "context"
-	empty "github.com/golang/protobuf/ptypes/empty"
 	grpc "google.golang.org/grpc"
 	codes "google.golang.org/grpc/codes"
 	status "google.golang.org/grpc/status"
+	emptypb "google.golang.org/protobuf/types/known/emptypb"
 )
 
 // This is a compile-time assertion to ensure that this generated file
@@ -25,7 +29,8 @@ type BorClient interface {
 	PeersList(ctx context.Context, in *PeersListRequest, opts ...grpc.CallOption) (*PeersListResponse, error)
 	PeersStatus(ctx context.Context, in *PeersStatusRequest, opts ...grpc.CallOption) (*PeersStatusResponse, error)
 	ChainSetHead(ctx context.Context, in *ChainSetHeadRequest, opts ...grpc.CallOption) (*ChainSetHeadResponse, error)
-	Status(ctx context.Context, in *empty.Empty, opts ...grpc.CallOption) (*StatusResponse, error)
+	Status(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*StatusResponse, error)
+	ChainWatch(ctx context.Context, in *ChainWatchRequest, opts ...grpc.CallOption) (Bor_ChainWatchClient, error)
 }
 
 type borClient struct {
@@ -113,7 +118,7 @@ func (c *borClient) ChainSetHead(ctx context.Context, in *ChainSetHeadRequest, o
 	return out, nil
 }
 
-func (c *borClient) Status(ctx context.Context, in *empty.Empty, opts ...grpc.CallOption) (*StatusResponse, error) {
+func (c *borClient) Status(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*StatusResponse, error) {
 	out := new(StatusResponse)
 	err := c.cc.Invoke(ctx, "/proto.Bor/Status", in, out, opts...)
 	if err != nil {
@@ -122,6 +127,38 @@ func (c *borClient) Status(ctx context.Context, in *empty.Empty, opts ...grpc.Ca
 	return out, nil
 }
 
+func (c *borClient) ChainWatch(ctx context.Context, in *ChainWatchRequest, opts ...grpc.CallOption) (Bor_ChainWatchClient, error) {
+	stream, err := c.cc.NewStream(ctx, &Bor_ServiceDesc.Streams[1], "/proto.Bor/ChainWatch", opts...)
+	if err != nil {
+		return nil, err
+	}
+	x := &borChainWatchClient{stream}
+	if err := x.ClientStream.SendMsg(in); err != nil {
+		return nil, err
+	}
+	if err := x.ClientStream.CloseSend(); err != nil {
+		return nil, err
+	}
+	return x, nil
+}
+
+type Bor_ChainWatchClient interface {
+	Recv() (*ChainWatchResponse, error)
+	grpc.ClientStream
+}
+
+type borChainWatchClient struct {
+	grpc.ClientStream
+}
+
+func (x *borChainWatchClient) Recv() (*ChainWatchResponse, error) {
+	m := new(ChainWatchResponse)
+	if err := x.ClientStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
 // BorServer is the server API for Bor service.
 // All implementations must embed UnimplementedBorServer
 // for forward compatibility
@@ -132,7 +169,8 @@ type BorServer interface {
 	PeersList(context.Context, *PeersListRequest) (*PeersListResponse, error)
 	PeersStatus(context.Context, *PeersStatusRequest) (*PeersStatusResponse, error)
 	ChainSetHead(context.Context, *ChainSetHeadRequest) (*ChainSetHeadResponse, error)
-	Status(context.Context, *empty.Empty) (*StatusResponse, error)
+	Status(context.Context, *emptypb.Empty) (*StatusResponse, error)
+	ChainWatch(*ChainWatchRequest, Bor_ChainWatchServer) error
 	mustEmbedUnimplementedBorServer()
 }
 
@@ -158,9 +196,12 @@ func (UnimplementedBorServer) PeersStatus(context.Context, *PeersStatusRequest)
 func (UnimplementedBorServer) ChainSetHead(context.Context, *ChainSetHeadRequest) (*ChainSetHeadResponse, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method ChainSetHead not implemented")
 }
-func (UnimplementedBorServer) Status(context.Context, *empty.Empty) (*StatusResponse, error) {
+func (UnimplementedBorServer) Status(context.Context, *emptypb.Empty) (*StatusResponse, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method Status not implemented")
 }
+func (UnimplementedBorServer) ChainWatch(*ChainWatchRequest, Bor_ChainWatchServer) error {
+	return status.Errorf(codes.Unimplemented, "method ChainWatch not implemented")
+}
 func (UnimplementedBorServer) mustEmbedUnimplementedBorServer() {}
 
 // UnsafeBorServer may be embedded to opt out of forward compatibility for this service.
@@ -286,7 +327,7 @@ func _Bor_ChainSetHead_Handler(srv interface{}, ctx context.Context, dec func(in
 }
 
 func _Bor_Status_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
-	in := new(empty.Empty)
+	in := new(emptypb.Empty)
 	if err := dec(in); err != nil {
 		return nil, err
 	}
@@ -298,11 +339,32 @@ func _Bor_Status_Handler(srv interface{}, ctx context.Context, dec func(interfac
 		FullMethod: "/proto.Bor/Status",
 	}
 	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
-		return srv.(BorServer).Status(ctx, req.(*empty.Empty))
+		return srv.(BorServer).Status(ctx, req.(*emptypb.Empty))
 	}
 	return interceptor(ctx, in, info, handler)
 }
 
+func _Bor_ChainWatch_Handler(srv interface{}, stream grpc.ServerStream) error {
+	m := new(ChainWatchRequest)
+	if err := stream.RecvMsg(m); err != nil {
+		return err
+	}
+	return srv.(BorServer).ChainWatch(m, &borChainWatchServer{stream})
+}
+
+type Bor_ChainWatchServer interface {
+	Send(*ChainWatchResponse) error
+	grpc.ServerStream
+}
+
+type borChainWatchServer struct {
+	grpc.ServerStream
+}
+
+func (x *borChainWatchServer) Send(m *ChainWatchResponse) error {
+	return x.ServerStream.SendMsg(m)
+}
+
 // Bor_ServiceDesc is the grpc.ServiceDesc for Bor service.
 // It's only intended for direct use with grpc.RegisterService,
 // and not to be introspected or modified (even as a copy)
@@ -341,6 +403,11 @@ var Bor_ServiceDesc = grpc.ServiceDesc{
 			Handler:       _Bor_Pprof_Handler,
 			ServerStreams: true,
 		},
+		{
+			StreamName:    "ChainWatch",
+			Handler:       _Bor_ChainWatch_Handler,
+			ServerStreams: true,
+		},
 	},
 	Metadata: "command/server/proto/server.proto",
 }

internal/cli/server/service.go

@@ -7,6 +7,7 @@ import (
 	"reflect"
 	"strings"
 
+	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/internal/cli/server/pprof"
 	"github.com/ethereum/go-ethereum/internal/cli/server/proto"
@@ -218,3 +219,39 @@ func gatherForks(configList ...interface{}) []*proto.StatusResponse_Fork {
 	}
 	return forks
 }
+
+func convertBlockToBlockStub(blocks []*types.Block) []*proto.BlockStub {
+
+	var blockStubs []*proto.BlockStub
+
+	for _, block := range blocks {
+		blockStub := &proto.BlockStub{
+			Hash:   block.Hash().String(),
+			Number: block.NumberU64(),
+		}
+		blockStubs = append(blockStubs, blockStub)
+	}
+
+	return blockStubs
+}
+
+func (s *Server) ChainWatch(req *proto.ChainWatchRequest, reply proto.Bor_ChainWatchServer) error {
+
+	chain2HeadChanSize := 10
+
+	chain2HeadCh := make(chan core.Chain2HeadEvent, chain2HeadChanSize)
+	headSub := s.backend.APIBackend.SubscribeChain2HeadEvent(chain2HeadCh)
+	defer headSub.Unsubscribe()
+
+	for {
+		msg := <-chain2HeadCh
+
+		err := reply.Send(&proto.ChainWatchResponse{Type: msg.Type,
+			Newchain: convertBlockToBlockStub(msg.NewChain),
+			Oldchain: convertBlockToBlockStub(msg.OldChain),
+		})
+		if err != nil {
+			return err
+		}
+	}
+}

internal/ethapi/backend.go

@@ -98,6 +98,7 @@ type Backend interface {
 	GetBorBlockLogs(ctx context.Context, hash common.Hash) ([]*types.Log, error)
 	GetBorBlockTransaction(ctx context.Context, txHash common.Hash) (*types.Transaction, common.Hash, uint64, uint64, error)
 	GetBorBlockTransactionWithBlockHash(ctx context.Context, txHash common.Hash, blockHash common.Hash) (*types.Transaction, common.Hash, uint64, uint64, error)
+	SubscribeChain2HeadEvent(ch chan<- core.Chain2HeadEvent) event.Subscription
 
 	ChainConfig() *params.ChainConfig
 	Engine() consensus.Engine

les/bor_api_backend.go

@@ -17,3 +17,8 @@ func (b *LesApiBackend) GetRootHash(ctx context.Context, starBlockNr uint64, end
 func (b *LesApiBackend) SubscribeStateSyncEvent(ch chan<- core.StateSyncEvent) event.Subscription {
 	return b.eth.blockchain.SubscribeStateSyncEvent(ch)
 }
+
+// SubscribeChain2HeadEvent subscribe head/fork/reorg events.
+func (b *LesApiBackend) SubscribeChain2HeadEvent(ch chan<- core.Chain2HeadEvent) event.Subscription {
+	return b.eth.BlockChain().SubscribeChain2HeadEvent(ch)
+}

light/lightchain.go

@@ -72,6 +72,9 @@ type LightChain struct {
 	running          int32 // whether LightChain is running or stopped
 	procInterrupt    int32 // interrupts chain insert
 	disableCheckFreq int32 // disables header verification
+
+	// Bor
+	chain2HeadFeed event.Feed
 }
 
 // NewLightChain returns a fully initialised light chain using information
@@ -561,6 +564,11 @@ func (lc *LightChain) SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent)
 	return lc.scope.Track(new(event.Feed).Subscribe(ch))
 }
 
+// SubscribeChain2HeadEvent registers a subscription of Reorg/head/fork events.
+func (lc *LightChain) SubscribeChain2HeadEvent(ch chan<- core.Chain2HeadEvent) event.Subscription {
+	return lc.scope.Track(lc.chain2HeadFeed.Subscribe(ch))
+}
+
 // DisableCheckFreq disables header validation. This is used for ultralight mode.
 func (lc *LightChain) DisableCheckFreq() {
 	atomic.StoreInt32(&lc.disableCheckFreq, 1)