diff --git a/management/server/grpcserver.go b/management/server/grpcserver.go index b639c6613..26cba1ba5 100644 --- a/management/server/grpcserver.go +++ b/management/server/grpcserver.go @@ -255,24 +255,39 @@ func (s *GRPCServer) Sync(req *proto.EncryptedMessage, srv proto.ManagementServi func (s *GRPCServer) handleUpdates(ctx context.Context, accountID string, peerKey wgtypes.Key, peer *nbpeer.Peer, updates *UpdateChannel, srv proto.ManagementService_SyncServer) error { log.WithContext(ctx).Tracef("starting to handle updates for peer %s", peerKey.String()) - // Channel to receive network map updates - networkMapCh := make(chan *UpdateMessage) + // Channel to receive network map updates with metadata + type networkMapUpdate struct { + msg *UpdateMessage + overwrites int + timeSinceLastPop time.Duration + } + networkMapCh := make(chan networkMapUpdate) // Start goroutine to Pop from buffer go func() { for { - update, ok := updates.NetworkMap.Pop(ctx) + update, overwrites, timeSinceLastPop, ok := updates.NetworkMap.Pop(ctx) if !ok { close(networkMapCh) return } start := time.Now() select { - case networkMapCh <- update: - log.WithContext(ctx).Debugf("forwarded an update for peer %s from the network map buffer in %v", peerKey.String(), time.Since(start)) - start = time.Now() - time.Sleep(time.Duration(s.debounce) * time.Second) - log.WithContext(ctx).Debugf("debounced for %v seconds for peer %s", time.Since(start), peerKey.String()) + case networkMapCh <- networkMapUpdate{ + msg: update, + overwrites: overwrites, + timeSinceLastPop: timeSinceLastPop, + }: + log.WithContext(ctx).Debugf("forwarded an update for peer %s from the network map buffer in %v (overwrites: %d)", peerKey.String(), time.Since(start), overwrites) + + // Adaptive debounce: increase delay based on overwrite rate + // If many overwrites happened, wait longer to let things settle + debounce := s.calculateDebounce(overwrites, timeSinceLastPop) + if debounce > 0 { + start = time.Now() + time.Sleep(debounce) + log.WithContext(ctx).Debugf("debounced for %v for peer %s (overwrites: %d)", time.Since(start), peerKey.String(), overwrites) + } case <-ctx.Done(): return } @@ -299,14 +314,14 @@ func (s *GRPCServer) handleUpdates(ctx context.Context, accountID string, peerKe return err } - case update, ok := <-networkMapCh: + case updateData, ok := <-networkMapCh: if !ok { log.WithContext(ctx).Debugf("update buffer for peer %s closed", peerKey.String()) s.cancelPeerRoutines(ctx, accountID, peer) return nil } - log.WithContext(ctx).Debugf("sending latest update to peer %s", peerKey.String()) - if err := s.sendUpdate(ctx, accountID, peerKey, peer, update, srv); err != nil { + log.WithContext(ctx).Debugf("sending latest update to peer %s (overwrites: %d)", peerKey.String(), updateData.overwrites) + if err := s.sendUpdate(ctx, accountID, peerKey, peer, updateData.msg, srv); err != nil { return err } @@ -320,6 +335,33 @@ func (s *GRPCServer) handleUpdates(ctx context.Context, accountID string, peerKe } } +// calculateDebounce calculates adaptive debounce duration based on overwrite rate +// More overwrites = longer debounce to let updates settle +func (s *GRPCServer) calculateDebounce(overwrites int, timeSinceLastPop time.Duration) time.Duration { + if overwrites == 0 { + // No overwrites, use base debounce + return time.Duration(s.debounce) * time.Second + } + + var rate float64 + if timeSinceLastPop > 0 { + rate = float64(overwrites) / timeSinceLastPop.Seconds() + } else { + rate = float64(overwrites) + } + + multiplier := 2.0 + adaptiveDebounce := float64(s.debounce) + (rate * multiplier) + + // Cap at 10x base debounce + maxDebounce := float64(s.debounce) * 10 + if adaptiveDebounce > maxDebounce { + adaptiveDebounce = maxDebounce + } + + return time.Duration(adaptiveDebounce * float64(time.Second)) +} + // sendUpdate encrypts the update message using the peer key and the server's wireguard key, // then sends the encrypted message to the connected peer via the sync server. func (s *GRPCServer) sendUpdate(ctx context.Context, accountID string, peerKey wgtypes.Key, peer *nbpeer.Peer, update *UpdateMessage, srv proto.ManagementService_SyncServer) error { diff --git a/management/server/update_buffer.go b/management/server/update_buffer.go index 09bd8145d..26e5507c6 100644 --- a/management/server/update_buffer.go +++ b/management/server/update_buffer.go @@ -3,16 +3,19 @@ package server import ( "context" "sync" + "time" "github.com/netbirdio/netbird/management/server/telemetry" ) type UpdateBuffer struct { - mu sync.Mutex - cond *sync.Cond - update *UpdateMessage - closed bool - metrics *telemetry.UpdateChannelMetrics + mu sync.Mutex + cond *sync.Cond + update *UpdateMessage + closed bool + metrics *telemetry.UpdateChannelMetrics + overwriteCount int // Number of overwrites since last Pop + lastPopTime time.Time // Time of last Pop } func NewUpdateBuffer(metrics *telemetry.UpdateChannelMetrics) *UpdateBuffer { @@ -30,6 +33,7 @@ func (b *UpdateBuffer) Push(update *UpdateMessage) { b.metrics.CountBufferPush() } else { b.metrics.CountBufferOverwrite() + b.overwriteCount++ } b.update = update @@ -41,7 +45,7 @@ func (b *UpdateBuffer) Push(update *UpdateMessage) { b.metrics.CountBufferIgnore() } -func (b *UpdateBuffer) Pop(ctx context.Context) (*UpdateMessage, bool) { +func (b *UpdateBuffer) Pop(ctx context.Context) (*UpdateMessage, int, time.Duration, bool) { b.mu.Lock() defer b.mu.Unlock() @@ -60,11 +64,25 @@ func (b *UpdateBuffer) Pop(ctx context.Context) (*UpdateMessage, bool) { } if b.closed { - return nil, false + return nil, 0, 0, false } + msg := b.update + overwrites := b.overwriteCount + + // Calculate time since last pop + now := time.Now() + var timeSinceLastPop time.Duration + if !b.lastPopTime.IsZero() { + timeSinceLastPop = now.Sub(b.lastPopTime) + } + + // Reset counters b.update = nil - return msg, true + b.overwriteCount = 0 + b.lastPopTime = now + + return msg, overwrites, timeSinceLastPop, true } func (b *UpdateBuffer) Close() {