feat(dot/sync): Implement state sync strategy

chain/kusama/defaults.go

@@ -25,6 +25,7 @@ func DefaultConfig() *cfg.Config {
 	config.Core.GrandpaAuthority = false
 	config.Core.Role = 1
 	config.Network.NoMDNS = false
+	config.Core.SyncMode = cfg.FullSync
 
 	return config
 }

chain/paseo/defaults.go

@@ -24,6 +24,7 @@ func DefaultConfig() *cfg.Config {
 	config.Core.GrandpaAuthority = false
 	config.Core.Role = 1
 	config.Network.NoMDNS = false
+	config.Core.SyncMode = cfg.FullSync
 
 	return config
 }

chain/polkadot/defaults.go

@@ -24,6 +24,7 @@ func DefaultConfig() *cfg.Config {
 	config.Core.GrandpaAuthority = false
 	config.Core.Role = 1
 	config.Network.NoMDNS = false
+	config.Core.SyncMode = cfg.FullSync
 
 	return config
 }

chain/westend-dev/defaults.go

@@ -24,6 +24,7 @@ func DefaultConfig() *cfg.Config {
 	config.RPC.UnsafeRPC = true
 	config.RPC.WSExternal = true
 	config.RPC.UnsafeWSExternal = true
+	config.Core.SyncMode = cfg.FullSync
 
 	return config
 }

chain/westend-local/default.go

@@ -29,6 +29,7 @@ func DefaultConfig() *cfg.Config {
 	config.RPC.UnsafeRPC = true
 	config.RPC.WSExternal = true
 	config.RPC.UnsafeWSExternal = true
+	config.Core.SyncMode = cfg.FullSync
 
 	return config
 }

chain/westend/defaults.go

@@ -24,6 +24,7 @@ func DefaultConfig() *cfg.Config {
 	config.Core.GrandpaAuthority = false
 	config.Core.Role = 1
 	config.Network.NoMDNS = false
+	config.Core.SyncMode = cfg.FullSync
 
 	return config
 }

cmd/gossamer/commands/root.go

@@ -39,6 +39,8 @@ var (
 	role string
 	// validator when set, the node will be an authority
 	validator bool
+	// Sync mode [warp | full]
+	syncMode string
 
 	// Account Config
 	// key to use for the node
@@ -102,6 +104,10 @@ Usage:
 				return fmt.Errorf("failed to parse role: %s", err)
 			}
 
+			if err := parseSyncMode(); err != nil {
+				return fmt.Errorf("failed to parse sync mode: %s", err)
+			}
+
 			if err := parseTelemetryURL(); err != nil {
 				return fmt.Errorf("failed to parse telemetry-url: %s", err.Error())
 			}
@@ -529,6 +535,11 @@ func addCoreFlags(cmd *cobra.Command) error {
 		return fmt.Errorf("failed to add --grandpa-interval flag: %s", err)
 	}
 
+	cmd.Flags().StringVar(&syncMode,
+		"sync",
+		cfg.FullSync.String(),
+		"Sync mode. One of 'full' or 'warp'.")
+
 	return nil
 }
 

cmd/gossamer/commands/utils.go

@@ -407,6 +407,23 @@ func parseRole() error {
 	return nil
 }
 
+// parseSyncMode parses the sync mode from the command line flags
+func parseSyncMode() error {
+	var selectedSyncMode cfg.SyncMode
+	switch syncMode {
+	case cfg.FullSync.String():
+		selectedSyncMode = cfg.FullSync
+	case cfg.WarpSync.String():
+		selectedSyncMode = cfg.WarpSync
+	default:
+		return fmt.Errorf("invalid sync mode: %s", role)
+	}
+
+	config.Core.SyncMode = selectedSyncMode
+	viper.Set("core.syncMode", config.Core.SyncMode)
+	return nil
+}
+
 // parseTelemetryURL parses the telemetry-url from the command line flag
 func parseTelemetryURL() error {
 	if telemetryURLs == "" {

config/config.go

@@ -61,6 +61,9 @@ const (
 	DefaultSystemName = "Gossamer"
 	// DefaultSystemVersion is the default system version
 	DefaultSystemVersion = "0.0.0"
+
+	// DefaultSyncMode is the default block sync mode
+	DefaultSyncMode = FullSync
 )
 
 // DefaultRPCModules the default RPC modules
@@ -188,6 +191,7 @@ type CoreConfig struct {
 	GrandpaAuthority bool               `mapstructure:"grandpa-authority"`
 	WasmInterpreter  string             `mapstructure:"wasm-interpreter,omitempty"`
 	GrandpaInterval  time.Duration      `mapstructure:"grandpa-interval,omitempty"`
+	SyncMode         SyncMode           `mapstructure:"sync,omitempty"`
 }
 
 // StateConfig contains the configuration for the state.
@@ -363,6 +367,7 @@ func DefaultConfig() *Config {
 			GrandpaAuthority: true,
 			WasmInterpreter:  DefaultWasmInterpreter,
 			GrandpaInterval:  DefaultDiscoveryInterval,
+			SyncMode:         DefaultSyncMode,
 		},
 		Network: &NetworkConfig{
 			Port:              DefaultNetworkPort,
@@ -444,6 +449,7 @@ func DefaultConfigFromSpec(nodeSpec *genesis.Genesis) *Config {
 			GrandpaAuthority: true,
 			WasmInterpreter:  DefaultWasmInterpreter,
 			GrandpaInterval:  DefaultDiscoveryInterval,
+			SyncMode:         DefaultSyncMode,
 		},
 		Network: &NetworkConfig{
 			Port:              DefaultNetworkPort,
@@ -525,6 +531,7 @@ func Copy(c *Config) Config {
 			GrandpaAuthority: c.Core.GrandpaAuthority,
 			WasmInterpreter:  c.Core.WasmInterpreter,
 			GrandpaInterval:  c.Core.GrandpaInterval,
+			SyncMode:         c.Core.SyncMode,
 		},
 		Network: &NetworkConfig{
 			Port:              c.Network.Port,
@@ -604,6 +611,19 @@ func (c Chain) String() string {
 	return string(c)
 }
 
+// SyncMode is a string representing a sync mode
+type SyncMode string
+
+const (
+	FullSync  SyncMode = "full"
+	WarpSync  SyncMode = "warp"
+	StateSync SyncMode = "state"
+)
+
+func (n SyncMode) String() string {
+	return string(n)
+}
+
 // NetworkRole is a string representing a network role
 type NetworkRole string
 

dot/network/discovery.go

@@ -171,7 +171,7 @@ func (d *discovery) advertise() {
 
 			ttl, err = d.rd.Advertise(d.ctx, string(d.pid))
 			if err != nil {
-				logger.Warnf("failed to advertise in the DHT: %s", err)
+				logger.Debugf("failed to advertise in the DHT: %s", err)
 				ttl = tryAdvertiseTimeout
 			}
 		}

dot/network/host_integration_test.go

@@ -7,6 +7,7 @@ package network
 
 import (
 	"fmt"
+	"strings"
 	"testing"
 	"time"
 
@@ -396,12 +397,21 @@ func Test_PeerSupportsProtocol(t *testing.T) {
 	}
 	require.NoError(t, err)
 
+	genesisHash := nodeA.blockState.GenesisHash().String()
+	genesisHash = strings.TrimPrefix(genesisHash, "0x")
+	fullSyncProtocolId := fmt.Sprintf("/%s%s", genesisHash, SyncID)
+	warpSyncProtocolId := fmt.Sprintf("/%s%s", genesisHash, WarpSyncID)
+
 	tests := []struct {
 		protocol protocol.ID
 		expect   bool
 	}{
 		{
-			protocol: protocol.ID("/gossamer/test/0/sync/2"),
+			protocol: protocol.ID(fullSyncProtocolId),
+			expect:   true,
+		},
+		{
+			protocol: protocol.ID(warpSyncProtocolId),
 			expect:   true,
 		},
 		{

dot/network/messages/state.go

@@ -22,10 +22,18 @@ type StateRequest struct {
 	NoProof bool
 }
 
+func NewStateRequest(block common.Hash, start [][]byte, noProof bool) *StateRequest {
+	return &StateRequest{
+		Block:   block,
+		Start:   start,
+		NoProof: noProof,
+	}
+}
+
 func (s *StateRequest) String() string {
-	return fmt.Sprintf("StateRequest Block=%s Start=[0x%x, 0x%x] NoProof=%v",
+	return fmt.Sprintf("StateRequest Block=%s Start=[%v] NoProof=%v",
 		s.Block.String(),
-		s.Start[0], s.Start[1],
+		s.Start,
 		s.NoProof,
 	)
 }
@@ -98,3 +106,14 @@ func (s *StateResponse) Decode(in []byte) error {
 
 	return nil
 }
+
+func (s *StateResponse) Encode() ([]byte, error) {
+	panic("not implemented")
+}
+
+func (s *StateResponse) String() string {
+	return fmt.Sprintf("StateResponse Entries=[%v] Proof=[%v]",
+		s.Entries,
+		s.Proof,
+	)
+}

dot/network/messages/warp_sync.go

@@ -15,6 +15,12 @@ type WarpProofRequest struct {
 	Begin common.Hash
 }
 
+func NewWarpProofRequest(from common.Hash) *WarpProofRequest {
+	return &WarpProofRequest{
+		Begin: from,
+	}
+}
+
 // Decode decodes the message into a WarpProofRequest
 func (wpr *WarpProofRequest) Decode(in []byte) error {
 	return scale.Unmarshal(in, wpr)

dot/network/notifications.go

@@ -282,7 +282,7 @@ func (s *Service) sendData(peer peer.ID, hs Handshake, info *notificationsProtoc
 	// we've completed the handshake with the peer, send message directly
 	logger.Tracef("sending message to peer %s using protocol %s: %s", peer, info.protocolID, msg)
 	if err := s.host.writeToStream(stream, msg); err != nil {
-		logger.Errorf("failed to send message to peer %s: %s", peer, err)
+		logger.Debugf("failed to send message to peer %s: %s", peer, err)
 
 		// the stream was closed or reset, close it on our end and delete it from our peer's data
 		if errors.Is(err, io.EOF) || errors.Is(err, network.ErrReset) {

dot/network/request_response.go

@@ -30,6 +30,19 @@ type RequestResponseProtocol struct {
 	responseBuf     []byte
 }
 
+func NewRequestResponseProtocol(ctx context.Context, host *host, protocolID protocol.ID,
+	requestTimeout time.Duration, maxResponseSize uint64) *RequestResponseProtocol {
+	return &RequestResponseProtocol{
+		ctx:             ctx,
+		host:            host,
+		requestTimeout:  requestTimeout,
+		maxResponseSize: maxResponseSize,
+		protocolID:      protocolID,
+		responseBuf:     make([]byte, maxResponseSize),
+		responseBufMu:   sync.Mutex{},
+	}
+}
+
 func (rrp *RequestResponseProtocol) Do(to peer.ID, req, res messages.P2PMessage) error {
 	rrp.host.p2pHost.ConnManager().Protect(to, "")
 	defer rrp.host.p2pHost.ConnManager().Unprotect(to, "")

dot/network/service.go

@@ -34,6 +34,7 @@ const (
 	// the following are sub-protocols used by the node
 	SyncID          = "/sync/2"
 	WarpSyncID      = "/sync/warp"
+	StateSyncID     = "/state/2"
 	lightID         = "/light/2"
 	blockAnnounceID = "/block-announces/1"
 	transactionsID  = "/transactions/1"
@@ -259,11 +260,14 @@ func (s *Service) Start() error {
 		s.ctx, s.cancel = context.WithCancel(context.Background())
 	}
 
-	genesisHashProtocolId := protocol.ID(s.cfg.BlockState.GenesisHash().String())
+	genesisHash := s.blockState.GenesisHash().String()
+	genesisHash = strings.TrimPrefix(genesisHash, "0x")
+	fullSyncProtocolId := fmt.Sprintf("/%s%s", genesisHash, SyncID)
+	warpSyncProtocolId := fmt.Sprintf("/%s%s", genesisHash, WarpSyncID)
 
-	s.host.registerStreamHandler(s.host.protocolID+SyncID, s.handleSyncStream)
+	s.host.registerStreamHandler(protocol.ID(fullSyncProtocolId), s.handleSyncStream)
 	s.host.registerStreamHandler(s.host.protocolID+lightID, s.handleLightStream)
-	s.host.registerStreamHandler(genesisHashProtocolId+WarpSyncID, s.handleWarpSyncStream)
+	s.host.registerStreamHandler(protocol.ID(warpSyncProtocolId), s.handleWarpSyncStream)
 
 	// register block announce protocol
 	err := s.RegisterNotificationsProtocol(
@@ -622,16 +626,11 @@ func (s *Service) SendMessage(to peer.ID, msg NotificationsMessage) error {
 func (s *Service) GetRequestResponseProtocol(subprotocol string, requestTimeout time.Duration,
 	maxResponseSize uint64) *RequestResponseProtocol {
 
-	protocolID := s.host.protocolID + protocol.ID(subprotocol)
-	return &RequestResponseProtocol{
-		ctx:             s.ctx,
-		host:            s.host,
-		requestTimeout:  requestTimeout,
-		maxResponseSize: maxResponseSize,
-		protocolID:      protocolID,
-		responseBuf:     make([]byte, maxResponseSize),
-		responseBufMu:   sync.Mutex{},
-	}
+	genesisHash := s.blockState.GenesisHash().String()
+	genesisHash = strings.TrimPrefix(genesisHash, "0x")
+	protocolId := fmt.Sprintf("/%s%s", genesisHash, subprotocol)
+
+	return NewRequestResponseProtocol(s.ctx, s.host, protocol.ID(protocolId), requestTimeout, maxResponseSize)
 }
 
 // Health returns information about host needed for the rpc server
@@ -759,7 +758,7 @@ func (s *Service) processMessage(msg peerset.Message) {
 		err := s.host.connect(addrInfo)
 		if err != nil {
 			// TODO: if error happens here outgoing (?) slot is occupied but no peer is really connected
-			logger.Warnf("failed to open connection for peer %s: %s", peerID, err)
+			logger.Debugf("failed to open connection for peer %s: %s", peerID, err)
 			return
 		}
 		logger.Debugf("connection successful with peer %s", peerID)

dot/network/warp_sync.go

@@ -8,23 +8,16 @@ import (
 	"fmt"
 
 	"github.com/ChainSafe/gossamer/dot/network/messages"
-	"github.com/ChainSafe/gossamer/dot/types"
 	"github.com/ChainSafe/gossamer/internal/primitives/consensus/grandpa"
 	primitives "github.com/ChainSafe/gossamer/internal/primitives/consensus/grandpa"
 	"github.com/ChainSafe/gossamer/lib/common"
+	"github.com/ChainSafe/gossamer/lib/grandpa/warpsync"
 	libp2pnetwork "github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
 )
 
 const MaxAllowedSameRequestPerPeer = 5
 
-type WarpSyncVerificationResult struct {
-	SetId         grandpa.SetID
-	AuthorityList primitives.AuthorityList
-	Header        types.Header
-	Completed     bool
-}
-
 // WarpSyncProvider is an interface for generating warp sync proofs
 type WarpSyncProvider interface {
 	// Generate proof starting at given block hash. The proof is accumulated until maximum proof
@@ -34,7 +27,7 @@ type WarpSyncProvider interface {
 		encodedProof []byte,
 		setId grandpa.SetID,
 		authorities primitives.AuthorityList,
-	) (*WarpSyncVerificationResult, error)
+	) (*warpsync.WarpSyncVerificationResult, error)
 }
 
 func (s *Service) handleWarpSyncRequest(req messages.WarpProofRequest) ([]byte, error) {