Merge branch 'main' into prototype/reverse-proxy

management/internals/shared/grpc/server.go

@@ -77,8 +77,9 @@ type Server struct {
 
 	oAuthConfigProvider idp.OAuthConfigProvider
 
-	syncSem atomic.Int32
-	syncLim int32
+	syncSem        atomic.Int32
+	syncLimEnabled bool
+	syncLim        int32
 }
 
 // NewServer creates a new Management server
@@ -108,6 +109,7 @@ func NewServer(
 	blockPeersWithSameConfig := strings.ToLower(os.Getenv(envBlockPeers)) == "true"
 
 	syncLim := int32(defaultSyncLim)
+	syncLimEnabled := true
 	if syncLimStr := os.Getenv(envConcurrentSyncs); syncLimStr != "" {
 		syncLimParsed, err := strconv.Atoi(syncLimStr)
 		if err != nil {
@@ -115,6 +117,9 @@ func NewServer(
 		} else {
 			//nolint:gosec
 			syncLim = int32(syncLimParsed)
+			if syncLim < 0 {
+				syncLimEnabled = false
+			}
 		}
 	}
 
@@ -134,7 +139,8 @@ func NewServer(
 
 		loginFilter: newLoginFilter(),
 
-		syncLim: syncLim,
+		syncLim:        syncLim,
+		syncLimEnabled: syncLimEnabled,
 	}, nil
 }
 
@@ -212,7 +218,7 @@ func (s *Server) Job(srv proto.ManagementService_JobServer) error {
 // Sync validates the existence of a connecting peer, sends an initial state (all available for the connecting peers) and
 // notifies the connected peer of any updates (e.g. new peers under the same account)
 func (s *Server) Sync(req *proto.EncryptedMessage, srv proto.ManagementService_SyncServer) error {
-	if s.syncSem.Load() >= s.syncLim {
+	if s.syncLimEnabled && s.syncSem.Load() >= s.syncLim {
 		return status.Errorf(codes.ResourceExhausted, "too many concurrent sync requests, please try again later")
 	}
 	s.syncSem.Add(1)
@@ -294,7 +300,7 @@ func (s *Server) Sync(req *proto.EncryptedMessage, srv proto.ManagementService_S
 	metahash := metaHash(peerMeta, realIP.String())
 	s.loginFilter.addLogin(peerKey.String(), metahash)
 
-	peer, netMap, postureChecks, dnsFwdPort, err := s.accountManager.SyncAndMarkPeer(ctx, accountID, peerKey.String(), peerMeta, realIP)
+	peer, netMap, postureChecks, dnsFwdPort, err := s.accountManager.SyncAndMarkPeer(ctx, accountID, peerKey.String(), peerMeta, realIP, reqStart)
 	if err != nil {
 		log.WithContext(ctx).Debugf("error while syncing peer %s: %v", peerKey.String(), err)
 		s.syncSem.Add(-1)
@@ -305,7 +311,7 @@ func (s *Server) Sync(req *proto.EncryptedMessage, srv proto.ManagementService_S
 	if err != nil {
 		log.WithContext(ctx).Debugf("error while sending initial sync for %s: %v", peerKey.String(), err)
 		s.syncSem.Add(-1)
-		s.cancelPeerRoutines(ctx, accountID, peer)
+		s.cancelPeerRoutinesWithoutLock(ctx, accountID, peer, reqStart)
 		return err
 	}
 
@@ -313,7 +319,7 @@ func (s *Server) Sync(req *proto.EncryptedMessage, srv proto.ManagementService_S
 	if err != nil {
 		log.WithContext(ctx).Debugf("error while notify peer connected for %s: %v", peerKey.String(), err)
 		s.syncSem.Add(-1)
-		s.cancelPeerRoutines(ctx, accountID, peer)
+		s.cancelPeerRoutinesWithoutLock(ctx, accountID, peer, reqStart)
 		return err
 	}
 
@@ -330,7 +336,7 @@ func (s *Server) Sync(req *proto.EncryptedMessage, srv proto.ManagementService_S
 
 	s.syncSem.Add(-1)
 
-	return s.handleUpdates(ctx, accountID, peerKey, peer, updates, srv)
+	return s.handleUpdates(ctx, accountID, peerKey, peer, updates, srv, reqStart)
 }
 
 func (s *Server) handleHandshake(ctx context.Context, srv proto.ManagementService_JobServer) (wgtypes.Key, error) {
@@ -398,11 +404,20 @@ func (s *Server) sendJobsLoop(ctx context.Context, accountID string, peerKey wgt
 }
 
 // handleUpdates sends updates to the connected peer until the updates channel is closed.
-func (s *Server) handleUpdates(ctx context.Context, accountID string, peerKey wgtypes.Key, peer *nbpeer.Peer, updates chan *network_map.UpdateMessage, srv proto.ManagementService_SyncServer) error {
+// It implements a backpressure mechanism that sends the first update immediately,
+// then debounces subsequent rapid updates, ensuring only the latest update is sent
+// after a quiet period.
+func (s *Server) handleUpdates(ctx context.Context, accountID string, peerKey wgtypes.Key, peer *nbpeer.Peer, updates chan *network_map.UpdateMessage, srv proto.ManagementService_SyncServer, streamStartTime time.Time) error {
 	log.WithContext(ctx).Tracef("starting to handle updates for peer %s", peerKey.String())
+
+	// Create a debouncer for this peer connection
+	debouncer := NewUpdateDebouncer(1000 * time.Millisecond)
+	defer debouncer.Stop()
+
 	for {
 		select {
 		// condition when there are some updates
+		// todo set the updates channel size to 1
 		case update, open := <-updates:
 			if s.appMetrics != nil {
 				s.appMetrics.GRPCMetrics().UpdateChannelQueueLength(len(updates) + 1)
@@ -410,20 +425,38 @@ func (s *Server) handleUpdates(ctx context.Context, accountID string, peerKey wg
 
 			if !open {
 				log.WithContext(ctx).Debugf("updates channel for peer %s was closed", peerKey.String())
-				s.cancelPeerRoutines(ctx, accountID, peer)
+				s.cancelPeerRoutines(ctx, accountID, peer, streamStartTime)
 				return nil
 			}
+
 			log.WithContext(ctx).Debugf("received an update for peer %s", peerKey.String())
-			if err := s.sendUpdate(ctx, accountID, peerKey, peer, update, srv); err != nil {
-				log.WithContext(ctx).Debugf("error while sending an update to peer %s: %v", peerKey.String(), err)
-				return err
+			if debouncer.ProcessUpdate(update) {
+				// Send immediately (first update or after quiet period)
+				if err := s.sendUpdate(ctx, accountID, peerKey, peer, update, srv, streamStartTime); err != nil {
+					log.WithContext(ctx).Debugf("error while sending an update to peer %s: %v", peerKey.String(), err)
+					return err
+				}
+			}
+
+		// Timer expired - quiet period reached, send pending updates if any
+		case <-debouncer.TimerChannel():
+			pendingUpdates := debouncer.GetPendingUpdates()
+			if len(pendingUpdates) == 0 {
+				continue
+			}
+			log.WithContext(ctx).Debugf("sending %d debounced update(s) for peer %s", len(pendingUpdates), peerKey.String())
+			for _, pendingUpdate := range pendingUpdates {
+				if err := s.sendUpdate(ctx, accountID, peerKey, peer, pendingUpdate, srv, streamStartTime); err != nil {
+					log.WithContext(ctx).Debugf("error while sending an update to peer %s: %v", peerKey.String(), err)
+					return err
+				}
 			}
 
 		// condition when client <-> server connection has been terminated
 		case <-srv.Context().Done():
 			// happens when connection drops, e.g. client disconnects
 			log.WithContext(ctx).Debugf("stream of peer %s has been closed", peerKey.String())
-			s.cancelPeerRoutines(ctx, accountID, peer)
+			s.cancelPeerRoutines(ctx, accountID, peer, streamStartTime)
 			return srv.Context().Err()
 		}
 	}
@@ -431,16 +464,16 @@ func (s *Server) handleUpdates(ctx context.Context, accountID string, peerKey wg
 
 // 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 *Server) sendUpdate(ctx context.Context, accountID string, peerKey wgtypes.Key, peer *nbpeer.Peer, update *network_map.UpdateMessage, srv proto.ManagementService_SyncServer) error {
+func (s *Server) sendUpdate(ctx context.Context, accountID string, peerKey wgtypes.Key, peer *nbpeer.Peer, update *network_map.UpdateMessage, srv proto.ManagementService_SyncServer, streamStartTime time.Time) error {
 	key, err := s.secretsManager.GetWGKey()
 	if err != nil {
-		s.cancelPeerRoutines(ctx, accountID, peer)
+		s.cancelPeerRoutines(ctx, accountID, peer, streamStartTime)
 		return status.Errorf(codes.Internal, "failed processing update message")
 	}
 
 	encryptedResp, err := encryption.EncryptMessage(peerKey, key, update.Update)
 	if err != nil {
-		s.cancelPeerRoutines(ctx, accountID, peer)
+		s.cancelPeerRoutines(ctx, accountID, peer, streamStartTime)
 		return status.Errorf(codes.Internal, "failed processing update message")
 	}
 	err = srv.Send(&proto.EncryptedMessage{
@@ -448,7 +481,7 @@ func (s *Server) sendUpdate(ctx context.Context, accountID string, peerKey wgtyp
 		Body:     encryptedResp,
 	})
 	if err != nil {
-		s.cancelPeerRoutines(ctx, accountID, peer)
+		s.cancelPeerRoutines(ctx, accountID, peer, streamStartTime)
 		return status.Errorf(codes.Internal, "failed sending update message")
 	}
 	log.WithContext(ctx).Debugf("sent an update to peer %s", peerKey.String())
@@ -480,11 +513,15 @@ func (s *Server) sendJob(ctx context.Context, peerKey wgtypes.Key, job *job.Even
 	return nil
 }
 
-func (s *Server) cancelPeerRoutines(ctx context.Context, accountID string, peer *nbpeer.Peer) {
+func (s *Server) cancelPeerRoutines(ctx context.Context, accountID string, peer *nbpeer.Peer, streamStartTime time.Time) {
 	unlock := s.acquirePeerLockByUID(ctx, peer.Key)
 	defer unlock()
 
-	err := s.accountManager.OnPeerDisconnected(ctx, accountID, peer.Key)
+	s.cancelPeerRoutinesWithoutLock(ctx, accountID, peer, streamStartTime)
+}
+
+func (s *Server) cancelPeerRoutinesWithoutLock(ctx context.Context, accountID string, peer *nbpeer.Peer, streamStartTime time.Time) {
+	err := s.accountManager.OnPeerDisconnected(ctx, accountID, peer.Key, streamStartTime)
 	if err != nil {
 		log.WithContext(ctx).Errorf("failed to disconnect peer %s properly: %v", peer.Key, err)
 	}

management/internals/shared/grpc/token_mgr.go

@@ -242,7 +242,10 @@ func (m *TimeBasedAuthSecretsManager) pushNewTURNAndRelayTokens(ctx context.Cont
 	m.extendNetbirdConfig(ctx, peerID, accountID, update)
 
 	log.WithContext(ctx).Debugf("sending new TURN credentials to peer %s", peerID)
-	m.updateManager.SendUpdate(ctx, peerID, &network_map.UpdateMessage{Update: update})
+	m.updateManager.SendUpdate(ctx, peerID, &network_map.UpdateMessage{
+		Update:      update,
+		MessageType: network_map.MessageTypeControlConfig,
+	})
 }
 
 func (m *TimeBasedAuthSecretsManager) pushNewRelayTokens(ctx context.Context, accountID, peerID string) {
@@ -266,7 +269,10 @@ func (m *TimeBasedAuthSecretsManager) pushNewRelayTokens(ctx context.Context, ac
 	m.extendNetbirdConfig(ctx, peerID, accountID, update)
 
 	log.WithContext(ctx).Debugf("sending new relay credentials to peer %s", peerID)
-	m.updateManager.SendUpdate(ctx, peerID, &network_map.UpdateMessage{Update: update})
+	m.updateManager.SendUpdate(ctx, peerID, &network_map.UpdateMessage{
+		Update:      update,
+		MessageType: network_map.MessageTypeControlConfig,
+	})
 }
 
 func (m *TimeBasedAuthSecretsManager) extendNetbirdConfig(ctx context.Context, peerID, accountID string, update *proto.SyncResponse) {

management/internals/shared/grpc/update_debouncer.go

@@ -0,0 +1,103 @@
+package grpc
+
+import (
+	"time"
+
+	"github.com/netbirdio/netbird/management/internals/controllers/network_map"
+)
+
+// UpdateDebouncer implements a backpressure mechanism that:
+// - Sends the first update immediately
+// - Coalesces rapid subsequent network map updates (only latest matters)
+// - Queues control/config updates (all must be delivered)
+// - Preserves the order of messages (important for control configs between network maps)
+// - Ensures pending updates are sent after a quiet period
+type UpdateDebouncer struct {
+	debounceInterval time.Duration
+	timer            *time.Timer
+	pendingUpdates   []*network_map.UpdateMessage // Queue that preserves order
+	timerC           <-chan time.Time
+}
+
+// NewUpdateDebouncer creates a new debouncer with the specified interval
+func NewUpdateDebouncer(interval time.Duration) *UpdateDebouncer {
+	return &UpdateDebouncer{
+		debounceInterval: interval,
+	}
+}
+
+// ProcessUpdate handles an incoming update and returns whether it should be sent immediately
+func (d *UpdateDebouncer) ProcessUpdate(update *network_map.UpdateMessage) bool {
+	if d.timer == nil {
+		// No active debounce timer, signal to send immediately
+		// and start the debounce period
+		d.startTimer()
+		return true
+	}
+
+	// Already in debounce period, accumulate this update preserving order
+	// Check if we should coalesce with the last pending update
+	if len(d.pendingUpdates) > 0 &&
+		update.MessageType == network_map.MessageTypeNetworkMap &&
+		d.pendingUpdates[len(d.pendingUpdates)-1].MessageType == network_map.MessageTypeNetworkMap {
+		// Replace the last network map with this one (coalesce consecutive network maps)
+		d.pendingUpdates[len(d.pendingUpdates)-1] = update
+	} else {
+		// Append to the queue (preserves order for control configs and non-consecutive network maps)
+		d.pendingUpdates = append(d.pendingUpdates, update)
+	}
+	d.resetTimer()
+	return false
+}
+
+// TimerChannel returns the timer channel for select statements
+func (d *UpdateDebouncer) TimerChannel() <-chan time.Time {
+	if d.timer == nil {
+		return nil
+	}
+	return d.timerC
+}
+
+// GetPendingUpdates returns and clears all pending updates after timer expiration.
+// Updates are returned in the order they were received, with consecutive network maps
+// already coalesced to only the latest one.
+// If there were pending updates, it restarts the timer to continue debouncing.
+// If there were no pending updates, it clears the timer (true quiet period).
+func (d *UpdateDebouncer) GetPendingUpdates() []*network_map.UpdateMessage {
+	updates := d.pendingUpdates
+	d.pendingUpdates = nil
+
+	if len(updates) > 0 {
+		// There were pending updates, so updates are still coming rapidly
+		// Restart the timer to continue debouncing mode
+		if d.timer != nil {
+			d.timer.Reset(d.debounceInterval)
+		}
+	} else {
+		// No pending updates means true quiet period - return to immediate mode
+		d.timer = nil
+		d.timerC = nil
+	}
+
+	return updates
+}
+
+// Stop stops the debouncer and cleans up resources
+func (d *UpdateDebouncer) Stop() {
+	if d.timer != nil {
+		d.timer.Stop()
+		d.timer = nil
+		d.timerC = nil
+	}
+	d.pendingUpdates = nil
+}
+
+func (d *UpdateDebouncer) startTimer() {
+	d.timer = time.NewTimer(d.debounceInterval)
+	d.timerC = d.timer.C
+}
+
+func (d *UpdateDebouncer) resetTimer() {
+	d.timer.Stop()
+	d.timer.Reset(d.debounceInterval)
+}

management/internals/shared/grpc/update_debouncer_test.go

@@ -0,0 +1,587 @@
+package grpc
+
+import (
+	"testing"
+	"time"
+
+	"github.com/netbirdio/netbird/management/internals/controllers/network_map"
+	"github.com/netbirdio/netbird/shared/management/proto"
+)
+
+func TestUpdateDebouncer_FirstUpdateSentImmediately(t *testing.T) {
+	debouncer := NewUpdateDebouncer(50 * time.Millisecond)
+	defer debouncer.Stop()
+
+	update := &network_map.UpdateMessage{
+		Update:      &proto.SyncResponse{},
+		MessageType: network_map.MessageTypeNetworkMap,
+	}
+
+	shouldSend := debouncer.ProcessUpdate(update)
+
+	if !shouldSend {
+		t.Error("First update should be sent immediately")
+	}
+
+	if debouncer.TimerChannel() == nil {
+		t.Error("Timer should be started after first update")
+	}
+}
+
+func TestUpdateDebouncer_RapidUpdatesCoalesced(t *testing.T) {
+	debouncer := NewUpdateDebouncer(50 * time.Millisecond)
+	defer debouncer.Stop()
+
+	update1 := &network_map.UpdateMessage{
+		Update:      &proto.SyncResponse{},
+		MessageType: network_map.MessageTypeNetworkMap,
+	}
+	update2 := &network_map.UpdateMessage{
+		Update:      &proto.SyncResponse{},
+		MessageType: network_map.MessageTypeNetworkMap,
+	}
+	update3 := &network_map.UpdateMessage{
+		Update:      &proto.SyncResponse{},
+		MessageType: network_map.MessageTypeNetworkMap,
+	}
+
+	// First update should be sent immediately
+	if !debouncer.ProcessUpdate(update1) {
+		t.Error("First update should be sent immediately")
+	}
+
+	// Rapid subsequent updates should be coalesced
+	if debouncer.ProcessUpdate(update2) {
+		t.Error("Second rapid update should not be sent immediately")
+	}
+
+	if debouncer.ProcessUpdate(update3) {
+		t.Error("Third rapid update should not be sent immediately")
+	}
+
+	// Wait for debounce period
+	select {
+	case <-debouncer.TimerChannel():
+		pendingUpdates := debouncer.GetPendingUpdates()
+		if len(pendingUpdates) != 1 {
+			t.Errorf("Should get exactly 1 pending update, got %d", len(pendingUpdates))
+		}
+		if pendingUpdates[0] != update3 {
+			t.Error("Should get the last update (update3)")
+		}
+	case <-time.After(100 * time.Millisecond):
+		t.Error("Timer should have fired")
+	}
+}
+
+func TestUpdateDebouncer_LastUpdateAlwaysSent(t *testing.T) {
+	debouncer := NewUpdateDebouncer(30 * time.Millisecond)
+	defer debouncer.Stop()
+
+	update1 := &network_map.UpdateMessage{
+		Update:      &proto.SyncResponse{},
+		MessageType: network_map.MessageTypeNetworkMap,
+	}
+	update2 := &network_map.UpdateMessage{
+		Update:      &proto.SyncResponse{},
+		MessageType: network_map.MessageTypeNetworkMap,
+	}
+
+	// Send first update
+	debouncer.ProcessUpdate(update1)
+
+	// Send second update within debounce period
+	debouncer.ProcessUpdate(update2)
+
+	// Wait for timer
+	select {
+	case <-debouncer.TimerChannel():
+		pendingUpdates := debouncer.GetPendingUpdates()
+		if len(pendingUpdates) != 1 {
+			t.Errorf("Should get exactly 1 pending update, got %d", len(pendingUpdates))
+		}
+		if pendingUpdates[0] != update2 {
+			t.Error("Should get the last update")
+		}
+		if pendingUpdates[0] == update1 {
+			t.Error("Should not get the first update")
+		}
+	case <-time.After(100 * time.Millisecond):
+		t.Error("Timer should have fired")
+	}
+}
+
+func TestUpdateDebouncer_TimerResetOnNewUpdate(t *testing.T) {
+	debouncer := NewUpdateDebouncer(50 * time.Millisecond)
+	defer debouncer.Stop()
+
+	update1 := &network_map.UpdateMessage{
+		Update:      &proto.SyncResponse{},
+		MessageType: network_map.MessageTypeNetworkMap,
+	}
+	update2 := &network_map.UpdateMessage{
+		Update:      &proto.SyncResponse{},
+		MessageType: network_map.MessageTypeNetworkMap,
+	}
+	update3 := &network_map.UpdateMessage{
+		Update:      &proto.SyncResponse{},
+		MessageType: network_map.MessageTypeNetworkMap,
+	}
+
+	// Send first update
+	debouncer.ProcessUpdate(update1)
+
+	// Wait a bit, but not the full debounce period
+	time.Sleep(30 * time.Millisecond)
+
+	// Send second update - should reset timer
+	debouncer.ProcessUpdate(update2)
+
+	// Wait a bit more
+	time.Sleep(30 * time.Millisecond)
+
+	// Send third update - should reset timer again
+	debouncer.ProcessUpdate(update3)
+
+	// Now wait for the timer (should fire after last update's reset)
+	select {
+	case <-debouncer.TimerChannel():
+		pendingUpdates := debouncer.GetPendingUpdates()
+		if len(pendingUpdates) != 1 {
+			t.Errorf("Should get exactly 1 pending update, got %d", len(pendingUpdates))
+		}
+		if pendingUpdates[0] != update3 {
+			t.Error("Should get the last update (update3)")
+		}
+		// Timer should be restarted since there was a pending update
+		if debouncer.TimerChannel() == nil {
+			t.Error("Timer should be restarted after sending pending update")
+		}
+	case <-time.After(150 * time.Millisecond):
+		t.Error("Timer should have fired")
+	}
+}
+
+func TestUpdateDebouncer_TimerRestartsAfterPendingUpdateSent(t *testing.T) {
+	debouncer := NewUpdateDebouncer(30 * time.Millisecond)
+	defer debouncer.Stop()
+
+	update1 := &network_map.UpdateMessage{
+		Update:      &proto.SyncResponse{},
+		MessageType: network_map.MessageTypeNetworkMap,
+	}
+	update2 := &network_map.UpdateMessage{
+		Update:      &proto.SyncResponse{},
+		MessageType: network_map.MessageTypeNetworkMap,
+	}
+	update3 := &network_map.UpdateMessage{
+		Update:      &proto.SyncResponse{},
+		MessageType: network_map.MessageTypeNetworkMap,
+	}
+
+	// First update sent immediately
+	debouncer.ProcessUpdate(update1)
+
+	// Second update coalesced
+	debouncer.ProcessUpdate(update2)
+
+	// Wait for timer to expire
+	select {
+	case <-debouncer.TimerChannel():
+		pendingUpdates := debouncer.GetPendingUpdates()
+
+		if len(pendingUpdates) == 0 {
+			t.Fatal("Should have pending update")
+		}
+
+		// After sending pending update, timer is restarted, so next update is NOT immediate
+		if debouncer.ProcessUpdate(update3) {
+			t.Error("Update after debounced send should not be sent immediately (timer restarted)")
+		}
+
+		// Wait for the restarted timer and verify update3 is pending
+		select {
+		case <-debouncer.TimerChannel():
+			finalUpdates := debouncer.GetPendingUpdates()
+			if len(finalUpdates) != 1 || finalUpdates[0] != update3 {
+				t.Error("Should get update3 as pending")
+			}
+		case <-time.After(100 * time.Millisecond):
+			t.Error("Timer should have fired for restarted timer")
+		}
+	case <-time.After(100 * time.Millisecond):
+		t.Error("Timer should have fired")
+	}
+}
+
+func TestUpdateDebouncer_StopCleansUp(t *testing.T) {
+	debouncer := NewUpdateDebouncer(50 * time.Millisecond)
+
+	update := &network_map.UpdateMessage{
+		Update:      &proto.SyncResponse{},
+		MessageType: network_map.MessageTypeNetworkMap,
+	}
+
+	// Send update to start timer
+	debouncer.ProcessUpdate(update)
+
+	// Stop should clean up
+	debouncer.Stop()
+
+	// Multiple stops should be safe
+	debouncer.Stop()
+}
+
+func TestUpdateDebouncer_HighFrequencyUpdates(t *testing.T) {
+	debouncer := NewUpdateDebouncer(50 * time.Millisecond)
+	defer debouncer.Stop()
+
+	// Simulate high-frequency updates
+	var lastUpdate *network_map.UpdateMessage
+	sentImmediately := 0
+	for i := 0; i < 100; i++ {
+		update := &network_map.UpdateMessage{
+			Update: &proto.SyncResponse{
+				NetworkMap: &proto.NetworkMap{
+					Serial: uint64(i),
+				},
+			},
+			MessageType: network_map.MessageTypeNetworkMap,
+		}
+		lastUpdate = update
+		if debouncer.ProcessUpdate(update) {
+			sentImmediately++
+		}
+		time.Sleep(1 * time.Millisecond) // Very rapid updates
+	}
+
+	// Only first update should be sent immediately
+	if sentImmediately != 1 {
+		t.Errorf("Expected only 1 update sent immediately, got %d", sentImmediately)
+	}
+
+	// Wait for debounce period
+	select {
+	case <-debouncer.TimerChannel():
+		pendingUpdates := debouncer.GetPendingUpdates()
+		if len(pendingUpdates) != 1 {
+			t.Errorf("Should get exactly 1 pending update, got %d", len(pendingUpdates))
+		}
+		if pendingUpdates[0] != lastUpdate {
+			t.Error("Should get the very last update")
+		}
+		if pendingUpdates[0].Update.NetworkMap.Serial != 99 {
+			t.Errorf("Expected serial 99, got %d", pendingUpdates[0].Update.NetworkMap.Serial)
+		}
+	case <-time.After(200 * time.Millisecond):
+		t.Error("Timer should have fired")
+	}
+}
+
+func TestUpdateDebouncer_NoUpdatesAfterFirst(t *testing.T) {
+	debouncer := NewUpdateDebouncer(30 * time.Millisecond)
+	defer debouncer.Stop()
+
+	update := &network_map.UpdateMessage{
+		Update:      &proto.SyncResponse{},
+		MessageType: network_map.MessageTypeNetworkMap,
+	}
+
+	// Send first update
+	if !debouncer.ProcessUpdate(update) {
+		t.Error("First update should be sent immediately")
+	}
+
+	// Wait for timer to expire with no additional updates (true quiet period)
+	select {
+	case <-debouncer.TimerChannel():
+		pendingUpdates := debouncer.GetPendingUpdates()
+		if len(pendingUpdates) != 0 {
+			t.Error("Should have no pending updates")
+		}
+		// After true quiet period, timer should be cleared
+		if debouncer.TimerChannel() != nil {
+			t.Error("Timer should be cleared after quiet period")
+		}
+	case <-time.After(100 * time.Millisecond):
+		t.Error("Timer should have fired")
+	}
+}
+
+func TestUpdateDebouncer_IntermediateUpdatesDropped(t *testing.T) {
+	debouncer := NewUpdateDebouncer(50 * time.Millisecond)
+	defer debouncer.Stop()
+
+	updates := make([]*network_map.UpdateMessage, 5)
+	for i := range updates {
+		updates[i] = &network_map.UpdateMessage{
+			Update: &proto.SyncResponse{
+				NetworkMap: &proto.NetworkMap{
+					Serial: uint64(i),
+				},
+			},
+			MessageType: network_map.MessageTypeNetworkMap,
+		}
+	}
+
+	// First update sent immediately
+	debouncer.ProcessUpdate(updates[0])
+
+	// Send updates 1, 2, 3, 4 rapidly - only last one should remain pending
+	debouncer.ProcessUpdate(updates[1])
+	debouncer.ProcessUpdate(updates[2])
+	debouncer.ProcessUpdate(updates[3])
+	debouncer.ProcessUpdate(updates[4])
+
+	// Wait for debounce
+	<-debouncer.TimerChannel()
+	pendingUpdates := debouncer.GetPendingUpdates()
+
+	if len(pendingUpdates) != 1 {
+		t.Errorf("Should get exactly 1 pending update, got %d", len(pendingUpdates))
+	}
+	if pendingUpdates[0].Update.NetworkMap.Serial != 4 {
+		t.Errorf("Expected only the last update (serial 4), got serial %d", pendingUpdates[0].Update.NetworkMap.Serial)
+	}
+}
+
+func TestUpdateDebouncer_TrueQuietPeriodResetsToImmediateMode(t *testing.T) {
+	debouncer := NewUpdateDebouncer(30 * time.Millisecond)
+	defer debouncer.Stop()
+
+	update1 := &network_map.UpdateMessage{
+		Update:      &proto.SyncResponse{},
+		MessageType: network_map.MessageTypeNetworkMap,
+	}
+	update2 := &network_map.UpdateMessage{
+		Update:      &proto.SyncResponse{},
+		MessageType: network_map.MessageTypeNetworkMap,
+	}
+
+	// First update sent immediately
+	if !debouncer.ProcessUpdate(update1) {
+		t.Error("First update should be sent immediately")
+	}
+
+	// Wait for timer without sending any more updates (true quiet period)
+	<-debouncer.TimerChannel()
+	pendingUpdates := debouncer.GetPendingUpdates()
+
+	if len(pendingUpdates) != 0 {
+		t.Error("Should have no pending updates during quiet period")
+	}
+
+	// After true quiet period, next update should be sent immediately
+	if !debouncer.ProcessUpdate(update2) {
+		t.Error("Update after true quiet period should be sent immediately")
+	}
+}
+
+func TestUpdateDebouncer_ContinuousHighFrequencyStaysInDebounceMode(t *testing.T) {
+	debouncer := NewUpdateDebouncer(50 * time.Millisecond)
+	defer debouncer.Stop()
+
+	// Simulate continuous high-frequency updates
+	for i := 0; i < 10; i++ {
+		update := &network_map.UpdateMessage{
+			Update: &proto.SyncResponse{
+				NetworkMap: &proto.NetworkMap{
+					Serial: uint64(i),
+				},
+			},
+			MessageType: network_map.MessageTypeNetworkMap,
+		}
+
+		if i == 0 {
+			// First one sent immediately
+			if !debouncer.ProcessUpdate(update) {
+				t.Error("First update should be sent immediately")
+			}
+		} else {
+			// All others should be coalesced (not sent immediately)
+			if debouncer.ProcessUpdate(update) {
+				t.Errorf("Update %d should not be sent immediately", i)
+			}
+		}
+
+		// Wait a bit but send next update before debounce expires
+		time.Sleep(20 * time.Millisecond)
+	}
+
+	// Now wait for final debounce
+	select {
+	case <-debouncer.TimerChannel():
+		pendingUpdates := debouncer.GetPendingUpdates()
+		if len(pendingUpdates) == 0 {
+			t.Fatal("Should have the last update pending")
+		}
+		if pendingUpdates[0].Update.NetworkMap.Serial != 9 {
+			t.Errorf("Expected serial 9, got %d", pendingUpdates[0].Update.NetworkMap.Serial)
+		}
+	case <-time.After(200 * time.Millisecond):
+		t.Error("Timer should have fired")
+	}
+}
+
+func TestUpdateDebouncer_ControlConfigMessagesQueued(t *testing.T) {
+	debouncer := NewUpdateDebouncer(50 * time.Millisecond)
+	defer debouncer.Stop()
+
+	netmapUpdate := &network_map.UpdateMessage{
+		Update:      &proto.SyncResponse{NetworkMap: &proto.NetworkMap{Serial: 1}},
+		MessageType: network_map.MessageTypeNetworkMap,
+	}
+	tokenUpdate1 := &network_map.UpdateMessage{
+		Update:      &proto.SyncResponse{NetbirdConfig: &proto.NetbirdConfig{}},
+		MessageType: network_map.MessageTypeControlConfig,
+	}
+	tokenUpdate2 := &network_map.UpdateMessage{
+		Update:      &proto.SyncResponse{NetbirdConfig: &proto.NetbirdConfig{}},
+		MessageType: network_map.MessageTypeControlConfig,
+	}
+
+	// First update sent immediately
+	debouncer.ProcessUpdate(netmapUpdate)
+
+	// Send multiple control config updates - they should all be queued
+	debouncer.ProcessUpdate(tokenUpdate1)
+	debouncer.ProcessUpdate(tokenUpdate2)
+
+	// Wait for debounce period
+	select {
+	case <-debouncer.TimerChannel():
+		pendingUpdates := debouncer.GetPendingUpdates()
+		// Should get both control config updates
+		if len(pendingUpdates) != 2 {
+			t.Errorf("Expected 2 control config updates, got %d", len(pendingUpdates))
+		}
+		// Control configs should come first
+		if pendingUpdates[0] != tokenUpdate1 {
+			t.Error("First pending update should be tokenUpdate1")
+		}
+		if pendingUpdates[1] != tokenUpdate2 {
+			t.Error("Second pending update should be tokenUpdate2")
+		}
+	case <-time.After(200 * time.Millisecond):
+		t.Error("Timer should have fired")
+	}
+}
+
+func TestUpdateDebouncer_MixedMessageTypes(t *testing.T) {
+	debouncer := NewUpdateDebouncer(50 * time.Millisecond)
+	defer debouncer.Stop()
+
+	netmapUpdate1 := &network_map.UpdateMessage{
+		Update:      &proto.SyncResponse{NetworkMap: &proto.NetworkMap{Serial: 1}},
+		MessageType: network_map.MessageTypeNetworkMap,
+	}
+	netmapUpdate2 := &network_map.UpdateMessage{
+		Update:      &proto.SyncResponse{NetworkMap: &proto.NetworkMap{Serial: 2}},
+		MessageType: network_map.MessageTypeNetworkMap,
+	}
+	tokenUpdate := &network_map.UpdateMessage{
+		Update:      &proto.SyncResponse{NetbirdConfig: &proto.NetbirdConfig{}},
+		MessageType: network_map.MessageTypeControlConfig,
+	}
+
+	// First update sent immediately
+	debouncer.ProcessUpdate(netmapUpdate1)
+
+	// Send token update and network map update
+	debouncer.ProcessUpdate(tokenUpdate)
+	debouncer.ProcessUpdate(netmapUpdate2)
+
+	// Wait for debounce period
+	select {
+	case <-debouncer.TimerChannel():
+		pendingUpdates := debouncer.GetPendingUpdates()
+		// Should get 2 updates in order: token, then network map
+		if len(pendingUpdates) != 2 {
+			t.Errorf("Expected 2 pending updates, got %d", len(pendingUpdates))
+		}
+		// Token update should come first (preserves order)
+		if pendingUpdates[0] != tokenUpdate {
+			t.Error("First pending update should be tokenUpdate")
+		}
+		// Network map update should come second
+		if pendingUpdates[1] != netmapUpdate2 {
+			t.Error("Second pending update should be netmapUpdate2")
+		}
+	case <-time.After(200 * time.Millisecond):
+		t.Error("Timer should have fired")
+	}
+}
+
+func TestUpdateDebouncer_OrderPreservation(t *testing.T) {
+	debouncer := NewUpdateDebouncer(50 * time.Millisecond)
+	defer debouncer.Stop()
+
+	// Simulate: 50 network maps -> 1 control config -> 50 network maps
+	// Expected result: 3 messages (netmap, controlConfig, netmap)
+
+	// Send first network map immediately
+	firstNetmap := &network_map.UpdateMessage{
+		Update:      &proto.SyncResponse{NetworkMap: &proto.NetworkMap{Serial: 0}},
+		MessageType: network_map.MessageTypeNetworkMap,
+	}
+	if !debouncer.ProcessUpdate(firstNetmap) {
+		t.Error("First update should be sent immediately")
+	}
+
+	// Send 49 more network maps (will be coalesced to last one)
+	var lastNetmapBatch1 *network_map.UpdateMessage
+	for i := 1; i < 50; i++ {
+		lastNetmapBatch1 = &network_map.UpdateMessage{
+			Update:      &proto.SyncResponse{NetworkMap: &proto.NetworkMap{Serial: uint64(i)}},
+			MessageType: network_map.MessageTypeNetworkMap,
+		}
+		debouncer.ProcessUpdate(lastNetmapBatch1)
+	}
+
+	// Send 1 control config
+	controlConfig := &network_map.UpdateMessage{
+		Update:      &proto.SyncResponse{NetbirdConfig: &proto.NetbirdConfig{}},
+		MessageType: network_map.MessageTypeControlConfig,
+	}
+	debouncer.ProcessUpdate(controlConfig)
+
+	// Send 50 more network maps (will be coalesced to last one)
+	var lastNetmapBatch2 *network_map.UpdateMessage
+	for i := 50; i < 100; i++ {
+		lastNetmapBatch2 = &network_map.UpdateMessage{
+			Update:      &proto.SyncResponse{NetworkMap: &proto.NetworkMap{Serial: uint64(i)}},
+			MessageType: network_map.MessageTypeNetworkMap,
+		}
+		debouncer.ProcessUpdate(lastNetmapBatch2)
+	}
+
+	// Wait for debounce period
+	select {
+	case <-debouncer.TimerChannel():
+		pendingUpdates := debouncer.GetPendingUpdates()
+		// Should get exactly 3 updates: netmap, controlConfig, netmap
+		if len(pendingUpdates) != 3 {
+			t.Errorf("Expected 3 pending updates, got %d", len(pendingUpdates))
+		}
+		// First should be the last netmap from batch 1
+		if pendingUpdates[0] != lastNetmapBatch1 {
+			t.Error("First pending update should be last netmap from batch 1")
+		}
+		if pendingUpdates[0].Update.NetworkMap.Serial != 49 {
+			t.Errorf("Expected serial 49, got %d", pendingUpdates[0].Update.NetworkMap.Serial)
+		}
+		// Second should be the control config
+		if pendingUpdates[1] != controlConfig {
+			t.Error("Second pending update should be control config")
+		}
+		// Third should be the last netmap from batch 2
+		if pendingUpdates[2] != lastNetmapBatch2 {
+			t.Error("Third pending update should be last netmap from batch 2")
+		}
+		if pendingUpdates[2].Update.NetworkMap.Serial != 99 {
+			t.Errorf("Expected serial 99, got %d", pendingUpdates[2].Update.NetworkMap.Serial)
+		}
+	case <-time.After(200 * time.Millisecond):
+		t.Error("Timer should have fired")
+	}
+}