Skip to content

Commit bcc7935

Browse files
nootnoot
authored
fix(dot/network, lib/grandpa): fix node sync, improve devnet finality
1 parent ce9a8a4 commit bcc7935

26 files changed

+924
-500
lines changed

‎dot/digest/digest_test.go

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -138,14 +138,14 @@ func TestHandler_GrandpaScheduledChange(t *testing.T) {
138138
require.NoError(t, err)
139139

140140
headers := addTestBlocksToState(t, 2, handler.blockState)
141-
for _, h := range headers {
142-
handler.blockState.(*state.BlockState).SetFinalizedHash(h.Hash(), 0, 0)
141+
for i, h := range headers {
142+
handler.blockState.(*state.BlockState).SetFinalizedHash(h.Hash(), uint64(i), 0)
143143
}
144144

145145
// authorities should change on start of block 3 from start
146146
headers = addTestBlocksToState(t, 1, handler.blockState)
147147
for _, h := range headers {
148-
handler.blockState.(*state.BlockState).SetFinalizedHash(h.Hash(), 0, 0)
148+
handler.blockState.(*state.BlockState).SetFinalizedHash(h.Hash(), 3, 0)
149149
}
150150

151151
time.Sleep(time.Millisecond * 500)
@@ -231,8 +231,8 @@ func TestHandler_GrandpaPauseAndResume(t *testing.T) {
231231
require.Equal(t, big.NewInt(int64(p.Delay)), nextPause)
232232

233233
headers := addTestBlocksToState(t, 3, handler.blockState)
234-
for _, h := range headers {
235-
handler.blockState.(*state.BlockState).SetFinalizedHash(h.Hash(), 0, 0)
234+
for i, h := range headers {
235+
handler.blockState.(*state.BlockState).SetFinalizedHash(h.Hash(), uint64(i), 0)
236236
}
237237

238238
time.Sleep(time.Millisecond * 100)

‎dot/network/notifications.go

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -281,7 +281,12 @@ func (s *Service) sendData(peer peer.ID, hs Handshake, info *notificationsProtoc
281281
return
282282
}
283283

284-
if !added {
284+
// TODO: ensure grandpa stores *all* previously received votes and discards them
285+
// only when they are for already finalised rounds; currently this causes issues
286+
// because a vote might be received slightly too early, causing a round mismatch err,
287+
// causing grandpa to discard the vote.
288+
_, isConsensusMsg := msg.(*ConsensusMessage)
289+
if !added && !isConsensusMsg {
285290
return
286291
}
287292
}

‎dot/network/sync.go

Lines changed: 28 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,16 @@ import (
3636
"github.com/libp2p/go-libp2p-core/peer"
3737
)
3838

39+
// SendBlockReqestByHash sends a block request to the network with the given block hash
40+
func (s *Service) SendBlockReqestByHash(hash common.Hash) {
41+
req := createBlockRequestWithHash(hash, blockRequestSize)
42+
s.syncQueue.requestDataByHash.Delete(hash)
43+
s.syncQueue.trySync(&syncRequest{
44+
req: req,
45+
to: "",
46+
})
47+
}
48+
3949
// handleSyncStream handles streams with the <protocol-id>/sync/2 protocol ID
4050
func (s *Service) handleSyncStream(stream libp2pnetwork.Stream) {
4151
if stream == nil {
@@ -537,7 +547,11 @@ func (q *syncQueue) pushResponse(resp *BlockResponseMessage, pid peer.ID) error
537547
}
538548

539549
q.responses = sortResponses(q.responses)
540-
logger.Debug("pushed block data to queue", "start", start, "end", end, "queue", q.stringifyResponseQueue())
550+
logger.Debug("pushed block data to queue", "start", start, "end", end,
551+
"start hash", q.responses[0].Hash,
552+
"end hash", q.responses[len(q.responses)-1].Hash,
553+
"queue", q.stringifyResponseQueue(),
554+
)
541555
return nil
542556
}
543557

@@ -611,9 +625,10 @@ func (q *syncQueue) trySync(req *syncRequest) {
611625
logger.Trace("trying peers in prioritised order...")
612626
syncPeers := q.getSortedPeers()
613627

614-
for _, peer := range syncPeers {
628+
for i, peer := range syncPeers {
615629
// if peer doesn't respond multiple times, then ignore them TODO: determine best values for this
616-
if peer.score <= badPeerThreshold {
630+
// TODO: if we only have a few peers, should we do this check at all?
631+
if peer.score <= badPeerThreshold && i > q.s.cfg.MinPeers {
617632
break
618633
}
619634

@@ -647,9 +662,6 @@ func (q *syncQueue) trySync(req *syncRequest) {
647662

648663
q.justificationRequestData.Store(startingBlockHash, reqdata)
649664
}
650-
651-
req.to = ""
652-
q.requestCh <- req
653665
}
654666

655667
func (q *syncQueue) syncWithPeer(peer peer.ID, req *BlockRequestMessage) (*BlockResponseMessage, error) {
@@ -737,7 +749,7 @@ func (q *syncQueue) handleBlockData(data []*types.BlockData) {
737749

738750
end := data[len(data)-1].Number().Int64()
739751
if end <= finalised.Number.Int64() {
740-
logger.Debug("ignoring block data that is below our head", "got", end, "head", finalised.Number.Int64())
752+
logger.Debug("ignoring block data that is below our finalised head", "got", end, "head", finalised.Number.Int64())
741753
q.pushRequest(uint64(end+1), blockRequestBufferSize, "")
742754
return
743755
}
@@ -844,21 +856,16 @@ func (q *syncQueue) handleBlockAnnounce(msg *BlockAnnounceMessage, from peer.ID)
844856
return
845857
}
846858

847-
if header.Number.Int64() <= q.goal {
848-
return
859+
if header.Number.Int64() > q.goal {
860+
q.goal = header.Number.Int64()
849861
}
850862

851-
q.goal = header.Number.Int64()
852-
853-
bestNum, err := q.s.blockState.BestBlockNumber()
854-
if err != nil {
855-
logger.Error("failed to get best block number", "error", err)
856-
return
863+
req := createBlockRequestWithHash(header.Hash(), blockRequestSize)
864+
q.requestDataByHash.Delete(req)
865+
q.requestCh <- &syncRequest{
866+
req: req,
867+
to: from,
857868
}
858-
859-
// TODO: if we're at the head, this should request by hash instead of number, since there will
860-
// certainly be blocks with the same number.
861-
q.pushRequest(uint64(bestNum.Int64()+1), blockRequestBufferSize, from)
862869
}
863870

864871
func createBlockRequest(startInt int64, size uint32) *BlockRequestMessage {
@@ -875,7 +882,7 @@ func createBlockRequest(startInt int64, size uint32) *BlockRequestMessage {
875882
RequestedData: RequestedDataHeader + RequestedDataBody + RequestedDataJustification,
876883
StartingBlock: start,
877884
EndBlockHash: optional.NewHash(false, common.Hash{}),
878-
Direction: 0, // ascending
885+
Direction: 0, // TODO: define this somewhere
879886
Max: max,
880887
}
881888

@@ -896,7 +903,7 @@ func createBlockRequestWithHash(startHash common.Hash, size uint32) *BlockReques
896903
RequestedData: RequestedDataHeader + RequestedDataBody + RequestedDataJustification,
897904
StartingBlock: start,
898905
EndBlockHash: optional.NewHash(false, common.Hash{}),
899-
Direction: 0, // ascending
906+
Direction: 0, // TODO: define this somewhere
900907
Max: max,
901908
}
902909

‎dot/network/sync_test.go

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -271,11 +271,12 @@ func TestSyncQueue_HandleBlockAnnounce(t *testing.T) {
271271
require.True(t, ok)
272272
require.Equal(t, 1, score.(int))
273273
require.Equal(t, testBlockAnnounceMessage.Number.Int64(), q.goal)
274-
require.Equal(t, 6, len(q.requestCh))
274+
require.Equal(t, 1, len(q.requestCh))
275275

276-
head, err := q.s.blockState.BestBlockNumber()
277-
require.NoError(t, err)
278-
expected := createBlockRequest(head.Int64(), blockRequestSize)
276+
header := &types.Header{
277+
Number: testBlockAnnounceMessage.Number,
278+
}
279+
expected := createBlockRequestWithHash(header.Hash(), blockRequestSize)
279280
req := <-q.requestCh
280281
require.Equal(t, &syncRequest{req: expected, to: testPeerID}, req)
281282
}

‎dot/state/block.go

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -432,8 +432,9 @@ func (bs *BlockState) SetFinalizedHash(hash common.Hash, round, setID uint64) er
432432
}
433433
}
434434

435-
go bs.notifyFinalized(hash, round, setID)
436435
if round > 0 {
436+
go bs.notifyFinalized(hash, round, setID)
437+
437438
err := bs.SetRound(round)
438439
if err != nil {
439440
return err
@@ -452,7 +453,7 @@ func (bs *BlockState) SetFinalizedHash(hash common.Hash, round, setID uint64) er
452453
return err
453454
}
454455

455-
logger.Trace("pruned block", "hash", rem)
456+
logger.Trace("pruned block", "hash", rem, "number", header.Number)
456457
bs.pruneKeyCh <- header
457458
}
458459

‎dot/state/block_notify_test.go

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -61,7 +61,7 @@ func TestFinalizedChannel(t *testing.T) {
6161
chain, _ := AddBlocksToState(t, bs, 3)
6262

6363
for _, b := range chain {
64-
bs.SetFinalizedHash(b.Hash(), 0, 0)
64+
bs.SetFinalizedHash(b.Hash(), 1, 0)
6565
}
6666

6767
for i := 0; i < 1; i++ {
@@ -146,7 +146,7 @@ func TestFinalizedChannel_Multi(t *testing.T) {
146146
}
147147

148148
time.Sleep(time.Millisecond * 10)
149-
bs.SetFinalizedHash(chain[0].Hash(), 0, 0)
149+
bs.SetFinalizedHash(chain[0].Hash(), 1, 0)
150150
wg.Wait()
151151

152152
for _, id := range ids {

‎dot/sync/message.go

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -101,16 +101,16 @@ func (s *Service) CreateBlockResponse(blockRequest *network.BlockRequestMessage)
101101
responseData := []*types.BlockData{}
102102

103103
switch blockRequest.Direction {
104-
case 0: // ascending (ie child to parent)
105-
for i := endHeader.Number.Int64(); i >= startHeader.Number.Int64(); i-- {
104+
case 0: // ascending (ie parent to child)
105+
for i := startHeader.Number.Int64(); i <= endHeader.Number.Int64(); i++ {
106106
blockData, err := s.getBlockData(big.NewInt(i), blockRequest.RequestedData)
107107
if err != nil {
108108
return nil, err
109109
}
110110
responseData = append(responseData, blockData)
111111
}
112-
case 1: // descending (ie parent to child)
113-
for i := startHeader.Number.Int64(); i <= endHeader.Number.Int64(); i++ {
112+
case 1: // descending (ie child to parent)
113+
for i := endHeader.Number.Int64(); i >= startHeader.Number.Int64(); i-- {
114114
blockData, err := s.getBlockData(big.NewInt(i), blockRequest.RequestedData)
115115
if err != nil {
116116
return nil, err

‎dot/sync/message_test.go

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@ func TestService_CreateBlockResponse_MaxSize(t *testing.T) {
4848
RequestedData: 3,
4949
StartingBlock: start,
5050
EndBlockHash: optional.NewHash(false, common.Hash{}),
51-
Direction: 1,
51+
Direction: 0,
5252
Max: optional.NewUint32(false, 0),
5353
}
5454

@@ -62,7 +62,7 @@ func TestService_CreateBlockResponse_MaxSize(t *testing.T) {
6262
RequestedData: 3,
6363
StartingBlock: start,
6464
EndBlockHash: optional.NewHash(false, common.Hash{}),
65-
Direction: 1,
65+
Direction: 0,
6666
Max: optional.NewUint32(true, maxResponseSize+100),
6767
}
6868

@@ -87,7 +87,7 @@ func TestService_CreateBlockResponse_StartHash(t *testing.T) {
8787
RequestedData: 3,
8888
StartingBlock: start,
8989
EndBlockHash: optional.NewHash(false, common.Hash{}),
90-
Direction: 1,
90+
Direction: 0,
9191
Max: optional.NewUint32(false, 0),
9292
}
9393

@@ -98,7 +98,7 @@ func TestService_CreateBlockResponse_StartHash(t *testing.T) {
9898
require.Equal(t, big.NewInt(128), resp.BlockData[127].Number())
9999
}
100100

101-
func TestService_CreateBlockResponse_Ascending(t *testing.T) {
101+
func TestService_CreateBlockResponse_Descending(t *testing.T) {
102102
s := NewTestSyncer(t, false)
103103
addTestBlocksToState(t, int(maxResponseSize), s.blockState)
104104

@@ -112,7 +112,7 @@ func TestService_CreateBlockResponse_Ascending(t *testing.T) {
112112
RequestedData: 3,
113113
StartingBlock: start,
114114
EndBlockHash: optional.NewHash(false, common.Hash{}),
115-
Direction: 0,
115+
Direction: 1,
116116
Max: optional.NewUint32(false, 0),
117117
}
118118

@@ -169,7 +169,7 @@ func TestService_CreateBlockResponse(t *testing.T) {
169169
RequestedData: 3,
170170
StartingBlock: start,
171171
EndBlockHash: optional.NewHash(true, endHash),
172-
Direction: 1,
172+
Direction: 0,
173173
Max: optional.NewUint32(false, 0),
174174
},
175175
expectedMsgValue: &network.BlockResponseMessage{
@@ -188,7 +188,7 @@ func TestService_CreateBlockResponse(t *testing.T) {
188188
RequestedData: 1,
189189
StartingBlock: start,
190190
EndBlockHash: optional.NewHash(true, endHash),
191-
Direction: 1,
191+
Direction: 0,
192192
Max: optional.NewUint32(false, 0),
193193
},
194194
expectedMsgValue: &network.BlockResponseMessage{
@@ -207,7 +207,7 @@ func TestService_CreateBlockResponse(t *testing.T) {
207207
RequestedData: 4,
208208
StartingBlock: start,
209209
EndBlockHash: optional.NewHash(true, endHash),
210-
Direction: 1,
210+
Direction: 0,
211211
Max: optional.NewUint32(false, 0),
212212
},
213213
expectedMsgValue: &network.BlockResponseMessage{
@@ -227,7 +227,7 @@ func TestService_CreateBlockResponse(t *testing.T) {
227227
RequestedData: 8,
228228
StartingBlock: start,
229229
EndBlockHash: optional.NewHash(true, endHash),
230-
Direction: 1,
230+
Direction: 0,
231231
Max: optional.NewUint32(false, 0),
232232
},
233233
expectedMsgValue: &network.BlockResponseMessage{

‎go.sum

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -273,6 +273,7 @@ github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfb
273273
github.com/golang/mock v1.2.0/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
274274
github.com/golang/mock v1.3.1/go.mod h1:sBzyDLLjw3U8JLTeZvSv8jJB+tU5PVekmnlKIyFUx0Y=
275275
github.com/golang/mock v1.4.0/go.mod h1:UOMv5ysSaYNkG+OFQykRIcU/QvvxJf3p21QfJ2Bt3cw=
276+
github.com/golang/mock v1.4.4 h1:l75CXGRSwbaYNpl/Z2X1XIIAMSCquvXgpVZDhwEIJsc=
276277
github.com/golang/mock v1.4.4/go.mod h1:l3mdAwkq5BuhzHwde/uurv3sEJeZMXNpwsxVWU71h+4=
277278
github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
278279
github.com/golang/protobuf v1.3.0/go.mod h1:Qd/q+1AKNOZr9uGQzbzCmRO6sUih6GTPZv6a1/R87v0=

‎lib/common/hash.go

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,9 @@ const (
3535
// Hash used to store a blake2b hash
3636
type Hash [32]byte
3737

38+
// EmptyHash is an empty [32]byte{}
39+
var EmptyHash = Hash{}
40+
3841
// NewHash casts a byte array to a Hash
3942
// if the input is longer than 32 bytes, it takes the first 32 bytes
4043
func NewHash(in []byte) (res Hash) {

0 commit comments

Comments
 (0)