log the most busy peers

signal/server/send_tracker.go

@@ -0,0 +1,77 @@
+package server
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+)
+
+const sendRateLogInterval = 5 * time.Minute
+
+// sendRateTracker tracks per-key message counts and logs the busiest peers periodically.
+type sendRateTracker struct {
+	mu     sync.Mutex
+	counts map[string]int64
+}
+
+func newSendRateTracker() *sendRateTracker {
+	return &sendRateTracker{counts: make(map[string]int64)}
+}
+
+func (t *sendRateTracker) increment(key string) {
+	t.mu.Lock()
+	t.counts[key]++
+	t.mu.Unlock()
+}
+
+// resetAndSnapshot atomically returns current counts and resets the tracker.
+func (t *sendRateTracker) resetAndSnapshot() map[string]int64 {
+	t.mu.Lock()
+	snap := t.counts
+	t.counts = make(map[string]int64, len(snap))
+	t.mu.Unlock()
+	return snap
+}
+
+// logSendRates periodically logs peers that have at least half the rate of the busiest peer.
+func (t *sendRateTracker) logSendRates(ctx context.Context) {
+	ticker := time.NewTicker(sendRateLogInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			snap := t.resetAndSnapshot()
+			if len(snap) == 0 {
+				continue
+			}
+
+			var maxCount int64
+			for _, count := range snap {
+				if count > maxCount {
+					maxCount = count
+				}
+			}
+
+			threshold := int64(float64(maxCount) * 0.95)
+			intervalMin := sendRateLogInterval.Minutes()
+
+			log.Debugf("send rate stats: %d unique peers in last %.0fs, max rate %.1f msg/min",
+				len(snap), sendRateLogInterval.Seconds(), float64(maxCount)/intervalMin)
+			logged := 0
+			for key, count := range snap {
+				if count >= threshold {
+					log.Debugf("peer [%s] %.1f msg/min", key, float64(count)/intervalMin)
+					logged++
+					if logged >= 100 {
+						break
+					}
+				}
+			}
+		}
+	}
+}

signal/server/send_tracker_test.go

@@ -0,0 +1,56 @@
+package server
+
+import (
+	"sync"
+	"testing"
+)
+
+func TestSendRateTracker_Increment(t *testing.T) {
+	tracker := newSendRateTracker()
+
+	tracker.increment("peer-a")
+	tracker.increment("peer-a")
+	tracker.increment("peer-b")
+
+	snap := tracker.resetAndSnapshot()
+	if snap["peer-a"] != 2 {
+		t.Errorf("expected peer-a count 2, got %d", snap["peer-a"])
+	}
+	if snap["peer-b"] != 1 {
+		t.Errorf("expected peer-b count 1, got %d", snap["peer-b"])
+	}
+}
+
+func TestSendRateTracker_ResetAndSnapshot_Resets(t *testing.T) {
+	tracker := newSendRateTracker()
+	tracker.increment("peer-a")
+
+	snap1 := tracker.resetAndSnapshot()
+	if snap1["peer-a"] != 1 {
+		t.Fatalf("expected 1, got %d", snap1["peer-a"])
+	}
+
+	snap2 := tracker.resetAndSnapshot()
+	if len(snap2) != 0 {
+		t.Errorf("expected empty snapshot after reset, got %v", snap2)
+	}
+}
+
+func TestSendRateTracker_ConcurrentIncrement(t *testing.T) {
+	tracker := newSendRateTracker()
+
+	var wg sync.WaitGroup
+	for i := 0; i < 100; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			tracker.increment("peer-x")
+		}()
+	}
+	wg.Wait()
+
+	snap := tracker.resetAndSnapshot()
+	if snap["peer-x"] != 100 {
+		t.Errorf("expected 100, got %d", snap["peer-x"])
+	}
+}

signal/server/signal.go

@@ -59,6 +59,8 @@ type Server struct {
 	successHeader metadata.MD
 
 	sendTimeout time.Duration
+
+	sendTracker *sendRateTracker
 }
 
 // NewServer creates a new Signal server
@@ -86,8 +88,11 @@ func NewServer(ctx context.Context, meter metric.Meter, metricsPrefix ...string)
 		metrics:       appMetrics,
 		successHeader: metadata.Pairs(proto.HeaderRegistered, "1"),
 		sendTimeout:   sTimeout,
+		sendTracker:   newSendRateTracker(),
 	}
 
+	go s.sendTracker.logSendRates(ctx)
+
 	return s, nil
 }
 
@@ -95,6 +100,8 @@ func NewServer(ctx context.Context, meter metric.Meter, metricsPrefix ...string)
 func (s *Server) Send(ctx context.Context, msg *proto.EncryptedMessage) (*proto.EncryptedMessage, error) {
 	log.Tracef("received a new message to send from peer [%s] to peer [%s]", msg.Key, msg.RemoteKey)
 
+	s.sendTracker.increment(msg.Key)
+
 	if _, found := s.registry.Get(msg.RemoteKey); found {
 		s.forwardMessageToPeer(ctx, msg)
 		return &proto.EncryptedMessage{}, nil