increase grpc timeouts

client/firewall/uspfilter/conntrack/common_test.go

@@ -12,7 +12,7 @@ import (
 )
 
 var logger = log.NewFromLogrus(logrus.StandardLogger())
-var flowLogger = netflow.NewManager(context.Background(), nil, []byte{}).GetLogger()
+var flowLogger = netflow.NewManager(context.Background(), nil, []byte{}, nil).GetLogger()
 
 // Memory pressure tests
 func BenchmarkMemoryPressure(b *testing.B) {

client/firewall/uspfilter/uspfilter_test.go

@@ -24,7 +24,7 @@ import (
 )
 
 var logger = log.NewFromLogrus(logrus.StandardLogger())
-var flowLogger = netflow.NewManager(context.Background(), nil, []byte{}).GetLogger()
+var flowLogger = netflow.NewManager(context.Background(), nil, []byte{}, nil).GetLogger()
 
 type IFaceMock struct {
 	SetFilterFunc   func(device.PacketFilter) error

client/internal/acl/manager_test.go

@@ -15,7 +15,7 @@ import (
 	mgmProto "github.com/netbirdio/netbird/management/proto"
 )
 
-var flowLogger = netflow.NewManager(context.Background(), nil, []byte{}).GetLogger()
+var flowLogger = netflow.NewManager(context.Background(), nil, []byte{}, nil).GetLogger()
 
 func TestDefaultManager(t *testing.T) {
 	networkMap := &mgmProto.NetworkMap{

client/internal/dns/server_test.go

@@ -31,7 +31,7 @@ import (
 	"github.com/netbirdio/netbird/formatter"
 )
 
-var flowLogger = netflow.NewManager(context.Background(), nil, []byte{}).GetLogger()
+var flowLogger = netflow.NewManager(context.Background(), nil, []byte{}, nil).GetLogger()
 
 type mocWGIface struct {
 	filter device.PacketFilter

client/internal/engine.go

@@ -353,7 +353,7 @@ func (e *Engine) Start() error {
 
 	// start flow manager right after interface creation
 	publicKey := e.config.WgPrivateKey.PublicKey()
-	e.flowManager = netflow.NewManager(e.ctx, e.wgInterface, publicKey[:])
+	e.flowManager = netflow.NewManager(e.ctx, e.wgInterface, publicKey[:], e.statusRecorder)
 
 	if e.config.RosenpassEnabled {
 		log.Infof("rosenpass is enabled")

client/internal/netflow/logger/logger.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/netbirdio/netbird/client/internal/netflow/store"
 	"github.com/netbirdio/netbird/client/internal/netflow/types"
+	"github.com/netbirdio/netbird/client/internal/peer"
 )
 
 type rcvChan chan *types.EventFields
@@ -21,15 +22,17 @@ type Logger struct {
 	enabled        atomic.Bool
 	rcvChan        atomic.Pointer[rcvChan]
 	cancelReceiver context.CancelFunc
+	statusRecorder *peer.Status
 	Store          types.Store
 }
 
-func New(ctx context.Context) *Logger {
+func New(ctx context.Context, statusRecorder *peer.Status) *Logger {
 	ctx, cancel := context.WithCancel(ctx)
 	return &Logger{
-		ctx:    ctx,
-		cancel: cancel,
-		Store:  store.NewMemoryStore(),
+		ctx:            ctx,
+		cancel:         cancel,
+		statusRecorder: statusRecorder,
+		Store:          store.NewMemoryStore(),
 	}
 }
 
@@ -80,6 +83,9 @@ func (l *Logger) startReceiver() {
 				EventFields: *eventFields,
 				Timestamp:   time.Now(),
 			}
+			srcResId, dstResId := l.statusRecorder.CheckRoutes(event.SourceIP, event.DestIP, event.Direction)
+			event.SourceResourceID = []byte(srcResId)
+			event.DestResourceID = []byte(dstResId)
 			l.Store.StoreEvent(&event)
 		}
 	}

client/internal/netflow/logger/logger_test.go

@@ -12,7 +12,7 @@ import (
 )
 
 func TestStore(t *testing.T) {
-	logger := logger.New(context.Background())
+	logger := logger.New(context.Background(), nil)
 	logger.Enable()
 
 	event := types.EventFields{

client/internal/netflow/manager.go

@@ -15,6 +15,7 @@ import (
 	"github.com/netbirdio/netbird/client/internal/netflow/conntrack"
 	"github.com/netbirdio/netbird/client/internal/netflow/logger"
 	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
+	"github.com/netbirdio/netbird/client/internal/peer"
 	"github.com/netbirdio/netbird/flow/client"
 	"github.com/netbirdio/netbird/flow/proto"
 )
@@ -31,8 +32,8 @@ type Manager struct {
 }
 
 // NewManager creates a new netflow manager
-func NewManager(ctx context.Context, iface nftypes.IFaceMapper, publicKey []byte) *Manager {
-	flowLogger := logger.New(ctx)
+func NewManager(ctx context.Context, iface nftypes.IFaceMapper, publicKey []byte, statusRecorder *peer.Status) *Manager {
+	flowLogger := logger.New(ctx, statusRecorder)
 
 	var ct nftypes.ConnTracker
 	if runtime.GOOS == "linux" && iface != nil && !iface.IsUserspaceBind() {
@@ -197,17 +198,19 @@ func toProtoEvent(publicKey []byte, event *nftypes.Event) *proto.FlowEvent {
 		Timestamp: timestamppb.New(event.Timestamp),
 		PublicKey: publicKey,
 		FlowFields: &proto.FlowFields{
-			FlowId:    event.FlowID[:],
-			RuleId:    event.RuleID,
-			Type:      proto.Type(event.Type),
-			Direction: proto.Direction(event.Direction),
-			Protocol:  uint32(event.Protocol),
-			SourceIp:  event.SourceIP.AsSlice(),
-			DestIp:    event.DestIP.AsSlice(),
-			RxPackets: event.RxPackets,
-			TxPackets: event.TxPackets,
-			RxBytes:   event.RxBytes,
-			TxBytes:   event.TxBytes,
+			FlowId:           event.FlowID[:],
+			RuleId:           event.RuleID,
+			Type:             proto.Type(event.Type),
+			Direction:        proto.Direction(event.Direction),
+			Protocol:         uint32(event.Protocol),
+			SourceIp:         event.SourceIP.AsSlice(),
+			DestIp:           event.DestIP.AsSlice(),
+			RxPackets:        event.RxPackets,
+			TxPackets:        event.TxPackets,
+			RxBytes:          event.RxBytes,
+			TxBytes:          event.TxBytes,
+			SourceResourceId: event.SourceResourceID,
+			DestResourceId:   event.DestResourceID,
 		},
 	}
 

client/internal/netflow/types/types.go

@@ -70,21 +70,23 @@ type Event struct {
 }
 
 type EventFields struct {
-	FlowID     uuid.UUID
-	Type       Type
-	RuleID     []byte
-	Direction  Direction
-	Protocol   Protocol
-	SourceIP   netip.Addr
-	DestIP     netip.Addr
-	SourcePort uint16
-	DestPort   uint16
-	ICMPType   uint8
-	ICMPCode   uint8
-	RxPackets  uint64
-	TxPackets  uint64
-	RxBytes    uint64
-	TxBytes    uint64
+	FlowID           uuid.UUID
+	Type             Type
+	RuleID           []byte
+	Direction        Direction
+	Protocol         Protocol
+	SourceIP         netip.Addr
+	DestIP           netip.Addr
+	SourceResourceID []byte
+	DestResourceID   []byte
+	SourcePort       uint16
+	DestPort         uint16
+	ICMPType         uint8
+	ICMPCode         uint8
+	RxPackets        uint64
+	TxPackets        uint64
+	RxBytes          uint64
+	TxBytes          uint64
 }
 
 type FlowConfig struct {

client/internal/peer/route.go

@@ -0,0 +1,100 @@
+package peer
+
+import (
+	"net/netip"
+	"sync"
+
+	log "github.com/sirupsen/logrus"
+
+	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
+)
+
+type routeIDLookup struct {
+	localMap    sync.Map
+	remoteMap   sync.Map
+	resolvedIPs sync.Map
+}
+
+func (r *routeIDLookup) AddLocalRouteID(resourceID string, route netip.Prefix) {
+	_, exists := r.localMap.LoadOrStore(route, resourceID)
+	if exists {
+		log.Tracef("resourceID %s already exists in local map", resourceID)
+	}
+}
+
+func (r *routeIDLookup) RemoveLocalRouteID(route netip.Prefix) {
+	r.localMap.Delete(route)
+}
+
+func (r *routeIDLookup) AddRemoteRouteID(resourceID string, route netip.Prefix) {
+	_, exists := r.remoteMap.LoadOrStore(route, resourceID)
+	if exists {
+		log.Tracef("resourceID %s already exists in remote map", resourceID)
+	}
+}
+
+func (r *routeIDLookup) RemoveRemoteRouteID(route netip.Prefix) {
+	r.remoteMap.Delete(route)
+}
+
+func (r *routeIDLookup) AddResolvedIP(resourceID string, route netip.Prefix) {
+
+	r.resolvedIPs.Store(route.Addr(), resourceID)
+}
+
+func (r *routeIDLookup) RemoveResolvedIP(route netip.Prefix) {
+	r.resolvedIPs.Delete(route.Addr())
+}
+
+func (r *routeIDLookup) Lookup(src, dst netip.Addr, direction nftypes.Direction) (srcResourceID, dstResourceID string) {
+
+	// check resolved ip's first
+	resId, ok := r.resolvedIPs.Load(src)
+	if ok {
+		srcResourceID = resId.(string)
+	} else {
+		resId, ok := r.resolvedIPs.Load(dst)
+		if ok {
+			dstResourceID = resId.(string)
+		}
+	}
+
+	switch direction {
+	case nftypes.Ingress:
+		if srcResourceID == "" || dstResourceID == "" {
+			r.localMap.Range(func(key, value interface{}) bool {
+				if srcResourceID == "" && key.(netip.Prefix).Contains(src) {
+					srcResourceID = value.(string)
+
+				} else if dstResourceID == "" && key.(netip.Prefix).Contains(dst) {
+					dstResourceID = value.(string)
+				}
+
+				if srcResourceID != "" && dstResourceID != "" {
+					return false
+				}
+
+				return true
+			})
+		}
+	case nftypes.Egress:
+		if srcResourceID == "" || dstResourceID == "" {
+			r.remoteMap.Range(func(key, value interface{}) bool {
+				if srcResourceID == "" && key.(netip.Prefix).Contains(src) {
+					srcResourceID = value.(string)
+
+				} else if dstResourceID == "" && key.(netip.Prefix).Contains(dst) {
+					dstResourceID = value.(string)
+				}
+
+				if srcResourceID != "" && dstResourceID != "" {
+					return false
+				}
+
+				return true
+			})
+		}
+	}
+
+	return srcResourceID, dstResourceID
+}

client/internal/peer/status.go

@@ -17,6 +17,7 @@ import (
 	firewall "github.com/netbirdio/netbird/client/firewall/manager"
 	"github.com/netbirdio/netbird/client/iface/configurer"
 	"github.com/netbirdio/netbird/client/internal/ingressgw"
+	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
 	"github.com/netbirdio/netbird/client/internal/relay"
 	"github.com/netbirdio/netbird/client/proto"
 	"github.com/netbirdio/netbird/management/domain"
@@ -176,6 +177,8 @@ type Status struct {
 	eventQueue   *EventQueue
 
 	ingressGwMgr *ingressgw.Manager
+
+	routeIDLookup routeIDLookup
 }
 
 // NewRecorder returns a new Status instance
@@ -311,7 +314,7 @@ func (d *Status) UpdatePeerState(receivedState State) error {
 	return nil
 }
 
-func (d *Status) AddPeerStateRoute(peer string, route string) error {
+func (d *Status) AddPeerStateRoute(peer string, route string, resourceId string) error {
 	d.mux.Lock()
 	defer d.mux.Unlock()
 
@@ -323,6 +326,14 @@ func (d *Status) AddPeerStateRoute(peer string, route string) error {
 	peerState.AddRoute(route)
 	d.peers[peer] = peerState
 
+	pref, err := netip.ParsePrefix(route)
+	if err != nil {
+		log.Errorf("failed to parse prefix %s: %v", route, err)
+	} else {
+
+		d.routeIDLookup.AddRemoteRouteID(resourceId, pref)
+	}
+
 	// todo: consider to make sense of this notification or not
 	d.notifyPeerListChanged()
 	return nil
@@ -340,11 +351,28 @@ func (d *Status) RemovePeerStateRoute(peer string, route string) error {
 	peerState.DeleteRoute(route)
 	d.peers[peer] = peerState
 
+	pref, err := netip.ParsePrefix(route)
+	if err != nil {
+		log.Errorf("failed to parse prefix %s: %v", route, err)
+	} else {
+		d.routeIDLookup.RemoveRemoteRouteID(pref)
+	}
+
 	// todo: consider to make sense of this notification or not
 	d.notifyPeerListChanged()
 	return nil
 }
 
+// CheckRoutes checks if the source and destination addresses are within the same route
+// and returns the resource ID of the route that contains the addresses
+func (d *Status) CheckRoutes(src, dst netip.Addr, direction nftypes.Direction) (srcResId string, dstResId string) {
+	if d == nil {
+		return
+	}
+
+	return d.routeIDLookup.Lookup(src, dst, direction)
+}
+
 func (d *Status) UpdatePeerICEState(receivedState State) error {
 	d.mux.Lock()
 	defer d.mux.Unlock()
@@ -558,6 +586,50 @@ func (d *Status) UpdateLocalPeerState(localPeerState LocalPeerState) {
 	d.notifyAddressChanged()
 }
 
+// AddLocalPeerStateRoute adds a route to the local peer state
+func (d *Status) AddLocalPeerStateRoute(route, resourceId string) {
+	d.mux.Lock()
+	defer d.mux.Unlock()
+
+	pref, err := netip.ParsePrefix(route)
+	if err != nil {
+		log.Errorf("failed to parse prefix %s: %v", route, err)
+		return
+	}
+
+	if d.localPeer.Routes == nil {
+		d.localPeer.Routes = map[string]struct{}{}
+	}
+
+	d.localPeer.Routes[route] = struct{}{}
+
+	d.routeIDLookup.AddLocalRouteID(resourceId, pref)
+}
+
+// RemoveLocalPeerStateRoute removes a route from the local peer state
+func (d *Status) RemoveLocalPeerStateRoute(route string) {
+	d.mux.Lock()
+	defer d.mux.Unlock()
+
+	pref, err := netip.ParsePrefix(route)
+	if err != nil {
+		log.Errorf("failed to parse prefix %s: %v", route, err)
+		return
+	}
+
+	delete(d.localPeer.Routes, route)
+
+	d.routeIDLookup.RemoveLocalRouteID(pref)
+}
+
+// CleanLocalPeerStateRoutes cleans all routes from the local peer state
+func (d *Status) CleanLocalPeerStateRoutes() {
+	d.mux.Lock()
+	defer d.mux.Unlock()
+
+	d.localPeer.Routes = map[string]struct{}{}
+}
+
 // CleanLocalPeerState cleans local peer status
 func (d *Status) CleanLocalPeerState() {
 	d.mux.Lock()
@@ -641,7 +713,7 @@ func (d *Status) UpdateDNSStates(dnsStates []NSGroupState) {
 	d.nsGroupStates = dnsStates
 }
 
-func (d *Status) UpdateResolvedDomainsStates(originalDomain domain.Domain, resolvedDomain domain.Domain, prefixes []netip.Prefix) {
+func (d *Status) UpdateResolvedDomainsStates(originalDomain domain.Domain, resolvedDomain domain.Domain, prefixes []netip.Prefix, resourceId string) {
 	d.mux.Lock()
 	defer d.mux.Unlock()
 
@@ -650,6 +722,10 @@ func (d *Status) UpdateResolvedDomainsStates(originalDomain domain.Domain, resol
 		Prefixes:     prefixes,
 		ParentDomain: originalDomain,
 	}
+
+	for _, prefix := range prefixes {
+		d.routeIDLookup.AddResolvedIP(resourceId, prefix)
+	}
 }
 
 func (d *Status) DeleteResolvedDomainsStates(domain domain.Domain) {
@@ -660,6 +736,10 @@ func (d *Status) DeleteResolvedDomainsStates(domain domain.Domain) {
 	for k, v := range d.resolvedDomainsStates {
 		if v.ParentDomain == domain {
 			delete(d.resolvedDomainsStates, k)
+
+			for _, prefix := range v.Prefixes {
+				d.routeIDLookup.RemoveResolvedIP(prefix)
+			}
 		}
 	}
 }

client/internal/routemanager/client.go

@@ -330,7 +330,7 @@ func (c *clientNetwork) recalculateRouteAndUpdatePeerAndSystem(rsn reason) error
 		c.connectEvent()
 	}
 
-	err := c.statusRecorder.AddPeerStateRoute(c.currentChosen.Peer, c.handler.String())
+	err := c.statusRecorder.AddPeerStateRoute(c.currentChosen.Peer, c.handler.String(), c.currentChosen.GetResourceID())
 	if err != nil {
 		return fmt.Errorf("add peer state route: %w", err)
 	}

client/internal/routemanager/dnsinterceptor/handler.go

@@ -321,7 +321,7 @@ func (d *DnsInterceptor) updateDomainPrefixes(resolvedDomain, originalDomain dom
 	if len(toAdd) > 0 || len(toRemove) > 0 {
 		d.interceptedDomains[resolvedDomain] = newPrefixes
 		originalDomain = domain.Domain(strings.TrimSuffix(string(originalDomain), "."))
-		d.statusRecorder.UpdateResolvedDomainsStates(originalDomain, resolvedDomain, newPrefixes)
+		d.statusRecorder.UpdateResolvedDomainsStates(originalDomain, resolvedDomain, newPrefixes, d.route.GetResourceID())
 
 		if len(toAdd) > 0 {
 			log.Debugf("added dynamic route(s) for domain=%s (pattern: domain=%s): %s",

client/internal/routemanager/dynamic/route.go

@@ -288,7 +288,7 @@ func (r *Route) updateDynamicRoutes(ctx context.Context, newDomains domainMap) e
 		updatedPrefixes := combinePrefixes(oldPrefixes, removedPrefixes, addedPrefixes)
 		r.dynamicDomains[domain] = updatedPrefixes
 
-		r.statusRecorder.UpdateResolvedDomainsStates(domain, domain, updatedPrefixes)
+		r.statusRecorder.UpdateResolvedDomainsStates(domain, domain, updatedPrefixes, r.route.GetResourceID())
 	}
 
 	return nberrors.FormatErrorOrNil(merr)

client/internal/routemanager/server_nonandroid.go

@@ -103,9 +103,7 @@ func (m *serverRouter) removeFromServerNetwork(route *route.Route) error {
 
 	delete(m.routes, route.ID)
 
-	state := m.statusRecorder.GetLocalPeerState()
-	delete(state.Routes, route.Network.String())
-	m.statusRecorder.UpdateLocalPeerState(state)
+	m.statusRecorder.RemoveLocalPeerStateRoute(route.Network.String())
 
 	return nil
 }
@@ -131,18 +129,12 @@ func (m *serverRouter) addToServerNetwork(route *route.Route) error {
 
 	m.routes[route.ID] = route
 
-	state := m.statusRecorder.GetLocalPeerState()
-	if state.Routes == nil {
-		state.Routes = map[string]struct{}{}
-	}
-
 	routeStr := route.Network.String()
 	if route.IsDynamic() {
 		routeStr = route.Domains.SafeString()
 	}
-	state.Routes[routeStr] = struct{}{}
 
-	m.statusRecorder.UpdateLocalPeerState(state)
+	m.statusRecorder.AddLocalPeerStateRoute(routeStr, route.GetResourceID())
 
 	return nil
 }
@@ -164,9 +156,7 @@ func (m *serverRouter) cleanUp() {
 
 	}
 
-	state := m.statusRecorder.GetLocalPeerState()
-	state.Routes = nil
-	m.statusRecorder.UpdateLocalPeerState(state)
+	m.statusRecorder.CleanLocalPeerStateRoutes()
 }
 
 func routeToRouterPair(route *route.Route) (firewall.RouterPair, error) {

flow/proto/flow.pb.go

@@ -278,6 +278,9 @@ type FlowFields struct {
 	// Number of bytes
 	RxBytes uint64 `protobuf:"varint,12,opt,name=rx_bytes,json=rxBytes,proto3" json:"rx_bytes,omitempty"`
 	TxBytes uint64 `protobuf:"varint,13,opt,name=tx_bytes,json=txBytes,proto3" json:"tx_bytes,omitempty"`
+	// Resource ID
+	SourceResourceId []byte `protobuf:"bytes,14,opt,name=source_resource_id,json=sourceResourceId,proto3" json:"source_resource_id,omitempty"`
+	DestResourceId   []byte `protobuf:"bytes,15,opt,name=dest_resource_id,json=destResourceId,proto3" json:"dest_resource_id,omitempty"`
 }
 
 func (x *FlowFields) Reset() {
@@ -410,6 +413,20 @@ func (x *FlowFields) GetTxBytes() uint64 {
 	return 0
 }
 
+func (x *FlowFields) GetSourceResourceId() []byte {
+	if x != nil {
+		return x.SourceResourceId
+	}
+	return nil
+}
+
+func (x *FlowFields) GetDestResourceId() []byte {
+	if x != nil {
+		return x.DestResourceId
+	}
+	return nil
+}
+
 type isFlowFields_ConnectionInfo interface {
 	isFlowFields_ConnectionInfo()
 }
@@ -560,7 +577,7 @@ var file_flow_proto_rawDesc = []byte{
 	0x6f, 0x77, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x73, 0x22, 0x29, 0x0a, 0x0c, 0x46, 0x6c, 0x6f, 0x77,
 	0x45, 0x76, 0x65, 0x6e, 0x74, 0x41, 0x63, 0x6b, 0x12, 0x19, 0x0a, 0x08, 0x65, 0x76, 0x65, 0x6e,
 	0x74, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x07, 0x65, 0x76, 0x65, 0x6e,
-	0x74, 0x49, 0x64, 0x22, 0xc4, 0x03, 0x0a, 0x0a, 0x46, 0x6c, 0x6f, 0x77, 0x46, 0x69, 0x65, 0x6c,
+	0x74, 0x49, 0x64, 0x22, 0x9c, 0x04, 0x0a, 0x0a, 0x46, 0x6c, 0x6f, 0x77, 0x46, 0x69, 0x65, 0x6c,
 	0x64, 0x73, 0x12, 0x17, 0x0a, 0x07, 0x66, 0x6c, 0x6f, 0x77, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20,
 	0x01, 0x28, 0x0c, 0x52, 0x06, 0x66, 0x6c, 0x6f, 0x77, 0x49, 0x64, 0x12, 0x1e, 0x0a, 0x04, 0x74,
 	0x79, 0x70, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x0a, 0x2e, 0x66, 0x6c, 0x6f, 0x77,
@@ -587,31 +604,36 @@ var file_flow_proto_rawDesc = []byte{
 	0x73, 0x12, 0x19, 0x0a, 0x08, 0x72, 0x78, 0x5f, 0x62, 0x79, 0x74, 0x65, 0x73, 0x18, 0x0c, 0x20,
 	0x01, 0x28, 0x04, 0x52, 0x07, 0x72, 0x78, 0x42, 0x79, 0x74, 0x65, 0x73, 0x12, 0x19, 0x0a, 0x08,
 	0x74, 0x78, 0x5f, 0x62, 0x79, 0x74, 0x65, 0x73, 0x18, 0x0d, 0x20, 0x01, 0x28, 0x04, 0x52, 0x07,
-	0x74, 0x78, 0x42, 0x79, 0x74, 0x65, 0x73, 0x42, 0x11, 0x0a, 0x0f, 0x63, 0x6f, 0x6e, 0x6e, 0x65,
-	0x63, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x69, 0x6e, 0x66, 0x6f, 0x22, 0x48, 0x0a, 0x08, 0x50, 0x6f,
-	0x72, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x1f, 0x0a, 0x0b, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
-	0x5f, 0x70, 0x6f, 0x72, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0a, 0x73, 0x6f, 0x75,
-	0x72, 0x63, 0x65, 0x50, 0x6f, 0x72, 0x74, 0x12, 0x1b, 0x0a, 0x09, 0x64, 0x65, 0x73, 0x74, 0x5f,
-	0x70, 0x6f, 0x72, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x08, 0x64, 0x65, 0x73, 0x74,
-	0x50, 0x6f, 0x72, 0x74, 0x22, 0x44, 0x0a, 0x08, 0x49, 0x43, 0x4d, 0x50, 0x49, 0x6e, 0x66, 0x6f,
-	0x12, 0x1b, 0x0a, 0x09, 0x69, 0x63, 0x6d, 0x70, 0x5f, 0x74, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20,
-	0x01, 0x28, 0x0d, 0x52, 0x08, 0x69, 0x63, 0x6d, 0x70, 0x54, 0x79, 0x70, 0x65, 0x12, 0x1b, 0x0a,
-	0x09, 0x69, 0x63, 0x6d, 0x70, 0x5f, 0x63, 0x6f, 0x64, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0d,
-	0x52, 0x08, 0x69, 0x63, 0x6d, 0x70, 0x43, 0x6f, 0x64, 0x65, 0x2a, 0x45, 0x0a, 0x04, 0x54, 0x79,
-	0x70, 0x65, 0x12, 0x10, 0x0a, 0x0c, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x55, 0x4e, 0x4b, 0x4e, 0x4f,
-	0x57, 0x4e, 0x10, 0x00, 0x12, 0x0e, 0x0a, 0x0a, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x53, 0x54, 0x41,
-	0x52, 0x54, 0x10, 0x01, 0x12, 0x0c, 0x0a, 0x08, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x45, 0x4e, 0x44,
-	0x10, 0x02, 0x12, 0x0d, 0x0a, 0x09, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x44, 0x52, 0x4f, 0x50, 0x10,
-	0x03, 0x2a, 0x3b, 0x0a, 0x09, 0x44, 0x69, 0x72, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x15,
-	0x0a, 0x11, 0x44, 0x49, 0x52, 0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x5f, 0x55, 0x4e, 0x4b, 0x4e,
-	0x4f, 0x57, 0x4e, 0x10, 0x00, 0x12, 0x0b, 0x0a, 0x07, 0x49, 0x4e, 0x47, 0x52, 0x45, 0x53, 0x53,
-	0x10, 0x01, 0x12, 0x0a, 0x0a, 0x06, 0x45, 0x47, 0x52, 0x45, 0x53, 0x53, 0x10, 0x02, 0x32, 0x42,
-	0x0a, 0x0b, 0x46, 0x6c, 0x6f, 0x77, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x33, 0x0a,
-	0x06, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x12, 0x0f, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x46,
-	0x6c, 0x6f, 0x77, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x1a, 0x12, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e,
-	0x46, 0x6c, 0x6f, 0x77, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x41, 0x63, 0x6b, 0x22, 0x00, 0x28, 0x01,
-	0x30, 0x01, 0x42, 0x08, 0x5a, 0x06, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x06, 0x70, 0x72,
-	0x6f, 0x74, 0x6f, 0x33,
+	0x74, 0x78, 0x42, 0x79, 0x74, 0x65, 0x73, 0x12, 0x2c, 0x0a, 0x12, 0x73, 0x6f, 0x75, 0x72, 0x63,
+	0x65, 0x5f, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x0e, 0x20,
+	0x01, 0x28, 0x0c, 0x52, 0x10, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x52, 0x65, 0x73, 0x6f, 0x75,
+	0x72, 0x63, 0x65, 0x49, 0x64, 0x12, 0x28, 0x0a, 0x10, 0x64, 0x65, 0x73, 0x74, 0x5f, 0x72, 0x65,
+	0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x0f, 0x20, 0x01, 0x28, 0x0c, 0x52,
+	0x0e, 0x64, 0x65, 0x73, 0x74, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x49, 0x64, 0x42,
+	0x11, 0x0a, 0x0f, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x69, 0x6e,
+	0x66, 0x6f, 0x22, 0x48, 0x0a, 0x08, 0x50, 0x6f, 0x72, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x1f,
+	0x0a, 0x0b, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x70, 0x6f, 0x72, 0x74, 0x18, 0x01, 0x20,
+	0x01, 0x28, 0x0d, 0x52, 0x0a, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x50, 0x6f, 0x72, 0x74, 0x12,
+	0x1b, 0x0a, 0x09, 0x64, 0x65, 0x73, 0x74, 0x5f, 0x70, 0x6f, 0x72, 0x74, 0x18, 0x02, 0x20, 0x01,
+	0x28, 0x0d, 0x52, 0x08, 0x64, 0x65, 0x73, 0x74, 0x50, 0x6f, 0x72, 0x74, 0x22, 0x44, 0x0a, 0x08,
+	0x49, 0x43, 0x4d, 0x50, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x1b, 0x0a, 0x09, 0x69, 0x63, 0x6d, 0x70,
+	0x5f, 0x74, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x08, 0x69, 0x63, 0x6d,
+	0x70, 0x54, 0x79, 0x70, 0x65, 0x12, 0x1b, 0x0a, 0x09, 0x69, 0x63, 0x6d, 0x70, 0x5f, 0x63, 0x6f,
+	0x64, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x08, 0x69, 0x63, 0x6d, 0x70, 0x43, 0x6f,
+	0x64, 0x65, 0x2a, 0x45, 0x0a, 0x04, 0x54, 0x79, 0x70, 0x65, 0x12, 0x10, 0x0a, 0x0c, 0x54, 0x59,
+	0x50, 0x45, 0x5f, 0x55, 0x4e, 0x4b, 0x4e, 0x4f, 0x57, 0x4e, 0x10, 0x00, 0x12, 0x0e, 0x0a, 0x0a,
+	0x54, 0x59, 0x50, 0x45, 0x5f, 0x53, 0x54, 0x41, 0x52, 0x54, 0x10, 0x01, 0x12, 0x0c, 0x0a, 0x08,
+	0x54, 0x59, 0x50, 0x45, 0x5f, 0x45, 0x4e, 0x44, 0x10, 0x02, 0x12, 0x0d, 0x0a, 0x09, 0x54, 0x59,
+	0x50, 0x45, 0x5f, 0x44, 0x52, 0x4f, 0x50, 0x10, 0x03, 0x2a, 0x3b, 0x0a, 0x09, 0x44, 0x69, 0x72,
+	0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x15, 0x0a, 0x11, 0x44, 0x49, 0x52, 0x45, 0x43, 0x54,
+	0x49, 0x4f, 0x4e, 0x5f, 0x55, 0x4e, 0x4b, 0x4e, 0x4f, 0x57, 0x4e, 0x10, 0x00, 0x12, 0x0b, 0x0a,
+	0x07, 0x49, 0x4e, 0x47, 0x52, 0x45, 0x53, 0x53, 0x10, 0x01, 0x12, 0x0a, 0x0a, 0x06, 0x45, 0x47,
+	0x52, 0x45, 0x53, 0x53, 0x10, 0x02, 0x32, 0x42, 0x0a, 0x0b, 0x46, 0x6c, 0x6f, 0x77, 0x53, 0x65,
+	0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x33, 0x0a, 0x06, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x12,
+	0x0f, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x46, 0x6c, 0x6f, 0x77, 0x45, 0x76, 0x65, 0x6e, 0x74,
+	0x1a, 0x12, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x46, 0x6c, 0x6f, 0x77, 0x45, 0x76, 0x65, 0x6e,
+	0x74, 0x41, 0x63, 0x6b, 0x22, 0x00, 0x28, 0x01, 0x30, 0x01, 0x42, 0x08, 0x5a, 0x06, 0x2f, 0x70,
+	0x72, 0x6f, 0x74, 0x6f, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (

flow/proto/flow.proto

@@ -67,6 +67,11 @@ message FlowFields {
   // Number of bytes
   uint64 rx_bytes = 12;
   uint64 tx_bytes = 13;
+
+  // Resource ID
+  bytes source_resource_id = 14;
+  bytes dest_resource_id = 15;
+
 }
 
 // Flow event types

management/client/grpc.go

@@ -272,9 +272,9 @@ func (c *GrpcClient) GetServerPublicKey() (*wgtypes.Key, error) {
 		return nil, errors.New(errMsgNoMgmtConnection)
 	}
 
-	mgmCtx, cancel := context.WithTimeout(c.ctx, 5*time.Second)
-	defer cancel()
-	resp, err := c.realClient.GetServerKey(mgmCtx, &proto.Empty{})
+	// mgmCtx, cancel := context.WithTimeout(c.ctx, 30*time.Second)
+	// defer cancel()
+	resp, err := c.realClient.GetServerKey(c.ctx, &proto.Empty{})
 	if err != nil {
 		log.Errorf("failed while getting Management Service public key: %v", err)
 		return nil, fmt.Errorf("failed while getting Management Service public key")
@@ -327,11 +327,9 @@ func (c *GrpcClient) login(serverKey wgtypes.Key, req *proto.LoginRequest) (*pro
 
 	var resp *proto.EncryptedMessage
 	operation := func() error {
-		mgmCtx, cancel := context.WithTimeout(context.Background(), ConnectTimeout)
-		defer cancel()
 
 		var err error
-		resp, err = c.realClient.Login(mgmCtx, &proto.EncryptedMessage{
+		resp, err = c.realClient.Login(context.Background(), &proto.EncryptedMessage{
 			WgPubKey: c.key.PublicKey().String(),
 			Body:     loginReq,
 		})

management/cmd/management.go

@@ -11,6 +11,7 @@ import (
 	"io/fs"
 	"net"
 	"net/http"
+	_ "net/http/pprof"
 	"net/netip"
 	"net/url"
 	"os"
@@ -34,6 +35,7 @@ import (
 	"github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/realip"
 
 	"github.com/netbirdio/management-integrations/integrations"
+
 	"github.com/netbirdio/netbird/management/server/peers"
 
 	"github.com/netbirdio/netbird/encryption"
@@ -90,6 +92,11 @@ var (
 		PreRunE: func(cmd *cobra.Command, args []string) error {
 			flag.Parse()
 
+			go func() {
+				log.Infof("Starting pprof on :6060")
+				http.ListenAndServe("localhost:6060", nil)
+			}()
+
 			//nolint
 			ctx := context.WithValue(cmd.Context(), hook.ExecutionContextKey, hook.SystemSource)
 
@@ -132,6 +139,7 @@ var (
 			return nil
 		},
 		RunE: func(cmd *cobra.Command, args []string) error {
+
 			flag.Parse()
 
 			ctx, cancel := context.WithCancel(cmd.Context())

management/server/account.go

@@ -68,7 +68,8 @@ type DefaultAccountManager struct {
 	eventStore           activity.Store
 	geo                  geolocation.Geolocation
 
-	requestBuffer *AccountRequestBuffer
+	accountUpdateLocks sync.Map
+	requestBuffer      *AccountRequestBuffer
 
 	proxyController port_forwarding.Controller
 	settingsManager settings.Manager
@@ -1222,7 +1223,7 @@ func (am *DefaultAccountManager) SyncUserJWTGroups(ctx context.Context, userAuth
 
 		if removedGroupAffectsPeers || newGroupsAffectsPeers {
 			log.WithContext(ctx).Tracef("user %s: JWT group membership changed, updating account peers", userAuth.UserId)
-			am.UpdateAccountPeers(ctx, userAuth.AccountId)
+			am.BufferUpdateAccountPeers(ctx, userAuth.AccountId)
 		}
 	}
 
@@ -1461,7 +1462,7 @@ func (am *DefaultAccountManager) GetDNSDomain() string {
 
 func (am *DefaultAccountManager) onPeersInvalidated(ctx context.Context, accountID string) {
 	log.WithContext(ctx).Debugf("validated peers has been invalidated for account %s", accountID)
-	am.UpdateAccountPeers(ctx, accountID)
+	am.BufferUpdateAccountPeers(ctx, accountID)
 }
 
 func (am *DefaultAccountManager) FindExistingPostureCheck(accountID string, checks *posture.ChecksDefinition) (*posture.Checks, error) {

management/server/grpcserver.go

@@ -10,13 +10,12 @@ import (
 	"time"
 
 	pb "github.com/golang/protobuf/proto" // nolint
-	"github.com/golang/protobuf/ptypes/timestamp"
 	"github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/realip"
 	log "github.com/sirupsen/logrus"
 	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
 	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/peer"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/types/known/timestamppb"
 
 	integrationsConfig "github.com/netbirdio/management-integrations/integrations/config"
 	"github.com/netbirdio/netbird/encryption"
@@ -38,6 +37,7 @@ type GRPCServer struct {
 	accountManager  account.Manager
 	settingsManager settings.Manager
 	wgKey           wgtypes.Key
+	wgPubKeySting   string
 	proto.UnimplementedManagementServiceServer
 	peersUpdateManager *PeersUpdateManager
 	config             *Config
@@ -76,7 +76,8 @@ func NewServer(
 	}
 
 	return &GRPCServer{
-		wgKey: key,
+		wgKey:         key,
+		wgPubKeySting: key.PublicKey().String(),
 		// peerKey -> event channel
 		peersUpdateManager: peersUpdateManager,
 		accountManager:     accountManager,
@@ -90,30 +91,16 @@ func NewServer(
 }
 
 func (s *GRPCServer) GetServerKey(ctx context.Context, req *proto.Empty) (*proto.ServerKeyResponse, error) {
-	ip := ""
-	p, ok := peer.FromContext(ctx)
-	if ok {
-		ip = p.Addr.String()
-	}
-
-	log.WithContext(ctx).Tracef("GetServerKey request from %s", ip)
-	start := time.Now()
-	defer func() {
-		log.WithContext(ctx).Tracef("GetServerKey from %s took %v", ip, time.Since(start))
-	}()
-
 	// todo introduce something more meaningful with the key expiration/rotation
 	if s.appMetrics != nil {
 		s.appMetrics.GRPCMetrics().CountGetKeyRequest()
 	}
-	now := time.Now().Add(24 * time.Hour)
-	secs := int64(now.Second())
-	nanos := int32(now.Nanosecond())
-	expiresAt := &timestamp.Timestamp{Seconds: secs, Nanos: nanos}
+
+	expiresAt := time.Now().Add(24 * time.Hour)
 
 	return &proto.ServerKeyResponse{
-		Key:       s.wgKey.PublicKey().String(),
-		ExpiresAt: expiresAt,
+		Key:       s.wgPubKeySting,
+		ExpiresAt: timestamppb.New(expiresAt),
 	}, nil
 }
 
@@ -242,7 +229,7 @@ func (s *GRPCServer) sendUpdate(ctx context.Context, accountID string, peerKey w
 		return status.Errorf(codes.Internal, "failed processing update message")
 	}
 	err = srv.SendMsg(&proto.EncryptedMessage{
-		WgPubKey: s.wgKey.PublicKey().String(),
+		WgPubKey: s.wgPubKeySting,
 		Body:     encryptedResp,
 	})
 	if err != nil {
@@ -500,7 +487,7 @@ func (s *GRPCServer) Login(ctx context.Context, req *proto.EncryptedMessage) (*p
 	}
 
 	return &proto.EncryptedMessage{
-		WgPubKey: s.wgKey.PublicKey().String(),
+		WgPubKey: s.wgPubKeySting,
 		Body:     encryptedResp,
 	}, nil
 }
@@ -713,7 +700,7 @@ func (s *GRPCServer) sendInitialSync(ctx context.Context, peerKey wgtypes.Key, p
 	}
 
 	err = srv.Send(&proto.EncryptedMessage{
-		WgPubKey: s.wgKey.PublicKey().String(),
+		WgPubKey: s.wgPubKeySting,
 		Body:     encryptedResp,
 	})
 
@@ -778,7 +765,7 @@ func (s *GRPCServer) GetDeviceAuthorizationFlow(ctx context.Context, req *proto.
 	}
 
 	return &proto.EncryptedMessage{
-		WgPubKey: s.wgKey.PublicKey().String(),
+		WgPubKey: s.wgPubKeySting,
 		Body:     encryptedResp,
 	}, nil
 }
@@ -830,7 +817,7 @@ func (s *GRPCServer) GetPKCEAuthorizationFlow(ctx context.Context, req *proto.En
 	}
 
 	return &proto.EncryptedMessage{
-		WgPubKey: s.wgKey.PublicKey().String(),
+		WgPubKey: s.wgPubKeySting,
 		Body:     encryptedResp,
 	}, nil
 }

management/server/networks/resources/types/resource.go

@@ -20,9 +20,9 @@ import (
 type NetworkResourceType string
 
 const (
-	host   NetworkResourceType = "host"
-	subnet NetworkResourceType = "subnet"
-	domain NetworkResourceType = "domain"
+	Host   NetworkResourceType = "host"
+	Subnet NetworkResourceType = "subnet"
+	Domain NetworkResourceType = "domain"
 )
 
 func (p NetworkResourceType) String() string {
@@ -66,7 +66,7 @@ func NewNetworkResource(accountID, networkID, name, description, address string,
 
 func (n *NetworkResource) ToAPIResponse(groups []api.GroupMinimum) *api.NetworkResource {
 	addr := n.Prefix.String()
-	if n.Type == domain {
+	if n.Type == Domain {
 		addr = n.Domain
 	}
 
@@ -125,7 +125,7 @@ func (n *NetworkResource) ToRoute(peer *nbpeer.Peer, router *routerTypes.Network
 		AccessControlGroups: nil,
 	}
 
-	if n.Type == host || n.Type == subnet {
+	if n.Type == Host || n.Type == Subnet {
 		r.Network = n.Prefix
 
 		r.NetworkType = route.IPv4Network
@@ -134,7 +134,7 @@ func (n *NetworkResource) ToRoute(peer *nbpeer.Peer, router *routerTypes.Network
 		}
 	}
 
-	if n.Type == domain {
+	if n.Type == Domain {
 		domainList, err := nbDomain.FromStringList([]string{n.Domain})
 		if err != nil {
 			return nil
@@ -157,18 +157,18 @@ func (n *NetworkResource) EventMeta(network *networkTypes.Network) map[string]an
 func GetResourceType(address string) (NetworkResourceType, string, netip.Prefix, error) {
 	if prefix, err := netip.ParsePrefix(address); err == nil {
 		if prefix.Bits() == 32 || prefix.Bits() == 128 {
-			return host, "", prefix, nil
+			return Host, "", prefix, nil
 		}
-		return subnet, "", prefix, nil
+		return Subnet, "", prefix, nil
 	}
 
 	if ip, err := netip.ParseAddr(address); err == nil {
-		return host, "", netip.PrefixFrom(ip, ip.BitLen()), nil
+		return Host, "", netip.PrefixFrom(ip, ip.BitLen()), nil
 	}
 
 	domainRegex := regexp.MustCompile(`^(\*\.)?([a-zA-Z0-9-]+\.)+[a-zA-Z]{2,}$`)
 	if domainRegex.MatchString(address) {
-		return domain, address, netip.Prefix{}, nil
+		return Domain, address, netip.Prefix{}, nil
 	}
 
 	return "", "", netip.Prefix{}, errors.New("not a valid host, subnet, or domain")

management/server/networks/resources/types/resource_test.go

@@ -14,15 +14,15 @@ func TestGetResourceType(t *testing.T) {
 		expectedPrefix netip.Prefix
 	}{
 		// Valid host IPs
-		{"1.1.1.1", host, false, "", netip.MustParsePrefix("1.1.1.1/32")},
-		{"1.1.1.1/32", host, false, "", netip.MustParsePrefix("1.1.1.1/32")},
+		{"1.1.1.1", Host, false, "", netip.MustParsePrefix("1.1.1.1/32")},
+		{"1.1.1.1/32", Host, false, "", netip.MustParsePrefix("1.1.1.1/32")},
 		// Valid subnets
-		{"192.168.1.0/24", subnet, false, "", netip.MustParsePrefix("192.168.1.0/24")},
-		{"10.0.0.0/16", subnet, false, "", netip.MustParsePrefix("10.0.0.0/16")},
+		{"192.168.1.0/24", Subnet, false, "", netip.MustParsePrefix("192.168.1.0/24")},
+		{"10.0.0.0/16", Subnet, false, "", netip.MustParsePrefix("10.0.0.0/16")},
 		// Valid domains
-		{"example.com", domain, false, "example.com", netip.Prefix{}},
-		{"*.example.com", domain, false, "*.example.com", netip.Prefix{}},
-		{"sub.example.com", domain, false, "sub.example.com", netip.Prefix{}},
+		{"example.com", Domain, false, "example.com", netip.Prefix{}},
+		{"*.example.com", Domain, false, "*.example.com", netip.Prefix{}},
+		{"sub.example.com", Domain, false, "sub.example.com", netip.Prefix{}},
 		// Invalid inputs
 		{"invalid", "", true, "", netip.Prefix{}},
 		{"1.1.1.1/abc", "", true, "", netip.Prefix{}},
@@ -32,7 +32,7 @@ func TestGetResourceType(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.input, func(t *testing.T) {
 			result, domain, prefix, err := GetResourceType(tt.input)
-      
+
 			if result != tt.expectedType {
 				t.Errorf("Expected type %v, got %v", tt.expectedType, result)
 			}

management/server/peer.go

@@ -6,7 +6,9 @@ import (
 	b64 "encoding/base64"
 	"fmt"
 	"net"
+	"os"
 	"slices"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -135,7 +137,7 @@ func (am *DefaultAccountManager) MarkPeerConnected(ctx context.Context, peerPubK
 	if expired {
 		// we need to update other peers because when peer login expires all other peers are notified to disconnect from
 		// the expired one. Here we notify them that connection is now allowed again.
-		am.UpdateAccountPeers(ctx, accountID)
+		am.BufferUpdateAccountPeers(ctx, accountID)
 	}
 
 	return nil
@@ -302,7 +304,7 @@ func (am *DefaultAccountManager) UpdatePeer(ctx context.Context, accountID, user
 	}
 
 	if peerLabelChanged || requiresPeerUpdates {
-		am.UpdateAccountPeers(ctx, accountID)
+		am.BufferUpdateAccountPeers(ctx, accountID)
 	} else if sshChanged {
 		am.UpdateAccountPeer(ctx, accountID, peer.ID)
 	}
@@ -383,7 +385,7 @@ func (am *DefaultAccountManager) DeletePeer(ctx context.Context, accountID, peer
 	}
 
 	if updateAccountPeers {
-		am.UpdateAccountPeers(ctx, accountID)
+		am.BufferUpdateAccountPeers(ctx, accountID)
 	}
 
 	return nil
@@ -653,7 +655,7 @@ func (am *DefaultAccountManager) AddPeer(ctx context.Context, setupKey, userID s
 	unlock = nil
 
 	if updateAccountPeers {
-		am.UpdateAccountPeers(ctx, accountID)
+		am.BufferUpdateAccountPeers(ctx, accountID)
 	}
 
 	return am.getValidatedPeerWithMap(ctx, false, accountID, newPeer)
@@ -748,7 +750,7 @@ func (am *DefaultAccountManager) SyncPeer(ctx context.Context, sync types.PeerSy
 	}
 
 	if isStatusChanged || sync.UpdateAccountPeers || (updated && len(postureChecks) > 0) {
-		am.UpdateAccountPeers(ctx, accountID)
+		am.BufferUpdateAccountPeers(ctx, accountID)
 	}
 
 	return am.getValidatedPeerWithMap(ctx, peerNotValid, accountID, peer)
@@ -893,7 +895,7 @@ func (am *DefaultAccountManager) LoginPeer(ctx context.Context, login types.Peer
 	unlockPeer = nil
 
 	if updateRemotePeers || isStatusChanged || (isPeerUpdated && len(postureChecks) > 0) {
-		am.UpdateAccountPeers(ctx, accountID)
+		am.BufferUpdateAccountPeers(ctx, accountID)
 	}
 
 	return am.getValidatedPeerWithMap(ctx, isRequiresApproval, accountID, peer)
@@ -1226,6 +1228,28 @@ func (am *DefaultAccountManager) UpdateAccountPeers(ctx context.Context, account
 	}
 }
 
+func (am *DefaultAccountManager) BufferUpdateAccountPeers(ctx context.Context, accountID string) {
+	mu, _ := am.accountUpdateLocks.LoadOrStore(accountID, &sync.Mutex{})
+	lock := mu.(*sync.Mutex)
+
+	if !lock.TryLock() {
+		return
+	}
+
+	go func() {
+		intervalStr := os.Getenv("PEER_UPDATE_INTERVAL_MS")
+		interval, err := strconv.Atoi(intervalStr)
+		if err != nil {
+			interval = 10000
+		}
+		time.Sleep(time.Duration(interval) * time.Millisecond)
+		am.UpdateAccountPeers(ctx, accountID)
+
+		am.accountUpdateLocks.Delete(accountID)
+		lock.Unlock()
+	}()
+}
+
 // UpdateAccountPeer updates a single peer that belongs to an account.
 // Should be called when changes need to be synced to a specific peer only.
 func (am *DefaultAccountManager) UpdateAccountPeer(ctx context.Context, accountId string, peerId string) {

management/server/updatechannel.go

@@ -57,8 +57,13 @@ func (p *PeersUpdateManager) SendUpdate(ctx context.Context, peerID string, upda
 		case channel <- update:
 			log.WithContext(ctx).Debugf("update was sent to channel for peer %s", peerID)
 		default:
-			dropped = true
-			log.WithContext(ctx).Warnf("channel for peer %s is %d full or closed", peerID, len(channel))
+			select {
+			case <-channel:
+				log.WithContext(ctx).Trace("dropped oldest message from channel for peer %s", peerID)
+			default:
+				channel <- update
+				log.WithContext(ctx).Debugf("update was sent to channel for peer %s", peerID)
+			}
 		}
 	} else {
 		log.WithContext(ctx).Debugf("peer %s has no channel", peerID)
@@ -85,7 +90,7 @@ func (p *PeersUpdateManager) CreateChannel(ctx context.Context, peerID string) c
 		close(channel)
 	}
 	// mbragin: todo shouldn't it be more? or configurable?
-	channel := make(chan *UpdateMessage, channelBufferSize)
+	channel := make(chan *UpdateMessage, 2)
 	p.peerChannels[peerID] = channel
 
 	log.WithContext(ctx).Debugf("opened updates channel for a peer %s", peerID)

util/grpc/dialer.go

@@ -71,18 +71,15 @@ func CreateConnection(addr string, tlsEnabled bool) (*grpc.ClientConn, error) {
 		}))
 	}
 
-	connCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
-	defer cancel()
-
 	conn, err := grpc.DialContext(
-		connCtx,
+		context.Background(),
 		addr,
 		transportOption,
 		WithCustomDialer(),
 		grpc.WithBlock(),
 		grpc.WithKeepaliveParams(keepalive.ClientParameters{
-			Time:    30 * time.Second,
-			Timeout: 10 * time.Second,
+			Time:    10 * time.Minute,
+			Timeout: 5 * time.Minute,
 		}),
 	)
 	if err != nil {