// Copyright 2016 The go-ethereum Authors // This file is part of the go-ethereum library. // // The go-ethereum library is free software: you can redistribute it and/or modify // it under the terms of the GNU Lesser General Public License as published by // the Free Software Foundation, either version 3 of the License, or // (at your option) any later version. // // The go-ethereum library is distributed in the hope that it will be useful, // but WITHOUT ANY WARRANTY; without even the implied warranty of // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the // GNU Lesser General Public License for more details. // // You should have received a copy of the GNU Lesser General Public License // along with the go-ethereum library. If not, see . // Package light implements on-demand retrieval capable state and chain objects // for the Ethereum Light Client. package light import ( "context" "errors" "math/big" "sync" "sync/atomic" "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/consensus" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/core/state" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/params" "github.com/ethereum/go-ethereum/rlp" lru "github.com/hashicorp/golang-lru" ) var ( bodyCacheLimit = 256 blockCacheLimit = 256 ) // LightChain represents a canonical chain that by default only handles block // headers, downloading block bodies and receipts on demand through an ODR // interface. It only does header validation during chain insertion. type LightChain struct { hc *core.HeaderChain indexerConfig *IndexerConfig chainDb ethdb.Database engine consensus.Engine odr OdrBackend chainFeed event.Feed chainSideFeed event.Feed chainHeadFeed event.Feed scope event.SubscriptionScope genesisBlock *types.Block bodyCache *lru.Cache // Cache for the most recent block bodies bodyRLPCache *lru.Cache // Cache for the most recent block bodies in RLP encoded format blockCache *lru.Cache // Cache for the most recent entire blocks chainmu sync.RWMutex // protects header inserts quit chan struct{} wg sync.WaitGroup // Atomic boolean switches: running int32 // whether LightChain is running or stopped procInterrupt int32 // interrupts chain insert disableCheckFreq int32 // disables header verification } // NewLightChain returns a fully initialised light chain using information // available in the database. It initialises the default Ethereum header // validator. func NewLightChain(odr OdrBackend, config *params.ChainConfig, engine consensus.Engine, checkpoint *params.TrustedCheckpoint) (*LightChain, error) { bodyCache, _ := lru.New(bodyCacheLimit) bodyRLPCache, _ := lru.New(bodyCacheLimit) blockCache, _ := lru.New(blockCacheLimit) bc := &LightChain{ chainDb: odr.Database(), indexerConfig: odr.IndexerConfig(), odr: odr, quit: make(chan struct{}), bodyCache: bodyCache, bodyRLPCache: bodyRLPCache, blockCache: blockCache, engine: engine, } var err error bc.hc, err = core.NewHeaderChain(odr.Database(), config, bc.engine, bc.getProcInterrupt) if err != nil { return nil, err } bc.genesisBlock, _ = bc.GetBlockByNumber(NoOdr, 0) if bc.genesisBlock == nil { return nil, core.ErrNoGenesis } if checkpoint != nil { bc.AddTrustedCheckpoint(checkpoint) } if err := bc.loadLastState(); err != nil { return nil, err } // Check the current state of the block hashes and make sure that we do not have any of the bad blocks in our chain for hash := range core.BadHashes { if header := bc.GetHeaderByHash(hash); header != nil { log.Error("Found bad hash, rewinding chain", "number", header.Number, "hash", header.ParentHash) bc.SetHead(header.Number.Uint64() - 1) log.Error("Chain rewind was successful, resuming normal operation") } } return bc, nil } // AddTrustedCheckpoint adds a trusted checkpoint to the blockchain func (lc *LightChain) AddTrustedCheckpoint(cp *params.TrustedCheckpoint) { if lc.odr.ChtIndexer() != nil { StoreChtRoot(lc.chainDb, cp.SectionIndex, cp.SectionHead, cp.CHTRoot) lc.odr.ChtIndexer().AddCheckpoint(cp.SectionIndex, cp.SectionHead) } if lc.odr.BloomTrieIndexer() != nil { StoreBloomTrieRoot(lc.chainDb, cp.SectionIndex, cp.SectionHead, cp.BloomRoot) lc.odr.BloomTrieIndexer().AddCheckpoint(cp.SectionIndex, cp.SectionHead) } if lc.odr.BloomIndexer() != nil { lc.odr.BloomIndexer().AddCheckpoint(cp.SectionIndex, cp.SectionHead) } log.Info("Added trusted checkpoint", "block", (cp.SectionIndex+1)*lc.indexerConfig.ChtSize-1, "hash", cp.SectionHead) } func (lc *LightChain) getProcInterrupt() bool { return atomic.LoadInt32(&lc.procInterrupt) == 1 } // Odr returns the ODR backend of the chain func (lc *LightChain) Odr() OdrBackend { return lc.odr } // loadLastState loads the last known chain state from the database. This method // assumes that the chain manager mutex is held. func (lc *LightChain) loadLastState() error { if head := rawdb.ReadHeadHeaderHash(lc.chainDb); head == (common.Hash{}) { // Corrupt or empty database, init from scratch lc.Reset() } else { if header := lc.GetHeaderByHash(head); header != nil { lc.hc.SetCurrentHeader(header) } } // Issue a status log and return header := lc.hc.CurrentHeader() headerTd := lc.GetTd(header.Hash(), header.Number.Uint64()) log.Info("Loaded most recent local header", "number", header.Number, "hash", header.Hash(), "td", headerTd, "age", common.PrettyAge(time.Unix(int64(header.Time), 0))) return nil } // SetHead rewinds the local chain to a new head. Everything above the new // head will be deleted and the new one set. func (lc *LightChain) SetHead(head uint64) error { lc.chainmu.Lock() defer lc.chainmu.Unlock() lc.hc.SetHead(head, nil, nil) return lc.loadLastState() } // GasLimit returns the gas limit of the current HEAD block. func (lc *LightChain) GasLimit() uint64 { return lc.hc.CurrentHeader().GasLimit } // Reset purges the entire blockchain, restoring it to its genesis state. func (lc *LightChain) Reset() { lc.ResetWithGenesisBlock(lc.genesisBlock) } // ResetWithGenesisBlock purges the entire blockchain, restoring it to the // specified genesis state. func (lc *LightChain) ResetWithGenesisBlock(genesis *types.Block) { // Dump the entire block chain and purge the caches lc.SetHead(0) lc.chainmu.Lock() defer lc.chainmu.Unlock() // Prepare the genesis block and reinitialise the chain rawdb.WriteTd(lc.chainDb, genesis.Hash(), genesis.NumberU64(), genesis.Difficulty()) rawdb.WriteBlock(lc.chainDb, genesis) lc.genesisBlock = genesis lc.hc.SetGenesis(lc.genesisBlock.Header()) lc.hc.SetCurrentHeader(lc.genesisBlock.Header()) } // Accessors // Engine retrieves the light chain's consensus engine. func (lc *LightChain) Engine() consensus.Engine { return lc.engine } // Genesis returns the genesis block func (lc *LightChain) Genesis() *types.Block { return lc.genesisBlock } func (lc *LightChain) StateCache() state.Database { panic("not implemented") } // GetBody retrieves a block body (transactions and uncles) from the database // or ODR service by hash, caching it if found. func (lc *LightChain) GetBody(ctx context.Context, hash common.Hash) (*types.Body, error) { // Short circuit if the body's already in the cache, retrieve otherwise if cached, ok := lc.bodyCache.Get(hash); ok { body := cached.(*types.Body) return body, nil } number := lc.hc.GetBlockNumber(hash) if number == nil { return nil, errors.New("unknown block") } body, err := GetBody(ctx, lc.odr, hash, *number) if err != nil { return nil, err } // Cache the found body for next time and return lc.bodyCache.Add(hash, body) return body, nil } // GetBodyRLP retrieves a block body in RLP encoding from the database or // ODR service by hash, caching it if found. func (lc *LightChain) GetBodyRLP(ctx context.Context, hash common.Hash) (rlp.RawValue, error) { // Short circuit if the body's already in the cache, retrieve otherwise if cached, ok := lc.bodyRLPCache.Get(hash); ok { return cached.(rlp.RawValue), nil } number := lc.hc.GetBlockNumber(hash) if number == nil { return nil, errors.New("unknown block") } body, err := GetBodyRLP(ctx, lc.odr, hash, *number) if err != nil { return nil, err } // Cache the found body for next time and return lc.bodyRLPCache.Add(hash, body) return body, nil } // HasBlock checks if a block is fully present in the database or not, caching // it if present. func (lc *LightChain) HasBlock(hash common.Hash, number uint64) bool { blk, _ := lc.GetBlock(NoOdr, hash, number) return blk != nil } // GetBlock retrieves a block from the database or ODR service by hash and number, // caching it if found. func (lc *LightChain) GetBlock(ctx context.Context, hash common.Hash, number uint64) (*types.Block, error) { // Short circuit if the block's already in the cache, retrieve otherwise if block, ok := lc.blockCache.Get(hash); ok { return block.(*types.Block), nil } block, err := GetBlock(ctx, lc.odr, hash, number) if err != nil { return nil, err } // Cache the found block for next time and return lc.blockCache.Add(block.Hash(), block) return block, nil } // GetBlockByHash retrieves a block from the database or ODR service by hash, // caching it if found. func (lc *LightChain) GetBlockByHash(ctx context.Context, hash common.Hash) (*types.Block, error) { number := lc.hc.GetBlockNumber(hash) if number == nil { return nil, errors.New("unknown block") } return lc.GetBlock(ctx, hash, *number) } // GetBlockByNumber retrieves a block from the database or ODR service by // number, caching it (associated with its hash) if found. func (lc *LightChain) GetBlockByNumber(ctx context.Context, number uint64) (*types.Block, error) { hash, err := GetCanonicalHash(ctx, lc.odr, number) if hash == (common.Hash{}) || err != nil { return nil, err } return lc.GetBlock(ctx, hash, number) } // Stop stops the blockchain service. If any imports are currently in progress // it will abort them using the procInterrupt. func (lc *LightChain) Stop() { if !atomic.CompareAndSwapInt32(&lc.running, 0, 1) { return } close(lc.quit) atomic.StoreInt32(&lc.procInterrupt, 1) lc.wg.Wait() log.Info("Blockchain manager stopped") } // Rollback is designed to remove a chain of links from the database that aren't // certain enough to be valid. func (lc *LightChain) Rollback(chain []common.Hash) { lc.chainmu.Lock() defer lc.chainmu.Unlock() for i := len(chain) - 1; i >= 0; i-- { hash := chain[i] if head := lc.hc.CurrentHeader(); head.Hash() == hash { lc.hc.SetCurrentHeader(lc.GetHeader(head.ParentHash, head.Number.Uint64()-1)) } } } // postChainEvents iterates over the events generated by a chain insertion and // posts them into the event feed. func (lc *LightChain) postChainEvents(events []interface{}) { for _, event := range events { switch ev := event.(type) { case core.ChainEvent: if lc.CurrentHeader().Hash() == ev.Hash { lc.chainHeadFeed.Send(core.ChainHeadEvent{Block: ev.Block}) } lc.chainFeed.Send(ev) case core.ChainSideEvent: lc.chainSideFeed.Send(ev) } } } // InsertHeaderChain attempts to insert the given header chain in to the local // chain, possibly creating a reorg. If an error is returned, it will return the // index number of the failing header as well an error describing what went wrong. // // The verify parameter can be used to fine tune whether nonce verification // should be done or not. The reason behind the optional check is because some // of the header retrieval mechanisms already need to verfy nonces, as well as // because nonces can be verified sparsely, not needing to check each. // // In the case of a light chain, InsertHeaderChain also creates and posts light // chain events when necessary. func (lc *LightChain) InsertHeaderChain(chain []*types.Header, checkFreq int) (int, error) { if atomic.LoadInt32(&lc.disableCheckFreq) == 1 { checkFreq = 0 } start := time.Now() if i, err := lc.hc.ValidateHeaderChain(chain, checkFreq); err != nil { return i, err } // Make sure only one thread manipulates the chain at once lc.chainmu.Lock() defer lc.chainmu.Unlock() lc.wg.Add(1) defer lc.wg.Done() var events []interface{} whFunc := func(header *types.Header) error { status, err := lc.hc.WriteHeader(header) switch status { case core.CanonStatTy: log.Debug("Inserted new header", "number", header.Number, "hash", header.Hash()) events = append(events, core.ChainEvent{Block: types.NewBlockWithHeader(header), Hash: header.Hash()}) case core.SideStatTy: log.Debug("Inserted forked header", "number", header.Number, "hash", header.Hash()) events = append(events, core.ChainSideEvent{Block: types.NewBlockWithHeader(header)}) } return err } i, err := lc.hc.InsertHeaderChain(chain, whFunc, start) lc.postChainEvents(events) return i, err } // CurrentHeader retrieves the current head header of the canonical chain. The // header is retrieved from the HeaderChain's internal cache. func (lc *LightChain) CurrentHeader() *types.Header { return lc.hc.CurrentHeader() } // GetTd retrieves a block's total difficulty in the canonical chain from the // database by hash and number, caching it if found. func (lc *LightChain) GetTd(hash common.Hash, number uint64) *big.Int { return lc.hc.GetTd(hash, number) } // GetTdByHash retrieves a block's total difficulty in the canonical chain from the // database by hash, caching it if found. func (lc *LightChain) GetTdByHash(hash common.Hash) *big.Int { return lc.hc.GetTdByHash(hash) } // GetHeader retrieves a block header from the database by hash and number, // caching it if found. func (lc *LightChain) GetHeader(hash common.Hash, number uint64) *types.Header { return lc.hc.GetHeader(hash, number) } // GetHeaderByHash retrieves a block header from the database by hash, caching it if // found. func (lc *LightChain) GetHeaderByHash(hash common.Hash) *types.Header { return lc.hc.GetHeaderByHash(hash) } // HasHeader checks if a block header is present in the database or not, caching // it if present. func (lc *LightChain) HasHeader(hash common.Hash, number uint64) bool { return lc.hc.HasHeader(hash, number) } // GetBlockHashesFromHash retrieves a number of block hashes starting at a given // hash, fetching towards the genesis block. func (lc *LightChain) GetBlockHashesFromHash(hash common.Hash, max uint64) []common.Hash { return lc.hc.GetBlockHashesFromHash(hash, max) } // GetAncestor retrieves the Nth ancestor of a given block. It assumes that either the given block or // a close ancestor of it is canonical. maxNonCanonical points to a downwards counter limiting the // number of blocks to be individually checked before we reach the canonical chain. // // Note: ancestor == 0 returns the same block, 1 returns its parent and so on. func (lc *LightChain) GetAncestor(hash common.Hash, number, ancestor uint64, maxNonCanonical *uint64) (common.Hash, uint64) { return lc.hc.GetAncestor(hash, number, ancestor, maxNonCanonical) } // GetHeaderByNumber retrieves a block header from the database by number, // caching it (associated with its hash) if found. func (lc *LightChain) GetHeaderByNumber(number uint64) *types.Header { return lc.hc.GetHeaderByNumber(number) } // GetHeaderByNumberOdr retrieves a block header from the database or network // by number, caching it (associated with its hash) if found. func (lc *LightChain) GetHeaderByNumberOdr(ctx context.Context, number uint64) (*types.Header, error) { if header := lc.hc.GetHeaderByNumber(number); header != nil { return header, nil } return GetHeaderByNumber(ctx, lc.odr, number) } // Config retrieves the header chain's chain configuration. func (lc *LightChain) Config() *params.ChainConfig { return lc.hc.Config() } // SyncCheckpoint fetches the checkpoint point block header according to // the checkpoint provided by the remote peer. // // Note if we are running the clique, fetches the last epoch snapshot header // which covered by checkpoint. func (lc *LightChain) SyncCheckpoint(ctx context.Context, checkpoint *params.TrustedCheckpoint) bool { // Ensure the remote checkpoint head is ahead of us head := lc.CurrentHeader().Number.Uint64() latest := (checkpoint.SectionIndex+1)*lc.indexerConfig.ChtSize - 1 if clique := lc.hc.Config().Clique; clique != nil { latest -= latest % clique.Epoch // epoch snapshot for clique } if head >= latest { return true } // Retrieve the latest useful header and update to it if header, err := GetHeaderByNumber(ctx, lc.odr, latest); header != nil && err == nil { lc.chainmu.Lock() defer lc.chainmu.Unlock() // Ensure the chain didn't move past the latest block while retrieving it if lc.hc.CurrentHeader().Number.Uint64() < header.Number.Uint64() { log.Info("Updated latest header based on CHT", "number", header.Number, "hash", header.Hash(), "age", common.PrettyAge(time.Unix(int64(header.Time), 0))) lc.hc.SetCurrentHeader(header) } return true } return false } // LockChain locks the chain mutex for reading so that multiple canonical hashes can be // retrieved while it is guaranteed that they belong to the same version of the chain func (lc *LightChain) LockChain() { lc.chainmu.RLock() } // UnlockChain unlocks the chain mutex func (lc *LightChain) UnlockChain() { lc.chainmu.RUnlock() } // SubscribeChainEvent registers a subscription of ChainEvent. func (lc *LightChain) SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription { return lc.scope.Track(lc.chainFeed.Subscribe(ch)) } // SubscribeChainHeadEvent registers a subscription of ChainHeadEvent. func (lc *LightChain) SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) event.Subscription { return lc.scope.Track(lc.chainHeadFeed.Subscribe(ch)) } // SubscribeChainSideEvent registers a subscription of ChainSideEvent. func (lc *LightChain) SubscribeChainSideEvent(ch chan<- core.ChainSideEvent) event.Subscription { return lc.scope.Track(lc.chainSideFeed.Subscribe(ch)) } // SubscribeLogsEvent implements the interface of filters.Backend // LightChain does not send logs events, so return an empty subscription. func (lc *LightChain) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription { return lc.scope.Track(new(event.Feed).Subscribe(ch)) } // SubscribeRemovedLogsEvent implements the interface of filters.Backend // LightChain does not send core.RemovedLogsEvent, so return an empty subscription. func (lc *LightChain) SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription { return lc.scope.Track(new(event.Feed).Subscribe(ch)) } // DisableCheckFreq disables header validation. This is used for ultralight mode. func (lc *LightChain) DisableCheckFreq() { atomic.StoreInt32(&lc.disableCheckFreq, 1) } // EnableCheckFreq enables header validation. func (lc *LightChain) EnableCheckFreq() { atomic.StoreInt32(&lc.disableCheckFreq, 0) }