forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-20 10:52:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

core, accounts, eth, trie: handle genesis state missing

Browse source
This commit is contained in:
Gary Rong 2023-09-20 16:42:13 +08:00
parent b85c183ea7
commit d02618b55d
19 changed files with 317 additions and 160 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
accounts/abi/bind/backends/simulated.go
View file

@ -199,7 +199,6 @@ func (b *SimulatedBackend) CodeAt(ctx context.Context, contract common.Address,
if err != nil {
return nil, err
}
return stateDB.GetCode(contract), nil
}
@ -212,7 +211,6 @@ func (b *SimulatedBackend) BalanceAt(ctx context.Context, contract common.Addres
if err != nil {
return nil, err
}
return stateDB.GetBalance(contract), nil
}
@ -225,7 +223,6 @@ func (b *SimulatedBackend) NonceAt(ctx context.Context, contract common.Address,
if err != nil {
return 0, err
}
return stateDB.GetNonce(contract), nil
}
@ -238,7 +235,6 @@ func (b *SimulatedBackend) StorageAt(ctx context.Context, contract common.Addres
if err != nil {
return nil, err
}
val := stateDB.GetState(contract, key)
return val[:], nil
}
@ -700,8 +696,10 @@ func (b *SimulatedBackend) SendTransaction(ctx context.Context, tx *types.Transa
}
block.AddTxWithChain(b.blockchain, tx)
})
stateDB, _ := b.blockchain.State()
stateDB, err := b.blockchain.State()
if err != nil {
return err
}
b.pendingBlock = blocks[0]
b.pendingState, _ = state.New(b.pendingBlock.Root(), stateDB.Database(), nil)
b.pendingReceipts = receipts[0]
@ -821,11 +819,12 @@ func (b *SimulatedBackend) AdjustTime(adjustment time.Duration) error {
blocks, _ := core.GenerateChain(b.config, block, ethash.NewFaker(), b.database, 1, func(number int, block *core.BlockGen) {
block.OffsetTime(int64(adjustment.Seconds()))
})
stateDB, _ := b.blockchain.State()
stateDB, err := b.blockchain.State()
if err != nil {
return err
}
b.pendingBlock = blocks[0]
b.pendingState, _ = state.New(b.pendingBlock.Root(), stateDB.Database(), nil)
return nil
}

55
core/blockchain.go
View file

@ -337,17 +337,17 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, genesis *Genesis
if err := bc.loadLastState(); err != nil {
return nil, err
}
// Make sure the state associated with the block is available
// Make sure the state associated with the block is available, or log out
// if there is no available state, waiting for state sync.
head := bc.CurrentBlock()
if !bc.HasState(head.Root) {
if head.Number.Uint64() == 0 {
// The genesis state is missing, which is only possible in the path-based
// scheme. This situation occurs when the state syncer overwrites it.
//
// The solution is to reset the state to the genesis state. Although it may not
// match the sync target, the state healer will later address and correct any
// inconsistencies.
bc.resetState()
// scheme. This situation occurs when the initial state sync is not finished
// yet, or the chain head is rewound below the pivot point. In both scenario,
// there is no possible recovery approach except for rerunning a snap sync.
// Do nothing here until the state syncer picks it up.
log.Info("Genesis state is missing, wait state sync")
} else {
// Head state is missing, before the state recovery, find out the
// disk layer point of snapshot(if it's enabled). Make sure the
@ -630,28 +630,6 @@ func (bc *BlockChain) SetSafe(header *types.Header) {
}
}
// resetState resets the persistent state to genesis state if it's not present.
func (bc *BlockChain) resetState() {
// Short circuit if the genesis state is already present.
root := bc.genesisBlock.Root()
if bc.HasState(root) {
return
}
// Reset the state database to empty for committing genesis state.
// Note, it should only happen in path-based scheme and Reset function
// is also only call-able in this mode.
if bc.triedb.Scheme() == rawdb.PathScheme {
if err := bc.triedb.Reset(types.EmptyRootHash); err != nil {
log.Crit("Failed to clean state", "err", err) // Shouldn't happen
}
}
// Write genesis state into database.
if err := CommitGenesisState(bc.db, bc.triedb, bc.genesisBlock.Hash()); err != nil {
log.Crit("Failed to commit genesis state", "err", err)
}
log.Info("Reset state to genesis", "root", root)
}
// setHeadBeyondRoot rewinds the local chain to a new head with the extra condition
// that the rewind must pass the specified state root. This method is meant to be
// used when rewinding with snapshots enabled to ensure that we go back further than
@ -678,16 +656,15 @@ func (bc *BlockChain) setHeadBeyondRoot(head uint64, time uint64, root common.Ha
pivot := rawdb.ReadLastPivotNumber(bc.db)
frozen, _ := bc.db.Ancients()
updateFn := func(db ethdb.KeyValueWriter, header *types.Header) (*types.Header, bool) {
// Rewind the blockchain, ensuring we don't end up with a stateless head
// block. Note, depth equality is permitted to allow using SetHead as a
// chain reparation mechanism without deleting any data!
updateFn := func(db ethdb.KeyValueWriter, header *types.Header) (*types.Header, bool) {
if currentBlock := bc.CurrentBlock(); currentBlock != nil && header.Number.Uint64() <= currentBlock.Number.Uint64() {
newHeadBlock := bc.GetBlock(header.Hash(), header.Number.Uint64())
if newHeadBlock == nil {
log.Error("Gap in the chain, rewinding to genesis", "number", header.Number, "hash", header.Hash())
newHeadBlock = bc.genesisBlock
bc.resetState()
} else {
// Block exists, keep rewinding until we find one with state,
// keeping rewinding until we exceed the optional threshold
@ -715,16 +692,14 @@ func (bc *BlockChain) setHeadBeyondRoot(head uint64, time uint64, root common.Ha
}
}
if beyondRoot || newHeadBlock.NumberU64() == 0 {
if newHeadBlock.NumberU64() == 0 {
bc.resetState()
} else if !bc.HasState(newHeadBlock.Root()) {
if !bc.HasState(newHeadBlock.Root()) && bc.stateRecoverable(newHeadBlock.Root()) {
// Rewind to a block with recoverable state. If the state is
// missing, run the state recovery here.
if err := bc.triedb.Recover(newHeadBlock.Root()); err != nil {
log.Crit("Failed to rollback state", "err", err) // Shouldn't happen
}
}
log.Debug("Rewound to block with state", "number", newHeadBlock.NumberU64(), "hash", newHeadBlock.Hash())
}
break
}
log.Debug("Skipping block with threshold state", "number", newHeadBlock.NumberU64(), "hash", newHeadBlock.Hash(), "root", newHeadBlock.Root())
@ -733,6 +708,14 @@ func (bc *BlockChain) setHeadBeyondRoot(head uint64, time uint64, root common.Ha
}
rawdb.WriteHeadBlockHash(db, newHeadBlock.Hash())
// The genesis state is missing, which is only possible in the path-based
// scheme. This situation occurs when the chain head is rewound below the
// pivot point. In this scenario, there is no possible recovery approach
// except for rerunning a snap sync. Do nothing here until the state syncer
// picks it up.
if newHeadBlock.NumberU64() == 0 && !bc.HasState(newHeadBlock.Root()) {
log.Info("Genesis state is missing, wait state sync")
}
// Degrade the chain markers if they are explicitly reverted.
// In theory we should update all in-memory markers in the
// last step, however the direction of SetHead is from high
@ -838,7 +821,7 @@ func (bc *BlockChain) SnapSyncCommitHead(hash common.Hash) error {
// Reset the trie database with the fresh snap synced state.
root := block.Root()
if bc.triedb.Scheme() == rawdb.PathScheme {
if err := bc.triedb.Reset(root); err != nil {
if err := bc.triedb.Activate(root); err != nil {
return err
}
}

21
core/rawdb/accessors_sync.go
View file

@ -76,3 +76,24 @@ func DeleteSkeletonHeader(db ethdb.KeyValueWriter, number uint64) {
log.Crit("Failed to delete skeleton header", "err", err)
}
}
const (
StateSyncing = uint8(1) // flags the state snap sync is not completed yet
StateSynced = uint8(2) // flags the state snap sync is completed
)
// ReadSnapSyncStatusFlag retrieves the state snap sync status flag.
func ReadSnapSyncStatusFlag(db ethdb.KeyValueReader) uint8 {
blob, err := db.Get(syncStatusFlagKey)
if err != nil || len(blob) != 1 {
return 0
}
return blob[0]
}
// WriteSnapSyncStatusFlag stores the state snap sync status flag into database.
func WriteSnapSyncStatusFlag(db ethdb.KeyValueWriter, flag uint8) {
if err := db.Put(syncStatusFlagKey, []byte{flag}); err != nil {
log.Crit("Failed to store sync status flag", "err", err)
}
}

2
core/rawdb/database.go
View file

@ -555,7 +555,7 @@ func InspectDatabase(db ethdb.Database, keyPrefix, keyStart []byte) error {
lastPivotKey, fastTrieProgressKey, snapshotDisabledKey, SnapshotRootKey, snapshotJournalKey,
snapshotGeneratorKey, snapshotRecoveryKey, txIndexTailKey, fastTxLookupLimitKey,
uncleanShutdownKey, badBlockKey, transitionStatusKey, skeletonSyncStatusKey,
persistentStateIDKey, trieJournalKey, snapshotSyncStatusKey,
persistentStateIDKey, trieJournalKey, snapshotSyncStatusKey, syncStatusFlagKey,
} {
if bytes.Equal(key, meta) {
metadata.Add(size)

3
core/rawdb/schema.go
View file

@ -91,6 +91,9 @@ var (
// transitionStatusKey tracks the eth2 transition status.
transitionStatusKey = []byte("eth2-transition")
// syncStatusFlagKey flags that status of state sync.
syncStatusFlagKey = []byte("sync-status")
// Data item prefixes (use single byte to avoid mixing data types, avoid `i`, used for indexes).
headerPrefix = []byte("h") // headerPrefix + num (uint64 big endian) + hash -> header
headerTDSuffix = []byte("t") // headerPrefix + num (uint64 big endian) + hash + headerTDSuffix -> td

109
core/txpool/txpool.go
View file

@ -21,6 +21,7 @@ import (
"fmt"
"math/big"
"sync"
"sync/atomic"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core"
@ -55,6 +56,9 @@ type BlockChain interface {
// CurrentBlock returns the current head of the chain.
CurrentBlock() *types.Header
// HasState checks if state is present in the database or not.
HasState(common.Hash) bool
// SubscribeChainHeadEvent subscribes to new blocks being added to the chain.
SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) event.Subscription
}
@ -65,6 +69,7 @@ type BlockChain interface {
// They exit the pool when they are included in the blockchain or evicted due to
// resource constraints.
type TxPool struct {
inited atomic.Bool // Flag whether the subpools are initialized
subpools []SubPool // List of subpools for specialized transaction handling
reservations map[common.Address]SubPool // Map with the account to pool reservations
@ -87,18 +92,57 @@ func New(gasTip *big.Int, chain BlockChain, subpools []SubPool) (*TxPool, error)
reservations: make(map[common.Address]SubPool),
quit: make(chan chan error),
}
for i, subpool := range subpools {
if err := subpool.Init(gasTip, head, pool.reserver(i, subpool)); err != nil {
for j := i - 1; j >= 0; j-- {
subpools[j].Close()
}
return nil, err
}
}
if chain.HasState(head.Root) {
pool.init(gasTip, head)
go pool.loop(head, chain)
} else {
go pool.lazyInit(gasTip, chain)
}
return pool, nil
}
// init performs the initialization for subpools.
func (p *TxPool) init(gasTip *big.Int, head *types.Header) {
for i, subpool := range p.subpools {
if err := subpool.Init(gasTip, head, p.reserver(i, subpool)); err != nil {
for j := i - 1; j >= 0; j-- {
p.subpools[j].Close()
}
// TODO(rjl493456442) can we shutdown the node gracefully?
log.Crit("Failed to initialize subpool", "err", err)
}
}
p.inited.Store(true)
}
// lazyInit waits the signal that state sync is completed and initializes the subpools.
func (p *TxPool) lazyInit(gasTip *big.Int, chain BlockChain) {
var (
newHeadCh = make(chan core.ChainHeadEvent)
newHeadSub = chain.SubscribeChainHeadEvent(newHeadCh)
)
defer newHeadSub.Unsubscribe()
var errc chan error
for errc == nil {
select {
case event := <-newHeadCh:
head := event.Block.Header()
if !chain.HasState(head.Root) {
continue // shouldn't happen
}
p.init(gasTip, head)
go p.loop(head, chain)
return
case errc = <-p.quit:
// Termination requested, break out on the next loop round
}
}
// Notify the closer of termination (no error possible for now)
errc <- nil
}
// reserver is a method to create an address reservation callback to exclusively
// assign/deassign addresses to/from subpools. This can ensure that at any point
// in time, only a single subpool is able to manage an account, avoiding cross
@ -156,14 +200,16 @@ func (p *TxPool) Close() error {
errs = append(errs, err)
}
// Terminate each subpool
// Terminate each subpool if they are initialized
if p.inited.Load() {
for _, subpool := range p.subpools {
if err := subpool.Close(); err != nil {
errs = append(errs, err)
}
}
}
if len(errs) > 0 {
return fmt.Errorf("subpool close errors: %v", errs)
return fmt.Errorf("txpool close errors: %v", errs)
}
return nil
}
@ -232,6 +278,10 @@ func (p *TxPool) loop(head *types.Header, chain BlockChain) {
// SetGasTip updates the minimum gas tip required by the transaction pool for a
// new transaction, and drops all transactions below this threshold.
func (p *TxPool) SetGasTip(tip *big.Int) {
if !p.inited.Load() {
log.Info("Skip tip adjustment as txpool hasn't been initialized")
return
}
for _, subpool := range p.subpools {
subpool.SetGasTip(tip)
}
@ -240,6 +290,9 @@ func (p *TxPool) SetGasTip(tip *big.Int) {
// Has returns an indicator whether the pool has a transaction cached with the
// given hash.
func (p *TxPool) Has(hash common.Hash) bool {
if !p.inited.Load() {
return false
}
for _, subpool := range p.subpools {
if subpool.Has(hash) {
return true
@ -250,6 +303,9 @@ func (p *TxPool) Has(hash common.Hash) bool {
// Get returns a transaction if it is contained in the pool, or nil otherwise.
func (p *TxPool) Get(hash common.Hash) *types.Transaction {
if !p.inited.Load() {
return nil
}
for _, subpool := range p.subpools {
if tx := subpool.Get(hash); tx != nil {
return tx
@ -262,6 +318,13 @@ func (p *TxPool) Get(hash common.Hash) *types.Transaction {
// to the large transaction churn, add may postpone fully integrating the tx
// to a later point to batch multiple ones together.
func (p *TxPool) Add(txs []*types.Transaction, local bool, sync bool) []error {
if !p.inited.Load() {
errs := make([]error, len(txs))
for i := 0; i < len(errs); i++ {
errs[i] = errors.New("txpool is not initialized")
}
return errs
}
// Split the input transactions between the subpools. It shouldn't really
// happen that we receive merged batches, but better graceful than strange
// errors.
@ -307,6 +370,9 @@ func (p *TxPool) Add(txs []*types.Transaction, local bool, sync bool) []error {
// Pending retrieves all currently processable transactions, grouped by origin
// account and sorted by nonce.
func (p *TxPool) Pending(enforceTips bool) map[common.Address][]*LazyTransaction {
if !p.inited.Load() {
return nil
}
txs := make(map[common.Address][]*LazyTransaction)
for _, subpool := range p.subpools {
for addr, set := range subpool.Pending(enforceTips) {
@ -329,6 +395,9 @@ func (p *TxPool) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.Subscrip
// Nonce returns the next nonce of an account, with all transactions executable
// by the pool already applied on top.
func (p *TxPool) Nonce(addr common.Address) uint64 {
if !p.inited.Load() {
return 0
}
// Since (for now) accounts are unique to subpools, only one pool will have
// (at max) a non-state nonce. To avoid stateful lookups, just return the
// highest nonce for now.
@ -344,6 +413,9 @@ func (p *TxPool) Nonce(addr common.Address) uint64 {
// Stats retrieves the current pool stats, namely the number of pending and the
// number of queued (non-executable) transactions.
func (p *TxPool) Stats() (int, int) {
if !p.inited.Load() {
return 0, 0
}
var runnable, blocked int
for _, subpool := range p.subpools {
run, block := subpool.Stats()
@ -357,6 +429,9 @@ func (p *TxPool) Stats() (int, int) {
// Content retrieves the data content of the transaction pool, returning all the
// pending as well as queued transactions, grouped by account and sorted by nonce.
func (p *TxPool) Content() (map[common.Address][]*types.Transaction, map[common.Address][]*types.Transaction) {
if !p.inited.Load() {
return nil, nil
}
var (
runnable = make(map[common.Address][]*types.Transaction)
blocked = make(map[common.Address][]*types.Transaction)
@ -377,6 +452,9 @@ func (p *TxPool) Content() (map[common.Address][]*types.Transaction, map[common.
// ContentFrom retrieves the data content of the transaction pool, returning the
// pending as well as queued transactions of this address, grouped by nonce.
func (p *TxPool) ContentFrom(addr common.Address) ([]*types.Transaction, []*types.Transaction) {
if !p.inited.Load() {
return nil, nil
}
for _, subpool := range p.subpools {
run, block := subpool.ContentFrom(addr)
if len(run) != 0 || len(block) != 0 {
@ -388,6 +466,9 @@ func (p *TxPool) ContentFrom(addr common.Address) ([]*types.Transaction, []*type
// Locals retrieves the accounts currently considered local by the pool.
func (p *TxPool) Locals() []common.Address {
if !p.inited.Load() {
return nil
}
// Retrieve the locals from each subpool and deduplicate them
locals := make(map[common.Address]struct{})
for _, subpool := range p.subpools {
@ -406,6 +487,9 @@ func (p *TxPool) Locals() []common.Address {
// Status returns the known status (unknown/pending/queued) of a transaction
// identified by their hashes.
func (p *TxPool) Status(hash common.Hash) TxStatus {
if !p.inited.Load() {
return TxStatusUnknown
}
for _, subpool := range p.subpools {
if status := subpool.Status(hash); status != TxStatusUnknown {
return status
@ -413,3 +497,8 @@ func (p *TxPool) Status(hash common.Hash) TxStatus {
}
return TxStatusUnknown
}
// Inited returns the indicator if txpool is fully initialized.
func (p *TxPool) Inited() bool {
return p.inited.Load()
}

10
eth/api_backend.go
View file

@ -204,7 +204,10 @@ func (b *EthAPIBackend) StateAndHeaderByNumber(ctx context.Context, number rpc.B
return nil, nil, errors.New("header not found")
}
stateDb, err := b.eth.BlockChain().StateAt(header.Root)
return stateDb, header, err
if err != nil {
return nil, nil, err
}
return stateDb, header, nil
}
func (b *EthAPIBackend) StateAndHeaderByNumberOrHash(ctx context.Context, blockNrOrHash rpc.BlockNumberOrHash) (*state.StateDB, *types.Header, error) {
@ -223,7 +226,10 @@ func (b *EthAPIBackend) StateAndHeaderByNumberOrHash(ctx context.Context, blockN
return nil, nil, errors.New("hash is not currently canonical")
}
stateDb, err := b.eth.BlockChain().StateAt(header.Root)
return stateDb, header, err
if err != nil {
return nil, nil, err
}
return stateDb, header, nil
}
return nil, nil, errors.New("invalid arguments; neither block nor hash specified")
}

6
eth/backend.go
View file

@ -440,10 +440,6 @@ func (s *Ethereum) StartMining() error {
}
cli.Authorize(eb, wallet.SignData)
}
// If mining is started, we can disable the transaction rejection mechanism
// introduced to speed sync times.
s.handler.enableSyncedFeatures()
go s.miner.Start()
}
return nil
@ -474,7 +470,7 @@ func (s *Ethereum) Engine() consensus.Engine { return s.engine }
func (s *Ethereum) ChainDb() ethdb.Database { return s.chainDb }
func (s *Ethereum) IsListening() bool { return true } // Always listening
func (s *Ethereum) Downloader() *downloader.Downloader { return s.handler.downloader }
func (s *Ethereum) Synced() bool { return s.handler.acceptTxs.Load() }
func (s *Ethereum) Synced() bool { return s.handler.synced.Load() }
func (s *Ethereum) SetSynced() { s.handler.enableSyncedFeatures() }
func (s *Ethereum) ArchiveMode() bool { return s.config.NoPruning }
func (s *Ethereum) BloomIndexer() *core.ChainIndexer { return s.bloomIndexer }

2
eth/downloader/downloader.go
View file

@ -403,7 +403,7 @@ func (d *Downloader) synchronise(id string, hash common.Hash, td, ttd *big.Int,
// subsequent state reads, explicitly disable the trie database and state
// syncer is responsible to address and correct any state missing.
if d.blockchain.TrieDB().Scheme() == rawdb.PathScheme {
d.blockchain.TrieDB().Reset(types.EmptyRootHash)
d.blockchain.TrieDB().Deactivate()
}
// Snap sync uses the snapshot namespace to store potentially flaky data until
// sync completely heals and finishes. Pause snapshot maintenance in the mean-

40
eth/handler.go
View file

@ -101,7 +101,7 @@ type handler struct {
forkFilter forkid.Filter // Fork ID filter, constant across the lifetime of the node
snapSync atomic.Bool // Flag whether snap sync is enabled (gets disabled if we already have blocks)
acceptTxs atomic.Bool // Flag whether we're considered synchronised (enables transaction processing)
synced atomic.Bool // Flag whether we're considered synchronised (enables transaction processing)
database ethdb.Database
txpool txPool
@ -163,10 +163,14 @@ func newHandler(config *handlerConfig) (*handler, error) {
fullBlock, snapBlock := h.chain.CurrentBlock(), h.chain.CurrentSnapBlock()
if fullBlock.Number.Uint64() == 0 && snapBlock.Number.Uint64() > 0 {
h.snapSync.Store(true)
log.Warn("Switch sync mode from full sync to snap sync")
log.Warn("Switch sync mode from full sync to snap sync", "reason", "snap sync incomplete")
} else if !h.chain.HasState(fullBlock.Root) {
h.snapSync.Store(true)
log.Warn("Switch sync mode from full sync to snap sync", "reason", "head state missing")
}
} else {
if h.chain.CurrentBlock().Number.Uint64() > 0 {
head := h.chain.CurrentBlock()
if head.Number.Uint64() > 0 && h.chain.HasState(head.Root) {
// Print warning log if database is not empty to run snap sync.
log.Warn("Switch sync mode from snap sync to full sync")
} else {
@ -174,17 +178,9 @@ func newHandler(config *handlerConfig) (*handler, error) {
h.snapSync.Store(true)
}
}
// If sync succeeds, pass a callback to potentially disable snap sync mode
// and enable transaction propagation.
// If the sync succeeds, mark the local node as synced and enable all features
// with state synchronization requirements.
success := func() {
// If we were running snap sync and it finished, disable doing another
// round on next sync cycle
if h.snapSync.Load() {
log.Info("Snap sync complete, auto disabling")
h.snapSync.Store(false)
}
// If we've successfully finished a sync cycle, accept transactions from
// the network
h.enableSyncedFeatures()
}
// Construct the downloader (long sync)
@ -245,7 +241,7 @@ func newHandler(config *handlerConfig) (*handler, error) {
// accept each others' blocks until a restart. Unfortunately we haven't figured
// out a way yet where nodes can decide unilaterally whether the network is new
// or not. This should be fixed if we figure out a solution.
if h.snapSync.Load() {
if !h.synced.Load() {
log.Warn("Snap syncing, discarded propagated block", "number", blocks[0].Number(), "hash", blocks[0].Hash())
return 0, nil
}
@ -272,11 +268,7 @@ func newHandler(config *handlerConfig) (*handler, error) {
}
return 0, nil
}
n, err := h.chain.InsertChain(blocks)
if err == nil {
h.enableSyncedFeatures() // Mark initial sync done on any fetcher import
}
return n, err
return h.chain.InsertChain(blocks)
}
h.blockFetcher = fetcher.NewBlockFetcher(false, nil, h.chain.GetBlockByHash, validator, h.BroadcastBlock, heighter, nil, inserter, h.removePeer)
@ -680,7 +672,15 @@ func (h *handler) txBroadcastLoop() {
// enableSyncedFeatures enables the post-sync functionalities when the initial
// sync is finished.
func (h *handler) enableSyncedFeatures() {
h.acceptTxs.Store(true)
// Mark the local node as synced.
h.synced.Store(true)
// If we were running snap sync and it finished, disable doing another
// round on next sync cycle
if h.snapSync.Load() {
log.Info("Snap sync complete, auto disabling")
h.snapSync.Store(false)
}
if h.chain.TrieDB().Scheme() == rawdb.PathScheme {
h.chain.TrieDB().SetBufferSize(pathdb.DefaultBufferSize)
}

2
eth/handler_eth.go
View file

@ -51,7 +51,7 @@ func (h *ethHandler) PeerInfo(id enode.ID) interface{} {
// AcceptTxs retrieves whether transaction processing is enabled on the node
// or if inbound transactions should simply be dropped.
func (h *ethHandler) AcceptTxs() bool {
return h.acceptTxs.Load()
return h.synced.Load()
}
// Handle is invoked from a peer's message handler when it receives a new remote

4
eth/handler_eth_test.go
View file

@ -248,7 +248,7 @@ func testRecvTransactions(t *testing.T, protocol uint) {
handler := newTestHandler()
defer handler.close()
handler.handler.acceptTxs.Store(true) // mark synced to accept transactions
handler.handler.synced.Store(true) // mark synced to accept transactions
txs := make(chan core.NewTxsEvent)
sub := handler.txpool.SubscribeNewTxsEvent(txs)
@ -401,7 +401,7 @@ func testTransactionPropagation(t *testing.T, protocol uint) {
sinks[i] = newTestHandler()
defer sinks[i].close()
sinks[i].handler.acceptTxs.Store(true) // mark synced to accept transactions
sinks[i].handler.synced.Store(true) // mark synced to accept transactions
}
// Interconnect all the sink handlers with the source handler
for i, sink := range sinks {

22
eth/sync.go
View file

@ -197,16 +197,24 @@ func (cs *chainSyncer) modeAndLocalHead() (downloader.SyncMode, *big.Int) {
return downloader.SnapSync, td
}
// We are probably in full sync, but we might have rewound to before the
// snap sync pivot, check if we should reenable
// snap sync pivot, check if we should re-enable snap sync.
head := cs.handler.chain.CurrentBlock()
if pivot := rawdb.ReadLastPivotNumber(cs.handler.database); pivot != nil {
if head := cs.handler.chain.CurrentBlock(); head.Number.Uint64() < *pivot {
if head.Number.Uint64() < *pivot {
block := cs.handler.chain.CurrentSnapBlock()
td := cs.handler.chain.GetTd(block.Hash(), block.Number.Uint64())
return downloader.SnapSync, td
}
}
// We are in a full sync, but the associated head state is missing. To complete
// the head state, forcefully rerun the snap sync. Note it doesn't mean the
// persistent state is corrupted, just mismatch with the head block.
if !cs.handler.chain.HasState(head.Root) {
block := cs.handler.chain.CurrentSnapBlock()
td := cs.handler.chain.GetTd(block.Hash(), block.Number.Uint64())
return downloader.SnapSync, td
}
// Nope, we're really full syncing
head := cs.handler.chain.CurrentBlock()
td := cs.handler.chain.GetTd(head.Hash(), head.Number.Uint64())
return downloader.FullSync, td
}
@ -242,13 +250,7 @@ func (h *handler) doSync(op *chainSyncOp) error {
if err != nil {
return err
}
if h.snapSync.Load() {
log.Info("Snap sync complete, auto disabling")
h.snapSync.Store(false)
}
// If we've successfully finished a sync cycle, enable accepting transactions
// from the network.
h.acceptTxs.Store(true)
h.enableSyncedFeatures()
head := h.chain.CurrentBlock()
if head.Number.Uint64() > 0 {

7
miner/miner_test.go
View file

@ -64,6 +64,7 @@ func (m *mockBackend) StateAtBlock(block *types.Block, reexec uint64, base *stat
}
type testBlockChain struct {
root common.Hash
config *params.ChainConfig
statedb *state.StateDB
gasLimit uint64
@ -89,6 +90,10 @@ func (bc *testBlockChain) StateAt(common.Hash) (*state.StateDB, error) {
return bc.statedb, nil
}
func (bc *testBlockChain) HasState(root common.Hash) bool {
return bc.root == root
}
func (bc *testBlockChain) SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) event.Subscription {
return bc.chainHeadFeed.Subscribe(ch)
}
@ -302,7 +307,7 @@ func createMiner(t *testing.T) (*Miner, *event.TypeMux, func(skipMiner bool)) {
t.Fatalf("can't create new chain %v", err)
}
statedb, _ := state.New(bc.Genesis().Root(), bc.StateCache(), nil)
blockchain := &testBlockChain{chainConfig, statedb, 10000000, new(event.Feed)}
blockchain := &testBlockChain{bc.Genesis().Root(), chainConfig, statedb, 10000000, new(event.Feed)}
pool := legacypool.New(testTxPoolConfig, blockchain)
txpool, _ := txpool.New(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain, []txpool.SubPool{pool})

20
trie/database.go
View file

@ -273,15 +273,27 @@ func (db *Database) Recoverable(root common.Hash) (bool, error) {
return pdb.Recoverable(root), nil
}
// Reset wipes all available journal from the persistent database and discard
// all caches and diff layers. Using the given root to create a new disk layer.
// Deactivate disables the database and invalidates all available state layers
// as stale to prevent access to the persistent state, which is in the syncing
// stage.
//
// It's only supported by path-based database and will return an error for others.
func (db *Database) Reset(root common.Hash) error {
func (db *Database) Deactivate() error {
pdb, ok := db.backend.(*pathdb.Database)
if !ok {
return errors.New("not supported")
}
return pdb.Reset(root)
return pdb.Deactivate()
}
// Activate re-enables database and resets the state tree with the provided.
// persistent state root once the state sync is finished.
func (db *Database) Activate(root common.Hash) error {
pdb, ok := db.backend.(*pathdb.Database)
if !ok {
return errors.New("not supported")
}
return pdb.Activate(root)
}
// Journal commits an entire diff hierarchy to disk into a single journal entry.

98
trie/triedb/pathdb/database.go
View file

@ -128,7 +128,8 @@ type Database struct {
// readOnly is the flag whether the mutation is allowed to be applied.
// It will be set automatically when the database is journaled during
// the shutdown to reject all following unexpected mutations.
readOnly bool // Indicator if database is opened in read only mode
readOnly bool // Flag if database is opened in read only mode
disabled bool // Flag if database is deactivated due to initial state sync
bufferSize int // Memory allowance (in bytes) for caching dirty nodes
config *Config // Configuration for database
diskdb ethdb.Database // Persistent storage for matured trie nodes
@ -179,6 +180,10 @@ func New(diskdb ethdb.Database, config *Config) *Database {
log.Warn("Truncated extra state histories", "number", pruned)
}
}
// Disable database in case node is still in the initial state sync stage.
if rawdb.ReadSnapSyncStatusFlag(diskdb) == rawdb.StateSyncing && !db.readOnly {
db.Deactivate()
}
log.Warn("Path-based state scheme is an experimental feature")
return db
}
@ -204,9 +209,9 @@ func (db *Database) Update(root common.Hash, parentRoot common.Hash, block uint6
db.lock.Lock()
defer db.lock.Unlock()
// Short circuit if the database is in read only mode.
if db.readOnly {
return errSnapshotReadOnly
// Short circuit if the mutation is not allowed.
if err := db.modifyAllowed(); err != nil {
return err
}
if err := db.tree.add(root, parentRoot, block, nodes, states); err != nil {
return err
@ -227,45 +232,58 @@ func (db *Database) Commit(root common.Hash, report bool) error {
db.lock.Lock()
defer db.lock.Unlock()
// Short circuit if the database is in read only mode.
if db.readOnly {
return errSnapshotReadOnly
// Short circuit if the mutation is not allowed.
if err := db.modifyAllowed(); err != nil {
return err
}
return db.tree.cap(root, 0)
}
// Reset rebuilds the database with the specified state as the base.
//
// - if target state is empty, clear the stored state and all layers on top
// - if target state is non-empty, ensure the stored state matches with it
// and clear all other layers on top.
func (db *Database) Reset(root common.Hash) error {
// Deactivate disables the database and invalidates all available state layers
// as stale to prevent access to the persistent state, which is in the syncing
// stage.
func (db *Database) Deactivate() error {
db.lock.Lock()
defer db.lock.Unlock()
// Short circuit if the database is in read only mode.
if db.readOnly {
return errSnapshotReadOnly
return errDatabaseReadOnly
}
batch := db.diskdb.NewBatch()
root = types.TrieRootHash(root)
if root == types.EmptyRootHash {
// Empty state is requested as the target, nuke out
// the root node and leave all others as dangling.
rawdb.DeleteAccountTrieNode(batch, nil)
} else {
// Ensure the requested state is existent before any
// action is applied.
_, hash := rawdb.ReadAccountTrieNode(db.diskdb, nil)
if hash != root {
return fmt.Errorf("state is mismatched, local: %x, target: %x", hash, root)
// Prevent duplicated disable operation.
if db.disabled {
return nil
}
}
// Mark the disk layer as stale before applying any mutation.
db.disabled = true
// Mark the disk layer as stale to prevent access to persistent state.
db.tree.bottom().markStale()
// Write the initial sync flag to persist it across restarts.
rawdb.WriteSnapSyncStatusFlag(db.diskdb, rawdb.StateSyncing)
log.Info("Disabled trie database")
return nil
}
// Activate re-enables database and resets the state tree with the provided.
// persistent state root once the state sync is finished.
func (db *Database) Activate(root common.Hash) error {
db.lock.Lock()
defer db.lock.Unlock()
// Short circuit if the database is in read only mode.
if db.readOnly {
return errDatabaseReadOnly
}
// Ensure the provided state root matches the stored one.
root = types.TrieRootHash(root)
_, stored := rawdb.ReadAccountTrieNode(db.diskdb, nil)
if stored != root {
return fmt.Errorf("state is mismatched, stored: %x, target: %x", stored, root)
}
// Drop the stale state journal in persistent database and
// reset the persistent state id back to zero.
batch := db.diskdb.NewBatch()
rawdb.DeleteTrieJournal(batch)
rawdb.WritePersistentStateID(batch, 0)
if err := batch.Write(); err != nil {
@ -282,8 +300,11 @@ func (db *Database) Reset(root common.Hash) error {
}
// Re-construct a new disk layer backed by persistent state
// with **empty clean cache and node buffer**.
dl := newDiskLayer(root, 0, db, nil, newNodeBuffer(db.bufferSize, nil, 0))
db.tree.reset(dl)
db.tree.reset(newDiskLayer(root, 0, db, nil, newNodeBuffer(db.bufferSize, nil, 0)))
// Re-enable the database as the final step.
db.disabled = false
rawdb.WriteSnapSyncStatusFlag(db.diskdb, rawdb.StateSynced)
log.Info("Rebuilt trie database", "root", root)
return nil
}
@ -296,7 +317,10 @@ func (db *Database) Recover(root common.Hash, loader triestate.TrieLoader) error
defer db.lock.Unlock()
// Short circuit if rollback operation is not supported.
if db.readOnly || db.freezer == nil {
if err := db.modifyAllowed(); err != nil {
return err
}
if db.freezer == nil {
return errors.New("state rollback is non-supported")
}
// Short circuit if the target state is not recoverable.
@ -424,3 +448,15 @@ func (db *Database) SetBufferSize(size int) error {
func (db *Database) Scheme() string {
return rawdb.PathScheme
}
// modifyAllowed returns the indicator if mutation is allowed. This function
// assumes the db.lock is already held.
func (db *Database) modifyAllowed() error {
if db.readOnly {
return errDatabaseReadOnly
}
if db.disabled {
return errDatabaseDisabled
}
return nil
}

35
trie/triedb/pathdb/database_test.go
View file

@ -439,38 +439,39 @@ func TestDatabaseRecoverable(t *testing.T) {
}
}
func TestReset(t *testing.T) {
var (
tester = newTester(t)
index = tester.bottomIndex()
)
func TestDisable(t *testing.T) {
tester := newTester(t)
defer tester.release()
// Reset database to unknown target, should reject it
if err := tester.db.Reset(testutil.RandomHash()); err == nil {
t.Fatal("Failed to reject invalid reset")
_, stored := rawdb.ReadAccountTrieNode(tester.db.diskdb, nil)
if err := tester.db.Deactivate(); err != nil {
t.Fatal("Failed to deactivate database")
}
// Reset database to state persisted in the disk
if err := tester.db.Reset(types.EmptyRootHash); err != nil {
t.Fatalf("Failed to reset database %v", err)
if err := tester.db.Activate(types.EmptyRootHash); err == nil {
t.Fatalf("Invalid activation should be rejected")
}
if err := tester.db.Activate(stored); err != nil {
t.Fatal("Failed to activate database")
}
// Ensure journal is deleted from disk
if blob := rawdb.ReadTrieJournal(tester.db.diskdb); len(blob) != 0 {
t.Fatal("Failed to clean journal")
}
// Ensure all trie histories are removed
for i := 0; i <= index; i++ {
_, err := readHistory(tester.db.freezer, uint64(i+1))
if err == nil {
t.Fatalf("Failed to clean state history, index %d", i+1)
n, err := tester.db.freezer.Ancients()
if err != nil {
t.Fatal("Failed to clean state history")
}
if n != 0 {
t.Fatal("Failed to clean state history")
}
// Verify layer tree structure, single disk layer is expected
if tester.db.tree.len() != 1 {
t.Fatalf("Extra layer kept %d", tester.db.tree.len())
}
if tester.db.tree.bottom().rootHash() != types.EmptyRootHash {
t.Fatalf("Root hash is not matched exp %x got %x", types.EmptyRootHash, tester.db.tree.bottom().rootHash())
if tester.db.tree.bottom().rootHash() != stored {
t.Fatalf("Root hash is not matched exp %x got %x", stored, tester.db.tree.bottom().rootHash())
}
}

10
trie/triedb/pathdb/errors.go
View file

@ -25,9 +25,13 @@ import (
)
var (
// errSnapshotReadOnly is returned if the database is opened in read only mode
// and mutation is requested.
errSnapshotReadOnly = errors.New("read only")
// errDatabaseReadOnly is returned if the database is opened in read only mode
// to prevent any mutation.
errDatabaseReadOnly = errors.New("read only")
// errDatabaseDisabled is returned if database is disabled due to an ongoing
// state sync process.
errDatabaseDisabled = errors.New("disabled")
// errSnapshotStale is returned from data accessors if the underlying layer
// layer had been invalidated due to the chain progressing forward far enough

2
trie/triedb/pathdb/journal.go
View file

@ -356,7 +356,7 @@ func (db *Database) Journal(root common.Hash) error {
// Short circuit if the database is in read only mode.
if db.readOnly {
return errSnapshotReadOnly
return errDatabaseReadOnly
}
// Firstly write out the metadata of journal
journal := new(bytes.Buffer)