*: refactor for generic queue version

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

pkg/consensus/consensus.go

@@ -20,6 +20,7 @@ import (
 	"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
 	"github.com/nspcc-dev/neo-go/pkg/encoding/address"
 	"github.com/nspcc-dev/neo-go/pkg/io"
+	"github.com/nspcc-dev/neo-go/pkg/network/bqueue"
 	npayload "github.com/nspcc-dev/neo-go/pkg/network/payload"
 	"github.com/nspcc-dev/neo-go/pkg/smartcontract"
 	"github.com/nspcc-dev/neo-go/pkg/util"
@@ -58,7 +59,7 @@ type Ledger interface {
 
 // BlockQueuer is an interface to the block queue manager sufficient for Service.
 type BlockQueuer interface {
-	Put(block any) error
+	Put(block bqueue.Queueable) error
 }
 
 // Service represents a consensus instance.

pkg/core/blockchain.go

@@ -33,6 +33,7 @@ import (
 	"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
 	"github.com/nspcc-dev/neo-go/pkg/encoding/fixedn"
 	"github.com/nspcc-dev/neo-go/pkg/io"
+	"github.com/nspcc-dev/neo-go/pkg/network/bqueue"
 	"github.com/nspcc-dev/neo-go/pkg/smartcontract"
 	"github.com/nspcc-dev/neo-go/pkg/smartcontract/callflag"
 	"github.com/nspcc-dev/neo-go/pkg/smartcontract/manifest"
@@ -128,7 +129,7 @@ var (
 // adding new blocks or headers.
 type Blockchain struct {
 	HeaderHashes
-
+	bqueue.Blockqueuer[bqueue.Queueable]
 	config config.Blockchain
 
 	// The only way chain state changes is by adding blocks, so we can't
@@ -1608,6 +1609,19 @@ func (bc *Blockchain) GetStateSyncModule() *statesync.Module {
 	return statesync.NewModule(bc, bc.stateRoot, bc.log, bc.dao, bc.jumpToState)
 }
 
+func (bc *Blockchain) AddItem(b bqueue.Queueable) error {
+	return bc.AddBlock(b.(*block.Block))
+}
+
+func (bc *Blockchain) AddItems(hdrs ...bqueue.Queueable) error {
+	for _, h := range hdrs {
+		if err := bc.AddHeaders(h.(*block.Header)); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 // storeBlock performs chain update using the block given, it executes all
 // transactions with all appropriate side-effects and updates Blockchain state.
 // This is the only way to change Blockchain state.

pkg/core/statesync/module.go

@@ -30,6 +30,7 @@ import (
 	"github.com/nspcc-dev/neo-go/pkg/core/stateroot"
 	"github.com/nspcc-dev/neo-go/pkg/core/storage"
 	"github.com/nspcc-dev/neo-go/pkg/io"
+	"github.com/nspcc-dev/neo-go/pkg/network/bqueue"
 	"github.com/nspcc-dev/neo-go/pkg/util"
 	"go.uber.org/zap"
 )
@@ -62,7 +63,7 @@ const (
 
 // Ledger is the interface required from Blockchain for Module to operate.
 type Ledger interface {
-	AddHeaders(...*block.Header) error
+	bqueue.Blockqueuer[bqueue.Queueable]
 	BlockHeight() uint32
 	GetConfig() config.Blockchain
 	GetHeader(hash util.Uint256) (*block.Header, error)
@@ -95,6 +96,24 @@ type Module struct {
 	jumpCallback func(p uint32) error
 }
 
+func (s *Module) AddItem(b bqueue.Queueable) error {
+	return s.AddBlock(b.(*block.Block))
+
+}
+
+func (s *Module) AddItems(hdrs ...bqueue.Queueable) error {
+	for _, h := range hdrs {
+		if err := s.AddHeaders(h.(*block.Header)); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (s *Module) Height() uint32 {
+	return s.BlockHeight()
+}
+
 // 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) {
@@ -293,8 +312,13 @@ func (s *Module) AddHeaders(hdrs ...*block.Header) error {
 	if s.syncStage != initialized {
 		return errors.New("headers were not requested")
 	}
-
-	hdrsErr := s.bc.AddHeaders(hdrs...)
+	var hdrsErr error
+	for _, hdr := range hdrs {
+		hdrsErr = s.AddItems(hdr)
+		if hdrsErr != nil {
+			break
+		}
+	}
 	if s.bc.HeaderHeight() > s.syncPoint {
 		err := s.defineSyncStage()
 		if err != nil {

pkg/network/bqueue/queue.go

@@ -10,7 +10,7 @@ import (
 
 // Blockqueuer is an interface for a block queue.
 type Blockqueuer[Q Queueable] interface {
-	AddItem(Q Queueable) error
+	AddItem(Q) error
 	AddItems(...Q) error
 	Height() uint32
 }
@@ -108,7 +108,7 @@ func (bq *Queue[Q]) Run() {
 				break
 			}
 
-			err := bq.chain.AddItem(b)
+			err := bq.chain.AddItem(b.(Q))
 			if err != nil {
 				// The block might already be added by the consensus.
 				if bq.chain.Height() < b.GetIndex() {

pkg/network/server.go

@@ -59,7 +59,7 @@ type (
 	Ledger interface {
 		extpool.Ledger
 		mempool.Feer
-		bqueue.Blockqueuer
+		bqueue.Blockqueuer[bqueue.Queueable]
 		GetBlock(hash util.Uint256) (*block.Block, error)
 		GetConfig() config.Blockchain
 		GetHeader(hash util.Uint256) (*block.Header, error)
@@ -102,9 +102,9 @@ type (
 		transports        []Transporter
 		discovery         Discoverer
 		chain             Ledger
-		bQueue            *bqueue.Queue
-		bSyncQueue        *bqueue.Queue
-		bFetcherQueue     *bqueue.Queue
+		bQueue            *bqueue.Queue[*block.Block]
+		bSyncQueue        *bqueue.Queue[*block.Block]
+		bFetcherQueue     *bqueue.Queue[*block.Block]
 		mempool           *mempool.Pool
 		notaryRequestPool *mempool.Pool
 		extensiblePool    *extpool.Pool
@@ -219,15 +219,13 @@ func newServerFromConstructors(config ServerConfig, chain Ledger, stSync StateSy
 			}, s.notaryFeer)
 		})
 	}
-	s.bQueue = bqueue.New(chain, log, func(b *block.Block) {
-		s.tryStartServices()
-	}, bqueue.DefaultCacheSize, updateBlockQueueLenMetric, bqueue.NonBlocking, bqueue.Blocks)
+	s.bQueue = bqueue.New[*block.Block](chainAdapter{chain}, log, func(b *block.Block) { s.tryStartServices() }, bqueue.DefaultCacheSize, updateBlockQueueLenMetric, bqueue.NonBlocking)
 
-	s.bSyncQueue = bqueue.New(s.stateSync, log, nil, bqueue.DefaultCacheSize, updateBlockQueueLenMetric, bqueue.NonBlocking, bqueue.Blocks)
+	s.bSyncQueue = bqueue.New[*block.Block](blockAdapter{s.stateSync}, log, nil, bqueue.DefaultCacheSize, updateBlockQueueLenMetric, bqueue.NonBlocking)
 	if s.NeoFSBlockFetcherCfg.BQueueSize <= 0 {
 		s.NeoFSBlockFetcherCfg.BQueueSize = blockfetcher.DefaultQueueCacheSize
 	}
-	s.bFetcherQueue = bqueue.New(chain, log, nil, s.NeoFSBlockFetcherCfg.BQueueSize, updateBlockQueueLenMetric, bqueue.Blocking, bqueue.Blocks)
+	s.bFetcherQueue = bqueue.New[*block.Block](chainAdapter{chain}, log, nil, s.NeoFSBlockFetcherCfg.BQueueSize, updateBlockQueueLenMetric, bqueue.Blocking)
 	var err error
 	s.blockFetcher, err = blockfetcher.New(chain, s.NeoFSBlockFetcherCfg, log, s.bFetcherQueue.Put, sync.OnceFunc(func() { close(s.blockFetcherFin) }))
 	if err != nil {
@@ -281,6 +279,48 @@ func newServerFromConstructors(config ServerConfig, chain Ledger, stSync StateSy
 	return s, nil
 }
 
+type blockAdapter struct {
+	bc StateSync
+}
+
+func (a blockAdapter) AddItem(b *block.Block) error {
+	return a.bc.AddItem(b)
+}
+
+func (a blockAdapter) AddItems(bs ...*block.Block) error {
+	for _, b := range bs {
+		if err := a.bc.AddItem(b); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (a blockAdapter) Height() uint32 {
+	return a.bc.Height()
+}
+
+type chainAdapter struct {
+	chain Ledger
+}
+
+func (a chainAdapter) AddItem(b *block.Block) error {
+	return a.chain.AddItem(b)
+}
+
+func (a chainAdapter) AddItems(bs ...*block.Block) error {
+	for _, b := range bs {
+		if err := a.chain.AddItem(b); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (a chainAdapter) Height() uint32 {
+	return a.chain.Height()
+}
+
 // ID returns the servers ID.
 func (s *Server) ID() uint32 {
 	return s.id
@@ -373,7 +413,7 @@ func (s *Server) addService(svc Service) {
 }
 
 // GetBlockQueue returns the block queue instance managed by Server.
-func (s *Server) GetBlockQueue() *bqueue.Queue {
+func (s *Server) GetBlockQueue() *bqueue.Queue[*block.Block] {
 	return s.bQueue
 }
 
@@ -824,14 +864,14 @@ func (s *Server) requestBlocksOrHeaders(p Peer) error {
 		return nil
 	}
 	var (
-		bq              bqueue.Blockqueuer = s.chain
+		bq              bqueue.Blockqueuer[*block.Block] = chainAdapter{s.chain}
 		requestMPTNodes bool
 	)
 	if s.stateSync.IsActive() {
-		bq = s.stateSync
+		bq = blockAdapter{s.stateSync}
 		requestMPTNodes = s.stateSync.NeedMPTNodes()
 	}
-	if bq.BlockHeight() >= p.LastBlockIndex() {
+	if bq.Height() >= p.LastBlockIndex() {
 		return nil
 	}
 	err := s.requestBlocks(bq, p)
@@ -1138,7 +1178,13 @@ func (s *Server) handleHeadersCmd(p Peer, h *payload.Headers) error {
 	if s.blockFetcher.IsActive() {
 		return nil
 	}
-	return s.stateSync.AddHeaders(h.Hdrs...)
+	for _, header := range h.Hdrs {
+		err := s.stateSync.AddItems(header)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
 }
 
 // handleExtensibleCmd processes the received extensible payload.
@@ -1333,8 +1379,8 @@ func (s *Server) handleGetAddrCmd(p Peer) error {
 // 1. Block range is divided into chunks of payload.MaxHashesCount.
 // 2. Send requests for chunk in increasing order.
 // 3. After all requests have been sent, request random height.
-func (s *Server) requestBlocks(bq bqueue.Blockqueuer, p Peer) error {
-	pl := getRequestBlocksPayload(p, bq.BlockHeight(), &s.lastRequestedBlock)
+func (s *Server) requestBlocks(bq bqueue.Blockqueuer[*block.Block], p Peer) error {
+	pl := getRequestBlocksPayload(p, bq.Height(), &s.lastRequestedBlock)
 	lq, capLeft := s.bQueue.LastQueued()
 	if capLeft == 0 {
 		// No more blocks will fit into the queue.

pkg/network/state_sync.go

@@ -9,7 +9,7 @@ import (
 // StateSync represents state sync module.
 type StateSync interface {
 	AddMPTNodes([][]byte) error
-	bqueue.Blockqueuer
+	bqueue.Blockqueuer[bqueue.Queueable]
 	Init(currChainHeight uint32) error
 	IsActive() bool
 	IsInitialized() bool

pkg/services/blockfetcher/blockfetcher.go

@@ -15,6 +15,7 @@ import (
 	"github.com/nspcc-dev/neo-go/pkg/config"
 	"github.com/nspcc-dev/neo-go/pkg/core/block"
 	gio "github.com/nspcc-dev/neo-go/pkg/io"
+	"github.com/nspcc-dev/neo-go/pkg/network/bqueue"
 	"github.com/nspcc-dev/neo-go/pkg/services/helpers/neofs"
 	"github.com/nspcc-dev/neo-go/pkg/wallet"
 	"github.com/nspcc-dev/neofs-sdk-go/client"
@@ -59,7 +60,7 @@ type Service struct {
 
 	chain        Ledger
 	pool         poolWrapper
-	enqueueBlock func(any) error
+	enqueueBlock func(block bqueue.Queueable) error
 	account      *wallet.Account
 
 	oidsCh chan oid.ID
@@ -81,7 +82,7 @@ type Service struct {
 }
 
 // New creates a new BlockFetcher Service.
-func New(chain Ledger, cfg config.NeoFSBlockFetcher, logger *zap.Logger, putBlock func(any) error, shutdownCallback func()) (*Service, error) {
+func New(chain Ledger, cfg config.NeoFSBlockFetcher, logger *zap.Logger, putBlock func(block bqueue.Queueable) error, shutdownCallback func()) (*Service, error) {
 	var (
 		account *wallet.Account
 		err     error