chore(lib/grandpa, dot/network): send CommitMessage directly to peer on round mismatch; cleanup grandpa receiveMessages

dot/core/interface.go

@@ -70,7 +70,7 @@ type TransactionState interface {
 
 // Network is the interface for the network service
 type Network interface {
-	SendMessage(network.NotificationsMessage)
+	GossipMessage(network.NotificationsMessage)
 }
 
 // EpochState is the interface for state.EpochState

dot/core/messages_test.go

@@ -110,8 +110,7 @@ func TestService_ProcessBlockAnnounceMessage(t *testing.T) {
 		BestBlock:      true,
 	}
 
-	//setup the SendMessage function
-	net.On("SendMessage", expected)
+	net.On("GossipMessage", expected)
 
 	state, err := s.storageState.TrieState(nil)
 	require.NoError(t, err)
@@ -120,7 +119,7 @@ func TestService_ProcessBlockAnnounceMessage(t *testing.T) {
 	require.NoError(t, err)
 
 	time.Sleep(time.Second)
-	net.AssertCalled(t, "SendMessage", expected)
+	net.AssertCalled(t, "GossipMessage", expected)
 }
 
 func TestService_HandleTransactionMessage(t *testing.T) {

dot/core/service.go

@@ -204,7 +204,7 @@ func (s *Service) HandleBlockProduced(block *types.Block, state *rtstorage.TrieS
 		BestBlock:      true,
 	}
 
-	s.net.SendMessage(msg)
+	s.net.GossipMessage(msg)
 	return s.handleBlock(block, state)
 }
 
@@ -553,7 +553,7 @@ func (s *Service) HandleSubmittedExtrinsic(ext types.Extrinsic) error {
 
 	// broadcast transaction
 	msg := &network.TransactionMessage{Extrinsics: []types.Extrinsic{ext}}
-	s.net.SendMessage(msg)
+	s.net.GossipMessage(msg)
 	return nil
 }
 

dot/network/notifications_test.go

@@ -286,7 +286,7 @@ func Test_HandshakeTimeout(t *testing.T) {
 		BestBlockHash:   common.Hash{1},
 		GenesisHash:     common.Hash{2},
 	}
-	nodeA.SendMessage(testHandshakeMsg)
+	nodeA.GossipMessage(testHandshakeMsg)
 
 	go nodeA.sendData(nodeB.host.id(), testHandshakeMsg, info, nil)
 

dot/network/service.go

@@ -480,8 +480,8 @@ func (s *Service) IsStopped() bool {
 	return s.ctx.Err() != nil
 }
 
-// SendMessage implementation of interface to handle receiving messages
-func (s *Service) SendMessage(msg NotificationsMessage) {
+// GossipMessage gossips a notifications protocol message to our peers
+func (s *Service) GossipMessage(msg NotificationsMessage) {
 	if s.host == nil || msg == nil || s.IsStopped() {
 		return
 	}
@@ -509,6 +509,23 @@ func (s *Service) SendMessage(msg NotificationsMessage) {
 	logger.Error("message not supported by any notifications protocol", "msg type", msg.Type())
 }
 
+// SendMessage sends a message to the given peer
+func (s *Service) SendMessage(to peer.ID, msg NotificationsMessage) error {
+	s.notificationsMu.Lock()
+	defer s.notificationsMu.Unlock()
+	for msgID, prtl := range s.notificationsProtocols {
+		if msg.Type() != msgID || prtl == nil {
+			continue
+		}
+		hs, err := prtl.getHandshake()
+		if err != nil {
+			return err
+		}
+		s.sendData(to, hs, prtl, msg)
+	}
+	return errors.New("message not supported by any notifications protocol")
+}
+
 // handleLightStream handles streams with the <protocol-id>/light/2 protocol ID
 func (s *Service) handleLightStream(stream libp2pnetwork.Stream) {
 	s.readStream(stream, s.decodeLightMessage, s.handleLightMsg)

dot/network/service_test.go

@@ -175,7 +175,7 @@ func TestBroadcastMessages(t *testing.T) {
 	require.NoError(t, err)
 
 	// simulate message sent from core service
-	nodeA.SendMessage(testBlockAnnounceMessage)
+	nodeA.GossipMessage(testBlockAnnounceMessage)
 	time.Sleep(time.Second * 2)
 	require.NotNil(t, handler.messages[nodeA.host.id()])
 }
@@ -232,7 +232,7 @@ func TestBroadcastDuplicateMessage(t *testing.T) {
 
 	// Only one message will be sent.
 	for i := 0; i < 5; i++ {
-		nodeA.SendMessage(testBlockAnnounceMessage)
+		nodeA.GossipMessage(testBlockAnnounceMessage)
 		time.Sleep(time.Millisecond * 10)
 	}
 
@@ -243,7 +243,7 @@ func TestBroadcastDuplicateMessage(t *testing.T) {
 
 	// All 5 message will be sent since cache is disabled.
 	for i := 0; i < 5; i++ {
-		nodeA.SendMessage(testBlockAnnounceMessage)
+		nodeA.GossipMessage(testBlockAnnounceMessage)
 		time.Sleep(time.Millisecond * 10)
 	}
 	require.Equal(t, 6, len(handler.messages[nodeA.host.id()]))

dot/rpc/modules/system_test.go

@@ -338,7 +338,7 @@ func setupSystemModule(t *testing.T) *SystemModule {
 
 type mockNetwork struct{}
 
-func (n *mockNetwork) SendMessage(_ network.NotificationsMessage) {}
+func (n *mockNetwork) GossipMessage(_ network.NotificationsMessage) {}
 
 func newCoreService(t *testing.T, srvc *state.Service) *core.Service {
 	// setup service

dot/state/block.go

@@ -94,31 +94,26 @@ func NewBlockStateFromGenesis(db chaindb.Database, header *types.Header) (*Block
 		pruneKeyCh: make(chan *types.Header, pruneKeyBufferSize),
 	}
 
-	err := bs.setArrivalTime(header.Hash(), time.Now())
-	if err != nil {
+	if err := bs.setArrivalTime(header.Hash(), time.Now()); err != nil {
 		return nil, err
 	}
 
-	err = bs.SetHeader(header)
-	if err != nil {
+	if err := bs.SetHeader(header); err != nil {
 		return nil, err
 	}
 
-	err = bs.db.Put(headerHashKey(header.Number.Uint64()), header.Hash().ToBytes())
-	if err != nil {
+	if err := bs.db.Put(headerHashKey(header.Number.Uint64()), header.Hash().ToBytes()); err != nil {
 		return nil, err
 	}
 
-	err = bs.SetBlockBody(header.Hash(), types.NewBody([]byte{}))
-	if err != nil {
+	if err := bs.SetBlockBody(header.Hash(), types.NewBody([]byte{})); err != nil {
 		return nil, err
 	}
 
 	bs.genesisHash = header.Hash()
 
 	// set the latest finalised head to the genesis header
-	err = bs.SetFinalisedHash(bs.genesisHash, 0, 0)
-	if err != nil {
+	if err := bs.SetFinalisedHash(bs.genesisHash, 0, 0); err != nil {
 		return nil, err
 	}
 

lib/grandpa/grandpa.go

@@ -76,7 +76,7 @@ type Service struct {
 	bestFinalCandidate map[uint64]*Vote // map of round number -> best final candidate
 
 	// channels for communication with other services
-	in               chan GrandpaMessage // only used to receive *VoteMessage
+	in               chan *networkVoteMessage // only used to receive *VoteMessage
 	finalisedCh      chan *types.FinalisationInfo
 	finalisedChID    byte
 	neighbourMessage *NeighbourMessage // cached neighbour message
@@ -166,7 +166,7 @@ func NewService(cfg *Config) (*Service, error) {
 		preVotedBlock:      make(map[uint64]*Vote),
 		bestFinalCandidate: make(map[uint64]*Vote),
 		head:               head,
-		in:                 make(chan GrandpaMessage, 128),
+		in:                 make(chan *networkVoteMessage, 128),
 		resumed:            make(chan struct{}),
 		network:            cfg.Network,
 		finalisedCh:        finalisedCh,
@@ -421,7 +421,7 @@ func (s *Service) handleIsPrimary() (bool, error) {
 		return false, fmt.Errorf("failed to encode finalisation message: %w", err)
 	}
 
-	s.network.SendMessage(msg)
+	s.network.GossipMessage(msg)
 	return true, nil
 }
 
@@ -439,42 +439,24 @@ func (s *Service) primaryBroadcastCommitMessage() {
 		logger.Warn("failed to encode finalisation message", "error", err)
 	}
 
-	s.network.SendMessage(msg)
+	s.network.GossipMessage(msg)
 }
 
 // playGrandpaRound executes a round of GRANDPA
 // at the end of this round, a block will be finalised.
 func (s *Service) playGrandpaRound() error {
 	logger.Debug("starting round", "round", s.state.round, "setID", s.state.setID)
-
-	// save start time
-	start := time.Now()
+	ctx, cancel := context.WithCancel(s.ctx)
+	defer cancel()
 
 	isPrimary, err := s.handleIsPrimary()
 	if err != nil {
 		return err
 	}
 
 	logger.Debug("receiving pre-vote messages...")
-
-	go s.receiveMessages(func() bool {
-		if s.paused.Load().(bool) {
-			return true
-		}
-
-		end := start.Add(interval * 2)
-
-		// ignore err, since if round isn't completable then this will continue
-		completable, _ := s.isCompletable()
-
-		if time.Since(end) >= 0 || completable {
-			return true
-		}
-
-		return false
-	})
-
-	time.Sleep(interval * 2)
+	go s.receiveMessages(ctx)
+	time.Sleep(interval)
 
 	if s.paused.Load().(bool) {
 		return ErrServicePaused
@@ -494,28 +476,14 @@ func (s *Service) playGrandpaRound() error {
 	if !isPrimary {
 		s.prevotes.Store(s.publicKeyBytes(), spv)
 	}
-	logger.Debug("sending pre-vote message...", "vote", pv)
 
+	logger.Debug("sending pre-vote message...", "vote", pv)
 	roundComplete := make(chan struct{})
 
 	// continue to send prevote messages until round is done
 	go s.sendVoteMessage(prevote, vm, roundComplete)
 
 	logger.Debug("receiving pre-commit messages...")
-
-	go s.receiveMessages(func() bool {
-		end := start.Add(interval * 4)
-
-		// ignore err, since if round isn't completable then this will continue
-		completable, _ := s.isCompletable()
-
-		if time.Since(end) >= 0 || completable {
-			return true
-		}
-
-		return false
-	})
-
 	time.Sleep(interval)
 
 	if s.paused.Load().(bool) {
@@ -539,44 +507,6 @@ func (s *Service) playGrandpaRound() error {
 	// continue to send precommit messages until round is done
 	go s.sendVoteMessage(precommit, pcm, roundComplete)
 
-	go func() {
-		// receive messages until current round is completable and previous round is finalisable
-		// and the last finalised block is greater than the best final candidate from the previous round
-		s.receiveMessages(func() bool {
-			if s.paused.Load().(bool) {
-				return true
-			}
-
-			completable, err := s.isCompletable() //nolint
-			if err != nil {
-				return false
-			}
-
-			round := s.state.round
-			finalisable, err := s.isFinalisable(round)
-			if err != nil {
-				return false
-			}
-
-			s.mapLock.Lock()
-			prevBfc := s.bestFinalCandidate[s.state.round-1]
-			s.mapLock.Unlock()
-
-			// this shouldn't happen as long as playGrandpaRound is called through initiate
-			if prevBfc == nil {
-				return false
-			}
-
-			if completable && finalisable && uint32(s.head.Number.Int64()) >= prevBfc.Number {
-				return true
-			}
-
-			return false
-		})
-	}()
-
-	time.Sleep(interval)
-
 	err = s.attemptToFinalize()
 	if err != nil {
 		logger.Error("failed to finalise", "error", err)
@@ -671,7 +601,7 @@ func (s *Service) attemptToFinalize() error {
 		}
 
 		logger.Debug("sending CommitMessage", "msg", cm)
-		s.network.SendMessage(msg)
+		s.network.GossipMessage(msg)
 		return nil
 	}
 }
@@ -947,10 +877,6 @@ func (s *Service) getBestFinalCandidate() (*Vote, error) {
 
 	// if there are multiple blocks, get the one with the highest number
 	// that is also an ancestor of the prevoted block (or is the prevoted block)
-	if blocks[prevoted.Hash] != 0 {
-		return &prevoted, nil
-	}
-
 	bfc := &Vote{
 		Number: 0,
 	}

lib/grandpa/message_handler.go

@@ -56,7 +56,10 @@ func (h *MessageHandler) handleMessage(from peer.ID, m GrandpaMessage) (network.
 		vm, ok := m.(*VoteMessage)
 		if h.grandpa != nil && ok {
 			// send vote message to grandpa service
-			h.grandpa.in <- vm
+			h.grandpa.in <- &networkVoteMessage{
+				from: from,
+				msg:  vm,
+			}
 		}
 		return nil, nil
 	case commitType:

lib/grandpa/message_handler_test.go

@@ -182,7 +182,7 @@ func TestMessageHandler_VoteMessage(t *testing.T) {
 
 	select {
 	case vote := <-gs.in:
-		require.Equal(t, vm, vote)
+		require.Equal(t, vm, vote.msg)
 	case <-time.After(time.Second):
 		t.Fatal("did not receive VoteMessage")
 	}