getValidatedPeerWithMap get account from cache for exp

client/grpc/dialer.go

@@ -4,12 +4,15 @@ 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"
@@ -17,6 +20,9 @@ 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()
@@ -25,6 +31,26 @@ 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) {
@@ -42,22 +68,24 @@ func CreateConnection(ctx context.Context, addr string, tlsEnabled bool, compone
 		}))
 	}
 
-	connCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
-	defer cancel()
-
-	conn, err := grpc.DialContext(
-		connCtx,
+	conn, err := grpc.NewClient(
 		addr,
 		transportOption,
 		WithCustomDialer(tlsEnabled, component),
-		grpc.WithBlock(),
 		grpc.WithKeepaliveParams(keepalive.ClientParameters{
 			Time:    30 * time.Second,
 			Timeout: 10 * time.Second,
 		}),
 	)
 	if err != nil {
-		log.Printf("DialContext error: %v", err)
+		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()
 		return nil, err
 	}
 

client/grpc/dialer_generic.go

@@ -18,7 +18,7 @@ import (
 	nbnet "github.com/netbirdio/netbird/client/net"
 )
 
-func WithCustomDialer(tlsEnabled bool, component string) grpc.DialOption {
+func WithCustomDialer(_ bool, _ string) grpc.DialOption {
 	return grpc.WithContextDialer(func(ctx context.Context, addr string) (net.Conn, error) {
 		if runtime.GOOS == "linux" {
 			currentUser, err := user.Current()
@@ -36,7 +36,6 @@ func WithCustomDialer(tlsEnabled bool, component 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,8 +17,7 @@ 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 before closing the connection.
+// Close overrides the net.Conn Close method to execute all registered hooks after closing the connection.
 func (c *Conn) Close() error {
 	return closeConn(c.ID, c.Conn)
 }
@@ -29,7 +28,7 @@ type TCPConn struct {
 	ID hooks.ConnectionID
 }
 
-// Close overrides the net.TCPConn Close method to execute all registered hooks before closing the connection.
+// Close overrides the net.TCPConn Close method to execute all registered hooks after closing the connection.
 func (c *TCPConn) Close() error {
 	return closeConn(c.ID, c.TCPConn)
 }
@@ -37,13 +36,16 @@ 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,7 +74,6 @@ 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,6 +30,7 @@ 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)
 	}
 
@@ -64,7 +65,7 @@ func callDialerHooks(ctx context.Context, connID hooks.ConnectionID, address str
 
 	ips, err := resolver.LookupIPAddr(ctx, host)
 	if err != nil {
-		return fmt.Errorf("failed to resolve address %s: %w", address, err)
+		return fmt.Errorf("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 before closing the connection.
+// Close overrides the net.PacketConn Close method to execute all registered hooks after 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 before closing the connection.
+// Close overrides the net.UDPConn Close method to execute all registered hooks after closing the connection.
 func (c *UDPConn) Close() error {
 	defer c.seenAddrs.Clear()
 	return closeConn(c.ID, c.UDPConn)

client/ui/client_ui.go

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

go.mod

@@ -56,7 +56,6 @@ require (
 	github.com/hashicorp/go-multierror v1.1.1
 	github.com/hashicorp/go-secure-stdlib/base62 v0.1.2
 	github.com/hashicorp/go-version v1.6.0
-	github.com/jackc/pgx/v5 v5.5.5
 	github.com/libdns/route53 v1.5.0
 	github.com/libp2p/go-netroute v0.2.1
 	github.com/mdlayher/socket v0.5.1
@@ -184,6 +183,7 @@ require (
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
+	github.com/jackc/pgx/v5 v5.5.5 // indirect
 	github.com/jackc/puddle/v2 v2.2.1 // indirect
 	github.com/jeandeaual/go-locale v0.0.0-20240223122105-ce5225dcaa49 // indirect
 	github.com/jinzhu/inflection v1.0.0 // indirect

management/server/account_test.go

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

management/server/grpcserver.go

@@ -68,7 +68,6 @@ type GRPCServer struct {
 	logBlockedPeers          bool
 	blockPeersWithSameConfig bool
 	integratedPeerValidator  integrated_validator.IntegratedValidator
-	debounce                 int
 
 	syncSem atomic.Int32
 	syncLim int32
@@ -95,7 +94,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(peersUpdateManager.GetChannelCount())
+			return int64(len(peersUpdateManager.peerChannels))
 		})
 		if err != nil {
 			return nil, err
@@ -105,14 +104,6 @@ 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)
@@ -137,7 +128,6 @@ func NewServer(
 		logBlockedPeers:          logBlockedPeers,
 		blockPeersWithSameConfig: blockPeersWithSameConfig,
 		integratedPeerValidator:  integratedPeerValidator,
-		debounce:                 debounce,
 
 		syncLim: syncLim,
 	}, nil
@@ -284,56 +274,14 @@ 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 *UpdateChannel, srv proto.ManagementService_SyncServer) error {
+func (s *GRPCServer) handleUpdates(ctx context.Context, accountID string, peerKey wgtypes.Key, peer *nbpeer.Peer, updates chan *UpdateMessage, 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.Important:
+		case update, open := <-updates:
 			if s.appMetrics != nil {
-				s.appMetrics.GRPCMetrics().UpdateChannelQueueLength(len(updates.Important) + 1)
+				s.appMetrics.GRPCMetrics().UpdateChannelQueueLength(len(updates) + 1)
 			}
 
 			if !open {
@@ -341,24 +289,13 @@ 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
@@ -369,33 +306,6 @@ 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 {

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.NetworkMap)
+			peerShouldNotReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldNotReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			close(done)
 		}()
 

management/server/peer.go

@@ -6,7 +6,6 @@ import (
 	b64 "encoding/base64"
 	"fmt"
 	"net"
-	"runtime"
 	"slices"
 	"strings"
 	"sync"
@@ -1260,6 +1259,7 @@ 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,64 +1285,57 @@ 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
 		}
-		wch <- peer
+
+		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)
 	}
 
-	close(wch)
+	//
+
 	wg.Wait()
 	if am.metrics != nil {
 		am.metrics.AccountManagerMetrics().CountUpdateAccountPeersDuration(time.Since(globalStart))
@@ -1350,11 +1343,9 @@ func (am *DefaultAccountManager) UpdateAccountPeers(ctx context.Context, account
 }
 
 type bufferUpdate struct {
-	mu            sync.Mutex
-	next          *time.Timer
-	update        atomic.Bool
-	requestCount  atomic.Int32
-	lastResetTime atomic.Int64 // Unix timestamp in milliseconds
+	mu     sync.Mutex
+	next   *time.Timer
+	update atomic.Bool
 }
 
 func (am *DefaultAccountManager) BufferUpdateAccountPeers(ctx context.Context, accountID string) {
@@ -1365,7 +1356,6 @@ func (am *DefaultAccountManager) BufferUpdateAccountPeers(ctx context.Context, a
 
 	if !b.mu.TryLock() {
 		b.update.Store(true)
-		b.requestCount.Add(1)
 		return
 	}
 
@@ -1375,41 +1365,18 @@ 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(adaptiveDelay, func() {
+			b.next = time.AfterFunc(time.Duration(am.updateAccountPeersBufferInterval.Load()), func() {
 				am.UpdateAccountPeers(ctx, accountID)
 			})
 			return
 		}
-		b.next.Reset(adaptiveDelay)
+		b.next.Reset(time.Duration(am.updateAccountPeersBufferInterval.Load()))
 	}()
 }
 
@@ -1480,7 +1447,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.SendNetworkMapUpdate(ctx, peer.ID, &UpdateMessage{Update: update})
+	am.peersUpdateManager.SendUpdate(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.
@@ -1677,7 +1644,7 @@ func deletePeers(ctx context.Context, am *DefaultAccountManager, transaction sto
 			return nil, err
 		}
 
-		am.peersUpdateManager.SendImportantUpdate(ctx, peer.ID, &UpdateMessage{
+		am.peersUpdateManager.SendUpdate(ctx, peer.ID, &UpdateMessage{
 			Update: &proto.SyncResponse{
 				RemotePeers:        []*proto.RemotePeerConfig{},
 				RemotePeersIsEmpty: true,

management/server/peer_test.go

@@ -26,6 +26,7 @@ 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"
@@ -953,15 +954,14 @@ func BenchmarkUpdateAccountPeers(b *testing.B) {
 		minMsPerOpCICD  float64
 		maxMsPerOpCICD  float64
 	}{
-		{"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},
+		{"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},
 	}
-	b.Setenv("NB_EXPERIMENT_NETWORK_MAP", "false")
 
 	log.SetOutput(io.Discard)
 	defer log.SetOutput(os.Stderr)
@@ -980,39 +980,38 @@ func BenchmarkUpdateAccountPeers(b *testing.B) {
 				b.Fatalf("Failed to get account: %v", err)
 			}
 
-			peerChannels := make(map[string]*UpdateChannel)
+			peerChannels := make(map[string]chan *UpdateMessage)
 
 			for peerID := range account.Peers {
-				peerChannels[peerID] = &UpdateChannel{
-					Important:  make(chan *UpdateMessage, channelBufferSize),
-					NetworkMap: NewUpdateBuffer(manager.metrics.UpdateChannelMetrics()),
-				}
+				peerChannels[peerID] = make(chan *UpdateMessage, channelBufferSize)
 			}
 
 			manager.peersUpdateManager.peerChannels = peerChannels
 
-			for i := 0; i < 1000; i++ {
-				b.Logf("Run %d", i)
-				manager.UpdateAccountPeers(ctx, account.Id)
-				mm(b)
-				manager.Store.IncrementNetworkSerial(ctx, account.Id)
-			}
+			b.ResetTimer()
+			start := time.Now()
 
 			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)
@@ -1045,30 +1044,22 @@ 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]*UpdateChannel)
+			peerChannels := make(map[string]chan *UpdateMessage)
 
 			for peerID := range account.Peers {
-				peerChannels[peerID] = &UpdateChannel{
-					Important:  make(chan *UpdateMessage, channelBufferSize),
-					NetworkMap: NewUpdateBuffer(metrics.UpdateChannelMetrics()),
-				}
+				peerChannels[peerID] = make(chan *UpdateMessage, channelBufferSize)
 			}
 
 			manager.peersUpdateManager.peerChannels = peerChannels
 			manager.UpdateAccountPeers(ctx, account.Id)
 
 			for _, channel := range peerChannels {
-				update, _, _, _ := channel.NetworkMap.Pop(ctx)
+				update := <-channel
 				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))
@@ -1800,7 +1791,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.NetworkMap)
+			peerShouldNotReceiveUpdate(t, updMsg)
 			close(done)
 		}()
 
@@ -1818,7 +1809,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.NetworkMap) //
+			peerShouldReceiveUpdate(t, updMsg) //
 			close(done)
 		}()
 
@@ -1843,7 +1834,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.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			close(done)
 		}()
 
@@ -1861,7 +1852,7 @@ func TestPeerAccountPeersUpdate(t *testing.T) {
 	t.Run("updating peer label", func(t *testing.T) {
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			close(done)
 		}()
 
@@ -1884,7 +1875,7 @@ func TestPeerAccountPeersUpdate(t *testing.T) {
 		manager.integratedPeerValidator = MockIntegratedValidator{ValidatePeerFunc: requireUpdateFunc}
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			close(done)
 		}()
 
@@ -1906,7 +1897,7 @@ func TestPeerAccountPeersUpdate(t *testing.T) {
 		manager.integratedPeerValidator = MockIntegratedValidator{ValidatePeerFunc: requireNoUpdateFunc}
 		done := make(chan struct{})
 		go func() {
-			peerShouldNotReceiveUpdate(t, updMsg.NetworkMap)
+			peerShouldNotReceiveUpdate(t, updMsg)
 			close(done)
 		}()
 
@@ -1939,7 +1930,7 @@ func TestPeerAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			close(done)
 		}()
 
@@ -1965,7 +1956,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.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			close(done)
 		}()
 
@@ -2003,7 +1994,7 @@ func TestPeerAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			close(done)
 		}()
 
@@ -2029,7 +2020,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.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			close(done)
 		}()
 
@@ -2058,7 +2049,7 @@ func TestPeerAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			close(done)
 		}()
 
@@ -2084,7 +2075,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.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldNotReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldNotReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldNotReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldNotReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldNotReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			close(done)
 		}()
 
@@ -264,7 +264,7 @@ func TestPostureCheckAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldNotReceiveUpdate(t, updMsg)
 			close(done)
 		}()
 
@@ -337,7 +337,7 @@ func TestPostureCheckAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldNotReceiveUpdate(t, updMsg.NetworkMap)
+			peerShouldNotReceiveUpdate(t, updMsg)
 			close(done)
 		}()
 
@@ -381,7 +381,7 @@ func TestPostureCheckAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg1.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg1)
 			close(done)
 		}()
 
@@ -420,7 +420,7 @@ func TestPostureCheckAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldNotReceiveUpdate(t, updMsg)
 			close(done)
 		}()
 
@@ -2034,7 +2034,7 @@ func TestRouteAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			close(done)
 		}()
 
@@ -2095,7 +2095,7 @@ func TestRouteAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			close(done)
 		}()
 
@@ -2149,7 +2149,7 @@ func TestRouteAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			close(done)
 		}()
 
@@ -2189,7 +2189,7 @@ func TestRouteAccountPeersUpdate(t *testing.T) {
 
 		done := make(chan struct{})
 		go func() {
-			peerShouldReceiveUpdate(t, updMsg.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldNotReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldNotReceiveUpdate(t, updMsg)
 			close(done)
 		}()
 

management/server/telemetry/updatechannel_metrics.go

@@ -22,9 +22,6 @@ type UpdateChannelMetrics struct {
 	calcPeerNetworkMapDurationMs      metric.Int64Histogram
 	mergeNetworkMapDurationMicro      metric.Int64Histogram
 	toSyncResponseDurationMicro       metric.Int64Histogram
-	bufferPushCounter                 metric.Int64Counter
-	bufferOverwriteCounter            metric.Int64Counter
-	bufferIgnoreCounter               metric.Int64Counter
 	ctx                               context.Context
 }
 
@@ -128,27 +125,6 @@ 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,
@@ -162,9 +138,6 @@ func NewUpdateChannelMetrics(ctx context.Context, meter metric.Meter) (*UpdateCh
 		calcPeerNetworkMapDurationMs:      calcPeerNetworkMapDurationMs,
 		mergeNetworkMapDurationMicro:      mergeNetworkMapDurationMicro,
 		toSyncResponseDurationMicro:       toSyncResponseDurationMicro,
-		bufferPushCounter:                 bufferPushCounter,
-		bufferOverwriteCounter:            bufferOverwriteCounter,
-		bufferIgnoreCounter:               bufferIgnoreCounter,
 		ctx:                               ctx,
 	}, nil
 }
@@ -220,18 +193,3 @@ 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.SendImportantUpdate(ctx, peerID, &UpdateMessage{Update: update})
+	m.updateManager.SendUpdate(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.SendImportantUpdate(ctx, peerID, &UpdateMessage{Update: update})
+	m.updateManager.SendUpdate(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.Important:
+		case update := <-updateChannel:
 			updates = append(updates, update)
 		case <-timeout:
 			break loop

management/server/types/networkmap.go

@@ -31,33 +31,6 @@ 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/update_buffer.go

@@ -1,106 +0,0 @@
-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,33 +13,13 @@ 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]*UpdateChannel
+	peerChannels map[string]chan *UpdateMessage
 	// channelsMux keeps the mutex to access peerChannels
 	channelsMux *sync.RWMutex
 	// metrics provides method to collect application metrics
@@ -49,14 +29,14 @@ type PeersUpdateManager struct {
 // NewPeersUpdateManager returns a new instance of PeersUpdateManager
 func NewPeersUpdateManager(metrics telemetry.AppMetrics) *PeersUpdateManager {
 	return &PeersUpdateManager{
-		peerChannels: make(map[string]*UpdateChannel),
+		peerChannels: make(map[string]chan *UpdateMessage),
 		channelsMux:  &sync.RWMutex{},
 		metrics:      metrics,
 	}
 }
 
-// SendImportantUpdate sends update message to the peer that needs to be received
-func (p *PeersUpdateManager) SendImportantUpdate(ctx context.Context, peerID string, update *UpdateMessage) {
+// SendUpdate sends update message to the peer's channel
+func (p *PeersUpdateManager) SendUpdate(ctx context.Context, peerID string, update *UpdateMessage) {
 	start := time.Now()
 	var found, dropped bool
 
@@ -72,42 +52,19 @@ func (p *PeersUpdateManager) SendImportantUpdate(ctx context.Context, peerID str
 	if channel, ok := p.peerChannels[peerID]; ok {
 		found = true
 		select {
-		case channel.Important <- update:
-			// log.WithContext(ctx).Debugf("update was sent to important channel for peer %s", peerID)
+		case channel <- update:
+			log.WithContext(ctx).Debugf("update was sent to channel for peer %s", peerID)
 		default:
 			dropped = true
-			// log.WithContext(ctx).Warnf("important channel for peer %s is %d full or closed", peerID, len(channel.Important))
+			log.WithContext(ctx).Warnf("channel for peer %s is %d full or closed", peerID, len(channel))
 		}
 	} else {
-		// 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)
+		log.WithContext(ctx).Debugf("peer %s has no channel", 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) *UpdateChannel {
+func (p *PeersUpdateManager) CreateChannel(ctx context.Context, peerID string) chan *UpdateMessage {
 	start := time.Now()
 
 	closed := false
@@ -123,21 +80,21 @@ func (p *PeersUpdateManager) CreateChannel(ctx context.Context, peerID string) *
 	if channel, ok := p.peerChannels[peerID]; ok {
 		closed = true
 		delete(p.peerChannels, peerID)
-		channel.Close()
+		close(channel)
 	}
-
-	newChannel := NewUpdateChannel(p.metrics.UpdateChannelMetrics())
-	p.peerChannels[peerID] = newChannel
+	// mbragin: todo shouldn't it be more? or configurable?
+	channel := make(chan *UpdateMessage, channelBufferSize)
+	p.peerChannels[peerID] = channel
 
 	log.WithContext(ctx).Debugf("opened updates channel for a peer %s", peerID)
 
-	return newChannel
+	return channel
 }
 
 func (p *PeersUpdateManager) closeChannel(ctx context.Context, peerID string) {
 	if channel, ok := p.peerChannels[peerID]; ok {
 		delete(p.peerChannels, peerID)
-		channel.Close()
+		close(channel)
 
 		log.WithContext(ctx).Debugf("closed updates channel of a peer %s", peerID)
 		return
@@ -217,10 +174,3 @@ 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.SendImportantUpdate(context.Background(), peer, update1)
+	peersUpdater.SendUpdate(context.Background(), peer, update1)
 	select {
-	case <-peersUpdater.peerChannels[peer].Important:
+	case <-peersUpdater.peerChannels[peer]:
 	default:
 		t.Error("Update wasn't send")
 	}
 
 	for range [channelBufferSize]int{} {
-		peersUpdater.SendImportantUpdate(context.Background(), peer, update1)
+		peersUpdater.SendUpdate(context.Background(), peer, update1)
 	}
 
 	update2 := &UpdateMessage{Update: &proto.SyncResponse{
@@ -50,13 +50,13 @@ func TestSendUpdate(t *testing.T) {
 		},
 	}}
 
-	peersUpdater.SendImportantUpdate(context.Background(), peer, update2)
+	peersUpdater.SendUpdate(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].Important:
+		case updateReader := <-peersUpdater.peerChannels[peer]:
 			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.NetworkMap)
+			peerShouldNotReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldNotReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldNotReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldReceiveUpdate(t, updMsg)
 			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.NetworkMap)
+			peerShouldReceiveUpdate(t, peer4UpdMsg)
 			close(done)
 		}()
 

route/route.go

@@ -124,7 +124,6 @@ func (r *Route) EventMeta() map[string]any {
 func (r *Route) Copy() *Route {
 	route := &Route{
 		ID:                  r.ID,
-		AccountID:           r.AccountID,
 		Description:         r.Description,
 		NetID:               r.NetID,
 		Network:             r.Network,

shared/management/client/grpc.go

@@ -55,8 +55,7 @@ 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 {
-			log.Printf("createConnection error: %v", err)
-			return err
+			return fmt.Errorf("create connection: %w", err)
 		}
 		return nil
 	}

shared/signal/client/grpc.go

@@ -60,8 +60,7 @@ 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 {
-			log.Printf("createConnection error: %v", err)
-			return err
+			return fmt.Errorf("create connection: %w", err)
 		}
 		return nil
 	}