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/txpool/validation.go

@@ -28,7 +28,6 @@ import (
 	"github.com/ethereum/go-ethereum/core/state"
 	"github.com/ethereum/go-ethereum/core/systemcontracts"
 	"github.com/ethereum/go-ethereum/core/types"
-	"github.com/ethereum/go-ethereum/crypto/kzg4844"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/params"
 )
@@ -155,7 +154,7 @@ func ValidateTransaction(tx *types.Transaction, head *types.Header, signer types
 			return fmt.Errorf("too many blobs in transaction: have %d, permitted %d", len(hashes), maxBlobs)
 		}
 		// Ensure commitments, proofs and hashes are valid
-		if err := validateBlobSidecar(hashes, sidecar); err != nil {
+		if err := core.ValidateBlobSidecar(hashes, sidecar); err != nil {
 			return err
 		}
 	}
@@ -167,26 +166,6 @@ func ValidateTransaction(tx *types.Transaction, head *types.Header, signer types
 	return nil
 }
 
-func validateBlobSidecar(hashes []common.Hash, sidecar *types.BlobTxSidecar) error {
-	if len(sidecar.Blobs) != len(hashes) {
-		return fmt.Errorf("invalid number of %d blobs compared to %d blob hashes", len(sidecar.Blobs), len(hashes))
-	}
-	if len(sidecar.Proofs) != len(hashes) {
-		return fmt.Errorf("invalid number of %d blob proofs compared to %d blob hashes", len(sidecar.Proofs), len(hashes))
-	}
-	if err := sidecar.ValidateBlobCommitmentHashes(hashes); err != nil {
-		return err
-	}
-	// Blob commitments match with the hashes in the transaction, verify the
-	// blobs themselves via KZG
-	for i := range sidecar.Blobs {
-		if err := kzg4844.VerifyBlobProof(&sidecar.Blobs[i], sidecar.Commitments[i], sidecar.Proofs[i]); err != nil {
-			return fmt.Errorf("invalid blob %d: %v", i, err)
-		}
-	}
-	return nil
-}
-
 // ValidationOptionsWithState define certain differences between stateful transaction
 // validation across the different pools without having to duplicate those checks.
 type ValidationOptionsWithState struct {

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

@@ -473,10 +473,6 @@ func (b *Block) Sidecars() BlobSidecars {
 	return b.sidecars
 }
 
-func (b *Block) CleanSidecars() {
-	b.sidecars = make(BlobSidecars, 0)
-}
-
 type writeCounter uint64
 
 func (c *writeCounter) Write(b []byte) (int, error) {
@@ -602,6 +598,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