refactor: switch ConnectToNode to use a retries counter and make it configurable via P2PConfig
This commit is contained in:
parent
a87bfe7ba6
commit
8f32074667
|
@ -17,6 +17,7 @@ type P2PConfig struct {
|
||||||
Network string `mapstructure:"network"`
|
Network string `mapstructure:"network"`
|
||||||
Peers PeersConfig `mapstructure:"peers"`
|
Peers PeersConfig `mapstructure:"peers"`
|
||||||
MaxOutgoingPeerFailures uint `mapstructure:"max_outgoing_peer_failures"`
|
MaxOutgoingPeerFailures uint `mapstructure:"max_outgoing_peer_failures"`
|
||||||
|
MaxConnectionAttempts uint `mapstructure:"max_connection_attempts"`
|
||||||
}
|
}
|
||||||
|
|
||||||
type PeersConfig struct {
|
type PeersConfig struct {
|
||||||
|
|
|
@ -66,7 +66,7 @@ func (m MediatorDefault) RegistryGet(pk []byte) (protocol.SignedRegistryEntry, e
|
||||||
}
|
}
|
||||||
|
|
||||||
func (m MediatorDefault) ConnectToNode(connectionUris []*url.URL, retried bool, fromPeer net.Peer) error {
|
func (m MediatorDefault) ConnectToNode(connectionUris []*url.URL, retried bool, fromPeer net.Peer) error {
|
||||||
return m.Services().P2P().ConnectToNode(connectionUris, retried, fromPeer)
|
return m.Services().P2P().ConnectToNode(connectionUris, 0, fromPeer)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (m MediatorDefault) ServicesStarted() bool {
|
func (m MediatorDefault) ServicesStarted() bool {
|
||||||
|
|
|
@ -98,7 +98,7 @@ func (p *P2PServiceDefault) Start(ctx context.Context) error {
|
||||||
|
|
||||||
peer := peer
|
peer := peer
|
||||||
go func() {
|
go func() {
|
||||||
err := p.ConnectToNode([]*url.URL{u}, false, nil)
|
err := p.ConnectToNode([]*url.URL{u}, 0, nil)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
p.Logger().Error("failed to connect to initial peer", zap.Error(err), zap.String("peer", peer))
|
p.Logger().Error("failed to connect to initial peer", zap.Error(err), zap.String("peer", peer))
|
||||||
}
|
}
|
||||||
|
@ -129,7 +129,7 @@ func (p *P2PServiceDefault) Init(ctx context.Context) error {
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
func (p *P2PServiceDefault) ConnectToNode(connectionUris []*url.URL, retried bool, fromPeer net.Peer) error {
|
func (p *P2PServiceDefault) ConnectToNode(connectionUris []*url.URL, retry uint, fromPeer net.Peer) error {
|
||||||
if !p.Services().IsStarted() {
|
if !p.Services().IsStarted() {
|
||||||
if !p.Services().IsStarting() {
|
if !p.Services().IsStarting() {
|
||||||
return nil
|
return nil
|
||||||
|
@ -237,7 +237,7 @@ func (p *P2PServiceDefault) ConnectToNode(connectionUris []*url.URL, retried boo
|
||||||
|
|
||||||
socket, err := net.CreateTransportSocket(scheme, connectionUri)
|
socket, err := net.CreateTransportSocket(scheme, connectionUri)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if retried {
|
if retry >= p.Config().P2P.MaxConnectionAttempts {
|
||||||
p.Logger().Error("failed to connect, too many retries", zap.String("node", connectionUri.String()), zap.Error(err))
|
p.Logger().Error("failed to connect, too many retries", zap.String("node", connectionUri.String()), zap.Error(err))
|
||||||
counter := uint(0)
|
counter := uint(0)
|
||||||
if p.outgoingPeerFailures.Contains(idString) {
|
if p.outgoingPeerFailures.Contains(idString) {
|
||||||
|
@ -291,7 +291,7 @@ func (p *P2PServiceDefault) ConnectToNode(connectionUris []*url.URL, retried boo
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
retried = true
|
retry++
|
||||||
|
|
||||||
p.Logger().Error("failed to connect", zap.String("node", connectionUri.String()), zap.Error(err))
|
p.Logger().Error("failed to connect", zap.String("node", connectionUri.String()), zap.Error(err))
|
||||||
|
|
||||||
|
@ -305,7 +305,7 @@ func (p *P2PServiceDefault) ConnectToNode(connectionUris []*url.URL, retried boo
|
||||||
|
|
||||||
time.Sleep(time.Duration(delayDeref) * time.Second)
|
time.Sleep(time.Duration(delayDeref) * time.Second)
|
||||||
|
|
||||||
return p.ConnectToNode(connectionUris, retried, fromPeer)
|
return p.ConnectToNode(connectionUris, retry, fromPeer)
|
||||||
}
|
}
|
||||||
|
|
||||||
if p.outgoingPeerFailures.Contains(idString) {
|
if p.outgoingPeerFailures.Contains(idString) {
|
||||||
|
|
|
@ -12,7 +12,7 @@ import (
|
||||||
type P2PService interface {
|
type P2PService interface {
|
||||||
SelfConnectionUris() []*url.URL
|
SelfConnectionUris() []*url.URL
|
||||||
Peers() structs.Map
|
Peers() structs.Map
|
||||||
ConnectToNode(connectionUris []*url.URL, retried bool, fromPeer net.Peer) error
|
ConnectToNode(connectionUris []*url.URL, retry uint, fromPeer net.Peer) error
|
||||||
OnNewPeer(peer net.Peer, verifyId bool) error
|
OnNewPeer(peer net.Peer, verifyId bool) error
|
||||||
GetNodeScore(nodeId *encoding.NodeId) (float64, error)
|
GetNodeScore(nodeId *encoding.NodeId) (float64, error)
|
||||||
SortNodesByScore(nodes []*encoding.NodeId) ([]*encoding.NodeId, error)
|
SortNodesByScore(nodes []*encoding.NodeId) ([]*encoding.NodeId, error)
|
||||||
|
|
Loading…
Reference in New Issue