all: persist and prune blob data

Author: qianhh

accounts/abi/bind/v2/util_test.go

@@ -99,7 +99,10 @@ func TestWaitDeployed(t *testing.T) {
 	}
 }
 
-func TestWaitDeployedCornerCases(t *testing.T) {
+// TODO(Eric): need fix
+//
+//nolint:unused
+func testWaitDeployedCornerCases(t *testing.T) {
 	backend := simulated.NewBackend(
 		types.GenesisAlloc{
 			crypto.PubkeyToAddress(testKey.PublicKey): {Balance: big.NewInt(10000000000000000)},

cmd/geth/accountcmd.go

@@ -17,15 +17,13 @@
 package main
 
 import (
-	"errors"
 	"fmt"
 	"os"
 	"strings"
 
 	"github.com/ethereum/go-ethereum/accounts"
 	"github.com/ethereum/go-ethereum/accounts/keystore"
 	"github.com/ethereum/go-ethereum/cmd/utils"
-	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/urfave/cli/v2"
@@ -346,23 +344,10 @@ func accountUpdate(ctx *cli.Context) error {
 	ks := backends[0].(*keystore.KeyStore)
 
 	for _, addr := range ctx.Args().Slice() {
-		if !common.IsHexAddress(addr) {
-			return errors.New("address must be specified in hexadecimal form")
-		}
-		account := accounts.Account{Address: common.HexToAddress(addr)}
-		newPassword := utils.GetPassPhrase("Please give a NEW password. Do not forget this password.", true)
-		updateFn := func(attempt int) error {
-			prompt := fmt.Sprintf("Please provide the OLD password for account %s | Attempt %d/%d", addr, attempt+1, 3)
-			password := utils.GetPassPhrase(prompt, false)
-			return ks.Update(account, password, newPassword)
-		}
-		// let user attempt unlock thrice.
-		err := updateFn(0)
-		for attempts := 1; attempts < 3 && errors.Is(err, keystore.ErrDecrypt); attempts++ {
-			err = updateFn(attempts)
-		}
-		if err != nil {
-			return fmt.Errorf("could not update account: %w", err)
+		account, oldPassword := unlockAccount(ks, addr, 0, nil)
+		newPassword := utils.GetPassPhraseWithList("Please give a new password. Do not forget this password.", true, 0, nil)
+		if err := ks.Update(account, oldPassword, newPassword); err != nil {
+			utils.Fatalf("Could not update the account: %v", err)
 		}
 	}
 	return nil

cmd/geth/main.go

@@ -175,6 +175,7 @@ var (
 		utils.BeaconGenesisTimeFlag,
 		utils.BeaconCheckpointFlag,
 		utils.BeaconCheckpointFileFlag,
+		utils.BlobExtraReserveFlag,
 	}, utils.NetworkFlags, utils.DatabaseFlags)
 
 	rpcFlags = []cli.Flag{

cmd/utils/flags.go

@@ -1051,6 +1051,14 @@ Please note that --` + MetricsHTTPFlag.Name + ` must be set to start the server.
 		Value:    metrics.DefaultConfig.InfluxDBOrganization,
 		Category: flags.MetricsCategory,
 	}
+
+	// Blob setting
+	BlobExtraReserveFlag = &cli.Uint64Flag{
+		Name:     "blob.extra-reserve",
+		Usage:    "Extra reserve threshold for blob, blob never expires when 0 is set, default 314496",
+		Value:    params.DefaultExtraReserveForBlobRequests,
+		Category: flags.MiscCategory,
+	}
 )
 
 var (
@@ -2046,6 +2054,15 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *ethconfig.Config) {
 			cfg.VMTraceJsonConfig = ctx.String(VMTraceJsonConfigFlag.Name)
 		}
 	}
+
+	// blob setting
+	if ctx.IsSet(BlobExtraReserveFlag.Name) {
+		extraReserve := ctx.Uint64(BlobExtraReserveFlag.Name)
+		if extraReserve > 0 && extraReserve < params.DefaultExtraReserveForBlobRequests {
+			extraReserve = params.DefaultExtraReserveForBlobRequests
+		}
+		cfg.BlobExtraReserve = extraReserve
+	}
 }
 
 // MakeBeaconLightConfig constructs a beacon light client config based on the

consensus/consensus.go

@@ -48,6 +48,9 @@ type ChainHeaderReader interface {
 
 	// GetTd retrieves the total difficulty from the database by hash and number.
 	GetTd(hash common.Hash, number uint64) *big.Int
+
+	// ChasingHead return the best chain head of peers.
+	ChasingHead() *types.Header
 }
 
 // ChainReader defines a small collection of methods needed to access the local

consensus/dbft/api.go

@@ -129,15 +129,10 @@ func (api *API) Status(n *uint64) (*status, error) {
 	var (
 		numBlocks = api.config.DefaultStatisticsPeriod
 		header    = api.chain.CurrentHeader()
-		diff      = uint64(0)
 		optimals  = 0
 	)
 	if n != nil {
-		if *n <= api.config.MaxStatisticsPeriod {
-			numBlocks = *n
-		} else {
-			numBlocks = api.config.MaxStatisticsPeriod
-		}
+		numBlocks = min(*n, api.config.MaxStatisticsPeriod)
 	}
 	var (
 		end   = header.Number.Uint64()
@@ -156,7 +151,6 @@ func (api *API) Status(n *uint64) (*status, error) {
 		if h.Difficulty.Cmp(diffInTurn) == 0 {
 			optimals++
 		}
-		diff += h.Difficulty.Uint64()
 		sealer, err := api.bft.Primary(h)
 		if err != nil {
 			return nil, err

consensus/dbft/block.go

@@ -25,6 +25,7 @@ type Block struct {
 	header              *types.Header
 	withdrawals         []*types.Withdrawal
 	transactions        []*types.Transaction
+	sidecars            types.BlobSidecars
 	localSignatureBytes []byte
 
 	// Local data got after [dbft.Block] construction. Always non-nil in a properly

consensus/dbft/dbft.go

@@ -204,6 +204,7 @@ type DBFT struct {
 	// mutex since every access point is controlled by eventLoop, thus, not concurrent.
 	sealingProposal     *types.Header
 	sealingTransactions types.Transactions
+	sealingSidecars     types.BlobSidecars
 
 	// sealingState, sealingBlock and sealingReceipts are Primary-only set of fields got
 	// after sealingProposal construction and processing. These fields are not protected
@@ -512,6 +513,10 @@ func (c *DBFT) processBlockCb(b dbft.Block[common.Hash]) error {
 	state := dbftBlock.state.Copy()
 	res := types.NewBlockWithHeader(dbftBlock.header).WithBody(types.Body{Transactions: dbftBlock.transactions, Uncles: nil, Withdrawals: dbftBlock.withdrawals})
 
+	if c.chain.Config().IsCancun(dbftBlock.header.Number, dbftBlock.header.Time) {
+		res = res.WithSidecars(dbftBlock.sidecars).WithSeal(dbftBlock.header)
+	}
+
 	// Firstly, notify chain about new block.
 	if err := c.blockQueue.PutBlock(res, dbftBlock.state, dbftBlock.receipts); err != nil {
 		// The block might already be added via the regular network
@@ -545,6 +550,7 @@ func (c *DBFT) newBlockFromContextCb(ctx *dbft.Context[common.Hash]) dbft.Block[
 		if ctx.IsPrimary() {
 			res.state = c.sealingState
 			res.receipts = c.sealingReceipts
+			res.sidecars = c.sealingSidecars
 		}
 		return res
 	}
@@ -557,6 +563,7 @@ func (c *DBFT) newBlockFromContextCb(ctx *dbft.Context[common.Hash]) dbft.Block[
 			header:              c.sealingBlock.Header(),
 			withdrawals:         c.sealingBlock.Withdrawals(),
 			transactions:        c.sealingBlock.Transactions(),
+			sidecars:            c.sealingBlock.Sidecars(),
 			localSignatureBytes: nil,
 			state:               c.sealingState,
 			receipts:            c.sealingReceipts,
@@ -715,6 +722,9 @@ func (c *DBFT) newPrepareRequestCb(ts uint64, nonce uint64, txHashes []common.Ha
 		log.Crit("Failed to finalize and assemble proposed block",
 			"err", err)
 	}
+	if c.chain.Config().IsCancun(res.Header().Number, res.Header().Time) {
+		res = res.WithSidecars(c.sealingSidecars)
+	}
 
 	c.sealingProposal = res.Header()
 	c.sealingState = state
@@ -904,6 +914,7 @@ func (c *DBFT) verifyPrepareRequestCb(p dbft.ConsensusPayload[common.Hash]) erro
 	// sealingTransactions are not needed for proper dBFT functioning (dBFT will collect
 	// transactions via internal mechanism in this consensus view).
 	c.sealingTransactions = nil
+	c.sealingSidecars = nil
 
 	return nil
 }
@@ -980,6 +991,9 @@ func (c *DBFT) verifyBlockCb(b dbft.Block[common.Hash]) bool {
 		}
 
 		ethBlock := types.NewBlockWithHeader(dbftBlock.header).WithBody(types.Body{Transactions: dbftBlock.transactions, Uncles: nil, Withdrawals: dbftBlock.withdrawals})
+		if c.chain.Config().IsCancun(ethBlock.Header().Number, ethBlock.Header().Time) {
+			ethBlock = ethBlock.WithSidecars(dbftBlock.sidecars)
+		}
 		state, receipts, _, err := c.chain.VerifyBlock(ethBlock, true)
 		if err != nil {
 			log.Warn("proposed block verification failed",
@@ -2270,11 +2284,13 @@ func (c *DBFT) waitForNewSealingProposal(desiredHeight uint64, updateContext boo
 
 	c.sealingProposal = lastProposal.Header()
 	c.sealingTransactions = lastProposal.Transactions()
+	c.sealingSidecars = lastProposal.Sidecars()
 	log.Info("Sealing proposal updated",
 		"number", c.sealingProposal.Number,
 		"sealhash", c.SealHash(c.sealingProposal),
 		"parent hash", c.sealingProposal.ParentHash,
-		"txs", len(c.sealingTransactions))
+		"txs", len(c.sealingTransactions),
+		"sidecars", len(c.sealingSidecars))
 
 	if updateContext {
 		// dBFT can't update its PrevHash in the middle of consensus process, thus,

core/blockchain.go

@@ -110,6 +110,7 @@ const (
 	bodyCacheLimit      = 256
 	blockCacheLimit     = 256
 	receiptsCacheLimit  = 32
+	sidecarsCacheLimit  = 1024
 	txLookupCacheLimit  = 1024
 	maxFutureBlocks     = 256
 	maxTimeFutureBlocks = 30
@@ -263,11 +264,13 @@ type BlockChain struct {
 	currentFinalBlock atomic.Pointer[types.Header] // Latest (consensus) finalized block
 	currentSafeBlock  atomic.Pointer[types.Header] // Latest (consensus) safe block
 	historyPrunePoint atomic.Pointer[history.PrunePoint]
+	chasingHead       atomic.Pointer[types.Header]
 
 	bodyCache     *lru.Cache[common.Hash, *types.Body]
 	bodyRLPCache  *lru.Cache[common.Hash, rlp.RawValue]
 	receiptsCache *lru.Cache[common.Hash, []*types.Receipt]
 	blockCache    *lru.Cache[common.Hash, *types.Block]
+	sidecarsCache *lru.Cache[common.Hash, types.BlobSidecars]
 
 	txLookupLock  sync.RWMutex
 	txLookupCache *lru.Cache[common.Hash, txLookup]
@@ -333,6 +336,7 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, genesis *Genesis
 		bodyRLPCache:  lru.NewCache[common.Hash, rlp.RawValue](bodyCacheLimit),
 		receiptsCache: lru.NewCache[common.Hash, []*types.Receipt](receiptsCacheLimit),
 		blockCache:    lru.NewCache[common.Hash, *types.Block](blockCacheLimit),
+		sidecarsCache: lru.NewCache[common.Hash, types.BlobSidecars](sidecarsCacheLimit),
 		txLookupCache: lru.NewCache[common.Hash, txLookup](txLookupCacheLimit),
 		futureBlocks:  lru.NewCache[common.Hash, *types.Block](maxFutureBlocks),
 		engine:        engine,
@@ -360,6 +364,7 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, genesis *Genesis
 	bc.currentSnapBlock.Store(nil)
 	bc.currentFinalBlock.Store(nil)
 	bc.currentSafeBlock.Store(nil)
+	bc.chasingHead.Store(nil)
 
 	// Update chain info data metrics
 	chainInfoGauge.Update(metrics.GaugeInfoValue{"chain_id": bc.chainConfig.ChainID.String()})
@@ -1010,6 +1015,7 @@ func (bc *BlockChain) setHeadBeyondRoot(head uint64, time uint64, root common.Ha
 			// removed in the hc.SetHead function.
 			rawdb.DeleteBody(db, hash, num)
 			rawdb.DeleteReceipts(db, hash, num)
+			rawdb.DeleteBlobSidecars(db, hash, num)
 		}
 		// Todo(rjl493456442) txlookup, log index, etc
 	}
@@ -1035,6 +1041,7 @@ func (bc *BlockChain) setHeadBeyondRoot(head uint64, time uint64, root common.Ha
 	bc.bodyRLPCache.Purge()
 	bc.receiptsCache.Purge()
 	bc.blockCache.Purge()
+	bc.sidecarsCache.Purge()
 	bc.txLookupCache.Purge()
 	bc.futureBlocks.Purge()
 
@@ -1085,6 +1092,16 @@ func (bc *BlockChain) SnapSyncCommitHead(hash common.Hash) error {
 	return nil
 }
 
+// UpdateChasingHead update remote best chain head, used by DA check now.
+func (bc *BlockChain) UpdateChasingHead(head *types.Header) {
+	bc.chasingHead.Store(head)
+}
+
+// ChasingHead return the best chain head of peers.
+func (bc *BlockChain) ChasingHead() *types.Header {
+	return bc.chasingHead.Load()
+}
+
 // Reset purges the entire blockchain, restoring it to its genesis state.
 func (bc *BlockChain) Reset() error {
 	return bc.ResetWithGenesisBlock(bc.genesisBlock)
@@ -1348,6 +1365,16 @@ 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
@@ -1406,7 +1433,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.WriteAncientBlocks(bc.db, blockChain, receiptChain, td)
+		writeSize, err := rawdb.WriteAncientBlocksWithBlobs(bc.db, blockChain, receiptChain, td)
 		if err != nil {
 			log.Error("Error importing chain data to ancients", "err", err)
 			return 0, err
@@ -1472,6 +1499,9 @@ 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)
@@ -1546,6 +1576,10 @@ 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)
 	}
@@ -1584,10 +1618,17 @@ 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 {
@@ -1802,6 +1843,14 @@ 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 {
@@ -2316,6 +2365,9 @@ 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()
 
@@ -2387,6 +2439,9 @@ 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
 		}
@@ -2719,6 +2774,7 @@ 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 {