chore(lib/grandpa): update grandpa message tracker to store messages with map of maps, also store commit messages

dot/network/sync.go

@@ -53,10 +53,10 @@ func (s *Service) handleSyncStream(stream libp2pnetwork.Stream) {
 		return
 	}
 
-	s.readStream(stream, s.decodeSyncMessage, s.handleSyncMessage)
+	s.readStream(stream, decodeSyncMessage, s.handleSyncMessage)
 }
 
-func (s *Service) decodeSyncMessage(in []byte, peer peer.ID, inbound bool) (Message, error) {
+func decodeSyncMessage(in []byte, _ peer.ID, _ bool) (Message, error) {
 	msg := new(BlockRequestMessage)
 	err := msg.Decode(in)
 	return msg, err
@@ -198,7 +198,7 @@ func (q *syncQueue) syncAtHead() {
 
 	for {
 		select {
-		// sleep for average block time TODO: make this configurable from slot duration
+		// sleep for average block time
 		case <-t.C:
 		case <-q.ctx.Done():
 			return
@@ -209,7 +209,7 @@ func (q *syncQueue) syncAtHead() {
 			continue
 		}
 
-		goal := atomic.LoadInt64(&q.goal)
+		goal := atomic.LoadInt64(&q.goal) - int64(blockRequestSize)
 
 		// we aren't at the head yet, sleep
 		if curr.Number.Int64() < goal && curr.Number.Cmp(prev.Number) > 0 {
@@ -230,7 +230,7 @@ func (q *syncQueue) syncAtHead() {
 
 		prev = curr
 		start := uint64(curr.Number.Int64()) + 1
-		logger.Debug("haven't received new blocks since last check, pushing request", "start", start)
+		logger.Debug("pushing request for blocks...", "start", start)
 		q.requestData.Delete(start)
 		q.pushRequest(start, 1, "")
 	}
@@ -444,35 +444,29 @@ func (q *syncQueue) pushRequest(start uint64, numRequests int, to peer.ID) {
 	goal := atomic.LoadInt64(&q.goal)
 	if goal < best.Int64() {
 		atomic.StoreInt64(&q.goal, best.Int64())
+		return
 	}
 
-	goal = atomic.LoadInt64(&q.goal)
-	if goal-int64(start) < int64(blockRequestSize) {
-		start := best.Int64() + 1
-		req := createBlockRequest(start, 0)
-
-		logger.Debug("pushing request to queue", "start", start)
-		q.requestData.Store(start, requestData{
-			received: false,
-		})
-
-		q.requestCh <- &syncRequest{
-			req: req,
-			to:  to,
-		}
+	if goal == best.Int64() {
 		return
 	}
 
-	// all requests must start at a multiple of 128 + 1
-	m := start % uint64(blockRequestSize)
-	start = start - m + 1
+	reqSize := blockRequestSize
+	if goal-int64(start) < int64(blockRequestSize) {
+		start = best.Uint64() + 1
+		reqSize = uint32(goal) - uint32(start)
+	} else {
+		// all requests must start at a multiple of 128 + 1
+		m := start % uint64(blockRequestSize)
+		start = start - m + 1
+	}
 
 	for i := 0; i < numRequests; i++ {
 		if start > uint64(goal) {
 			return
 		}
 
-		req := createBlockRequest(int64(start), blockRequestSize)
+		req := createBlockRequest(int64(start), reqSize)
 
 		if d, has := q.requestData.Load(start); has {
 			data := d.(requestData)

dot/network/sync_test.go

@@ -59,17 +59,11 @@ func createBlockRequests(start, end int64) []*BlockRequestMessage {
 }
 
 func TestDecodeSyncMessage(t *testing.T) {
-	s := &Service{
-		ctx: context.Background(),
-	}
-
-	s.syncQueue = newSyncQueue(s)
 	testPeer := peer.ID("noot")
-
 	reqEnc, err := testBlockRequestMessage.Encode()
 	require.NoError(t, err)
 
-	msg, err := s.decodeSyncMessage(reqEnc, testPeer, true)
+	msg, err := decodeSyncMessage(reqEnc, testPeer, true)
 	require.NoError(t, err)
 
 	req, ok := msg.(*BlockRequestMessage)
@@ -468,6 +462,7 @@ func TestSyncQueue_SyncAtHead(t *testing.T) {
 	time.Sleep(time.Second)
 	q.ctx = context.Background()
 	q.slotDuration = time.Millisecond * 100
+	q.goal = 129
 
 	go q.syncAtHead()
 	time.Sleep(q.slotDuration * 3)
@@ -484,9 +479,9 @@ func TestSyncQueue_PushRequest_NearHead(t *testing.T) {
 	q.stop()
 	time.Sleep(time.Second)
 	q.ctx = context.Background()
-	q.goal = 0
+	q.goal = 129
 
-	q.pushRequest(1, 1, "")
+	q.pushRequest(2, 1, "")
 	select {
 	case req := <-q.requestCh:
 		require.Equal(t, uint64(2), req.req.StartingBlock.Uint64())

lib/grandpa/errors.go

@@ -103,4 +103,5 @@ var (
 
 	errVoteExists              = errors.New("already have vote")
 	errVoteToSignatureMismatch = errors.New("votes and authority count mismatch")
+	errInvalidVoteBlock        = errors.New("block in vote is not descendant of previously finalised block")
 )

lib/grandpa/grandpa.go

@@ -326,7 +326,7 @@ func (s *Service) initiateRound() error {
 	s.precommits = new(sync.Map)
 	s.pvEquivocations = make(map[ed25519.PublicKeyBytes][]*SignedVote)
 	s.pcEquivocations = make(map[ed25519.PublicKeyBytes][]*SignedVote)
-	s.tracker, err = newTracker(s.blockState, s.in)
+	s.tracker, err = newTracker(s.blockState, s.messageHandler)
 	if err != nil {
 		return err
 	}

lib/grandpa/message_handler.go

@@ -18,12 +18,14 @@ package grandpa
 
 import (
 	"bytes"
+	"errors"
 	"fmt"
 	"math/big"
 	"reflect"
 
 	"github.com/ChainSafe/gossamer/dot/network"
 	"github.com/ChainSafe/gossamer/dot/types"
+	"github.com/ChainSafe/gossamer/lib/blocktree"
 	"github.com/ChainSafe/gossamer/lib/common"
 	"github.com/ChainSafe/gossamer/lib/crypto/ed25519"
 	"github.com/ChainSafe/gossamer/lib/scale"
@@ -51,41 +53,26 @@ func NewMessageHandler(grandpa *Service, blockState BlockState) *MessageHandler
 func (h *MessageHandler) handleMessage(from peer.ID, m GrandpaMessage) (network.NotificationsMessage, error) {
 	logger.Trace("handling grandpa message", "msg", m)
 
-	switch m.Type() {
-	case voteType:
-		vm, ok := m.(*VoteMessage)
-		if h.grandpa != nil && ok {
-			// send vote message to grandpa service
-			h.grandpa.in <- &networkVoteMessage{
-				from: from,
-				msg:  vm,
-			}
-		}
-		return nil, nil
-	case commitType:
-		if fm, ok := m.(*CommitMessage); ok {
-			return nil, h.handleCommitMessage(fm)
-		}
-	case neighbourType:
-		nm, ok := m.(*NeighbourMessage)
-		if !ok {
-			return nil, nil
+	switch msg := m.(type) {
+	case *VoteMessage:
+		// send vote message to grandpa service
+		h.grandpa.in <- &networkVoteMessage{
+			from: from,
+			msg:  msg,
 		}
 
-		return nil, h.handleNeighbourMessage(from, nm)
-	case catchUpRequestType:
-		if r, ok := m.(*catchUpRequest); ok {
-			return h.handleCatchUpRequest(r)
-		}
-	case catchUpResponseType:
-		if r, ok := m.(*catchUpResponse); ok {
-			return nil, h.handleCatchUpResponse(r)
-		}
+		return nil, nil
+	case *CommitMessage:
+		return nil, h.handleCommitMessage(msg)
+	case *NeighbourMessage:
+		return nil, h.handleNeighbourMessage(from, msg)
+	case *catchUpRequest:
+		return h.handleCatchUpRequest(msg)
+	case *catchUpResponse:
+		return nil, h.handleCatchUpResponse(msg)
 	default:
 		return nil, ErrInvalidMessageType
 	}
-
-	return nil, nil
 }
 
 func (h *MessageHandler) handleNeighbourMessage(from peer.ID, msg *NeighbourMessage) error {
@@ -125,6 +112,11 @@ func (h *MessageHandler) handleCommitMessage(msg *CommitMessage) error {
 
 	// check justification here
 	if err := h.verifyCommitMessageJustification(msg); err != nil {
+		if errors.Is(err, blocktree.ErrStartNodeNotFound) {
+			// TODO: make this synchronous
+			go h.grandpa.network.SendBlockReqestByHash(msg.Vote.Hash)
+			h.grandpa.tracker.addCommit(msg)
+		}
 		return err
 	}
 

lib/grandpa/message_tracker.go

@@ -21,35 +21,38 @@ import (
 
 	"github.com/ChainSafe/gossamer/dot/types"
 	"github.com/ChainSafe/gossamer/lib/common"
+	"github.com/ChainSafe/gossamer/lib/crypto/ed25519"
 )
 
 // tracker keeps track of messages that have been received that have failed to validate with ErrBlockDoesNotExist
 // these messages may be needed again in the case that we are slightly out of sync with the rest of the network
 type tracker struct {
-	blockState BlockState
-	messages   map[common.Hash][]*networkVoteMessage // map of vote block hash -> array of VoteMessages for that hash
-	mapLock    sync.Mutex
-	in         chan *types.Block          // receive imported block from BlockState
-	chanID     byte                       // BlockState channel ID
-	out        chan<- *networkVoteMessage // send a VoteMessage back to grandpa. corresponds to grandpa's in channel
-	stopped    chan struct{}
+	blockState     BlockState
+	handler        *MessageHandler
+	voteMessages   map[common.Hash]map[ed25519.PublicKeyBytes]*networkVoteMessage // map of vote block hash -> array of VoteMessages for that hash
+	commitMessages map[common.Hash]*CommitMessage                                 // map of commit block hash to commit message
+	mapLock        sync.Mutex
+	in             chan *types.Block // receive imported block from BlockState
+	chanID         byte              // BlockState channel ID
+	stopped        chan struct{}
 }
 
-func newTracker(bs BlockState, out chan<- *networkVoteMessage) (*tracker, error) {
+func newTracker(bs BlockState, handler *MessageHandler) (*tracker, error) {
 	in := make(chan *types.Block, 16)
 	id, err := bs.RegisterImportedChannel(in)
 	if err != nil {
 		return nil, err
 	}
 
 	return &tracker{
-		blockState: bs,
-		messages:   make(map[common.Hash][]*networkVoteMessage),
-		mapLock:    sync.Mutex{},
-		in:         in,
-		chanID:     id,
-		out:        out,
-		stopped:    make(chan struct{}),
+		blockState:     bs,
+		handler:        handler,
+		voteMessages:   make(map[common.Hash]map[ed25519.PublicKeyBytes]*networkVoteMessage),
+		commitMessages: make(map[common.Hash]*CommitMessage),
+		mapLock:        sync.Mutex{},
+		in:             in,
+		chanID:         id,
+		stopped:        make(chan struct{}),
 	}, nil
 }
 
@@ -63,15 +66,27 @@ func (t *tracker) stop() {
 	close(t.in)
 }
 
-func (t *tracker) add(v *networkVoteMessage) {
+func (t *tracker) addVote(v *networkVoteMessage) {
 	if v.msg == nil {
 		return
 	}
 
 	t.mapLock.Lock()
-	// TODO: change to map of maps, this allows duplicates
-	t.messages[v.msg.Message.Hash] = append(t.messages[v.msg.Message.Hash], v)
-	t.mapLock.Unlock()
+	defer t.mapLock.Unlock()
+
+	msgs, has := t.voteMessages[v.msg.Message.Hash]
+	if !has {
+		msgs = make(map[ed25519.PublicKeyBytes]*networkVoteMessage)
+		t.voteMessages[v.msg.Message.Hash] = msgs
+	}
+
+	msgs[v.msg.Message.AuthorityID] = v
+}
+
+func (t *tracker) addCommit(cm *CommitMessage) {
+	t.mapLock.Lock()
+	defer t.mapLock.Unlock()
+	t.commitMessages[cm.Vote.Hash] = cm
 }
 
 func (t *tracker) handleBlocks() {
@@ -82,18 +97,35 @@ func (t *tracker) handleBlocks() {
 				continue
 			}
 
-			t.mapLock.Lock()
+			t.handleBlock(b)
+		case <-t.stopped:
+			return
+		}
+	}
+}
+
+func (t *tracker) handleBlock(b *types.Block) {
+	t.mapLock.Lock()
+	defer t.mapLock.Unlock()
 
-			h := b.Header.Hash()
-			if t.messages[h] != nil {
-				for _, v := range t.messages[h] {
-					t.out <- v
-				}
+	h := b.Header.Hash()
+	if vms, has := t.voteMessages[h]; has {
+		for _, v := range vms {
+			_, err := t.handler.handleMessage(v.from, v.msg)
+			if err != nil {
+				logger.Warn("failed to handle vote message", "message", v, "error", err)
 			}
+		}
 
-			t.mapLock.Unlock()
-		case <-t.stopped:
-			return
+		delete(t.voteMessages, h)
+	}
+
+	if cm, has := t.commitMessages[h]; has {
+		_, err := t.handler.handleMessage("", cm)
+		if err != nil {
+			logger.Warn("failed to handle commit message", "message", cm, "error", err)
 		}
+
+		delete(t.commitMessages, h)
 	}
 }