package peers import ( "crypto/ecdsa" "errors" "sync" "time" "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/discv5" "github.com/ethereum/go-ethereum/p2p/enode" "github.com/status-im/status-go/common" "github.com/status-im/status-go/discovery" "github.com/status-im/status-go/params" "github.com/status-im/status-go/peers/verifier" "github.com/status-im/status-go/signal" ) var ( // ErrDiscv5NotRunning returned when pool is started but discover v5 is not running or not enabled. ErrDiscv5NotRunning = errors.New("Discovery v5 is not running") ) // PoolEvent is a type used to for peer pool events. type PoolEvent string const ( immediately = 0 * time.Minute // expirationPeriod is an amount of time while peer is considered as a connectable expirationPeriod = 60 * time.Minute // discoveryRestartTimeout defines how often loop will try to start discovery server discoveryRestartTimeout = 2 * time.Second // DefaultFastSync is a recommended value for aggressive peers search. DefaultFastSync = 3 * time.Second // DefaultSlowSync is a recommended value for slow (background) peers search. DefaultSlowSync = 30 * time.Second // DefaultDiscV5Timeout is a timeout after which Discv5 is stopped. DefaultDiscV5Timeout = 3 * time.Minute // DefaultTopicFastModeTimeout is a timeout after which sync mode is switched to slow mode. DefaultTopicFastModeTimeout = 30 * time.Second // DefaultTopicStopSearchDelay is the default delay when stopping a topic search. DefaultTopicStopSearchDelay = 10 * time.Second ) // Options is a struct with PeerPool configuration. type Options struct { FastSync time.Duration SlowSync time.Duration // After this time, Discovery is stopped even if max peers is not reached. DiscServerTimeout time.Duration // AllowStop allows stopping Discovery when reaching max peers or after timeout. AllowStop bool // TopicStopSearchDelay time stopSearch will be waiting for max cached peers to be // filled before really stopping the search. TopicStopSearchDelay time.Duration // TrustedMailServers is a list of trusted nodes. TrustedMailServers []enode.ID } // NewDefaultOptions returns a struct with default Options. func NewDefaultOptions() *Options { return &Options{ FastSync: DefaultFastSync, SlowSync: DefaultSlowSync, DiscServerTimeout: DefaultDiscV5Timeout, AllowStop: false, TopicStopSearchDelay: DefaultTopicStopSearchDelay, } } type peerInfo struct { // discoveredTime last time when node was found by v5 discoveredTime time.Time // dismissed is true when our node requested a disconnect dismissed bool // added is true when the node tries to add this peer to a server added bool node *discv5.Node // store public key separately to make peerInfo more independent from discv5 publicKey *ecdsa.PublicKey } func (p *peerInfo) NodeID() enode.ID { return enode.PubkeyToIDV4(p.publicKey) } // PeerPool manages discovered peers and connects them to p2p server type PeerPool struct { opts *Options discovery discovery.Discovery // config can be set only once per pool life cycle config map[discv5.Topic]params.Limits cache *Cache mu sync.RWMutex timeoutMu sync.RWMutex topics []TopicPoolInterface serverSubscription event.Subscription events chan *p2p.PeerEvent quit chan struct{} wg sync.WaitGroup timeout <-chan time.Time updateTopic chan *updateTopicRequest } // NewPeerPool creates instance of PeerPool func NewPeerPool(discovery discovery.Discovery, config map[discv5.Topic]params.Limits, cache *Cache, options *Options) *PeerPool { return &PeerPool{ opts: options, discovery: discovery, config: config, cache: cache, } } func (p *PeerPool) setDiscoveryTimeout() { p.timeoutMu.Lock() defer p.timeoutMu.Unlock() if p.opts.AllowStop && p.opts.DiscServerTimeout > 0 { p.timeout = time.After(p.opts.DiscServerTimeout) } } // Start creates topic pool for each topic in config and subscribes to server events. func (p *PeerPool) Start(server *p2p.Server) error { if !p.discovery.Running() { return ErrDiscv5NotRunning } p.mu.Lock() defer p.mu.Unlock() // init channels p.quit = make(chan struct{}) p.updateTopic = make(chan *updateTopicRequest) p.setDiscoveryTimeout() // subscribe to peer events p.events = make(chan *p2p.PeerEvent, 20) p.serverSubscription = server.SubscribeEvents(p.events) p.wg.Add(1) go func() { defer common.LogOnPanic() p.handleServerPeers(server, p.events) p.wg.Done() }() // collect topics and start searching for nodes p.topics = make([]TopicPoolInterface, 0, len(p.config)) for topic, limits := range p.config { var topicPool TopicPoolInterface t := newTopicPool(p.discovery, topic, limits, p.opts.SlowSync, p.opts.FastSync, p.cache) if topic == MailServerDiscoveryTopic { v, err := p.initVerifier() if err != nil { return err } topicPool = newCacheOnlyTopicPool(t, v) } else { topicPool = t } if err := topicPool.StartSearch(server); err != nil { return err } p.topics = append(p.topics, topicPool) } // discovery must be already started when pool is started signal.SendDiscoveryStarted() return nil } func (p *PeerPool) initVerifier() (v Verifier, err error) { return verifier.NewLocalVerifier(p.opts.TrustedMailServers), nil } func (p *PeerPool) startDiscovery() error { if p.discovery.Running() { return nil } if err := p.discovery.Start(); err != nil { return err } p.mu.Lock() p.setDiscoveryTimeout() p.mu.Unlock() signal.SendDiscoveryStarted() return nil } func (p *PeerPool) stopDiscovery(server *p2p.Server) { if !p.discovery.Running() { return } if err := p.discovery.Stop(); err != nil { log.Error("discovery errored when stopping", "err", err) } for _, t := range p.topics { t.StopSearch(server) } p.timeoutMu.Lock() p.timeout = nil p.timeoutMu.Unlock() signal.SendDiscoveryStopped() } // restartDiscovery and search for topics that have peer count below min func (p *PeerPool) restartDiscovery(server *p2p.Server) error { if !p.discovery.Running() { if err := p.startDiscovery(); err != nil { return err } log.Debug("restarted discovery from peer pool") } for _, t := range p.topics { if !t.BelowMin() || t.SearchRunning() { continue } err := t.StartSearch(server) if err != nil { log.Error("search failed to start", "error", err) } } return nil } // handleServerPeers watches server peer events, notifies topic pools about changes // in the peer set and stops the discv5 if all topic pools collected enough peers. // // @TODO(adam): split it into peers and discovery management loops. This should // simplify the whole logic and allow to remove `timeout` field from `PeerPool`. func (p *PeerPool) handleServerPeers(server *p2p.Server, events <-chan *p2p.PeerEvent) { retryDiscv5 := make(chan struct{}, 1) stopDiscv5 := make(chan struct{}, 1) queueRetry := func(d time.Duration) { go func() { defer common.LogOnPanic() time.Sleep(d) select { case retryDiscv5 <- struct{}{}: default: } }() } queueStop := func() { go func() { defer common.LogOnPanic() select { case stopDiscv5 <- struct{}{}: default: } }() } for { // We use a separate lock for timeout, as this loop should // always be running, otherwise the p2p.Server will hang. // Because the handler of events might potentially hang on the // server, deadlocking if this loop is waiting for the global lock. // NOTE: this code probably needs to be refactored and simplified // as it's difficult to follow the asynchronous nature of it. p.timeoutMu.RLock() timeout := p.timeout p.timeoutMu.RUnlock() select { case <-p.quit: log.Debug("stopping DiscV5 because of quit") p.stopDiscovery(server) return case <-timeout: log.Info("DiscV5 timed out") p.stopDiscovery(server) case <-retryDiscv5: if err := p.restartDiscovery(server); err != nil { log.Error("starting discv5 failed", "error", err, "retry", discoveryRestartTimeout) queueRetry(discoveryRestartTimeout) } case <-stopDiscv5: p.handleStopTopics(server) case req := <-p.updateTopic: if p.updateTopicLimits(server, req) == nil { if !p.discovery.Running() { queueRetry(immediately) } } case event := <-events: // NOTE: handlePeerEventType needs to be called asynchronously // as it publishes on the <-events channel, leading to a deadlock // if events channel is full. go p.handlePeerEventType(server, event, queueRetry, queueStop) } } } func (p *PeerPool) handlePeerEventType(server *p2p.Server, event *p2p.PeerEvent, queueRetry func(time.Duration), queueStop func()) { defer common.LogOnPanic() p.mu.Lock() defer p.mu.Unlock() var shouldRetry bool var shouldStop bool switch event.Type { case p2p.PeerEventTypeDrop: log.Debug("confirm peer dropped", "ID", event.Peer) if p.handleDroppedPeer(server, event.Peer) { shouldRetry = true } case p2p.PeerEventTypeAdd: // skip other events log.Debug("confirm peer added", "ID", event.Peer) p.handleAddedPeer(server, event.Peer) shouldStop = true default: return } // First we send the discovery summary SendDiscoverySummary(server.PeersInfo()) // then we send the stop event if shouldRetry { queueRetry(immediately) } else if shouldStop { queueStop() } } // handleAddedPeer notifies all topics about added peer. func (p *PeerPool) handleAddedPeer(server *p2p.Server, nodeID enode.ID) { for _, t := range p.topics { t.ConfirmAdded(server, nodeID) if p.opts.AllowStop && t.MaxReached() { t.setStopSearchTimeout(p.opts.TopicStopSearchDelay) } } } // handleStopTopics stops the search on any topics having reached its max cached // limit or its delay stop is expired, additionally will stop discovery if all // peers are stopped. func (p *PeerPool) handleStopTopics(server *p2p.Server) { if !p.opts.AllowStop { return } for _, t := range p.topics { if t.readyToStopSearch() { t.StopSearch(server) } } if p.allTopicsStopped() { log.Debug("closing discv5 connection because all topics reached max limit") p.stopDiscovery(server) } } // allTopicsStopped returns true if all topics are stopped. func (p *PeerPool) allTopicsStopped() (all bool) { if !p.opts.AllowStop { return false } all = true for _, t := range p.topics { if !t.isStopped() { all = false } } return all } // handleDroppedPeer notifies every topic about dropped peer and returns true if any peer have connections // below min limit func (p *PeerPool) handleDroppedPeer(server *p2p.Server, nodeID enode.ID) (any bool) { for _, t := range p.topics { confirmed := t.ConfirmDropped(server, nodeID) if confirmed { newPeer := t.AddPeerFromTable(server) if newPeer != nil { log.Debug("added peer from local table", "ID", newPeer.ID) } } log.Debug("search", "topic", t.Topic(), "below min", t.BelowMin()) if t.BelowMin() && !t.SearchRunning() { any = true } } return any } // Stop closes pool quit channel and all channels that are watched by search queries // and waits till all goroutines will exit. func (p *PeerPool) Stop() { // pool wasn't started if p.quit == nil { return } select { case <-p.quit: return default: log.Debug("started closing peer pool") close(p.quit) } p.serverSubscription.Unsubscribe() p.wg.Wait() } type updateTopicRequest struct { Topic string Limits params.Limits } // UpdateTopic updates the pre-existing TopicPool limits. func (p *PeerPool) UpdateTopic(topic string, limits params.Limits) error { if _, err := p.getTopic(topic); err != nil { return err } p.updateTopic <- &updateTopicRequest{ Topic: topic, Limits: limits, } return nil } func (p *PeerPool) updateTopicLimits(server *p2p.Server, req *updateTopicRequest) error { t, err := p.getTopic(req.Topic) if err != nil { return err } t.SetLimits(req.Limits) return nil } func (p *PeerPool) getTopic(topic string) (TopicPoolInterface, error) { for _, t := range p.topics { if t.Topic() == discv5.Topic(topic) { return t, nil } } return nil, errors.New("topic not found") }