Merge remote-tracking branch 'origin/main' into dmitri-event-aggregation

Signed-off-by: Dmitri Dolguikh <dmitri.external@netbird.io>

client/internal/dns/mgmt/mgmt.go

@@ -51,13 +51,20 @@ type cachedRecord struct {
 }
 
 // Resolver caches critical NetBird infrastructure domains.
-// records, refreshing, mgmtDomain and serverDomains are all guarded by mutex.
+// records, refreshing, failedResolves, mgmtDomain and serverDomains are all
+// guarded by mutex.
 type Resolver struct {
 	records       map[dns.Question]*cachedRecord
 	mgmtDomain    *domain.Domain
 	serverDomains *dnsconfig.ServerDomains
 	mutex         sync.RWMutex
 
+	// failedResolves records the last failed initial resolve per domain so a
+	// domain that never resolves isn't retried on every server-domains update
+	// until refreshBackoff elapses. Entries are cleared on success and pruned
+	// to the current server-domains set.
+	failedResolves map[domain.Domain]time.Time
+
 	chain            ChainResolver
 	chainMaxPriority int
 	refreshGroup     singleflight.Group
@@ -76,9 +83,10 @@ type Resolver struct {
 // NewResolver creates a new management domains cache resolver.
 func NewResolver() *Resolver {
 	return &Resolver{
-		records:    make(map[dns.Question]*cachedRecord),
-		refreshing: make(map[dns.Question]*atomic.Bool),
-		cacheTTL:   resolveCacheTTL(),
+		records:        make(map[dns.Question]*cachedRecord),
+		refreshing:     make(map[dns.Question]*atomic.Bool),
+		failedResolves: make(map[domain.Domain]time.Time),
+		cacheTTL:       resolveCacheTTL(),
 	}
 }
 
@@ -173,7 +181,9 @@ func (m *Resolver) continueToNext(w dns.ResponseWriter, r *dns.Msg) {
 
 // AddDomain resolves a domain and stores its A/AAAA records in the cache.
 // A family that resolves NODATA (nil err, zero records) evicts any stale
-// entry for that qtype.
+// entry for that qtype. When one family hard-errors while the other succeeds,
+// the resolved family is still cached but AddDomain returns an error so the
+// caller retries the incomplete resolve rather than treating it as complete.
 func (m *Resolver) AddDomain(ctx context.Context, d domain.Domain) error {
 	dnsName := strings.ToLower(dns.Fqdn(d.PunycodeString()))
 
@@ -203,6 +213,10 @@ func (m *Resolver) AddDomain(ctx context.Context, d domain.Domain) error {
 	log.Debugf("added/updated domain=%s with %d A records and %d AAAA records",
 		d.SafeString(), len(aRecords), len(aaaaRecords))
 
+	if errA != nil || errAAAA != nil {
+		return fmt.Errorf("resolve %s: incomplete, a family failed: %w", d.SafeString(), errors.Join(errA, errAAAA))
+	}
+
 	return nil
 }
 
@@ -462,6 +476,7 @@ func (m *Resolver) RemoveDomain(d domain.Domain) error {
 	delete(m.records, qAAAA)
 	delete(m.refreshing, qA)
 	delete(m.refreshing, qAAAA)
+	delete(m.failedResolves, d)
 
 	log.Debugf("removed domain=%s from cache", d.SafeString())
 	return nil
@@ -505,6 +520,7 @@ func (m *Resolver) UpdateFromServerDomains(ctx context.Context, serverDomains dn
 		allDomains := m.extractDomainsFromServerDomains(updatedServerDomains)
 		currentDomains := m.GetCachedDomains()
 		removedDomains = m.removeStaleDomains(currentDomains, allDomains)
+		m.pruneFailedResolves(allDomains)
 	}
 
 	m.addNewDomains(ctx, newDomains)
@@ -577,13 +593,85 @@ func (m *Resolver) isManagementDomain(domain domain.Domain) bool {
 	return m.mgmtDomain != nil && domain == *m.mgmtDomain
 }
 
-// addNewDomains resolves and caches all domains from the update
+// addNewDomains resolves and caches domains that are not yet in the cache,
+// running the lookups concurrently. Domains already cached are skipped and left
+// to the stale-while-revalidate refresh path, so a sync never re-resolves them
+// synchronously: once NetBird owns the OS resolver the resolve runs through the
+// handler chain and would otherwise dial the managed upstreams under the engine
+// sync lock on every update.
 func (m *Resolver) addNewDomains(ctx context.Context, newDomains domain.List) {
+	var wg sync.WaitGroup
+	seen := make(map[domain.Domain]struct{}, len(newDomains))
 	for _, newDomain := range newDomains {
-		if err := m.AddDomain(ctx, newDomain); err != nil {
-			log.Warnf("failed to add/update domain=%s: %v", newDomain.SafeString(), err)
-		} else {
-			log.Debugf("added/updated management cache domain=%s", newDomain.SafeString())
+		if _, dup := seen[newDomain]; dup {
+			continue
+		}
+		seen[newDomain] = struct{}{}
+
+		if !m.needsResolve(newDomain) {
+			continue
+		}
+
+		wg.Add(1)
+		go func(d domain.Domain) {
+			defer wg.Done()
+			if err := m.AddDomain(ctx, d); err != nil {
+				m.markResolveFailed(d)
+				log.Warnf("failed to add/update domain=%s: %v", d.SafeString(), err)
+				return
+			}
+			m.clearResolveFailed(d)
+			log.Debugf("added/updated management cache domain=%s", d.SafeString())
+		}(newDomain)
+	}
+	wg.Wait()
+}
+
+// needsResolve reports whether d should be resolved now. A recent failed or
+// incomplete resolve gates retries on the backoff even when one family is
+// already cached, so a transiently-failed family is retried instead of being
+// treated as fully resolved. Otherwise a domain with any cached record is left
+// to the stale-while-revalidate refresh path.
+func (m *Resolver) needsResolve(d domain.Domain) bool {
+	dnsName := strings.ToLower(dns.Fqdn(d.PunycodeString()))
+
+	m.mutex.RLock()
+	defer m.mutex.RUnlock()
+
+	if failedAt, ok := m.failedResolves[d]; ok {
+		return time.Since(failedAt) >= refreshBackoff
+	}
+
+	for _, qtype := range []uint16{dns.TypeA, dns.TypeAAAA} {
+		q := dns.Question{Name: dnsName, Qtype: qtype, Qclass: dns.ClassINET}
+		if _, ok := m.records[q]; ok {
+			return false
+		}
+	}
+	return true
+}
+
+func (m *Resolver) markResolveFailed(d domain.Domain) {
+	m.mutex.Lock()
+	m.failedResolves[d] = time.Now()
+	m.mutex.Unlock()
+}
+
+func (m *Resolver) clearResolveFailed(d domain.Domain) {
+	m.mutex.Lock()
+	delete(m.failedResolves, d)
+	m.mutex.Unlock()
+}
+
+// pruneFailedResolves drops failure markers for domains no longer present in
+// the server-domains set, keeping the map bounded to the current set (a
+// failed-only domain has no cached record, so RemoveDomain never sees it).
+func (m *Resolver) pruneFailedResolves(domains domain.List) {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+	for d := range m.failedResolves {
+		if !slices.Contains(domains, d) {
+			delete(m.failedResolves, d)
 		}
 	}
 }

client/internal/dns/mgmt/mgmt_refresh_test.go

@@ -21,6 +21,7 @@ type fakeChain struct {
 	mu       sync.Mutex
 	calls    map[string]int
 	answers  map[string][]dns.RR
+	qErr     map[string]error
 	err      error
 	hasRoot  bool
 	onLookup func()
@@ -30,6 +31,7 @@ func newFakeChain() *fakeChain {
 	return &fakeChain{
 		calls:   map[string]int{},
 		answers: map[string][]dns.RR{},
+		qErr:    map[string]error{},
 		hasRoot: true,
 	}
 }
@@ -47,6 +49,9 @@ func (f *fakeChain) ResolveInternal(ctx context.Context, msg *dns.Msg, maxPriori
 	f.calls[key]++
 	answers := f.answers[key]
 	err := f.err
+	if err == nil {
+		err = f.qErr[key]
+	}
 	onLookup := f.onLookup
 	f.mu.Unlock()
 
@@ -75,6 +80,12 @@ func (f *fakeChain) setAnswer(name string, qtype uint16, ip string) {
 	}
 }
 
+func (f *fakeChain) setErr(name string, qtype uint16, err error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.qErr[name+"|"+dns.TypeToString[qtype]] = err
+}
+
 func (f *fakeChain) callCount(name string, qtype uint16) int {
 	f.mu.Lock()
 	defer f.mu.Unlock()

client/internal/dns/mgmt/mgmt_resolve_test.go

@@ -0,0 +1,183 @@
+package mgmt
+
+import (
+	"context"
+	"errors"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/miekg/dns"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	dnsconfig "github.com/netbirdio/netbird/client/internal/dns/config"
+	"github.com/netbirdio/netbird/shared/management/domain"
+)
+
+// A domain already in the cache must not be re-resolved on a subsequent server
+// domains update; it is left to the stale-while-revalidate refresh path.
+func TestResolver_UpdateFromServerDomains_SkipsCached(t *testing.T) {
+	r := NewResolver()
+	chain := newFakeChain()
+	chain.setAnswer("signal.example.com.", dns.TypeA, "10.0.0.2")
+	r.SetChainResolver(chain, 50)
+
+	sd := dnsconfig.ServerDomains{Signal: domain.Domain("signal.example.com")}
+
+	_, err := r.UpdateFromServerDomains(context.Background(), sd)
+	require.NoError(t, err)
+	require.Equal(t, 1, chain.callCount("signal.example.com.", dns.TypeA),
+		"first update must resolve the domain")
+
+	_, err = r.UpdateFromServerDomains(context.Background(), sd)
+	require.NoError(t, err)
+	assert.Equal(t, 1, chain.callCount("signal.example.com.", dns.TypeA),
+		"cached domain must not be re-resolved on a subsequent update")
+}
+
+// New domains in a single update must resolve concurrently rather than serially.
+func TestResolver_AddNewDomains_ResolvesConcurrently(t *testing.T) {
+	r := NewResolver()
+	chain := newFakeChain()
+
+	var inflight, maxInflight atomic.Int32
+	chain.onLookup = func() {
+		n := inflight.Add(1)
+		for {
+			old := maxInflight.Load()
+			if n <= old || maxInflight.CompareAndSwap(old, n) {
+				break
+			}
+		}
+		time.Sleep(50 * time.Millisecond)
+		inflight.Add(-1)
+	}
+
+	relays := []domain.Domain{"a.example.com", "b.example.com", "c.example.com", "d.example.com"}
+	for _, d := range relays {
+		chain.setAnswer(dns.Fqdn(string(d)), dns.TypeA, "10.0.0.2")
+	}
+	r.SetChainResolver(chain, 50)
+
+	start := time.Now()
+	_, err := r.UpdateFromServerDomains(context.Background(), dnsconfig.ServerDomains{Relay: relays})
+	require.NoError(t, err)
+	elapsed := time.Since(start)
+
+	assert.GreaterOrEqual(t, int(maxInflight.Load()), 2, "domains must resolve concurrently")
+	// Serial resolution of 4 domains would take at least 4*50ms; concurrent is far less.
+	assert.Less(t, elapsed, 300*time.Millisecond, "resolution should not be serial")
+}
+
+// A domain that fails to resolve must not be retried on every update; the
+// failure backoff suppresses re-resolution until it expires.
+func TestResolver_UpdateFromServerDomains_BacksOffFailures(t *testing.T) {
+	r := NewResolver()
+	chain := newFakeChain()
+	chain.err = errors.New("resolve boom")
+	r.SetChainResolver(chain, 50)
+
+	sd := dnsconfig.ServerDomains{Signal: domain.Domain("signal.example.com")}
+
+	_, err := r.UpdateFromServerDomains(context.Background(), sd)
+	require.NoError(t, err)
+	require.Equal(t, 1, chain.callCount("signal.example.com.", dns.TypeA),
+		"first update must attempt the resolve")
+
+	_, err = r.UpdateFromServerDomains(context.Background(), sd)
+	require.NoError(t, err)
+	assert.Equal(t, 1, chain.callCount("signal.example.com.", dns.TypeA),
+		"failed resolve must back off and not retry on the next update")
+}
+
+// A domain listed under more than one server-domain type (e.g. STUN and TURN on
+// the same host) must be resolved once per update, not once per occurrence.
+func TestResolver_AddNewDomains_DedupesDuplicateDomains(t *testing.T) {
+	r := NewResolver()
+	chain := newFakeChain()
+	chain.setAnswer("dup.example.com.", dns.TypeA, "10.0.0.9")
+	r.SetChainResolver(chain, 50)
+
+	sd := dnsconfig.ServerDomains{
+		Stuns: []domain.Domain{"dup.example.com"},
+		Turns: []domain.Domain{"dup.example.com"},
+	}
+
+	_, err := r.UpdateFromServerDomains(context.Background(), sd)
+	require.NoError(t, err)
+	assert.Equal(t, 1, chain.callCount("dup.example.com.", dns.TypeA),
+		"a domain appearing under multiple server-domain types must resolve once")
+}
+
+// A failure marker must be dropped once its domain leaves the server-domains set
+// so the map stays bounded to the current set.
+func TestResolver_UpdateFromServerDomains_PrunesFailedResolves(t *testing.T) {
+	r := NewResolver()
+	chain := newFakeChain()
+	chain.err = errors.New("resolve boom")
+	r.SetChainResolver(chain, 50)
+
+	_, err := r.UpdateFromServerDomains(context.Background(), dnsconfig.ServerDomains{Signal: domain.Domain("gone.example.com")})
+	require.NoError(t, err)
+	r.mutex.RLock()
+	_, marked := r.failedResolves[domain.Domain("gone.example.com")]
+	r.mutex.RUnlock()
+	require.True(t, marked, "failed resolve must be recorded")
+
+	_, err = r.UpdateFromServerDomains(context.Background(), dnsconfig.ServerDomains{Signal: domain.Domain("other.example.com")})
+	require.NoError(t, err)
+	r.mutex.RLock()
+	_, stillMarked := r.failedResolves[domain.Domain("gone.example.com")]
+	r.mutex.RUnlock()
+	assert.False(t, stillMarked, "failure marker for a domain no longer in the set must be pruned")
+}
+
+// When one family hard-errors while the other resolves, the domain is cached
+// for the working family but recorded as incomplete so the failed family is
+// retried under backoff instead of being treated as fully resolved forever.
+func TestResolver_AddNewDomains_RetriesPartialFamilyFailure(t *testing.T) {
+	d := domain.Domain("relay.example.com")
+	r := NewResolver()
+	chain := newFakeChain()
+	chain.setAnswer("relay.example.com.", dns.TypeA, "10.0.0.2")
+	chain.setErr("relay.example.com.", dns.TypeAAAA, errors.New("servfail"))
+	r.SetChainResolver(chain, 50)
+
+	_, err := r.UpdateFromServerDomains(context.Background(), dnsconfig.ServerDomains{Relay: []domain.Domain{d}})
+	require.NoError(t, err)
+
+	r.mutex.RLock()
+	_, aCached := r.records[dns.Question{Name: "relay.example.com.", Qtype: dns.TypeA, Qclass: dns.ClassINET}]
+	_, marked := r.failedResolves[d]
+	r.mutex.RUnlock()
+	require.True(t, aCached, "the working family must still be cached")
+	require.True(t, marked, "a partial failure must be recorded so the failed family is retried")
+
+	assert.False(t, r.needsResolve(d), "within the backoff window the domain is not retried")
+
+	r.mutex.Lock()
+	r.failedResolves[d] = time.Now().Add(-2 * refreshBackoff)
+	r.mutex.Unlock()
+	assert.True(t, r.needsResolve(d), "after the backoff elapses the domain is retried to pick up the missing family")
+}
+
+// A family that returns NODATA (legitimately absent, e.g. an IPv4-only host) is
+// not a failure: the domain must not be marked for retry, otherwise it would be
+// re-resolved on every sync.
+func TestResolver_AddNewDomains_NodataIsNotFailure(t *testing.T) {
+	d := domain.Domain("v4only.example.com")
+	r := NewResolver()
+	chain := newFakeChain()
+	chain.setAnswer("v4only.example.com.", dns.TypeA, "10.0.0.2")
+	r.SetChainResolver(chain, 50)
+
+	_, err := r.UpdateFromServerDomains(context.Background(), dnsconfig.ServerDomains{Relay: []domain.Domain{d}})
+	require.NoError(t, err)
+
+	r.mutex.RLock()
+	_, marked := r.failedResolves[d]
+	r.mutex.RUnlock()
+	assert.False(t, marked, "a NODATA family must not be recorded as a failure")
+	assert.False(t, r.needsResolve(d), "an IPv4-only host must not be re-resolved on later syncs")
+}

client/internal/netflow/logger/logger.go

@@ -27,7 +27,7 @@ type Logger struct {
 	wgIfaceNetV6       netip.Prefix
 	dnsCollection      atomic.Bool
 	exitNodeCollection atomic.Bool
-	Store              types.Store
+	Store              types.AggregatingStore
 }
 
 func New(statusRecorder *peer.Status, wgIfaceIPNet, wgIfaceIPNetV6 netip.Prefix) *Logger {
@@ -35,7 +35,7 @@ func New(statusRecorder *peer.Status, wgIfaceIPNet, wgIfaceIPNetV6 netip.Prefix)
 		statusRecorder: statusRecorder,
 		wgIfaceNet:     wgIfaceIPNet,
 		wgIfaceNetV6:   wgIfaceIPNetV6,
-		Store:          store.NewMemoryStore(),
+		Store:          store.NewAggregatingMemoryStore(),
 	}
 }
 
@@ -125,6 +125,10 @@ func (l *Logger) stop() {
 	l.mux.Unlock()
 }
 
+func (l *Logger) ResetAggregationWindow() types.FlowEventAggregator {
+	return l.Store.ResetAggregationWindow()
+}
+
 func (l *Logger) GetEvents() []*types.Event {
 	return l.Store.GetEvents()
 }

client/internal/netflow/manager.go

@@ -9,12 +9,14 @@ import (
 	"sync"
 	"time"
 
+	"github.com/cenkalti/backoff/v4"
 	"github.com/google/uuid"
 	log "github.com/sirupsen/logrus"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"github.com/netbirdio/netbird/client/internal/netflow/conntrack"
 	"github.com/netbirdio/netbird/client/internal/netflow/logger"
+	"github.com/netbirdio/netbird/client/internal/netflow/store"
 	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
 	"github.com/netbirdio/netbird/client/internal/peer"
 	"github.com/netbirdio/netbird/flow/client"
@@ -23,14 +25,16 @@ import (
 
 // Manager handles netflow tracking and logging
 type Manager struct {
-	mux            sync.Mutex
-	shutdownWg     sync.WaitGroup
-	logger         nftypes.FlowLogger
-	flowConfig     *nftypes.FlowConfig
-	conntrack      nftypes.ConnTracker
-	receiverClient *client.GRPCClient
-	publicKey      []byte
-	cancel         context.CancelFunc
+	mux               sync.Mutex
+	shutdownWg        sync.WaitGroup
+	logger            nftypes.FlowLogger
+	flowConfig        *nftypes.FlowConfig
+	conntrack         nftypes.ConnTracker
+	receiverClient    *client.GRPCClient
+	eventsWithoutAcks nftypes.Store
+	publicKey         []byte
+	cancel            context.CancelFunc
+	retryInterval     time.Duration
 }
 
 // NewManager creates a new netflow manager
@@ -48,9 +52,11 @@ func NewManager(iface nftypes.IFaceMapper, publicKey []byte, statusRecorder *pee
 	}
 
 	return &Manager{
-		logger:    flowLogger,
-		conntrack: ct,
-		publicKey: publicKey,
+		logger:            flowLogger,
+		conntrack:         ct,
+		publicKey:         publicKey,
+		retryInterval:     time.Second,
+		eventsWithoutAcks: store.NewMemoryStore(),
 	}
 }
 
@@ -66,6 +72,7 @@ func (m *Manager) needsNewClient(previous *nftypes.FlowConfig) bool {
 }
 
 // enableFlow starts components for flow tracking
+// must be called under m.mux lock
 func (m *Manager) enableFlow(previous *nftypes.FlowConfig) error {
 	// first make sender ready so events don't pile up
 	if m.needsNewClient(previous) {
@@ -85,6 +92,7 @@ func (m *Manager) enableFlow(previous *nftypes.FlowConfig) error {
 	return nil
 }
 
+// must be called under m.mux lock
 func (m *Manager) resetClient() error {
 	if m.receiverClient != nil {
 		if err := m.receiverClient.Close(); err != nil {
@@ -107,14 +115,19 @@ func (m *Manager) resetClient() error {
 	ctx, cancel := context.WithCancel(context.Background())
 	m.cancel = cancel
 
-	m.shutdownWg.Add(2)
+	m.shutdownWg.Add(3)
+	flowConfigInterval := m.flowConfig.Interval
 	go func() {
 		defer m.shutdownWg.Done()
-		m.receiveACKs(ctx, flowClient)
+		m.receiveACKs(ctx, flowClient, flowConfigInterval)
 	}()
 	go func() {
 		defer m.shutdownWg.Done()
-		m.startSender(ctx)
+		m.startSender(ctx, flowConfigInterval)
+	}()
+	go func() {
+		defer m.shutdownWg.Done()
+		m.startRetries(ctx, flowConfigInterval)
 	}()
 
 	return nil
@@ -198,8 +211,8 @@ func (m *Manager) GetLogger() nftypes.FlowLogger {
 	return m.logger
 }
 
-func (m *Manager) startSender(ctx context.Context) {
-	ticker := time.NewTicker(m.flowConfig.Interval)
+func (m *Manager) startSender(ctx context.Context, flowConfigInterval time.Duration) {
+	ticker := time.NewTicker(flowConfigInterval)
 	defer ticker.Stop()
 
 	for {
@@ -207,27 +220,29 @@ func (m *Manager) startSender(ctx context.Context) {
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
-			events := m.logger.GetEvents()
+			collectedEvents := m.logger.ResetAggregationWindow()
+			events := collectedEvents.GetAggregatedEvents()
 			for _, event := range events {
 				if err := m.send(event); err != nil {
 					log.Errorf("failed to send flow event to server: %v", err)
-					continue
+				} else {
+					log.Tracef("sent flow event: %s", event.ID)
 				}
-				log.Tracef("sent flow event: %s", event.ID)
+				m.eventsWithoutAcks.StoreEvent(event)
 			}
 		}
 	}
 }
 
-func (m *Manager) receiveACKs(ctx context.Context, client *client.GRPCClient) {
-	err := client.Receive(ctx, m.flowConfig.Interval, func(ack *proto.FlowEventAck) error {
+func (m *Manager) receiveACKs(ctx context.Context, client *client.GRPCClient, flowConfigInterval time.Duration) {
+	err := client.Receive(ctx, flowConfigInterval, func(ack *proto.FlowEventAck) error {
 		id, err := uuid.FromBytes(ack.EventId)
 		if err != nil {
 			log.Warnf("failed to convert ack event id to uuid: %v", err)
 			return nil
 		}
 		log.Tracef("received flow event ack: %s", id)
-		m.logger.DeleteEvents([]uuid.UUID{id})
+		m.eventsWithoutAcks.DeleteEvents([]uuid.UUID{id})
 		return nil
 	})
 
@@ -236,6 +251,43 @@ func (m *Manager) receiveACKs(ctx context.Context, client *client.GRPCClient) {
 	}
 }
 
+// We effectively never drop events (see MaxInterval), which makes eventsWithoutAcks unbounded.
+// We may want to limit the max size of the store, and start dropping oldest events when the threshold is reached.
+func (m *Manager) startRetries(ctx context.Context, flowConfigInterval time.Duration) {
+	timer := time.NewTimer(m.retryInterval)
+	retryBackoff := backoff.WithContext(&backoff.ExponentialBackOff{
+		InitialInterval:     1 * time.Second,
+		RandomizationFactor: 0.5,
+		Multiplier:          1.7,
+		MaxInterval:         flowConfigInterval / 2,
+		MaxElapsedTime:      3 * 30 * 24 * time.Hour, // 3 months
+		Stop:                backoff.Stop,
+		Clock:               backoff.SystemClock,
+	}, ctx)
+	defer timer.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-timer.C:
+			for _, e := range m.eventsWithoutAcks.GetEvents() {
+				if e.Timestamp.Add(time.Second).After(time.Now()) {
+					// grace period on retries to avoid early retries
+					// do not retry if the event is less than 1 sec old
+					continue
+				}
+				if err := m.send(e); err != nil {
+					timer = time.NewTimer(retryBackoff.NextBackOff()) //nolint:staticcheck,wastedassign
+					break
+				}
+			}
+			retryBackoff.Reset()
+			timer = time.NewTimer(m.retryInterval)
+		}
+	}
+}
+
 func (m *Manager) send(event *nftypes.Event) error {
 	m.mux.Lock()
 	client := m.receiverClient
@@ -250,9 +302,11 @@ func (m *Manager) send(event *nftypes.Event) error {
 
 func toProtoEvent(publicKey []byte, event *nftypes.Event) *proto.FlowEvent {
 	protoEvent := &proto.FlowEvent{
-		EventId:   event.ID[:],
-		Timestamp: timestamppb.New(event.Timestamp),
-		PublicKey: publicKey,
+		EventId:     event.ID[:],
+		Timestamp:   timestamppb.New(event.Timestamp),
+		PublicKey:   publicKey,
+		WindowStart: timestamppb.New(event.WindowStart),
+		WindowEnd:   timestamppb.New(event.WindowEnd),
 		FlowFields: &proto.FlowFields{
 			FlowId:           event.FlowID[:],
 			RuleId:           event.RuleID,
@@ -267,6 +321,9 @@ func toProtoEvent(publicKey []byte, event *nftypes.Event) *proto.FlowEvent {
 			TxBytes:          event.TxBytes,
 			SourceResourceId: event.SourceResourceID,
 			DestResourceId:   event.DestResourceID,
+			NumOfStarts:      event.NumOfStarts,
+			NumOfEnds:        event.NumOfEnds,
+			NumOfDrops:       event.NumOfDrops,
 		},
 	}
 

client/internal/netflow/manager_integration_test.go

@@ -0,0 +1,291 @@
+package netflow
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"net/netip"
+	"slices"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/netbirdio/netbird/client/iface/wgaddr"
+	"github.com/netbirdio/netbird/client/internal/netflow/types"
+	"github.com/netbirdio/netbird/flow/proto"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"google.golang.org/grpc"
+)
+
+type testServer struct {
+	proto.UnimplementedFlowServiceServer
+	events         chan *proto.FlowEvent
+	acks           chan *proto.FlowEventAck
+	grpcSrv        *grpc.Server
+	addr           string
+	handlerDone    chan struct{} // signaled each time Events() exits
+	handlerStarted chan struct{} // signaled each time Events() begins
+}
+
+func newTestServer(t *testing.T) *testServer {
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+
+	s := &testServer{
+		events:         make(chan *proto.FlowEvent, 100),
+		acks:           make(chan *proto.FlowEventAck, 100),
+		grpcSrv:        grpc.NewServer(),
+		addr:           listener.Addr().String(),
+		handlerDone:    make(chan struct{}, 10),
+		handlerStarted: make(chan struct{}, 10),
+	}
+
+	proto.RegisterFlowServiceServer(s.grpcSrv, s)
+
+	go func() {
+		if err := s.grpcSrv.Serve(listener); err != nil && !errors.Is(err, grpc.ErrServerStopped) {
+			t.Logf("server error: %v", err)
+		}
+	}()
+
+	t.Cleanup(func() {
+		s.grpcSrv.Stop()
+	})
+
+	return s
+}
+
+func (s *testServer) Events(stream proto.FlowService_EventsServer) error {
+	defer func() {
+		select {
+		case s.handlerDone <- struct{}{}:
+		default:
+		}
+	}()
+
+	err := stream.Send(&proto.FlowEventAck{IsInitiator: true})
+	if err != nil {
+		return err
+	}
+
+	select {
+	case s.handlerStarted <- struct{}{}:
+	default:
+	}
+
+	ctx, cancel := context.WithCancel(stream.Context())
+	defer cancel()
+
+	go func() {
+		defer cancel()
+		for {
+			event, err := stream.Recv()
+			if err != nil {
+				return
+			}
+
+			if !event.IsInitiator {
+				select {
+				case s.events <- event:
+				case <-ctx.Done():
+					return
+				}
+			}
+		}
+	}()
+
+	for {
+		select {
+		case ack := <-s.acks:
+			if err := stream.Send(ack); err != nil {
+				return err
+			}
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+}
+
+func TestSendEventReceiveAck(t *testing.T) {
+	_, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	t.Cleanup(cancel)
+
+	server := newTestServer(t)
+	manager := createManager(t, server.addr, 60*time.Second) // set high to prevent retries in this test
+	defer manager.Close()
+
+	assert.Eventually(t, func() bool {
+		select {
+		case <-server.handlerStarted:
+			return true
+		default:
+			return false
+		}
+	}, 3*time.Second, 100*time.Millisecond)
+
+	event1 := types.EventFields{
+		FlowID:    uuid.New(),
+		Type:      types.TypeStart,
+		Direction: types.Ingress,
+		DestIP:    ipAddr("172.16.1.2"),
+		DestPort:  2345,
+		Protocol:  6,
+	}
+	manager.logger.StoreEvent(event1)
+	event2 := types.EventFields{
+		FlowID:    uuid.New(),
+		Type:      types.TypeStart,
+		Direction: types.Ingress,
+		DestIP:    ipAddr("172.16.1.1"),
+		DestPort:  1234,
+		Protocol:  6,
+	}
+	manager.logger.StoreEvent(event2)
+
+	// verify the server received logged events
+	serverSideEvents := make([]*proto.FlowEvent, 0)
+	assert.Eventually(t, func() bool {
+		select {
+		case event := <-server.events:
+			serverSideEvents = append(serverSideEvents, event)
+			if len(serverSideEvents) == 2 {
+				return true
+			}
+		default:
+			if len(serverSideEvents) == 2 {
+				return true
+			}
+		}
+		return false
+	}, 5*time.Second, 100*time.Millisecond)
+
+	serverSideFlowIds := make([]uuid.UUID, 0, 2)
+	slices.Values(serverSideEvents)(func(e *proto.FlowEvent) bool {
+		id, err := uuid.FromBytes(e.FlowFields.FlowId)
+		assert.NoError(t, err)
+		serverSideFlowIds = append(serverSideFlowIds, id)
+		return true
+	})
+	assert.ElementsMatch(t, []uuid.UUID{event1.FlowID, event2.FlowID}, serverSideFlowIds)
+
+	// verify the manager tracks un-acked events
+	unackedEvents := manager.eventsWithoutAcks.GetEvents()
+	assert.Len(t, unackedEvents, 2)
+	flowIds := make([]uuid.UUID, 0)
+	slices.Values(unackedEvents)(func(e *types.Event) bool {
+		flowIds = append(flowIds, e.FlowID)
+		return true
+	})
+	assert.ElementsMatch(t, flowIds, []uuid.UUID{event1.FlowID, event2.FlowID})
+}
+
+// verify handling of retries:
+//   - unacked events are retried
+//   - when acks arrive, events are removed from the un-acked event tracker
+func TestRetryEvents(t *testing.T) {
+	_, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	t.Cleanup(cancel)
+
+	server := newTestServer(t)
+	manager := createManager(t, server.addr, time.Second) // set low to start retries sooner
+	defer manager.Close()
+
+	assert.Eventually(t, func() bool {
+		select {
+		case <-server.handlerStarted:
+			return true
+		default:
+			return false
+		}
+	}, 3*time.Second, 100*time.Millisecond)
+
+	event1 := types.EventFields{
+		FlowID:    uuid.New(),
+		Type:      types.TypeStart,
+		Direction: types.Ingress,
+		DestIP:    ipAddr("172.16.1.2"),
+		DestPort:  2345,
+		Protocol:  6,
+	}
+	manager.logger.StoreEvent(event1)
+	event2 := types.EventFields{
+		FlowID:    uuid.New(),
+		Type:      types.TypeStart,
+		Direction: types.Ingress,
+		DestIP:    ipAddr("172.16.1.1"),
+		DestPort:  1234,
+		Protocol:  6,
+	}
+	manager.logger.StoreEvent(event2)
+
+	// verify the server received retries of logged events
+	serverSideEvents := make([]*proto.FlowEvent, 0)
+	func() {
+		c := time.After(2500 * time.Millisecond)
+		for {
+			select {
+			case event := <-server.events:
+				serverSideEvents = append(serverSideEvents, event)
+			case <-c:
+				return
+			}
+		}
+	}()
+	assert.True(t, len(serverSideEvents) > 2) // must see retries
+
+	uniqueServerSideEvents := make(map[uuid.UUID]*proto.FlowEvent)
+	slices.Values(serverSideEvents)(func(e *proto.FlowEvent) bool {
+		id, err := uuid.FromBytes(e.FlowFields.FlowId)
+		assert.NoError(t, err)
+		uniqueServerSideEvents[id] = e
+		return true
+	})
+	assert.Contains(t, uniqueServerSideEvents, event1.FlowID)
+	assert.Contains(t, uniqueServerSideEvents, event2.FlowID)
+
+	// ack events
+	server.acks <- &proto.FlowEventAck{EventId: uniqueServerSideEvents[event1.FlowID].EventId}
+	server.acks <- &proto.FlowEventAck{EventId: uniqueServerSideEvents[event2.FlowID].EventId}
+
+	assert.EventuallyWithT(t, func(c *assert.CollectT) {
+		unackedEvents := manager.eventsWithoutAcks.GetEvents()
+		assert.Empty(c, unackedEvents)
+
+	}, 3*time.Second, 100*time.Millisecond)
+}
+
+func createManager(t *testing.T, serverAddr string, retryInterval time.Duration) *Manager {
+	t.Helper()
+
+	mockIFace := &mockIFaceMapper{
+		address: wgaddr.Address{
+			Network: netip.MustParsePrefix("192.168.1.1/32"),
+		},
+		isUserspaceBind: true,
+	}
+
+	publicKey := []byte("test-public-key")
+	manager := NewManager(mockIFace, publicKey, nil)
+	manager.retryInterval = retryInterval
+
+	initialConfig := &types.FlowConfig{
+		Enabled:        true,
+		URL:            fmt.Sprintf("http://%s", serverAddr),
+		TokenPayload:   "initial-payload",
+		TokenSignature: "initial-signature",
+		Interval:       500 * time.Millisecond,
+	}
+
+	err := manager.Update(initialConfig)
+	require.NoError(t, err)
+
+	return manager
+}
+
+func ipAddr(a string) netip.Addr {
+	addr, _ := netip.ParseAddr(a)
+	return addr
+}

client/internal/netflow/store/event_aggregation_test.go

@@ -0,0 +1,334 @@
+package store
+
+import (
+	"math/rand"
+	"net/netip"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/netbirdio/netbird/client/internal/netflow/types"
+	"github.com/stretchr/testify/assert"
+)
+
+var random = rand.New(rand.NewSource(time.Now().UnixNano()))
+
+func TestFlowAggregation(t *testing.T) {
+	var protocols = []types.Protocol{types.ICMP, types.ICMPv6, types.TCP, types.UDP}
+	var tests = []struct {
+		description string
+		addresses   [][]netip.Addr
+		dstPort     uint16
+		eventTypes  []types.Type
+	}{
+		{
+			description: "start and stop",
+			addresses:   [][]netip.Addr{{ipAddr("1.1.1.1"), ipAddr("2.2.2.2")}, {ipAddr("3.3.3.3"), ipAddr("2.2.2.2")}},
+			dstPort:     uint16(random.Uint32() >> 16),
+			eventTypes:  []types.Type{types.TypeStart, types.TypeEnd},
+		},
+		{
+			description: "start and drop",
+			addresses:   [][]netip.Addr{{ipAddr("1.1.1.1"), ipAddr("2.2.2.2")}, {ipAddr("3.3.3.3"), ipAddr("2.2.2.2")}},
+			dstPort:     uint16(random.Uint32() >> 16),
+			eventTypes:  []types.Type{types.TypeStart, types.TypeDrop},
+		},
+		{
+			description: "start only",
+			addresses:   [][]netip.Addr{{ipAddr("1.1.1.1"), ipAddr("2.2.2.2")}, {ipAddr("3.3.3.3"), ipAddr("2.2.2.2")}},
+			dstPort:     uint16(random.Uint32() >> 16),
+			eventTypes:  []types.Type{types.TypeStart},
+		},
+		{
+			description: "drop only",
+			addresses:   [][]netip.Addr{{ipAddr("1.1.1.1"), ipAddr("2.2.2.2")}, {ipAddr("3.3.3.3"), ipAddr("2.2.2.2")}},
+			dstPort:     uint16(random.Uint32() >> 16),
+			eventTypes:  []types.Type{types.TypeDrop},
+		}}
+
+	for _, protocol := range protocols {
+		for _, tt := range tests {
+			t.Run(tt.description+" "+protocol.String(), func(t *testing.T) {
+				store := NewAggregatingMemoryStore()
+				store.WindowEnd = time.Now().Add(5 * time.Second)
+
+				allExpected := make([]*types.Event, 0)
+
+				for _, srcAndDst := range tt.addresses {
+					inEvents, expected := generateEvents(srcAndDst[0], srcAndDst[1], tt.dstPort, tt.eventTypes, protocol, types.Ingress, 0, store.WindowStart, store.WindowEnd)
+					for _, e := range inEvents {
+						store.StoreEvent(e)
+					}
+					allExpected = append(allExpected, expected)
+				}
+
+				events := store.GetAggregatedEvents()
+				assert.ElementsMatch(t, events, allExpected)
+			})
+		}
+	}
+}
+
+func TestIcmpEventAggregation(t *testing.T) {
+	var protocols = []types.Protocol{types.ICMP, types.ICMPv6}
+	var icmpTypes = []uint8{1, 2, 3}
+
+	var tests = []struct {
+		description string
+		addresses   [][]netip.Addr
+		eventTypes  []types.Type
+	}{
+		{
+			description: "start and stop",
+			addresses:   [][]netip.Addr{{ipAddr("1.1.1.1"), ipAddr("2.2.2.2")}},
+			eventTypes:  []types.Type{types.TypeStart, types.TypeEnd},
+		},
+		{
+			description: "start and drop",
+			addresses:   [][]netip.Addr{{ipAddr("1.1.1.1"), ipAddr("2.2.2.2")}},
+			eventTypes:  []types.Type{types.TypeStart, types.TypeDrop},
+		},
+		{
+			description: "start only",
+			addresses:   [][]netip.Addr{{ipAddr("1.1.1.1"), ipAddr("2.2.2.2")}},
+			eventTypes:  []types.Type{types.TypeStart},
+		},
+		{
+			description: "drop only",
+			addresses:   [][]netip.Addr{{ipAddr("1.1.1.1"), ipAddr("2.2.2.2")}},
+			eventTypes:  []types.Type{types.TypeDrop},
+		}}
+
+	for _, protocol := range protocols {
+		for _, tt := range tests {
+			t.Run(tt.description+" "+protocol.String(), func(t *testing.T) {
+				store := NewAggregatingMemoryStore()
+				store.WindowEnd = time.Now().Add(5 * time.Second)
+
+				allExpected := make([]*types.Event, 0)
+				for _, icmpType := range icmpTypes {
+					events, expected := generateEvents(tt.addresses[0][0], tt.addresses[0][1], 0, tt.eventTypes, protocol, types.Ingress, icmpType, store.WindowStart, store.WindowEnd)
+					for _, e := range events {
+						store.StoreEvent(e)
+					}
+					allExpected = append(allExpected, expected)
+				}
+				aggregatedEvents := store.GetAggregatedEvents()
+				assert.Len(t, aggregatedEvents, len(allExpected))
+				assert.ElementsMatch(t, aggregatedEvents, allExpected)
+			})
+		}
+	}
+}
+
+func TestFlowAggregationOfUnknownProtocols(t *testing.T) {
+	var tests = []struct {
+		description string
+		addresses   [][]netip.Addr
+		dstPort     uint16
+		eventTypes  []types.Type
+	}{
+		{
+			description: "start and stop",
+			addresses:   [][]netip.Addr{{ipAddr("1.1.1.1"), ipAddr("2.2.2.2")}, {ipAddr("3.3.3.3"), ipAddr("2.2.2.2")}},
+			dstPort:     uint16(random.Uint32() >> 16),
+			eventTypes:  []types.Type{types.TypeStart, types.TypeEnd},
+		},
+		{
+			description: "start and drop",
+			addresses:   [][]netip.Addr{{ipAddr("1.1.1.1"), ipAddr("2.2.2.2")}, {ipAddr("3.3.3.3"), ipAddr("2.2.2.2")}},
+			dstPort:     uint16(random.Uint32() >> 16),
+			eventTypes:  []types.Type{types.TypeStart, types.TypeDrop},
+		},
+		{
+			description: "start only",
+			addresses:   [][]netip.Addr{{ipAddr("1.1.1.1"), ipAddr("2.2.2.2")}, {ipAddr("3.3.3.3"), ipAddr("2.2.2.2")}},
+			dstPort:     uint16(random.Uint32() >> 16),
+			eventTypes:  []types.Type{types.TypeStart},
+		},
+		{
+			description: "drop only",
+			addresses:   [][]netip.Addr{{ipAddr("1.1.1.1"), ipAddr("2.2.2.2")}, {ipAddr("3.3.3.3"), ipAddr("2.2.2.2")}},
+			dstPort:     uint16(random.Uint32() >> 16),
+			eventTypes:  []types.Type{types.TypeDrop},
+		}}
+
+	for _, tt := range tests {
+		t.Run(tt.description+" "+types.ProtocolUnknown.String(), func(t *testing.T) {
+			store := NewAggregatingMemoryStore()
+			store.WindowEnd = time.Now().Add(5 * time.Second)
+
+			allExpected := make([]*types.Event, 0)
+
+			for _, srcAndDst := range tt.addresses {
+				inEvents, expected := generateEventsForUnknownProtocol(srcAndDst[0], srcAndDst[1], tt.dstPort, tt.eventTypes, types.ProtocolUnknown, types.Ingress, store.WindowStart, store.WindowEnd)
+				for _, e := range inEvents {
+					store.StoreEvent(e)
+				}
+				allExpected = append(allExpected, expected...)
+			}
+
+			events := store.GetAggregatedEvents()
+			assert.ElementsMatch(t, events, allExpected)
+		})
+	}
+}
+
+func ipAddr(a string) netip.Addr {
+	addr, _ := netip.ParseAddr(a)
+	return addr
+}
+
+func generateEvents(srcIp, dstIp netip.Addr, dstPort uint16, eventTypes []types.Type, protocol types.Protocol,
+	direction types.Direction, icmpType uint8, windowStart, windowEnd time.Time) ([]*types.Event, *types.Event) {
+	var rxPackets, txPackets, rxBytes, txBytes uint64
+	inEvents := make([]*types.Event, 0)
+	ts := time.Now()
+	flowId := uuid.New()
+	srcPort := uint16(random.Uint32() >> 16)
+
+	for idx, eventType := range eventTypes {
+		e := &types.Event{
+			ID:        uuid.New(),
+			Timestamp: ts.Add(time.Duration(idx) * time.Second),
+			EventFields: types.EventFields{
+				FlowID:           flowId,
+				Type:             eventType,
+				Protocol:         protocol,
+				RuleID:           []byte("rule-id-1"),
+				Direction:        direction,
+				SourceIP:         srcIp,
+				SourcePort:       srcPort,
+				DestIP:           dstIp,
+				DestPort:         dstPort,
+				SourceResourceID: []byte("source-resource-id"),
+				DestResourceID:   []byte("dest-resource-id"),
+				RxPackets:        random.Uint64(),
+				TxPackets:        random.Uint64(),
+				RxBytes:          random.Uint64(),
+				TxBytes:          random.Uint64(),
+			}}
+		rxBytes += e.RxBytes
+		txBytes += e.TxBytes
+		rxPackets += e.RxPackets
+		txPackets += e.TxPackets
+		inEvents = append(inEvents, e)
+		if protocol == types.ICMP || protocol == types.ICMPv6 {
+			e.ICMPType = icmpType
+		}
+	}
+
+	var start, end, drop uint64
+	for _, eventType := range eventTypes {
+		switch eventType {
+		case types.TypeStart:
+			start += 1
+		case types.TypeDrop:
+			drop += 1
+		case types.TypeEnd:
+			end += 1
+		}
+	}
+	aggregatedEvent := &types.Event{
+		ID:          inEvents[0].ID,
+		Timestamp:   inEvents[0].Timestamp,
+		WindowStart: windowStart,
+		WindowEnd:   windowEnd,
+		EventFields: types.EventFields{
+			FlowID:           flowId,
+			Type:             types.TypeUnknown,
+			Protocol:         inEvents[0].Protocol,
+			RuleID:           []byte("rule-id-1"),
+			Direction:        inEvents[0].Direction,
+			SourceIP:         srcIp,
+			SourcePort:       srcPort,
+			DestIP:           dstIp,
+			DestPort:         dstPort,
+			SourceResourceID: []byte("source-resource-id"),
+			DestResourceID:   []byte("dest-resource-id"),
+			RxPackets:        rxPackets,
+			TxPackets:        txPackets,
+			RxBytes:          rxBytes,
+			TxBytes:          txBytes,
+			NumOfStarts:      start,
+			NumOfEnds:        end,
+			NumOfDrops:       drop,
+		}}
+	if protocol == types.ICMP || protocol == types.ICMPv6 {
+		aggregatedEvent.ICMPType = icmpType
+	}
+
+	return inEvents, aggregatedEvent
+}
+
+func generateEventsForUnknownProtocol(srcIp, dstIp netip.Addr, dstPort uint16, eventTypes []types.Type, protocol types.Protocol,
+	direction types.Direction, windowStart, windowEnd time.Time) ([]*types.Event, []*types.Event) {
+	inEvents := make([]*types.Event, 0)
+	expectedEvents := make([]*types.Event, 0)
+
+	ts := time.Now()
+	flowId := uuid.New()
+	srcPort := uint16(random.Uint32() >> 16)
+
+	for idx, eventType := range eventTypes {
+		e := &types.Event{
+			ID:        uuid.New(),
+			Timestamp: ts.Add(time.Duration(idx) * time.Second),
+			EventFields: types.EventFields{
+				FlowID:           flowId,
+				Type:             eventType,
+				Protocol:         protocol,
+				RuleID:           []byte("rule-id-1"),
+				Direction:        direction,
+				SourceIP:         srcIp,
+				SourcePort:       srcPort,
+				DestIP:           dstIp,
+				DestPort:         dstPort,
+				SourceResourceID: []byte("source-resource-id"),
+				DestResourceID:   []byte("dest-resource-id"),
+				RxPackets:        random.Uint64(),
+				TxPackets:        random.Uint64(),
+				RxBytes:          random.Uint64(),
+				TxBytes:          random.Uint64(),
+			}}
+		inEvents = append(inEvents, e)
+
+		var start, end, drop uint64
+		switch eventType {
+		case types.TypeStart:
+			start = 1
+		case types.TypeDrop:
+			drop = 1
+		case types.TypeEnd:
+			end = 1
+		}
+
+		expectedEvents = append(expectedEvents, &types.Event{
+			ID:          e.ID,
+			Timestamp:   e.Timestamp,
+			WindowStart: windowStart,
+			WindowEnd:   windowEnd,
+			EventFields: types.EventFields{
+				FlowID:           flowId,
+				Type:             types.TypeUnknown,
+				Protocol:         e.Protocol,
+				RuleID:           []byte("rule-id-1"),
+				Direction:        e.Direction,
+				SourceIP:         srcIp,
+				SourcePort:       srcPort,
+				DestIP:           dstIp,
+				DestPort:         dstPort,
+				SourceResourceID: []byte("source-resource-id"),
+				DestResourceID:   []byte("dest-resource-id"),
+				RxPackets:        e.RxPackets,
+				TxPackets:        e.TxPackets,
+				RxBytes:          e.RxBytes,
+				TxBytes:          e.TxBytes,
+				NumOfStarts:      start,
+				NumOfEnds:        end,
+				NumOfDrops:       drop,
+			}})
+	}
+
+	return inEvents, expectedEvents
+}

client/internal/netflow/store/memory.go

@@ -1,10 +1,15 @@
 package store
 
 import (
+	"maps"
+	"math/rand"
+	v2 "math/rand/v2"
+	"net/netip"
+	"slices"
 	"sync"
+	"time"
 
 	"github.com/google/uuid"
-
 	"github.com/netbirdio/netbird/client/internal/netflow/types"
 )
 
@@ -19,6 +24,13 @@ type Memory struct {
 	events map[uuid.UUID]*types.Event
 }
 
+type AggregatingMemory struct {
+	Memory
+	WindowStart time.Time
+	WindowEnd   time.Time
+	rnd         *v2.PCG
+}
+
 func (m *Memory) StoreEvent(event *types.Event) {
 	m.mux.Lock()
 	defer m.mux.Unlock()
@@ -48,3 +60,92 @@ func (m *Memory) DeleteEvents(ids []uuid.UUID) {
 		delete(m.events, id)
 	}
 }
+
+func NewAggregatingMemoryStore() *AggregatingMemory {
+	return &AggregatingMemory{WindowStart: time.Now(), Memory: Memory{events: make(map[uuid.UUID]*types.Event)}, rnd: v2.NewPCG(rand.Uint64(), rand.Uint64())}
+}
+
+func (am *AggregatingMemory) ResetAggregationWindow() types.FlowEventAggregator {
+	am.mux.Lock()
+	defer am.mux.Unlock()
+
+	now := time.Now()
+	toret := AggregatingMemory{WindowStart: am.WindowStart, WindowEnd: now, Memory: Memory{events: am.events}}
+
+	am.events = make(map[uuid.UUID]*types.Event)
+	am.WindowStart = now
+
+	return &toret
+}
+
+type aggregationKey struct {
+	srcAddr   netip.Addr
+	destAddr  netip.Addr
+	destPort  uint16
+	direction int
+	protocol  uint8
+	icmpType  uint8
+	unique    uint64 // used to prevent aggregation on non icmp/udp/tcp events
+}
+
+func (am *AggregatingMemory) GetAggregatedEvents() []*types.Event {
+	am.mux.Lock()
+	defer am.mux.Unlock()
+
+	aggregated := make(map[aggregationKey]*types.Event)
+	for _, v := range am.events {
+		lookupKey := aggregationKey{srcAddr: v.SourceIP, destAddr: v.DestIP, destPort: v.DestPort, direction: int(v.Direction), protocol: uint8(v.Protocol), icmpType: v.ICMPType}
+		if _, ok := aggregated[lookupKey]; !ok {
+			event := v.Clone()
+
+			switch event.Type {
+			case types.TypeStart:
+				event.NumOfStarts += 1
+			case types.TypeDrop:
+				event.NumOfDrops += 1
+			case types.TypeEnd:
+				event.NumOfEnds += 1
+			}
+			event.Type = types.TypeUnknown
+
+			// Please note that ICMPCode field isn't propagated by the manager (see flow/proto/flow.pb.go, FlowFields struct)
+			// so the field value in an icmp event in the "aggregated" doesn't matter
+
+			event.WindowStart = am.WindowStart
+			event.WindowEnd = am.WindowEnd
+
+			if event.Protocol != types.ICMP && event.Protocol != types.ICMPv6 && event.Protocol != types.UDP && event.Protocol != types.TCP {
+				lookupKey.unique = am.rnd.Uint64() // to make the lookup key unique so we don't aggregate on it
+			}
+
+			aggregated[lookupKey] = event
+			continue
+		}
+
+		aggregatedEvent := aggregated[lookupKey]
+		if aggregatedEvent.Protocol != types.ICMP && aggregatedEvent.Protocol != types.ICMPv6 && aggregatedEvent.Protocol != types.UDP && aggregatedEvent.Protocol != types.TCP {
+			continue // we don't aggregate this type of events; shouldn't ever get here
+		}
+
+		// track the number of connections, duration?, open and close events?
+		aggregatedEvent.RxBytes += v.RxBytes
+		aggregatedEvent.RxPackets += v.RxPackets
+		aggregatedEvent.TxBytes += v.TxBytes
+		aggregatedEvent.TxPackets += v.TxPackets
+		switch v.Type {
+		case types.TypeStart:
+			aggregatedEvent.NumOfStarts += 1
+		case types.TypeDrop:
+			aggregatedEvent.NumOfDrops += 1
+		case types.TypeEnd:
+			aggregatedEvent.NumOfEnds += 1
+		}
+		if aggregatedEvent.Timestamp.Compare(v.Timestamp) > 0 {
+			aggregatedEvent.Timestamp = v.Timestamp
+			aggregatedEvent.ID = v.ID
+			aggregatedEvent.SourcePort = v.SourcePort
+		}
+	}
+
+	return slices.Collect(maps.Values(aggregated)) // could return an iterator instead here
+}

client/internal/netflow/types/types.go

@@ -2,6 +2,7 @@ package types
 
 import (
 	"net/netip"
+	"slices"
 	"strconv"
 	"time"
 
@@ -69,8 +70,10 @@ const (
 )
 
 type Event struct {
-	ID        uuid.UUID
-	Timestamp time.Time
+	ID          uuid.UUID
+	Timestamp   time.Time
+	WindowStart time.Time
+	WindowEnd   time.Time
 	EventFields
 }
 
@@ -92,6 +95,17 @@ type EventFields struct {
 	TxPackets        uint64
 	RxBytes          uint64
 	TxBytes          uint64
+	NumOfStarts      uint64
+	NumOfEnds        uint64
+	NumOfDrops       uint64
+}
+
+func (e *Event) Clone() *Event {
+	toret := *e
+	toret.RuleID = slices.Clone(e.RuleID)
+	toret.SourceResourceID = slices.Clone(e.SourceResourceID)
+	toret.DestResourceID = slices.Clone(e.DestResourceID)
+	return &toret
 }
 
 type FlowConfig struct {
@@ -114,13 +128,15 @@ type FlowManager interface {
 	GetLogger() FlowLogger
 }
 
+type FlowEventAggregator interface {
+	ResetAggregationWindow() FlowEventAggregator
+	GetAggregatedEvents() []*Event
+}
+
 type FlowLogger interface {
+	ResetAggregationWindow() FlowEventAggregator
 	// StoreEvent stores a flow event
 	StoreEvent(flowEvent EventFields)
-	// GetEvents returns all stored events
-	GetEvents() []*Event
-	// DeleteEvents deletes events from the store
-	DeleteEvents([]uuid.UUID)
 	// Close closes the logger
 	Close()
 	// Enable enables the flow logger receiver
@@ -140,6 +156,11 @@ type Store interface {
 	Close()
 }
 
+type AggregatingStore interface {
+	FlowEventAggregator
+	Store
+}
+
 // ConnTracker defines the interface for connection tracking functionality
 type ConnTracker interface {
 	// Start begins tracking connections by listening for conntrack events.

client/ui/client_ui.go

@@ -418,7 +418,14 @@ func newServiceClient(args *newServiceClientArgs) *serviceClient {
 	case args.showProfiles:
 		s.showProfilesUI()
 	case args.showQuickActions:
-		s.showQuickActionsUI()
+		// Suppress the on-boot Quick Actions popup when the daemon
+		// reports DisableAutoConnect=true — that flag carries both the
+		// user's "Connect on Startup = off" preference AND any MDM-
+		// enforced override (applyMDMPolicy writes the policy value
+		// into the same Config field). See netbirdio/netbird#5744.
+		if !s.disableAutoConnectFromDaemon() {
+			s.showQuickActionsUI()
+		}
 	case args.showUpdate:
 		s.showUpdateProgress(ctx, args.showUpdateVersion)
 	}
@@ -1338,6 +1345,40 @@ func (s *serviceClient) getFeatures() (*proto.GetFeaturesResponse, error) {
 	return features, nil
 }
 
+// disableAutoConnectFromDaemon returns true when the daemon reports
+// the active profile has DisableAutoConnect=true. Used by the
+// --quick-actions startup path to suppress the on-boot popup when the
+// user (or an MDM admin) opted out of auto-connecting; both cases
+// converge on the same Config field because applyMDMPolicy writes the
+// policy value into it. Returns false on any RPC / lookup failure so a
+// daemon hiccup does not silently swallow the popup.
+func (s *serviceClient) disableAutoConnectFromDaemon() bool {
+	activeProf, err := s.profileManager.GetActiveProfile()
+	if err != nil {
+		log.Warnf("disableAutoConnectFromDaemon: get active profile: %v", err)
+		return false
+	}
+	currUser, err := user.Current()
+	if err != nil {
+		log.Warnf("disableAutoConnectFromDaemon: get current user: %v", err)
+		return false
+	}
+	conn, err := s.getSrvClient(failFastTimeout)
+	if err != nil {
+		log.Warnf("disableAutoConnectFromDaemon: get daemon client: %v", err)
+		return false
+	}
+	srvCfg, err := conn.GetConfig(s.ctx, &proto.GetConfigRequest{
+		ProfileName: activeProf.ID.String(),
+		Username:    currUser.Username,
+	})
+	if err != nil {
+		log.Warnf("disableAutoConnectFromDaemon: GetConfig RPC: %v", err)
+		return false
+	}
+	return srvCfg.GetDisableAutoConnect()
+}
+
 // getSrvConfig from the service to show it in the settings window.
 func (s *serviceClient) getSrvConfig() {
 	s.managementURL = profilemanager.DefaultManagementURL

flow/proto/flow.pb.go

@@ -134,9 +134,11 @@ type FlowEvent struct {
 	// When the event occurred
 	Timestamp *timestamppb.Timestamp `protobuf:"bytes,2,opt,name=timestamp,proto3" json:"timestamp,omitempty"`
 	// Public key of the sending peer
-	PublicKey   []byte      `protobuf:"bytes,3,opt,name=public_key,json=publicKey,proto3" json:"public_key,omitempty"`
-	FlowFields  *FlowFields `protobuf:"bytes,4,opt,name=flow_fields,json=flowFields,proto3" json:"flow_fields,omitempty"`
-	IsInitiator bool        `protobuf:"varint,5,opt,name=isInitiator,proto3" json:"isInitiator,omitempty"`
+	PublicKey   []byte                 `protobuf:"bytes,3,opt,name=public_key,json=publicKey,proto3" json:"public_key,omitempty"`
+	FlowFields  *FlowFields            `protobuf:"bytes,4,opt,name=flow_fields,json=flowFields,proto3" json:"flow_fields,omitempty"`
+	IsInitiator bool                   `protobuf:"varint,5,opt,name=isInitiator,proto3" json:"isInitiator,omitempty"`
+	WindowStart *timestamppb.Timestamp `protobuf:"bytes,6,opt,name=window_start,json=windowStart,proto3" json:"window_start,omitempty"`
+	WindowEnd   *timestamppb.Timestamp `protobuf:"bytes,7,opt,name=window_end,json=windowEnd,proto3" json:"window_end,omitempty"`
 }
 
 func (x *FlowEvent) Reset() {
@@ -206,6 +208,20 @@ func (x *FlowEvent) GetIsInitiator() bool {
 	return false
 }
 
+func (x *FlowEvent) GetWindowStart() *timestamppb.Timestamp {
+	if x != nil {
+		return x.WindowStart
+	}
+	return nil
+}
+
+func (x *FlowEvent) GetWindowEnd() *timestamppb.Timestamp {
+	if x != nil {
+		return x.WindowEnd
+	}
+	return nil
+}
+
 type FlowEventAck struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -284,7 +300,6 @@ type FlowFields struct {
 	// Layer 4 -specific information
 	//
 	// Types that are assignable to ConnectionInfo:
-	//
 	//	*FlowFields_PortInfo
 	//	*FlowFields_IcmpInfo
 	ConnectionInfo isFlowFields_ConnectionInfo `protobuf_oneof:"connection_info"`
@@ -297,6 +312,9 @@ type FlowFields struct {
 	// 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"`
+	NumOfStarts      uint64 `protobuf:"varint,16,opt,name=num_of_starts,json=numOfStarts,proto3" json:"num_of_starts,omitempty"`
+	NumOfEnds        uint64 `protobuf:"varint,17,opt,name=num_of_ends,json=numOfEnds,proto3" json:"num_of_ends,omitempty"`
+	NumOfDrops       uint64 `protobuf:"varint,18,opt,name=num_of_drops,json=numOfDrops,proto3" json:"num_of_drops,omitempty"`
 }
 
 func (x *FlowFields) Reset() {
@@ -443,6 +461,27 @@ func (x *FlowFields) GetDestResourceId() []byte {
 	return nil
 }
 
+func (x *FlowFields) GetNumOfStarts() uint64 {
+	if x != nil {
+		return x.NumOfStarts
+	}
+	return 0
+}
+
+func (x *FlowFields) GetNumOfEnds() uint64 {
+	if x != nil {
+		return x.NumOfEnds
+	}
+	return 0
+}
+
+func (x *FlowFields) GetNumOfDrops() uint64 {
+	if x != nil {
+		return x.NumOfDrops
+	}
+	return 0
+}
+
 type isFlowFields_ConnectionInfo interface {
 	isFlowFields_ConnectionInfo()
 }
@@ -579,7 +618,7 @@ var file_flow_proto_rawDesc = []byte{
 	0x0a, 0x0a, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x04, 0x66, 0x6c,
 	0x6f, 0x77, 0x1a, 0x1f, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f,
 	0x62, 0x75, 0x66, 0x2f, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x70, 0x72,
-	0x6f, 0x74, 0x6f, 0x22, 0xd4, 0x01, 0x0a, 0x09, 0x46, 0x6c, 0x6f, 0x77, 0x45, 0x76, 0x65, 0x6e,
+	0x6f, 0x74, 0x6f, 0x22, 0xce, 0x02, 0x0a, 0x09, 0x46, 0x6c, 0x6f, 0x77, 0x45, 0x76, 0x65, 0x6e,
 	0x74, 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, 0x12, 0x38, 0x0a, 0x09,
 	0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32,
@@ -592,45 +631,59 @@ var file_flow_proto_rawDesc = []byte{
 	0x77, 0x2e, 0x46, 0x6c, 0x6f, 0x77, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x73, 0x52, 0x0a, 0x66, 0x6c,
 	0x6f, 0x77, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x73, 0x12, 0x20, 0x0a, 0x0b, 0x69, 0x73, 0x49, 0x6e,
 	0x69, 0x74, 0x69, 0x61, 0x74, 0x6f, 0x72, 0x18, 0x05, 0x20, 0x01, 0x28, 0x08, 0x52, 0x0b, 0x69,
-	0x73, 0x49, 0x6e, 0x69, 0x74, 0x69, 0x61, 0x74, 0x6f, 0x72, 0x22, 0x4b, 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, 0x12, 0x20, 0x0a, 0x0b, 0x69, 0x73, 0x49, 0x6e, 0x69, 0x74, 0x69,
-	0x61, 0x74, 0x6f, 0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x08, 0x52, 0x0b, 0x69, 0x73, 0x49, 0x6e,
-	0x69, 0x74, 0x69, 0x61, 0x74, 0x6f, 0x72, 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, 0x2e, 0x54, 0x79, 0x70, 0x65, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12,
-	0x17, 0x0a, 0x07, 0x72, 0x75, 0x6c, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c,
-	0x52, 0x06, 0x72, 0x75, 0x6c, 0x65, 0x49, 0x64, 0x12, 0x2d, 0x0a, 0x09, 0x64, 0x69, 0x72, 0x65,
-	0x63, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x0f, 0x2e, 0x66, 0x6c,
-	0x6f, 0x77, 0x2e, 0x44, 0x69, 0x72, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x09, 0x64, 0x69,
-	0x72, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x1a, 0x0a, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f,
-	0x63, 0x6f, 0x6c, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f,
-	0x63, 0x6f, 0x6c, 0x12, 0x1b, 0x0a, 0x09, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x69, 0x70,
-	0x18, 0x06, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x08, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x49, 0x70,
-	0x12, 0x17, 0x0a, 0x07, 0x64, 0x65, 0x73, 0x74, 0x5f, 0x69, 0x70, 0x18, 0x07, 0x20, 0x01, 0x28,
-	0x0c, 0x52, 0x06, 0x64, 0x65, 0x73, 0x74, 0x49, 0x70, 0x12, 0x2d, 0x0a, 0x09, 0x70, 0x6f, 0x72,
-	0x74, 0x5f, 0x69, 0x6e, 0x66, 0x6f, 0x18, 0x08, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0e, 0x2e, 0x66,
-	0x6c, 0x6f, 0x77, 0x2e, 0x50, 0x6f, 0x72, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x48, 0x00, 0x52, 0x08,
-	0x70, 0x6f, 0x72, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x2d, 0x0a, 0x09, 0x69, 0x63, 0x6d, 0x70,
-	0x5f, 0x69, 0x6e, 0x66, 0x6f, 0x18, 0x09, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0e, 0x2e, 0x66, 0x6c,
-	0x6f, 0x77, 0x2e, 0x49, 0x43, 0x4d, 0x50, 0x49, 0x6e, 0x66, 0x6f, 0x48, 0x00, 0x52, 0x08, 0x69,
-	0x63, 0x6d, 0x70, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x1d, 0x0a, 0x0a, 0x72, 0x78, 0x5f, 0x70, 0x61,
-	0x63, 0x6b, 0x65, 0x74, 0x73, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x04, 0x52, 0x09, 0x72, 0x78, 0x50,
-	0x61, 0x63, 0x6b, 0x65, 0x74, 0x73, 0x12, 0x1d, 0x0a, 0x0a, 0x74, 0x78, 0x5f, 0x70, 0x61, 0x63,
-	0x6b, 0x65, 0x74, 0x73, 0x18, 0x0b, 0x20, 0x01, 0x28, 0x04, 0x52, 0x09, 0x74, 0x78, 0x50, 0x61,
-	0x63, 0x6b, 0x65, 0x74, 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, 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,
+	0x73, 0x49, 0x6e, 0x69, 0x74, 0x69, 0x61, 0x74, 0x6f, 0x72, 0x12, 0x3d, 0x0a, 0x0c, 0x77, 0x69,
+	0x6e, 0x64, 0x6f, 0x77, 0x5f, 0x73, 0x74, 0x61, 0x72, 0x74, 0x18, 0x06, 0x20, 0x01, 0x28, 0x0b,
+	0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62,
+	0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0b, 0x77, 0x69,
+	0x6e, 0x64, 0x6f, 0x77, 0x53, 0x74, 0x61, 0x72, 0x74, 0x12, 0x39, 0x0a, 0x0a, 0x77, 0x69, 0x6e,
+	0x64, 0x6f, 0x77, 0x5f, 0x65, 0x6e, 0x64, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e,
+	0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e,
+	0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x09, 0x77, 0x69, 0x6e, 0x64, 0x6f,
+	0x77, 0x45, 0x6e, 0x64, 0x22, 0x4b, 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, 0x12,
+	0x20, 0x0a, 0x0b, 0x69, 0x73, 0x49, 0x6e, 0x69, 0x74, 0x69, 0x61, 0x74, 0x6f, 0x72, 0x18, 0x02,
+	0x20, 0x01, 0x28, 0x08, 0x52, 0x0b, 0x69, 0x73, 0x49, 0x6e, 0x69, 0x74, 0x69, 0x61, 0x74, 0x6f,
+	0x72, 0x22, 0x82, 0x05, 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, 0x2e, 0x54,
+	0x79, 0x70, 0x65, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12, 0x17, 0x0a, 0x07, 0x72, 0x75, 0x6c,
+	0x65, 0x5f, 0x69, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x72, 0x75, 0x6c, 0x65,
+	0x49, 0x64, 0x12, 0x2d, 0x0a, 0x09, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x18,
+	0x04, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x0f, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x44, 0x69, 0x72,
+	0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x09, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x69, 0x6f,
+	0x6e, 0x12, 0x1a, 0x0a, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x18, 0x05, 0x20,
+	0x01, 0x28, 0x0d, 0x52, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x12, 0x1b, 0x0a,
+	0x09, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x69, 0x70, 0x18, 0x06, 0x20, 0x01, 0x28, 0x0c,
+	0x52, 0x08, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x49, 0x70, 0x12, 0x17, 0x0a, 0x07, 0x64, 0x65,
+	0x73, 0x74, 0x5f, 0x69, 0x70, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x64, 0x65, 0x73,
+	0x74, 0x49, 0x70, 0x12, 0x2d, 0x0a, 0x09, 0x70, 0x6f, 0x72, 0x74, 0x5f, 0x69, 0x6e, 0x66, 0x6f,
+	0x18, 0x08, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0e, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x50, 0x6f,
+	0x72, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x48, 0x00, 0x52, 0x08, 0x70, 0x6f, 0x72, 0x74, 0x49, 0x6e,
+	0x66, 0x6f, 0x12, 0x2d, 0x0a, 0x09, 0x69, 0x63, 0x6d, 0x70, 0x5f, 0x69, 0x6e, 0x66, 0x6f, 0x18,
+	0x09, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0e, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x49, 0x43, 0x4d,
+	0x50, 0x49, 0x6e, 0x66, 0x6f, 0x48, 0x00, 0x52, 0x08, 0x69, 0x63, 0x6d, 0x70, 0x49, 0x6e, 0x66,
+	0x6f, 0x12, 0x1d, 0x0a, 0x0a, 0x72, 0x78, 0x5f, 0x70, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x73, 0x18,
+	0x0a, 0x20, 0x01, 0x28, 0x04, 0x52, 0x09, 0x72, 0x78, 0x50, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x73,
+	0x12, 0x1d, 0x0a, 0x0a, 0x74, 0x78, 0x5f, 0x70, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x73, 0x18, 0x0b,
+	0x20, 0x01, 0x28, 0x04, 0x52, 0x09, 0x74, 0x78, 0x50, 0x61, 0x63, 0x6b, 0x65, 0x74, 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, 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, 0x12, 0x22, 0x0a,
+	0x0d, 0x6e, 0x75, 0x6d, 0x5f, 0x6f, 0x66, 0x5f, 0x73, 0x74, 0x61, 0x72, 0x74, 0x73, 0x18, 0x10,
+	0x20, 0x01, 0x28, 0x04, 0x52, 0x0b, 0x6e, 0x75, 0x6d, 0x4f, 0x66, 0x53, 0x74, 0x61, 0x72, 0x74,
+	0x73, 0x12, 0x1e, 0x0a, 0x0b, 0x6e, 0x75, 0x6d, 0x5f, 0x6f, 0x66, 0x5f, 0x65, 0x6e, 0x64, 0x73,
+	0x18, 0x11, 0x20, 0x01, 0x28, 0x04, 0x52, 0x09, 0x6e, 0x75, 0x6d, 0x4f, 0x66, 0x45, 0x6e, 0x64,
+	0x73, 0x12, 0x20, 0x0a, 0x0c, 0x6e, 0x75, 0x6d, 0x5f, 0x6f, 0x66, 0x5f, 0x64, 0x72, 0x6f, 0x70,
+	0x73, 0x18, 0x12, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0a, 0x6e, 0x75, 0x6d, 0x4f, 0x66, 0x44, 0x72,
+	0x6f, 0x70, 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,
@@ -683,17 +736,19 @@ var file_flow_proto_goTypes = []interface{}{
 var file_flow_proto_depIdxs = []int32{
 	7, // 0: flow.FlowEvent.timestamp:type_name -> google.protobuf.Timestamp
 	4, // 1: flow.FlowEvent.flow_fields:type_name -> flow.FlowFields
-	0, // 2: flow.FlowFields.type:type_name -> flow.Type
-	1, // 3: flow.FlowFields.direction:type_name -> flow.Direction
-	5, // 4: flow.FlowFields.port_info:type_name -> flow.PortInfo
-	6, // 5: flow.FlowFields.icmp_info:type_name -> flow.ICMPInfo
-	2, // 6: flow.FlowService.Events:input_type -> flow.FlowEvent
-	3, // 7: flow.FlowService.Events:output_type -> flow.FlowEventAck
-	7, // [7:8] is the sub-list for method output_type
-	6, // [6:7] is the sub-list for method input_type
-	6, // [6:6] is the sub-list for extension type_name
-	6, // [6:6] is the sub-list for extension extendee
-	0, // [0:6] is the sub-list for field type_name
+	7, // 2: flow.FlowEvent.window_start:type_name -> google.protobuf.Timestamp
+	7, // 3: flow.FlowEvent.window_end:type_name -> google.protobuf.Timestamp
+	0, // 4: flow.FlowFields.type:type_name -> flow.Type
+	1, // 5: flow.FlowFields.direction:type_name -> flow.Direction
+	5, // 6: flow.FlowFields.port_info:type_name -> flow.PortInfo
+	6, // 7: flow.FlowFields.icmp_info:type_name -> flow.ICMPInfo
+	2, // 8: flow.FlowService.Events:input_type -> flow.FlowEvent
+	3, // 9: flow.FlowService.Events:output_type -> flow.FlowEventAck
+	9, // [9:10] is the sub-list for method output_type
+	8, // [8:9] is the sub-list for method input_type
+	8, // [8:8] is the sub-list for extension type_name
+	8, // [8:8] is the sub-list for extension extendee
+	0, // [0:8] is the sub-list for field type_name
 }
 
 func init() { file_flow_proto_init() }

flow/proto/flow.proto

@@ -24,6 +24,9 @@ message FlowEvent {
   FlowFields flow_fields = 4;
 
   bool isInitiator = 5;
+
+  google.protobuf.Timestamp window_start = 6;
+  google.protobuf.Timestamp window_end = 7;
 }
 
 message FlowEventAck {
@@ -75,6 +78,9 @@ message FlowFields {
   bytes source_resource_id = 14;
   bytes dest_resource_id = 15;
 
+  uint64 num_of_starts = 16;
+  uint64 num_of_ends = 17;
+  uint64 num_of_drops = 18;
 }
 
 // Flow event types

flow/proto/flow_grpc.pb.go

@@ -1,4 +1,8 @@
 // Code generated by protoc-gen-go-grpc. DO NOT EDIT.
+// versions:
+// - protoc-gen-go-grpc v1.6.1
+// - protoc             v3.21.9
+// source: flow.proto
 
 package proto
 
@@ -11,15 +15,19 @@ import (
 
 // This is a compile-time assertion to ensure that this generated file
 // is compatible with the grpc package it is being compiled against.
-// Requires gRPC-Go v1.32.0 or later.
-const _ = grpc.SupportPackageIsVersion7
+// Requires gRPC-Go v1.64.0 or later.
+const _ = grpc.SupportPackageIsVersion9
+
+const (
+	FlowService_Events_FullMethodName = "/flow.FlowService/Events"
+)
 
 // FlowServiceClient is the client API for FlowService service.
 //
 // For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
 type FlowServiceClient interface {
 	// Client to receiver streams of events and acknowledgements
-	Events(ctx context.Context, opts ...grpc.CallOption) (FlowService_EventsClient, error)
+	Events(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[FlowEvent, FlowEventAck], error)
 }
 
 type flowServiceClient struct {
@@ -30,54 +38,40 @@ func NewFlowServiceClient(cc grpc.ClientConnInterface) FlowServiceClient {
 	return &flowServiceClient{cc}
 }
 
-func (c *flowServiceClient) Events(ctx context.Context, opts ...grpc.CallOption) (FlowService_EventsClient, error) {
-	stream, err := c.cc.NewStream(ctx, &FlowService_ServiceDesc.Streams[0], "/flow.FlowService/Events", opts...)
+func (c *flowServiceClient) Events(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[FlowEvent, FlowEventAck], error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	stream, err := c.cc.NewStream(ctx, &FlowService_ServiceDesc.Streams[0], FlowService_Events_FullMethodName, cOpts...)
 	if err != nil {
 		return nil, err
 	}
-	x := &flowServiceEventsClient{stream}
+	x := &grpc.GenericClientStream[FlowEvent, FlowEventAck]{ClientStream: stream}
 	return x, nil
 }
 
-type FlowService_EventsClient interface {
-	Send(*FlowEvent) error
-	Recv() (*FlowEventAck, error)
-	grpc.ClientStream
-}
-
-type flowServiceEventsClient struct {
-	grpc.ClientStream
-}
-
-func (x *flowServiceEventsClient) Send(m *FlowEvent) error {
-	return x.ClientStream.SendMsg(m)
-}
-
-func (x *flowServiceEventsClient) Recv() (*FlowEventAck, error) {
-	m := new(FlowEventAck)
-	if err := x.ClientStream.RecvMsg(m); err != nil {
-		return nil, err
-	}
-	return m, nil
-}
+// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
+type FlowService_EventsClient = grpc.BidiStreamingClient[FlowEvent, FlowEventAck]
 
 // FlowServiceServer is the server API for FlowService service.
 // All implementations must embed UnimplementedFlowServiceServer
-// for forward compatibility
+// for forward compatibility.
 type FlowServiceServer interface {
 	// Client to receiver streams of events and acknowledgements
-	Events(FlowService_EventsServer) error
+	Events(grpc.BidiStreamingServer[FlowEvent, FlowEventAck]) error
 	mustEmbedUnimplementedFlowServiceServer()
 }
 
-// UnimplementedFlowServiceServer must be embedded to have forward compatible implementations.
-type UnimplementedFlowServiceServer struct {
-}
+// UnimplementedFlowServiceServer must be embedded to have
+// forward compatible implementations.
+//
+// NOTE: this should be embedded by value instead of pointer to avoid a nil
+// pointer dereference when methods are called.
+type UnimplementedFlowServiceServer struct{}
 
-func (UnimplementedFlowServiceServer) Events(FlowService_EventsServer) error {
-	return status.Errorf(codes.Unimplemented, "method Events not implemented")
+func (UnimplementedFlowServiceServer) Events(grpc.BidiStreamingServer[FlowEvent, FlowEventAck]) error {
+	return status.Error(codes.Unimplemented, "method Events not implemented")
 }
 func (UnimplementedFlowServiceServer) mustEmbedUnimplementedFlowServiceServer() {}
+func (UnimplementedFlowServiceServer) testEmbeddedByValue()                     {}
 
 // UnsafeFlowServiceServer may be embedded to opt out of forward compatibility for this service.
 // Use of this interface is not recommended, as added methods to FlowServiceServer will
@@ -87,34 +81,22 @@ type UnsafeFlowServiceServer interface {
 }
 
 func RegisterFlowServiceServer(s grpc.ServiceRegistrar, srv FlowServiceServer) {
+	// If the following call panics, it indicates UnimplementedFlowServiceServer was
+	// embedded by pointer and is nil.  This will cause panics if an
+	// unimplemented method is ever invoked, so we test this at initialization
+	// time to prevent it from happening at runtime later due to I/O.
+	if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
+		t.testEmbeddedByValue()
+	}
 	s.RegisterService(&FlowService_ServiceDesc, srv)
 }
 
 func _FlowService_Events_Handler(srv interface{}, stream grpc.ServerStream) error {
-	return srv.(FlowServiceServer).Events(&flowServiceEventsServer{stream})
+	return srv.(FlowServiceServer).Events(&grpc.GenericServerStream[FlowEvent, FlowEventAck]{ServerStream: stream})
 }
 
-type FlowService_EventsServer interface {
-	Send(*FlowEventAck) error
-	Recv() (*FlowEvent, error)
-	grpc.ServerStream
-}
-
-type flowServiceEventsServer struct {
-	grpc.ServerStream
-}
-
-func (x *flowServiceEventsServer) Send(m *FlowEventAck) error {
-	return x.ServerStream.SendMsg(m)
-}
-
-func (x *flowServiceEventsServer) Recv() (*FlowEvent, error) {
-	m := new(FlowEvent)
-	if err := x.ServerStream.RecvMsg(m); err != nil {
-		return nil, err
-	}
-	return m, nil
-}
+// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
+type FlowService_EventsServer = grpc.BidiStreamingServer[FlowEvent, FlowEventAck]
 
 // FlowService_ServiceDesc is the grpc.ServiceDesc for FlowService service.
 // It's only intended for direct use with grpc.RegisterService,

management/internals/controllers/network_map/controller/controller.go

@@ -497,7 +497,7 @@ func (c *Controller) BufferUpdateAffectedPeers(ctx context.Context, accountID st
 		c.accountManagerMetrics.CountUpdateAccountPeersTriggered(string(reason.Resource), string(reason.Operation))
 	}
 
-	log.WithContext(ctx).Tracef("buffer updating %d affected peers for account %s from %s", len(peerIDs), accountID, util.GetCallerName())
+	log.WithContext(ctx).Tracef("buffer updating %d affected peers for account %s from %s with reason %s/%s", len(peerIDs), accountID, util.GetCallerName(), reason.Operation, reason.Resource)
 
 	bufUpd, _ := c.affectedPeerUpdateLocks.LoadOrStore(accountID, &bufferAffectedUpdate{
 		peerIDs: make(map[string]struct{}),
@@ -610,12 +610,10 @@ func (c *Controller) GetValidatedPeerWithMap(ctx context.Context, isRequiresAppr
 		return nil, nil, 0, err
 	}
 
-	startPosture := time.Now()
 	postureChecks, err := c.getPeerPostureChecks(account, peerID)
 	if err != nil {
 		return nil, nil, 0, err
 	}
-	log.WithContext(ctx).Debugf("getPeerPostureChecks took %s", time.Since(startPosture))
 
 	accountZones, err := c.repo.GetAccountZones(ctx, account.Id)
 	if err != nil {

management/internals/shared/grpc/loginfilter.go

@@ -11,7 +11,7 @@ import (
 
 const (
 	reconnThreshold   = 5 * time.Minute
-	baseBlockDuration = 10 * time.Minute // Duration for which a peer is banned after exceeding the reconnection limit
+	baseBlockDuration = 30 * time.Minute // Duration for which a peer is banned after exceeding the reconnection limit
 	reconnLimitForBan = 30               // Number of reconnections within the reconnTreshold that triggers a ban
 	metaChangeLimit   = 3                // Number of reconnections with different metadata that triggers a ban of one peer
 )
@@ -139,22 +139,13 @@ func (l *loginFilter) addLogin(wgPubKey string, metaHash uint64) {
 	state.lastSeen = now
 }
 
-func metaHash(meta nbpeer.PeerSystemMeta, pubip string) uint64 {
+func metaHash(meta nbpeer.PeerSystemMeta) uint64 {
 	h := fnv.New64a()
 
-	h.Write([]byte(meta.WtVersion))
 	h.Write([]byte(meta.OSVersion))
 	h.Write([]byte(meta.KernelVersion))
 	h.Write([]byte(meta.Hostname))
 	h.Write([]byte(meta.SystemSerialNumber))
-	h.Write([]byte(pubip))
 
-	macs := uint64(0)
-	for _, na := range meta.NetworkAddresses {
-		for _, r := range na.Mac {
-			macs += uint64(r)
-		}
-	}
-
-	return h.Sum64() + macs
+	return h.Sum64()
 }

management/internals/shared/grpc/loginfilter_test.go

@@ -164,9 +164,7 @@ func BenchmarkHashingMethods(b *testing.B) {
 		KernelVersion:      "5.15.0-76-generic",
 		Hostname:           "prod-server-database-01",
 		SystemSerialNumber: "PC-1234567890",
-		NetworkAddresses:   []nbpeer.NetworkAddress{{Mac: "00:1B:44:11:3A:B7"}, {Mac: "00:1B:44:11:3A:B8"}},
 	}
-	pubip := "8.8.8.8"
 
 	var resultString string
 	var resultUint uint64
@@ -175,7 +173,7 @@ func BenchmarkHashingMethods(b *testing.B) {
 		b.ReportAllocs()
 		b.ResetTimer()
 		for i := 0; i < b.N; i++ {
-			resultString = builderString(meta, pubip)
+			resultString = builderString(meta)
 		}
 	})
 
@@ -183,7 +181,7 @@ func BenchmarkHashingMethods(b *testing.B) {
 		b.ReportAllocs()
 		b.ResetTimer()
 		for i := 0; i < b.N; i++ {
-			resultString = fnvHashToString(meta, pubip)
+			resultString = fnvHashToString(meta)
 		}
 	})
 
@@ -191,7 +189,7 @@ func BenchmarkHashingMethods(b *testing.B) {
 		b.ReportAllocs()
 		b.ResetTimer()
 		for i := 0; i < b.N; i++ {
-			resultUint = metaHash(meta, pubip)
+			resultUint = metaHash(meta)
 		}
 	})
 
@@ -199,29 +197,20 @@ func BenchmarkHashingMethods(b *testing.B) {
 	_ = resultUint
 }
 
-func fnvHashToString(meta nbpeer.PeerSystemMeta, pubip string) string {
+func fnvHashToString(meta nbpeer.PeerSystemMeta) string {
 	h := fnv.New64a()
 
-	if len(meta.NetworkAddresses) != 0 {
-		for _, na := range meta.NetworkAddresses {
-			h.Write([]byte(na.Mac))
-		}
-	}
-
 	h.Write([]byte(meta.WtVersion))
 	h.Write([]byte(meta.OSVersion))
 	h.Write([]byte(meta.KernelVersion))
 	h.Write([]byte(meta.Hostname))
 	h.Write([]byte(meta.SystemSerialNumber))
-	h.Write([]byte(pubip))
 
 	return strconv.FormatUint(h.Sum64(), 16)
 }
 
-func builderString(meta nbpeer.PeerSystemMeta, pubip string) string {
-	mac := getMacAddress(meta.NetworkAddresses)
-	estimatedSize := len(meta.WtVersion) + len(meta.OSVersion) + len(meta.KernelVersion) + len(meta.Hostname) + len(meta.SystemSerialNumber) +
-		len(pubip) + len(mac) + 6
+func builderString(meta nbpeer.PeerSystemMeta) string {
+	estimatedSize := len(meta.WtVersion) + len(meta.OSVersion) + len(meta.KernelVersion) + len(meta.Hostname) + len(meta.SystemSerialNumber) + 4
 
 	var b strings.Builder
 	b.Grow(estimatedSize)
@@ -235,23 +224,10 @@ func builderString(meta nbpeer.PeerSystemMeta, pubip string) string {
 	b.WriteString(meta.Hostname)
 	b.WriteByte('|')
 	b.WriteString(meta.SystemSerialNumber)
-	b.WriteByte('|')
-	b.WriteString(pubip)
 
 	return b.String()
 }
 
-func getMacAddress(nas []nbpeer.NetworkAddress) string {
-	if len(nas) == 0 {
-		return ""
-	}
-	macs := make([]string, 0, len(nas))
-	for _, na := range nas {
-		macs = append(macs, na.Mac)
-	}
-	return strings.Join(macs, "/")
-}
-
 func BenchmarkLoginFilter_ParallelLoad(b *testing.B) {
 	filter := newLoginFilterWithCfg(testAdvancedCfg())
 	numKeys := 100000

management/internals/shared/grpc/server.go

@@ -254,7 +254,7 @@ func (s *Server) Sync(req *proto.EncryptedMessage, srv proto.ManagementService_S
 		return mapError(ctx, err)
 	}
 
-	metahashed := metaHash(peerMeta, sRealIP)
+	metahashed := metaHash(peerMeta)
 	if userID == "" && !s.loginFilter.allowLogin(peerKey.String(), metahashed) {
 		if s.appMetrics != nil {
 			s.appMetrics.GRPCMetrics().CountSyncRequestBlocked()
@@ -306,7 +306,7 @@ func (s *Server) Sync(req *proto.EncryptedMessage, srv proto.ManagementService_S
 		log.WithContext(ctx).Tracef("peer system meta has to be provided on sync. Peer %s, remote addr %s", peerKey.String(), realIP)
 	}
 
-	metahash := metaHash(peerMeta, realIP.String())
+	metahash := metaHash(peerMeta)
 	s.loginFilter.addLogin(peerKey.String(), metahash)
 
 	peer, netMap, postureChecks, dnsFwdPort, err := s.accountManager.SyncAndMarkPeer(ctx, accountID, peerKey.String(), peerMeta, realIP, syncStart)
@@ -732,7 +732,7 @@ func (s *Server) Login(ctx context.Context, req *proto.EncryptedMessage) (*proto
 	}
 
 	peerMeta := extractPeerMeta(ctx, loginReq.GetMeta())
-	metahashed := metaHash(peerMeta, sRealIP)
+	metahashed := metaHash(peerMeta)
 	if !s.loginFilter.allowLogin(peerKey.String(), metahashed) {
 		if s.logBlockedPeers {
 			log.WithContext(ctx).Tracef("peer %s with meta hash %d is blocked from login", peerKey.String(), metahashed)
@@ -788,7 +788,11 @@ func (s *Server) Login(ctx context.Context, req *proto.EncryptedMessage) (*proto
 		ExtraDNSLabels:  loginReq.GetDnsLabels(),
 	})
 	if err != nil {
-		log.WithContext(ctx).Warnf("failed logging in peer %s: %s", peerKey, err)
+		if errors.Is(err, internalStatus.ErrNoAuthMethodProvided) {
+			log.WithContext(ctx).Tracef("failed logging in peer %s: %s", peerKey, err)
+		} else {
+			log.WithContext(ctx).Warnf("failed logging in peer %s: %s", peerKey, err)
+		}
 		return nil, mapError(ctx, err)
 	}
 

management/server/affected_peers_coverage_test.go

@@ -41,7 +41,7 @@ func TestAffectedPeers_DependencyCoverageMatrix(t *testing.T) {
 				_, err := s.manager.SavePolicy(ctx, s.accountID, userID, peerToResourcePolicyByGroup(s.sourceGroupID, s.resourceGroupID), true)
 				require.NoError(t, err)
 				return affectedpeers.Change{ChangedPeerIDs: []string{s.routerPeerID}},
-					[]string{s.sourcePeerID}, []string{s.unrelatedPeerID}
+					[]string{s.sourcePeerID, s.routerPeerID}, []string{s.unrelatedPeerID}
 			},
 		},
 		{
@@ -106,11 +106,9 @@ func TestAffectedPeers_DependencyCoverageMatrix(t *testing.T) {
 			change, mustContain, mustExclude := r.build(t, s, ctx)
 			affected := resolveAffected(t, s.manager.Store, s.accountID, change)
 
-			for _, id := range mustContain {
-				assert.Contains(t, affected, id, "expected peer to be affected")
-			}
-			for _, id := range mustExclude {
-				assert.NotContains(t, affected, id, "peer must not be affected")
+			assert.ElementsMatch(t, affected, mustContain, "expected peer to be affected")
+			for _, peerID := range mustExclude {
+				assert.NotContains(t, affected, peerID, "peer must not be affected")
 			}
 		})
 	}

management/server/affected_peers_router_paths_test.go

@@ -251,7 +251,9 @@ func TestAffectedPeers_E2E_UpdateResource_DestinationResourcePolicy_RefreshesSou
 	}
 }
 
-func TestAffectedPeers_E2E_UpdateResource_DisabledSiblingRouter_StillBridged(t *testing.T) {
+// A disabled sibling router routes to nobody, so updating a resource on its network
+// must NOT refresh its peer (the enabled router carries the bridge instead).
+func TestAffectedPeers_E2E_UpdateResource_DisabledSiblingRouterNotBridged(t *testing.T) {
 	s := setupRouterScenario(t, true)
 	ctx := context.Background()
 
@@ -274,13 +276,18 @@ func TestAffectedPeers_E2E_UpdateResource_DisabledSiblingRouter_StillBridged(t *
 	require.NoError(t, err)
 
 	disabledCh := s.updateManager.CreateChannel(ctx, disabledRouterPeer.ID)
-	t.Cleanup(func() { s.updateManager.CloseChannel(ctx, disabledRouterPeer.ID) })
+	enabledCh := s.updateManager.CreateChannel(ctx, s.routerPeerID)
+	t.Cleanup(func() {
+		s.updateManager.CloseChannel(ctx, disabledRouterPeer.ID)
+		s.updateManager.CloseChannel(ctx, s.routerPeerID)
+	})
 
-	settleAffectedUpdates(disabledCh)
+	settleAffectedUpdates(disabledCh, enabledCh)
 
 	done := make(chan struct{})
 	go func() {
-		peerShouldReceiveUpdate(t, disabledCh)
+		peerShouldReceiveUpdate(t, enabledCh)
+		peerShouldNotReceiveUpdate(t, disabledCh)
 		close(done)
 	}()
 
@@ -298,7 +305,7 @@ func TestAffectedPeers_E2E_UpdateResource_DisabledSiblingRouter_StillBridged(t *
 	select {
 	case <-done:
 	case <-time.After(peerUpdateTimeout):
-		t.Error("timeout: resource update did not refresh the disabled sibling router's peer")
+		t.Error("timeout")
 	}
 }
 

management/server/affected_peers_router_test.go

@@ -682,6 +682,9 @@ func TestAffectedPeers_AllRoutingPeers_Network(t *testing.T) {
 	assert.Contains(t, affected, secondRouterPeer.ID, "second routing peer on the same network must also be affected")
 }
 
+// A disabled router in the snapshot routes to nobody, so it is skipped when the
+// walk scans existing account data: a policy edit still folds the literal source
+// group, but not the disabled router's peer.
 func TestAffectedPeers_DisabledRouter(t *testing.T) {
 	s := setupRouterScenario(t, true)
 	ctx := context.Background()
@@ -694,11 +697,13 @@ func TestAffectedPeers_DisabledRouter(t *testing.T) {
 
 	affected := s.resolvePolicyAffected(ctx, peerToResourcePolicyByGroup(s.sourceGroupID, s.resourceGroupID))
 
-	assert.Contains(t, affected, s.sourcePeerID, "source peer must be affected")
-	assert.Contains(t, affected, s.routerPeerID,
-		"disabled router's peer must still be affected: Enabled must not gate affected-peers")
+	assert.Contains(t, affected, s.sourcePeerID, "source peer (literal policy source group) must be affected")
+	assert.NotContains(t, affected, s.routerPeerID,
+		"a disabled router routes to nobody, so its peer must not be folded from snapshot data")
 }
 
+// A disabled resource in the snapshot is skipped: the policy edit still folds the
+// literal source group, but the resource no longer bridges to its network's router.
 func TestAffectedPeers_DisabledResource(t *testing.T) {
 	s := setupRouterScenario(t, true)
 	ctx := context.Background()
@@ -710,9 +715,9 @@ func TestAffectedPeers_DisabledResource(t *testing.T) {
 
 	affected := s.resolvePolicyAffected(ctx, peerToResourcePolicyByGroup(s.sourceGroupID, s.resourceGroupID))
 
-	assert.Contains(t, affected, s.sourcePeerID, "source peer must be affected")
-	assert.Contains(t, affected, s.routerPeerID,
-		"disabled resource must still resolve the routing peer: Enabled must not gate affected-peers")
+	assert.Contains(t, affected, s.sourcePeerID, "source peer (literal policy source group) must be affected")
+	assert.NotContains(t, affected, s.routerPeerID,
+		"a disabled resource routes to nobody, so its network's router must not be folded from snapshot data")
 }
 
 func TestAffectedPeers_DisabledRule(t *testing.T) {

management/server/affected_peers_test.go

@@ -96,33 +96,54 @@ func affectedGroupID(i int) string   { return fmt.Sprintf("affected-grp-%d", i)
 func affectedGroupName(i int) string { return fmt.Sprintf("AffectedGroup%d", i) }
 
 func TestCollectGroupChange_PolicyLinked(t *testing.T) {
-	manager, s, accountID, _, groupIDs := setupAffectedPeersTest(t)
+	manager, s, accountID, peerIDs, groupIDs := setupAffectedPeersTest(t)
 	ctx := context.Background()
 
 	_, err := manager.SavePolicy(ctx, accountID, userID, &types.Policy{
 		Enabled: true,
 		Rules: []*types.PolicyRule{
 			{
-				Enabled:       true,
-				Sources:       []string{groupIDs[0]},
-				Destinations:  []string{groupIDs[1]},
-				Bidirectional: true,
-				Action:        types.PolicyTrafficActionAccept,
+				Enabled:             true,
+				Sources:             []string{groupIDs[0]},
+				Destinations:        []string{groupIDs[1]},
+				SourceResource:      types.Resource{ID: peerIDs[0], Type: types.ResourceTypePeer},
+				DestinationResource: types.Resource{ID: peerIDs[1], Type: types.ResourceTypePeer},
+				Bidirectional:       true,
+				Action:              types.PolicyTrafficActionAccept,
+			},
+			{
+				Enabled:             true,
+				Sources:             []string{groupIDs[0]},
+				Destinations:        []string{groupIDs[1]},
+				SourceResource:      types.Resource{ID: peerIDs[2], Type: types.ResourceTypeHost},
+				DestinationResource: types.Resource{ID: peerIDs[3], Type: types.ResourceTypeHost},
+				Bidirectional:       true,
+				Action:              types.PolicyTrafficActionAccept,
+			},
+			{
+				Enabled:             true,
+				Sources:             []string{groupIDs[0]},
+				Destinations:        []string{groupIDs[1]},
+				SourceResource:      types.Resource{ID: "", Type: types.ResourceTypePeer},
+				DestinationResource: types.Resource{ID: "", Type: types.ResourceTypePeer},
+				Bidirectional:       true,
+				Action:              types.PolicyTrafficActionAccept,
 			},
 		},
 	}, true)
 	require.NoError(t, err)
 
-	groups, _ := collectGroupChangeAffectedGroups(ctx, s, accountID, []string{groupIDs[0]})
-	assert.Contains(t, groups, groupIDs[0])
-	assert.Contains(t, groups, groupIDs[1])
+	groups, directPeers := collectGroupChangeAffectedGroups(ctx, s, accountID, []string{groupIDs[0]})
+	assert.ElementsMatch(t, groups, []string{groupIDs[0], groupIDs[1]})
+	assert.ElementsMatch(t, directPeers, []string{peerIDs[1]})
 
-	groups, _ = collectGroupChangeAffectedGroups(ctx, s, accountID, []string{groupIDs[1]})
-	assert.Contains(t, groups, groupIDs[0])
-	assert.Contains(t, groups, groupIDs[1])
+	groups, directPeers = collectGroupChangeAffectedGroups(ctx, s, accountID, []string{groupIDs[1]})
+	assert.ElementsMatch(t, groups, []string{groupIDs[0], groupIDs[1]})
+	assert.ElementsMatch(t, directPeers, []string{peerIDs[0]})
 
-	groups, _ = collectGroupChangeAffectedGroups(ctx, s, accountID, []string{groupIDs[2]})
+	groups, directPeers = collectGroupChangeAffectedGroups(ctx, s, accountID, []string{groupIDs[2]})
 	assert.Empty(t, groups)
+	assert.Empty(t, directPeers)
 }
 
 func TestCollectGroupChange_PolicyWithDirectPeerResource(t *testing.T) {
@@ -133,20 +154,44 @@ func TestCollectGroupChange_PolicyWithDirectPeerResource(t *testing.T) {
 		Enabled: true,
 		Rules: []*types.PolicyRule{
 			{
-				Enabled:        true,
-				Sources:        []string{groupIDs[0]},
-				SourceResource: types.Resource{ID: peerIDs[3], Type: types.ResourceTypePeer},
-				Destinations:   []string{groupIDs[1]},
-				Action:         types.PolicyTrafficActionAccept,
+				Enabled:             true,
+				Sources:             []string{groupIDs[0]},
+				SourceResource:      types.Resource{ID: peerIDs[3], Type: types.ResourceTypePeer},
+				DestinationResource: types.Resource{ID: peerIDs[4], Type: types.ResourceTypePeer},
+				Destinations:        []string{groupIDs[1]},
+				Action:              types.PolicyTrafficActionAccept,
+			},
+			{
+				Enabled:             true,
+				Sources:             []string{groupIDs[0]},
+				SourceResource:      types.Resource{ID: peerIDs[1], Type: types.ResourceTypeHost},
+				DestinationResource: types.Resource{ID: peerIDs[2], Type: types.ResourceTypeHost},
+				Destinations:        []string{groupIDs[1]},
+				Action:              types.PolicyTrafficActionAccept,
+			},
+			{
+				Enabled:             true,
+				Sources:             []string{groupIDs[0]},
+				SourceResource:      types.Resource{ID: "", Type: types.ResourceTypePeer},
+				DestinationResource: types.Resource{ID: "", Type: types.ResourceTypePeer},
+				Destinations:        []string{groupIDs[1]},
+				Action:              types.PolicyTrafficActionAccept,
 			},
 		},
 	}, true)
 	require.NoError(t, err)
 
 	groups, directPeers := collectGroupChangeAffectedGroups(ctx, s, accountID, []string{groupIDs[0]})
-	assert.Contains(t, groups, groupIDs[0])
-	assert.Contains(t, groups, groupIDs[1])
-	assert.Contains(t, directPeers, peerIDs[3])
+	assert.ElementsMatch(t, groups, []string{groupIDs[0], groupIDs[1]})
+	assert.ElementsMatch(t, directPeers, []string{peerIDs[4]})
+
+	groups, directPeers = collectGroupChangeAffectedGroups(ctx, s, accountID, []string{groupIDs[1]})
+	assert.ElementsMatch(t, groups, []string{groupIDs[0], groupIDs[1]})
+	assert.ElementsMatch(t, directPeers, []string{peerIDs[3]})
+
+	groups, directPeers = collectGroupChangeAffectedGroups(ctx, s, accountID, []string{groupIDs[2]})
+	assert.Empty(t, groups)
+	assert.Empty(t, directPeers)
 }
 
 func TestCollectGroupChange_PolicyWithNonPeerResource_NoDirectPeers(t *testing.T) {
@@ -168,8 +213,7 @@ func TestCollectGroupChange_PolicyWithNonPeerResource_NoDirectPeers(t *testing.T
 	require.NoError(t, err)
 
 	groups, directPeers := collectGroupChangeAffectedGroups(ctx, s, accountID, []string{groupIDs[0]})
-	assert.Contains(t, groups, groupIDs[0])
-	assert.Contains(t, groups, groupIDs[1])
+	assert.ElementsMatch(t, groups, []string{groupIDs[0], groupIDs[1]})
 	assert.Empty(t, directPeers, "non-peer resources should not produce direct peer IDs")
 }
 
@@ -294,6 +338,7 @@ func TestCollectGroupChange_NetworkRouterLinked(t *testing.T) {
 		AccountID:  accountID,
 		PeerGroups: []string{groupIDs[0]},
 		Peer:       peerIDs[3],
+		Enabled:    true,
 	})
 	require.NoError(t, err)
 
@@ -324,6 +369,7 @@ func TestCollectGroupChange_NetworkRouterPeerOnlyNoGroups(t *testing.T) {
 		NetworkID: net1.ID,
 		AccountID: accountID,
 		Peer:      peerIDs[4],
+		Enabled:   true,
 	})
 	require.NoError(t, err)
 
@@ -373,17 +419,11 @@ func TestCollectGroupChange_MultipleEntities(t *testing.T) {
 	require.NoError(t, err)
 
 	groups, directPeers := collectGroupChangeAffectedGroups(ctx, s, accountID, []string{groupIDs[0]})
-	assert.Contains(t, groups, groupIDs[0])
-	assert.Contains(t, groups, groupIDs[1])
-	assert.NotContains(t, groups, groupIDs[2])
-	assert.NotContains(t, groups, groupIDs[3])
+	assert.ElementsMatch(t, groups, []string{groupIDs[0], groupIDs[1]})
 	assert.Empty(t, directPeers)
 
 	groups, directPeers = collectGroupChangeAffectedGroups(ctx, s, accountID, []string{groupIDs[3]})
-	assert.Contains(t, groups, groupIDs[2])
-	assert.Contains(t, groups, groupIDs[3])
-	assert.NotContains(t, groups, groupIDs[0])
-	assert.NotContains(t, groups, groupIDs[1])
+	assert.ElementsMatch(t, groups, []string{groupIDs[2], groupIDs[3]})
 	assert.Empty(t, directPeers)
 }
 
@@ -452,8 +492,9 @@ func TestResolveAffectedPeers_PolicyBetweenTwoGroups(t *testing.T) {
 	result = manager.resolveAffectedPeersForPeerChanges(ctx, s, accountID, []string{peerIDs[1]})
 	assert.ElementsMatch(t, []string{peerIDs[0], peerIDs[1]}, result)
 
+	// peerIDs[2] is unrelated to the route; only its own map can change.
 	result = manager.resolveAffectedPeersForPeerChanges(ctx, s, accountID, []string{peerIDs[2]})
-	assert.Empty(t, result)
+	assert.ElementsMatch(t, []string{peerIDs[2]}, result)
 }
 
 func TestResolveAffectedPeers_PolicyThreeGroups(t *testing.T) {
@@ -474,7 +515,7 @@ func TestResolveAffectedPeers_PolicyThreeGroups(t *testing.T) {
 	require.NoError(t, err)
 
 	result := manager.resolveAffectedPeersForPeerChanges(ctx, s, accountID, []string{peerIDs[0]})
-	assert.ElementsMatch(t, []string{peerIDs[0], peerIDs[1], peerIDs[2]}, result)
+	assert.ElementsMatch(t, []string{peerIDs[0], peerIDs[2]}, result)
 }
 
 func TestResolveAffectedPeers_RoutePeerGroups(t *testing.T) {
@@ -506,8 +547,9 @@ func TestResolveAffectedPeers_RoutePeerGroups(t *testing.T) {
 	result = manager.resolveAffectedPeersForPeerChanges(ctx, s, accountID, []string{peerIDs[1]})
 	assert.ElementsMatch(t, []string{peerIDs[0], peerIDs[1]}, result)
 
+	// peerIDs[2] is in no policy; only its own map can change, so it refreshes itself.
 	result = manager.resolveAffectedPeersForPeerChanges(ctx, s, accountID, []string{peerIDs[2]})
-	assert.Empty(t, result)
+	assert.ElementsMatch(t, []string{peerIDs[2]}, result)
 }
 
 func TestResolveAffectedPeers_RouteWithDirectPeer(t *testing.T) {
@@ -564,9 +606,9 @@ func TestResolveAffectedPeers_RouteWithAccessControlGroups(t *testing.T) {
 	result := manager.resolveAffectedPeersForPeerChanges(ctx, s, accountID, []string{peerIDs[2]})
 	assert.ElementsMatch(t, []string{peerIDs[0], peerIDs[1], peerIDs[2]}, result)
 
-	// peer3 is unrelated
+	// peer3 is unrelated to the route; only its own map can change.
 	result = manager.resolveAffectedPeersForPeerChanges(ctx, s, accountID, []string{peerIDs[3]})
-	assert.Empty(t, result)
+	assert.ElementsMatch(t, []string{peerIDs[3]}, result)
 }
 
 func TestResolveAffectedPeers_NetworkRouter(t *testing.T) {
@@ -587,6 +629,7 @@ func TestResolveAffectedPeers_NetworkRouter(t *testing.T) {
 		AccountID:  accountID,
 		PeerGroups: []string{groupIDs[0]},
 		Peer:       peerIDs[3],
+		Enabled:    true,
 	})
 	require.NoError(t, err)
 
@@ -659,9 +702,13 @@ func TestResolveAffectedPeers_PeerInMultipleGroups(t *testing.T) {
 	}, true)
 	require.NoError(t, err)
 
-	// peer0 is in group0 AND group1, so both policies apply
+	// peer0 is in group0 AND group1, so both policies apply. A peer change folds
+	// only the changed peer plus the opposite side of each rule: group2 (peer2) via
+	// the group0 policy and group3 (peer3) via the group1 policy. peer1, a co-member
+	// of group1, is a sibling of the changed peer and must NOT refresh.
 	result := manager.resolveAffectedPeersForPeerChanges(ctx, s, accountID, []string{peerIDs[0]})
-	assert.ElementsMatch(t, []string{peerIDs[0], peerIDs[1], peerIDs[2], peerIDs[3]}, result)
+	assert.ElementsMatch(t, []string{peerIDs[0], peerIDs[2], peerIDs[3]}, result)
+	assert.NotContains(t, result, peerIDs[1], "co-member of the changed peer's group must not refresh")
 }
 
 func TestResolveAffectedPeers_MultipleChangedPeers(t *testing.T) {
@@ -697,7 +744,7 @@ func TestResolveAffectedPeers_MultipleChangedPeers(t *testing.T) {
 	require.NoError(t, err)
 
 	result := manager.resolveAffectedPeersForPeerChanges(ctx, s, accountID, []string{peerIDs[0], peerIDs[2]})
-	assert.ElementsMatch(t, []string{peerIDs[0], peerIDs[1], peerIDs[2], peerIDs[3]}, result)
+	assert.ElementsMatch(t, []string{peerIDs[0], peerIDs[2], peerIDs[1], peerIDs[3]}, result)
 }
 
 func TestResolveAffectedPeers_SharedGroupAcrossPolicyAndRoute(t *testing.T) {
@@ -854,8 +901,9 @@ func TestAffectedPeers_IsolatedPolicies(t *testing.T) {
 	assert.NotContains(t, result, peerIDs[0])
 	assert.NotContains(t, result, peerIDs[1])
 
+	// peerIDs[4] is in neither isolated policy; only its own map can change.
 	result = manager.resolveAffectedPeersForPeerChanges(ctx, s, accountID, []string{peerIDs[4]})
-	assert.Empty(t, result)
+	assert.ElementsMatch(t, []string{peerIDs[4]}, result)
 }
 
 func TestAffectedPeers_IsolatedRouteAndPolicy(t *testing.T) {
@@ -977,12 +1025,13 @@ func TestAffectedPeers_GroupUpdateOnlyAffectsLinkedPeers(t *testing.T) {
 	})
 }
 
-func TestAffectedPeers_UnlinkedGroupChange_NoUpdates(t *testing.T) {
+// A peer in no policy/route refreshes only itself — no other peer is affected.
+func TestAffectedPeers_UnlinkedPeerChange_RefreshesSelfOnly(t *testing.T) {
 	manager, s, accountID, peerIDs, _ := setupAffectedPeersTest(t)
 	ctx := context.Background()
 
 	result := manager.resolveAffectedPeersForPeerChanges(ctx, s, accountID, []string{peerIDs[0]})
-	assert.Empty(t, result)
+	assert.ElementsMatch(t, []string{peerIDs[0]}, result)
 }
 
 // TestAffectedPeers_PolicyChange_UnrelatedPeerNoUpdate verifies that creating/deleting a
@@ -1332,6 +1381,7 @@ func TestAffectedPeers_NetworkRouterUnlinkedPeerNoUpdate(t *testing.T) {
 		NetworkID:  net1.ID,
 		AccountID:  accountID,
 		PeerGroups: []string{"nr-grpA"},
+		Enabled:    true,
 	})
 	require.NoError(t, err)
 
@@ -1755,7 +1805,9 @@ func TestCollectAffectedFromProxyServices_GroupContainingTargetPeerChanged(t *te
 	assert.Contains(t, directPeers, peerIDs[1], "target peer must be refreshed")
 }
 
-func TestCollectAffectedFromProxyServices_DisabledServiceStillMatches(t *testing.T) {
+// A disabled service in the snapshot proxies nothing, so it is skipped: a changed
+// target peer does not pull in the service's proxy peer.
+func TestCollectAffectedFromProxyServices_DisabledServiceSkipped(t *testing.T) {
 	manager, s, accountID, peerIDs, _ := setupAffectedPeersTest(t)
 	ctx := context.Background()
 
@@ -1781,8 +1833,7 @@ func TestCollectAffectedFromProxyServices_DisabledServiceStillMatches(t *testing
 	require.NoError(t, s.CreateService(ctx, svc))
 
 	_, directPeers := collectPeerChangeAffectedGroups(ctx, manager.Store, accountID, nil, []string{peerIDs[1]})
-	assert.Contains(t, directPeers, peerIDs[0], "disabled service should still trigger a refresh so peers are ready when re-enabled")
-	assert.Contains(t, directPeers, peerIDs[1], "disabled target should still trigger a refresh")
+	assert.NotContains(t, directPeers, peerIDs[0], "a disabled service proxies nothing, so its proxy peer must not be folded")
 }
 
 func TestCollectAffectedFromProxyServices_NonPeerTargetType(t *testing.T) {

management/server/affectedpeers/resolver.go

@@ -6,7 +6,12 @@
 //     and before a delete/removal severs the old state).
 //   - Snapshot.Expand: in-memory walk, no store access. Run AFTER the tx commits.
 //
-// Enabled is never consulted: toggling it is itself an observable change.
+// Enabled handling differs by source. Disabled objects in the SNAPSHOT (existing
+// account policies/resources/routers/routes/proxy services and their rules/targets)
+// route to nobody and are skipped — they cannot affect any peer's map. Objects in
+// the CHANGE itself are processed regardless of Enabled, so disabling one still
+// refreshes the peers that lose access (the toggle is the observable change, and the
+// update carries the old∪new state).
 package affectedpeers
 
 import (
@@ -61,7 +66,8 @@ func Load(ctx context.Context, s store.Store, accountID string, c Change) (*Snap
 // loadCollections reads the policy/route/nameserver/dns/router/resource/proxy
 // collections a Change can touch, gated to what the walk needs.
 func (snap *Snapshot) loadCollections(ctx context.Context, s store.Store, accountID string, c Change) error {
-	hasGroupOrPeerChange := len(c.ChangedGroupIDs) > 0 || len(c.ChangedPeerIDs) > 0 || len(c.Resources) > 0
+	// LinkGroups drive the same policy/route/dns walk as a changed group or peer.
+	hasGroupOrPeerChange := len(c.ChangedGroupIDs) > 0 || len(c.ChangedPeerIDs) > 0 || len(c.LinkGroups) > 0 || len(c.Resources) > 0
 	hasNetworkObject := len(c.Routers) > 0 || len(c.Resources) > 0 || len(c.Networks) > 0
 	// the resource<->router bridge can fire for any of these
 	needsRoutersResources := hasGroupOrPeerChange || len(c.PostureCheckIDs) > 0 || len(c.Policies) > 0 || hasNetworkObject
@@ -76,7 +82,7 @@ func (snap *Snapshot) loadCollections(ctx context.Context, s store.Store, accoun
 			return err
 		}
 	}
-	if len(c.ChangedGroupIDs) > 0 || len(c.ChangedPeerIDs) > 0 {
+	if len(c.ChangedGroupIDs) > 0 || len(c.ChangedPeerIDs) > 0 || len(c.LinkGroups) > 0 {
 		if err := snap.loadDNS(ctx, s, accountID); err != nil {
 			return err
 		}
@@ -174,6 +180,24 @@ type Change struct {
 	// folded in — but only when the group is linked (an unlinked group has no map
 	// impact), matching how current members are handled.
 	RemovedPeersByGroup map[string][]string
+
+	// OutputPeerIDs are peers folded straight into the result without seeding their
+	// group memberships into the walk. Use for the peer whose group membership changed:
+	// the peer itself must refresh, but its OTHER groups did not change, so they must
+	// not be walked. Contrast ChangedPeerIDs, which seeds ALL of the peer's groups
+	// (correct when the peer's own attributes changed, e.g. IP/status).
+	OutputPeerIDs []string
+
+	// LinkGroups are groups used ONLY to match policies/routes/routers and walk to the
+	// OPPOSITE side — they are never expanded to their own members. Use this when a
+	// peer's group membership changed: pass the peer in ChangedPeerIDs and its
+	// group(s) here. The opposite side of the policies the group participates in
+	// refreshes, but the group's other members (siblings) do not — nothing changed for
+	// them. For an intra-group policy (A→A) the opposite side IS the group, so its
+	// members still refresh via the opposite-side fold, exactly when they genuinely
+	// gain/lose the changed peer. Unlike ChangedGroupIDs, a LinkGroup is not added to
+	// the output, so a one-sided membership change never wakes the whole group.
+	LinkGroups []string
 }
 
 func (c Change) isEmpty() bool {
@@ -186,7 +210,9 @@ func (c Change) isEmpty() bool {
 		len(c.Networks) == 0 &&
 		len(c.PostureCheckIDs) == 0 &&
 		len(c.DistributionGroupIDs) == 0 &&
-		len(c.RemovedPeersByGroup) == 0
+		len(c.RemovedPeersByGroup) == 0 &&
+		len(c.LinkGroups) == 0 &&
+		len(c.OutputPeerIDs) == 0
 }
 
 // Expand returns the deduplicated affected peer IDs from the preloaded Snapshot,
@@ -197,8 +223,8 @@ func (snap *Snapshot) Expand(ctx context.Context, accountID string, c Change) []
 		return nil
 	}
 	r := newResolver(ctx, snap, accountID, c)
-	log.WithContext(ctx).Tracef("affectedpeers expand start: account=%s changedGroups=%v changedPeers=%v policies=%d routes=%d routers=%d resources=%d networks=%d postureChecks=%v distributionGroups=%v",
-		accountID, c.ChangedGroupIDs, c.ChangedPeerIDs, len(c.Policies), len(c.Routes), len(c.Routers), len(c.Resources), len(c.Networks), c.PostureCheckIDs, c.DistributionGroupIDs)
+	log.WithContext(ctx).Tracef("affectedpeers expand start: account=%s changedGroups=%v changedPeers=%v linkGroups=%v policies=%d routes=%d routers=%d resources=%d networks=%d postureChecks=%v distributionGroups=%v",
+		accountID, c.ChangedGroupIDs, c.ChangedPeerIDs, c.LinkGroups, len(c.Policies), len(c.Routes), len(c.Routers), len(c.Resources), len(c.Networks), c.PostureCheckIDs, c.DistributionGroupIDs)
 	r.walk()
 	return r.expand()
 }
@@ -216,57 +242,84 @@ func Collect(ctx context.Context, s store.Store, accountID string, c Change) (gr
 	}
 	r := newResolver(ctx, snap, accountID, c)
 	r.walk()
-	return setToSlice(r.groupSet), setToSlice(r.peerSet)
+	return setToSlice(r.affectedGroups), setToSlice(r.affectedPeers)
 }
 
 func newResolver(ctx context.Context, snap *Snapshot, accountID string, c Change) *resolver {
 	r := &resolver{
-		ctx:             ctx,
-		snap:            snap,
-		accountID:       accountID,
-		change:          c,
-		changedGroupSet: toSet(c.ChangedGroupIDs),
-		changedPeerSet:  toSet(c.ChangedPeerIDs),
-		groupSet:        make(map[string]struct{}),
-		peerSet:         make(map[string]struct{}),
-		networkIDs:      make(map[string]struct{}),
+		ctx:            ctx,
+		snap:           snap,
+		accountID:      accountID,
+		change:         c,
+		linkGroups:     toSet(c.ChangedGroupIDs),
+		outputGroups:   toSet(c.ChangedGroupIDs),
+		changedPeers:   toSet(c.ChangedPeerIDs),
+		affectedGroups: make(map[string]struct{}),
+		affectedPeers:  make(map[string]struct{}),
 	}
+	// LinkGroups match policies/routes to find the opposite side but are NOT output:
+	// they go into linkGroups only, never outputGroups, so their members never fold in.
+	addAll(r.linkGroups, c.LinkGroups)
 	// Resolve each changed peer to its groups here so callers pass only ChangedPeerIDs.
 	r.seedChangedGroupsFromPeers()
-	r.matchedPolicies = append(r.matchedPolicies, c.Policies...)
 	return r
 }
 
-// seedChangedGroupsFromPeers adds each changed peer's groups to changedGroupSet so
+// seedChangedGroupsFromPeers adds each changed peer's groups to linkGroups so
 // the group-driven walkers fire for memberships, not just direct peer references.
+// These seeded groups are for MATCHING only — folding the changed entity's own
+// side is gated on outputGroups (the caller-reported groups), so a seeded group
+// never folds its whole membership; only the changed peer itself folds in.
 func (r *resolver) seedChangedGroupsFromPeers() {
-	if len(r.changedPeerSet) == 0 {
+	if len(r.changedPeers) == 0 {
 		return
 	}
 	for groupID, members := range r.snap.groupPeers {
-		for pID := range r.changedPeerSet {
+		for pID := range r.changedPeers {
 			if _, ok := members[pID]; ok {
-				r.changedGroupSet[groupID] = struct{}{}
+				r.linkGroups[groupID] = struct{}{}
 				break
 			}
 		}
 	}
 }
 
+// policySide selects which side of a policy rule to walk.
+type policySide int
+
+const (
+	sideSource policySide = iota
+	sideDestination
+)
+
+func (s policySide) opposite() policySide {
+	if s == sideSource {
+		return sideDestination
+	}
+	return sideSource
+}
+
+// walk resolves affected peers in two buckets, by how far each change propagates.
+//
+// BOTH-SIDES — the rule itself changed (an explicit policy edit, or a policy whose
+// posture check changed). Source AND destination refresh, so each such policy is
+// walked on both sides.
+//
+// OPPOSITE-SIDE — an endpoint moved but no rule changed. For each policy the change
+// touches we fold only the side AWAY from the change:
+//   - a changed peer/group sits ON a policy side -> fold the opposite side;
+//   - a changed router/resource/network sits on a NETWORK -> fold the SOURCE side of
+//     the policies whose destination reaches it (and the routers it implies).
+//
+// Routes, nameserver groups, DNS and embedded-proxy services distribute to their own
+// member peers, outside the policy graph, and are folded here too.
 func (r *resolver) walk() {
-	r.collectFromExplicitPolicies()
-	r.collectFromExplicitRoutes(r.change.Routes)
-	r.collectFromExplicitRouters(r.change.Routers)
-	r.collectFromExplicitResources(r.change.Resources)
-	r.collectFromExplicitNetworks(r.change.Networks)
-	r.collectFromPostureChecks(r.change.PostureCheckIDs)
+	for _, policy := range r.bothSidesPolicies() {
+		r.foldPolicySide(policy, sideSource)
+		r.foldPolicySide(policy, sideDestination)
+	}
 
-	// Distribution groups (nameserver/DNS) affect only their member peers: fold them
-	// straight into groupSet so expand() maps them to members, without the policy/
-	// route walk that changedGroupSet would trigger.
-	addAll(r.groupSet, r.change.DistributionGroupIDs)
-
-	if len(r.changedGroupSet) > 0 || len(r.changedPeerSet) > 0 {
+	if len(r.linkGroups) > 0 || len(r.changedPeers) > 0 {
 		r.collectFromPolicies()
 		r.collectFromRoutes()
 		r.collectFromNameServers()
@@ -275,7 +328,31 @@ func (r *resolver) walk() {
 		r.collectFromProxyServices()
 	}
 
-	r.collectResourceRouterBridge()
+	r.collectFromChangedRoutes(r.change.Routes)
+	r.collectFromChangedRouters(r.change.Routers)
+	r.collectFromChangedResources(r.change.Resources)
+	r.collectFromChangedNetworks(r.change.Networks)
+
+	// The explicitly changed peers always refresh their own maps. OnPeersUpdated only
+	// refreshes the resolver's output (it ignores the separately-passed changed peers),
+	// so the changed peer reaches its own new map only via here. An offline/deleted
+	// peer in the set is filtered downstream (filterConnectedAffectedPeers).
+	addAll(r.affectedPeers, setToSlice(r.changedPeers))
+	// OutputPeerIDs refresh themselves too, but unlike changedPeers their group
+	// memberships were not seeded into the walk (only the changed group was).
+	addAll(r.affectedPeers, r.change.OutputPeerIDs)
+
+	// Distribution groups (nameserver/DNS) affect only their member peers: fold them
+	// straight into affectedGroups so expand() maps them to members, without the
+	// policy/route walk that linkGroups would trigger.
+	addAll(r.affectedGroups, r.change.DistributionGroupIDs)
+}
+
+// bothSidesPolicies are the policies whose rule changed: the explicitly edited ones
+// plus those gated by a changed posture check. walk folds both their sides.
+func (r *resolver) bothSidesPolicies() []*types.Policy {
+	policies := append([]*types.Policy(nil), r.change.Policies...)
+	return r.appendPoliciesForPostureChecks(policies, r.change.PostureCheckIDs)
 }
 
 type resolver struct {
@@ -284,27 +361,71 @@ type resolver struct {
 	accountID string
 	change    Change
 
-	changedGroupSet map[string]struct{}
-	changedPeerSet  map[string]struct{}
+	// Inputs — what changed. Set once at construction, read-only during the walk
+	// (except linkGroups, which collectFromExplicitResources also seeds).
+	//
+	// linkGroups is the MATCH set: caller-changed groups ∪ the groups of changed
+	// peers ∪ changed-resource groups. A rule/route/router matches the change when
+	// one of its groups is here — used only to find the opposite side to fold.
+	//
+	// outputGroups is the FOLD-WHOLE-GROUP set: ONLY Change.ChangedGroupIDs. When a
+	// matched group is here, its whole membership is affected. A peer-seeded group
+	// is in linkGroups but NOT outputGroups, so it folds only the changed peer
+	// (changedPeers), never its siblings.
+	linkGroups   map[string]struct{}
+	outputGroups map[string]struct{}
+	changedPeers map[string]struct{}
 
-	groupSet map[string]struct{}
-	peerSet  map[string]struct{}
-
-	matchedPolicies []*types.Policy
-	networkIDs      map[string]struct{}
+	// Outputs — the answer. The only sets the walk accumulates into. affectedGroups
+	// is expanded to its member peers in expand().
+	affectedGroups map[string]struct{}
+	affectedPeers  map[string]struct{}
 }
 
-func (r *resolver) policies() []*types.Policy { return r.snap.policies }
+// policies returns the account's ENABLED policies from the snapshot. Disabled
+// policies grant no access, so the walk skips them when scanning existing account
+// data. Explicitly changed policies (Change.Policies, via bothSidesPolicies) are
+// processed regardless of Enabled, so disabling one still refreshes its peers.
+func (r *resolver) policies() []*types.Policy {
+	enabled := make([]*types.Policy, 0, len(r.snap.policies))
+	for _, policy := range r.snap.policies {
+		if policy != nil && policy.Enabled {
+			enabled = append(enabled, policy)
+		}
+	}
+	return enabled
+}
 
-func (r *resolver) networkResources() []*resourceTypes.NetworkResource { return r.snap.resources }
+// networkResources / networkRouters return the account's ENABLED resources/routers
+// from the snapshot. Disabled objects route to nobody, so the walk skips them when
+// it scans existing account data. The explicitly changed objects in the Change are
+// processed regardless of Enabled (collectFromChanged*), so disabling one still
+// refreshes the peers that lose access.
+func (r *resolver) networkResources() []*resourceTypes.NetworkResource {
+	enabled := make([]*resourceTypes.NetworkResource, 0, len(r.snap.resources))
+	for _, resource := range r.snap.resources {
+		if resource.Enabled {
+			enabled = append(enabled, resource)
+		}
+	}
+	return enabled
+}
 
-func (r *resolver) networkRouters() []*routerTypes.NetworkRouter { return r.snap.routers }
+func (r *resolver) networkRouters() []*routerTypes.NetworkRouter {
+	enabled := make([]*routerTypes.NetworkRouter, 0, len(r.snap.routers))
+	for _, router := range r.snap.routers {
+		if router.Enabled {
+			enabled = append(enabled, router)
+		}
+	}
+	return enabled
+}
 
 // peerIDsForGroups maps a group set to its member peer IDs via the preloaded index.
-func (r *resolver) peerIDsForGroups(groupSet map[string]struct{}) []string {
+func (r *resolver) peerIDsForGroups(groups map[string]struct{}) []string {
 	seen := make(map[string]struct{})
 	var ids []string
-	for gID := range groupSet {
+	for gID := range groups {
 		for pID := range r.snap.groupPeers[gID] {
 			if _, ok := seen[pID]; ok {
 				continue
@@ -317,25 +438,25 @@ func (r *resolver) peerIDsForGroups(groupSet map[string]struct{}) []string {
 }
 
 func (r *resolver) expand() []string {
-	peerIDs := r.peerIDsForGroups(r.groupSet)
+	peerIDs := r.peerIDsForGroups(r.affectedGroups)
 
 	log.WithContext(r.ctx).Tracef("affectedpeers expand: account=%s affectedGroups=%v -> %d group-member peers; direct peers=%v",
-		r.accountID, setToSlice(r.groupSet), len(peerIDs), setToSlice(r.peerSet))
+		r.accountID, setToSlice(r.affectedGroups), len(peerIDs), setToSlice(r.affectedPeers))
 
 	seen := make(map[string]struct{}, len(peerIDs))
 	for _, id := range peerIDs {
 		seen[id] = struct{}{}
 	}
-	for id := range r.peerSet {
+	for id := range r.affectedPeers {
 		if _, ok := seen[id]; !ok {
 			peerIDs = append(peerIDs, id)
 			seen[id] = struct{}{}
 		}
 	}
 
-	// Fold in removed peers only when their group is linked (in groupSet).
+	// Fold in removed peers only when their group is linked (in affectedGroups).
 	for groupID, removed := range r.change.RemovedPeersByGroup {
-		if _, linked := r.groupSet[groupID]; !linked {
+		if _, linked := r.affectedGroups[groupID]; !linked {
 			continue
 		}
 		for _, id := range removed {
@@ -351,169 +472,349 @@ func (r *resolver) expand() []string {
 	return peerIDs
 }
 
-func (r *resolver) collectFromExplicitPolicies() {
-	for _, policy := range r.matchedPolicies {
-		if policy == nil {
-			continue
+// ruleSideGroups / ruleSideResource return the groups and the resource on the given
+// side of a rule.
+func ruleSideGroups(rule *types.PolicyRule, side policySide) []string {
+	if side == sideDestination {
+		return rule.Destinations
+	}
+	return rule.Sources
+}
+
+func ruleSideResource(rule *types.PolicyRule, side policySide) types.Resource {
+	if side == sideDestination {
+		return rule.DestinationResource
+	}
+	return rule.SourceResource
+}
+
+// foldPolicySide folds one side of a policy down to affected peers: its groups
+// (resolved to members in expand) and its direct peer. When the side is the
+// DESTINATION and references a network resource (directly or via a destination
+// group's resources), it also folds the routers that serve that resource's network
+// — a destination resource is reached through its routers. A resource on the SOURCE
+// side routes to nobody (GetPoliciesForNetworkResource matches destinations only),
+// so the router hop is destination-only.
+func (r *resolver) foldPolicySide(policy *types.Policy, side policySide) {
+	if policy == nil {
+		return
+	}
+	for _, rule := range policy.Rules {
+		addAll(r.affectedGroups, ruleSideGroups(rule, side))
+		res := ruleSideResource(rule, side)
+		if res.Type == types.ResourceTypePeer && res.ID != "" {
+			r.affectedPeers[res.ID] = struct{}{}
 		}
-		log.WithContext(r.ctx).Tracef("collectFromExplicitPolicies: changed policy %s (%s) -> folding rule groups %v + direct peers",
-			policy.ID, policy.Name, policy.RuleGroups())
-		addAll(r.groupSet, policy.RuleGroups())
-		collectPolicyDirectPeers(policy, r.peerSet)
+	}
+	if side == sideDestination {
+		r.foldRoutersForResources(r.policyDestinationResourceIDs(policy))
 	}
 }
 
-func (r *resolver) collectFromExplicitRoutes(routes []*route.Route) {
+// appendPoliciesForPostureChecks appends every policy that references a changed
+// posture check (a rule change, so walk both sides).
+func (r *resolver) appendPoliciesForPostureChecks(policies []*types.Policy, postureCheckIDs []string) []*types.Policy {
+	if len(postureCheckIDs) == 0 {
+		return policies
+	}
+	ids := toSet(postureCheckIDs)
+	for _, policy := range r.policies() {
+		if !policyReferencesPostureChecks(policy, ids) || !policy.Enabled {
+			continue
+		}
+		log.WithContext(r.ctx).Tracef("appendPoliciesForPostureChecks: policy %s (%s) references changed posture checks %v -> both-sides policy",
+			policy.ID, policy.Name, postureCheckIDs)
+		policies = append(policies, policy)
+	}
+	return policies
+}
+
+// collectFromPolicies folds, for every policy whose rule a changed group or peer
+// touches, only the OPPOSITE side (down to peers, incl. destination routers), plus
+// the changed entity's own side: the changed group's whole membership when the
+// group itself changed (outputGroups), or the changed peer alone when matched via a
+// peer-seeded group (never its co-members).
+func (r *resolver) collectFromPolicies() {
+	for _, policy := range r.policies() {
+		for _, rule := range policy.Rules {
+			if !rule.Enabled {
+				continue // a disabled rule grants no access
+			}
+			r.foldRuleSideIfChanged(policy, rule, sideSource)
+			r.foldRuleSideIfChanged(policy, rule, sideDestination)
+		}
+	}
+}
+
+// foldRuleSideIfChanged: when a changed group or direct peer sits on `side` of the
+// rule, fold the opposite side fully (groups/peers + destination routers) and fold
+// the changed entity's own side (the whole changed group, or the changed peer alone).
+func (r *resolver) foldRuleSideIfChanged(policy *types.Policy, rule *types.PolicyRule, side policySide) {
+	nearGroups := ruleSideGroups(rule, side)
+	nearResource := ruleSideResource(rule, side)
+
+	matchedByGroup := anyInSet(nearGroups, r.linkGroups)
+	matchedByPeer := isDirectPeerInSet(nearResource, r.changedPeers)
+	if !matchedByGroup && !matchedByPeer {
+		return
+	}
+
+	// Opposite side, fully down to peers (a destination opposite also folds routers).
+	r.foldPolicySideForRule(policy, rule, side.opposite())
+
+	// Own side: fold the whole changed group's members only when the group itself
+	// changed (outputGroups). A peer-seeded or link-only group is not folded here —
+	// its siblings never refresh. The changed peers themselves are folded once, after
+	// the walk (see walk()).
+	for _, gID := range nearGroups {
+		if _, ok := r.outputGroups[gID]; ok {
+			r.affectedGroups[gID] = struct{}{}
+		}
+	}
+
+	// When the changed side IS a destination, the resources it targets are reached
+	// through their network's routers, so those routers refresh too (e.g. attaching a
+	// resource to a destination group, or a changed destination group/resource).
+	if side == sideDestination {
+		r.foldRoutersForResources(r.ruleDestinationResourceIDs(rule))
+	}
+}
+
+// foldPolicySideForRule folds one side of a single rule (groups + direct peer), and
+// for a destination side the routers of that rule's destination resources.
+func (r *resolver) foldPolicySideForRule(policy *types.Policy, rule *types.PolicyRule, side policySide) {
+	addAll(r.affectedGroups, ruleSideGroups(rule, side))
+	res := ruleSideResource(rule, side)
+	if res.Type == types.ResourceTypePeer && res.ID != "" {
+		r.affectedPeers[res.ID] = struct{}{}
+	}
+	if side == sideDestination {
+		r.foldRoutersForResources(r.ruleDestinationResourceIDs(rule))
+	}
+}
+
+// collectFromChangedRoutes folds an explicitly changed route's own groups and peer.
+func (r *resolver) collectFromChangedRoutes(routes []*route.Route) {
 	for _, rt := range routes {
 		if rt == nil {
 			continue
 		}
-		log.WithContext(r.ctx).Tracef("collectFromExplicitRoutes: changed route %s -> folding groups=%v peerGroups=%v accessControlGroups=%v peer=%q",
+		log.WithContext(r.ctx).Tracef("collectFromChangedRoutes: changed route %s -> folding groups=%v peerGroups=%v accessControlGroups=%v peer=%q",
 			rt.ID, rt.Groups, rt.PeerGroups, rt.AccessControlGroups, rt.Peer)
-		addAll(r.groupSet, rt.Groups, rt.PeerGroups, rt.AccessControlGroups)
+		addAll(r.affectedGroups, rt.Groups, rt.PeerGroups, rt.AccessControlGroups)
 		if rt.Peer != "" {
-			r.peerSet[rt.Peer] = struct{}{}
+			r.affectedPeers[rt.Peer] = struct{}{}
 		}
 	}
 }
 
-// collectFromExplicitRouters folds changed routers' peers and marks their networks
-// for the bridge. Passing the old router keeps a repointed router's previous peers
-// affected without a post-commit read.
-func (r *resolver) collectFromExplicitRouters(routers []*routerTypes.NetworkRouter) {
+// collectFromChangedRouters: a changed router refreshes its OWN backing peer/groups
+// (the changed entity) and the SOURCE side of every policy reaching a resource on
+// its network (the router serves the whole network). Sibling routers on the network
+// are independent and are NOT folded. Passing the old router state keeps a repointed
+// router's previous backing affected without a post-commit read.
+func (r *resolver) collectFromChangedRouters(routers []*routerTypes.NetworkRouter) {
 	for _, router := range routers {
 		if router == nil {
 			continue
 		}
-		log.WithContext(r.ctx).Tracef("collectFromExplicitRouters: changed router %s on network %s -> folding peerGroups=%v peer=%q and marking network for source bridge",
+		log.WithContext(r.ctx).Tracef("collectFromChangedRouters: changed router %s on network %s -> folding its own peerGroups=%v peer=%q + sources reaching network resources",
 			router.ID, router.NetworkID, router.PeerGroups, router.Peer)
-		addAll(r.groupSet, router.PeerGroups)
+		addAll(r.affectedGroups, router.PeerGroups)
 		if router.Peer != "" {
-			r.peerSet[router.Peer] = struct{}{}
+			r.affectedPeers[router.Peer] = struct{}{}
 		}
 		if router.NetworkID != "" {
-			r.networkIDs[router.NetworkID] = struct{}{}
+			r.foldPolicySourcesForResources(r.networkResourceIDs(router.NetworkID))
 		}
 	}
 }
 
-// collectFromExplicitResources marks changed resources' networks for the bridge and
-// treats their group IDs as changed, so policies targeting the resource via a
-// now-detached (old) group still refresh.
-func (r *resolver) collectFromExplicitResources(resources []*resourceTypes.NetworkResource) {
+// collectFromChangedResources: a changed resource refreshes the SOURCE side of the
+// policies targeting EXACTLY that resource — directly, or via one of the resource's
+// own groups (old∪new across the change, so a now-detached group's sources still
+// refresh) — plus the routers serving its network (the resource is reached through
+// them). It does not touch sibling resources on the same network.
+func (r *resolver) collectFromChangedResources(resources []*resourceTypes.NetworkResource) {
 	for _, resource := range resources {
 		if resource == nil {
 			continue
 		}
-		log.WithContext(r.ctx).Tracef("collectFromExplicitResources: changed resource %s on network %s -> marking network for bridge and treating groups %v as changed",
+		log.WithContext(r.ctx).Tracef("collectFromChangedResources: changed resource %s on network %s (groups %v) -> folding sources of policies targeting it + its network's routers",
 			resource.ID, resource.NetworkID, resource.GroupIDs)
-		addAll(r.changedGroupSet, resource.GroupIDs)
+		r.foldPolicySourcesForResource(resource.ID, resource.GroupIDs)
 		if resource.NetworkID != "" {
-			r.networkIDs[resource.NetworkID] = struct{}{}
+			r.foldRoutersOnNetworks(map[string]struct{}{resource.NetworkID: {}})
 		}
 	}
 }
 
-// collectFromExplicitNetworks marks changed networks for the bridge. A network has
-// no groups/peers of its own.
-func (r *resolver) collectFromExplicitNetworks(networks []*networkTypes.Network) {
-	for _, network := range networks {
-		if network == nil {
+// foldPolicySourcesForResource folds the source side of every policy whose
+// destination is the given resource — referenced directly, or via any of the given
+// groups (the resource's own old∪new groups, which captures a detached group).
+func (r *resolver) foldPolicySourcesForResource(resourceID string, groupIDs []string) {
+	groups := toSet(groupIDs)
+	for _, policy := range r.policies() {
+		if !policyTargetsResourceOrGroups(policy, resourceID, groups) {
 			continue
 		}
-		log.WithContext(r.ctx).Tracef("collectFromExplicitNetworks: changed network %s -> marking for bridge", network.ID)
-		if network.ID != "" {
-			r.networkIDs[network.ID] = struct{}{}
-		}
+		log.WithContext(r.ctx).Tracef("foldPolicySourcesForResource: policy %s (%s) targets changed resource %s -> folding its source groups/peers", policy.ID, policy.Name, resourceID)
+		collectPolicySources(policy, r.affectedGroups, r.affectedPeers)
 	}
 }
 
-func (r *resolver) collectFromPostureChecks(postureCheckIDs []string) {
-	if len(postureCheckIDs) == 0 {
+// policyTargetsResourceOrGroups reports whether a policy's destination is the given
+// resource directly, or one of the given destination groups.
+func policyTargetsResourceOrGroups(policy *types.Policy, resourceID string, groups map[string]struct{}) bool {
+	if policy == nil {
+		return false
+	}
+	for _, rule := range policy.Rules {
+		if !rule.Enabled {
+			continue
+		}
+		if rule.DestinationResource.Type != types.ResourceTypePeer && rule.DestinationResource.ID == resourceID && resourceID != "" {
+			return true
+		}
+		if anyInSet(rule.Destinations, groups) {
+			return true
+		}
+	}
+	return false
+}
+
+// collectFromChangedNetworks: a changed network refreshes the SOURCE side of the
+// policies reaching any of its resources, plus its routers. A network has no
+// groups/peers of its own.
+func (r *resolver) collectFromChangedNetworks(networks []*networkTypes.Network) {
+	for _, network := range networks {
+		if network == nil || network.ID == "" {
+			continue
+		}
+		log.WithContext(r.ctx).Tracef("collectFromChangedNetworks: changed network %s -> folding sources reaching its resources + its routers", network.ID)
+		resourceIDs := r.networkResourceIDs(network.ID)
+		r.foldPolicySourcesForResources(resourceIDs)
+		r.foldRoutersOnNetworks(map[string]struct{}{network.ID: {}})
+	}
+}
+
+// foldPolicySourcesForResources folds the source groups/peers of every policy whose
+// destination targets one of resourceIDs (directly or via a destination group).
+func (r *resolver) foldPolicySourcesForResources(resourceIDs map[string]struct{}) {
+	if len(resourceIDs) == 0 {
 		return
 	}
-	ids := toSet(postureCheckIDs)
 	for _, policy := range r.policies() {
-		if !policyReferencesPostureChecks(policy, ids) {
-			continue
+		if r.policyTargetsResources(policy, resourceIDs) {
+			log.WithContext(r.ctx).Tracef("foldPolicySourcesForResources: policy %s (%s) targets a changed resource -> folding its source groups/peers", policy.ID, policy.Name)
+			collectPolicySources(policy, r.affectedGroups, r.affectedPeers)
 		}
-		log.WithContext(r.ctx).Tracef("collectFromPostureChecks: policy %s (%s) references changed posture checks %v -> folding rule groups %v + direct peers",
-			policy.ID, policy.Name, postureCheckIDs, policy.RuleGroups())
-		addAll(r.groupSet, policy.RuleGroups())
-		collectPolicyDirectPeers(policy, r.peerSet)
-		r.matchedPolicies = append(r.matchedPolicies, policy)
-	}
-}
-
-func (r *resolver) collectFromPolicies() {
-	for _, policy := range r.policies() {
-		matchedByGroup := policyReferencesGroups(policy, r.changedGroupSet)
-		matchedByPeer := len(r.changedPeerSet) > 0 && policyReferencesDirectPeers(policy, r.changedPeerSet)
-		if !matchedByGroup && !matchedByPeer {
-			continue
-		}
-		log.WithContext(r.ctx).Tracef("collectFromPolicies: policy %s (%s) matched (byGroup=%t byPeer=%t) -> folding rule groups %v + direct peers",
-			policy.ID, policy.Name, matchedByGroup, matchedByPeer, policy.RuleGroups())
-		addAll(r.groupSet, policy.RuleGroups())
-		collectPolicyDirectPeers(policy, r.peerSet)
-		r.matchedPolicies = append(r.matchedPolicies, policy)
 	}
 }
 
+// collectFromRoutes folds, per matched route, the OPPOSITE side(s) fully and the
+// matched side's own groups only on a whole-group change (outputGroups). A route has
+// three peer sides — routing (Peer/PeerGroups), consumer (Groups) and ACL
+// (AccessControlGroups) — that each refresh the others; the changed side's own group
+// folds its siblings only when the group itself changed, never on a one-peer move.
 func (r *resolver) collectFromRoutes() {
 	for _, rt := range r.snap.routes {
-		matchedByGroup := anyInSet(rt.Groups, r.changedGroupSet) || anyInSet(rt.PeerGroups, r.changedGroupSet) || anyInSet(rt.AccessControlGroups, r.changedGroupSet)
-		matchedByPeer := rt.Peer != "" && len(r.changedPeerSet) > 0 && isInSet(rt.Peer, r.changedPeerSet)
-		if !matchedByGroup && !matchedByPeer {
+		if !rt.Enabled {
+			continue // disabled routes route to nobody; skip existing account data
+		}
+		routing := anyInSet(rt.PeerGroups, r.linkGroups) || (rt.Peer != "" && isInSet(rt.Peer, r.changedPeers))
+		consumer := anyInSet(rt.Groups, r.linkGroups)
+		acl := anyInSet(rt.AccessControlGroups, r.linkGroups)
+		if !routing && !consumer && !acl {
 			continue
 		}
-		log.WithContext(r.ctx).Tracef("collectFromRoutes: route %s matched (byGroup=%t byPeer=%t) -> folding groups=%v peerGroups=%v accessControlGroups=%v peer=%q",
-			rt.ID, matchedByGroup, matchedByPeer, rt.Groups, rt.PeerGroups, rt.AccessControlGroups, rt.Peer)
-		addAll(r.groupSet, rt.Groups, rt.PeerGroups, rt.AccessControlGroups)
-		if rt.Peer != "" {
-			r.peerSet[rt.Peer] = struct{}{}
+		log.WithContext(r.ctx).Tracef("collectFromRoutes: route %s matched (routing=%t consumer=%t acl=%t) -> folding opposite sides; own side gated on outputGroups",
+			rt.ID, routing, consumer, acl)
+		r.foldRouteSide(rt.PeerGroups, routing)
+		r.foldRouteSide(rt.Groups, consumer)
+		r.foldRouteSide(rt.AccessControlGroups, acl)
+		// The single routing Peer folds when the routing side is the OPPOSITE of the
+		// match (consumer/acl need it), or when that very peer is the change.
+		if rt.Peer != "" && (consumer || acl || isInSet(rt.Peer, r.changedPeers)) {
+			r.affectedPeers[rt.Peer] = struct{}{}
+		}
+	}
+}
+
+// foldRouteSide folds a route side: when this side is the one that matched, fold its
+// groups only on a whole-group change (outputGroups) so siblings of a single moved
+// peer stay put; otherwise it is an opposite side and folds fully.
+func (r *resolver) foldRouteSide(groups []string, matchedHere bool) {
+	if matchedHere {
+		r.foldOutputGroups(groups)
+		return
+	}
+	addAll(r.affectedGroups, groups)
+}
+
+// foldOutputGroups folds only the groups that the caller reported as wholly changed
+// (outputGroups). Used for a matched object's OWN side, where a peer-seeded or
+// link-only group must not pull in its siblings.
+func (r *resolver) foldOutputGroups(groups ...[]string) {
+	for _, gs := range groups {
+		for _, gID := range gs {
+			if _, ok := r.outputGroups[gID]; ok {
+				r.affectedGroups[gID] = struct{}{}
+			}
 		}
 	}
 }
 
 func (r *resolver) collectFromNameServers() {
-	if len(r.changedGroupSet) == 0 {
+	if len(r.linkGroups) == 0 {
 		return
 	}
 	for _, ns := range r.snap.nsGroups {
-		if anyInSet(ns.Groups, r.changedGroupSet) {
-			log.WithContext(r.ctx).Tracef("collectFromNameServers: nameserver group %s references a changed group -> folding its groups %v", ns.ID, ns.Groups)
-			addAll(r.groupSet, ns.Groups)
+		if anyInSet(ns.Groups, r.linkGroups) {
+			// A nameserver group has no opposite side: a peer's DNS config depends only
+			// on its own membership, so a one-peer move refreshes that peer alone (folded
+			// elsewhere). Fold the referenced groups only on a whole-group change.
+			log.WithContext(r.ctx).Tracef("collectFromNameServers: nameserver group %s references a linked group -> folding its groups %v (outputGroups only)", ns.ID, ns.Groups)
+			r.foldOutputGroups(ns.Groups)
 		}
 	}
 }
 
 func (r *resolver) collectFromDNSSettings() {
-	if len(r.changedGroupSet) == 0 || r.snap.dnsSettings == nil {
+	if len(r.linkGroups) == 0 || r.snap.dnsSettings == nil {
 		return
 	}
 	for _, gID := range r.snap.dnsSettings.DisabledManagementGroups {
-		if _, ok := r.changedGroupSet[gID]; ok {
+		if _, ok := r.linkGroups[gID]; ok {
 			log.WithContext(r.ctx).Tracef("collectFromDNSSettings: changed group %s is in DisabledManagementGroups -> folding it", gID)
-			r.groupSet[gID] = struct{}{}
+			r.affectedGroups[gID] = struct{}{}
 		}
 	}
 }
 
+// collectFromNetworkRouters handles a changed group/peer that BACKS a router (the
+// routing peer set moved): the router's own peers refresh and so do the sources of
+// the policies reaching its network's resources. Sibling routers on the network are
+// independent and are not folded.
 func (r *resolver) collectFromNetworkRouters() {
 	for _, router := range r.networkRouters() {
-		matchedByGroup := anyInSet(router.PeerGroups, r.changedGroupSet)
-		matchedByPeer := router.Peer != "" && len(r.changedPeerSet) > 0 && isInSet(router.Peer, r.changedPeerSet)
+		matchedByGroup := anyInSet(router.PeerGroups, r.linkGroups)
+		matchedByPeer := router.Peer != "" && len(r.changedPeers) > 0 && isInSet(router.Peer, r.changedPeers)
 		if !matchedByGroup && !matchedByPeer {
 			continue
 		}
-		log.WithContext(r.ctx).Tracef("collectFromNetworkRouters: router %s on network %s matched (byGroup=%t byPeer=%t) -> folding peerGroups=%v peer=%q and marking network for source bridge",
+		log.WithContext(r.ctx).Tracef("collectFromNetworkRouters: router %s on network %s matched (byGroup=%t byPeer=%t) -> folding its peerGroups=%v peer=%q (own groups on outputGroups) + sources reaching network resources",
 			router.ID, router.NetworkID, matchedByGroup, matchedByPeer, router.PeerGroups, router.Peer)
-		addAll(r.groupSet, router.PeerGroups)
+		// The backing PeerGroups are the matched (own) side: fold them only on a
+		// whole-group change so a one-peer move does not wake sibling backing peers. The
+		// opposite side (policy sources reaching the network) is folded below.
+		r.foldOutputGroups(router.PeerGroups)
 		if router.Peer != "" {
-			r.peerSet[router.Peer] = struct{}{}
+			r.affectedPeers[router.Peer] = struct{}{}
+		}
+		if router.NetworkID != "" {
+			r.foldPolicySourcesForResources(r.networkResourceIDs(router.NetworkID))
 		}
-		r.networkIDs[router.NetworkID] = struct{}{}
 	}
 }
 
@@ -526,42 +827,48 @@ func (r *resolver) collectFromProxyServices() {
 	expanded := r.expandChangedPeersWithGroups()
 
 	for _, svc := range services {
-		if svc == nil {
-			continue
+		if svc == nil || !svc.Enabled {
+			continue // a disabled service proxies nothing; skip existing account data
 		}
 		proxyPeers := proxyByCluster[svc.ProxyCluster]
 		if len(proxyPeers) == 0 {
 			continue
 		}
 		matchedByPeer := serviceMatchesChangedPeers(svc, proxyPeers, expanded)
-		matchedByAccessGroup := anyInSet(svc.AccessGroups, r.changedGroupSet)
+		matchedByAccessGroup := anyInSet(svc.AccessGroups, r.linkGroups)
 		if !matchedByPeer && !matchedByAccessGroup {
 			continue
 		}
-		log.WithContext(r.ctx).Tracef("collectFromProxyServices: service %s (cluster=%s) matched (byProxyOrTargetPeer=%t byAccessGroup=%t) -> folding %d proxy peers, peer targets and access groups %v",
+		log.WithContext(r.ctx).Tracef("collectFromProxyServices: service %s (cluster=%s) matched (byProxyOrTargetPeer=%t byAccessGroup=%t) -> folding %d proxy peers, peer targets; access groups %v on outputGroups only",
 			svc.ID, svc.ProxyCluster, matchedByPeer, matchedByAccessGroup, len(proxyPeers), svc.AccessGroups)
 		for _, pid := range proxyPeers {
-			r.peerSet[pid] = struct{}{}
+			r.affectedPeers[pid] = struct{}{}
 		}
 		for _, target := range svc.Targets {
+			if !target.Enabled {
+				continue // a disabled target forwards nothing
+			}
 			if target.TargetType == rpservice.TargetTypePeer && target.TargetId != "" {
-				r.peerSet[target.TargetId] = struct{}{}
+				r.affectedPeers[target.TargetId] = struct{}{}
 			}
 		}
-		addAll(r.groupSet, svc.AccessGroups)
+		// AccessGroups are the matched (own) side with no opposite to fold: a member's
+		// proxy access is self-contained, so a one-peer move refreshes that peer alone.
+		// Fold the groups only on a whole-group change.
+		r.foldOutputGroups(svc.AccessGroups)
 	}
 }
 
 func (r *resolver) expandChangedPeersWithGroups() map[string]struct{} {
-	if len(r.changedGroupSet) == 0 {
-		return r.changedPeerSet
+	if len(r.linkGroups) == 0 {
+		return r.changedPeers
 	}
-	ids := r.peerIDsForGroups(r.changedGroupSet)
+	ids := r.peerIDsForGroups(r.linkGroups)
 	if len(ids) == 0 {
-		return r.changedPeerSet
+		return r.changedPeers
 	}
-	merged := make(map[string]struct{}, len(r.changedPeerSet)+len(ids))
-	for id := range r.changedPeerSet {
+	merged := make(map[string]struct{}, len(r.changedPeers)+len(ids))
+	for id := range r.changedPeers {
 		merged[id] = struct{}{}
 	}
 	for _, id := range ids {
@@ -570,54 +877,36 @@ func (r *resolver) expandChangedPeersWithGroups() map[string]struct{} {
 	return merged
 }
 
-// collectResourceRouterBridge crosses between source peers and routing peers, which
-// are reachable only via resource -> network -> router, not through the policy's own
-// groups: source -> router (targeted resources' networks), then router -> source.
-func (r *resolver) collectResourceRouterBridge() {
-	r.bridgeSourceToRouters()
-	r.bridgeRoutersToSources()
-}
-
-func (r *resolver) bridgeSourceToRouters() {
-	resourceIDs := r.policyDestinationResourceIDs(r.matchedPolicies...)
+// foldRoutersForResources folds the routers serving the networks of the given
+// resources (a destination resource is reached through its network's routers). It is
+// the resource -> network -> router hop used by foldPolicySide for a destination.
+func (r *resolver) foldRoutersForResources(resourceIDs map[string]struct{}) {
 	if len(resourceIDs) == 0 {
 		return
 	}
-
-	networkIDs := r.resourceNetworkIDs(resourceIDs)
-	log.WithContext(r.ctx).Tracef("bridgeSourceToRouters: targeted resources %v -> networks %v (their routers become affected via the router->source pass)",
-		setToSlice(resourceIDs), setToSlice(networkIDs))
-	for id := range networkIDs {
-		r.networkIDs[id] = struct{}{}
-	}
+	r.foldRoutersOnNetworks(r.resourceNetworkIDs(resourceIDs))
 }
 
-func (r *resolver) bridgeRoutersToSources() {
-	if len(r.networkIDs) == 0 {
-		return
+// ruleDestinationResourceIDs returns the destination resource IDs of a single rule:
+// the direct DestinationResource plus the resources of its destination groups.
+func (r *resolver) ruleDestinationResourceIDs(rule *types.PolicyRule) map[string]struct{} {
+	resourceIDs := make(map[string]struct{})
+	if rule.DestinationResource.Type != types.ResourceTypePeer && rule.DestinationResource.ID != "" {
+		resourceIDs[rule.DestinationResource.ID] = struct{}{}
 	}
+	r.addGroupResourceIDs(toSet(rule.Destinations), resourceIDs)
+	return resourceIDs
+}
 
-	log.WithContext(r.ctx).Tracef("bridgeRoutersToSources: affected networks %v -> folding their routing peers and the source peers of policies targeting their resources",
-		setToSlice(r.networkIDs))
-
-	r.foldRoutersOnNetworks(r.networkIDs)
-
+// networkResourceIDs returns the IDs of all resources on the given network.
+func (r *resolver) networkResourceIDs(networkID string) map[string]struct{} {
 	resourceIDs := make(map[string]struct{})
 	for _, resource := range r.networkResources() {
-		if _, ok := r.networkIDs[resource.NetworkID]; ok {
+		if resource.NetworkID == networkID {
 			resourceIDs[resource.ID] = struct{}{}
 		}
 	}
-	if len(resourceIDs) == 0 {
-		return
-	}
-
-	for _, policy := range r.policies() {
-		if r.policyTargetsResources(policy, resourceIDs) {
-			log.WithContext(r.ctx).Tracef("bridgeRoutersToSources: policy %s (%s) targets an affected-network resource -> folding its source groups/peers", policy.ID, policy.Name)
-			collectPolicySources(policy, r.groupSet, r.peerSet)
-		}
-	}
+	return resourceIDs
 }
 
 func (r *resolver) foldRoutersOnNetworks(networkIDs map[string]struct{}) {
@@ -627,9 +916,9 @@ func (r *resolver) foldRoutersOnNetworks(networkIDs map[string]struct{}) {
 		}
 		log.WithContext(r.ctx).Tracef("bridgeRoutersToSources: router %s serves affected network %s -> folding peerGroups=%v peer=%q",
 			router.ID, router.NetworkID, router.PeerGroups, router.Peer)
-		addAll(r.groupSet, router.PeerGroups)
+		addAll(r.affectedGroups, router.PeerGroups)
 		if router.Peer != "" {
-			r.peerSet[router.Peer] = struct{}{}
+			r.affectedPeers[router.Peer] = struct{}{}
 		}
 	}
 }
@@ -650,6 +939,9 @@ func (r *resolver) policyTargetsResources(policy *types.Policy, resourceIDs map[
 	}
 	destGroupSet := make(map[string]struct{})
 	for _, rule := range policy.Rules {
+		if !rule.Enabled {
+			continue
+		}
 		if rule.DestinationResource.Type != types.ResourceTypePeer && isInSet(rule.DestinationResource.ID, resourceIDs) {
 			return true
 		}
@@ -714,44 +1006,20 @@ func (r *resolver) addGroupResourceIDs(groupIDs map[string]struct{}, resourceIDs
 	}
 }
 
-func collectPolicyDirectPeers(policy *types.Policy, peerSet map[string]struct{}) {
+// collectPolicySources folds the source groups/peers of a snapshot policy's enabled
+// rules (a disabled rule grants no access).
+func collectPolicySources(policy *types.Policy, groups, peers map[string]struct{}) {
 	for _, rule := range policy.Rules {
+		if !rule.Enabled {
+			continue
+		}
+		addAll(groups, rule.Sources)
 		if rule.SourceResource.Type == types.ResourceTypePeer && rule.SourceResource.ID != "" {
-			peerSet[rule.SourceResource.ID] = struct{}{}
-		}
-		if rule.DestinationResource.Type == types.ResourceTypePeer && rule.DestinationResource.ID != "" {
-			peerSet[rule.DestinationResource.ID] = struct{}{}
+			peers[rule.SourceResource.ID] = struct{}{}
 		}
 	}
 }
 
-func collectPolicySources(policy *types.Policy, groupSet, peerSet map[string]struct{}) {
-	for _, rule := range policy.Rules {
-		addAll(groupSet, rule.Sources)
-		if rule.SourceResource.Type == types.ResourceTypePeer && rule.SourceResource.ID != "" {
-			peerSet[rule.SourceResource.ID] = struct{}{}
-		}
-	}
-}
-
-func policyReferencesGroups(policy *types.Policy, groupSet map[string]struct{}) bool {
-	for _, rule := range policy.Rules {
-		if anyInSet(rule.Sources, groupSet) || anyInSet(rule.Destinations, groupSet) {
-			return true
-		}
-	}
-	return false
-}
-
-func policyReferencesDirectPeers(policy *types.Policy, changedSet map[string]struct{}) bool {
-	for _, rule := range policy.Rules {
-		if isDirectPeerInSet(rule.SourceResource, changedSet) || isDirectPeerInSet(rule.DestinationResource, changedSet) {
-			return true
-		}
-	}
-	return false
-}
-
 func policyReferencesPostureChecks(policy *types.Policy, ids map[string]struct{}) bool {
 	for _, id := range policy.SourcePostureChecks {
 		if _, ok := ids[id]; ok {
@@ -776,7 +1044,7 @@ func serviceMatchesChangedPeers(svc *rpservice.Service, proxyPeers []string, cha
 		}
 	}
 	for _, target := range svc.Targets {
-		if target.TargetType != rpservice.TargetTypePeer || target.TargetId == "" {
+		if !target.Enabled || target.TargetType != rpservice.TargetTypePeer || target.TargetId == "" {
 			continue
 		}
 		if _, ok := changedPeers[target.TargetId]; ok {

management/server/affectedpeers/resolver_test.go

@@ -10,8 +10,8 @@ import (
 	"github.com/netbirdio/netbird/management/server/types"
 )
 
-// policyGroupsAndPeers mirrors the explicit-policy extraction (RuleGroups +
-// direct peers) the resolver folds in, for asserting the pure logic.
+// policyGroupsAndPeers mirrors the both-sides extraction (RuleGroups + direct peers)
+// the resolver folds in for a changed policy, for asserting the pure logic.
 func policyGroupsAndPeers(policies ...*types.Policy) (groups []string, peers []string) {
 	peerSet := map[string]struct{}{}
 	for _, p := range policies {
@@ -19,7 +19,14 @@ func policyGroupsAndPeers(policies ...*types.Policy) (groups []string, peers []s
 			continue
 		}
 		groups = append(groups, p.RuleGroups()...)
-		collectPolicyDirectPeers(p, peerSet)
+		for _, rule := range p.Rules {
+			if rule.SourceResource.Type == types.ResourceTypePeer && rule.SourceResource.ID != "" {
+				peerSet[rule.SourceResource.ID] = struct{}{}
+			}
+			if rule.DestinationResource.Type == types.ResourceTypePeer && rule.DestinationResource.ID != "" {
+				peerSet[rule.DestinationResource.ID] = struct{}{}
+			}
+		}
 	}
 	for id := range peerSet {
 		peers = append(peers, id)
@@ -80,26 +87,6 @@ func TestChangeIsEmpty(t *testing.T) {
 	assert.False(t, Change{PostureCheckIDs: []string{"pc"}}.isEmpty())
 }
 
-func TestPolicyReferencesGroups(t *testing.T) {
-	policy := &types.Policy{Rules: []*types.PolicyRule{{Sources: []string{"g1", "g2"}, Destinations: []string{"g3"}}}}
-
-	assert.True(t, policyReferencesGroups(policy, map[string]struct{}{"g1": {}}))
-	assert.True(t, policyReferencesGroups(policy, map[string]struct{}{"g3": {}}))
-	assert.False(t, policyReferencesGroups(policy, map[string]struct{}{"g4": {}}))
-	assert.False(t, policyReferencesGroups(policy, map[string]struct{}{}))
-}
-
-func TestPolicyReferencesDirectPeers(t *testing.T) {
-	policy := &types.Policy{Rules: []*types.PolicyRule{{
-		SourceResource:      types.Resource{Type: types.ResourceTypePeer, ID: "p1"},
-		DestinationResource: types.Resource{Type: types.ResourceTypeHost, ID: "r1"},
-	}}}
-
-	assert.True(t, policyReferencesDirectPeers(policy, map[string]struct{}{"p1": {}}))
-	assert.False(t, policyReferencesDirectPeers(policy, map[string]struct{}{"r1": {}}))
-	assert.False(t, policyReferencesDirectPeers(policy, map[string]struct{}{"p2": {}}))
-}
-
 func TestPolicyReferencesPostureChecks(t *testing.T) {
 	policy := &types.Policy{SourcePostureChecks: []string{"pc1", "pc2"}}
 
@@ -107,24 +94,9 @@ func TestPolicyReferencesPostureChecks(t *testing.T) {
 	assert.False(t, policyReferencesPostureChecks(policy, map[string]struct{}{"pc3": {}}))
 }
 
-func TestCollectPolicyDirectPeers(t *testing.T) {
-	policy := &types.Policy{Rules: []*types.PolicyRule{{
-		SourceResource:      types.Resource{Type: types.ResourceTypePeer, ID: "p1"},
-		DestinationResource: types.Resource{Type: types.ResourceTypePeer, ID: "p2"},
-	}, {
-		DestinationResource: types.Resource{Type: types.ResourceTypeHost, ID: "r1"},
-	}}}
-
-	peerSet := map[string]struct{}{}
-	collectPolicyDirectPeers(policy, peerSet)
-
-	assert.Contains(t, peerSet, "p1")
-	assert.Contains(t, peerSet, "p2")
-	assert.NotContains(t, peerSet, "r1")
-}
-
 func TestCollectPolicySources(t *testing.T) {
 	policy := &types.Policy{Rules: []*types.PolicyRule{{
+		Enabled:        true,
 		Sources:        []string{"g1"},
 		SourceResource: types.Resource{Type: types.ResourceTypePeer, ID: "p1"},
 		Destinations:   []string{"g2"},

management/server/group.go

@@ -520,7 +520,12 @@ func collectDeletableGroups(ctx context.Context, transaction store.Store, accoun
 // GroupAddPeer appends peer to the group
 func (am *DefaultAccountManager) GroupAddPeer(ctx context.Context, accountID, groupID, peerID string) error {
 	var snap *affectedpeers.Snapshot
-	change := affectedpeers.Change{ChangedGroupIDs: []string{groupID}}
+	// A membership change affects only the peer itself and the opposite side of THIS
+	// group's policies — not the group's other members, and not the peer's other
+	// groups. LinkGroups walks only this group (matched, not expanded); OutputPeerIDs
+	// refreshes the peer without seeding its other group memberships. For an
+	// intra-group policy the opposite side is the group, so its members still refresh.
+	change := affectedpeers.Change{OutputPeerIDs: []string{peerID}, LinkGroups: []string{groupID}}
 
 	err := am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
 		if err := transaction.AddPeerToGroup(ctx, accountID, peerID, groupID); err != nil {
@@ -586,10 +591,11 @@ func (am *DefaultAccountManager) GroupAddResource(ctx context.Context, accountID
 // GroupDeletePeer removes peer from the group
 func (am *DefaultAccountManager) GroupDeletePeer(ctx context.Context, accountID, groupID, peerID string) error {
 	var snap *affectedpeers.Snapshot
-	change := affectedpeers.Change{
-		ChangedGroupIDs:     []string{groupID},
-		RemovedPeersByGroup: map[string][]string{groupID: {peerID}},
-	}
+	// Same as GroupAddPeer: the removed peer and the opposite side of THIS group's
+	// policies refresh, not the group's other members or the peer's other groups. The
+	// peer is no longer in the group's index, but LinkGroups still drives the
+	// opposite-side walk, and OutputPeerIDs refreshes the removed peer itself.
+	change := affectedpeers.Change{OutputPeerIDs: []string{peerID}, LinkGroups: []string{groupID}}
 
 	err := am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
 		if err := transaction.RemovePeerFromGroup(ctx, peerID, groupID); err != nil {
@@ -600,8 +606,6 @@ func (am *DefaultAccountManager) GroupDeletePeer(ctx context.Context, accountID,
 			return err
 		}
 
-		// The removed peer is carried in change.RemovedPeersByGroup and folded in
-		// only when the group is linked, so loading post-removal is correct.
 		var err error
 		if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil {
 			return err

management/server/http/handlers/users/users_handler.go

@@ -220,7 +220,7 @@ func (h *handler) getAllUsers(w http.ResponseWriter, r *http.Request) {
 		}
 
 		includeServiceUser, err := strconv.ParseBool(serviceUser)
-		log.WithContext(r.Context()).Debugf("Should include service user: %v", includeServiceUser)
+		log.WithContext(r.Context()).Tracef("Should include service user: %v", includeServiceUser)
 		if err != nil {
 			util.WriteError(r.Context(), status.Errorf(status.InvalidArgument, "invalid service_user query parameter"), w)
 			return

management/server/peer.go

@@ -209,14 +209,14 @@ func (am *DefaultAccountManager) resolvePeerLocation(ctx context.Context, peer *
 	if am.geo == nil || realIP == nil {
 		return nil
 	}
-	if peer.Location.ConnectionIP != nil && peer.Location.ConnectionIP.Equal(realIP) {
-		return nil
-	}
 	location, err := am.geo.Lookup(realIP)
 	if err != nil {
 		log.WithContext(ctx).Warnf("failed to get location for peer %s realip: [%s]: %v", peer.ID, realIP.String(), err)
 		return nil
 	}
+	if peer.Location.ConnectionIP != nil && peer.Location.ConnectionIP.Equal(realIP) && peer.Location.GeoNameID == location.City.GeonameID {
+		return nil
+	}
 	return &nbpeer.Location{
 		ConnectionIP: realIP,
 		CountryCode:  location.Country.ISOCode,
@@ -730,7 +730,7 @@ func (am *DefaultAccountManager) handleSetupKeyAddedPeer(ctx context.Context, en
 func (am *DefaultAccountManager) AddPeer(ctx context.Context, accountID, setupKey, userID string, peer *nbpeer.Peer, temporary bool) (*nbpeer.Peer, *types.Network, []*posture.Checks, bool, error) {
 	if setupKey == "" && userID == "" && !peer.ProxyMeta.Embedded {
 		// no auth method provided => reject access
-		return nil, nil, nil, false, status.Errorf(status.Unauthenticated, "no peer auth method provided, please use a setup key or interactive SSO login")
+		return nil, nil, nil, false, status.ErrNoAuthMethodProvided
 	}
 
 	upperKey := strings.ToUpper(setupKey)
@@ -1051,8 +1051,8 @@ func (am *DefaultAccountManager) SyncPeer(ctx context.Context, sync types.PeerSy
 		return nil, nil, nil, 0, err
 	}
 
-	metaDiffAffectsPosture := posture.AffectsPosture(&metaDiff, resPostureChecks)
-	if isStatusChanged || sync.UpdateAccountPeers || ipv6CapabilityChanged || metaDiffAffectsPosture || metaDiff.VersionChanged || metaDiff.Hostname {
+	metaDiffAffectsPosture := posture.AffectsPosture(ctx, &metaDiff, resPostureChecks)
+	if requiresPeerUpdate(ctx, isStatusChanged, sync.UpdateAccountPeers, ipv6CapabilityChanged, metaDiffAffectsPosture, metaDiff.VersionChanged(), metaDiff.HostnameChanged()) {
 		changedPeerIDs := []string{peer.ID}
 		affectedPeerIDs := am.syncPeerAffectedPeers(ctx, accountID, peer.ID, nmap, peerNotValid, metaDiffAffectsPosture)
 		if err = am.networkMapController.OnPeersUpdated(ctx, accountID, changedPeerIDs, affectedPeerIDs); err != nil {
@@ -1063,6 +1063,29 @@ func (am *DefaultAccountManager) SyncPeer(ctx context.Context, sync types.PeerSy
 	return peer, nmap, resPostureChecks, dnsFwdPort, nil
 }
 
+func requiresPeerUpdate(ctx context.Context, isStatusChanged, updateAccountPeers, ipv6CapabilityChanged, metaDiffAffectsPosture, versionChanged, hostname bool) bool {
+	var reason string
+	switch {
+	case isStatusChanged:
+		reason = "status changed"
+	case updateAccountPeers:
+		reason = "update account peers"
+	case ipv6CapabilityChanged:
+		reason = "ipv6 capability changed"
+	case metaDiffAffectsPosture:
+		reason = "meta diff affects posture"
+	case versionChanged:
+		reason = "version changed"
+	case hostname:
+		reason = "hostname changed"
+	default:
+		return false
+	}
+
+	log.WithContext(ctx).Tracef("peer update required: %s", reason)
+	return true
+}
+
 // syncPeerAffectedPeers resolves the peers affected by a SyncPeer change. The
 // peer's own validated network map is bidirectional for policy and routing
 // reachability, so when the peer stays valid and no source-posture gate is in

management/server/peer/peer.go

@@ -107,6 +107,15 @@ type Location struct {
 	GeoNameID    uint // city level geoname id
 }
 
+// equal reports whether two locations match. ConnectionIP is a net.IP slice, so it uses
+// IP.Equal, not ==.
+func (l Location) equal(other Location) bool {
+	return l.CountryCode == other.CountryCode &&
+		l.CityName == other.CityName &&
+		l.GeoNameID == other.GeoNameID &&
+		l.ConnectionIP.Equal(other.ConnectionIP)
+}
+
 // NetworkAddress is the IP address with network and MAC address of a network interface
 type NetworkAddress struct {
 	NetIP netip.Prefix `gorm:"serializer:json"`
@@ -267,185 +276,141 @@ func (p *Peer) UpdateMetaIfNew(ctx context.Context, meta PeerSystemMeta, newLoca
 		return MetaDiff{}
 	}
 
-	versionChanged := p.Meta.WtVersion != meta.WtVersion
-
 	// Avoid overwriting UIVersion if the update was triggered sole by the CLI client
 	if meta.UIVersion == "" {
 		meta.UIVersion = p.Meta.UIVersion
 	}
 
-	oldVersion := p.Meta.WtVersion
+	effectiveLocation := p.Location
+	if newLocation != nil {
+		effectiveLocation = *newLocation
+	}
 
-	diff := diffMeta(p.Meta, meta)
-	if diff.Any() {
+	diff := diffMeta(p.Meta, meta, p.Location, effectiveLocation)
+	if diff.Updated() {
 		p.Meta = meta
 	}
-	diff.VersionChanged = versionChanged
+	p.Location = effectiveLocation
 
-	locationInfo := ""
-	if newLocation != nil {
-		p.Location = *newLocation
-		diff.LocationChanged = true
-		locationInfo = fmt.Sprintf("location changed to %s, ", newLocation.ConnectionIP)
-	}
-
-	versionInfo := ""
-	if diff.VersionChanged {
-		versionInfo = fmt.Sprintf("version changed: %s -> %s, ", oldVersion, meta.WtVersion)
-	}
-
-	if diff.Any() || diff.VersionChanged || diff.LocationChanged {
-		log.WithContext(ctx).
-			Debugf("peer meta updated, %s%s%d field(s) changed: %s", versionInfo, locationInfo, len(diff.Changed), strings.Join(diff.Changed, ", "))
+	if diff.Updated() {
+		log.WithContext(ctx).Debug(diff.LogSummary())
 	}
 
 	return diff
 }
 
-// MetaDiff records which PeerSystemMeta fields differ between two metas. Each bool
-// maps to a single struct field, except Environment, which is split into Cloud and
-// Platform. Changed holds the human-readable `field: <old> -> <new>` entries so the
-// existing log line and isEqual can be derived from the same comparison.
-//
-// VersionChanged and LocationChanged sit outside the per-meta-field set:
-// VersionChanged tracks the WireGuard client version specifically (compared before
-// the UIVersion fixup, to signal client upgrades) and LocationChanged tracks the
-// peer's connection geo location, which lives on Peer rather than PeerSystemMeta.
-// Neither contributes an entry to Changed, so the field-coverage accounting stays
-// driven purely by the PeerSystemMeta comparison.
+// MetaDiff holds a peer's full before/after state across a sync: both metas and both
+// connection locations (the location lives on Peer, not PeerSystemMeta, but posture
+// checks read it). Changed lists what moved, for logging and the persistence decision;
+// the snapshots let a posture check be replayed against old and new. Everything is derived
+// from these fields, so there are no parallel per-field flags to keep in sync.
 type MetaDiff struct {
-	Hostname            bool
-	GoOS                bool
-	Kernel              bool
-	KernelVersion       bool
-	Core                bool
-	Platform            bool
-	OS                  bool
-	OSVersion           bool
-	WtVersion           bool
-	UIVersion           bool
-	SystemSerialNumber  bool
-	SystemProductName   bool
-	SystemManufacturer  bool
-	EnvironmentCloud    bool
-	EnvironmentPlatform bool
-	Flags               bool
-	Capabilities        bool
-	NetworkAddresses    bool
-	Files               bool
-
-	VersionChanged  bool
-	LocationChanged bool
+	OldMeta     PeerSystemMeta
+	NewMeta     PeerSystemMeta
+	OldLocation Location
+	NewLocation Location
 
 	Changed []string
 }
 
-// Any reports whether any PeerSystemMeta field changed.
-func (d MetaDiff) Any() bool {
+// Updated reports whether anything changed and the peer must be persisted. diffMeta fills
+// Changed in the pass that builds the diff, so this is a length check, not a re-comparison.
+// Pointer receiver: MetaDiff embeds two metas, so copying it per call is wasteful.
+func (d *MetaDiff) Updated() bool {
 	return len(d.Changed) != 0
 }
 
-// Updated reports whether the peer needs to be persisted: any meta field changed
-// or the geo location changed. The version flag alone does not imply a write,
-// since a version change is also reflected in the WtVersion meta field.
-func (d MetaDiff) Updated() bool {
-	return d.Any() || d.LocationChanged || d.VersionChanged
+// VersionChanged reports whether the WireGuard client version changed (a client upgrade).
+func (d *MetaDiff) VersionChanged() bool {
+	return d.OldMeta.WtVersion != d.NewMeta.WtVersion
+}
+
+// HostnameChanged reports whether the peer's hostname changed.
+func (d *MetaDiff) HostnameChanged() bool {
+	return d.OldMeta.Hostname != d.NewMeta.Hostname
+}
+
+// LogSummary renders the changed fields as a single human-readable line.
+func (d *MetaDiff) LogSummary() string {
+	return fmt.Sprintf("peer meta updated, %d field(s) changed: %s",
+		len(d.Changed), strings.Join(d.Changed, ", "))
 }
 
 func metaDiff(oldMeta, newMeta PeerSystemMeta) []string {
-	return diffMeta(oldMeta, newMeta).Changed
+	return diffMeta(oldMeta, newMeta, Location{}, Location{}).Changed
 }
 
-// diffMeta compares two metas field by field, returning both a per-field flag set
-// (for callers that need to know exactly what changed, e.g. matching against
-// posture checks) and the human-readable Changed list. It is the single source of
-// truth for meta comparison: isEqual reports equality as an empty diff, so the log
-// line, the change decision, and the flags can never disagree.
-func diffMeta(oldMeta, newMeta PeerSystemMeta) MetaDiff {
-	var d MetaDiff
+// diffMeta snapshots a peer's old and new state and records a Changed entry per field that
+// moved. It is the single source of truth for the comparison: isEqual is an empty Changed
+// list, so the log line and the persistence decision can never disagree.
+func diffMeta(oldMeta, newMeta PeerSystemMeta, oldLocation, newLocation Location) MetaDiff {
+	d := MetaDiff{OldMeta: oldMeta, NewMeta: newMeta, OldLocation: oldLocation, NewLocation: newLocation}
 	add := func(field string, oldVal, newVal any) {
 		d.Changed = append(d.Changed, fmt.Sprintf("%s: %v -> %v", field, oldVal, newVal))
 	}
 
 	if oldMeta.Hostname != newMeta.Hostname {
-		d.Hostname = true
 		add("hostname", oldMeta.Hostname, newMeta.Hostname)
 	}
 	if oldMeta.GoOS != newMeta.GoOS {
-		d.GoOS = true
 		add("goos", oldMeta.GoOS, newMeta.GoOS)
 	}
 	if oldMeta.Kernel != newMeta.Kernel {
-		d.Kernel = true
 		add("kernel", oldMeta.Kernel, newMeta.Kernel)
 	}
 	if oldMeta.KernelVersion != newMeta.KernelVersion {
-		d.KernelVersion = true
 		add("kernel_version", oldMeta.KernelVersion, newMeta.KernelVersion)
 	}
 	if oldMeta.Core != newMeta.Core {
-		d.Core = true
 		add("core", oldMeta.Core, newMeta.Core)
 	}
 	if oldMeta.Platform != newMeta.Platform {
-		d.Platform = true
 		add("platform", oldMeta.Platform, newMeta.Platform)
 	}
 	if oldMeta.OS != newMeta.OS {
-		d.OS = true
 		add("os", oldMeta.OS, newMeta.OS)
 	}
 	if oldMeta.OSVersion != newMeta.OSVersion {
-		d.OSVersion = true
 		add("os_version", oldMeta.OSVersion, newMeta.OSVersion)
 	}
 	if oldMeta.WtVersion != newMeta.WtVersion {
-		d.WtVersion = true
 		add("wt_version", oldMeta.WtVersion, newMeta.WtVersion)
 	}
 	if oldMeta.UIVersion != newMeta.UIVersion {
-		d.UIVersion = true
 		add("ui_version", oldMeta.UIVersion, newMeta.UIVersion)
 	}
 	if oldMeta.SystemSerialNumber != newMeta.SystemSerialNumber {
-		d.SystemSerialNumber = true
 		add("system_serial_number", oldMeta.SystemSerialNumber, newMeta.SystemSerialNumber)
 	}
 	if oldMeta.SystemProductName != newMeta.SystemProductName {
-		d.SystemProductName = true
 		add("system_product_name", oldMeta.SystemProductName, newMeta.SystemProductName)
 	}
 	if oldMeta.SystemManufacturer != newMeta.SystemManufacturer {
-		d.SystemManufacturer = true
 		add("system_manufacturer", oldMeta.SystemManufacturer, newMeta.SystemManufacturer)
 	}
 	if oldMeta.Environment.Cloud != newMeta.Environment.Cloud {
-		d.EnvironmentCloud = true
 		add("environment_cloud", oldMeta.Environment.Cloud, newMeta.Environment.Cloud)
 	}
 	if oldMeta.Environment.Platform != newMeta.Environment.Platform {
-		d.EnvironmentPlatform = true
 		add("environment_platform", oldMeta.Environment.Platform, newMeta.Environment.Platform)
 	}
 	if !oldMeta.Flags.isEqual(newMeta.Flags) {
-		d.Flags = true
 		add("flags", fmt.Sprintf("%+v", oldMeta.Flags), fmt.Sprintf("%+v", newMeta.Flags))
 	}
 	if !capabilitiesEqual(oldMeta.Capabilities, newMeta.Capabilities) {
-		d.Capabilities = true
 		add("capabilities", oldMeta.Capabilities, newMeta.Capabilities)
 	}
-
 	if !sameMultiset(oldMeta.NetworkAddresses, newMeta.NetworkAddresses) {
-		d.NetworkAddresses = true
 		add("network_addresses", fmt.Sprintf("%v", oldMeta.NetworkAddresses), fmt.Sprintf("%v", newMeta.NetworkAddresses))
 	}
-
 	if !sameMultiset(oldMeta.Files, newMeta.Files) {
-		d.Files = true
 		add("files", fmt.Sprintf("%v", oldMeta.Files), fmt.Sprintf("%v", newMeta.Files))
 	}
 
+	if !oldLocation.equal(newLocation) {
+		add("connection_ip", oldLocation.ConnectionIP, newLocation.ConnectionIP)
+	}
+
 	return d
 }
 

management/server/peer_test.go

@@ -49,6 +49,7 @@ import (
 
 	nbdns "github.com/netbirdio/netbird/dns"
 	"github.com/netbirdio/netbird/management/server/activity"
+	"github.com/netbirdio/netbird/management/server/geolocation"
 	nbpeer "github.com/netbirdio/netbird/management/server/peer"
 	"github.com/netbirdio/netbird/management/server/posture"
 	"github.com/netbirdio/netbird/management/server/store"
@@ -2893,3 +2894,141 @@ func TestUpdatePeer_DnsLabelUniqueName(t *testing.T) {
 	require.NoError(t, err, "renaming to unique FQDN should succeed")
 	assert.Equal(t, "api-server", updated.DNSLabel, "DNS label should be first label of FQDN")
 }
+
+// fakeGeo is a configurable geolocation.Geolocation implementation for tests. It
+// returns a record built from the configured city geoname id, or an error when set.
+type fakeGeo struct {
+	geoNameID uint
+	isoCode   string
+	cityName  string
+	err       error
+}
+
+func (g *fakeGeo) Lookup(net.IP) (*geolocation.Record, error) {
+	if g.err != nil {
+		return nil, g.err
+	}
+	record := &geolocation.Record{}
+	record.City.GeonameID = g.geoNameID
+	record.City.Names.En = g.cityName
+	record.Country.ISOCode = g.isoCode
+	return record, nil
+}
+
+func (g *fakeGeo) GetAllCountries() ([]geolocation.Country, error) { return nil, nil }
+
+func (g *fakeGeo) GetCitiesByCountry(string) ([]geolocation.City, error) { return nil, nil }
+
+func (g *fakeGeo) Stop() error { return nil }
+
+func TestResolvePeerLocation(t *testing.T) {
+	realIP := net.ParseIP("203.0.113.10")
+
+	tests := []struct {
+		name    string
+		geo     geolocation.Geolocation
+		peer    *nbpeer.Peer
+		realIP  net.IP
+		want    *nbpeer.Location
+		wantNil bool
+	}{
+		{
+			name:    "no geo configured returns nil",
+			geo:     nil,
+			peer:    &nbpeer.Peer{ID: "p1"},
+			realIP:  realIP,
+			wantNil: true,
+		},
+		{
+			name:    "nil real IP returns nil",
+			geo:     &fakeGeo{geoNameID: 100},
+			peer:    &nbpeer.Peer{ID: "p1"},
+			realIP:  nil,
+			wantNil: true,
+		},
+		{
+			name:    "lookup error returns nil",
+			geo:     &fakeGeo{err: fmt.Errorf("lookup boom")},
+			peer:    &nbpeer.Peer{ID: "p1"},
+			realIP:  realIP,
+			wantNil: true,
+		},
+		{
+			name: "same IP and same geoname returns nil",
+			geo:  &fakeGeo{geoNameID: 100, isoCode: "US", cityName: "City A"},
+			peer: &nbpeer.Peer{
+				ID: "p1",
+				Location: nbpeer.Location{
+					ConnectionIP: realIP,
+					GeoNameID:    100,
+				},
+			},
+			realIP:  realIP,
+			wantNil: true,
+		},
+		{
+			name: "same IP but changed geoname returns location",
+			geo:  &fakeGeo{geoNameID: 200, isoCode: "US", cityName: "City B"},
+			peer: &nbpeer.Peer{
+				ID: "p1",
+				Location: nbpeer.Location{
+					ConnectionIP: realIP,
+					GeoNameID:    100,
+				},
+			},
+			realIP: realIP,
+			want: &nbpeer.Location{
+				ConnectionIP: realIP,
+				CountryCode:  "US",
+				CityName:     "City B",
+				GeoNameID:    200,
+			},
+		},
+		{
+			name: "different IP returns location",
+			geo:  &fakeGeo{geoNameID: 100, isoCode: "US", cityName: "City A"},
+			peer: &nbpeer.Peer{
+				ID: "p1",
+				Location: nbpeer.Location{
+					ConnectionIP: net.ParseIP("198.51.100.7"),
+					GeoNameID:    100,
+				},
+			},
+			realIP: realIP,
+			want: &nbpeer.Location{
+				ConnectionIP: realIP,
+				CountryCode:  "US",
+				CityName:     "City A",
+				GeoNameID:    100,
+			},
+		},
+		{
+			name:   "no prior location returns location",
+			geo:    &fakeGeo{geoNameID: 100, isoCode: "US", cityName: "City A"},
+			peer:   &nbpeer.Peer{ID: "p1"},
+			realIP: realIP,
+			want: &nbpeer.Location{
+				ConnectionIP: realIP,
+				CountryCode:  "US",
+				CityName:     "City A",
+				GeoNameID:    100,
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			am := &DefaultAccountManager{geo: tt.geo}
+			got := am.resolvePeerLocation(context.Background(), tt.peer, tt.realIP)
+			if tt.wantNil {
+				assert.Nil(t, got, "resolved location should be nil")
+				return
+			}
+			require.NotNil(t, got, "resolved location should not be nil")
+			assert.True(t, tt.want.ConnectionIP.Equal(got.ConnectionIP), "connection IP should match")
+			assert.Equal(t, tt.want.CountryCode, got.CountryCode, "country code should match")
+			assert.Equal(t, tt.want.CityName, got.CityName, "city name should match")
+			assert.Equal(t, tt.want.GeoNameID, got.GeoNameID, "geoname id should match")
+		})
+	}
+}

management/server/posture/affects_posture_test.go

@@ -0,0 +1,202 @@
+package posture
+
+import (
+	"context"
+	"net"
+	"net/netip"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	nbpeer "github.com/netbirdio/netbird/management/server/peer"
+)
+
+// diffFrom builds a MetaDiff from the old/new snapshots AffectsPosture replays against.
+func diffFrom(oldMeta, newMeta nbpeer.PeerSystemMeta, oldLoc, newLoc nbpeer.Location) *nbpeer.MetaDiff {
+	return &nbpeer.MetaDiff{
+		OldMeta:     oldMeta,
+		NewMeta:     newMeta,
+		OldLocation: oldLoc,
+		NewLocation: newLoc,
+	}
+}
+
+func checks(def ChecksDefinition) []*Checks {
+	return []*Checks{{Checks: def}}
+}
+
+func TestAffectsPosture_NilDiff(t *testing.T) {
+	assert.False(t, AffectsPosture(context.Background(), nil, checks(ChecksDefinition{
+		NBVersionCheck: &NBVersionCheck{MinVersion: "1.0.0"},
+	})))
+}
+
+func TestAffectsPosture_NBVersion(t *testing.T) {
+	c := checks(ChecksDefinition{NBVersionCheck: &NBVersionCheck{MinVersion: "1.2.0"}})
+
+	tests := []struct {
+		name           string
+		oldVer, newVer string
+		want           bool
+	}{
+		{"both above min, no flip", "1.3.0", "1.4.0", false},
+		{"both below min, no flip", "1.0.0", "1.1.0", false},
+		{"crosses up below->above", "1.1.0", "1.3.0", true},
+		{"crosses down above->below", "1.3.0", "1.1.0", true},
+		{"unparsable old only -> flip", "garbage", "1.3.0", true},
+		{"unparsable both -> no flip", "garbage", "junk", false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			diff := diffFrom(
+				nbpeer.PeerSystemMeta{WtVersion: tt.oldVer},
+				nbpeer.PeerSystemMeta{WtVersion: tt.newVer},
+				nbpeer.Location{}, nbpeer.Location{},
+			)
+			assert.Equal(t, tt.want, AffectsPosture(context.Background(), diff, c))
+		})
+	}
+}
+
+func TestAffectsPosture_OSVersion_KernelBumpWithinMin(t *testing.T) {
+	c := checks(ChecksDefinition{OSVersionCheck: &OSVersionCheck{
+		Linux: &MinKernelVersionCheck{MinKernelVersion: "5.0.0"},
+	}})
+
+	// Kernel moves but stays above the minimum: verdict stays pass -> not affected.
+	withinMin := diffFrom(
+		nbpeer.PeerSystemMeta{GoOS: "linux", KernelVersion: "5.10.0-arch1"},
+		nbpeer.PeerSystemMeta{GoOS: "linux", KernelVersion: "5.15.0-arch2"},
+		nbpeer.Location{}, nbpeer.Location{},
+	)
+	assert.False(t, AffectsPosture(context.Background(), withinMin, c))
+
+	// Kernel drops below the minimum: verdict flips pass -> fail -> affected.
+	crossesDown := diffFrom(
+		nbpeer.PeerSystemMeta{GoOS: "linux", KernelVersion: "5.10.0-arch1"},
+		nbpeer.PeerSystemMeta{GoOS: "linux", KernelVersion: "4.19.0-arch1"},
+		nbpeer.Location{}, nbpeer.Location{},
+	)
+	assert.True(t, AffectsPosture(context.Background(), crossesDown, c))
+}
+
+func TestAffectsPosture_OSVersion_GoOSSwitchFlipsVerdict(t *testing.T) {
+	// Only Linux is constrained. An OS outside the switch (freebsd) passes; switching to a
+	// failing linux kernel flips the verdict pass -> fail.
+	c := checks(ChecksDefinition{OSVersionCheck: &OSVersionCheck{
+		Linux: &MinKernelVersionCheck{MinKernelVersion: "6.0.0"},
+	}})
+
+	diff := diffFrom(
+		nbpeer.PeerSystemMeta{GoOS: "freebsd"},
+		nbpeer.PeerSystemMeta{GoOS: "linux", KernelVersion: "4.19.0"},
+		nbpeer.Location{}, nbpeer.Location{},
+	)
+	assert.True(t, AffectsPosture(context.Background(), diff, c))
+}
+
+func TestAffectsPosture_Process_GoOSSwitchFlipsVerdict(t *testing.T) {
+	// Process runs at a linux path. Switching GoOS to windows (no WindowsPath configured)
+	// flips the verdict.
+	c := checks(ChecksDefinition{ProcessCheck: &ProcessCheck{
+		Processes: []Process{{LinuxPath: "/usr/bin/foo"}},
+	}})
+
+	files := []nbpeer.File{{Path: "/usr/bin/foo", ProcessIsRunning: true}}
+	diff := diffFrom(
+		nbpeer.PeerSystemMeta{GoOS: "linux", Files: files},
+		nbpeer.PeerSystemMeta{GoOS: "windows", Files: files},
+		nbpeer.Location{}, nbpeer.Location{},
+	)
+	assert.True(t, AffectsPosture(context.Background(), diff, c))
+}
+
+func TestAffectsPosture_Process_UnrelatedFileChange(t *testing.T) {
+	// A tracked process stays running while an unrelated file is added: the verdict does
+	// not move, so posture is not affected.
+	c := checks(ChecksDefinition{ProcessCheck: &ProcessCheck{
+		Processes: []Process{{LinuxPath: "/usr/bin/foo"}},
+	}})
+
+	diff := diffFrom(
+		nbpeer.PeerSystemMeta{GoOS: "linux", Files: []nbpeer.File{
+			{Path: "/usr/bin/foo", ProcessIsRunning: true},
+		}},
+		nbpeer.PeerSystemMeta{GoOS: "linux", Files: []nbpeer.File{
+			{Path: "/usr/bin/foo", ProcessIsRunning: true},
+			{Path: "/usr/bin/bar", ProcessIsRunning: true},
+		}},
+		nbpeer.Location{}, nbpeer.Location{},
+	)
+	assert.False(t, AffectsPosture(context.Background(), diff, c))
+}
+
+func TestAffectsPosture_GeoLocation(t *testing.T) {
+	c := checks(ChecksDefinition{GeoLocationCheck: &GeoLocationCheck{
+		Action:    CheckActionAllow,
+		Locations: []Location{{CountryCode: "DE"}},
+	}})
+
+	// Moving within allowed countries keeps the verdict; moving out flips it.
+	stayAllowed := diffFrom(
+		nbpeer.PeerSystemMeta{}, nbpeer.PeerSystemMeta{},
+		nbpeer.Location{CountryCode: "DE", CityName: "Berlin"},
+		nbpeer.Location{CountryCode: "DE", CityName: "Munich"},
+	)
+	assert.False(t, AffectsPosture(context.Background(), stayAllowed, c))
+
+	moveOut := diffFrom(
+		nbpeer.PeerSystemMeta{}, nbpeer.PeerSystemMeta{},
+		nbpeer.Location{CountryCode: "DE"},
+		nbpeer.Location{CountryCode: "FR"},
+	)
+	assert.True(t, AffectsPosture(context.Background(), moveOut, c))
+}
+
+func TestAffectsPosture_PeerNetworkRange_ConnectionIP(t *testing.T) {
+	// The check reads the connection IP. Moving out of the allowed range flips the verdict;
+	// moving within it does not.
+	_, allowed, _ := net.ParseCIDR("10.0.0.0/8")
+	c := checks(ChecksDefinition{PeerNetworkRangeCheck: &PeerNetworkRangeCheck{
+		Action: CheckActionAllow,
+		Ranges: []netip.Prefix{netip.MustParsePrefix(allowed.String())},
+	}})
+
+	movesOutOfRange := diffFrom(
+		nbpeer.PeerSystemMeta{}, nbpeer.PeerSystemMeta{},
+		nbpeer.Location{ConnectionIP: net.ParseIP("10.1.2.3")},
+		nbpeer.Location{ConnectionIP: net.ParseIP("8.8.8.8")},
+	)
+	assert.True(t, AffectsPosture(context.Background(), movesOutOfRange, c))
+
+	staysInRange := diffFrom(
+		nbpeer.PeerSystemMeta{}, nbpeer.PeerSystemMeta{},
+		nbpeer.Location{ConnectionIP: net.ParseIP("10.1.2.3")},
+		nbpeer.Location{ConnectionIP: net.ParseIP("10.9.9.9")},
+	)
+	assert.False(t, AffectsPosture(context.Background(), staysInRange, c))
+}
+
+func TestAffectsPosture_IrrelevantFieldChange(t *testing.T) {
+	// Hostname changes but no check reads it: not affected even with checks present.
+	c := checks(ChecksDefinition{
+		NBVersionCheck:   &NBVersionCheck{MinVersion: "1.0.0"},
+		GeoLocationCheck: &GeoLocationCheck{Action: CheckActionAllow, Locations: []Location{{CountryCode: "DE"}}},
+	})
+
+	diff := diffFrom(
+		nbpeer.PeerSystemMeta{Hostname: "old", WtVersion: "1.5.0"},
+		nbpeer.PeerSystemMeta{Hostname: "new", WtVersion: "1.5.0"},
+		nbpeer.Location{CountryCode: "DE"}, nbpeer.Location{CountryCode: "DE"},
+	)
+	assert.False(t, AffectsPosture(context.Background(), diff, c))
+}
+
+func TestAffectsPosture_NoChecks(t *testing.T) {
+	diff := diffFrom(
+		nbpeer.PeerSystemMeta{WtVersion: "1.0.0"},
+		nbpeer.PeerSystemMeta{WtVersion: "2.0.0"},
+		nbpeer.Location{}, nbpeer.Location{},
+	)
+	assert.False(t, AffectsPosture(context.Background(), diff, nil))
+}

management/server/posture/checks.go

@@ -7,6 +7,7 @@ import (
 	"regexp"
 
 	"github.com/hashicorp/go-version"
+	log "github.com/sirupsen/logrus"
 
 	nbpeer "github.com/netbirdio/netbird/management/server/peer"
 	"github.com/netbirdio/netbird/shared/management/http/api"
@@ -52,34 +53,46 @@ type Checks struct {
 	Checks ChecksDefinition `gorm:"serializer:json"`
 }
 
-// AffectsPosture reports whether the peer metadata changes described by diff can
-// alter the outcome of any of the given posture checks. It maps each check kind to
-// the metadata fields it inspects, so an unrelated change (e.g. a hostname update)
-// does not force a posture re-evaluation.
-func AffectsPosture(diff *nbpeer.MetaDiff, checks []*Checks) bool {
+// AffectsPosture reports whether the change in diff flips the verdict of any check. It
+// replays each check against the peer's old and new state and compares verdicts, so a
+// change that moves a field but stays the right side of a threshold (e.g. a kernel bump
+// still above the minimum) does not force a re-evaluation. See verdictChanged for how an
+// evaluation error counts.
+func AffectsPosture(ctx context.Context, diff *nbpeer.MetaDiff, checks []*Checks) bool {
 	if diff == nil {
 		return false
 	}
+
+	oldPeer := nbpeer.Peer{Meta: diff.OldMeta, Location: diff.OldLocation}
+	newPeer := nbpeer.Peer{Meta: diff.NewMeta, Location: diff.NewLocation}
+
 	for _, c := range checks {
-		if c.Checks.ProcessCheck != nil && diff.Files {
-			return true
-		}
-		if c.Checks.OSVersionCheck != nil && (diff.OSVersion || diff.OS || diff.KernelVersion) {
-			return true
-		}
-		if c.Checks.NBVersionCheck != nil && diff.WtVersion {
-			return true
-		}
-		if c.Checks.GeoLocationCheck != nil && diff.LocationChanged {
-			return true
-		}
-		if c.Checks.PeerNetworkRangeCheck != nil && diff.NetworkAddresses {
-			return true
+		for _, check := range c.GetChecks() {
+			if verdictChanged(ctx, check, oldPeer, newPeer) {
+				return true
+			}
 		}
 	}
 	return false
 }
 
+// verdictChanged replays check against old and new state and reports whether the verdict
+// differs. Like callers, it treats an evaluation error as deny: two errors are the same
+// verdict (no change), an error on one side only is a flip.
+func verdictChanged(ctx context.Context, check Check, oldPeer, newPeer nbpeer.Peer) bool {
+	oldPass, oldErr := check.Check(ctx, oldPeer)
+	newPass, newErr := check.Check(ctx, newPeer)
+
+	oldVerdict := oldPass && (oldErr == nil)
+	newVerdict := newPass && (newErr == nil)
+	changed := oldVerdict != newVerdict
+
+	log.WithContext(ctx).Tracef("posture check %s replay: verdict %t -> %t (changed=%t), errs: %v -> %v",
+		check.Name(), oldVerdict, newVerdict, changed, oldErr, newErr)
+
+	return changed
+}
+
 // ChecksDefinition contains definition of actual check
 type ChecksDefinition struct {
 	NBVersionCheck        *NBVersionCheck        `json:",omitempty"`

management/server/posture_checks_test.go

@@ -489,6 +489,7 @@ func TestArePostureCheckChangesAffectPeers(t *testing.T) {
 
 	policy := &types.Policy{
 		AccountID: account.Id,
+		Enabled:   true,
 		Rules: []*types.PolicyRule{
 			{
 				Enabled:      true,

management/server/user.go

@@ -1059,8 +1059,8 @@ func (am *DefaultAccountManager) BuildUserInfosForAccount(ctx context.Context, a
 		if err != nil {
 			return nil, err
 		}
-		log.WithContext(ctx).Debugf("Got %d users from ExternalCache for account %s", len(usersFromIntegration), accountID)
-		log.WithContext(ctx).Debugf("Got %d users from InternalCache for account %s", len(queriedUsers), accountID)
+		log.WithContext(ctx).Tracef("Got %d users from ExternalCache for account %s", len(usersFromIntegration), accountID)
+		log.WithContext(ctx).Tracef("Got %d users from InternalCache for account %s", len(queriedUsers), accountID)
 		queriedUsers = append(queriedUsers, usersFromIntegration...)
 	}
 

shared/management/http/api/openapi.yml

@@ -2765,6 +2765,28 @@ components:
           type: integer
           description: "Number of packets transmitted."
           example: 5
+        num_of_starts:
+          type: integer
+          description: "Number of start events."
+          example: 3
+        num_of_ends:
+          type: integer
+          description: "Number of end events."
+          example: 4
+        num_of_drops:
+          type: integer
+          description: "Number of drop events."
+          example: 5
+        window_start:
+          type: string
+          format: date-time
+          description: Timestamp of the start of the aggregation window.
+          example: 2025-03-20T16:23:58.125397Z
+        window_end:
+          type: string
+          format: date-time
+          description: Timestamp of the end of the aggregation window.
+          example: 2025-03-20T16:23:58.125397Z
         events:
           type: array
           description: "List of events that are correlated to this flow (e.g., start, end)."
@@ -2786,6 +2808,11 @@ components:
         - rx_packets
         - tx_bytes
         - tx_packets
+        - num_of_starts
+        - num_of_ends
+        - num_of_drops
+        - window_start
+        - window_end
         - events
     NetworkTrafficEventsResponse:
       type: object

shared/management/http/api/types.gen.go

@@ -2905,9 +2905,18 @@ type NetworkTrafficEvent struct {
 	Events []NetworkTrafficSubEvent `json:"events"`
 
 	// FlowId FlowID is the ID of the connection flow. Not unique because it can be the same for multiple events (e.g., start and end of the connection).
-	FlowId string               `json:"flow_id"`
-	Icmp   NetworkTrafficICMP   `json:"icmp"`
-	Policy NetworkTrafficPolicy `json:"policy"`
+	FlowId string             `json:"flow_id"`
+	Icmp   NetworkTrafficICMP `json:"icmp"`
+
+	// NumOfDrops Number of drop events.
+	NumOfDrops int `json:"num_of_drops"`
+
+	// NumOfEnds Number of end events.
+	NumOfEnds int `json:"num_of_ends"`
+
+	// NumOfStarts Number of start events.
+	NumOfStarts int                  `json:"num_of_starts"`
+	Policy      NetworkTrafficPolicy `json:"policy"`
 
 	// Protocol Protocol is the protocol of the traffic (e.g. 1 = ICMP, 6 = TCP, 17 = UDP, etc.).
 	Protocol int `json:"protocol"`
@@ -2928,6 +2937,12 @@ type NetworkTrafficEvent struct {
 	// TxPackets Number of packets transmitted.
 	TxPackets int                `json:"tx_packets"`
 	User      NetworkTrafficUser `json:"user"`
+
+	// WindowEnd Timestamp of the end of the aggregation window.
+	WindowEnd time.Time `json:"window_end"`
+
+	// WindowStart Timestamp of the start of the aggregation window.
+	WindowStart time.Time `json:"window_start"`
 }
 
 // NetworkTrafficEventsResponse defines model for NetworkTrafficEventsResponse.

shared/management/status/error.go

@@ -48,6 +48,10 @@ type Type int32
 var (
 	ErrExtraSettingsNotFound = errors.New("extra settings not found")
 	ErrPeerAlreadyLoggedIn   = errors.New("peer with the same public key is already logged in")
+
+	// ErrNoAuthMethodProvided is returned when a peer login attempt carries neither a
+	// setup key nor an SSO token. Match it with errors.Is.
+	ErrNoAuthMethodProvided = Errorf(Unauthenticated, "no peer auth method provided, please use a setup key or interactive SSO login")
 )
 
 // Error is an internal error
@@ -66,6 +70,16 @@ func (e *Error) Error() string {
 	return e.Message
 }
 
+// Is reports whether target is an *Error with the same type and message,
+// enabling matching with errors.Is against sentinel errors.
+func (e *Error) Is(target error) bool {
+	var t *Error
+	if !errors.As(target, &t) {
+		return false
+	}
+	return e.ErrorType == t.ErrorType && e.Message == t.Message
+}
+
 // Errorf returns Error(ErrorType, fmt.Sprintf(format, a...)).
 func Errorf(errorType Type, format string, a ...interface{}) error {
 	return &Error{