getValidatedPeerWithMap get account from cache for exp

client/grpc/dialer.go

@@ -4,15 +4,12 @@ import (
 	"context"
 	"crypto/tls"
 	"crypto/x509"
-	"errors"
-	"fmt"
 	"runtime"
 	"time"
 
 	"github.com/cenkalti/backoff/v4"
 	log "github.com/sirupsen/logrus"
 	"google.golang.org/grpc"
-	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/keepalive"
@@ -20,9 +17,6 @@ import (
 	"github.com/netbirdio/netbird/util/embeddedroots"
 )
 
-// ErrConnectionShutdown indicates that the connection entered shutdown state before becoming ready
-var ErrConnectionShutdown = errors.New("connection shutdown before ready")
-
 // Backoff returns a backoff configuration for gRPC calls
 func Backoff(ctx context.Context) backoff.BackOff {
 	b := backoff.NewExponentialBackOff()
@@ -31,26 +25,6 @@ func Backoff(ctx context.Context) backoff.BackOff {
 	return backoff.WithContext(b, ctx)
 }
 
-// waitForConnectionReady blocks until the connection becomes ready or fails.
-// Returns an error if the connection times out, is cancelled, or enters shutdown state.
-func waitForConnectionReady(ctx context.Context, conn *grpc.ClientConn) error {
-	conn.Connect()
-
-	state := conn.GetState()
-	for state != connectivity.Ready && state != connectivity.Shutdown {
-		if !conn.WaitForStateChange(ctx, state) {
-			return fmt.Errorf("wait state change from %s: %w", state, ctx.Err())
-		}
-		state = conn.GetState()
-	}
-
-	if state == connectivity.Shutdown {
-		return ErrConnectionShutdown
-	}
-
-	return nil
-}
-
 // CreateConnection creates a gRPC client connection with the appropriate transport options.
 // The component parameter specifies the WebSocket proxy component path (e.g., "/management", "/signal").
 func CreateConnection(ctx context.Context, addr string, tlsEnabled bool, component string) (*grpc.ClientConn, error) {
@@ -68,24 +42,22 @@ func CreateConnection(ctx context.Context, addr string, tlsEnabled bool, compone
 		}))
 	}
 
-	conn, err := grpc.NewClient(
+	connCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
+	defer cancel()
+
+	conn, err := grpc.DialContext(
+		connCtx,
 		addr,
 		transportOption,
 		WithCustomDialer(tlsEnabled, component),
+		grpc.WithBlock(),
 		grpc.WithKeepaliveParams(keepalive.ClientParameters{
 			Time:    30 * time.Second,
 			Timeout: 10 * time.Second,
 		}),
 	)
 	if err != nil {
-		return nil, fmt.Errorf("new client: %w", err)
-	}
-
-	ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
-	defer cancel()
-
-	if err := waitForConnectionReady(ctx, conn); err != nil {
-		_ = conn.Close()
+		log.Printf("DialContext error: %v", err)
 		return nil, err
 	}
 

client/grpc/dialer_generic.go

@@ -18,7 +18,7 @@ import (
 	nbnet "github.com/netbirdio/netbird/client/net"
 )
 
-func WithCustomDialer(_ bool, _ string) grpc.DialOption {
+func WithCustomDialer(tlsEnabled bool, component string) grpc.DialOption {
 	return grpc.WithContextDialer(func(ctx context.Context, addr string) (net.Conn, error) {
 		if runtime.GOOS == "linux" {
 			currentUser, err := user.Current()
@@ -36,6 +36,7 @@ func WithCustomDialer(_ bool, _ string) grpc.DialOption {
 
 		conn, err := nbnet.NewDialer().DialContext(ctx, "tcp", addr)
 		if err != nil {
+			log.Errorf("Failed to dial: %s", err)
 			return nil, fmt.Errorf("nbnet.NewDialer().DialContext: %w", err)
 		}
 		return conn, nil

client/net/conn.go

@@ -17,7 +17,8 @@ type Conn struct {
 	ID hooks.ConnectionID
 }
 
-// Close overrides the net.Conn Close method to execute all registered hooks after closing the connection.
+// Close overrides the net.Conn Close method to execute all registered hooks after closing the connection
+// Close overrides the net.Conn Close method to execute all registered hooks before closing the connection.
 func (c *Conn) Close() error {
 	return closeConn(c.ID, c.Conn)
 }
@@ -28,7 +29,7 @@ type TCPConn struct {
 	ID hooks.ConnectionID
 }
 
-// Close overrides the net.TCPConn Close method to execute all registered hooks after closing the connection.
+// Close overrides the net.TCPConn Close method to execute all registered hooks before closing the connection.
 func (c *TCPConn) Close() error {
 	return closeConn(c.ID, c.TCPConn)
 }
@@ -36,16 +37,13 @@ func (c *TCPConn) Close() error {
 // closeConn is a helper function to close connections and execute close hooks.
 func closeConn(id hooks.ConnectionID, conn io.Closer) error {
 	err := conn.Close()
-	cleanupConnID(id)
-	return err
-}
 
-// cleanupConnID executes close hooks for a connection ID.
-func cleanupConnID(id hooks.ConnectionID) {
 	closeHooks := hooks.GetCloseHooks()
 	for _, hook := range closeHooks {
 		if err := hook(id); err != nil {
 			log.Errorf("Error executing close hook: %v", err)
 		}
 	}
+
+	return err
 }

client/net/dial.go

@@ -74,6 +74,7 @@ func DialTCP(network string, laddr, raddr *net.TCPAddr) (transport.TCPConn, erro
 		}
 		return &TCPConn{TCPConn: tcpConn, ID: c.ID}, nil
 	}
+
 	if err := conn.Close(); err != nil {
 		log.Errorf("failed to close connection: %v", err)
 	}

client/net/dialer_dial.go

@@ -30,7 +30,6 @@ func (d *Dialer) DialContext(ctx context.Context, network, address string) (net.
 
 	conn, err := d.Dialer.DialContext(ctx, network, address)
 	if err != nil {
-		cleanupConnID(connID)
 		return nil, fmt.Errorf("d.Dialer.DialContext: %w", err)
 	}
 
@@ -65,7 +64,7 @@ func callDialerHooks(ctx context.Context, connID hooks.ConnectionID, address str
 
 	ips, err := resolver.LookupIPAddr(ctx, host)
 	if err != nil {
-		return fmt.Errorf("resolve address %s: %w", address, err)
+		return fmt.Errorf("failed to resolve address %s: %w", address, err)
 	}
 
 	log.Debugf("Dialer resolved IPs for %s: %v", address, ips)

client/net/listener_listen.go

@@ -48,7 +48,7 @@ func (c *PacketConn) WriteTo(b []byte, addr net.Addr) (n int, err error) {
 	return c.PacketConn.WriteTo(b, addr)
 }
 
-// Close overrides the net.PacketConn Close method to execute all registered hooks after closing the connection.
+// Close overrides the net.PacketConn Close method to execute all registered hooks before closing the connection.
 func (c *PacketConn) Close() error {
 	defer c.seenAddrs.Clear()
 	return closeConn(c.ID, c.PacketConn)
@@ -69,7 +69,7 @@ func (c *UDPConn) WriteTo(b []byte, addr net.Addr) (n int, err error) {
 	return c.UDPConn.WriteTo(b, addr)
 }
 
-// Close overrides the net.UDPConn Close method to execute all registered hooks after closing the connection.
+// Close overrides the net.UDPConn Close method to execute all registered hooks before closing the connection.
 func (c *UDPConn) Close() error {
 	defer c.seenAddrs.Clear()
 	return closeConn(c.ID, c.UDPConn)

client/ui/client_ui.go

@@ -833,7 +833,6 @@ func (s *serviceClient) onTrayReady() {
 
 	newProfileMenuArgs := &newProfileMenuArgs{
 		ctx:                  s.ctx,
-		serviceClient:        s,
 		profileManager:       s.profileManager,
 		eventHandler:         s.eventHandler,
 		profileMenuItem:      profileMenuItem,

management/main.go

@@ -11,7 +11,7 @@ import (
 
 func main() {
 	go func() {
-		log.Println(http.ListenAndServe(":6060", nil))
+		log.Println(http.ListenAndServe("localhost:6060", nil))
 	}()
 	if err := cmd.Execute(); err != nil {
 		os.Exit(1)

management/server/account_test.go

@@ -1190,7 +1190,7 @@ func testAccountManager_NetworkUpdates_SaveGroup(t *testing.T) {
 	}, true)
 	require.NoError(t, err)
 
-	updMsg := manager.peersUpdateManager.CreateChannel(context.Background(), peer1.ID)
+	updChan := manager.peersUpdateManager.CreateChannel(context.Background(), peer1.ID)
 	defer manager.peersUpdateManager.CloseChannel(context.Background(), peer1.ID)
 
 	wg := sync.WaitGroup{}
@@ -1198,7 +1198,12 @@ func testAccountManager_NetworkUpdates_SaveGroup(t *testing.T) {
 	go func() {
 		defer wg.Done()
 
-		message := <-updMsg
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		message, _, _, err := updChan.NetworkMap.Pop(ctx)
+		if err != nil {
+			t.Errorf("timeout waiting for network update")
+		}
 		networkMap := message.Update.GetNetworkMap()
 		if len(networkMap.RemotePeers) != 2 {
 			t.Errorf("mismatch peers count: 2 expected, got %v", len(networkMap.RemotePeers))
@@ -1229,12 +1234,11 @@ func testAccountManager_NetworkUpdates_DeletePolicy(t *testing.T) {
 	updMsg := manager.peersUpdateManager.CreateChannel(context.Background(), peer1.ID)
 	defer manager.peersUpdateManager.CloseChannel(context.Background(), peer1.ID)
 
-	// Ensure that we do not receive an update message before the policy is deleted
-	time.Sleep(time.Second)
-	select {
-	case <-updMsg:
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	_, _, _, err := updMsg.NetworkMap.Pop(ctx)
+	if err != nil {
 		t.Logf("received addPeer update message before policy deletion")
-	default:
 	}
 
 	wg := sync.WaitGroup{}
@@ -1242,7 +1246,12 @@ func testAccountManager_NetworkUpdates_DeletePolicy(t *testing.T) {
 	go func() {
 		defer wg.Done()
 
-		message := <-updMsg
+		ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+		defer cancel()
+		message, _, _, err := updMsg.NetworkMap.Pop(ctx)
+		if err != nil {
+			t.Errorf("timeout waiting for network update")
+		}
 		networkMap := message.Update.GetNetworkMap()
 		if len(networkMap.RemotePeers) != 0 {
 			t.Errorf("mismatch peers count: 0 expected, got %v", len(networkMap.RemotePeers))
@@ -1288,7 +1297,12 @@ func testAccountManager_NetworkUpdates_SavePolicy(t *testing.T) {
 	go func() {
 		defer wg.Done()
 
-		message := <-updMsg
+		ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+		defer cancel()
+		message, _, _, err := updMsg.NetworkMap.Pop(ctx)
+		if err != nil {
+			t.Errorf("timeout waiting for network update")
+		}
 		networkMap := message.Update.GetNetworkMap()
 		if len(networkMap.RemotePeers) != 2 {
 			t.Errorf("mismatch peers count: 2 expected, got %v", len(networkMap.RemotePeers))
@@ -1362,7 +1376,12 @@ func testAccountManager_NetworkUpdates_DeletePeer(t *testing.T) {
 	go func() {
 		defer wg.Done()
 
-		message := <-updMsg
+		ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+		defer cancel()
+		message, _, _, err := updMsg.NetworkMap.Pop(ctx)
+		if err != nil {
+			t.Errorf("timeout waiting for network update")
+		}
 		networkMap := message.Update.GetNetworkMap()
 		if len(networkMap.RemotePeers) != 1 {
 			t.Errorf("mismatch peers count: 1 expected, got %v", len(networkMap.RemotePeers))
@@ -1427,7 +1446,12 @@ func testAccountManager_NetworkUpdates_DeleteGroup(t *testing.T) {
 	go func() {
 		defer wg.Done()
 
-		message := <-updMsg
+		ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+		defer cancel()
+		message, _, _, err := updMsg.NetworkMap.Pop(ctx)
+		if err != nil {
+			t.Errorf("timeout waiting for network update")
+		}
 		networkMap := message.Update.GetNetworkMap()
 		if len(networkMap.RemotePeers) != 0 {
 			t.Errorf("mismatch peers count: 0 expected, got %v", len(networkMap.RemotePeers))
@@ -2939,7 +2963,7 @@ func createManager(t testing.TB) (*DefaultAccountManager, error) {
 
 	permissionsManager := permissions.NewManager(store)
 
-	manager, err := BuildManager(context.Background(), store, NewPeersUpdateManager(nil), nil, "", "netbird.cloud", eventStore, nil, false, MockIntegratedValidator{}, metrics, port_forwarding.NewControllerMock(), settingsMockManager, permissionsManager, false)
+	manager, err := BuildManager(context.Background(), store, NewPeersUpdateManager(metrics), nil, "", "netbird.cloud", eventStore, nil, false, MockIntegratedValidator{}, metrics, port_forwarding.NewControllerMock(), settingsMockManager, permissionsManager, false)
 	if err != nil {
 		return nil, err
 	}
@@ -3020,27 +3044,33 @@ func setupNetworkMapTest(t *testing.T) (*DefaultAccountManager, *types.Account,
 	return manager, account, peer1, peer2, peer3
 }
 
-func peerShouldNotReceiveUpdate(t *testing.T, updateMessage <-chan *UpdateMessage) {
+func peerShouldNotReceiveUpdate(t *testing.T, updateMessageBuffer *UpdateBuffer) {
 	t.Helper()
-	select {
-	case msg := <-updateMessage:
+	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+	defer cancel()
+	msg, _, _, _ := updateMessageBuffer.Pop(ctx)
+	if msg != nil {
 		t.Errorf("Unexpected message received: %+v", msg)
-	case <-time.After(500 * time.Millisecond):
-		return
 	}
+
+	return
 }
 
-func peerShouldReceiveUpdate(t *testing.T, updateMessage <-chan *UpdateMessage) {
+func peerShouldReceiveUpdate(t *testing.T, updateMessageBuffer *UpdateBuffer) {
 	t.Helper()
 
-	select {
-	case msg := <-updateMessage:
-		if msg == nil {
-			t.Errorf("Received nil update message, expected valid message")
-		}
-	case <-time.After(500 * time.Millisecond):
-		t.Error("Timed out waiting for update message")
+	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+	defer cancel()
+
+	msg, _, _, err := updateMessageBuffer.Pop(ctx)
+	if err != nil {
+		t.Errorf("Expected update message, but none received")
 	}
+	if msg == nil {
+		t.Errorf("Received nil update message, expected valid message")
+	}
+
+	return
 }
 
 func BenchmarkSyncAndMarkPeer(b *testing.B) {
@@ -3077,11 +3107,13 @@ func BenchmarkSyncAndMarkPeer(b *testing.B) {
 			if err != nil {
 				b.Fatalf("Failed to get account: %v", err)
 			}
-			peerChannels := make(map[string]chan *UpdateMessage)
+			peerChannels := make(map[string]*UpdateChannel)
 			for peerID := range account.Peers {
-				peerChannels[peerID] = make(chan *UpdateMessage, channelBufferSize)
+				peerChannels[peerID] = &UpdateChannel{
+					Important:  make(chan *UpdateMessage, channelBufferSize),
+					NetworkMap: NewUpdateBuffer(manager.metrics.UpdateChannelMetrics()),
+				}
 			}
-			manager.peersUpdateManager.peerChannels = peerChannels
 
 			b.ResetTimer()
 			start := time.Now()
@@ -3140,9 +3172,12 @@ func BenchmarkLoginPeer_ExistingPeer(b *testing.B) {
 			if err != nil {
 				b.Fatalf("Failed to get account: %v", err)
 			}
-			peerChannels := make(map[string]chan *UpdateMessage)
+			peerChannels := make(map[string]*UpdateChannel)
 			for peerID := range account.Peers {
-				peerChannels[peerID] = make(chan *UpdateMessage, channelBufferSize)
+				peerChannels[peerID] = &UpdateChannel{
+					Important:  make(chan *UpdateMessage, channelBufferSize),
+					NetworkMap: NewUpdateBuffer(manager.metrics.UpdateChannelMetrics()),
+				}
 			}
 			manager.peersUpdateManager.peerChannels = peerChannels
 
@@ -3210,9 +3245,12 @@ func BenchmarkLoginPeer_NewPeer(b *testing.B) {
 			if err != nil {
 				b.Fatalf("Failed to get account: %v", err)
 			}
-			peerChannels := make(map[string]chan *UpdateMessage)
+			peerChannels := make(map[string]*UpdateChannel)
 			for peerID := range account.Peers {
-				peerChannels[peerID] = make(chan *UpdateMessage, channelBufferSize)
+				peerChannels[peerID] = &UpdateChannel{
+					Important:  make(chan *UpdateMessage, channelBufferSize),
+					NetworkMap: NewUpdateBuffer(manager.metrics.UpdateChannelMetrics()),
+				}
 			}
 			manager.peersUpdateManager.peerChannels = peerChannels
 

management/server/auth/manager_test.go

@@ -231,7 +231,7 @@ func TestAuthManager_ValidateAndParseToken(t *testing.T) {
 		return fmt.Sprintf("%s/%s", audience, name)
 	}
 
-	lastLogin := time.Date(2025, 2, 12, 14, 25, 26, 0, time.UTC) //"2025-02-12T14:25:26.186Z"
+	lastLogin := time.Date(2025, 2, 12, 14, 25, 26, 0, time.UTC) // "2025-02-12T14:25:26.186Z"
 
 	tests := []struct {
 		name      string

management/server/dns_test.go

@@ -609,7 +609,7 @@ func TestDNSAccountPeersUpdate(t *testing.T) {
 	t.Run("saving dns setting with unused groups", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldNotReceiveUpdate(t, updMsg)
+			peerShouldNotReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -629,7 +629,7 @@ func TestDNSAccountPeersUpdate(t *testing.T) {
 	t.Run("creating dns setting with unused groups", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldNotReceiveUpdate(t, updMsg)
+			peerShouldNotReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -662,7 +662,7 @@ func TestDNSAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -688,7 +688,7 @@ func TestDNSAccountPeersUpdate(t *testing.T) {
 	t.Run("saving dns setting with used groups", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -708,7 +708,7 @@ func TestDNSAccountPeersUpdate(t *testing.T) {
 	t.Run("removing group with no peers from dns settings", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldNotReceiveUpdate(t, updMsg)
+			peerShouldNotReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -728,7 +728,7 @@ func TestDNSAccountPeersUpdate(t *testing.T) {
 	t.Run("removing group with peers from dns settings", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 

management/server/group_test.go

@@ -451,7 +451,7 @@ func TestGroupAccountPeersUpdate(t *testing.T) {
 	t.Run("saving unlinked group", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldNotReceiveUpdate(t, updMsg)
+			peerShouldNotReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -474,7 +474,7 @@ func TestGroupAccountPeersUpdate(t *testing.T) {
 	t.Run("adding peer to unlinked group", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldNotReceiveUpdate(t, updMsg)
+			peerShouldNotReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -493,7 +493,7 @@ func TestGroupAccountPeersUpdate(t *testing.T) {
 	t.Run("removing peer from unliked group", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldNotReceiveUpdate(t, updMsg)
+			peerShouldNotReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -511,7 +511,7 @@ func TestGroupAccountPeersUpdate(t *testing.T) {
 	t.Run("deleting group", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldNotReceiveUpdate(t, updMsg)
+			peerShouldNotReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -544,7 +544,7 @@ func TestGroupAccountPeersUpdate(t *testing.T) {
 	t.Run("saving linked group to policy", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -566,7 +566,7 @@ func TestGroupAccountPeersUpdate(t *testing.T) {
 	t.Run("adding peer to linked group", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -584,7 +584,7 @@ func TestGroupAccountPeersUpdate(t *testing.T) {
 	t.Run("removing peer from linked group", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -613,7 +613,7 @@ func TestGroupAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -654,7 +654,7 @@ func TestGroupAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -681,7 +681,7 @@ func TestGroupAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -728,7 +728,7 @@ func TestGroupAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 

management/server/grpcserver.go

@@ -68,6 +68,7 @@ type GRPCServer struct {
 	logBlockedPeers          bool
 	blockPeersWithSameConfig bool
 	integratedPeerValidator  integrated_validator.IntegratedValidator
+	debounce                 int
 
 	syncSem atomic.Int32
 	syncLim int32
@@ -94,7 +95,7 @@ func NewServer(
 	if appMetrics != nil {
 		// update gauge based on number of connected peers which is equal to open gRPC streams
 		err = appMetrics.GRPCMetrics().RegisterConnectedStreams(func() int64 {
-			return int64(len(peersUpdateManager.peerChannels))
+			return int64(peersUpdateManager.GetChannelCount())
 		})
 		if err != nil {
 			return nil, err
@@ -104,6 +105,14 @@ func NewServer(
 	logBlockedPeers := strings.ToLower(os.Getenv(envLogBlockedPeers)) == "true"
 	blockPeersWithSameConfig := strings.ToLower(os.Getenv(envBlockPeers)) == "true"
 
+	debounce := 1
+	if decounceStr := os.Getenv("NB_UPDATE_BUFFER_DEBOUNCE_S"); decounceStr != "" {
+		debounce, err = strconv.Atoi(decounceStr)
+		if err != nil {
+			log.Errorf("invalid value for NB_UPDATE_BUFFER_DEBOUNCE_S: %v using 1 second", err)
+		}
+	}
+
 	syncLim := int32(defaultSyncLim)
 	if syncLimStr := os.Getenv(envConcurrentSyncs); syncLimStr != "" {
 		syncLimParsed, err := strconv.Atoi(syncLimStr)
@@ -128,6 +137,7 @@ func NewServer(
 		logBlockedPeers:          logBlockedPeers,
 		blockPeersWithSameConfig: blockPeersWithSameConfig,
 		integratedPeerValidator:  integratedPeerValidator,
+		debounce:                 debounce,
 
 		syncLim: syncLim,
 	}, nil
@@ -274,14 +284,56 @@ func (s *GRPCServer) Sync(req *proto.EncryptedMessage, srv proto.ManagementServi
 }
 
 // handleUpdates sends updates to the connected peer until the updates channel is closed.
-func (s *GRPCServer) handleUpdates(ctx context.Context, accountID string, peerKey wgtypes.Key, peer *nbpeer.Peer, updates chan *UpdateMessage, srv proto.ManagementService_SyncServer) error {
+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 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 {
+			start := time.Now()
+			update, overwrites, timeSinceLastPop, err := updates.NetworkMap.Pop(ctx)
+			log.WithContext(ctx).Debugf("popped an update for peer %s from the network map buffer in %v (overwrites: %d)", peerKey.String(), time.Since(start), overwrites)
+			if err != nil {
+				close(networkMapCh)
+				return
+			}
+			// start := time.Now()
+			select {
+			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
+			}
+		}
+	}()
+
 	for {
 		select {
 		// condition when there are some updates
-		case update, open := <-updates:
+		case update, open := <-updates.Important:
 			if s.appMetrics != nil {
-				s.appMetrics.GRPCMetrics().UpdateChannelQueueLength(len(updates) + 1)
+				s.appMetrics.GRPCMetrics().UpdateChannelQueueLength(len(updates.Important) + 1)
 			}
 
 			if !open {
@@ -289,13 +341,24 @@ func (s *GRPCServer) handleUpdates(ctx context.Context, accountID string, peerKe
 				s.cancelPeerRoutines(ctx, accountID, peer)
 				return nil
 			}
-			log.WithContext(ctx).Debugf("received an update for peer %s", peerKey.String())
+			// 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
 			}
 
+		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 (overwrites: %d)", peerKey.String(), updateData.overwrites)
+			if err := s.sendUpdate(ctx, accountID, peerKey, peer, updateData.msg, srv); err != nil {
+				return err
+			}
+
 		// condition when client <-> server connection has been terminated
 		case <-srv.Context().Done():
 			// happens when connection drops, e.g. client disconnects
@@ -306,6 +369,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 {
@@ -503,12 +593,6 @@ func (s *GRPCServer) parseRequest(ctx context.Context, req *proto.EncryptedMessa
 // In case it isn't, the endpoint checks whether setup key is provided within the request and tries to register a peer.
 // In case of the successful registration login is also successful
 func (s *GRPCServer) Login(ctx context.Context, req *proto.EncryptedMessage) (*proto.EncryptedMessage, error) {
-	if s.syncSem.Load() >= s.syncLim {
-		return nil, status.Errorf(codes.ResourceExhausted, "too many concurrent login requests, please try again later")
-	}
-	s.syncSem.Add(1)
-	defer s.syncSem.Add(-1)
-
 	reqStart := time.Now()
 	realIP := getRealIP(ctx)
 	sRealIP := realIP.String()

management/server/nameserver_test.go

@@ -1004,7 +1004,7 @@ func TestNameServerAccountPeersUpdate(t *testing.T) {
 	t.Run("creating nameserver group with distribution group no peers", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldNotReceiveUpdate(t, updMsg)
+			peerShouldNotReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -1031,7 +1031,7 @@ func TestNameServerAccountPeersUpdate(t *testing.T) {
 	t.Run("saving nameserver group with distribution group no peers", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldNotReceiveUpdate(t, updMsg)
+			peerShouldNotReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -1049,7 +1049,7 @@ func TestNameServerAccountPeersUpdate(t *testing.T) {
 	t.Run("creating nameserver group with distribution group has peers", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -1075,7 +1075,7 @@ func TestNameServerAccountPeersUpdate(t *testing.T) {
 	t.Run("saving nameserver group with distribution group has peers", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -1105,7 +1105,7 @@ func TestNameServerAccountPeersUpdate(t *testing.T) {
 	t.Run("deleting nameserver group", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 

management/server/peer.go

@@ -6,6 +6,7 @@ import (
 	b64 "encoding/base64"
 	"fmt"
 	"net"
+	"runtime"
 	"slices"
 	"strings"
 	"sync"
@@ -1259,7 +1260,6 @@ func (am *DefaultAccountManager) UpdateAccountPeers(ctx context.Context, account
 	}
 
 	var wg sync.WaitGroup
-	semaphore := make(chan struct{}, 10)
 
 	dnsCache := &DNSConfigCache{}
 	dnsDomain := am.GetDNSDomain(account.Settings)
@@ -1285,57 +1285,64 @@ func (am *DefaultAccountManager) UpdateAccountPeers(ctx context.Context, account
 
 	dnsFwdPort := computeForwarderPort(maps.Values(account.Peers), dnsForwarderPortMinVersion)
 
+	processNetworkMap := func(p *nbpeer.Peer) {
+		start := time.Now()
+
+		postureChecks, err := am.getPeerPostureChecks(account, p.ID)
+		if err != nil {
+			log.WithContext(ctx).Debugf("failed to get posture checks for peer %s: %v", p.ID, err)
+			return
+		}
+
+		am.metrics.UpdateChannelMetrics().CountCalcPostureChecksDuration(time.Since(start))
+		start = time.Now()
+
+		var remotePeerNetworkMap *types.NetworkMap
+
+		if am.experimentalNetworkMap(accountID) {
+			remotePeerNetworkMap = am.getPeerNetworkMapExp(ctx, p.AccountID, p.ID, approvedPeersMap, customZone, am.metrics.AccountManagerMetrics())
+		} else {
+			remotePeerNetworkMap = account.GetPeerNetworkMap(ctx, p.ID, customZone, approvedPeersMap, resourcePolicies, routers, am.metrics.AccountManagerMetrics())
+		}
+
+		am.metrics.UpdateChannelMetrics().CountCalcPeerNetworkMapDuration(time.Since(start))
+		start = time.Now()
+
+		proxyNetworkMap, ok := proxyNetworkMaps[p.ID]
+		if ok {
+			remotePeerNetworkMap.Merge(proxyNetworkMap)
+		}
+		am.metrics.UpdateChannelMetrics().CountMergeNetworkMapDuration(time.Since(start))
+
+		peerGroups := account.GetPeerGroups(p.ID)
+		start = time.Now()
+		update := toSyncResponse(ctx, nil, p, nil, nil, remotePeerNetworkMap, dnsDomain, postureChecks, dnsCache, account.Settings, extraSetting, maps.Keys(peerGroups), dnsFwdPort)
+		am.metrics.UpdateChannelMetrics().CountToSyncResponseDuration(time.Since(start))
+
+		am.peersUpdateManager.SendNetworkMapUpdate(ctx, p.ID, &UpdateMessage{Update: update})
+	}
+
+	numWorkers := runtime.NumCPU()
+	wch := make(chan *nbpeer.Peer, numWorkers)
+	wg.Add(numWorkers)
+	for range numWorkers {
+		go func() {
+			for peer := range wch {
+				processNetworkMap(peer)
+			}
+			wg.Done()
+		}()
+	}
+
 	for _, peer := range account.Peers {
 		if !am.peersUpdateManager.HasChannel(peer.ID) {
 			log.WithContext(ctx).Tracef("peer %s doesn't have a channel, skipping network map update", peer.ID)
 			continue
 		}
-
-		wg.Add(1)
-		semaphore <- struct{}{}
-		go func(p *nbpeer.Peer) {
-			defer wg.Done()
-			defer func() { <-semaphore }()
-
-			start := time.Now()
-
-			postureChecks, err := am.getPeerPostureChecks(account, p.ID)
-			if err != nil {
-				log.WithContext(ctx).Debugf("failed to get posture checks for peer %s: %v", peer.ID, err)
-				return
-			}
-
-			am.metrics.UpdateChannelMetrics().CountCalcPostureChecksDuration(time.Since(start))
-			start = time.Now()
-
-			var remotePeerNetworkMap *types.NetworkMap
-
-			if am.experimentalNetworkMap(accountID) {
-				remotePeerNetworkMap = am.getPeerNetworkMapExp(ctx, p.AccountID, p.ID, approvedPeersMap, customZone, am.metrics.AccountManagerMetrics())
-			} else {
-				remotePeerNetworkMap = account.GetPeerNetworkMap(ctx, p.ID, customZone, approvedPeersMap, resourcePolicies, routers, am.metrics.AccountManagerMetrics())
-			}
-
-			am.metrics.UpdateChannelMetrics().CountCalcPeerNetworkMapDuration(time.Since(start))
-			start = time.Now()
-
-			proxyNetworkMap, ok := proxyNetworkMaps[p.ID]
-			if ok {
-				remotePeerNetworkMap.Merge(proxyNetworkMap)
-			}
-			am.metrics.UpdateChannelMetrics().CountMergeNetworkMapDuration(time.Since(start))
-
-			peerGroups := account.GetPeerGroups(p.ID)
-			start = time.Now()
-			update := toSyncResponse(ctx, nil, p, nil, nil, remotePeerNetworkMap, dnsDomain, postureChecks, dnsCache, account.Settings, extraSetting, maps.Keys(peerGroups), dnsFwdPort)
-			am.metrics.UpdateChannelMetrics().CountToSyncResponseDuration(time.Since(start))
-
-			am.peersUpdateManager.SendUpdate(ctx, p.ID, &UpdateMessage{Update: update})
-		}(peer)
+		wch <- peer
 	}
 
-	//
-
+	close(wch)
 	wg.Wait()
 	if am.metrics != nil {
 		am.metrics.AccountManagerMetrics().CountUpdateAccountPeersDuration(time.Since(globalStart))
@@ -1343,9 +1350,11 @@ func (am *DefaultAccountManager) UpdateAccountPeers(ctx context.Context, account
 }
 
 type bufferUpdate struct {
-	mu     sync.Mutex
-	next   *time.Timer
-	update atomic.Bool
+	mu            sync.Mutex
+	next          *time.Timer
+	update        atomic.Bool
+	requestCount  atomic.Int32
+	lastResetTime atomic.Int64 // Unix timestamp in milliseconds
 }
 
 func (am *DefaultAccountManager) BufferUpdateAccountPeers(ctx context.Context, accountID string) {
@@ -1356,6 +1365,7 @@ func (am *DefaultAccountManager) BufferUpdateAccountPeers(ctx context.Context, a
 
 	if !b.mu.TryLock() {
 		b.update.Store(true)
+		b.requestCount.Add(1)
 		return
 	}
 
@@ -1365,18 +1375,41 @@ func (am *DefaultAccountManager) BufferUpdateAccountPeers(ctx context.Context, a
 
 	go func() {
 		defer b.mu.Unlock()
+
+		start := time.Now()
+
+		b.requestCount.Store(0)
+
 		am.UpdateAccountPeers(ctx, accountID)
+
 		if !b.update.Load() {
 			return
 		}
 		b.update.Store(false)
+
+		requestsDuringProcessing := b.requestCount.Load()
+		executionTime := time.Since(start)
+
+		requestsPerMinute := float64(requestsDuringProcessing) / executionTime.Seconds() * 60
+
+		adaptiveDelay := time.Duration(requestsPerMinute) * 250 * time.Millisecond
+
+		// cap the maximum delay to avoid too long delays
+		maxDelay := 30 * time.Second
+		if adaptiveDelay > maxDelay {
+			adaptiveDelay = maxDelay
+		}
+
+		log.WithContext(ctx).Debugf("adaptive debounce for account %s: %d requests during %v execution, waiting %v",
+			accountID, requestsDuringProcessing, executionTime, adaptiveDelay)
+
 		if b.next == nil {
-			b.next = time.AfterFunc(time.Duration(am.updateAccountPeersBufferInterval.Load()), func() {
+			b.next = time.AfterFunc(adaptiveDelay, func() {
 				am.UpdateAccountPeers(ctx, accountID)
 			})
 			return
 		}
-		b.next.Reset(time.Duration(am.updateAccountPeersBufferInterval.Load()))
+		b.next.Reset(adaptiveDelay)
 	}()
 }
 
@@ -1447,7 +1480,7 @@ func (am *DefaultAccountManager) UpdateAccountPeer(ctx context.Context, accountI
 	dnsFwdPort := computeForwarderPort(maps.Values(account.Peers), dnsForwarderPortMinVersion)
 
 	update := toSyncResponse(ctx, nil, peer, nil, nil, remotePeerNetworkMap, dnsDomain, postureChecks, dnsCache, account.Settings, extraSettings, maps.Keys(peerGroups), dnsFwdPort)
-	am.peersUpdateManager.SendUpdate(ctx, peer.ID, &UpdateMessage{Update: update})
+	am.peersUpdateManager.SendNetworkMapUpdate(ctx, peer.ID, &UpdateMessage{Update: update})
 }
 
 // getNextPeerExpiration returns the minimum duration in which the next peer of the account will expire if it was found.
@@ -1644,7 +1677,7 @@ func deletePeers(ctx context.Context, am *DefaultAccountManager, transaction sto
 			return nil, err
 		}
 
-		am.peersUpdateManager.SendUpdate(ctx, peer.ID, &UpdateMessage{
+		am.peersUpdateManager.SendImportantUpdate(ctx, peer.ID, &UpdateMessage{
 			Update: &proto.SyncResponse{
 				RemotePeers:        []*proto.RemotePeerConfig{},
 				RemotePeersIsEmpty: true,

management/server/peer_test.go

@@ -26,7 +26,6 @@ import (
 	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
 
 	"github.com/netbirdio/netbird/management/internals/server/config"
-	"github.com/netbirdio/netbird/management/server/http/testing/testing_tools"
 	"github.com/netbirdio/netbird/management/server/integrations/port_forwarding"
 	"github.com/netbirdio/netbird/management/server/mock_server"
 	"github.com/netbirdio/netbird/management/server/permissions"
@@ -954,14 +953,15 @@ func BenchmarkUpdateAccountPeers(b *testing.B) {
 		minMsPerOpCICD  float64
 		maxMsPerOpCICD  float64
 	}{
-		{"Small", 50, 5, 90, 120, 90, 120},
-		{"Medium", 500, 100, 110, 150, 120, 260},
-		{"Large", 5000, 200, 800, 1700, 2500, 5000},
-		{"Small single", 50, 10, 90, 120, 90, 120},
-		{"Medium single", 500, 10, 110, 170, 120, 200},
-		{"Large 5", 5000, 15, 1300, 2100, 4900, 7000},
-		{"Extra Large", 2000, 2000, 1300, 2400, 3000, 6400},
+		{"Small", 350, 5, 90, 120, 90, 120},
+		// {"Medium", 500, 100, 110, 150, 120, 260},
+		// {"Large", 5000, 2, 800, 1700, 2500, 5000},
+		// {"Small single", 50, 10, 90, 120, 90, 120},
+		// {"Medium single", 500, 10, 110, 170, 120, 200},
+		// {"Large 5", 5000, 15, 1300, 2100, 4900, 7000},
+		// {"Extra Large", 2000, 2000, 1300, 2400, 3000, 6400},
 	}
+	b.Setenv("NB_EXPERIMENT_NETWORK_MAP", "false")
 
 	log.SetOutput(io.Discard)
 	defer log.SetOutput(os.Stderr)
@@ -980,38 +980,39 @@ func BenchmarkUpdateAccountPeers(b *testing.B) {
 				b.Fatalf("Failed to get account: %v", err)
 			}
 
-			peerChannels := make(map[string]chan *UpdateMessage)
+			peerChannels := make(map[string]*UpdateChannel)
 
 			for peerID := range account.Peers {
-				peerChannels[peerID] = make(chan *UpdateMessage, channelBufferSize)
+				peerChannels[peerID] = &UpdateChannel{
+					Important:  make(chan *UpdateMessage, channelBufferSize),
+					NetworkMap: NewUpdateBuffer(manager.metrics.UpdateChannelMetrics()),
+				}
 			}
 
 			manager.peersUpdateManager.peerChannels = peerChannels
 
-			b.ResetTimer()
-			start := time.Now()
+			for i := 0; i < 1000; i++ {
+				b.Logf("Run %d", i)
+				manager.UpdateAccountPeers(ctx, account.Id)
+				mm(b)
+				manager.Store.IncrementNetworkSerial(ctx, account.Id)
+			}
 
 			for i := 0; i < b.N; i++ {
 				manager.UpdateAccountPeers(ctx, account.Id)
 			}
 
-			duration := time.Since(start)
-			msPerOp := float64(duration.Nanoseconds()) / float64(b.N) / 1e6
-			b.ReportMetric(msPerOp, "ms/op")
-
-			maxExpected := bc.maxMsPerOpLocal
-			if os.Getenv("CI") == "true" {
-				maxExpected = bc.maxMsPerOpCICD
-				testing_tools.EvaluateBenchmarkResults(b, bc.name, time.Since(start), "login", "newPeer")
-			}
-
-			if msPerOp > maxExpected {
-				b.Logf("Benchmark %s: too slow (%.2f ms/op, max %.2f ms/op)", bc.name, msPerOp, maxExpected)
-			}
 		})
 	}
 }
 
+func mm(b *testing.B) {
+	var m runtime.MemStats
+	runtime.ReadMemStats(&m)
+	// runtime.GC()
+	b.Logf("Alloc: %v MB, TotalAlloc: %v MB, Sys: %v MB, NumGC: %v", m.Alloc/1024/1024, m.TotalAlloc/1024/1024, m.Sys/1024/1024, m.NumGC)
+}
+
 func TestUpdateAccountPeers_Experimental(t *testing.T) {
 	t.Setenv(envNewNetworkMapBuilder, "true")
 	testUpdateAccountPeers(t)
@@ -1044,22 +1045,30 @@ func testUpdateAccountPeers(t *testing.T) {
 
 			ctx := context.Background()
 
+			metrics, err := telemetry.NewDefaultAppMetrics(ctx)
+			if err != nil {
+				t.Fatalf("Failed to create metrics: %v", err)
+			}
+
 			account, err := manager.Store.GetAccount(ctx, accountID)
 			if err != nil {
 				t.Fatalf("Failed to get account: %v", err)
 			}
 
-			peerChannels := make(map[string]chan *UpdateMessage)
+			peerChannels := make(map[string]*UpdateChannel)
 
 			for peerID := range account.Peers {
-				peerChannels[peerID] = make(chan *UpdateMessage, channelBufferSize)
+				peerChannels[peerID] = &UpdateChannel{
+					Important:  make(chan *UpdateMessage, channelBufferSize),
+					NetworkMap: NewUpdateBuffer(metrics.UpdateChannelMetrics()),
+				}
 			}
 
 			manager.peersUpdateManager.peerChannels = peerChannels
 			manager.UpdateAccountPeers(ctx, account.Id)
 
 			for _, channel := range peerChannels {
-				update := <-channel
+				update, _, _, _ := channel.NetworkMap.Pop(ctx)
 				assert.Nil(t, update.Update.NetbirdConfig)
 				assert.Equal(t, tc.peers, len(update.Update.NetworkMap.RemotePeers))
 				assert.Equal(t, tc.peers*2, len(update.Update.NetworkMap.FirewallRules))
@@ -1791,7 +1800,7 @@ func TestPeerAccountPeersUpdate(t *testing.T) {
 	t.Run("updating not expired peer and peer expiration is enabled", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldNotReceiveUpdate(t, updMsg)
+			peerShouldNotReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -1809,7 +1818,7 @@ func TestPeerAccountPeersUpdate(t *testing.T) {
 	t.Run("adding peer to unlinked group", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg) //
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap) //
 			close(done)
 		}()
 
@@ -1834,7 +1843,7 @@ func TestPeerAccountPeersUpdate(t *testing.T) {
 	t.Run("deleting peer with unlinked group", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -1852,7 +1861,7 @@ func TestPeerAccountPeersUpdate(t *testing.T) {
 	t.Run("updating peer label", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -1875,7 +1884,7 @@ func TestPeerAccountPeersUpdate(t *testing.T) {
 		manager.integratedPeerValidator = MockIntegratedValidator{ValidatePeerFunc: requireUpdateFunc}
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -1897,7 +1906,7 @@ func TestPeerAccountPeersUpdate(t *testing.T) {
 		manager.integratedPeerValidator = MockIntegratedValidator{ValidatePeerFunc: requireNoUpdateFunc}
 		done := make(chan struct{})
 		go func() {
-			peerShouldNotReceiveUpdate(t, updMsg)
+			peerShouldNotReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -1930,7 +1939,7 @@ func TestPeerAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -1956,7 +1965,7 @@ func TestPeerAccountPeersUpdate(t *testing.T) {
 	t.Run("deleting peer with linked group to policy", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -1994,7 +2003,7 @@ func TestPeerAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -2020,7 +2029,7 @@ func TestPeerAccountPeersUpdate(t *testing.T) {
 	t.Run("deleting peer with linked group to route", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -2049,7 +2058,7 @@ func TestPeerAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -2075,7 +2084,7 @@ func TestPeerAccountPeersUpdate(t *testing.T) {
 	t.Run("deleting peer with linked group to route", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 

management/server/policy_test.go

@@ -1034,7 +1034,7 @@ func TestPolicyAccountPeersUpdate(t *testing.T) {
 	t.Run("saving policy with rule groups with no peers", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldNotReceiveUpdate(t, updMsg)
+			peerShouldNotReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -1065,7 +1065,7 @@ func TestPolicyAccountPeersUpdate(t *testing.T) {
 	t.Run("saving policy where source has peers but destination does not", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -1097,7 +1097,7 @@ func TestPolicyAccountPeersUpdate(t *testing.T) {
 	t.Run("saving policy where destination has peers but source does not", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -1129,7 +1129,7 @@ func TestPolicyAccountPeersUpdate(t *testing.T) {
 	t.Run("saving policy with source and destination groups with peers", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -1160,7 +1160,7 @@ func TestPolicyAccountPeersUpdate(t *testing.T) {
 	t.Run("disabling policy with source and destination groups with peers", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -1180,7 +1180,7 @@ func TestPolicyAccountPeersUpdate(t *testing.T) {
 	t.Run("updating disabled policy with source and destination groups with peers", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldNotReceiveUpdate(t, updMsg)
+			peerShouldNotReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -1201,7 +1201,7 @@ func TestPolicyAccountPeersUpdate(t *testing.T) {
 	t.Run("enabling policy with source and destination groups with peers", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -1220,7 +1220,7 @@ func TestPolicyAccountPeersUpdate(t *testing.T) {
 	t.Run("deleting policy with source and destination groups with peers", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -1240,7 +1240,7 @@ func TestPolicyAccountPeersUpdate(t *testing.T) {
 	t.Run("deleting policy where destination has peers but source does not", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -1258,7 +1258,7 @@ func TestPolicyAccountPeersUpdate(t *testing.T) {
 	t.Run("deleting policy with no peers in groups", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldNotReceiveUpdate(t, updMsg)
+			peerShouldNotReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 

management/server/posture_checks_test.go

@@ -180,7 +180,7 @@ func TestPostureCheckAccountPeersUpdate(t *testing.T) {
 	t.Run("saving unused posture check", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldNotReceiveUpdate(t, updMsg)
+			peerShouldNotReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -198,7 +198,7 @@ func TestPostureCheckAccountPeersUpdate(t *testing.T) {
 	t.Run("updating unused posture check", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldNotReceiveUpdate(t, updMsg)
+			peerShouldNotReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -235,7 +235,7 @@ func TestPostureCheckAccountPeersUpdate(t *testing.T) {
 	t.Run("linking posture check to policy with peers", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -264,7 +264,7 @@ func TestPostureCheckAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -282,7 +282,7 @@ func TestPostureCheckAccountPeersUpdate(t *testing.T) {
 	t.Run("removing posture check from policy", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -301,7 +301,7 @@ func TestPostureCheckAccountPeersUpdate(t *testing.T) {
 	t.Run("deleting unused posture check", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldNotReceiveUpdate(t, updMsg)
+			peerShouldNotReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -337,7 +337,7 @@ func TestPostureCheckAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldNotReceiveUpdate(t, updMsg)
+			peerShouldNotReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -381,7 +381,7 @@ func TestPostureCheckAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg1)
+			peerShouldReceiveUpdate(t, updMsg1.NetworkMap)
 			close(done)
 		}()
 
@@ -420,7 +420,7 @@ func TestPostureCheckAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 

management/server/route_test.go

@@ -1998,7 +1998,7 @@ func TestRouteAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldNotReceiveUpdate(t, updMsg)
+			peerShouldNotReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -2034,7 +2034,7 @@ func TestRouteAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -2070,7 +2070,7 @@ func TestRouteAccountPeersUpdate(t *testing.T) {
 	t.Run("creating route with a routing peer", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -2095,7 +2095,7 @@ func TestRouteAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -2113,7 +2113,7 @@ func TestRouteAccountPeersUpdate(t *testing.T) {
 	t.Run("deleting route", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -2149,7 +2149,7 @@ func TestRouteAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -2189,7 +2189,7 @@ func TestRouteAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 

management/server/setupkey_test.go

@@ -431,7 +431,7 @@ func TestSetupKeyAccountPeersUpdate(t *testing.T) {
 	t.Run("creating setup key", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldNotReceiveUpdate(t, updMsg)
+			peerShouldNotReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -449,7 +449,7 @@ func TestSetupKeyAccountPeersUpdate(t *testing.T) {
 	t.Run("saving setup key", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldNotReceiveUpdate(t, updMsg)
+			peerShouldNotReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 

management/server/telemetry/updatechannel_metrics.go

@@ -22,6 +22,9 @@ type UpdateChannelMetrics struct {
 	calcPeerNetworkMapDurationMs      metric.Int64Histogram
 	mergeNetworkMapDurationMicro      metric.Int64Histogram
 	toSyncResponseDurationMicro       metric.Int64Histogram
+	bufferPushCounter                 metric.Int64Counter
+	bufferOverwriteCounter            metric.Int64Counter
+	bufferIgnoreCounter               metric.Int64Counter
 	ctx                               context.Context
 }
 
@@ -125,6 +128,27 @@ func NewUpdateChannelMetrics(ctx context.Context, meter metric.Meter) (*UpdateCh
 		return nil, err
 	}
 
+	bufferPushCounter, err := meter.Int64Counter("management.updatechannel.buffer.push.counter",
+		metric.WithUnit("1"),
+		metric.WithDescription("Number of updates pushed to an empty buffer"))
+	if err != nil {
+		return nil, err
+	}
+
+	bufferOverwriteCounter, err := meter.Int64Counter("management.updatechannel.buffer.overwrite.counter",
+		metric.WithUnit("1"),
+		metric.WithDescription("Number of updates overwriting old unsent updates in the buffer"))
+	if err != nil {
+		return nil, err
+	}
+
+	bufferIgnoreCounter, err := meter.Int64Counter("management.updatechannel.buffer.ignore.counter",
+		metric.WithUnit("1"),
+		metric.WithDescription("Number of updates being ignored due to old network serial"))
+	if err != nil {
+		return nil, err
+	}
+
 	return &UpdateChannelMetrics{
 		createChannelDurationMicro:        createChannelDurationMicro,
 		closeChannelDurationMicro:         closeChannelDurationMicro,
@@ -138,6 +162,9 @@ func NewUpdateChannelMetrics(ctx context.Context, meter metric.Meter) (*UpdateCh
 		calcPeerNetworkMapDurationMs:      calcPeerNetworkMapDurationMs,
 		mergeNetworkMapDurationMicro:      mergeNetworkMapDurationMicro,
 		toSyncResponseDurationMicro:       toSyncResponseDurationMicro,
+		bufferPushCounter:                 bufferPushCounter,
+		bufferOverwriteCounter:            bufferOverwriteCounter,
+		bufferIgnoreCounter:               bufferIgnoreCounter,
 		ctx:                               ctx,
 	}, nil
 }
@@ -193,3 +220,18 @@ func (metrics *UpdateChannelMetrics) CountMergeNetworkMapDuration(duration time.
 func (metrics *UpdateChannelMetrics) CountToSyncResponseDuration(duration time.Duration) {
 	metrics.toSyncResponseDurationMicro.Record(metrics.ctx, duration.Microseconds())
 }
+
+// CountBufferPush counts how many buffer push operations are happening on an empty buffer
+func (metrics *UpdateChannelMetrics) CountBufferPush() {
+	metrics.bufferPushCounter.Add(metrics.ctx, 1)
+}
+
+// CountBufferOverwrite counts how many buffer overwrite operations are happening on a non-empty buffer
+func (metrics *UpdateChannelMetrics) CountBufferOverwrite() {
+	metrics.bufferOverwriteCounter.Add(metrics.ctx, 1)
+}
+
+// CountBufferIgnore counts how many buffer ignore operations are happening when a new update is pushed
+func (metrics *UpdateChannelMetrics) CountBufferIgnore() {
+	metrics.bufferIgnoreCounter.Add(metrics.ctx, 1)
+}

management/server/token_mgr.go

@@ -227,7 +227,7 @@ 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, &UpdateMessage{Update: update})
+	m.updateManager.SendImportantUpdate(ctx, peerID, &UpdateMessage{Update: update})
 }
 
 func (m *TimeBasedAuthSecretsManager) pushNewRelayTokens(ctx context.Context, accountID, peerID string) {
@@ -251,7 +251,7 @@ 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, &UpdateMessage{Update: update})
+	m.updateManager.SendImportantUpdate(ctx, peerID, &UpdateMessage{Update: update})
 }
 
 func (m *TimeBasedAuthSecretsManager) extendNetbirdConfig(ctx context.Context, peerID, accountID string, update *proto.SyncResponse) {

management/server/token_mgr_test.go

@@ -121,7 +121,7 @@ func TestTimeBasedAuthSecretsManager_SetupRefresh(t *testing.T) {
 loop:
 	for timeout := time.After(5 * time.Second); ; {
 		select {
-		case update := <-updateChannel:
+		case update := <-updateChannel.Important:
 			updates = append(updates, update)
 		case <-timeout:
 			break loop

management/server/types/networkmap.go

@@ -31,6 +31,33 @@ type NetworkMap struct {
 	ForwardingRules     []*ForwardingRule
 }
 
+// var nmapPool = sync.Pool{
+// 	New: func() any {
+// 		return &NetworkMap{}
+// 	},
+// }
+
+// func GetNetworkMap() *NetworkMap {
+// 	n := nmapPool.Get().(*NetworkMap)
+// 	n.reset()
+// 	return n
+// }
+
+// func PutNetworkMap(n *NetworkMap) {
+// 	nmapPool.Put(n)
+// }
+
+// func (n *NetworkMap) reset() {
+// 	n.Peers = n.Peers[:0]
+// 	n.Network = nil
+// 	n.Routes = n.Routes[:0]
+// 	n.DNSConfig = nbdns.Config{}
+// 	n.OfflinePeers = n.OfflinePeers[:0]
+// 	n.FirewallRules = n.FirewallRules[:0]
+// 	n.RoutesFirewallRules = n.RoutesFirewallRules[:0]
+// 	n.ForwardingRules = n.ForwardingRules[:0]
+// }
+
 func (nm *NetworkMap) Merge(other *NetworkMap) {
 	nm.Peers = mergeUniquePeersByID(nm.Peers, other.Peers)
 	nm.Routes = util.MergeUnique(nm.Routes, other.Routes)

management/server/types/networkmapbuilder.go

@@ -3,7 +3,6 @@ package types
 import (
 	"context"
 	"fmt"
-	"runtime"
 	"slices"
 	"strconv"
 	"strings"
@@ -1058,7 +1057,6 @@ func (b *NetworkMapBuilder) OnPeerAddedIncremental(peerID string) error {
 
 	log.Debugf("NetworkMapBuilder: Added peer %s to cache", peerID)
 
-	runtime.GC()
 	return nil
 }
 

management/server/update_buffer.go

@@ -0,0 +1,106 @@
+package server
+
+import (
+	"context"
+	"fmt"
+	"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
+	overwriteCount int       // Number of overwrites since last Pop
+	lastPopTime    time.Time // Time of last Pop
+}
+
+func NewUpdateBuffer(metrics *telemetry.UpdateChannelMetrics) *UpdateBuffer {
+	ub := &UpdateBuffer{metrics: metrics}
+	ub.cond = sync.NewCond(&ub.mu)
+	return ub
+}
+
+func (b *UpdateBuffer) Push(update *UpdateMessage) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	if b.update == nil || update.Update.NetworkMap.Serial > b.update.Update.NetworkMap.Serial || b.update.Update.NetworkMap.Serial == 0 {
+		if b.update == nil {
+			b.metrics.CountBufferPush()
+		} else {
+			b.metrics.CountBufferOverwrite()
+			b.overwriteCount++
+		}
+
+		b.update = update
+		b.cond.Signal()
+
+		return
+	}
+
+	b.metrics.CountBufferIgnore()
+}
+
+func (b *UpdateBuffer) Pop(ctx context.Context) (*UpdateMessage, int, time.Duration, error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	for b.update == nil && !b.closed {
+		select {
+		case <-ctx.Done():
+			return nil, 0, 0, fmt.Errorf("context cancelled")
+		default:
+		}
+
+		waitCh := make(chan struct{})
+		go func() {
+			select {
+			case <-ctx.Done():
+				b.cond.Broadcast()
+			case <-waitCh:
+				// noop
+			}
+		}()
+		b.cond.Wait()
+		close(waitCh)
+
+		select {
+		case <-ctx.Done():
+			return nil, 0, 0, fmt.Errorf("context cancelled")
+		default:
+		}
+	}
+
+	if b.closed {
+		return nil, 0, 0, fmt.Errorf("buffer closed")
+	}
+
+	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
+	b.overwriteCount = 0
+	b.lastPopTime = now
+
+	return msg, overwrites, timeSinceLastPop, nil
+}
+
+func (b *UpdateBuffer) Close() {
+	b.mu.Lock()
+	b.closed = true
+	b.cond.Broadcast()
+	b.mu.Unlock()
+}

management/server/updatechannel.go

@@ -13,13 +13,33 @@ import (
 
 const channelBufferSize = 100
 
+type UpdateChannel struct {
+	Important  chan *UpdateMessage
+	NetworkMap *UpdateBuffer
+}
+
+func NewUpdateChannel(metrics *telemetry.UpdateChannelMetrics) *UpdateChannel {
+	channel := make(chan *UpdateMessage, channelBufferSize)
+	buffer := NewUpdateBuffer(metrics)
+
+	return &UpdateChannel{
+		Important:  channel,
+		NetworkMap: buffer,
+	}
+}
+
+func (u *UpdateChannel) Close() {
+	close(u.Important)
+	u.NetworkMap.Close()
+}
+
 type UpdateMessage struct {
 	Update *proto.SyncResponse
 }
 
 type PeersUpdateManager struct {
 	// peerChannels is an update channel indexed by Peer.ID
-	peerChannels map[string]chan *UpdateMessage
+	peerChannels map[string]*UpdateChannel
 	// channelsMux keeps the mutex to access peerChannels
 	channelsMux *sync.RWMutex
 	// metrics provides method to collect application metrics
@@ -29,14 +49,14 @@ type PeersUpdateManager struct {
 // NewPeersUpdateManager returns a new instance of PeersUpdateManager
 func NewPeersUpdateManager(metrics telemetry.AppMetrics) *PeersUpdateManager {
 	return &PeersUpdateManager{
-		peerChannels: make(map[string]chan *UpdateMessage),
+		peerChannels: make(map[string]*UpdateChannel),
 		channelsMux:  &sync.RWMutex{},
 		metrics:      metrics,
 	}
 }
 
-// SendUpdate sends update message to the peer's channel
-func (p *PeersUpdateManager) SendUpdate(ctx context.Context, peerID string, update *UpdateMessage) {
+// SendImportantUpdate sends update message to the peer that needs to be received
+func (p *PeersUpdateManager) SendImportantUpdate(ctx context.Context, peerID string, update *UpdateMessage) {
 	start := time.Now()
 	var found, dropped bool
 
@@ -52,19 +72,42 @@ func (p *PeersUpdateManager) SendUpdate(ctx context.Context, peerID string, upda
 	if channel, ok := p.peerChannels[peerID]; ok {
 		found = true
 		select {
-		case channel <- update:
-			log.WithContext(ctx).Debugf("update was sent to channel for peer %s", peerID)
+		case channel.Important <- update:
+			// log.WithContext(ctx).Debugf("update was sent to important channel for peer %s", peerID)
 		default:
 			dropped = true
-			log.WithContext(ctx).Warnf("channel for peer %s is %d full or closed", peerID, len(channel))
+			// log.WithContext(ctx).Warnf("important channel for peer %s is %d full or closed", peerID, len(channel.Important))
 		}
 	} else {
-		log.WithContext(ctx).Debugf("peer %s has no channel", peerID)
+		// log.WithContext(ctx).Debugf("peer %s has no important channel", peerID)
+	}
+}
+
+// SendNetworkMapUpdate sends a network map update to the peer's channel
+func (p *PeersUpdateManager) SendNetworkMapUpdate(ctx context.Context, peerID string, update *UpdateMessage) {
+	start := time.Now()
+	var found, dropped bool
+
+	p.channelsMux.RLock()
+
+	defer func() {
+		p.channelsMux.RUnlock()
+		if p.metrics != nil {
+			p.metrics.UpdateChannelMetrics().CountSendUpdateDuration(time.Since(start), found, dropped)
+		}
+	}()
+
+	if channel, ok := p.peerChannels[peerID]; ok {
+		found = true
+		channel.NetworkMap.Push(update)
+		// log.WithContext(ctx).Debugf("update was sent to network map buffer for peer %s", peerID)
+	} else {
+		// log.WithContext(ctx).Debugf("peer %s has no network map buffer", peerID)
 	}
 }
 
 // CreateChannel creates a go channel for a given peer used to deliver updates relevant to the peer.
-func (p *PeersUpdateManager) CreateChannel(ctx context.Context, peerID string) chan *UpdateMessage {
+func (p *PeersUpdateManager) CreateChannel(ctx context.Context, peerID string) *UpdateChannel {
 	start := time.Now()
 
 	closed := false
@@ -80,21 +123,21 @@ func (p *PeersUpdateManager) CreateChannel(ctx context.Context, peerID string) c
 	if channel, ok := p.peerChannels[peerID]; ok {
 		closed = true
 		delete(p.peerChannels, peerID)
-		close(channel)
+		channel.Close()
 	}
-	// mbragin: todo shouldn't it be more? or configurable?
-	channel := make(chan *UpdateMessage, channelBufferSize)
-	p.peerChannels[peerID] = channel
+
+	newChannel := NewUpdateChannel(p.metrics.UpdateChannelMetrics())
+	p.peerChannels[peerID] = newChannel
 
 	log.WithContext(ctx).Debugf("opened updates channel for a peer %s", peerID)
 
-	return channel
+	return newChannel
 }
 
 func (p *PeersUpdateManager) closeChannel(ctx context.Context, peerID string) {
 	if channel, ok := p.peerChannels[peerID]; ok {
 		delete(p.peerChannels, peerID)
-		close(channel)
+		channel.Close()
 
 		log.WithContext(ctx).Debugf("closed updates channel of a peer %s", peerID)
 		return
@@ -174,3 +217,10 @@ func (p *PeersUpdateManager) HasChannel(peerID string) bool {
 
 	return ok
 }
+
+// GetChannelCount returns the number of active peer channels
+func (p *PeersUpdateManager) GetChannelCount() int {
+	p.channelsMux.RLock()
+	defer p.channelsMux.RUnlock()
+	return len(p.peerChannels)
+}

management/server/updatechannel_test.go

@@ -33,15 +33,15 @@ func TestSendUpdate(t *testing.T) {
 	if _, ok := peersUpdater.peerChannels[peer]; !ok {
 		t.Error("Error creating the channel")
 	}
-	peersUpdater.SendUpdate(context.Background(), peer, update1)
+	peersUpdater.SendImportantUpdate(context.Background(), peer, update1)
 	select {
-	case <-peersUpdater.peerChannels[peer]:
+	case <-peersUpdater.peerChannels[peer].Important:
 	default:
 		t.Error("Update wasn't send")
 	}
 
 	for range [channelBufferSize]int{} {
-		peersUpdater.SendUpdate(context.Background(), peer, update1)
+		peersUpdater.SendImportantUpdate(context.Background(), peer, update1)
 	}
 
 	update2 := &UpdateMessage{Update: &proto.SyncResponse{
@@ -50,13 +50,13 @@ func TestSendUpdate(t *testing.T) {
 		},
 	}}
 
-	peersUpdater.SendUpdate(context.Background(), peer, update2)
+	peersUpdater.SendImportantUpdate(context.Background(), peer, update2)
 	timeout := time.After(5 * time.Second)
 	for range [channelBufferSize]int{} {
 		select {
 		case <-timeout:
 			t.Error("timed out reading previously sent updates")
-		case updateReader := <-peersUpdater.peerChannels[peer]:
+		case updateReader := <-peersUpdater.peerChannels[peer].Important:
 			if updateReader.Update.NetworkMap.Serial == update2.Update.NetworkMap.Serial {
 				t.Error("got the update that shouldn't have been sent")
 			}

management/server/user_test.go

@@ -1366,7 +1366,7 @@ func TestUserAccountPeersUpdate(t *testing.T) {
 	t.Run("creating new regular user with no groups", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldNotReceiveUpdate(t, updMsg)
+			peerShouldNotReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -1389,7 +1389,7 @@ func TestUserAccountPeersUpdate(t *testing.T) {
 	t.Run("updating user with no linked peers", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldNotReceiveUpdate(t, updMsg)
+			peerShouldNotReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -1412,7 +1412,7 @@ func TestUserAccountPeersUpdate(t *testing.T) {
 	t.Run("deleting user with no linked peers", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldNotReceiveUpdate(t, updMsg)
+			peerShouldNotReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -1449,7 +1449,7 @@ func TestUserAccountPeersUpdate(t *testing.T) {
 	t.Run("updating user with linked peers", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg)
+			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
 			close(done)
 		}()
 
@@ -1477,7 +1477,7 @@ func TestUserAccountPeersUpdate(t *testing.T) {
 	t.Run("deleting user with linked peers", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, peer4UpdMsg)
+			peerShouldReceiveUpdate(t, peer4UpdMsg.NetworkMap)
 			close(done)
 		}()
 

shared/management/client/grpc.go

@@ -55,7 +55,8 @@ func NewClient(ctx context.Context, addr string, ourPrivateKey wgtypes.Key, tlsE
 		var err error
 		conn, err = nbgrpc.CreateConnection(ctx, addr, tlsEnabled, wsproxy.ManagementComponent)
 		if err != nil {
-			return fmt.Errorf("create connection: %w", err)
+			log.Printf("createConnection error: %v", err)
+			return err
 		}
 		return nil
 	}

shared/signal/client/grpc.go

@@ -60,7 +60,8 @@ func NewClient(ctx context.Context, addr string, key wgtypes.Key, tlsEnabled boo
 		var err error
 		conn, err = nbgrpc.CreateConnection(ctx, addr, tlsEnabled, wsproxy.SignalComponent)
 		if err != nil {
-			return fmt.Errorf("create connection: %w", err)
+			log.Printf("createConnection error: %v", err)
+			return err
 		}
 		return nil
 	}