From d02618b55d4b33244e9f9506e6fcf7e07ad5ad9b Mon Sep 17 00:00:00 2001 From: Gary Rong Date: Wed, 20 Sep 2023 16:42:13 +0800 Subject: [PATCH] core, accounts, eth, trie: handle genesis state missing --- accounts/abi/bind/backends/simulated.go | 17 ++-- core/blockchain.go | 59 +++++-------- core/rawdb/accessors_sync.go | 21 +++++ core/rawdb/database.go | 2 +- core/rawdb/schema.go | 3 + core/txpool/txpool.go | 113 +++++++++++++++++++++--- eth/api_backend.go | 10 ++- eth/backend.go | 6 +- eth/downloader/downloader.go | 2 +- eth/handler.go | 42 ++++----- eth/handler_eth.go | 2 +- eth/handler_eth_test.go | 4 +- eth/sync.go | 22 ++--- miner/miner_test.go | 7 +- trie/database.go | 20 ++++- trie/triedb/pathdb/database.go | 98 +++++++++++++------- trie/triedb/pathdb/database_test.go | 37 ++++---- trie/triedb/pathdb/errors.go | 10 ++- trie/triedb/pathdb/journal.go | 2 +- 19 files changed, 317 insertions(+), 160 deletions(-) diff --git a/accounts/abi/bind/backends/simulated.go b/accounts/abi/bind/backends/simulated.go index 0c4342c494..dbdcd17823 100644 --- a/accounts/abi/bind/backends/simulated.go +++ b/accounts/abi/bind/backends/simulated.go @@ -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 } diff --git a/core/blockchain.go b/core/blockchain.go index e371e8d926..90ee3a5fe3 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -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() + // 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) { - // 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! 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()) } - 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 } } diff --git a/core/rawdb/accessors_sync.go b/core/rawdb/accessors_sync.go index 7a7374e168..0366f930c5 100644 --- a/core/rawdb/accessors_sync.go +++ b/core/rawdb/accessors_sync.go @@ -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) + } +} diff --git a/core/rawdb/database.go b/core/rawdb/database.go index 3839e949ed..123021cd6b 100644 --- a/core/rawdb/database.go +++ b/core/rawdb/database.go @@ -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) diff --git a/core/rawdb/schema.go b/core/rawdb/schema.go index 7269fe5d56..02d47e7629 100644 --- a/core/rawdb/schema.go +++ b/core/rawdb/schema.go @@ -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 diff --git a/core/txpool/txpool.go b/core/txpool/txpool.go index e40b414054..ae94bfa70e 100644 --- a/core/txpool/txpool.go +++ b/core/txpool/txpool.go @@ -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,7 +69,8 @@ type BlockChain interface { // They exit the pool when they are included in the blockchain or evicted due to // resource constraints. type TxPool struct { - subpools []SubPool // List of subpools for specialized transaction handling + 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 reserveLock sync.Mutex // Lock protecting the account reservations @@ -87,16 +92,55 @@ 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 { + 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-- { - subpools[j].Close() + p.subpools[j].Close() } - return nil, err + // TODO(rjl493456442) can we shutdown the node gracefully? + log.Crit("Failed to initialize subpool", "err", err) } } - go pool.loop(head, chain) - return pool, nil + 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 @@ -156,14 +200,16 @@ func (p *TxPool) Close() error { errs = append(errs, err) } - // Terminate each subpool - for _, subpool := range p.subpools { - if err := subpool.Close(); err != nil { - errs = append(errs, err) + // 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() +} diff --git a/eth/api_backend.go b/eth/api_backend.go index dea745382e..a0c14f1338 100644 --- a/eth/api_backend.go +++ b/eth/api_backend.go @@ -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") } diff --git a/eth/backend.go b/eth/backend.go index b99ae7655b..e481e854e7 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -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 } diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index 732e79f8ba..180988f521 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -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- diff --git a/eth/handler.go b/eth/handler.go index a629ec5ee9..6254b3a01a 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -100,8 +100,8 @@ type handler struct { networkID uint64 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) + snapSync atomic.Bool // Flag whether snap sync is enabled (gets disabled if we already have blocks) + 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) } diff --git a/eth/handler_eth.go b/eth/handler_eth.go index 3a5e6608bb..2aba16f928 100644 --- a/eth/handler_eth.go +++ b/eth/handler_eth.go @@ -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 diff --git a/eth/handler_eth_test.go b/eth/handler_eth_test.go index 41619fe300..a16abc5ed6 100644 --- a/eth/handler_eth_test.go +++ b/eth/handler_eth_test.go @@ -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 { diff --git a/eth/sync.go b/eth/sync.go index ba7a7427a5..03357254e7 100644 --- a/eth/sync.go +++ b/eth/sync.go @@ -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 { diff --git a/miner/miner_test.go b/miner/miner_test.go index 489bc46a91..36d5166c6d 100644 --- a/miner/miner_test.go +++ b/miner/miner_test.go @@ -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}) diff --git a/trie/database.go b/trie/database.go index 535ad87d72..f37915aa6d 100644 --- a/trie/database.go +++ b/trie/database.go @@ -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. diff --git a/trie/triedb/pathdb/database.go b/trie/triedb/pathdb/database.go index 18cc36ffc3..99259538f2 100644 --- a/trie/triedb/pathdb/database.go +++ b/trie/triedb/pathdb/database.go @@ -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 +} diff --git a/trie/triedb/pathdb/database_test.go b/trie/triedb/pathdb/database_test.go index 6d346d20ea..b5b2873c65 100644 --- a/trie/triedb/pathdb/database_test.go +++ b/trie/triedb/pathdb/database_test.go @@ -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()) } } diff --git a/trie/triedb/pathdb/errors.go b/trie/triedb/pathdb/errors.go index f6ac0ec4a0..20dbd6689b 100644 --- a/trie/triedb/pathdb/errors.go +++ b/trie/triedb/pathdb/errors.go @@ -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 diff --git a/trie/triedb/pathdb/journal.go b/trie/triedb/pathdb/journal.go index ea90207f29..ac770763e3 100644 --- a/trie/triedb/pathdb/journal.go +++ b/trie/triedb/pathdb/journal.go @@ -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)