[backport] p2p: use decaying tags for fetch peers (#5564)

CHANGELOG.md

@@ -6,6 +6,8 @@ See [RELEASE](./RELEASE.md) for workflow instructions.
 
 ### Improvements
 
+* [#5564](https://github.com/spacemeshos/go-spacemesh/pull/5564) Use decaying tags for fetch peers. This prevents
+  libp2p's Connection Manager from breaking sync.
 * [#5548](https://github.com/spacemeshos/go-spacemesh/pull/5548) Disable mesh aggremenet sync protocol.
 
   It reduces number of requests for historical activation ids.

fetch/fetch.go

@@ -107,6 +107,7 @@ type Config struct {
 	ServersConfig        map[string]ServerConfig `mapstructure:"servers"`
 	PeersRateThreshold   float64                 `mapstructure:"peers-rate-threshold"`
 	GetAtxsConcurrency   int64                   // The maximum number of concurrent requests to get ATXs.
+	DecayingTag          server.DecayingTagSpec  `mapstructure:"decaying-tag"`
 }
 
 func (c Config) getServerConfig(protocol string) ServerConfig {
@@ -149,6 +150,12 @@ func DefaultConfig() Config {
 		},
 		PeersRateThreshold: 0.02,
 		GetAtxsConcurrency: 100,
+		DecayingTag: server.DecayingTagSpec{
+			Interval: time.Minute,
+			Inc:      1000,
+			Dec:      1000,
+			Cap:      10000,
+		},
 	}
 }
 
@@ -291,6 +298,7 @@ func (f *Fetch) registerServer(
 		server.WithTimeout(f.cfg.RequestTimeout),
 		server.WithHardTimeout(f.cfg.RequestHardTimeout),
 		server.WithLog(f.logger),
+		server.WithDecayingTag(f.cfg.DecayingTag),
 	}
 	if f.cfg.EnableServerMetrics {
 		opts = append(opts, server.WithMetrics())

p2p/server/interface.go

@@ -5,6 +5,7 @@ import (
 	"io"
 	"time"
 
+	"github.com/libp2p/go-libp2p/core/connmgr"
 	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/libp2p/go-libp2p/core/protocol"
@@ -17,6 +18,7 @@ type Host interface {
 	SetStreamHandler(protocol.ID, network.StreamHandler)
 	NewStream(context.Context, peer.ID, ...protocol.ID) (network.Stream, error)
 	Network() network.Network
+	ConnManager() connmgr.ConnManager
 }
 
 type peerStream interface {

p2p/server/server.go

@@ -9,6 +9,7 @@
 	"io"
 	"time"
 
+	"github.com/libp2p/go-libp2p/core/connmgr"
 	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/libp2p/go-libp2p/core/protocol"
@@ -20,6 +21,13 @@
 	"github.com/spacemeshos/go-spacemesh/log"
 )
 
+type DecayingTagSpec struct {
+	Interval time.Duration `mapstructure:"interval"`
+	Inc      int           `mapstructure:"inc"`
+	Dec      int           `mapstructure:"dec"`
+	Cap      int           `mapstructure:"cap"`
+}
+
 // ErrNotConnected is returned when peer is not connected.
 var ErrNotConnected = errors.New("peer is not connected")
 
@@ -88,6 +96,12 @@
 	}
 }
 
+func WithDecayingTag(tag DecayingTagSpec) Opt {
+	return func(s *Server) {
+		s.decayingTagSpec = &tag
+	}
+}
+
 // Handler is the handler to be defined by the application.
 type Handler func(context.Context, []byte) ([]byte, error)
 
@@ -110,6 +124,8 @@
 	queueSize           int
 	requestsPerInterval int
 	interval            time.Duration
+	decayingTagSpec     *DecayingTagSpec
+	decayingTag         connmgr.DecayingTag
 
 	metrics *tracker // metrics can be nil
 
@@ -133,6 +149,23 @@
 	for _, opt := range opts {
 		opt(srv)
 	}
+
+	if srv.decayingTagSpec != nil {
+		decayer, supported := connmgr.SupportsDecay(h.ConnManager())
+		if supported {
+			tag, err := decayer.RegisterDecayingTag(
+				"server:"+proto,
+				srv.decayingTagSpec.Interval,
+				connmgr.DecayFixed(srv.decayingTagSpec.Dec),
+				connmgr.BumpSumBounded(0, srv.decayingTagSpec.Cap))
+			if err != nil {
+				srv.logger.Error("error registering decaying tag", log.Err(err))
+			} else {
+				srv.decayingTag = tag
+			}
+		}
+	}
+
 	return srv
 }
 
@@ -176,6 +209,9 @@
 				return nil
 			}
 			eg.Go(func() error {
+				if s.decayingTag != nil {
+					s.decayingTag.Bump(req.stream.Conn().RemotePeer(), s.decayingTagSpec.Inc)
+				}
 				ok := s.queueHandler(ctx, req.stream)
 				if s.metrics != nil {
 					s.metrics.serverLatency.Observe(time.Since(req.received).Seconds())