fix(dot/sync): rework on bootstrap/tip sync

chain/westend/defaults.go

@@ -29,6 +29,14 @@ func DefaultConfig() *cfg.Config {
 	config.Core.GrandpaAuthority = false
 	config.Core.Role = 1
 	config.Network.NoMDNS = false
+	config.Network.MaxPeers = 128
+	config.PrometheusExternal = true
+	config.PrometheusPort = 9876
 
+	config.Log.Sync = "trace"
+	config.Log.Digest = "trace"
+
+	config.Pprof.Enabled = true
+	config.Pprof.ListeningAddress = "0.0.0.0:6060"
 	return config
 }

dot/network/discovery.go

@@ -28,7 +28,7 @@ var (
 	startDHTTimeout             = time.Second * 10
 	initialAdvertisementTimeout = time.Millisecond
 	tryAdvertiseTimeout         = time.Second * 30
-	connectToPeersTimeout       = time.Minute * 5
+	connectToPeersTimeout       = time.Minute
 	findPeersTimeout            = time.Minute
 )
 
@@ -183,7 +183,7 @@ func (d *discovery) checkPeerCount() {
 		case <-d.ctx.Done():
 			return
 		case <-ticker.C:
-			if len(d.h.Network().Peers()) > d.minPeers {
+			if len(d.h.Network().Peers()) >= d.maxPeers {
 				continue
 			}
 

dot/network/errors.go

@@ -8,6 +8,8 @@ import (
 )
 
 var (
+	ErrReceivedEmptyMessage = errors.New("received empty message")
+
 	errCannotValidateHandshake       = errors.New("failed to validate handshake")
 	errMessageTypeNotValid           = errors.New("message type is not valid")
 	errInvalidHandshakeForPeer       = errors.New("peer previously sent invalid handshake")

dot/network/message.go

@@ -17,6 +17,9 @@ import (
 	"github.com/ChainSafe/gossamer/pkg/scale"
 )
 
+// maxResponseSize is maximum number of block data a BlockResponse message can contain
+const MaxBlocksInResponse = 128
+
 type MessageType byte
 
 // Message types for notifications protocol messages. Used internally to map message to protocol.
@@ -44,6 +47,9 @@ const (
 	RequestedDataReceipt       = byte(4)
 	RequestedDataMessageQueue  = byte(8)
 	RequestedDataJustification = byte(16)
+	BootstrapRequestData       = RequestedDataHeader +
+		RequestedDataBody +
+		RequestedDataJustification
 )
 
 var _ Message = (*BlockRequestMessage)(nil)
@@ -354,3 +360,79 @@ func (cm *ConsensusMessage) Hash() (common.Hash, error) {
 	}
 	return common.Blake2bHash(encMsg)
 }
+
+// NewSingleBlockRequestMessage returns a request to retrieve a single block
+func NewSingleBlockRequestMessage(blockHash common.Hash, requestedData byte) *BlockRequestMessage {
+	one := uint32(1)
+	return &BlockRequestMessage{
+		RequestedData: requestedData,
+		StartingBlock: *variadic.MustNewUint32OrHash(blockHash),
+		Direction:     Descending,
+		Max:           &one,
+	}
+}
+
+// NewDescendingBlockRequest returns a descending block request message
+func NewDescendingBlockRequest(blockHash common.Hash, amount uint32, requestedData byte) *BlockRequestMessage {
+	return &BlockRequestMessage{
+		RequestedData: requestedData,
+		StartingBlock: *variadic.MustNewUint32OrHash(blockHash),
+		Direction:     Descending,
+		Max:           &amount,
+	}
+}
+
+func NewAscedingBlockRequests(startNumber, targetNumber uint, requestedData byte) []*BlockRequestMessage {
+	if startNumber > targetNumber {
+		return []*BlockRequestMessage{}
+	}
+
+	diff := targetNumber - startNumber
+
+	// start and end block are the same, just request 1 block
+	if diff == 0 {
+		one := uint32(1)
+		return []*BlockRequestMessage{
+			{
+				RequestedData: requestedData,
+				StartingBlock: *variadic.MustNewUint32OrHash(uint32(startNumber)),
+				Direction:     Ascending,
+				Max:           &one,
+			},
+		}
+	}
+
+	numRequests := diff / MaxBlocksInResponse
+	// we should check if the diff is in the maxResponseSize bounds
+	// otherwise we should increase the numRequests by one, take this
+	// example, we want to sync from 0 to 259, the diff is 259
+	// then the num of requests is 2 (uint(259)/uint(128)) however two requests will
+	// retrieve only 256 blocks (each request can retrieve a max of 128 blocks), so we should
+	// create one more request to retrieve those missing blocks, 3 in this example.
+	missingBlocks := diff % MaxBlocksInResponse
+	if missingBlocks != 0 {
+		numRequests++
+	}
+
+	reqs := make([]*BlockRequestMessage, numRequests)
+	for i := uint(0); i < numRequests; i++ {
+		max := uint32(MaxBlocksInResponse)
+
+		lastIteration := numRequests - 1
+		if i == lastIteration && missingBlocks != 0 {
+			max = uint32(missingBlocks)
+		}
+
+		start := variadic.MustNewUint32OrHash(startNumber)
+
+		reqs[i] = &BlockRequestMessage{
+			RequestedData: requestedData,
+			StartingBlock: *start,
+			Direction:     Ascending,
+			Max:           &max,
+		}
+		startNumber += uint(max)
+	}
+
+	return reqs
+}

dot/network/message_test.go

@@ -422,3 +422,122 @@ func TestDecodeConsensusMessage(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, encMsg, encodedMessage)
 }
+
+func TestAscendingBlockRequest(t *testing.T) {
+	one := uint32(1)
+	three := uint32(3)
+	maxResponseSize := uint32(MaxBlocksInResponse)
+	cases := map[string]struct {
+		startNumber, targetNumber   uint
+		expectedBlockRequestMessage []*BlockRequestMessage
+	}{
+		"start_greater_than_target": {
+			startNumber:                 10,
+			targetNumber:                0,
+			expectedBlockRequestMessage: []*BlockRequestMessage{},
+		},
+
+		"no_difference_between_start_and_target": {
+			startNumber:  10,
+			targetNumber: 10,
+			expectedBlockRequestMessage: []*BlockRequestMessage{
+				{
+					RequestedData: BootstrapRequestData,
+					StartingBlock: *variadic.MustNewUint32OrHash(uint32(10)),
+					Direction:     Ascending,
+					Max:           &one,
+				},
+			},
+		},
+
+		"requesting_128_blocks": {
+			startNumber:  0,
+			targetNumber: 128,
+			expectedBlockRequestMessage: []*BlockRequestMessage{
+				{
+					RequestedData: BootstrapRequestData,
+					StartingBlock: *variadic.MustNewUint32OrHash(uint32(0)),
+					Direction:     Ascending,
+					Max:           &maxResponseSize,
+				},
+			},
+		},
+
+		"requesting_4_chunks_of_128_blocks": {
+			startNumber:  0,
+			targetNumber: 512, // 128 * 4
+			expectedBlockRequestMessage: []*BlockRequestMessage{
+				{
+					RequestedData: BootstrapRequestData,
+					StartingBlock: *variadic.MustNewUint32OrHash(uint32(0)),
+					Direction:     Ascending,
+					Max:           &maxResponseSize,
+				},
+				{
+					RequestedData: BootstrapRequestData,
+					StartingBlock: *variadic.MustNewUint32OrHash(uint32(128)),
+					Direction:     Ascending,
+					Max:           &maxResponseSize,
+				},
+				{
+					RequestedData: BootstrapRequestData,
+					StartingBlock: *variadic.MustNewUint32OrHash(uint32(256)),
+					Direction:     Ascending,
+					Max:           &maxResponseSize,
+				},
+				{
+					RequestedData: BootstrapRequestData,
+					StartingBlock: *variadic.MustNewUint32OrHash(uint32(384)),
+					Direction:     Ascending,
+					Max:           &maxResponseSize,
+				},
+			},
+		},
+
+		"requesting_4_chunks_of_128_plus_3_blocks": {
+			startNumber:  0,
+			targetNumber: (128 * 4) + 3,
+			expectedBlockRequestMessage: []*BlockRequestMessage{
+				{
+					RequestedData: BootstrapRequestData,
+					StartingBlock: *variadic.MustNewUint32OrHash(uint32(0)),
+					Direction:     Ascending,
+					Max:           &maxResponseSize,
+				},
+				{
+					RequestedData: BootstrapRequestData,
+					StartingBlock: *variadic.MustNewUint32OrHash(uint32(128)),
+					Direction:     Ascending,
+					Max:           &maxResponseSize,
+				},
+				{
+					RequestedData: BootstrapRequestData,
+					StartingBlock: *variadic.MustNewUint32OrHash(uint32(256)),
+					Direction:     Ascending,
+					Max:           &maxResponseSize,
+				},
+				{
+					RequestedData: BootstrapRequestData,
+					StartingBlock: *variadic.MustNewUint32OrHash(uint32(384)),
+					Direction:     Ascending,
+					Max:           &maxResponseSize,
+				},
+				{
+					RequestedData: BootstrapRequestData,
+					StartingBlock: *variadic.MustNewUint32OrHash(uint32(512)),
+					Direction:     Ascending,
+					Max:           &three,
+				},
+			},
+		},
+	}
+
+	for tname, tt := range cases {
+		tt := tt
+
+		t.Run(tname, func(t *testing.T) {
+			requests := NewAscedingBlockRequests(tt.startNumber, tt.targetNumber, BootstrapRequestData)
+			require.Equal(t, tt.expectedBlockRequestMessage, requests)
+		})
+	}
+}

dot/network/service.go

@@ -605,6 +605,11 @@ func (s *Service) NetworkState() common.NetworkState {
 	}
 }
 
+// AllConnectedPeersID returns all the connected to the node instance
+func (s *Service) AllConnectedPeersID() []peer.ID {
+	return s.host.p2pHost.Network().Peers()
+}
+
 // Peers returns information about connected peers needed for the rpc server
 func (s *Service) Peers() []common.PeerInfo {
 	var peers []common.PeerInfo

dot/network/sync.go

@@ -4,16 +4,10 @@
 package network
 
 import (
-	"time"
-
 	libp2pnetwork "github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
 )
 
-var (
-	BlockRequestTimeout = time.Second * 20
-)
-
 // handleSyncStream handles streams with the <protocol-id>/sync/2 protocol ID
 func (s *Service) handleSyncStream(stream libp2pnetwork.Stream) {
 	if stream == nil {

dot/node_integration_test.go

@@ -57,7 +57,6 @@ func TestNewNode(t *testing.T) {
 	initConfig.Account.Key = "alice"
 	initConfig.Core.Role = common.FullNodeRole
 	initConfig.Core.WasmInterpreter = wasmer.Name
-
 	initConfig.Log.Digest = "critical"
 
 	networkConfig := &network.Config{
@@ -86,7 +85,7 @@ func TestNewNode(t *testing.T) {
 	assert.NoError(t, err)
 
 	mockServiceRegistry := NewMockServiceRegisterer(ctrl)
-	mockServiceRegistry.EXPECT().RegisterService(gomock.Any()).Times(8)
+	mockServiceRegistry.EXPECT().RegisterService(gomock.Any()).Times(9)
 
 	m := NewMocknodeBuilderIface(ctrl)
 	m.EXPECT().isNodeInitialised(initConfig.BasePath).Return(nil)

dot/rpc/modules/gomock_reflect_2744729207/prog.go

@@ -0,0 +1,87 @@
+// Copyright 2023 ChainSafe Systems (ON)
+// SPDX-License-Identifier: LGPL-3.0-only
+
+
+package main
+
+import (
+	"encoding/gob"
+	"flag"
+	"fmt"
+	"os"
+	"path"
+	"reflect"
+
+	"github.com/golang/mock/mockgen/model"
+
+	pkg_ "github.com/ChainSafe/gossamer/dot/rpc/modules"
+)
+
+var output = flag.String("output", "", "The output file name, or empty to use stdout.")
+
+func main() {
+	flag.Parse()
+
+	its := []struct{
+		sym string
+		typ reflect.Type
+	}{
+
+		{ "StorageAPI", reflect.TypeOf((*pkg_.StorageAPI)(nil)).Elem()},
+
+		{ "BlockAPI", reflect.TypeOf((*pkg_.BlockAPI)(nil)).Elem()},
+
+		{ "NetworkAPI", reflect.TypeOf((*pkg_.NetworkAPI)(nil)).Elem()},
+
+		{ "BlockProducerAPI", reflect.TypeOf((*pkg_.BlockProducerAPI)(nil)).Elem()},
+
+		{ "TransactionStateAPI", reflect.TypeOf((*pkg_.TransactionStateAPI)(nil)).Elem()},
+
+		{ "CoreAPI", reflect.TypeOf((*pkg_.CoreAPI)(nil)).Elem()},
+
+		{ "SystemAPI", reflect.TypeOf((*pkg_.SystemAPI)(nil)).Elem()},
+
+		{ "BlockFinalityAPI", reflect.TypeOf((*pkg_.BlockFinalityAPI)(nil)).Elem()},
+
+		{ "RuntimeStorageAPI", reflect.TypeOf((*pkg_.RuntimeStorageAPI)(nil)).Elem()},
+
+		{ "SyncStateAPI", reflect.TypeOf((*pkg_.SyncStateAPI)(nil)).Elem()},
+
+	}
+	pkg := &model.Package{
+		// NOTE: This behaves contrary to documented behaviour if the
+		// package name is not the final component of the import path.
+		// The reflect package doesn't expose the package name, though.
+		Name: path.Base("github.com/ChainSafe/gossamer/dot/rpc/modules"),
+	}
+
+	for _, it := range its {
+		intf, err := model.InterfaceFromInterfaceType(it.typ)
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "Reflection: %v\n", err)
+			os.Exit(1)
+		}
+		intf.Name = it.sym
+		pkg.Interfaces = append(pkg.Interfaces, intf)
+	}
+
+	outfile := os.Stdout
+	if len(*output) != 0 {
+		var err error
+		outfile, err = os.Create(*output)
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "failed to open output file %q", *output)
+		}
+		defer func() {
+			if err := outfile.Close(); err != nil {
+				fmt.Fprintf(os.Stderr, "failed to close output file %q", *output)
+				os.Exit(1)
+			}
+		}()
+	}
+
+	if err := gob.NewEncoder(outfile).Encode(pkg); err != nil {
+		fmt.Fprintf(os.Stderr, "gob encode: %v\n", err)
+		os.Exit(1)
+	}
+}