network: integrate state sync module with blockfetcher

Close #3574

Signed-off-by: Ekaterina Pavlova <[email protected]>

fix

Signed-off-by: Ekaterina Pavlova <[email protected]>

Author: AliceInHunterland

Diff for: internal/fakechain/fakechain.go

@@ -452,6 +452,8 @@ func (s *FakeStateSync) Init(currChainHeight uint32) error {
 // NeedHeaders implements the StateSync interface.
 func (s *FakeStateSync) NeedHeaders() bool { return s.RequestHeaders.Load() }
 
+func (s *FakeStateSync) NeedBlocks() bool { return false }
+
 // NeedMPTNodes implements the StateSync interface.
 func (s *FakeStateSync) NeedMPTNodes() bool {
 	panic("TODO")
@@ -469,3 +471,13 @@ func (s *FakeStateSync) Traverse(root util.Uint256, process func(node mpt.Node,
 func (s *FakeStateSync) GetUnknownMPTNodesBatch(limit int) []util.Uint256 {
 	panic("TODO")
 }
+
+// GetConfig implements the StateSync interface.
+func (s *FakeStateSync) GetConfig() config.Blockchain {
+	panic("TODO")
+}
+
+// SetOnStageChanged implements the StateSync interface.
+func (s *FakeStateSync) SetOnStageChanged(func()) {
+	panic("TODO")
+}

Diff for: pkg/config/blockfetcher_config.go

@@ -21,6 +21,7 @@ type NeoFSBlockFetcher struct {
 	BQueueSize             int           `yaml:"BQueueSize"`
 	SkipIndexFilesSearch   bool          `yaml:"SkipIndexFilesSearch"`
 	IndexFileSize          uint32        `yaml:"IndexFileSize"`
+	BlocksOnly             bool          `yaml:"BlocksOnly"`
 }
 
 // Validate checks NeoFSBlockFetcher for internal consistency and ensures

Diff for: pkg/config/config.go

@@ -64,6 +64,7 @@ func (c Config) Blockchain() Blockchain {
 	return Blockchain{
 		ProtocolConfiguration: c.ProtocolConfiguration,
 		Ledger:                c.ApplicationConfiguration.Ledger,
+		NeoFSBlockFetcher:     c.ApplicationConfiguration.NeoFSBlockFetcher,
 	}
 }
 

Diff for: pkg/config/ledger_config.go

@@ -29,4 +29,5 @@ type Ledger struct {
 type Blockchain struct {
 	ProtocolConfiguration
 	Ledger
+	NeoFSBlockFetcher
 }

Diff for: pkg/core/blockchain.go

@@ -285,9 +285,9 @@ func NewBlockchain(s storage.Store, cfg config.Blockchain, log *zap.Logger) (*Bl
 			zap.Uint32("MaxValidUntilBlockIncrement", cfg.MaxValidUntilBlockIncrement))
 	}
 	if cfg.P2PStateExchangeExtensions {
-		if !cfg.StateRootInHeader {
-			return nil, errors.New("P2PStatesExchangeExtensions are enabled, but StateRootInHeader is off")
-		}
+		//if !cfg.StateRootInHeader && !cfg.NeoFSBlockFetcher.Enabled {
+		//	return nil, errors.New("P2PStatesExchangeExtensions are enabled, but StateRootInHeader is off")
+		//}
 		if cfg.KeepOnlyLatestState && !cfg.RemoveUntraceableBlocks {
 			return nil, errors.New("P2PStateExchangeExtensions can be enabled either on MPT-complete node (KeepOnlyLatestState=false) or on light GC-enabled node (RemoveUntraceableBlocks=true)")
 		}

Diff for: pkg/core/statesync/module.go

@@ -37,6 +37,9 @@ import (
 // stateSyncStage is a type of state synchronisation stage.
 type stateSyncStage uint8
 
+// OnStageChanged is a callback function that is called when state sync stage is changed.
+type OnStageChanged func()
+
 const (
 	// inactive means that state exchange is disabled by the protocol configuration.
 	// Can't be combined with other states.
@@ -93,6 +96,10 @@ type Module struct {
 	billet *mpt.Billet
 
 	jumpCallback func(p uint32) error
+
+	// stageCB is an optional callback that is triggered whenever
+	// the sync stage changes.
+	stageCB func()
 }
 
 // NewModule returns new instance of statesync module.
@@ -120,6 +127,7 @@ func NewModule(bc Ledger, stateMod *stateroot.Module, log *zap.Logger, s *dao.Si
 // Init initializes state sync module for the current chain's height with given
 // callback for MPT nodes requests.
 func (s *Module) Init(currChainHeight uint32) error {
+	oldStage := s.syncStage
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
@@ -131,6 +139,9 @@ func (s *Module) Init(currChainHeight uint32) error {
 	if p < 2*s.syncInterval {
 		// chain is too low to start state exchange process, use the standard sync mechanism
 		s.syncStage = inactive
+		if s.syncStage != oldStage {
+			s.notifyStageChanged()
+		}
 		return nil
 	}
 	pOld, err := s.dao.GetStateSyncPoint()
@@ -142,6 +153,9 @@ func (s *Module) Init(currChainHeight uint32) error {
 			// chain has already been synchronised up to old state sync point and regular blocks processing was started.
 			// Current block height is enough to start regular blocks processing.
 			s.syncStage = inactive
+			if s.syncStage != oldStage {
+				s.notifyStageChanged()
+			}
 			return nil
 		}
 		if err == nil {
@@ -172,10 +186,26 @@ func (s *Module) Init(currChainHeight uint32) error {
 	s.log.Info("try to sync state for the latest state synchronisation point",
 		zap.Uint32("point", p),
 		zap.Uint32("evaluated chain's blockHeight", currChainHeight))
-
+	if s.syncStage != oldStage {
+		s.notifyStageChanged()
+	}
 	return s.defineSyncStage()
 }
 
+// SetOnStageChanged sets callback that is triggered whenever the sync stage changes.
+func (s *Module) SetOnStageChanged(cb func()) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	s.stageCB = cb
+}
+
+// notifyStageChanged triggers stage callback if it's set.
+func (s *Module) notifyStageChanged() {
+	if s.stageCB != nil {
+		go s.stageCB()
+	}
+}
+
 // TemporaryPrefix accepts current storage prefix and returns prefix
 // to use for storing intermediate items during synchronization.
 func TemporaryPrefix(currPrefix storage.KeyPrefix) storage.KeyPrefix {
@@ -192,12 +222,17 @@ func TemporaryPrefix(currPrefix storage.KeyPrefix) storage.KeyPrefix {
 // defineSyncStage sequentially checks and sets sync state process stage after Module
 // initialization. It also performs initialization of MPT Billet if necessary.
 func (s *Module) defineSyncStage() error {
+	oldStage := s.syncStage
 	// check headers sync stage first
 	ltstHeaderHeight := s.bc.HeaderHeight()
 	if ltstHeaderHeight > s.syncPoint {
 		s.syncStage = headersSynced
 		s.log.Info("headers are in sync",
 			zap.Uint32("headerHeight", s.bc.HeaderHeight()))
+		if s.syncStage != oldStage {
+			s.notifyStageChanged()
+			oldStage = s.syncStage
+		}
 	}
 
 	// check blocks sync stage
@@ -206,13 +241,21 @@ func (s *Module) defineSyncStage() error {
 		s.syncStage |= blocksSynced
 		s.log.Info("blocks are in sync",
 			zap.Uint32("blockHeight", s.blockHeight))
+		if s.syncStage != oldStage {
+			s.notifyStageChanged()
+			oldStage = s.syncStage
+		}
 	}
 
 	// check MPT sync stage
 	if s.blockHeight > s.syncPoint {
 		s.syncStage |= mptSynced
 		s.log.Info("MPT is in sync",
 			zap.Uint32("stateroot height", s.stateMod.CurrentLocalHeight()))
+		if s.syncStage != oldStage {
+			s.notifyStageChanged()
+			oldStage = s.syncStage
+		}
 	} else if s.syncStage&headersSynced != 0 {
 		header, err := s.bc.GetHeader(s.bc.GetHeaderHash(s.syncPoint + 1))
 		if err != nil {
@@ -255,12 +298,19 @@ func (s *Module) defineSyncStage() error {
 			s.syncStage |= mptSynced
 			s.log.Info("MPT is in sync",
 				zap.Uint32("stateroot height", s.syncPoint))
+			if s.syncStage != oldStage {
+				s.notifyStageChanged()
+				oldStage = s.syncStage
+			}
 		}
 	}
 
 	if s.syncStage == headersSynced|blocksSynced|mptSynced {
 		s.log.Info("state is in sync, starting regular blocks processing")
 		s.syncStage = inactive
+		if s.syncStage != oldStage {
+			s.notifyStageChanged()
+		}
 	}
 	return nil
 }
@@ -306,6 +356,7 @@ func (s *Module) AddHeaders(hdrs ...*block.Header) error {
 
 // AddBlock verifies and saves block skipping executable scripts.
 func (s *Module) AddBlock(block *block.Block) error {
+	oldStage := s.syncStage
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
@@ -351,6 +402,9 @@ func (s *Module) AddBlock(block *block.Block) error {
 		s.syncStage |= blocksSynced
 		s.log.Info("blocks are in sync",
 			zap.Uint32("blockHeight", s.blockHeight))
+		if s.syncStage != oldStage {
+			s.notifyStageChanged()
+		}
 		s.checkSyncIsCompleted()
 	}
 	return nil
@@ -425,6 +479,7 @@ func (s *Module) restoreNode(n mpt.Node) error {
 // If so, then jumping to P state sync point occurs. It is not protected by lock, thus caller
 // should take care of it.
 func (s *Module) checkSyncIsCompleted() {
+	oldStage := s.syncStage
 	if s.syncStage != headersSynced|mptSynced|blocksSynced {
 		return
 	}
@@ -436,6 +491,10 @@ func (s *Module) checkSyncIsCompleted() {
 	}
 	s.syncStage = inactive
 	s.dispose()
+
+	if s.syncStage != oldStage {
+		s.notifyStageChanged()
+	}
 }
 
 func (s *Module) dispose() {
@@ -492,6 +551,14 @@ func (s *Module) NeedMPTNodes() bool {
 	return s.syncStage&headersSynced != 0 && s.syncStage&mptSynced == 0
 }
 
+// NeedBlocks returns whether the module hasn't completed blocks synchronisation.
+func (s *Module) NeedBlocks() bool {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
+	return s.syncStage&headersSynced != 0 && s.syncStage&blocksSynced == 0
+}
+
 // Traverse traverses local MPT nodes starting from the specified root down to its
 // children calling `process` for each serialised node until stop condition is satisfied.
 func (s *Module) Traverse(root util.Uint256, process func(node mpt.Node, nodeBytes []byte) bool) error {
@@ -516,3 +583,8 @@ func (s *Module) GetUnknownMPTNodesBatch(limit int) []util.Uint256 {
 
 	return s.mptpool.GetBatch(limit)
 }
+
+// GetConfig returns current blockchain configuration.
+func (s *Module) GetConfig() config.Blockchain {
+	return s.bc.GetConfig()
+}

Diff for: pkg/network/bqueue/queue.go

@@ -5,6 +5,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/nspcc-dev/neo-go/pkg/config"
 	"github.com/nspcc-dev/neo-go/pkg/core/block"
 	"go.uber.org/zap"
 )
@@ -15,6 +16,7 @@ type Blockqueuer interface {
 	AddHeaders(...*block.Header) error
 	BlockHeight() uint32
 	HeaderHeight() uint32
+	GetConfig() config.Blockchain
 }
 
 // OperationMode is the mode of operation for the block queue.