refactor(lib/sync): introduce fullsync strategy

.github/workflows/integration-tests.yml

@@ -32,7 +32,7 @@ jobs:
             github.com/ChainSafe/gossamer/dot/state,
             github.com/ChainSafe/gossamer/dot/digest,
             github.com/ChainSafe/gossamer/dot/network,
-            github.com/ChainSafe/gossamer/dot/sync,
+            github.com/ChainSafe/gossamer/lib/sync,
             github.com/ChainSafe/gossamer/lib/babe,
             github.com/ChainSafe/gossamer/lib/grandpa,
           ]

dot/core/service_integration_test.go

@@ -15,7 +15,6 @@ import (
 
 	"github.com/ChainSafe/gossamer/dot/network"
 	"github.com/ChainSafe/gossamer/dot/state"
-	"github.com/ChainSafe/gossamer/dot/sync"
 	"github.com/ChainSafe/gossamer/dot/types"
 	"github.com/ChainSafe/gossamer/lib/babe/inherents"
 	"github.com/ChainSafe/gossamer/lib/common"
@@ -210,65 +209,6 @@ func TestHandleChainReorg_NoReorg(t *testing.T) {
 	require.NoError(t, err)
 }
 
-func TestHandleChainReorg_WithReorg_Trans(t *testing.T) {
-	t.Skip() // TODO: tx fails to validate in handleChainReorg() with "Invalid transaction" (#1026)
-	s := NewTestService(t, nil)
-	bs := s.blockState
-
-	parent, err := bs.BestBlockHeader()
-	require.NoError(t, err)
-
-	bestBlockHash := s.blockState.BestBlockHash()
-	rt, err := s.blockState.GetRuntime(bestBlockHash)
-	require.NoError(t, err)
-
-	block1 := sync.BuildBlock(t, rt, parent, nil)
-	bs.StoreRuntime(block1.Header.Hash(), rt)
-	err = bs.AddBlock(block1)
-	require.NoError(t, err)
-
-	block2 := sync.BuildBlock(t, rt, &block1.Header, nil)
-	bs.StoreRuntime(block2.Header.Hash(), rt)
-	err = bs.AddBlock(block2)
-	require.NoError(t, err)
-
-	block3 := sync.BuildBlock(t, rt, &block2.Header, nil)
-	bs.StoreRuntime(block3.Header.Hash(), rt)
-	err = bs.AddBlock(block3)
-	require.NoError(t, err)
-
-	block4 := sync.BuildBlock(t, rt, &block3.Header, nil)
-	bs.StoreRuntime(block4.Header.Hash(), rt)
-	err = bs.AddBlock(block4)
-	require.NoError(t, err)
-
-	block5 := sync.BuildBlock(t, rt, &block4.Header, nil)
-	bs.StoreRuntime(block5.Header.Hash(), rt)
-	err = bs.AddBlock(block5)
-	require.NoError(t, err)
-
-	block31 := sync.BuildBlock(t, rt, &block2.Header, nil)
-	bs.StoreRuntime(block31.Header.Hash(), rt)
-	err = bs.AddBlock(block31)
-	require.NoError(t, err)
-
-	nonce := uint64(0)
-
-	// Add extrinsic to block `block41`
-	ext := createExtrinsic(t, rt, bs.(*state.BlockState).GenesisHash(), nonce)
-
-	block41 := sync.BuildBlock(t, rt, &block31.Header, ext)
-	bs.StoreRuntime(block41.Header.Hash(), rt)
-	err = bs.AddBlock(block41)
-	require.NoError(t, err)
-
-	err = s.handleChainReorg(block41.Header.Hash(), block5.Header.Hash())
-	require.NoError(t, err)
-
-	pending := s.transactionState.(*state.TransactionState).Pending()
-	require.Equal(t, 1, len(pending))
-}
-
 func TestHandleChainReorg_WithReorg_NoTransactions(t *testing.T) {
 	s := NewTestService(t, nil)
 	const height = 5

dot/network/message_test.go

@@ -429,7 +429,7 @@ func TestAscendingBlockRequest(t *testing.T) {
 	three := uint32(3)
 	maxResponseSize := uint32(messages.MaxBlocksInResponse)
 	cases := map[string]struct {
-		startNumber, targetNumber      uint
+		startNumber, targetNumber      uint32
 		expectedBlockRequestMessage    []*messages.BlockRequestMessage
 		expectedTotalOfBlocksRequested uint32
 	}{
@@ -446,7 +446,7 @@ func TestAscendingBlockRequest(t *testing.T) {
 			expectedBlockRequestMessage: []*messages.BlockRequestMessage{
 				{
 					RequestedData: messages.BootstrapRequestData,
-					StartingBlock: *variadic.MustNewUint32OrHash(uint32(10)),
+					StartingBlock: *variadic.Uint32OrHashFrom(uint32(10)),
 					Direction:     messages.Ascending,
 					Max:           &one,
 				},
@@ -461,7 +461,7 @@ func TestAscendingBlockRequest(t *testing.T) {
 			expectedBlockRequestMessage: []*messages.BlockRequestMessage{
 				{
 					RequestedData: messages.BootstrapRequestData,
-					StartingBlock: *variadic.MustNewUint32OrHash(uint32(1)),
+					StartingBlock: *variadic.Uint32OrHashFrom(uint32(1)),
 					Direction:     messages.Ascending,
 					Max:           &maxResponseSize,
 				},
@@ -475,63 +475,62 @@ func TestAscendingBlockRequest(t *testing.T) {
 			expectedBlockRequestMessage: []*messages.BlockRequestMessage{
 				{
 					RequestedData: messages.BootstrapRequestData,
-					StartingBlock: *variadic.MustNewUint32OrHash(uint32(1)),
+					StartingBlock: *variadic.Uint32OrHashFrom(uint32(1)),
 					Direction:     messages.Ascending,
 					Max:           &maxResponseSize,
 				},
 				{
 					RequestedData: messages.BootstrapRequestData,
-					StartingBlock: *variadic.MustNewUint32OrHash(uint32(129)),
+					StartingBlock: *variadic.Uint32OrHashFrom(uint32(129)),
 					Direction:     messages.Ascending,
 					Max:           &maxResponseSize,
 				},
 				{
 					RequestedData: messages.BootstrapRequestData,
-					StartingBlock: *variadic.MustNewUint32OrHash(uint32(257)),
+					StartingBlock: *variadic.Uint32OrHashFrom(uint32(257)),
 					Direction:     messages.Ascending,
 					Max:           &maxResponseSize,
 				},
 				{
 					RequestedData: messages.BootstrapRequestData,
-					StartingBlock: *variadic.MustNewUint32OrHash(uint32(385)),
+					StartingBlock: *variadic.Uint32OrHashFrom(uint32(385)),
 					Direction:     messages.Ascending,
 					Max:           &maxResponseSize,
 				},
 			},
 		},
-
 		"requesting_4_chunks_of_128_plus_3_blocks": {
 			startNumber:                    1,
 			targetNumber:                   (128 * 4) + 3,
 			expectedTotalOfBlocksRequested: 515,
 			expectedBlockRequestMessage: []*messages.BlockRequestMessage{
 				{
 					RequestedData: messages.BootstrapRequestData,
-					StartingBlock: *variadic.MustNewUint32OrHash(uint32(1)),
+					StartingBlock: *variadic.Uint32OrHashFrom(uint32(1)),
 					Direction:     messages.Ascending,
 					Max:           &maxResponseSize,
 				},
 				{
 					RequestedData: messages.BootstrapRequestData,
-					StartingBlock: *variadic.MustNewUint32OrHash(uint32(129)),
+					StartingBlock: *variadic.Uint32OrHashFrom(uint32(129)),
 					Direction:     messages.Ascending,
 					Max:           &maxResponseSize,
 				},
 				{
 					RequestedData: messages.BootstrapRequestData,
-					StartingBlock: *variadic.MustNewUint32OrHash(uint32(257)),
+					StartingBlock: *variadic.Uint32OrHashFrom(uint32(257)),
 					Direction:     messages.Ascending,
 					Max:           &maxResponseSize,
 				},
 				{
 					RequestedData: messages.BootstrapRequestData,
-					StartingBlock: *variadic.MustNewUint32OrHash(uint32(385)),
+					StartingBlock: *variadic.Uint32OrHashFrom(uint32(385)),
 					Direction:     messages.Ascending,
 					Max:           &maxResponseSize,
 				},
 				{
 					RequestedData: messages.BootstrapRequestData,
-					StartingBlock: *variadic.MustNewUint32OrHash(uint32(513)),
+					StartingBlock: *variadic.Uint32OrHashFrom(uint32(513)),
 					Direction:     messages.Ascending,
 					Max:           &three,
 				},
@@ -543,7 +542,8 @@ func TestAscendingBlockRequest(t *testing.T) {
 		tt := tt
 
 		t.Run(tname, func(t *testing.T) {
-			requests := messages.NewAscendingBlockRequests(tt.startNumber, tt.targetNumber, messages.BootstrapRequestData)
+			requests := messages.NewAscendingBlockRequests(tt.startNumber, tt.targetNumber,
+				messages.BootstrapRequestData)
 			require.Equal(t, tt.expectedBlockRequestMessage, requests)
 
 			acc := uint32(0)

dot/network/messages/block.go

@@ -72,7 +72,7 @@ func NewBlockRequest(startingBlock variadic.Uint32OrHash, amount uint32,
 	}
 }
 
-func NewAscendingBlockRequests(startNumber, targetNumber uint, requestedData byte) []*BlockRequestMessage {
+func NewAscendingBlockRequests(startNumber, targetNumber uint32, requestedData byte) []*BlockRequestMessage {
 	if startNumber > targetNumber {
 		return []*BlockRequestMessage{}
 	}
@@ -82,7 +82,7 @@ func NewAscendingBlockRequests(startNumber, targetNumber uint, requestedData byt
 	// start and end block are the same, just request 1 block
 	if diff == 0 {
 		return []*BlockRequestMessage{
-			NewBlockRequest(*variadic.MustNewUint32OrHash(uint32(startNumber)), 1, requestedData, Ascending),
+			NewBlockRequest(*variadic.Uint32OrHashFrom(startNumber), 1, requestedData, Ascending),
 		}
 	}
 
@@ -99,22 +99,27 @@ func NewAscendingBlockRequests(startNumber, targetNumber uint, requestedData byt
 	}
 
 	reqs := make([]*BlockRequestMessage, numRequests)
-	for i := uint(0); i < numRequests; i++ {
+	for i := uint32(0); i < numRequests; i++ {
 		max := uint32(MaxBlocksInResponse)
 
 		lastIteration := numRequests - 1
 		if i == lastIteration && missingBlocks != 0 {
-			max = uint32(missingBlocks)
+			max = missingBlocks
 		}
 
-		start := variadic.MustNewUint32OrHash(startNumber)
+		start := variadic.Uint32OrHashFrom(startNumber)
 		reqs[i] = NewBlockRequest(*start, max, requestedData, Ascending)
-		startNumber += uint(max)
+		startNumber += max
 	}
 
 	return reqs
 }
 
+// RequestField returns true the field passed as argument exists in the request data
+func (bm *BlockRequestMessage) RequestField(field byte) bool {
+	return (bm.RequestedData & field) == field
+}
+
 // String formats a BlockRequestMessage as a string
 func (bm *BlockRequestMessage) String() string {
 	max := uint32(0)

dot/network/service.go

@@ -735,6 +735,7 @@ func (s *Service) processMessage(msg peerset.Message) {
 			return
 		}
 		logger.Debugf("connection dropped successfully for peer %s", peerID)
+		s.syncer.OnConnectionClosed(peerID)
 	}
 }
 

dot/network/state.go

@@ -34,6 +34,8 @@ type Syncer interface {
 
 	// CreateBlockResponse is called upon receipt of a BlockRequestMessage to create the response
 	CreateBlockResponse(peer.ID, *messages.BlockRequestMessage) (*messages.BlockResponseMessage, error)
+
+	OnConnectionClosed(peer.ID)
 }
 
 // TransactionHandler is the interface used by the transactions sub-protocol

dot/node.go

@@ -21,7 +21,6 @@ import (
 	"github.com/ChainSafe/gossamer/dot/rpc"
 	"github.com/ChainSafe/gossamer/dot/state"
 	"github.com/ChainSafe/gossamer/dot/state/pruner"
-	dotsync "github.com/ChainSafe/gossamer/dot/sync"
 	"github.com/ChainSafe/gossamer/dot/system"
 	"github.com/ChainSafe/gossamer/dot/telemetry"
 	"github.com/ChainSafe/gossamer/dot/types"
@@ -64,7 +63,7 @@ type nodeBuilderIface interface {
 		net *network.Service, telemetryMailer Telemetry) (*grandpa.Service, error)
 	newSyncService(config *cfg.Config, st *state.Service, finalityGadget BlockJustificationVerifier,
 		verifier *babe.VerificationManager, cs *core.Service, net *network.Service,
-		telemetryMailer Telemetry) (*dotsync.Service, error)
+		telemetryMailer Telemetry) (network.Syncer, error)
 	createBABEService(config *cfg.Config, st *state.Service, ks KeyStore, cs *core.Service,
 		telemetryMailer Telemetry) (service *babe.Service, err error)
 	createSystemService(cfg *types.SystemInfo, stateSrvc *state.Service) (*system.Service, error)
@@ -382,7 +381,7 @@ func newNode(config *cfg.Config,
 		networkSrvc.SetSyncer(syncer)
 		networkSrvc.SetTransactionHandler(coreSrvc)
 	}
-	nodeSrvcs = append(nodeSrvcs, syncer)
+	nodeSrvcs = append(nodeSrvcs, syncer.(service))
 
 	bp, err := builder.createBABEService(config, stateSrvc, ks.Babe, coreSrvc, telemetryMailer)
 	if err != nil {
@@ -402,7 +401,7 @@ func newNode(config *cfg.Config,
 			blockProducer: bp,
 			system:        sysSrvc,
 			blockFinality: fg,
-			syncer:        syncer,
+			syncer:        syncer.(rpc.SyncAPI),
 		}
 		rpcSrvc, err = builder.createRPCService(cRPCParams)
 		if err != nil {

dot/node_integration_test.go

@@ -22,7 +22,7 @@ import (
 	digest "github.com/ChainSafe/gossamer/dot/digest"
 	network "github.com/ChainSafe/gossamer/dot/network"
 	"github.com/ChainSafe/gossamer/dot/state"
-	dotsync "github.com/ChainSafe/gossamer/dot/sync"
+	"github.com/ChainSafe/gossamer/dot/sync"
 	system "github.com/ChainSafe/gossamer/dot/system"
 	"github.com/ChainSafe/gossamer/dot/telemetry"
 	"github.com/ChainSafe/gossamer/dot/types"
@@ -139,7 +139,7 @@ func TestNewNode(t *testing.T) {
 	m.EXPECT().newSyncService(initConfig, gomock.AssignableToTypeOf(&state.Service{}), &grandpa.Service{},
 		&babe.VerificationManager{}, &core.Service{}, gomock.AssignableToTypeOf(&network.Service{}),
 		gomock.AssignableToTypeOf(&telemetry.Mailer{})).
-		Return(&dotsync.Service{}, nil)
+		Return(&sync.SyncService{}, nil)
 	m.EXPECT().createBABEService(initConfig, gomock.AssignableToTypeOf(&state.Service{}), ks.Babe,
 		&core.Service{}, gomock.AssignableToTypeOf(&telemetry.Mailer{})).
 		Return(&babe.Service{}, nil)

dot/peerset/constants.go

@@ -38,6 +38,11 @@ const (
 	// GoodTransactionReason is the reason for used for good transaction.
 	GoodTransactionReason = "Good Transaction"
 
+	// NotRelevantBlockAnnounce when peer sends us a not relevant block
+	NotRelevantBlockAnnounceValue Reputation = -(1 << 2)
+	// BadTransactionReason when transaction import was not performed.
+	NotRelevantBlockAnnounceReason = "Not Relevant Block Announce"
+
 	// BadTransactionValue used when transaction import was not performed.
 	BadTransactionValue Reputation = -(1 << 12)
 	// BadTransactionReason when transaction import was not performed.

dot/peerset/peerstate.go

@@ -240,7 +240,8 @@ func (ps *PeersState) addReputation(peerID peer.ID, change ReputationChange) (
 
 	node, has := ps.nodes[peerID]
 	if !has {
-		return 0, fmt.Errorf("%w: for peer id %s", ErrPeerDoesNotExist, peerID)
+		ps.insertPeer(0, peerID)
+		node = ps.nodes[peerID]
 	}
 
 	newReputation = node.addReputation(change.Value)