Merge remote-tracking branch 'origin/revert/component-types' into revert/component-types

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

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

@@ -21,6 +21,7 @@ import (
 	networkmapdb "github.com/netbirdio/netbird/management/internals/network_map_db"
 	"github.com/netbirdio/netbird/management/internals/server/config"
 	"github.com/netbirdio/netbird/management/internals/shared/grpc"
+	"github.com/netbirdio/netbird/management/internals/shared/requestbuffer"
 	"github.com/netbirdio/netbird/management/server/account"
 	"github.com/netbirdio/netbird/management/server/integrations/integrated_validator"
 	"github.com/netbirdio/netbird/management/server/integrations/port_forwarding"
@@ -39,6 +40,8 @@ import (
 	"github.com/netbirdio/netbird/version"
 )
 
+const defaultNetworkMapDataBufferInterval = 100 * time.Millisecond
+
 type Controller struct {
 	repo    Repository
 	metrics *metrics
@@ -65,7 +68,8 @@ type Controller struct {
 
 	perAccountServerSupportedSyncMessageVersions map[string]sharedgrpc.SyncMessageVersion
 
-	nmdataStore *networkmapdb.NetworkMapDBStoreImpl
+	nmdataStore  *networkmapdb.NetworkMapDBStoreImpl
+	nmdataBuffer *requestbuffer.Buffer[*networkmap.NetworkMapData]
 }
 
 type bufferUpdate struct {
@@ -89,7 +93,7 @@ func NewController(ctx context.Context, store store.Store, metrics telemetry.App
 		log.Fatal(fmt.Errorf("error creating metrics: %w", err))
 	}
 
-	return &Controller{
+	c := &Controller{
 		repo:                    newRepository(store),
 		metrics:                 nMetrics,
 		accountManagerMetrics:   metrics.AccountManagerMetrics(),
@@ -106,6 +110,14 @@ func NewController(ctx context.Context, store store.Store, metrics telemetry.App
 		perAccountServerSupportedSyncMessageVersions: sharedgrpc.SyncMessageVersionsFromMap(config.PerAccountHighestSupportedSyncMessageVersion),
 		nmdataStore:                                  nmdataStore,
 	}
+
+	if nmdataStore != nil {
+		interval := requestbuffer.Interval(ctx, "NB_NETWORK_MAP_DATA_BUFFER_INTERVAL", defaultNetworkMapDataBufferInterval)
+		log.WithContext(ctx).Infof("set network map data request buffer interval to %s", interval)
+		c.nmdataBuffer = requestbuffer.New(ctx, "network map data request buffer", interval, c.fetchNetworkMapData)
+	}
+
+	return c
 }
 
 func (c *Controller) OnPeerConnected(ctx context.Context, accountID string, peerID string) (chan *network_map.UpdateMessage, error) {
@@ -392,8 +404,6 @@ func (c *Controller) sendUpdatesFromData(ctx context.Context, accountID string,
 		return fmt.Errorf("failed to get flow enabled status: %v", err)
 	}
 
-	nmData.PrecomputePostureValidation()
-
 	dnsCache := &cache.DNSConfigCache{}
 	dnsDomain := c.getDNSDomainFromData(nmData.AccountSettings)
 	peersCustomZone := networkmap.PeersCustomZone(ctx, accountID, dnsDomain, nmData.Peers, IPv6AllowedPeersFromData(nmData))
@@ -452,7 +462,7 @@ func (c *Controller) sendUpdatesFromData(ctx context.Context, accountID string,
 				return
 			}
 
-			nmap := NetworkMapFromData(ctx, nmData, p.ID, peersCustomZone)
+			nmap := NetworkMapFromData(ctx, nmData, p.ID, peersCustomZone, c.accountManagerMetrics)
 
 			c.metrics.CountCalcPeerNetworkMapDuration(time.Since(start))
 
@@ -476,21 +486,37 @@ func (c *Controller) sendUpdatesFromData(ctx context.Context, accountID string,
 }
 
 func (c *Controller) getNetworkMapData(ctx context.Context, accountID string) *networkmap.NetworkMapData {
-	if c.nmdataStore == nil {
+	if c.nmdataBuffer == nil {
 		return nil
 	}
 
-	nmData, err := c.nmdataStore.GetNetworkMapData(ctx, accountID)
+	nmData, err := c.nmdataBuffer.Get(ctx, accountID)
 	if err != nil {
 		log.WithContext(ctx).Errorf("failed to get network map data for account %s, falling back to account-based computation: %v", accountID, err)
 		return nil
 	}
 
-	nmData.Services = c.proxyServicesFromRepo(ctx, accountID)
-
 	return nmData
 }
 
+// fetchNetworkMapData reads the twin once per buffer window. Its result is
+// shared by every waiter of that window, so the mutating steps run here, before
+// it is handed out: the twin the callers see is read-only. Injected proxy
+// policies carry no posture checks, so precomputing after the injection yields
+// the same validation as precomputing before it.
+func (c *Controller) fetchNetworkMapData(ctx context.Context, accountID string) (*networkmap.NetworkMapData, error) {
+	nmData, err := c.nmdataStore.GetNetworkMapData(ctx, accountID)
+	if err != nil {
+		return nil, err
+	}
+
+	nmData.Services = c.proxyServicesFromRepo(ctx, accountID)
+	nmData.InjectProxyPolicies()
+	nmData.PrecomputePostureValidation()
+
+	return nmData, nil
+}
+
 func (c *Controller) getDNSDomainFromData(settings *nmdata.AccountSettingsInfo) string {
 	if settings == nil || settings.DNSDomain == "" {
 		return c.dnsDomain
@@ -500,15 +526,18 @@ func (c *Controller) getDNSDomainFromData(settings *nmdata.AccountSettingsInfo)
 
 func IPv6AllowedPeersFromData(nmData *networkmap.NetworkMapData) map[string]struct{} {
 	result := make(map[string]struct{})
-	if nmData.AccountSettings != nil {
-		for _, groupID := range nmData.AccountSettings.IPv6EnabledGroups {
-			group := nmData.Groups[groupID]
-			if group == nil {
-				continue
-			}
-			for _, peerID := range group.Peers {
-				result[peerID] = struct{}{}
-			}
+	// An account with no IPv6-enabled group runs no overlay at all, so the
+	// embedded-proxy carve-out below has nothing to reach and stays shut.
+	if nmData.AccountSettings == nil || len(nmData.AccountSettings.IPv6EnabledGroups) == 0 {
+		return result
+	}
+	for _, groupID := range nmData.AccountSettings.IPv6EnabledGroups {
+		group := nmData.Groups[groupID]
+		if group == nil {
+			continue
+		}
+		for _, peerID := range group.Peers {
+			result[peerID] = struct{}{}
 		}
 	}
 	for id, p := range nmData.Peers {
@@ -519,12 +548,22 @@ func IPv6AllowedPeersFromData(nmData *networkmap.NetworkMapData) map[string]stru
 	return result
 }
 
-func NetworkMapFromData(ctx context.Context, nmData *networkmap.NetworkMapData, peerID string, peersCustomZone nmdata.CustomZone) *types.NetworkMap {
+func NetworkMapFromData(ctx context.Context, nmData *networkmap.NetworkMapData, peerID string, peersCustomZone nmdata.CustomZone, metrics *telemetry.AccountManagerMetrics) *types.NetworkMap {
+	start := time.Now()
+
 	components := nmData.GetPeerNetworkMapComponents(peerID, peersCustomZone)
 	if components.IsEmpty() {
 		return &types.NetworkMap{Network: components.Network}
 	}
-	return types.CalculateNetworkMapFromComponents(ctx, components)
+	nm := types.CalculateNetworkMapFromComponents(ctx, components)
+
+	if metrics != nil {
+		objectCount := int64(len(nm.Peers) + len(nm.OfflinePeers) + len(nm.Routes) + len(nm.FirewallRules) + len(nm.RoutesFirewallRules))
+		metrics.CountNetworkMapObjects(objectCount)
+		metrics.CountGetPeerNetworkMapDuration(time.Since(start))
+	}
+
+	return nm
 }
 
 // peerPostureChecksFromData mirrors getPeerPostureChecks on the twin store. The
@@ -1171,7 +1210,7 @@ func (c *Controller) getValidatedPeerWithMapFromData(ctx context.Context, accoun
 	dnsDomain := c.getDNSDomainFromData(nmData.AccountSettings)
 	peersCustomZone := networkmap.PeersCustomZone(ctx, accountID, dnsDomain, nmData.Peers, IPv6AllowedPeersFromData(nmData))
 
-	networkMap := NetworkMapFromData(ctx, nmData, peerID, peersCustomZone)
+	networkMap := NetworkMapFromData(ctx, nmData, peerID, peersCustomZone, c.accountManagerMetrics)
 	dnsFwdPort := ComputeForwarderPortFromData(nmData.Peers, network_map.DnsForwarderPortMinVersion)
 
 	return networkMap, postureChecks, dnsFwdPort, nil
@@ -1402,7 +1441,12 @@ func (c *Controller) GetNetworkMap(ctx context.Context, peerID string) (*types.N
 		groups[groupID] = group.Peers
 	}
 
-	validatedPeers, err := c.integratedPeerValidator.GetValidatedPeers(ctx, account.Id, types.TwinGroups(maps.Values(account.Groups)), types.TwinPeers(maps.Values(account.Peers)), account.Settings.Extra)
+	extraSettings, err := c.settingsManager.GetExtraSettings(ctx, account.Id)
+	if err != nil {
+		return nil, err
+	}
+
+	validatedPeers, err := c.integratedPeerValidator.GetValidatedPeers(ctx, account.Id, types.TwinGroups(maps.Values(account.Groups)), types.TwinPeers(maps.Values(account.Peers)), extraSettings)
 	if err != nil {
 		return nil, err
 	}

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

@@ -0,0 +1,47 @@
+package controller
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/netbirdio/netbird/shared/management/networkmap"
+	"github.com/netbirdio/netbird/shared/management/networkmap/nmdata"
+)
+
+// The account-side builder (types.Account.peerIPv6AllowedSet) is the reference:
+// an account with no IPv6-enabled group runs no IPv6 overlay at all, embedded
+// proxy peers included — see TestPeerIPv6AllowedEmbeddedProxy. Both builders
+// gate the same AAAA records, so the store-backed one has to agree.
+func TestIPv6AllowedPeersFromData(t *testing.T) {
+	data := func(enabledGroups []string) *networkmap.NetworkMapData {
+		return &networkmap.NetworkMapData{
+			AccountSettings: &nmdata.AccountSettingsInfo{IPv6EnabledGroups: enabledGroups},
+			Peers: map[string]*nmdata.Peer{
+				"peer1":  {ID: "peer1"},
+				"lonely": {ID: "lonely"},
+				"proxy":  {ID: "proxy", ProxyMeta: nmdata.ProxyMeta{Embedded: true, Cluster: "netbird.test"}},
+			},
+			Groups: map[string]*nmdata.Group{
+				"group-devs": {ID: "group-devs", Peers: []string{"peer1"}},
+			},
+		}
+	}
+
+	t.Run("embedded proxy allowed when any v6 group exists, without group membership", func(t *testing.T) {
+		allowed := IPv6AllowedPeersFromData(data([]string{"group-devs"}))
+		assert.Contains(t, allowed, "proxy", "embedded proxy participates in v6 overlay")
+		assert.Contains(t, allowed, "peer1", "regular peer in enabled group still allowed")
+	})
+
+	t.Run("embedded proxy denied when no v6 group enabled", func(t *testing.T) {
+		allowed := IPv6AllowedPeersFromData(data(nil))
+		assert.NotContains(t, allowed, "proxy", "v6 disabled account-wide denies embedded proxies too")
+		assert.Empty(t, allowed, "no peer participates in the v6 overlay")
+	})
+
+	t.Run("non-embedded peer outside any enabled group is not pulled in", func(t *testing.T) {
+		allowed := IPv6AllowedPeersFromData(data([]string{"group-devs"}))
+		assert.NotContains(t, allowed, "lonely", "embedded-proxy bypass must not leak to regular peers")
+	})
+}

management/internals/controllers/network_map/nmaptest/runner.go

@@ -236,7 +236,7 @@ func computeMode(t *testing.T, ctx context.Context, mode Mode, nmData *networkma
 	case ModeLegacy:
 		return computeLegacy(t, ctx, legacy, peerID, zone, dnsDomain, dnsFwdPort)
 	case ModeFull:
-		nmap := controller.NetworkMapFromData(ctx, nmData, peerID, zone)
+		nmap := controller.NetworkMapFromData(ctx, nmData, peerID, zone, nil)
 		return mgmtgrpc.ToSyncResponse(ctx, nil, nil, nil, peer, nil, nil, nmap, dnsDomain, nil,
 			&cache.DNSConfigCache{}, nmData.AccountSettings, nil, nil, dnsFwdPort).NetworkMap
 	case ModeEnvelope:

management/internals/controllers/network_map/nmaptest/testdata/cases/dns-config/case.json

@@ -1,5 +1,5 @@
 {
-  "description": "Nameserver group and applied custom zone distributed to grp-dev; peer-a (with an extra DNS label) receives them, peer-c outside the group receives neither.",
+  "description": "Nameserver group and applied custom zones distributed to grp-dev; peer-a (with an extra DNS label) receives them, peer-c is outside that group and receives only the zone distributed to grp-ops. Zone flags travel per zone: both grp-dev zones are match-only (NonAuthoritative), only search-off.internal. disables the search domain, and the built-in peer zone stays authoritative.",
   "peers": [
     "peer-a",
     "peer-c"

management/internals/controllers/network_map/nmaptest/testdata/cases/dns-config/golden/peer-a.json

@@ -45,6 +45,27 @@
           }
         ]
       },
+      {
+        "Domain": "search-off.internal.",
+        "SearchDomainDisabled": true,
+        "NonAuthoritative": true,
+        "Records": [
+          {
+            "Name": "alias.search-off.internal.",
+            "Type": "5",
+            "Class": "IN",
+            "TTL": "300",
+            "RData": "app.search-off.internal."
+          },
+          {
+            "Name": "app.search-off.internal.",
+            "Type": "1",
+            "Class": "IN",
+            "TTL": "300",
+            "RData": "10.10.0.6"
+          }
+        ]
+      },
       {
         "Domain": "netbird.test.",
         "Records": [

management/internals/controllers/network_map/nmaptest/testdata/cases/dns-config/golden/peer-c.json

@@ -22,6 +22,19 @@
             "RData": "100.64.0.3"
           }
         ]
+      },
+      {
+        "Domain": "ops-only.internal.",
+        "NonAuthoritative": true,
+        "Records": [
+          {
+            "Name": "tool.ops-only.internal.",
+            "Type": "1",
+            "Class": "IN",
+            "TTL": "300",
+            "RData": "10.10.0.7"
+          }
+        ]
       }
     ],
     "ForwarderPort": "22054"

management/internals/controllers/network_map/nmaptest/testdata/cases/dns-config/nmdata.json

@@ -47,6 +47,28 @@
           {"Name": "db.corp.internal.", "Type": 1, "Class": "IN", "TTL": 300, "RData": "10.10.0.5"}
         ]
       }
+    },
+    {
+      "DistributionGroups": ["grp-dev"],
+      "Zone": {
+        "Domain": "search-off.internal.",
+        "NonAuthoritative": true,
+        "SearchDomainDisabled": true,
+        "Records": [
+          {"Name": "app.search-off.internal.", "Type": 1, "Class": "IN", "TTL": 300, "RData": "10.10.0.6"},
+          {"Name": "alias.search-off.internal.", "Type": 5, "Class": "IN", "TTL": 300, "RData": "app.search-off.internal."}
+        ]
+      }
+    },
+    {
+      "DistributionGroups": ["grp-ops"],
+      "Zone": {
+        "Domain": "ops-only.internal.",
+        "NonAuthoritative": true,
+        "Records": [
+          {"Name": "tool.ops-only.internal.", "Type": 1, "Class": "IN", "TTL": 300, "RData": "10.10.0.7"}
+        ]
+      }
     }
   ]
 }

management/internals/network_map_db/pgsql/dns.go

@@ -13,8 +13,8 @@ const (
 	select zones.id as id, domain, not enable_search_domain as search_domain_disabled, distribution_groups,
 	r.name as record_name, r.type as record_type, 'IN' record_class, r.ttl as record_ttl, r.content as record_rdata
 	from zones
-	left join records as r on r.zone_id = zones.id 
-	where zones.account_id=$1
+	left join records as r on r.zone_id = zones.id
+	where zones.account_id=$1 and zones.enabled
 	`
 )
 

management/internals/network_map_db/shared_types.go

@@ -248,6 +248,12 @@ func ZonesToAppliedZoneCandidates(zones []Zone) ([]networkmap.AppliedZoneCandida
 		}
 
 		if z.Id != currentZoneId {
+			// The account-side builder (types.buildAppliedZoneCandidates) states
+			// the shape of an applied zone: names fully qualified, served
+			// non-authoritatively. Both builders feed the same client-facing map,
+			// so this one has to produce the same value.
+			zone.Domain = dns.Fqdn(zone.Domain)
+			zone.NonAuthoritative = true
 			zone.Records = []nmdata.SimpleRecord{}
 			toret = append(toret, AppliedZoneCandidateFromZone(zone, distributionGroups))
 			currentZoneId = z.Id
@@ -263,7 +269,7 @@ func ZonesToAppliedZoneCandidates(zones []Zone) ([]networkmap.AppliedZoneCandida
 
 		lastZone := &toret[len(toret)-1]
 		lastZone.Zone.Records = append(lastZone.Zone.Records, nmdata.SimpleRecord{
-			Name:  z.RecordName.String,
+			Name:  dns.Fqdn(z.RecordName.String),
 			Class: z.RecordClass.String,
 			TTL:   int(z.RecordTTL.Int64),
 			RData: rdata,

management/internals/network_map_db/sqlite/dns.go

@@ -12,8 +12,8 @@ const (
 	select zones.id as id, domain, not enable_search_domain as search_domain_disabled, distribution_groups,
 	r.name as record_name, r.type as record_type, 'IN' record_class, r.ttl as record_ttl, r.content as record_rdata
 	from zones
-	left join records as r on r.zone_id = zones.id 
-	where zones.account_id=?
+	left join records as r on r.zone_id = zones.id
+	where zones.account_id=? and zones.enabled
 	`
 )
 

management/internals/shared/requestbuffer/buffer.go

@@ -0,0 +1,102 @@
+// Package requestbuffer coalesces concurrent reads of the same expensive
+// resource into a single fetch.
+package requestbuffer
+
+import (
+	"context"
+	"os"
+	"sync"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+)
+
+// FetchFunc reads the resource identified by key.
+type FetchFunc[T any] func(ctx context.Context, key string) (T, error)
+
+// Buffer batches requests per key: the first request opens a window, every
+// request arriving within it joins the batch, and a single fetch serves them
+// all. The fetch starts only after the window closed, so a caller never
+// observes data read before its own request.
+type Buffer[T any] struct {
+	ctx      context.Context
+	name     string
+	fetch    FetchFunc[T]
+	interval time.Duration
+
+	mu      sync.Mutex
+	waiting map[string][]chan result[T]
+}
+
+type result[T any] struct {
+	value T
+	err   error
+}
+
+// New returns a Buffer serving batched requests through fetch. ctx bounds the
+// fetches, not the callers, and must outlive them.
+func New[T any](ctx context.Context, name string, interval time.Duration, fetch FetchFunc[T]) *Buffer[T] {
+	return &Buffer[T]{
+		ctx:      ctx,
+		name:     name,
+		fetch:    fetch,
+		interval: interval,
+		waiting:  make(map[string][]chan result[T]),
+	}
+}
+
+// Get returns the value for key, sharing one fetch with the other callers of
+// the current batch. The value is shared as is, so callers must treat it as
+// read-only unless the fetch hands out copies.
+func (b *Buffer[T]) Get(ctx context.Context, key string) (T, error) {
+	ch := make(chan result[T], 1)
+
+	b.mu.Lock()
+	b.waiting[key] = append(b.waiting[key], ch)
+	first := len(b.waiting[key]) == 1
+	b.mu.Unlock()
+
+	if first {
+		time.AfterFunc(b.interval, func() { b.flush(key) })
+	}
+
+	select {
+	case res := <-ch:
+		return res.value, res.err
+	case <-ctx.Done():
+		var zero T
+		return zero, ctx.Err()
+	}
+}
+
+func (b *Buffer[T]) flush(key string) {
+	b.mu.Lock()
+	waiting := b.waiting[key]
+	delete(b.waiting, key)
+	b.mu.Unlock()
+
+	if len(waiting) == 0 {
+		return
+	}
+
+	start := time.Now()
+	value, err := b.fetch(b.ctx, key)
+	log.WithContext(b.ctx).Tracef("%s: fetched %s for %d waiters in %s", b.name, key, len(waiting), time.Since(start))
+
+	for _, ch := range waiting {
+		ch <- result[T]{value: value, err: err}
+	}
+}
+
+// Interval reads a buffer interval from envVar, falling back to def.
+func Interval(ctx context.Context, envVar string, def time.Duration) time.Duration {
+	value := os.Getenv(envVar)
+	interval, err := time.ParseDuration(value)
+	if err != nil {
+		if value != "" {
+			log.WithContext(ctx).Warnf("failed to parse %s: %s", envVar, err)
+		}
+		return def
+	}
+	return interval
+}

management/internals/shared/requestbuffer/buffer_test.go

@@ -0,0 +1,106 @@
+package requestbuffer
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestBufferCoalescesConcurrentRequests(t *testing.T) {
+	var fetches atomic.Int32
+	buffer := New(context.Background(), "test", 50*time.Millisecond,
+		func(ctx context.Context, key string) (string, error) {
+			fetches.Add(1)
+			return key, nil
+		})
+
+	var wg sync.WaitGroup
+	for range 10 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			value, err := buffer.Get(context.Background(), "account")
+			assert.NoError(t, err)
+			assert.Equal(t, "account", value)
+		}()
+	}
+	wg.Wait()
+
+	assert.Equal(t, int32(1), fetches.Load())
+}
+
+func TestBufferSeparatesKeys(t *testing.T) {
+	keys := make(chan string, 2)
+	buffer := New(context.Background(), "test", 10*time.Millisecond,
+		func(ctx context.Context, key string) (string, error) {
+			keys <- key
+			return key, nil
+		})
+
+	var wg sync.WaitGroup
+	for _, key := range []string{"a", "b"} {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			_, err := buffer.Get(context.Background(), key)
+			assert.NoError(t, err)
+		}()
+	}
+	wg.Wait()
+	close(keys)
+
+	var fetched []string
+	for key := range keys {
+		fetched = append(fetched, key)
+	}
+	assert.ElementsMatch(t, []string{"a", "b"}, fetched)
+}
+
+func TestBufferFetchesAfterRequest(t *testing.T) {
+	var version atomic.Int32
+	buffer := New(context.Background(), "test", 10*time.Millisecond,
+		func(ctx context.Context, key string) (int32, error) {
+			return version.Load(), nil
+		})
+
+	first, err := buffer.Get(context.Background(), "account")
+	require.NoError(t, err)
+	assert.Equal(t, int32(0), first)
+
+	version.Store(1)
+
+	second, err := buffer.Get(context.Background(), "account")
+	require.NoError(t, err)
+	assert.Equal(t, int32(1), second)
+}
+
+func TestBufferPropagatesError(t *testing.T) {
+	fetchErr := errors.New("fetch failed")
+	buffer := New(context.Background(), "test", 10*time.Millisecond,
+		func(ctx context.Context, key string) (*int, error) {
+			return nil, fetchErr
+		})
+
+	value, err := buffer.Get(context.Background(), "account")
+	assert.ErrorIs(t, err, fetchErr)
+	assert.Nil(t, value)
+}
+
+func TestBufferHonorsCallerContext(t *testing.T) {
+	buffer := New(context.Background(), "test", time.Minute,
+		func(ctx context.Context, key string) (string, error) {
+			return key, nil
+		})
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
+	defer cancel()
+
+	_, err := buffer.Get(ctx, "account")
+	assert.ErrorIs(t, err, context.DeadlineExceeded)
+}

management/server/account_request_buffer.go

@@ -2,117 +2,38 @@ package server
 
 import (
 	"context"
-	"os"
-	"sync"
 	"time"
 
 	log "github.com/sirupsen/logrus"
 
+	"github.com/netbirdio/netbird/management/internals/shared/requestbuffer"
 	"github.com/netbirdio/netbird/management/server/store"
 	"github.com/netbirdio/netbird/management/server/types"
 )
 
-// AccountRequest holds the result channel to return the requested account.
-type AccountRequest struct {
-	AccountID  string
-	ResultChan chan *AccountResult
-}
-
-// AccountResult holds the account data or an error.
-type AccountResult struct {
-	Account *types.Account
-	Err     error
-}
+const defaultAccountBufferInterval = 100 * time.Millisecond
 
 type AccountRequestBuffer struct {
-	store               store.Store
-	getAccountRequests  map[string][]*AccountRequest
-	mu                  sync.Mutex
-	getAccountRequestCh chan *AccountRequest
-	bufferInterval      time.Duration
+	buffer *requestbuffer.Buffer[*types.Account]
 }
 
 func NewAccountRequestBuffer(ctx context.Context, store store.Store) *AccountRequestBuffer {
-	bufferIntervalStr := os.Getenv("NB_GET_ACCOUNT_BUFFER_INTERVAL")
-	bufferInterval, err := time.ParseDuration(bufferIntervalStr)
-	if err != nil {
-		if bufferIntervalStr != "" {
-			log.WithContext(ctx).Warnf("failed to parse account request buffer interval: %s", err)
-		}
-		bufferInterval = 100 * time.Millisecond
+	interval := requestbuffer.Interval(ctx, "NB_GET_ACCOUNT_BUFFER_INTERVAL", defaultAccountBufferInterval)
+	log.WithContext(ctx).Infof("set account request buffer interval to %s", interval)
+
+	return &AccountRequestBuffer{
+		buffer: requestbuffer.New(ctx, "account request buffer", interval, store.GetAccount),
 	}
-
-	log.WithContext(ctx).Infof("set account request buffer interval to %s", bufferInterval)
-
-	ac := AccountRequestBuffer{
-		store:               store,
-		getAccountRequests:  make(map[string][]*AccountRequest),
-		getAccountRequestCh: make(chan *AccountRequest),
-		bufferInterval:      bufferInterval,
-	}
-
-	go ac.processGetAccountRequests(ctx)
-
-	return &ac
 }
+
 func (ac *AccountRequestBuffer) GetAccountWithBackpressure(ctx context.Context, accountID string) (*types.Account, error) {
-	req := &AccountRequest{
-		AccountID:  accountID,
-		ResultChan: make(chan *AccountResult, 1),
+	account, err := ac.buffer.Get(ctx, accountID)
+	if err != nil || account == nil {
+		return account, err
 	}
 
-	log.WithContext(ctx).Tracef("requesting account %s with backpressure", accountID)
-	startTime := time.Now()
-	ac.getAccountRequestCh <- req
-
-	result := <-req.ResultChan
-	log.WithContext(ctx).Tracef("got account with backpressure after %s", time.Since(startTime))
-	return result.Account, result.Err
-}
-
-func (ac *AccountRequestBuffer) processGetAccountBatch(ctx context.Context, accountID string) {
-	ac.mu.Lock()
-	requests := ac.getAccountRequests[accountID]
-	delete(ac.getAccountRequests, accountID)
-	ac.mu.Unlock()
-
-	if len(requests) == 0 {
-		return
-	}
-
-	startTime := time.Now()
-	account, err := ac.store.GetAccount(ctx, accountID)
-	log.WithContext(ctx).Tracef("getting account %s in batch took %s", accountID, time.Since(startTime))
-	result := &AccountResult{Account: account, Err: err}
-
-	for _, req := range requests {
-		if account != nil {
-			// Shallow copy the account so each goroutine gets its own struct value.
-			// This prevents data races when callers mutate fields like Policies.
-			accountCopy := *account
-			req.ResultChan <- &AccountResult{Account: &accountCopy, Err: err}
-		} else {
-			req.ResultChan <- result
-		}
-		close(req.ResultChan)
-	}
-}
-
-func (ac *AccountRequestBuffer) processGetAccountRequests(ctx context.Context) {
-	for {
-		select {
-		case req := <-ac.getAccountRequestCh:
-			ac.mu.Lock()
-			ac.getAccountRequests[req.AccountID] = append(ac.getAccountRequests[req.AccountID], req)
-			if len(ac.getAccountRequests[req.AccountID]) == 1 {
-				go func(ctx context.Context, accountID string) {
-					time.Sleep(ac.bufferInterval)
-					ac.processGetAccountBatch(ctx, accountID)
-				}(ctx, req.AccountID)
-			}
-			ac.mu.Unlock()
-		case <-ctx.Done():
-			return
-		}
-	}
+	// Shallow copy the account so each caller gets its own struct value.
+	// This prevents data races when callers mutate fields like Policies.
+	accountCopy := *account
+	return &accountCopy, nil
 }

management/server/types/legacynmap/equivalence_test.go

@@ -50,7 +50,7 @@ import (
 	"gorm.io/gorm"
 	gormlogger "gorm.io/gorm/logger"
 
-	nbdns "github.com/netbirdio/netbird/dns"
+	"github.com/netbirdio/netbird/management/internals/controllers/network_map/controller"
 	"github.com/netbirdio/netbird/management/internals/controllers/network_map/controller/cache"
 	networkmapdb "github.com/netbirdio/netbird/management/internals/network_map_db"
 	networkmap_pgsql "github.com/netbirdio/netbird/management/internals/network_map_db/pgsql"
@@ -60,7 +60,7 @@ import (
 	"github.com/netbirdio/netbird/management/server/store"
 	"github.com/netbirdio/netbird/management/server/types"
 	"github.com/netbirdio/netbird/management/server/types/legacynmap"
-	"github.com/netbirdio/netbird/shared/management/networkmap/nmdata"
+	"github.com/netbirdio/netbird/shared/management/networkmap"
 	"github.com/netbirdio/netbird/shared/management/proto"
 )
 
@@ -108,7 +108,7 @@ func TestNetworkMapProtoEquivalence(t *testing.T) {
 			continue
 		}
 
-		checkAccount(ctx, t, nmStore, account, maxPeers, stats)
+		checkAccount(ctx, t, testStore, nmStore, account, maxPeers, stats)
 
 		account = nil
 		debug.FreeOSMemory()
@@ -126,7 +126,7 @@ func TestNetworkMapProtoEquivalence(t *testing.T) {
 
 // checkAccount compares both paths for every peer of one account. Nothing is
 // retained across peers, so memory stays flat within an account.
-func checkAccount(ctx context.Context, t *testing.T, nmStore *networkmapdb.NetworkMapDBStoreImpl, account *types.Account, maxPeers int, stats *equivStats) {
+func checkAccount(ctx context.Context, t *testing.T, accountStore store.Store, nmStore *networkmapdb.NetworkMapDBStoreImpl, account *types.Account, maxPeers int, stats *equivStats) {
 	t.Helper()
 
 	if len(account.Peers) == 0 {
@@ -150,11 +150,14 @@ func checkAccount(ctx context.Context, t *testing.T, nmStore *networkmapdb.Netwo
 	// Production fills ValidatedPeers via the integrated-validator wrapper; here
 	// every peer counts as validated, matching the legacy side's map.
 	nmData.ValidatedPeers = validated
-	// The legacy side receives no account zones (main sourced them from the
-	// external zones manager), so the DB-sourced applied-zone candidates must be
-	// dropped to keep the comparison surface identical. PrivateServiceCandidates
-	// stay: all paths derive them from account/DB data.
-	nmData.AppliedZoneCandidates = nil
+
+	// Custom DNS zones are built twice from the same rows — the account side
+	// from the zones manager, the store side in SQL — so both are fed in and
+	// compared rather than dropped. The same goes for the peers zone below:
+	// each side computes it with its own helper, which is where an AAAA gate
+	// that disagrees between the two would show up.
+	accountZones, err := accountStore.GetAccountZones(ctx, store.LockingStrengthNone, account.Id)
+	require.NoError(t, err, "account %s: load account zones", account.Id)
 
 	resourcePolicies := account.GetResourcePoliciesMap()
 	routers := account.GetResourceRoutersMap()
@@ -176,6 +179,9 @@ func checkAccount(ctx context.Context, t *testing.T, nmStore *networkmapdb.Netwo
 		settings = &types.Settings{}
 	}
 
+	accountPeersZone := account.GetPeersCustomZone(ctx, equivDNSName)
+	storePeersZone := networkmap.PeersCustomZone(ctx, account.Id, equivDNSName, nmData.Peers, controller.IPv6AllowedPeersFromData(nmData))
+
 	for _, peerID := range peerIDs {
 		peer := account.Peers[peerID]
 		if peer == nil {
@@ -188,7 +194,7 @@ func checkAccount(ctx context.Context, t *testing.T, nmStore *networkmapdb.Netwo
 
 		// STORE PATH — nmdata store through the production computation, mirroring
 		// the controller's networkMapFromData.
-		components := nmData.GetPeerNetworkMapComponents(peerID, nmdata.CustomZone{})
+		components := nmData.GetPeerNetworkMapComponents(peerID, storePeersZone)
 		storeNM := &types.NetworkMap{Network: components.Network}
 		if !components.IsEmpty() {
 			storeNM = types.CalculateNetworkMapFromComponents(ctx, components)
@@ -202,7 +208,7 @@ func checkAccount(ctx context.Context, t *testing.T, nmStore *networkmapdb.Netwo
 
 		// ACCOUNT PATH — Account → toNetworkMapData twins → components.
 		acctNM := account.GetPeerNetworkMapFromComponents(
-			ctx, peerID, nbdns.CustomZone{}, nil, validated, resourcePolicies, routers, nil, groupUsers,
+			ctx, peerID, accountPeersZone, accountZones, validated, resourcePolicies, routers, nil, groupUsers,
 		)
 		acctProto := mgmtgrpc.ToSyncResponse(
 			ctx, nil, nil, nil, types.TwinPeer(peer), nil, nil, acctNM, equivDNSName, nil,
@@ -211,7 +217,7 @@ func checkAccount(ctx context.Context, t *testing.T, nmStore *networkmapdb.Netwo
 
 		// LEGACY PATH — main's frozen copy.
 		legacyNM := legacynmap.GetPeerNetworkMapFromComponents(
-			&legacyAccount, ctx, peerID, nbdns.CustomZone{}, nil, validated, legacyResourcePolicies, routers, nil, groupUsers,
+			&legacyAccount, ctx, peerID, accountPeersZone, accountZones, validated, legacyResourcePolicies, routers, nil, groupUsers,
 		)
 		if legacyNM == nil {
 			t.Fatalf("after %d peers: account=%s peer=%s legacy NetworkMap nil, new non-nil", stats.peersChecked, account.Id, peerID)

management/server/types/legacynmap/proto_legacy.go

@@ -10,7 +10,7 @@ import (
 	"github.com/netbirdio/netbird/client/ssh/auth"
 	nbconfig "github.com/netbirdio/netbird/management/internals/server/config"
 	nbpeer "github.com/netbirdio/netbird/management/server/peer"
-	types "github.com/netbirdio/netbird/management/server/types"
+	"github.com/netbirdio/netbird/management/server/types"
 	nbroute "github.com/netbirdio/netbird/route"
 	"github.com/netbirdio/netbird/shared/management/networkmap"
 	"github.com/netbirdio/netbird/shared/management/proto"
@@ -89,7 +89,7 @@ func buildJWTConfig(config *nbconfig.HttpServerConfig, deviceFlowConfig *nbconfi
 
 	return &proto.JWTConfig{
 		Issuer:       issuer,
-		Audience:     audience,
+		Audience:     audience, //nolint:staticcheck
 		Audiences:    audiences,
 		KeysLocation: keysLocation,
 	}