report received block height (#4670)

* report received block height

* filter future blocks

* fix unit tests

Author: CoderZhi

chainservice/chainservice.go

@@ -13,6 +13,7 @@ import (
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
+	"go.uber.org/zap"
 	"google.golang.org/protobuf/proto"
 
 	"github.com/iotexproject/go-pkgs/hash"
@@ -127,9 +128,39 @@ func (cs *ChainService) Pause(pause bool) {
 	cs.paused.Store(pause)
 }
 
+func (cs *ChainService) Filter(messageType iotexrpc.MessageType, msg proto.Message, cap int) bool {
+	// Filter out messages that are not relevant to the chain service
+	if messageType != iotexrpc.MessageType_BLOCK {
+		return true
+	}
+	blk, ok := msg.(*iotextypes.Block)
+	if !ok || blk == nil {
+		return false
+	}
+	if blk.Header.Core.Height > atomic.LoadUint64(&cs.lastReceivedBlockHeight) {
+		atomic.StoreUint64(&cs.lastReceivedBlockHeight, blk.Header.Core.Height)
+	}
+	tip, err := cs.blockdao.Height()
+	if err != nil {
+		log.L().Error("failed to get tip height from blockdao", zap.Error(err))
+		return true
+	}
+	return blk.Header.Core.Height < tip+uint64(cap)
+}
+
 // ReportFullness switch on or off block sync
-func (cs *ChainService) ReportFullness(_ context.Context, messageType iotexrpc.MessageType, fullness float32) {
+func (cs *ChainService) ReportFullness(_ context.Context, messageType iotexrpc.MessageType, msg proto.Message, fullness float32) {
 	_blockchainFullnessMtc.WithLabelValues(iotexrpc.MessageType_name[int32(messageType)]).Set(float64(fullness))
+	switch messageType {
+	case iotexrpc.MessageType_BLOCK:
+		blk, ok := msg.(*iotextypes.Block)
+		if !ok || blk == nil {
+			return
+		}
+		if blk.Header.Core.Height > atomic.LoadUint64(&cs.lastReceivedBlockHeight) {
+			atomic.StoreUint64(&cs.lastReceivedBlockHeight, blk.Header.Core.Height)
+		}
+	}
 }
 
 // HandleAction handles incoming action request.
@@ -186,9 +217,6 @@ func (cs *ChainService) HandleBlock(ctx context.Context, peer string, pbBlock *i
 	if err != nil {
 		return err
 	}
-	if atomic.LoadUint64(&cs.lastReceivedBlockHeight) < blk.Height() {
-		atomic.StoreUint64(&cs.lastReceivedBlockHeight, blk.Height())
-	}
 	return cs.blocksync.ProcessBlock(ctx, peer, blk)
 }
 

dispatcher/dispatcher.go

@@ -232,7 +232,16 @@ func (d *IotxDispatcher) ValidateMessage(pMsg proto.Message) (bool, error) {
 }
 
 func (d *IotxDispatcher) queueMessage(msg *message) {
+	subscriber := d.subscriber(msg.chainID)
+	if subscriber == nil {
+		log.L().Warn("chainID has not been registered in dispatcher.", zap.Uint32("chainID", msg.chainID))
+		return
+	}
 	queue := d.queueMgr.Queue(msg)
+	if !subscriber.Filter(msg.msgType, msg.msg, cap(queue)) {
+		log.L().Debug("Message filtered by subscriber.", zap.Uint32("chainID", msg.chainID), zap.String("msgType", msg.msgType.String()))
+		return
+	}
 	select {
 	case queue <- msg:
 	default:
@@ -309,7 +318,7 @@ func (d *IotxDispatcher) updateMetrics(msg *message, queue chan *message) {
 	d.updateEventAudit(msg.msgType)
 	subscriber := d.subscriber(msg.chainID)
 	if subscriber != nil {
-		subscriber.ReportFullness(msg.ctx, msg.msgType, float32(len(queue))/float32(cap(queue)))
+		subscriber.ReportFullness(msg.ctx, msg.msgType, msg.msg, float32(len(queue))/float32(cap(queue)))
 	}
 }
 

dispatcher/dispatcher_test.go

@@ -181,7 +181,12 @@ func dispatcherIsClean(dsp *IotxDispatcher) bool {
 
 type dummySubscriber struct{}
 
-func (ds *dummySubscriber) ReportFullness(context.Context, iotexrpc.MessageType, float32) {}
+func (ds *dummySubscriber) Filter(iotexrpc.MessageType, proto.Message, int) bool {
+	return true
+}
+
+func (ds *dummySubscriber) ReportFullness(context.Context, iotexrpc.MessageType, proto.Message, float32) {
+}
 
 func (ds *dummySubscriber) HandleBlock(context.Context, string, *iotextypes.Block) error { return nil }
 
@@ -220,7 +225,12 @@ type counterSubscriber struct {
 	actionHash  atomic.Int32
 }
 
-func (cs *counterSubscriber) ReportFullness(context.Context, iotexrpc.MessageType, float32) {}
+func (cs *counterSubscriber) Filter(iotexrpc.MessageType, proto.Message, int) bool {
+	return true
+}
+
+func (cs *counterSubscriber) ReportFullness(context.Context, iotexrpc.MessageType, proto.Message, float32) {
+}
 
 func (cs *counterSubscriber) HandleBlock(context.Context, string, *iotextypes.Block) error {
 	cs.block.Inc()

dispatcher/subscriber.go

@@ -7,11 +7,13 @@ import (
 	"github.com/iotexproject/iotex-proto/golang/iotexrpc"
 	"github.com/iotexproject/iotex-proto/golang/iotextypes"
 	"github.com/libp2p/go-libp2p/core/peer"
+	"google.golang.org/protobuf/proto"
 )
 
 // Subscriber is the dispatcher subscriber interface
 type Subscriber interface {
-	ReportFullness(context.Context, iotexrpc.MessageType, float32)
+	Filter(iotexrpc.MessageType, proto.Message, int) bool
+	ReportFullness(context.Context, iotexrpc.MessageType, proto.Message, float32)
 	HandleAction(context.Context, *iotextypes.Action) error
 	HandleBlock(context.Context, string, *iotextypes.Block) error
 	HandleSyncRequest(context.Context, peer.AddrInfo, *iotexrpc.BlockSync) error