Merge pull request #3789 from nspcc-dev/headers

Headers fetching via NeoFS BlockFetcher service

Author: roman-khimov

internal/fakechain/fakechain.go

@@ -49,6 +49,11 @@ type FakeStateSync struct {
 	AddMPTNodesFunc   func(nodes [][]byte) error
 }
 
+// HeaderHeight returns the height of the latest stored header.
+func (s *FakeStateSync) HeaderHeight() uint32 {
+	return 0
+}
+
 // NewFakeChain returns a new FakeChain structure.
 func NewFakeChain() *FakeChain {
 	return NewFakeChainWithCustomCfg(nil)
@@ -447,6 +452,9 @@ func (s *FakeStateSync) Init(currChainHeight uint32) error {
 // NeedHeaders implements the StateSync interface.
 func (s *FakeStateSync) NeedHeaders() bool { return s.RequestHeaders.Load() }
 
+// NeedBlocks implements the StateSync interface.
+func (s *FakeStateSync) NeedBlocks() bool { return false }
+
 // NeedMPTNodes implements the StateSync interface.
 func (s *FakeStateSync) NeedMPTNodes() bool {
 	panic("TODO")
@@ -464,3 +472,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")
+}

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,
 	}
 }
 

pkg/config/ledger_config.go

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

pkg/config/protocol_config.go

@@ -49,6 +49,8 @@ type (
 		P2PSigExtensions bool `yaml:"P2PSigExtensions"`
 		// P2PStateExchangeExtensions enables additional P2P MPT state data exchange logic.
 		P2PStateExchangeExtensions bool `yaml:"P2PStateExchangeExtensions"`
+		// NeoFSStateSyncExtensions enables state data exchange logic via NeoFS.
+		NeoFSStateSyncExtensions bool `yaml:"NeoFSStateSyncExtensions"`
 		// ReservedAttributes allows to have reserved attributes range for experimental or private purposes.
 		ReservedAttributes bool `yaml:"ReservedAttributes"`
 

pkg/consensus/consensus.go

@@ -58,7 +58,7 @@ type Ledger interface {
 
 // BlockQueuer is an interface to the block queue manager sufficient for Service.
 type BlockQueuer interface {
-	PutBlock(block *coreb.Block) error
+	Put(queueable *coreb.Block) error
 }
 
 // Service represents a consensus instance.
@@ -623,7 +623,7 @@ func (s *service) processBlock(b dbft.Block[util.Uint256]) error {
 	bb := &b.(*neoBlock).Block
 	bb.Script = *(s.getBlockWitness(bb))
 
-	if err := s.BlockQueue.PutBlock(bb); err != nil {
+	if err := s.BlockQueue.Put(bb); err != nil {
 		// The block might already be added via the regular network
 		// interaction.
 		if _, errget := s.Chain.GetBlock(bb.Hash()); errget != nil {

pkg/consensus/consensus_test.go

@@ -528,7 +528,7 @@ type testBlockQueuer struct {
 var _ = BlockQueuer(testBlockQueuer{})
 
 // PutBlock implements BlockQueuer interface.
-func (bq testBlockQueuer) PutBlock(b *coreb.Block) error {
+func (bq testBlockQueuer) Put(b *coreb.Block) error {
 	return bq.bc.AddBlock(b)
 }
 

pkg/core/block/block.go

@@ -51,6 +51,13 @@ type auxBlockIn struct {
 	Transactions []json.RawMessage `json:"tx"`
 }
 
+// GetIndex returns the index of the block. This method should be used
+// for interfaces only. As generics don't support structural types
+// ref. golang/go#51259.
+func (b *Block) GetIndex() uint32 {
+	return b.Index
+}
+
 // ComputeMerkleRoot computes Merkle tree root hash based on actual block's data.
 func (b *Block) ComputeMerkleRoot() util.Uint256 {
 	hashes := make([]util.Uint256, len(b.Transactions))

pkg/core/block/header.go

@@ -10,6 +10,7 @@ import (
 	"github.com/nspcc-dev/neo-go/pkg/crypto/hash"
 	"github.com/nspcc-dev/neo-go/pkg/encoding/address"
 	"github.com/nspcc-dev/neo-go/pkg/io"
+	"github.com/nspcc-dev/neo-go/pkg/smartcontract"
 	"github.com/nspcc-dev/neo-go/pkg/util"
 )
 
@@ -80,6 +81,13 @@ type baseAux struct {
 	Witnesses     []transaction.Witness `json:"witnesses"`
 }
 
+// GetIndex returns the index of the block. This method should be used
+// for interfaces only. As generics don't support structural types
+// ref. golang/go#51259.
+func (b *Header) GetIndex() uint32 {
+	return b.Index
+}
+
 // Hash returns the hash of the block. Notice that it is cached internally,
 // so no matter how you change the [Header] after the first invocation of this
 // method it won't change. To get an updated hash in case you're changing
@@ -228,3 +236,23 @@ func (b *Header) UnmarshalJSON(data []byte) error {
 	}
 	return nil
 }
+
+// GetExpectedHeaderSize returns the expected Header size with the given number of validators.
+func GetExpectedHeaderSize(stateRootInHeader bool, numOfValidators int) int {
+	m := smartcontract.GetDefaultHonestNodeCount(numOfValidators)
+	// expectedHeaderSizeWithEmptyWitness contains 2 bytes for zero-length (new(Header)).Script.Invocation/Verification
+	// InvocationScript:
+	// 64 is the size of the default signature length + 2 bytes length and opcode
+	// 2 = 1 push opcode + 1 length
+	// VerifcationScript:
+	// m = 1 bytes
+	// 33 =  1 push opcode + 1 length + 33 bytes for public key
+	// n = 1 bytes
+	// 5 for SYSCALL
+	size := expectedHeaderSizeWithEmptyWitness + (1+1+64)*m + 2 + numOfValidators*(1+1+33) + 2 + 5
+
+	if stateRootInHeader {
+		size += util.Uint256Size
+	}
+	return size
+}

pkg/core/blockchain.go

@@ -299,6 +299,9 @@ func NewBlockchain(s storage.Store, cfg config.Blockchain, log *zap.Logger) (*Bl
 		log.Info("MaxValidUntilBlockIncrement is not set or wrong, using default value",
 			zap.Uint32("MaxValidUntilBlockIncrement", cfg.MaxValidUntilBlockIncrement))
 	}
+	if cfg.P2PStateExchangeExtensions && cfg.NeoFSStateSyncExtensions {
+		return nil, errors.New("P2PStateExchangeExtensions and NeoFSStateSyncExtensions cannot be enabled simultaneously")
+	}
 	if cfg.P2PStateExchangeExtensions {
 		if !cfg.StateRootInHeader {
 			return nil, errors.New("P2PStatesExchangeExtensions are enabled, but StateRootInHeader is off")
@@ -312,6 +315,16 @@ func NewBlockchain(s storage.Store, cfg config.Blockchain, log *zap.Logger) (*Bl
 				zap.Int("StateSyncInterval", cfg.StateSyncInterval))
 		}
 	}
+	if cfg.NeoFSStateSyncExtensions {
+		if !cfg.NeoFSBlockFetcher.Enabled {
+			return nil, errors.New("NeoFSStateSyncExtensions are enabled, but NeoFSBlockFetcher is off")
+		}
+		if cfg.StateSyncInterval <= 0 {
+			cfg.StateSyncInterval = defaultStateSyncInterval
+			log.Info("StateSyncInterval is not set or wrong, using default value",
+				zap.Int("StateSyncInterval", cfg.StateSyncInterval))
+		}
+	}
 	if cfg.RemoveUntraceableHeaders && !cfg.RemoveUntraceableBlocks {
 		return nil, errors.New("RemoveUntraceableHeaders is enabled, but RemoveUntraceableBlocks is not")
 	}

pkg/core/statesync/module.go

@@ -93,11 +93,15 @@ type Module struct {
 	billet *mpt.Billet
 
 	jumpCallback func(p uint32) error
+
+	// stageChangedCallback is an optional callback that is triggered whenever
+	// the sync stage changes.
+	stageChangedCallback func()
 }
 
 // NewModule returns new instance of statesync module.
 func NewModule(bc Ledger, stateMod *stateroot.Module, log *zap.Logger, s *dao.Simple, jumpCallback func(p uint32) error) *Module {
-	if !(bc.GetConfig().P2PStateExchangeExtensions && bc.GetConfig().Ledger.RemoveUntraceableBlocks) {
+	if !(bc.GetConfig().P2PStateExchangeExtensions && bc.GetConfig().Ledger.RemoveUntraceableBlocks) && !bc.GetConfig().NeoFSStateSyncExtensions {
 		return &Module{
 			dao:       s,
 			bc:        bc,
@@ -120,7 +124,13 @@ 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 func() {
+		if s.syncStage != oldStage {
+			s.notifyStageChanged()
+		}
+	}()
 	defer s.lock.Unlock()
 
 	if s.syncStage != none {
@@ -176,6 +186,20 @@ func (s *Module) Init(currChainHeight uint32) error {
 	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.stageChangedCallback = cb
+}
+
+// notifyStageChanged triggers stage change callback if it's set.
+func (s *Module) notifyStageChanged() {
+	if s.stageChangedCallback != nil {
+		s.stageChangedCallback()
+	}
+}
+
 // TemporaryPrefix accepts current storage prefix and returns prefix
 // to use for storing intermediate items during synchronization.
 func TemporaryPrefix(currPrefix storage.KeyPrefix) storage.KeyPrefix {
@@ -287,7 +311,13 @@ func (s *Module) getLatestSavedBlock(p uint32) uint32 {
 
 // AddHeaders validates and adds specified headers to the chain.
 func (s *Module) AddHeaders(hdrs ...*block.Header) error {
+	oldStage := s.syncStage
 	s.lock.Lock()
+	defer func() {
+		if s.syncStage != oldStage {
+			s.notifyStageChanged()
+		}
+	}()
 	defer s.lock.Unlock()
 
 	if s.syncStage != initialized {
@@ -306,7 +336,13 @@ 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 func() {
+		if s.syncStage != oldStage {
+			s.notifyStageChanged()
+		}
+	}()
 	defer s.lock.Unlock()
 
 	if s.syncStage&headersSynced == 0 || s.syncStage&blocksSynced != 0 {
@@ -359,7 +395,13 @@ func (s *Module) AddBlock(block *block.Block) error {
 // AddMPTNodes tries to add provided set of MPT nodes to the MPT billet if they are
 // not yet collected.
 func (s *Module) AddMPTNodes(nodes [][]byte) error {
+	oldStage := s.syncStage
 	s.lock.Lock()
+	defer func() {
+		if s.syncStage != oldStage {
+			s.notifyStageChanged()
+		}
+	}()
 	defer s.lock.Unlock()
 
 	if s.syncStage&headersSynced == 0 || s.syncStage&mptSynced != 0 {
@@ -425,6 +467,12 @@ 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
+	defer func() {
+		if s.syncStage != oldStage {
+			s.notifyStageChanged()
+		}
+	}()
 	if s.syncStage != headersSynced|mptSynced|blocksSynced {
 		return
 	}
@@ -484,6 +532,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 {
@@ -508,3 +564,13 @@ func (s *Module) GetUnknownMPTNodesBatch(limit int) []util.Uint256 {
 
 	return s.mptpool.GetBatch(limit)
 }
+
+// HeaderHeight returns the height of the latest stored header.
+func (s *Module) HeaderHeight() uint32 {
+	return s.bc.HeaderHeight()
+}
+
+// GetConfig returns current blockchain configuration.
+func (s *Module) GetConfig() config.Blockchain {
+	return s.bc.GetConfig()
+}