[GH-659] Basic protocol version change handling.

Author: Un1oR

nil/cmd/exporter/internal/clickhouse/clickhouse.go

@@ -6,6 +6,7 @@ import (
 	"database/sql"
 	"errors"
 	"fmt"
+
 	"github.com/ClickHouse/clickhouse-go/v2"
 	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
 	"github.com/NilFoundation/nil/nil/cmd/exporter/internal"

nil/internal/collate/bootstrap.go

@@ -10,6 +10,7 @@ import (
 	"github.com/NilFoundation/nil/nil/internal/db"
 	"github.com/NilFoundation/nil/nil/internal/network"
 	"github.com/NilFoundation/nil/nil/internal/types"
+	"github.com/libp2p/go-libp2p/core/peer"
 )
 
 const topicVersion = "/nil/version"
@@ -119,12 +120,7 @@ func fetchSnapshot(
 	return fetchShardSnap(ctx, nm, peerId, db, logger)
 }
 
-func fetchGenesisBlockHash(ctx context.Context, nm *network.Manager, peerAddr network.AddrInfo) (common.Hash, error) {
-	peerId, err := nm.Connect(ctx, peerAddr)
-	if err != nil {
-		return common.EmptyHash, fmt.Errorf("failed to connect to %s: %w", peerAddr, err)
-	}
-
+func fetchGenesisBlockHash(ctx context.Context, nm *network.Manager, peerId peer.ID) (common.Hash, error) {
 	resp, err := nm.SendRequestAndGetResponse(ctx, peerId, topicVersion, nil)
 	if err != nil {
 		return common.EmptyHash, fmt.Errorf("failed to fetch genesis block hash: %w", err)

nil/internal/collate/syncer.go

@@ -22,6 +22,15 @@ import (
 	"google.golang.org/protobuf/proto"
 )
 
+type ProtocolVersionMismatchError struct {
+	LocalVersion  string
+	RemoteVersion string
+}
+
+func (e *ProtocolVersionMismatchError) Error() string {
+	return fmt.Sprintf("protocol version mismatch; local: %s, remote: %s", e.LocalVersion, e.RemoteVersion)
+}
+
 type SyncerConfig struct {
 	execution.BlockGeneratorParams
 
@@ -79,30 +88,53 @@ func (s *Syncer) WaitComplete() {
 	s.waitForSync.Wait()
 }
 
-func (s *Syncer) readLocalVersion(ctx context.Context) (common.Hash, error) {
+func (s *Syncer) readLocalVersion(ctx context.Context) (*NodeVersion, error) {
+	protocolVersion := s.networkManager.ProtocolVersion()
+
 	rotx, err := s.db.CreateRoTx(ctx)
 	if err != nil {
-		return common.EmptyHash, err
+		return nil, err
 	}
 	defer rotx.Rollback()
 
 	res, err := db.ReadBlockHashByNumber(rotx, types.MainShardId, 0)
-	if errors.Is(err, db.ErrKeyNotFound) {
-		return common.EmptyHash, nil
+	if err != nil {
+		if errors.Is(err, db.ErrKeyNotFound) {
+			return &NodeVersion{protocolVersion, common.EmptyHash}, nil
+		} else {
+			return nil, err
+		}
 	}
-	return res, err
+	return &NodeVersion{protocolVersion, res}, err
+}
+
+type NodeVersion struct {
+	ProtocolVersion  string
+	GenesisBlockHash common.Hash
 }
 
-func (s *Syncer) fetchRemoteVersion(ctx context.Context) (common.Hash, error) {
+func (s *Syncer) fetchRemoteVersion(ctx context.Context) (NodeVersion, error) {
 	var err error
 	for _, peer := range s.config.BootstrapPeers {
+		var peerId network.PeerID
+		peerId, err = s.networkManager.Connect(ctx, peer)
+		if err != nil {
+			continue
+		}
+
+		var protocolVersion string
+		protocolVersion, err = s.networkManager.GetPeerProtocolVersion(peerId)
+		if err != nil {
+			continue
+		}
+
 		var res common.Hash
-		res, err = fetchGenesisBlockHash(ctx, s.networkManager, peer)
+		res, err = fetchGenesisBlockHash(ctx, s.networkManager, peerId)
 		if err == nil {
-			return res, nil
+			return NodeVersion{protocolVersion, res}, nil
 		}
 	}
-	return common.EmptyHash, fmt.Errorf("failed to fetch version from all peers; last error: %w", err)
+	return NodeVersion{"", common.EmptyHash}, fmt.Errorf("failed to fetch version from all peers; last error: %w", err)
 }
 
 func (s *Syncer) fetchSnapshot(ctx context.Context) error {
@@ -130,10 +162,6 @@ func (s *Syncer) Init(ctx context.Context, allowDbDrop bool) error {
 	if err != nil {
 		return err
 	}
-	if version.Empty() {
-		s.logger.Info().Msg("Local version is empty. Fetching snapshot...")
-		return s.fetchSnapshot(ctx)
-	}
 
 	remoteVersion, err := s.fetchRemoteVersion(ctx)
 	if err != nil {
@@ -142,7 +170,20 @@ func (s *Syncer) Init(ctx context.Context, allowDbDrop bool) error {
 			"Failed to fetch remote version. For now we assume that local version %s is up to date", version)
 		return nil
 	}
-	if version == remoteVersion {
+
+	if version.ProtocolVersion != remoteVersion.ProtocolVersion {
+		return &ProtocolVersionMismatchError{
+			version.ProtocolVersion,
+			remoteVersion.ProtocolVersion,
+		}
+	}
+
+	if version.GenesisBlockHash.Empty() {
+		s.logger.Info().Msg("Local version is empty. Fetching snapshot...")
+		return s.fetchSnapshot(ctx)
+	}
+
+	if version.GenesisBlockHash == remoteVersion.GenesisBlockHash {
 		s.logger.Info().Msgf("Local version %s is up to date. Finished initialization", version)
 		return nil
 	}
@@ -297,6 +338,10 @@ func (s *Syncer) fetchBlocksRange(ctx context.Context) <-chan *types.BlockWithEx
 	if err != nil {
 		return nil
 	}
+	if lastBlock == nil {
+		s.logger.Warn().Msg("No last block found")
+		return nil
+	}
 
 	for _, p := range peers {
 		s.logger.Trace().Msgf("Requesting blocks from %d from peer %s", lastBlock.Id+1, p)
@@ -335,7 +380,7 @@ func (s *Syncer) saveBlock(ctx context.Context, block *types.BlockWithExtractedD
 	return nil
 }
 
-func (s *Syncer) GenerateZerostate(ctx context.Context) error {
+func (s *Syncer) GenerateZerostateIfShardIsEmpty(ctx context.Context) error {
 	ctx, cancel := context.WithTimeout(ctx, s.config.Timeout)
 	defer cancel()
 

nil/internal/network/config.go

@@ -10,7 +10,8 @@ import (
 type PeerID = peer.ID
 
 type Config struct {
-	PrivateKey PrivateKey `yaml:"-"`
+	PrivateKey      PrivateKey `yaml:"-"`
+	ProtocolVersion string     `yaml:"-"`
 
 	KeysPath string `yaml:"keysPath,omitempty"`
 

nil/internal/network/host.go

@@ -53,6 +53,7 @@ func getCommonOptions(ctx context.Context, conf *Config) ([]libp2p.Option, loggi
 	}
 
 	return []libp2p.Option{
+		libp2p.ProtocolVersion(conf.ProtocolVersion),
 		libp2p.Security(noise.ID, noise.New),
 		libp2p.ConnectionManager(cm),
 		libp2p.Identity(conf.PrivateKey),

nil/internal/network/manager.go

@@ -2,6 +2,7 @@ package network
 
 import (
 	"context"
+	"fmt"
 	"slices"
 	"strings"
 
@@ -16,8 +17,9 @@ import (
 )
 
 type Manager struct {
-	ctx    context.Context
-	prefix string
+	ctx             context.Context
+	prefix          string
+	protocolVersion string
 
 	host   Host
 	pubSub *PubSub
@@ -73,13 +75,14 @@ func newManagerFromHost(
 	}
 
 	return &Manager{
-		ctx:    ctx,
-		prefix: conf.Prefix,
-		host:   h,
-		pubSub: ps,
-		dht:    dht,
-		meter:  telemetry.NewMeter("github.com/NilFoundation/nil/nil/internal/network"),
-		logger: logger,
+		ctx:             ctx,
+		prefix:          conf.Prefix,
+		protocolVersion: conf.ProtocolVersion,
+		host:            h,
+		pubSub:          ps,
+		dht:             dht,
+		meter:           telemetry.NewMeter("github.com/NilFoundation/nil/nil/internal/network"),
+		logger:          logger,
 	}, nil
 }
 
@@ -123,6 +126,22 @@ func (m *Manager) PubSub() *PubSub {
 	return m.pubSub
 }
 
+func (m *Manager) ProtocolVersion() string {
+	return m.protocolVersion
+}
+
+func (m *Manager) GetPeerProtocolVersion(peer peer.ID) (string, error) {
+	pv, err := m.host.Peerstore().Get(peer, "ProtocolVersion")
+	if err != nil {
+		return "", err
+	}
+	versionString, ok := pv.(string)
+	if !ok {
+		return "", fmt.Errorf("failed to convert protocol version to string for peer %s", peer)
+	}
+	return versionString, nil
+}
+
 func (m *Manager) AllKnownPeers() []peer.ID {
 	return slices.DeleteFunc(m.host.Peerstore().PeersWithAddrs(), func(i PeerID) bool {
 		return m.host.ID() == i

nil/internal/network/testaide.go

@@ -93,6 +93,36 @@ func topLevelTestName(t *testing.T) string {
 	return strings.Split(t.Name(), "/")[0]
 }
 
+// ConfigOverrides is a helper structure for configuring the config in tests.
+//
+// Why was it needed at all?
+// To understand this, let's consider how the config is filled when the application is running.
+// We have several layers of overrides:
+// - Default values set in the code
+// - They are merged with the values from the yaml file (overwriting in favor of the file)
+// - And they can be overridden by command line options (again with overwriting)
+//
+// What do the yaml file and command line options have in common?
+// Both of these sources have the semantics of absence of value.
+// That is, we can clearly distinguish, for example, false from the absence of a value,
+// and if a true was set at an earlier layer and there was no override at a later layer,
+// we should not turn true into false.
+//
+// It is important that the semantics of the absence of a value are not represented at the level
+// of the config structure itself in Go, so the task of merging two configs presented in this way becomes non-trivial.
+// We can easily accidentally overwrite the value from the base config with the default value from the override config.
+//
+// This is exactly why a separate structure was introduced, having an explicit semantics of absence of value.
+type ConfigOverrides struct {
+	ProtocolVersion *string
+}
+
+func (c *ConfigOverrides) ApplyToConfig(cfg *Config) {
+	if c.ProtocolVersion != nil {
+		cfg.ProtocolVersion = *c.ProtocolVersion
+	}
+}
+
 func GenerateConfig(t *testing.T, port int) (*Config, AddrInfo) {
 	t.Helper()
 

nil/services/nilservice/service.go

@@ -222,7 +222,7 @@ func initSyncers(ctx context.Context, syncers []*collate.Syncer, allowDbDrop boo
 		return err
 	}
 	for _, syncer := range syncers {
-		if err := syncer.GenerateZerostate(ctx); err != nil {
+		if err := syncer.GenerateZerostateIfShardIsEmpty(ctx); err != nil {
 			return err
 		}
 	}
@@ -315,7 +315,16 @@ type Node struct {
 
 func (i *Node) Run() error {
 	if err := concurrent.Run(i.ctx, i.funcs...); err != nil {
-		i.logger.Error().Err(err).Msg("App encountered an error and will be terminated.")
+		var executionErr *concurrent.ExecutionError
+		var protocolVersionMismatchErr *collate.ProtocolVersionMismatchError
+		if errors.As(err, &executionErr) && errors.As(executionErr.Err, &protocolVersionMismatchErr) {
+			i.logger.Error().
+				Str("localVersion", protocolVersionMismatchErr.LocalVersion).
+				Str("remoteVersion", protocolVersionMismatchErr.RemoteVersion).
+				Msg("Protocol version mismatch. Probably nild executable is outdated.")
+		} else {
+			i.logger.Error().Err(err).Msg("App encountered an error and will be terminated.")
+		}
 		return err
 	}
 	i.logger.Info().Msg("App is terminated.")