Peer health monitoring routine is notified about unhealthy peer. That…

… triggers restarting of the peer.

interface.go

@@ -31,6 +31,7 @@ type PeerI interface {
 	RequestBlock(blockHash *chainhash.Hash)
 	Network() wire.BitcoinNet
 	IsHealthy() bool
+	IsUnhealthyCh() <-chan struct{}
 	Shutdown()
 	Restart()
 }

peer.go

@@ -40,8 +40,8 @@ const (
 	retryReadWriteMessageAttempts        = 5
 	reconnectInterval                    = 10 * time.Second
 
-	pingInterval                   = 2 * time.Minute
-	connectionHealthTickerDuration = 3 * time.Minute
+	pingInterval                   = 30 * time.Second
+	connectionHealthTickerDuration = 1 * time.Minute
 )
 
 type Block struct {
@@ -76,6 +76,7 @@ type Peer struct {
 	userAgentVersion              *string
 	retryReadWriteMessageInterval time.Duration
 	nrWriteHandlers               int
+	isUnhealthyCh                 chan struct{}
 
 	ctx context.Context
 
@@ -105,6 +106,7 @@ func NewPeer(logger *slog.Logger, address string, peerHandler PeerHandlerI, netw
 		address:                       address,
 		writeChan:                     writeChan,
 		pingPongAlive:                 make(chan struct{}, 1),
+		isUnhealthyCh:                 make(chan struct{}),
 		peerHandler:                   peerHandler,
 		logger:                        peerLogger,
 		dial:                          net.Dial,
@@ -421,6 +423,7 @@ func (p *Peer) startReadHandler(ctx context.Context) {
 					return
 				}
 
+				//p.anySignalReceived <- struct{}{}
 				commandLogger := p.logger.With(slog.String(commandKey, strings.ToUpper(msg.Command())))
 
 				// we could check this based on type (switch msg.(type)) but that would not allow
@@ -668,7 +671,10 @@ func (p *Peer) writeRetry(ctx context.Context, msg wire.Message) error {
 }
 
 func (p *Peer) startWriteChannelHandler(ctx context.Context, instance int) {
-	p.writerWg.Add(1)
+	defer func() {
+		p.logger.Debug("Shutting down write handler", slog.Int("instance", instance))
+		p.writerWg.Done()
+	}()
 	go func() {
 		p.logger.Debug("Starting write handler", slog.Int("instance", instance))
 
@@ -784,11 +790,11 @@ func (p *Peer) startMonitorPingPong() {
 
 	go func() {
 		// if no ping/pong signal is received for certain amount of time, mark peer as unhealthy
-		checkConnectionHealthTicker := time.NewTicker(connectionHealthTickerDuration)
+		monitorConnectionTicker := time.NewTicker(connectionHealthTickerDuration)
 
 		defer func() {
 			p.healthMonitorWg.Done()
-			checkConnectionHealthTicker.Stop()
+			monitorConnectionTicker.Stop()
 		}()
 
 		for {
@@ -801,16 +807,19 @@ func (p *Peer) startMonitorPingPong() {
 				}
 				p.writeChan <- wire.NewMsgPing(nonce)
 			case <-p.pingPongAlive:
-				p.mu.Lock()
-				p.isHealthy = true
-				p.mu.Unlock()
-
-				// if ping/pong is received signal reset the ticker
-				checkConnectionHealthTicker.Reset(connectionHealthTickerDuration)
-			case <-checkConnectionHealthTicker.C:
+				// if ping/pong signal is received reset the ticker
+				monitorConnectionTicker.Reset(connectionHealthTickerDuration)
+				p.setHealthy()
+			case <-monitorConnectionTicker.C:
 
 				p.mu.Lock()
 				p.isHealthy = false
+
+				select {
+				case p.isUnhealthyCh <- struct{}{}:
+				default:
+				}
+
 				p.logger.Warn("peer unhealthy")
 				p.mu.Unlock()
 			case <-p.ctx.Done():
@@ -820,6 +829,22 @@ func (p *Peer) startMonitorPingPong() {
 	}()
 }
 
+func (p *Peer) IsUnhealthyCh() <-chan struct{} {
+	return p.isUnhealthyCh
+}
+
+func (p *Peer) setHealthy() {
+
+	p.mu.Lock()
+	if p.isHealthy {
+		p.mu.Unlock()
+		return
+	}
+	p.logger.Warn("peer healthy")
+	p.isHealthy = true
+	p.mu.Unlock()
+}
+
 func (p *Peer) IsHealthy() bool {
 	p.mu.Lock()
 	defer p.mu.Unlock()

peer_manager.go

@@ -96,24 +96,24 @@ func (pm *PeerManager) Shutdown() {
 }
 
 func (pm *PeerManager) StartMonitorPeerHealth() {
-	ticker := time.NewTicker(pm.monitorPeersInterval)
-	pm.waitGroup.Add(1)
-	go func() {
-		defer pm.waitGroup.Done()
-		for {
-			select {
-			case <-pm.ctx.Done():
-				return
-			case <-ticker.C:
-				for _, peer := range pm.GetPeers() {
-					if !peer.IsHealthy() {
-						pm.logger.Warn("peer unhealthy - restarting", slog.String("address", peer.String()), slog.Bool("connected", peer.Connected()))
-						peer.Restart()
-					}
+
+	for _, peer := range pm.peers {
+		pm.waitGroup.Add(1)
+		go func(p PeerI) {
+			defer func() {
+				pm.waitGroup.Done()
+			}()
+			for {
+				select {
+				case <-pm.ctx.Done():
+					return
+				case <-p.IsUnhealthyCh():
+					pm.logger.Warn("peer unhealthy - restarting", slog.String("address", p.String()), slog.Bool("connected", p.Connected()))
+					p.Restart()
 				}
 			}
-		}
-	}()
+		}(peer)
+	}
 }
 
 // AnnounceTransaction will send an INV message to the provided peers or to selected peers if peers is nil