123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411 |
- package eth
- import (
- "errors"
- "fmt"
- "sync/atomic"
- "time"
- "github.com/ethereum/go-ethereum/common"
- "github.com/ethereum/go-ethereum/core"
- "github.com/ethereum/go-ethereum/core/forkid"
- "github.com/ethereum/go-ethereum/core/types"
- "github.com/ethereum/go-ethereum/eth/downloader"
- "github.com/ethereum/go-ethereum/eth/fetcher"
- "github.com/ethereum/go-ethereum/eth/protocols/eth"
- qlightproto "github.com/ethereum/go-ethereum/eth/protocols/qlight"
- "github.com/ethereum/go-ethereum/event"
- "github.com/ethereum/go-ethereum/log"
- "github.com/ethereum/go-ethereum/p2p"
- "github.com/ethereum/go-ethereum/p2p/enode"
- "github.com/ethereum/go-ethereum/params"
- "github.com/ethereum/go-ethereum/trie"
- )
- type qlightClientHandler ethHandler
- func (h *qlightClientHandler) Chain() *core.BlockChain { return h.chain }
- func (h *qlightClientHandler) StateBloom() *trie.SyncBloom { return h.stateBloom }
- func (h *qlightClientHandler) TxPool() eth.TxPool { return h.txpool }
- func (h *qlightClientHandler) RunPeer(peer *eth.Peer, handler eth.Handler) error {
- return nil
- }
- func (h *qlightClientHandler) Handle(peer *eth.Peer, packet eth.Packet) error {
- return (*ethHandler)(h).Handle(peer, packet)
- }
- func (h *qlightClientHandler) RunQPeer(peer *qlightproto.Peer, hand qlightproto.Handler) error {
- return (*handler)(h).runQLightClientPeer(peer, hand)
- }
- // PeerInfo retrieves all known `eth` information about a peer.
- func (h *qlightClientHandler) PeerInfo(id enode.ID) interface{} {
- if p := h.peers.peer(id.String()); p != nil {
- return p.info()
- }
- return nil
- }
- // AcceptTxs retrieves whether transaction processing is enabled on the node
- // or if inbound transactions should simply be dropped.
- func (h *qlightClientHandler) AcceptTxs() bool {
- return atomic.LoadUint32(&h.acceptTxs) == 1
- }
- // newHandler returns a handler for all Ethereum chain management protocol.
- func newQLightClientHandler(config *handlerConfig) (*handler, error) {
- // Create the protocol manager with the base fields
- if config.EventMux == nil {
- config.EventMux = new(event.TypeMux) // Nicety initialization for tests
- }
- h := &handler{
- networkID: config.Network,
- forkFilter: forkid.NewFilter(config.Chain),
- eventMux: config.EventMux,
- database: config.Database,
- txpool: config.TxPool,
- chain: config.Chain,
- peers: newPeerSet(),
- authorizationList: config.AuthorizationList,
- txsyncCh: make(chan *txsync),
- quitSync: make(chan struct{}),
- raftMode: config.RaftMode,
- engine: config.Engine,
- psi: config.psi,
- privateClientCache: config.privateClientCache,
- tokenHolder: config.tokenHolder,
- }
- if config.Sync == downloader.FullSync {
- // The database seems empty as the current block is the genesis. Yet the fast
- // block is ahead, so fast sync was enabled for this node at a certain point.
- // The scenarios where this can happen is
- // * if the user manually (or via a bad block) rolled back a fast sync node
- // below the sync point.
- // * the last fast sync is not finished while user specifies a full sync this
- // time. But we don't have any recent state for full sync.
- // In these cases however it's safe to reenable fast sync.
- fullBlock, fastBlock := h.chain.CurrentBlock(), h.chain.CurrentFastBlock()
- if fullBlock.NumberU64() == 0 && fastBlock.NumberU64() > 0 {
- h.fastSync = uint32(1)
- log.Warn("Switch sync mode from full sync to fast sync")
- }
- } else {
- if h.chain.CurrentBlock().NumberU64() > 0 {
- // Print warning log if database is not empty to run fast sync.
- log.Warn("Switch sync mode from fast sync to full sync")
- } else {
- // If fast sync was requested and our database is empty, grant it
- h.fastSync = uint32(1)
- if config.Sync == downloader.SnapSync {
- h.snapSync = uint32(1)
- }
- }
- }
- // If we have trusted checkpoints, enforce them on the chain
- if config.Checkpoint != nil {
- h.checkpointNumber = (config.Checkpoint.SectionIndex+1)*params.CHTFrequency - 1
- h.checkpointHash = config.Checkpoint.SectionHead
- }
- // Construct the downloader (long sync) and its backing state bloom if fast
- // sync is requested. The downloader is responsible for deallocating the state
- // bloom when it's done.
- if atomic.LoadUint32(&h.fastSync) == 1 {
- h.stateBloom = trie.NewSyncBloom(config.BloomCache, config.Database)
- }
- h.downloader = downloader.New(h.checkpointNumber, config.Database, h.stateBloom, h.eventMux, h.chain, nil, h.removePeer)
- // Construct the fetcher (short sync)
- validator := func(header *types.Header) error {
- return h.chain.Engine().VerifyHeader(h.chain, header, true)
- }
- heighter := func() uint64 {
- return h.chain.CurrentBlock().NumberU64()
- }
- inserter := func(blocks types.Blocks) (int, error) {
- // If sync hasn't reached the checkpoint yet, deny importing weird blocks.
- //
- // Ideally we would also compare the head block's timestamp and similarly reject
- // the propagated block if the head is too old. Unfortunately there is a corner
- // case when starting new networks, where the genesis might be ancient (0 unix)
- // which would prevent full nodes from accepting it.
- if h.chain.CurrentBlock().NumberU64() < h.checkpointNumber {
- log.Warn("Unsynced yet, discarded propagated block", "number", blocks[0].Number(), "hash", blocks[0].Hash())
- return 0, nil
- }
- // If fast sync is running, deny importing weird blocks. This is a problematic
- // clause when starting up a new network, because fast-syncing miners might not
- // 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 atomic.LoadUint32(&h.fastSync) == 1 {
- log.Warn("Fast syncing, discarded propagated block", "number", blocks[0].Number(), "hash", blocks[0].Hash())
- return 0, nil
- }
- n, err := h.chain.InsertChain(blocks)
- if err == nil {
- atomic.StoreUint32(&h.acceptTxs, 1) // Mark initial sync done on any fetcher import
- }
- return n, err
- }
- h.blockFetcher = fetcher.NewBlockFetcher(false, nil, h.chain.GetBlockByHash, validator, h.BroadcastBlockQLightClient, heighter, nil, inserter, h.removePeer)
- fetchTx := func(peer string, hashes []common.Hash) error {
- p := h.peers.peer(peer)
- if p == nil {
- return errors.New("unknown peer")
- }
- return p.RequestTxs(hashes)
- }
- h.txFetcher = fetcher.NewTxFetcher(h.txpool.Has, h.txpool.AddRemotes, fetchTx)
- h.chainSync = newChainSyncer(h)
- return h, nil
- }
- // runEthPeer registers an eth peer into the joint eth/snap peerset, adds it to
- // various subsistems and starts handling messages.
- func (h *handler) runQLightClientPeer(peer *qlightproto.Peer, handler qlightproto.Handler) error {
- // If the peer has a `snap` extension, wait for it to connect so we can have
- // a uniform initialization/teardown mechanism
- snap, err := h.peers.waitSnapExtension(peer.EthPeer)
- if err != nil {
- peer.Log().Error("Snapshot extension barrier failed", "err", err)
- return err
- }
- // TODO(karalabe): Not sure why this is needed
- if !h.chainSync.handlePeerEvent(peer.EthPeer) {
- return p2p.DiscQuitting
- }
- h.peerWG.Add(1)
- defer h.peerWG.Done()
- // Execute the Ethereum handshake
- var (
- genesis = h.chain.Genesis()
- head = h.chain.CurrentHeader()
- hash = head.Hash()
- number = head.Number.Uint64()
- td = h.chain.GetTd(hash, number)
- )
- forkID := forkid.NewID(h.chain.Config(), h.chain.Genesis().Hash(), h.chain.CurrentHeader().Number.Uint64())
- if err := peer.EthPeer.Handshake(h.networkID, td, hash, genesis.Hash(), forkID, h.forkFilter); err != nil {
- peer.Log().Debug("Ethereum handshake failed", "err", err)
- // Quorum
- // When the Handshake() returns an error, the Run method corresponding to `eth` protocol returns with the error, causing the peer to drop, signal subprotocol as well to exit the `Run` method
- peer.EthPeerDisconnected <- struct{}{}
- // End Quorum
- return err
- }
- log.Info("QLight attempting handshake")
- if err := peer.QLightHandshake(false, h.psi, h.tokenHolder.CurrentToken()); err != nil {
- peer.Log().Debug("QLight handshake failed", "err", err)
- log.Info("QLight handshake failed", "err", err)
- // Quorum
- // When the Handshake() returns an error, the Run method corresponding to `eth` protocol returns with the error, causing the peer to drop, signal subprotocol as well to exit the `Run` method
- peer.EthPeerDisconnected <- struct{}{}
- // End Quorum
- return err
- }
- peer.Log().Debug("QLight handshake result for peer", "peer", peer.ID(), "server", peer.QLightServer(), "psi", peer.QLightPSI(), "token", peer.QLightToken())
- log.Info("QLight handshake result for peer", "peer", peer.ID(), "server", peer.QLightServer(), "psi", peer.QLightPSI(), "token", peer.QLightToken())
- // if we're not connected to a qlight server - disconnect the peer
- if !peer.QLightServer() {
- peer.Log().Debug("QLight connected to a non server peer. Disconnecting.")
- // Quorum
- // When the Handshake() returns an error, the Run method corresponding to `eth` protocol returns with the error, causing the peer to drop, signal subprotocol as well to exit the `Run` method
- peer.EthPeerDisconnected <- struct{}{}
- // End Quorum
- return fmt.Errorf("connected to a non server peer")
- }
- reject := false // reserved peer slots
- if atomic.LoadUint32(&h.snapSync) == 1 {
- if snap == nil {
- // If we are running snap-sync, we want to reserve roughly half the peer
- // slots for peers supporting the snap protocol.
- // The logic here is; we only allow up to 5 more non-snap peers than snap-peers.
- if all, snp := h.peers.len(), h.peers.snapLen(); all-snp > snp+5 {
- reject = true
- }
- }
- }
- // Ignore maxPeers if this is a trusted peer
- if !peer.Peer.Info().Network.Trusted {
- if reject || h.peers.len() >= h.maxPeers {
- return p2p.DiscTooManyPeers
- }
- }
- peer.Log().Debug("Ethereum peer connected", "name", peer.Name())
- // Register the peer locally
- if err := h.peers.registerQPeer(peer); err != nil {
- peer.Log().Error("Ethereum peer registration failed", "err", err)
- // Quorum
- // When the Register() returns an error, the Run method corresponding to `eth` protocol returns with the error, causing the peer to drop, signal subprotocol as well to exit the `Run` method
- peer.EthPeerDisconnected <- struct{}{}
- // End Quorum
- return err
- }
- defer h.removePeer(peer.ID())
- p := h.peers.peer(peer.ID())
- if p == nil {
- return errors.New("peer dropped during handling")
- }
- // Register the peer in the downloader. If the downloader considers it banned, we disconnect
- if err := h.downloader.RegisterPeer(peer.ID(), peer.Version(), peer.EthPeer); err != nil {
- peer.Log().Error("Failed to register peer in eth syncer", "err", err)
- return err
- }
- if snap != nil {
- if err := h.downloader.SnapSyncer.Register(snap); err != nil {
- peer.Log().Error("Failed to register peer in snap syncer", "err", err)
- return err
- }
- }
- h.chainSync.handlePeerEvent(peer.EthPeer)
- // Propagate existing transactions. new transactions appearing
- // after this will be sent via broadcasts.
- h.syncTransactions(peer.EthPeer)
- // If we have a trusted CHT, reject all peers below that (avoid fast sync eclipse)
- if h.checkpointHash != (common.Hash{}) {
- // Request the peer's checkpoint header for chain height/weight validation
- if err := peer.EthPeer.RequestHeadersByNumber(h.checkpointNumber, 1, 0, false); err != nil {
- return err
- }
- // Start a timer to disconnect if the peer doesn't reply in time
- p.syncDrop = time.AfterFunc(syncChallengeTimeout, func() {
- peer.Log().Warn("Checkpoint challenge timed out, dropping", "addr", peer.RemoteAddr(), "type", peer.Name())
- h.removePeer(peer.ID())
- })
- // Make sure it's cleaned up if the peer dies off
- defer func() {
- if p.syncDrop != nil {
- p.syncDrop.Stop()
- p.syncDrop = nil
- }
- }()
- }
- // If we have any explicit authorized block hashes, request them
- for number := range h.authorizationList {
- if err := peer.EthPeer.RequestHeadersByNumber(number, 1, 0, false); err != nil {
- return err
- }
- }
- // Quorum notify other subprotocols that the eth peer is ready, and has been added to the peerset.
- p.EthPeerRegistered <- struct{}{}
- // Quorum
- // Handle incoming messages until the connection is torn down
- return handler(peer)
- }
- func (h *handler) StartQLightClient() {
- h.maxPeers = 1
- // Quorum
- if h.raftMode {
- // We set this immediately in raft mode to make sure the miner never drops
- // incoming txes. Raft mode doesn't use the fetcher or downloader, and so
- // this would never be set otherwise.
- atomic.StoreUint32(&h.acceptTxs, 1)
- }
- // End Quorum
- // start sync handlers
- h.wg.Add(1)
- go h.chainSync.loop()
- }
- func (h *handler) StopQLightClient() {
- if h == nil {
- return
- }
- // Quit chainSync and txsync64.
- // After this is done, no new peers will be accepted.
- close(h.quitSync)
- h.wg.Wait()
- // Disconnect existing sessions.
- // This also closes the gate for any new registrations on the peer set.
- // sessions which are already established but not added to h.peers yet
- // will exit when they try to register.
- h.peers.close()
- h.peerWG.Wait()
- log.Info("QLight client protocol stopped")
- }
- // BroadcastBlock will either propagate a block to a subset of its peers, or
- // will only announce its availability (depending what's requested).
- func (h *handler) BroadcastBlockQLightClient(block *types.Block, propagate bool) {
- }
- // Handle is invoked from a peer's message handler when it receives a new remote
- // message that the handler couldn't consume and serve itself.
- func (h *qlightClientHandler) QHandle(peer *qlightproto.Peer, packet eth.Packet) error {
- // Consume any broadcasts and announces, forwarding the rest to the downloader
- switch packet := packet.(type) {
- case *eth.BlockHeadersPacket:
- return (*ethHandler)(h).Handle(peer.EthPeer, packet)
- case *eth.BlockBodiesPacket:
- txset, uncleset := packet.Unpack()
- h.handleBodiesQLight(txset)
- return (*ethHandler)(h).handleBodies(peer.EthPeer, txset, uncleset)
- case *eth.NewBlockHashesPacket:
- return (*ethHandler)(h).Handle(peer.EthPeer, packet)
- case *eth.NewBlockPacket:
- h.updateCacheWithNonPartyTxData(packet.Block.Transactions())
- return (*ethHandler)(h).handleBlockBroadcast(peer.EthPeer, packet.Block, packet.TD)
- case *qlightproto.BlockPrivateDataPacket:
- return h.handleBlockPrivateData(packet)
- case *eth.NewPooledTransactionHashesPacket:
- return (*ethHandler)(h).Handle(peer.EthPeer, packet)
- case *eth.TransactionsPacket:
- h.updateCacheWithNonPartyTxData(*packet)
- return (*ethHandler)(h).Handle(peer.EthPeer, packet)
- case *eth.PooledTransactionsPacket:
- return (*ethHandler)(h).Handle(peer.EthPeer, packet)
- default:
- return fmt.Errorf("unexpected eth packet type: %T", packet)
- }
- }
- // handleBodies is invoked from a peer's message handler when it transmits a batch
- // of block bodies for the local node to process.
- func (h *qlightClientHandler) handleBodiesQLight(txs [][]*types.Transaction) {
- for _, txArray := range txs {
- h.updateCacheWithNonPartyTxData(txArray)
- }
- }
- func (h *qlightClientHandler) updateCacheWithNonPartyTxData(transactions []*types.Transaction) {
- for _, tx := range transactions {
- if tx.IsPrivate() || tx.IsPrivacyMarker() {
- txHash := common.BytesToEncryptedPayloadHash(tx.Data())
- h.privateClientCache.CheckAndAddEmptyEntry(txHash)
- }
- }
- }
- func (h *qlightClientHandler) handleBlockPrivateData(blockPrivateData *qlightproto.BlockPrivateDataPacket) error {
- for _, b := range *blockPrivateData {
- if err := h.privateClientCache.AddPrivateBlock(b); err != nil {
- return fmt.Errorf("Unable to handle private block data: %v", err)
- }
- }
- return nil
- }
|