all: impl neox protocol to transfer blobs

Author: qianhh

consensus/dbft/dbft.go

@@ -190,6 +190,9 @@ type DBFT struct {
 	// about a new consensus payload to be sent.
 	broadcast func(m *dbftproto.Message) error
 
+	// blobsFeed is a feed to notify new blobs arrivals.
+	blobsFeed event.Feed
+
 	// requestTxs is a callback which is called to request the missing
 	// transactions from neighbor nodes. Requested transactions hashes are
 	// stored in txCbList and checked against the incoming transactions. If
@@ -572,6 +575,14 @@ func (c *DBFT) processBlockCb(b dbft.Block[common.Hash]) error {
 		}
 	}
 
+	// Notify new blobs arrival.
+	if len(dbftBlock.sidecars) > 0 {
+		c.blobsFeed.Send(core.NewBlobsEvent{Blob: &types.BlockBlobs{
+			BlockHash: res.Hash(),
+			Sidecars:  res.Sidecars(),
+		}})
+	}
+
 	c.postBlock(res.Header(), state)
 	return nil
 }

consensus/dbft/subscribe.go

@@ -0,0 +1,10 @@
+package dbft
+
+import (
+	"github.com/ethereum/go-ethereum/core"
+	"github.com/ethereum/go-ethereum/event"
+)
+
+func (dbft *DBFT) SubscribeBlobs(ch chan<- core.NewBlobsEvent) event.Subscription {
+	return dbft.blobsFeed.Subscribe(ch)
+}

core/blockchain.go

@@ -1365,16 +1365,6 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [
 	if n, err := bc.hc.ValidateHeaderChain(headers); err != nil {
 		return n, err
 	}
-
-	// check DA after cancun
-	lastBlk := blockChain[len(blockChain)-1]
-	if bc.chainConfig.DBFT != nil && bc.chainConfig.IsCancun(lastBlk.Number(), lastBlk.Time()) {
-		if _, err := CheckDataAvailableInBatch(bc, blockChain); err != nil {
-			log.Debug("CheckDataAvailableInBatch", "err", err)
-			return 0, err
-		}
-	}
-
 	// Hold the mutation lock
 	if !bc.chainmu.TryLock() {
 		return 0, errChainStopped
@@ -1433,7 +1423,7 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [
 			return 0, consensus.ErrUnknownAncestor
 		}
 		td := new(big.Int).Add(ptd, blockChain[0].Difficulty())
-		writeSize, err := rawdb.WriteAncientBlocksWithBlobs(bc.db, blockChain, receiptChain, td)
+		writeSize, err := rawdb.WriteAncientBlocks(bc.db, blockChain, receiptChain, td)
 		if err != nil {
 			log.Error("Error importing chain data to ancients", "err", err)
 			return 0, err
@@ -1499,9 +1489,6 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [
 			rawdb.WriteCanonicalHash(batch, block.Hash(), block.NumberU64())
 			rawdb.WriteBlock(batch, block)
 			rawdb.WriteReceipts(batch, block.Hash(), block.NumberU64(), receiptChain[i])
-			if bc.chainConfig.IsCancun(block.Number(), block.Time()) {
-				rawdb.WriteBlobSidecars(batch, block.Hash(), block.NumberU64(), block.Sidecars())
-			}
 
 			// Write everything belongs to the blocks into the database. So that
 			// we can ensure all components of body is completed(body, receipts)
@@ -1576,10 +1563,6 @@ func (bc *BlockChain) writeBlockWithoutState(block *types.Block, td *big.Int) (e
 	batch := bc.db.NewBatch()
 	rawdb.WriteTd(batch, block.Hash(), block.NumberU64(), td)
 	rawdb.WriteBlock(batch, block)
-	// if cancun is enabled, here need to write sidecars too
-	if bc.chainConfig.IsCancun(block.Number(), block.Time()) {
-		rawdb.WriteBlobSidecars(batch, block.Hash(), block.NumberU64(), block.Sidecars())
-	}
 	if err := batch.Write(); err != nil {
 		log.Crit("Failed to write block into disk", "err", err)
 	}
@@ -1618,17 +1601,10 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types.
 	rawdb.WriteTd(blockBatch, block.Hash(), block.NumberU64(), externTd)
 	rawdb.WriteBlock(blockBatch, block)
 	rawdb.WriteReceipts(blockBatch, block.Hash(), block.NumberU64(), receipts)
-	// if cancun is enabled, here need to write sidecars too
-	if bc.chainConfig.IsCancun(block.Number(), block.Time()) {
-		rawdb.WriteBlobSidecars(blockBatch, block.Hash(), block.NumberU64(), block.Sidecars())
-	}
 	rawdb.WritePreimages(blockBatch, statedb.Preimages())
 	if err := blockBatch.Write(); err != nil {
 		log.Crit("Failed to write block into disk", "err", err)
 	}
-	if bc.chainConfig.IsCancun(block.Number(), block.Time()) {
-		bc.sidecarsCache.Add(block.Hash(), block.Sidecars())
-	}
 	// Commit all cached state changes into underlying memory database.
 	root, err := statedb.Commit(block.NumberU64(), bc.chainConfig.IsEIP158(block.Number()), bc.chainConfig.IsCancun(block.Number(), block.Time()))
 	if err != nil {
@@ -1711,6 +1687,13 @@ func (bc *BlockChain) writeBlockAndSetHead(block *types.Block, receipts []*types
 	if err := bc.writeBlockWithState(block, receipts, state); err != nil {
 		return NonStatTy, err
 	}
+	// if cancun is enabled, here need to write sidecars too
+	if bc.chainConfig.IsCancun(block.Number(), block.Time()) && block.HasBlobTxs() {
+		if len(block.Sidecars()) > 0 {
+			rawdb.WriteBlobSidecars(bc.db, block.Hash(), block.NumberU64(), block.Sidecars())
+			bc.sidecarsCache.Add(block.Hash(), block.Sidecars())
+		}
+	}
 	currentBlock := bc.CurrentBlock()
 
 	// Reorganise the chain if the parent is not the head block
@@ -1843,14 +1826,6 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool, makeWitness
 			bc.chainHeadFeed.Send(ChainHeadEvent{Header: lastCanon.Header()})
 		}
 	}()
-
-	// check block data available first
-	if bc.chainConfig.DBFT != nil {
-		if index, err := CheckDataAvailableInBatch(bc, chain); err != nil {
-			return nil, index, err
-		}
-	}
-
 	// Start the parallel header verifier
 	headers := make([]*types.Header, len(chain))
 	for i, block := range chain {
@@ -2365,9 +2340,6 @@ func (bc *BlockChain) insertSideChain(block *types.Block, it *insertIterator, ma
 		// Append the next block to our batch
 		block := bc.GetBlock(hashes[i], numbers[i])
 
-		if bc.chainConfig.IsCancun(block.Number(), block.Time()) {
-			block = block.WithSidecars(bc.GetSidecarsByHash(hashes[i]))
-		}
 		blocks = append(blocks, block)
 		memory += block.Size()
 
@@ -2439,9 +2411,6 @@ func (bc *BlockChain) recoverAncestors(block *types.Block, makeWitness bool) (co
 		} else {
 			b = bc.GetBlock(hashes[i], numbers[i])
 		}
-		if bc.chainConfig.IsCancun(b.Number(), b.Time()) {
-			b = b.WithSidecars(bc.GetSidecarsByHash(b.Hash()))
-		}
 		if _, _, err := bc.insertChain(types.Blocks{b}, false, makeWitness && i == 0); err != nil {
 			return b.ParentHash(), err
 		}
@@ -2774,7 +2743,6 @@ func (bc *BlockChain) skipBlock(err error, it *insertIterator) bool {
 }
 
 // reportBlock logs a bad block error.
-// bad block need not save receipts & sidecars.
 func (bc *BlockChain) reportBlock(block *types.Block, res *ProcessResult, err error) {
 	var receipts types.Receipts
 	if res != nil {

core/blockchain_reader.go

@@ -171,8 +171,6 @@ func (bc *BlockChain) GetBlock(hash common.Hash, number uint64) *types.Block {
 	if block == nil {
 		return nil
 	}
-	sidecars := rawdb.ReadBlobSidecars(bc.db, hash, number)
-	block = block.WithSidecars(sidecars)
 	// Cache the found block for next time and return
 	bc.blockCache.Add(block.Hash(), block)
 	return block

core/events.go

@@ -29,6 +29,9 @@ type ReannoTxsEvent struct{ Txs []*types.Transaction }
 // NewMinedBlockEvent is posted when a block has been imported.
 type NewMinedBlockEvent struct{ Block *types.Block }
 
+// NewBlobsEvent is posted when new blobs are imported.
+type NewBlobsEvent struct{ Blob *types.BlockBlobs }
+
 // RemovedLogsEvent is posted when a reorg happens
 type RemovedLogsEvent struct{ Logs []*types.Log }
 

core/rawdb/accessors_chain.go

@@ -762,48 +762,6 @@ func WriteBlock(db ethdb.KeyValueWriter, block *types.Block) {
 	WriteHeader(db, block.Header())
 }
 
-// WriteAncientBlocksWithBlobs writes entire block data with blobs into ancient store and returns the total written size.
-func WriteAncientBlocksWithBlobs(db ethdb.AncientWriter, blocks []*types.Block, receipts []types.Receipts, td *big.Int) (int64, error) {
-	// find cancun index, it's used for new added blob ancient table
-	cancunIndex := -1
-	for i, block := range blocks {
-		if block.Sidecars() != nil {
-			cancunIndex = i
-			break
-		}
-	}
-	log.Debug("WriteAncientBlocks", "startAt", blocks[0].Number(), "cancunIndex", cancunIndex, "len", len(blocks))
-
-	var (
-		tdSum   = new(big.Int).Set(td)
-		preSize int64
-		err     error
-	)
-	if cancunIndex > 0 {
-		preSize, err = WriteAncientBlocks(db, blocks[:cancunIndex], receipts[:cancunIndex], td)
-		if err != nil {
-			return preSize, err
-		}
-		for i, block := range blocks[:cancunIndex] {
-			if i > 0 {
-				tdSum.Add(tdSum, block.Difficulty())
-			}
-		}
-		tdSum.Add(tdSum, blocks[cancunIndex].Difficulty())
-	}
-
-	// It will reset blob ancient table at cancunIndex
-	if cancunIndex >= 0 {
-		if err = ResetEmptyBlobAncientTable(db, blocks[cancunIndex].NumberU64()); err != nil {
-			return 0, err
-		}
-		blocks = blocks[cancunIndex:]
-		receipts = receipts[cancunIndex:]
-	}
-	postSize, err := WriteAncientBlocks(db, blocks, receipts, tdSum)
-	return preSize + postSize, err
-}
-
 // WriteAncientBlocks writes entire block data into ancient store and returns the total written size.
 func WriteAncientBlocks(db ethdb.AncientWriter, blocks []*types.Block, receipts []types.Receipts, td *big.Int) (int64, error) {
 	var (
@@ -846,11 +804,21 @@ func writeAncientBlock(op ethdb.AncientWriteOp, block *types.Block, header *type
 	if err := op.Append(ChainFreezerDifficultyTable, num, td); err != nil {
 		return fmt.Errorf("can't append block %d total difficulty: %v", num, err)
 	}
-	if block.Sidecars() != nil {
-		if err := op.Append(ChainFreezerBlobSidecarTable, num, block.Sidecars()); err != nil {
-			return fmt.Errorf("can't append block %d blobs: %v", num, err)
-		}
+	return nil
+}
+
+// WriteAncientSidecars writes the supplied blob sidecars into the ancient store and returns the total written size.
+func WriteAncientSidecars(db ethdb.AncientWriter, blockNum uint64, sidecars types.BlobSidecars) (int64, error) {
+	return db.ModifyAncients(func(op ethdb.AncientWriteOp) error {
+		return writeAncientSidecars(op, blockNum, sidecars)
+	})
+}
+
+func writeAncientSidecars(op ethdb.AncientWriteOp, blockNum uint64, sidecars types.BlobSidecars) error {
+	if err := op.Append(ChainFreezerBlobSidecarTable, blockNum, sidecars); err != nil {
+		return fmt.Errorf("can't append block %d blobs: %v", blockNum, err)
 	}
+
 	return nil
 }
 

core/types/blob_sidecar.go

@@ -61,3 +61,17 @@ func (s *BlobSidecar) SanityCheck(blockNumber *big.Int, blockHash common.Hash) e
 	}
 	return nil
 }
+
+// BlockBlobs represents a block hash and its associated sidecars.
+type BlockBlobs struct {
+	BlockHash common.Hash  `json:"blockHash"` // hash of the block
+	Sidecars  BlobSidecars `json:"sidecars"`  // sidecars associated with the block
+}
+
+// NewBlockBlobs creates a new BlockBlobs from a block.
+func NewBlockBlobs(block *Block) *BlockBlobs {
+	return &BlockBlobs{
+		BlockHash: block.Header().Hash(),
+		Sidecars:  block.Sidecars(),
+	}
+}

core/types/block.go

@@ -602,6 +602,16 @@ func (b *Block) Hash() common.Hash {
 	return h
 }
 
+// HasBlobTxs returns whether the block contains blob transactions.
+func (b *Block) HasBlobTxs() bool {
+	for _, tx := range b.transactions {
+		if tx.Type() == BlobTxType {
+			return true
+		}
+	}
+	return false
+}
+
 type Blocks []*Block
 
 // HeaderParentHashFromRLP returns the parentHash of an RLP-encoded

eth/api_backend.go

@@ -278,7 +278,23 @@ func (b *EthAPIBackend) GetReceipts(ctx context.Context, hash common.Hash) (type
 }
 
 func (b *EthAPIBackend) GetBlobSidecars(ctx context.Context, hash common.Hash) (types.BlobSidecars, error) {
-	return b.eth.blockchain.GetSidecarsByHash(hash), nil
+	sidecars := b.eth.blockchain.GetSidecarsByHash(hash)
+	if len(sidecars) > 0 {
+		return sidecars, nil
+	}
+
+	block, err := b.BlockByHash(ctx, hash)
+	if err != nil {
+		return nil, err
+	}
+	if !block.HasBlobTxs() {
+		return types.BlobSidecars{}, nil
+	}
+
+	if b.eth.neoxSrv != nil {
+		return b.eth.neoxSrv.RetrieveSidecarsByHash(hash)
+	}
+	return nil, errors.New("neox service not available")
 }
 
 func (b *EthAPIBackend) GetLogs(ctx context.Context, hash common.Hash, number uint64) ([][]*types.Log, error) {

eth/backend.go

@@ -49,6 +49,7 @@ import (
 	"github.com/ethereum/go-ethereum/eth/gasprice"
 	dbftproto "github.com/ethereum/go-ethereum/eth/protocols/dbft"
 	"github.com/ethereum/go-ethereum/eth/protocols/eth"
+	"github.com/ethereum/go-ethereum/eth/protocols/neox"
 	"github.com/ethereum/go-ethereum/eth/protocols/snap"
 	"github.com/ethereum/go-ethereum/eth/tracers"
 	"github.com/ethereum/go-ethereum/ethdb"
@@ -80,6 +81,7 @@ type Ethereum struct {
 	localTxTracker *locals.TxTracker
 	dbftSrv        *dbftproto.Service
 	blockchain     *core.BlockChain
+	neoxSrv        *neox.Service
 
 	handler *handler
 	discmix *enode.FairMix
@@ -369,6 +371,8 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 		if config.AntiMEVEnforceECDSABlockSignatureScheme {
 			bft.EnforceECDSASignatures()
 		}
+
+		eth.neoxSrv = neox.New(eth.blockchain, newNeoxCallback(chainDb, eth.blockchain), bft)
 	}
 
 	// Start the RPC service
@@ -642,6 +646,9 @@ func (s *Ethereum) SyncMode() downloader.SyncMode {
 func (s *Ethereum) Protocols() []p2p.Protocol {
 	protos := eth.MakeProtocols((*ethHandler)(s.handler), s.networkID, s.discmix)
 	protos = append(protos, s.dbftSrv.MakeProtocols()...)
+	if s.neoxSrv != nil {
+		protos = append(protos, s.neoxSrv.MakeProtocols()...)
+	}
 	if s.config.SnapshotCache > 0 {
 		protos = append(protos, snap.MakeProtocols((*snapHandler)(s.handler))...)
 	}
@@ -667,6 +674,12 @@ func (s *Ethereum) Start() error {
 	// start log indexer
 	s.filterMaps.Start()
 	go s.updateFilterMapsHeads()
+
+	// Start neox service
+	if s.neoxSrv != nil {
+		s.neoxSrv.Start()
+	}
+
 	return nil
 }
 
@@ -767,6 +780,9 @@ func (s *Ethereum) Stop() error {
 	s.discmix.Close()
 	s.dropper.Stop()
 	s.handler.Stop()
+	if s.neoxSrv != nil {
+		s.neoxSrv.Stop()
+	}
 
 	// Then stop everything else.
 	ch := make(chan struct{})